workflows-py/examples/eval_driven_prompt_refinement.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Eval-Driven Prompt Refinement\n",
        "\n",
        "This notebook demonstrates a self-contained workflow that refines prompts through an evaluate-and-iterate loop.\n",
        "It combines four key workflow patterns in a single cohesive scenario:\n",
        "\n",
        "- **Looping with conditional exit** — retry until the evaluation passes or max iterations are reached\n",
        "- **Branching** — approve, retry with feedback, or escalate to a human\n",
        "- **State management** — track prompt history, evaluation scores, and iteration count\n",
        "- **Human-in-the-loop** — escalation path when automated refinement fails\n",
        "\n",
        "The workflow uses a mock evaluator so **no API keys are needed**.\n",
        "\n",
        "### How it works\n",
        "\n",
        "```\n",
        "StartEvent\n",
        "    │\n",
        "    ▼\n",
        "generate  ◄──── GeneratePromptEvent (automated retry)\n",
        "    │          ◄──── HumanFeedbackResponse (HITL resume)\n",
        "    │\n",
        "    ▼\n",
        "EvalResultEvent\n",
        "    │\n",
        "    ▼\n",
        "evaluate_and_decide\n",
        "    ├── score >= threshold  ──►  handle_approval  ──►  StopEvent\n",
        "    ├── retries remaining   ──►  GeneratePromptEvent (loop)\n",
        "    └── retries exhausted   ──►  HumanFeedbackRequired (HITL)\n",
        "                                        │\n",
        "                                        ▼\n",
        "                               HumanFeedbackResponse → generate\n",
        "```"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {},
      "outputs": [],
      "source": [
        "!uv pip install llama-index-workflows -q"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 1. Define Events\n",
        "\n",
        "Each event type defines a distinct edge in the workflow graph."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {},
      "outputs": [],
      "source": [
        "from workflows.events import (\n",
        "    Event,\n",
        "    HumanResponseEvent,\n",
        "    InputRequiredEvent,\n",
        "    StartEvent,\n",
        "    StopEvent,\n",
        ")\n",
        "\n",
        "\n",
        "class GeneratePromptEvent(Event):\n",
        "    \"\"\"Triggers prompt generation. Carries optional feedback from a previous evaluation.\"\"\"\n",
        "\n",
        "    feedback: str = \"\"\n",
        "\n",
        "\n",
        "class EvalResultEvent(Event):\n",
        "    \"\"\"Carries the evaluation outcome for a candidate prompt.\"\"\"\n",
        "\n",
        "    prompt: str\n",
        "    score: float\n",
        "    feedback: str\n",
        "\n",
        "\n",
        "class ApprovedEvent(Event):\n",
        "    \"\"\"The prompt passed evaluation.\"\"\"\n",
        "\n",
        "    prompt: str\n",
        "    score: float\n",
        "\n",
        "\n",
        "class HumanFeedbackRequired(InputRequiredEvent):\n",
        "    \"\"\"Subclass of InputRequiredEvent so the caller can distinguish this prompt.\"\"\"\n",
        "\n",
        "    prompt: str\n",
        "    history: list[dict]\n",
        "\n",
        "\n",
        "class HumanFeedbackResponse(HumanResponseEvent):\n",
        "    \"\"\"Carries the human's feedback to resume the refinement loop.\"\"\"\n",
        "\n",
        "    response: str"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 2. Mock Evaluator\n",
        "\n",
        "A deterministic evaluator that scores prompts based on simple heuristics.\n",
        "Each criterion awards partial credit so the score increases as the prompt improves."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {},
      "outputs": [],
      "source": [
        "DATASET = [\n",
        "    {\"input\": \"What is 2+2?\", \"expected\": \"4\"},\n",
        "    {\"input\": \"Summarize: The cat sat on the mat.\", \"expected\": \"A cat sat on a mat.\"},\n",
        "    {\"input\": \"Translate to French: Hello\", \"expected\": \"Bonjour\"},\n",
        "]\n",
        "\n",
        "CRITERIA = [\n",
        "    (\"specific\", [\"specific\", \"precise\", \"exact\", \"concise\"]),\n",
        "    (\"structured\", [\"step\", \"first\", \"then\", \"finally\", \"1.\", \"2.\"]),\n",
        "    (\"role\", [\"you are\", \"act as\", \"your role\", \"as a\"]),\n",
        "    (\"examples\", [\"example\", \"e.g.\", \"for instance\", \"such as\"]),\n",
        "    (\"constraints\", [\"must\", \"do not\", \"avoid\", \"ensure\", \"always\"]),\n",
        "]\n",
        "\n",
        "\n",
        "class MockEvaluator:\n",
        "    \"\"\"Scores prompts 0.0-1.0 based on keyword heuristics.\"\"\"\n",
        "\n",
        "    def __init__(self, dataset: list[dict]):\n",
        "        self.dataset = dataset\n",
        "\n",
        "    def evaluate(self, prompt: str) -> tuple[float, str]:\n",
        "        lower = prompt.lower()\n",
        "        hits = []\n",
        "        misses = []\n",
        "        for name, keywords in CRITERIA:\n",
        "            if any(kw in lower for kw in keywords):\n",
        "                hits.append(name)\n",
        "            else:\n",
        "                misses.append(name)\n",
        "\n",
        "        score = len(hits) / len(CRITERIA)\n",
        "\n",
        "        if misses:\n",
        "            feedback = f\"Missing qualities: {', '.join(misses)}. Try adding language that is {', '.join(misses)}.\"\n",
        "        else:\n",
        "            feedback = \"All criteria met.\"\n",
        "\n",
        "        return round(score, 2), feedback"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 3. Mock Prompt Generator\n",
        "\n",
        "A template-based generator that incorporates feedback to produce\n",
        "incrementally better prompts. It adds at most **two** improvements per\n",
        "iteration, simulating realistic gradual refinement."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {},
      "outputs": [],
      "source": [
        "BASE_PROMPT = \"Answer the question.\"\n",
        "\n",
        "REFINEMENT_ADDITIONS = {\n",
        "    \"specific\": \"Be precise and concise in your answer.\",\n",
        "    \"structured\": \"First, understand the question. Then, provide a step-by-step answer.\",\n",
        "    \"role\": \"You are a helpful and knowledgeable assistant.\",\n",
        "    \"examples\": \"For instance, if asked about math, show your working (e.g. 2+2=4).\",\n",
        "    \"constraints\": \"You must answer accurately. Do not guess. Always verify your reasoning.\",\n",
        "}\n",
        "\n",
        "MAX_ADDITIONS_PER_ITERATION = 2\n",
        "\n",
        "\n",
        "def generate_prompt(feedback: str, previous_prompt: str | None = None) -> str:\n",
        "    \"\"\"Build a prompt by appending up to MAX_ADDITIONS_PER_ITERATION sentences\n",
        "    that address missing qualities mentioned in the feedback.\"\"\"\n",
        "    base = previous_prompt or BASE_PROMPT\n",
        "    additions = []\n",
        "    lower_feedback = feedback.lower()\n",
        "    for quality, sentence in REFINEMENT_ADDITIONS.items():\n",
        "        if quality in lower_feedback and sentence not in base:\n",
        "            additions.append(sentence)\n",
        "            if len(additions) >= MAX_ADDITIONS_PER_ITERATION:\n",
        "                break\n",
        "    if additions:\n",
        "        return base + \" \" + \" \".join(additions)\n",
        "    return base"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 4. Define the Workflow\n",
        "\n",
        "The `PromptRefinementWorkflow` ties everything together. HITL is modeled by `HumanFeedbackRequired` (a subclass of `InputRequiredEvent`) from the decide step and `HumanFeedbackResponse` (a `HumanResponseEvent`) feeding back into `generate`, without extra routing steps.\n",
        "\n",
        "| Step | Consumes | Produces | Pattern |\n",
        "|------|----------|----------|---------|\n",
        "| `generate` | `StartEvent`, `GeneratePromptEvent`, `HumanFeedbackResponse` | `EvalResultEvent` | **Loop** (re-entry); **HITL** resume |\n",
        "| `evaluate_and_decide` | `EvalResultEvent` | `ApprovedEvent` / `GeneratePromptEvent` / `HumanFeedbackRequired` | **Branching** (3-way) |\n",
        "| `handle_approval` | `ApprovedEvent` | `StopEvent` | Terminal |"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {},
      "outputs": [],
      "source": [
        "from pydantic import BaseModel\n",
        "from workflows import Context, Workflow, step\n",
        "\n",
        "\n",
        "class RefinementState(BaseModel):\n",
        "    iteration: int = 0\n",
        "    history: list[dict] = []\n",
        "    current_prompt: str = \"\"\n",
        "\n",
        "\n",
        "class PromptRefinementWorkflow(Workflow):\n",
        "    def __init__(\n",
        "        self,\n",
        "        evaluator: MockEvaluator,\n",
        "        max_retries: int = 3,\n",
        "        threshold: float = 0.8,\n",
        "        **kwargs,\n",
        "    ):\n",
        "        super().__init__(**kwargs)\n",
        "        self.evaluator = evaluator\n",
        "        self.max_retries = max_retries\n",
        "        self.threshold = threshold\n",
        "\n",
        "    # -- Step 1: Generate a candidate prompt --------------------------------\n",
        "\n",
        "    @step\n",
        "    async def generate(\n",
        "        self,\n",
        "        ctx: Context[RefinementState],\n",
        "        ev: StartEvent | GeneratePromptEvent | HumanFeedbackResponse,\n",
        "    ) -> EvalResultEvent:\n",
        "        if isinstance(ev, HumanFeedbackResponse):\n",
        "            async with ctx.store.edit_state() as state:\n",
        "                state.iteration = 0\n",
        "            feedback = ev.response\n",
        "            print(f\"  Human feedback received: {ev.response!r}\")\n",
        "        elif isinstance(ev, GeneratePromptEvent):\n",
        "            feedback = ev.feedback\n",
        "        else:\n",
        "            feedback = \"\"\n",
        "\n",
        "        async with ctx.store.edit_state() as state:\n",
        "            state.iteration += 1\n",
        "            previous = state.current_prompt or None\n",
        "            candidate = generate_prompt(feedback, previous)\n",
        "            state.current_prompt = candidate\n",
        "\n",
        "        print(f\"  [iteration {state.iteration}] Generated prompt: {candidate!r}\")\n",
        "\n",
        "        score, eval_feedback = self.evaluator.evaluate(candidate)\n",
        "        return EvalResultEvent(prompt=candidate, score=score, feedback=eval_feedback)\n",
        "\n",
        "    # -- Step 2: Decide — approve, retry, or HITL ---------------------------\n",
        "\n",
        "    @step\n",
        "    async def evaluate_and_decide(\n",
        "        self, ctx: Context[RefinementState], ev: EvalResultEvent\n",
        "    ) -> ApprovedEvent | GeneratePromptEvent | HumanFeedbackRequired:\n",
        "        async with ctx.store.edit_state() as state:\n",
        "            state.history.append(\n",
        "                {\"prompt\": ev.prompt, \"score\": ev.score, \"feedback\": ev.feedback}\n",
        "            )\n",
        "            iteration = state.iteration\n",
        "            history = list(state.history)\n",
        "\n",
        "        print(\n",
        "            f\"  [iteration {iteration}] Score: {ev.score} (threshold: {self.threshold})\"\n",
        "        )\n",
        "\n",
        "        if ev.score >= self.threshold:\n",
        "            return ApprovedEvent(prompt=ev.prompt, score=ev.score)\n",
        "\n",
        "        if iteration < self.max_retries:\n",
        "            print(f\"  [iteration {iteration}] Retrying with feedback: {ev.feedback}\")\n",
        "            return GeneratePromptEvent(feedback=ev.feedback)\n",
        "\n",
        "        print(f\"  [iteration {iteration}] Max retries reached — escalating to human.\")\n",
        "        return HumanFeedbackRequired(prompt=ev.prompt, history=history)\n",
        "\n",
        "    # -- Step 3: Approval — terminal step -----------------------------------\n",
        "\n",
        "    @step\n",
        "    async def handle_approval(\n",
        "        self, ctx: Context[RefinementState], ev: ApprovedEvent\n",
        "    ) -> StopEvent:\n",
        "        state = await ctx.store.get_state()\n",
        "        print(f\"  Approved after {state.iteration} iteration(s) with score {ev.score}.\")\n",
        "        return StopEvent(\n",
        "            result={\n",
        "                \"status\": \"approved\",\n",
        "                \"prompt\": ev.prompt,\n",
        "                \"score\": ev.score,\n",
        "                \"iterations\": state.iteration,\n",
        "            }\n",
        "        )"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 5. Run: Auto-Approve Path\n",
        "\n",
        "With `max_retries=5` and `threshold=0.8`, the mock evaluator will approve\n",
        "once the prompt accumulates enough quality keywords (4 out of 5 criteria = 0.8).\n",
        "Since the generator adds at most 2 improvements per iteration, this takes ~3 iterations."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "=== Auto-Approve Run ===\n",
            "  [iteration 1] Generated prompt: 'Answer the question.'\n",
            "  [iteration 1] Score: 0.0 (threshold: 0.8)\n",
            "  [iteration 1] Retrying with feedback: Missing qualities: specific, structured, role, examples, constraints. Try adding language that is specific, structured, role, examples, constraints.\n",
            "  [iteration 2] Generated prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer.'\n",
            "  [iteration 2] Score: 0.4 (threshold: 0.8)\n",
            "  [iteration 2] Retrying with feedback: Missing qualities: role, examples, constraints. Try adding language that is role, examples, constraints.\n",
            "  [iteration 3] Generated prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4).'\n",
            "  [iteration 3] Score: 0.8 (threshold: 0.8)\n",
            "  Approved after 3 iteration(s) with score 0.8.\n",
            "\n",
            "Result: {'status': 'approved', 'prompt': 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4).', 'score': 0.8, 'iterations': 3}\n"
          ]
        }
      ],
      "source": [
        "evaluator = MockEvaluator(dataset=DATASET)\n",
        "wf = PromptRefinementWorkflow(\n",
        "    evaluator=evaluator, max_retries=5, threshold=0.8, timeout=30\n",
        ")\n",
        "\n",
        "print(\"=== Auto-Approve Run ===\")\n",
        "result = await wf.run()\n",
        "print(f\"\\nResult: {result}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 6. Run: HITL Escalation Path\n",
        "\n",
        "Here we set `max_retries=2` so the workflow can only loop twice before\n",
        "escalating. With only 2 iterations adding 2 improvements each, the prompt\n",
        "reaches at most 4/5 criteria (score 0.8) which is below `threshold=1.0`.\n",
        "This forces escalation to a human.\n",
        "\n",
        "We consume events from the handler, simulate a human providing targeted\n",
        "feedback, and watch the loop resume."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "=== HITL Escalation Run ===\n",
            "  [iteration 1] Generated prompt: 'Answer the question.'\n",
            "  [iteration 1] Score: 0.0 (threshold: 1.0)\n",
            "  [iteration 1] Retrying with feedback: Missing qualities: specific, structured, role, examples, constraints. Try adding language that is specific, structured, role, examples, constraints.\n",
            "  [iteration 2] Generated prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer.'\n",
            "  [iteration 2] Score: 0.4 (threshold: 1.0)\n",
            "  [iteration 2] Max retries reached — escalating to human.\n",
            "\n",
            "  HITL requested! Current prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer.'\n",
            "  History (2 attempts):\n",
            "    score=0.0 | Answer the question.\n",
            "    score=0.4 | Answer the question. Be precise and concise in your answer. First, understand th\n",
            "\n",
            "  Sending human feedback: 'specific, structured, role, examples, constraints'\n",
            "  Human feedback received: 'specific, structured, role, examples, constraints'\n",
            "  [iteration 1] Generated prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4).'\n",
            "  [iteration 1] Score: 0.8 (threshold: 1.0)\n",
            "  [iteration 1] Retrying with feedback: Missing qualities: constraints. Try adding language that is constraints.\n",
            "  [iteration 2] Generated prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4). You must answer accurately. Do not guess. Always verify your reasoning.'\n",
            "  [iteration 2] Score: 1.0 (threshold: 1.0)\n",
            "  Approved after 2 iteration(s) with score 1.0.\n",
            "\n",
            "Result: {'status': 'approved', 'prompt': 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4). You must answer accurately. Do not guess. Always verify your reasoning.', 'score': 1.0, 'iterations': 2}\n"
          ]
        }
      ],
      "source": [
        "evaluator = MockEvaluator(dataset=DATASET)\n",
        "wf = PromptRefinementWorkflow(\n",
        "    evaluator=evaluator, max_retries=2, threshold=1.0, timeout=30\n",
        ")\n",
        "\n",
        "print(\"=== HITL Escalation Run ===\")\n",
        "handler = wf.run()\n",
        "\n",
        "async for ev in handler.stream_events():\n",
        "    if isinstance(ev, HumanFeedbackRequired):\n",
        "        print(f\"\\n  HITL requested! Current prompt: {ev.prompt!r}\")\n",
        "        print(f\"  History ({len(ev.history)} attempts):\")\n",
        "        for entry in ev.history:\n",
        "            print(f\"    score={entry['score']} | {entry['prompt'][:80]}\")\n",
        "\n",
        "        # Simulate a human providing targeted feedback\n",
        "        human_feedback = \"specific, structured, role, examples, constraints\"\n",
        "        print(f\"\\n  Sending human feedback: {human_feedback!r}\")\n",
        "        handler.ctx.send_event(HumanFeedbackResponse(response=human_feedback))\n",
        "\n",
        "result = await handler\n",
        "print(f\"\\nResult: {result}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 7. Inspect State\n",
        "\n",
        "After a run, we can read the workflow state to see the full refinement history\n",
        "— including the attempts before and after human intervention."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Total iterations: 2\n",
            "Final prompt: 'Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4). You must answer accurately. Do not guess. Always verify your reasoning.'\n",
            "\n",
            "Full history (4 entries):\n",
            "  1. score=0.00 | Answer the question.\n",
            "  2. score=0.40 | Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer.\n",
            "  3. score=0.80 | Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4).\n",
            "  4. score=1.00 | Answer the question. Be precise and concise in your answer. First, understand the question. Then, provide a step-by-step answer. You are a helpful and knowledgeable assistant. For instance, if asked about math, show your working (e.g. 2+2=4). You must answer accurately. Do not guess. Always verify your reasoning.\n"
          ]
        }
      ],
      "source": [
        "state = await handler.ctx.store.get_state()\n",
        "\n",
        "print(f\"Total iterations: {state.iteration}\")\n",
        "print(f\"Final prompt: {state.current_prompt!r}\")\n",
        "print(f\"\\nFull history ({len(state.history)} entries):\")\n",
        "for i, entry in enumerate(state.history, 1):\n",
        "    print(f\"  {i}. score={entry['score']:.2f} | {entry['prompt']}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summary\n",
        "\n",
        "This example demonstrated four workflow patterns working together:\n",
        "\n",
        "| Pattern | How it was used |\n",
        "|---------|----------------|\n",
        "| **Looping** | `evaluate_and_decide` returns `GeneratePromptEvent` to re-enter `generate` |\n",
        "| **Branching** | `evaluate_and_decide` chooses between `ApprovedEvent`, `GeneratePromptEvent`, or `HumanFeedbackRequired` |\n",
        "| **State management** | `RefinementState` tracks iteration count, prompt history, and current prompt via `ctx.store` |\n",
        "| **Human-in-the-loop** | At max retries, `evaluate_and_decide` emits `HumanFeedbackRequired`; the caller responds with `HumanFeedbackResponse`, which re-enters `generate` directly |\n",
        "\n",
        "### Next steps\n",
        "\n",
        "- Replace `MockEvaluator` with a real LLM-based judge (e.g., OpenAI)\n",
        "- Replace `generate_prompt()` with an LLM call that takes feedback as input\n",
        "- Add parallel evaluation of multiple prompt candidates using `ctx.send_event()` + `ctx.collect_events()`\n",
        "- Persist state across restarts using `ctx.to_dict()` / `Context.from_dict()`\n",
        "\n",
        "See also:\n",
        "- [Feature Walkthrough](feature_walkthrough.ipynb) for individual pattern walkthroughs\n",
        "- [Durable Workflows](durable_workflows.ipynb) for state persistence\n",
        "- [Document Processing](document_processing.ipynb) for a real-world HITL example with LlamaCloud"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": ".venv",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.10.19"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 4
}