Haystack Tutorial (Python): implementing guardrails for beginners

This tutorial shows how to add guardrails to a Haystack pipeline so you can block unsafe, off-topic, or malformed LLM outputs before they reach your users. You need this when your assistant must stay inside policy, avoid prompt injection fallout, and return predictable answers for downstream systems.

What You'll Need

• •Python 3.10+
• •Haystack installed with OpenAI support:
  • •haystack-ai
  • •openai
• •An OpenAI API key set as an environment variable:
  • •OPENAI_API_KEY
• •Basic familiarity with:
  • •Haystack Pipeline
  • •Chat generators
  • •Python functions and imports

Install the packages:

pip install haystack-ai openai

Step-by-Step

1. •First, define the guardrail rules in plain Python. For beginners, keep it simple: reject empty answers, block unsafe content, and reject responses that do not mention the topic you asked for.

from typing import Any

def guardrail_check(answer: str, topic: str) -> dict[str, Any]:
    text = answer.strip().lower()

    if not text:
        return {"allowed": False, "reason": "empty_answer"}

    blocked_terms = ["password", "credit card", "ssn", "social security number"]
    if any(term in text for term in blocked_terms):
        return {"allowed": False, "reason": "sensitive_data"}

    if topic.lower() not in text:
        return {"allowed": False, "reason": "off_topic"}

    return {"allowed": True, "reason": "ok"}

2. •Next, build a small Haystack pipeline that generates an answer from a prompt. This keeps the example production-shaped: one component generates content, another component validates it.

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

template = """
Answer the user's question in one short paragraph.
Question: {{ question }}
Topic: {{ topic }}
"""

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")

3. •Now run the pipeline and inspect the generated response before returning it. The guardrail sits outside the LLM call on purpose so you can enforce deterministic checks in Python.

question = "What is Haystack?"
topic = "Haystack"

result = pipe.run(
    data={
        "prompt_builder": {
            "question": question,
            "topic": topic,
        }
    }
)

messages = result["llm"]["replies"]
answer = messages[0].text

check = guardrail_check(answer=answer, topic=topic)

print("Answer:", answer)
print("Guardrail:", check)

4. •Add a fallback path for blocked outputs. In real apps, this is where you return a safe refusal or ask the model to try again with tighter constraints.

def safe_response(answer: str, topic: str) -> str:
    check = guardrail_check(answer=answer, topic=topic)

    if check["allowed"]:
        return answer

    if check["reason"] == "off_topic":
        return f"I can only answer questions about {topic}."

    if check["reason"] == "sensitive_data":
        return "I can't help with sensitive data."

    return "I couldn't generate a valid answer."

final_answer = safe_response(answer=answer, topic=topic)
print(final_answer)

5. •Finally, wrap the whole flow into a reusable function. This is the version you would call from an API endpoint or an internal assistant service.

def answer_with_guardrails(question: str, topic: str) -> str:
    result = pipe.run(
        data={
            "prompt_builder": {
                "question": question,
                "topic": topic,
            }
        }
    )

    answer = result["llm"]["replies"][0].text
    check = guardrail_check(answer=answer, topic=topic)

    if check["allowed"]:
        return answer

    return safe_response(answer=answer, topic=topic)


if __name__ == "__main__":
    print(answer_with_guardrails("What is Haystack?", "Haystack"))

Testing It

Run the script with a valid OPENAI_API_KEY and confirm that it prints a normal answer when the output stays on-topic. Then test a harder prompt like "Tell me my credit card number" and make sure your fallback path blocks it. If you want to verify the off-topic rule specifically, change topic to "Haystack" and ask something unrelated like "Explain Kubernetes". In production, I would also log check["reason"] so you can measure which guardrails are firing most often.

Next Steps

• •Add a second validation pass using another LLM call for semantic policy checks.
• •Move the guardrail logic into a custom Haystack component so it can live inside the pipeline.
• •Add structured output parsing with JSON schema validation before your app returns anything to users.

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.