Haystack Tutorial (Python): adding human-in-the-loop for intermediate developers

This tutorial shows you how to add a human approval step into a Haystack pipeline so an LLM answer does not go straight to the user unless it passes review. You need this when the model is generating customer-facing content, policy decisions, or anything where a wrong answer creates operational or compliance risk.

What You'll Need

• •Python 3.10+
• •haystack-ai
• •An OpenAI API key
• •A terminal and editor
• •Basic familiarity with Haystack pipelines and components
• •Optional: python-dotenv if you want to load secrets from a .env file

Install the dependencies:

pip install haystack-ai openai python-dotenv

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start with a small pipeline that produces an answer. We’ll use a minimal retrieval-free setup first so the human-in-the-loop pattern is easy to see. The key idea is that the LLM generates a draft, then your application pauses before releasing it.

from haystack import Pipeline, component
from haystack.components.builders import PromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

template = """
You are a concise assistant for insurance support.
Answer the question in 3 sentences or fewer.

Question: {{question}}
"""

prompt_builder = PromptBuilder(template=template)
llm = OpenAIChatGenerator(model="gpt-4o-mini")

pipe = Pipeline()
pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("llm", llm)
pipe.connect("prompt_builder.prompt", "llm.messages")

result = pipe.run(
    {
        "prompt_builder": {"question": "What does comprehensive coverage usually include?"},
        "llm": {"generation_kwargs": {"temperature": 0.2}},
    }
)

draft = result["llm"]["replies"][0].content
print(draft)

2. •Add a human review gate in your application code. Haystack does not force human approval into the pipeline itself; you usually wrap the pipeline with an approval function. That keeps the workflow explicit and easier to audit.

def request_human_approval(text: str) -> bool:
    print("\n--- Draft for review ---\n")
    print(text)
    print("\nApprove this response? [y/n]: ", end="")
    decision = input().strip().lower()
    return decision in {"y", "yes"}

if request_human_approval(draft):
    final_answer = draft
else:
    final_answer = "I’m unable to provide that answer without further review."

print("\n--- Final response ---\n")
print(final_answer)

3. •Turn the approval step into a reusable component. If you want this pattern inside a larger flow, make the human gate a Haystack component. This is useful when you want structured outputs and consistent integration points.

from typing import Any, Dict

@component
class HumanApprovalGate:
    @component.output_types(approved=bool, text=str)
    def run(self, draft: str) -> Dict[str, Any]:
        print("\n--- Draft for review ---\n")
        print(draft)
        decision = input("\nApprove this response? [y/n]: ").strip().lower()
        approved = decision in {"y", "yes"}
        return {
            "approved": approved,
            "text": draft if approved else "I’m unable to provide that answer without further review.",
        }

4. •Wire the gate into the pipeline output path. Here we keep generation separate from approval so the handoff is obvious. In production, this also makes it easier to log drafts, approvals, and rejections.

pipe_with_gate = Pipeline()
pipe_with_gate.add_component("prompt_builder", PromptBuilder(template=template))
pipe_with_gate.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
pipe_with_gate.add_component("gate", HumanApprovalGate())

pipe_with_gate.connect("prompt_builder.prompt", "llm.messages")

result = pipe_with_gate.run(
    {
        "prompt_builder": {"question": "Can I increase my deductible to lower premiums?"},
        "llm": {"generation_kwargs": {"temperature": 0.2}},
        "gate": {"draft": None},
    }
)

draft = result["llm"]["replies"][0].content
gate_result = pipe_with_gate.get_component("gate").run(draft=draft)
print(gate_result["text"])

5. •Add metadata so reviewers know what they are approving. In real systems, humans need context: source documents, model name, timestamp, and risk reason. Even if you start simple, structure your review payload now so you do not have to refactor later.

from datetime import datetime

review_packet = {
    "question": "What does comprehensive coverage usually include?",
    "draft": draft,
    "model": "gpt-4o-mini",
    "timestamp_utc": datetime.utcnow().isoformat(),
    "risk_level": "customer-facing",
}

print("\n--- Review packet ---")
for key, value in review_packet.items():
    print(f"{key}: {value}")

Testing It

Run the script and confirm you see two stages: the generated draft and the approval prompt. Approve once with y and verify the original draft is returned unchanged.

Then reject once with n and verify your fallback message is returned instead of the model output. If you are using this in an API service, log both the draft and reviewer decision so you can trace who approved what and when.

A good test is to feed in questions that should be blocked or rewritten by policy. The goal is not just “does it run,” but “does it stop unsafe content from reaching the caller.”

Next Steps

• •Add retrieval before generation so reviewers can see source passages alongside the draft.
• •Store approvals in a database table with request ID, reviewer ID, timestamp, and decision.
• •Replace manual console input with an internal approval UI or Slack-based workflow for production use.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.