Haystack Tutorial (Python): building conditional routing for advanced developers

This tutorial shows how to build a conditional routing pipeline in Haystack that sends queries down different paths based on the input. You need this when one model or tool is not enough and you want deterministic branching for things like question type, compliance checks, or escalation.

What You'll Need

• •Python 3.10+
• •haystack-ai installed
• •An OpenAI API key if you want to use an LLM-based router or generator
• •Basic familiarity with Haystack Pipeline, components, and Document
• •Optional: a local environment variable setup for secrets

pip install haystack-ai
export OPENAI_API_KEY="your-key"

Step-by-Step

1. •Start by defining two simple processing paths: one for normal questions and one for fallback/escalation. In production, these branches might map to different prompts, different retrievers, or a human-review queue.

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

@component
class SimpleRouter:
    @component.output_types(route=str)
    def run(self, question: str):
        q = question.lower()
        if any(word in q for word in ["refund", "chargeback", "dispute"]):
            return {"route": "escalate"}
        return {"route": "answer"}

router = SimpleRouter()

2. •Build two downstream components that represent the branches. For this example, both branches use an LLM generator, but they get different prompts so you can see the routing decision clearly.

answer_prompt = PromptBuilder(
    template="Answer the user directly and concisely:\nQuestion: {{question}}"
)

escalate_prompt = PromptBuilder(
    template=(
        "This looks like a sensitive banking issue. "
        "Explain that the case must be reviewed by support.\nQuestion: {{question}}"
    )
)

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

3. •Wire the router into a pipeline and connect each branch explicitly. Haystack pipelines are strict about inputs and outputs, which is exactly what you want when routing must be deterministic.

pipe = Pipeline()

pipe.add_component("router", router)
pipe.add_component("answer_prompt", answer_prompt)
pipe.add_component("escalate_prompt", escalate_prompt)
pipe.add_component("llm_answer", llm)
pipe.add_component("llm_escalate", OpenAIChatGenerator(model="gpt-4o-mini"))

pipe.connect("router.route", "answer_prompt.question")
pipe.connect("router.route", "escalate_prompt.question")

4. •The previous connection is not enough by itself because the router emits a route label, not the original question text. A better pattern is to pass the original input through both branches and use the route only as a selector inside your own orchestration logic.

@component
class RouteSelector:
    @component.output_types(answer_question=str, escalate_question=str)
    def run(self, question: str):
        return {
            "answer_question": question,
            "escalate_question": question,
        }

selector = RouteSelector()

5. •Now combine the selector with the router result in a small orchestration layer. This keeps your pipeline components clean and makes branch decisions easy to test without hiding logic inside prompt templates.

def run_routed(question: str):
    route_result = router.run(question=question)["route"]
    selector_result = selector.run(question=question)

    if route_result == "escalate":
        prompt_result = escalate_prompt.run(question=selector_result["escalate_question"])
        messages = [ChatMessage.from_user(prompt_result["prompt"])]
        return llm.run(messages=messages)

    prompt_result = answer_prompt.run(question=selector_result["answer_question"])
    messages = [ChatMessage.from_user(prompt_result["prompt"])]
    return llm.run(messages=messages)

6. •Execute it with both a normal query and a sensitive one. In real systems, replace the simple keyword router with an LLM classifier or rules engine once you have clear routing categories.

if __name__ == "__main__":
    print(run_routed("How do I reset my password?"))
    print(run_routed("I need a refund for a duplicate charge"))

Testing It

Run the script with two inputs and confirm they land on different branches. The password query should go through the direct-answer path, while the refund query should hit the escalation path.

If you want stronger verification, add unit tests around SimpleRouter.run() first. Then test run_routed() with mocked generators so you only validate routing behavior, not model output.

For production, log the selected route along with request IDs and user metadata allowed by your policy. That gives you traceability when compliance teams ask why a request was escalated.

Next Steps

• •Replace SimpleRouter with an LLM-based classifier using OpenAIChatGenerator
• •Add a third branch for document retrieval before generation
• •Move from ad hoc orchestration to a full Haystack pipeline with explicit branch components

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.