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

This tutorial shows you how to add a human approval step into a Haystack pipeline in Python. You’d use this when an LLM is about to answer something sensitive, and you want a person to review, edit, or block the response before it reaches the user.

What You'll Need

• •Python 3.10+
• •haystack-ai
• •An LLM provider package if you want to call a real model, for example:
  • •openai
• •An API key for your model provider, set as an environment variable
• •Basic familiarity with Haystack pipelines and components

Install the packages:

pip install haystack-ai openai

Step-by-Step

1. •Start with a simple pipeline that generates an answer from retrieved documents. The human-in-the-loop part will sit between generation and final delivery, so keep the first version small and readable.

from haystack import Document, Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Company policy: Refunds are allowed within 30 days."),
    Document(content="Company policy: Password resets require MFA verification."),
])

retriever = InMemoryBM25Retriever(document_store=document_store)
prompt_builder = PromptBuilder(
    template="""
You are a support assistant.
Use the documents to answer the question.

Question: {{ question }}

Documents:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

Answer:
"""
)

generator = OpenAIChatGenerator(model="gpt-4o-mini")

pipeline = Pipeline()
pipeline.add_component("retriever", retriever)
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("generator", generator)

pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder.prompt", "generator.messages")

2. •Add a small approval function that pauses execution and asks a human to approve or edit the draft. In production, this is usually replaced by a UI, ticketing workflow, or Slack approval, but the core logic stays the same.

def human_review(draft: str) -> str:
    print("\n--- DRAFT RESPONSE ---")
    print(draft)
    print("----------------------\n")

    decision = input("Approve this response? (y/n): ").strip().lower()
    if decision == "y":
        return draft

    edited = input("Enter revised response: ").strip()
    return edited if edited else draft

3. •Run the pipeline, extract the generated text, then send it through the review step before returning it to the user. This keeps generation automated while making final output explicitly human-controlled.

question = "What is our refund policy?"
result = pipeline.run({
    "retriever": {"query": question},
    "prompt_builder": {"question": question},
    "generator": {
        "messages": [
            {"role": "user", "content": question}
        ]
    }
})

draft_answer = result["generator"]["replies"][0].text
final_answer = human_review(draft_answer)

print("\nFINAL ANSWER:")
print(final_answer)

4. •If you want cleaner orchestration, wrap generation and review into a single function. This makes it easier to reuse in an API route or background job without scattering approval logic across your codebase.

def answer_with_review(question: str) -> str:
    result = pipeline.run({
        "retriever": {"query": question},
        "prompt_builder": {"question": question},
        "generator": {
            "messages": [
                {"role": "user", "content": question}
            ]
        }
    })

    draft = result["generator"]["replies"][0].text
    return human_review(draft)


if __name__ == "__main__":
    print(answer_with_review("What is our password reset policy?"))

5. •For a more realistic setup, store both the draft and the approved version. That gives you auditability, which matters when humans override model output in regulated environments.

review_log = []

def audited_human_review(question: str, draft: str) -> str:
    approved_text = human_review(draft)
    review_log.append({
        "question": question,
        "draft": draft,
        "approved_text": approved_text,
        "approved": approved_text == draft,
    })
    return approved_text


question = "What is our refund policy?"
result = pipeline.run({
    "retriever": {"query": question},
    "prompt_builder": {"question": question},
    "generator": {"messages": [{"role": "user", "content": question}]}
})

draft_answer = result["generator"]["replies"][0].text
final_answer = audited_human_review(question, draft_answer)

print(final_answer)
print(review_log[-1])

Testing It

Run the script and ask a question that matches one of the stored documents. You should see a draft answer printed first, then an approval prompt in your terminal.

Approve it with y and confirm the final answer matches the generated draft. Run it again and choose n so you can test manual editing.

If you want to verify retrieval is working, change one of the document strings and see whether the generated response changes accordingly. If it doesn’t, check your retriever wiring and prompt template first.

Next Steps

• •Replace input() with a web-based review queue or internal admin UI.
• •Add confidence-based routing so only low-confidence answers go to human review.
• •Store approvals in your database with document IDs, timestamps, and reviewer identity for audit trails.

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.