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

This tutorial shows you how to build a Haystack pipeline that routes queries conditionally based on simple rules like keyword detection and document count. You need this when one path should answer directly, while another should fall back to retrieval or a different generator.

What You'll Need

• •Python 3.10+
• •haystack-ai installed
• •An OpenAI API key if you want to use an LLM-based generator
• •A working internet connection for model calls
• •Basic familiarity with Haystack Pipeline, Component, and DocumentStore concepts

Install the packages:

pip install haystack-ai openai

Step-by-Step

1. •Start with a small set of documents and a document store. For this example, we’ll use an in-memory store so the whole pipeline is easy to run locally.

from haystack import Document
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
documents = [
    Document(content="Haystack is a framework for building LLM applications."),
    Document(content="Conditional routing helps choose different pipeline paths."),
    Document(content="Python developers often route based on query intent."),
]
document_store.write_documents(documents)

2. •Add a retriever so the pipeline can fetch relevant documents when the query needs context. We’ll use BM25 because it works without embeddings or extra setup.

from haystack.components.retrievers.in_memory import InMemoryBM25Retriever

retriever = InMemoryBM25Retriever(document_store=document_store)

3. •Create two small routing components: one decides whether the query is a greeting, and the other decides whether retrieved context is strong enough. These are plain Haystack components, so they plug into a pipeline cleanly.

from typing import Literal
from haystack import component

@component
class GreetingRouter:
    @component.output_types(route=Literal["greeting", "search"])
    def run(self, query: str):
        greetings = {"hi", "hello", "hey"}
        route = "greeting" if query.lower().strip() in greetings else "search"
        return {"route": route}

@component
class ContextRouter:
    @component.output_types(route=Literal["answer", "fallback"])
    def run(self, documents):
        route = "answer" if len(documents) >= 2 else "fallback"
        return {"route": route}

4. •Add two generators: one for direct greeting replies and one for question answering from retrieved context. If you do not want to call an external model yet, you can still wire the routing logic first and replace these later.

from haystack.components.builders import PromptBuilder

greeting_prompt = PromptBuilder(
    template="Reply briefly and politely to this user message: {{query}}"
)

qa_prompt = PromptBuilder(
    template="""
Use only the documents below to answer the question.

Question: {{query}}

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

Answer:
"""
)

5. •Build the pipeline and connect the conditional branches. The key idea is that the first router sends greetings directly to one prompt, while all other queries go through retrieval and then through a second router.

from haystack import Pipeline

pipeline = Pipeline()

pipeline.add_component("greeting_router", GreetingRouter())
pipeline.add_component("greeting_prompt", greeting_prompt)
pipeline.add_component("retriever", retriever)
pipeline.add_component("context_router", ContextRouter())
pipeline.add_component("qa_prompt", qa_prompt)

pipeline.connect("greeting_router.route", "greeting_prompt.query")
pipeline.connect("greeting_router.route", "retriever.query")
pipeline.connect("retriever.documents", "context_router.documents")
pipeline.connect("retriever.documents", "qa_prompt.documents")

6. •Run the pipeline with a few different inputs and inspect which branch gets used. This version prints prompts rather than calling an external generator, which keeps the control flow easy to verify.

result_hello = pipeline.run({"greeting_router": {"query": "hello"}})
result_query = pipeline.run({"greeting_router": {"query": "What is conditional routing?"}})

print(result_hello["greeting_prompt"]["prompt"])
print("---")
print(result_query["qa_prompt"]["prompt"])

Testing It

Run three inputs: a greeting like hello, a broad question like What is Haystack?, and a more specific question like How do I route based on intent?. You should see greetings going straight to greeting_prompt, while non-greetings go through retrieval and land in qa_prompt.

If you want stricter validation, print intermediate outputs from each component and confirm that GreetingRouter returns the expected route every time. Also check that BM25 returns at least one document for relevant queries; otherwise your second router will push traffic into fallback logic once you add it.

Next Steps

• •Replace the prompt-only branches with OpenAIChatGenerator or another chat generator.
• •Add a third branch for low-confidence retrieval results and send those to a human-review queue.
• •Extend routing beyond keywords by using an LLM classifier or embedding-based intent detector.

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.