Haystack agent pipeline tutorial: branching and loops

Haystack 2.0’s agent pipelines let you express multi-step reasoning as a graph instead of a linear chain. This haystack agent pipeline branching loops tutorial walks through a concrete example: a research assistant that plans, searches, evaluates, and either loops for more evidence or branches to a final answer. You’ll see the exact components, wiring, and runnable code to get this working locally.

Prerequisites

• Python 3.10+
• haystack-ai>=2.0.0 (not the legacy farm-haystack package)
• An OpenAI API key or any OpenAI-compatible endpoint
• Basic familiarity with Haystack components and pipelines

Install the dependencies:

pip install "haystack-ai>=2.0.0" python-dotenv

Create a .env file with your API key:

OPENAI_API_KEY=sk-...

Architecture Overview

The pipeline has four nodes:

1. Planner — An LLM that breaks the user question into a search plan (list of queries).
2. Search — A web search component (we’ll use a simple SerperDev wrapper) that executes each query.
3. Evaluator — An LLM that scores whether the collected evidence is sufficient.
4. Branching logic — If evidence is insufficient, loop back to the planner with a refined prompt; if sufficient, branch to the final answer generator.

Haystack 2.0 represents this as a Pipeline with conditional edges using Pipeline.add_edge() and the condition parameter on Pipeline.run().

Step 1: Define the Planner Component

The planner takes the user question and (optionally) previous evidence, then outputs a structured list of search queries.

# planner.py
from haystack import component
from haystack.dataclasses import ChatMessage
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import ChatPromptBuilder
from pydantic import BaseModel, Field
from typing import List
import json

class SearchPlan(BaseModel):
    queries: List[str] = Field(description="List of search queries to execute")
    reasoning: str = Field(description="Why these queries will help answer the question")

@component
class Planner:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.generator = OpenAIGenerator(model=model)
        self.prompt_builder = ChatPromptBuilder(
            template=[
                ChatMessage.from_system(
                    "You are a research planner. Given a question and any existing evidence, "
                    "produce a JSON object with 'queries' (list of strings) and 'reasoning' (string). "
                    "Output ONLY valid JSON."
                ),
                ChatMessage.from_user(
                    "Question: {{question}}\n\n"
                    "Existing evidence:\n{{evidence}}\n\n"
                    "Produce the search plan as JSON."
                ),
            ]
        )

    @component.output_types(plan=SearchPlan)
    def run(self, question: str, evidence: str = ""):
        prompt = self.prompt_builder.run(question=question, evidence=evidence)
        response = self.generator.run(prompt=prompt["prompt"])
        raw = response["replies"][0]
        plan = SearchPlan.model_validate_json(raw)
        return {"plan": plan}

Test it in isolation:

# test_planner.py
from planner import Planner

planner = Planner()
result = planner.run(question="What are the latest developments in nuclear fusion?")
print(result["plan"].model_dump_json(indent=2))

Expected output:

{
  "queries": [
    "nuclear fusion breakthrough 2024",
    "ITER project status 2024",
    "private fusion companies funding 2024"
  ],
  "reasoning": "Need recent breakthroughs, major project status, and private sector activity to cover the landscape."
}

Step 2: Implement the Search Component

We’ll wrap SerperDev’s API. You can swap this for any search provider.

# search.py
from haystack import component
from typing import List
import os
import requests

@component
class SerperSearch:
    def __init__(self, api_key: str | None = None, top_k: int = 5):
        self.api_key = api_key or os.getenv("SERPER_API_KEY")
        self.top_k = top_k
        if not self.api_key:
            raise ValueError("SERPER_API_KEY not set")

    @component.output_types(results=List[dict])
    def run(self, queries: List[str]):
        all_results = []
        for query in queries:
            resp = requests.post(
                "https://google.serper.dev/search",
                headers={"X-API-KEY": self.api_key, "Content-Type": "application/json"},
                json={"q": query, "num": self.top_k},
                timeout=15,
            )
            resp.raise_for_status()
            data = resp.json()
            for item in data.get("organic", []):
                all_results.append({
                    "query": query,
                    "title": item.get("title"),
                    "snippet": item.get("snippet"),
                    "link": item.get("link"),
                })
        return {"results": all_results}

Test it:

# test_search.py
from search import SerperSearch

search = SerperSearch()
results = search.run(queries=["nuclear fusion breakthrough 2024"])
for r in results["results"][:2]:
    print(f"- {r['title']}: {r['snippet'][:120]}...")

Step 3: Build the Evaluator

The evaluator decides whether we have enough evidence to answer. It returns a boolean sufficient and a critique explaining the gap.

# evaluator.py
from haystack import component
from haystack.dataclasses import ChatMessage
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import ChatPromptBuilder
from pydantic import BaseModel, Field

class Evaluation(BaseModel):
    sufficient: bool = Field(description="Whether evidence is sufficient to answer")
    critique: str = Field(description="What is missing or why it's sufficient")

@component
class Evaluator:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.generator = OpenAIGenerator(model=model)
        self.prompt_builder = ChatPromptBuilder(
            template=[
                ChatMessage.from_system(
                    "You are an evidence evaluator. Given a question and collected evidence, "
                    "decide if the evidence is sufficient to answer the question. "
                    "Output ONLY valid JSON with 'sufficient' (boolean) and 'critique' (string)."
                ),
                ChatMessage.from_user(
                    "Question: {{question}}\n\n"
                    "Evidence:\n{{evidence}}\n\n"
                    "Evaluate sufficiency."
                ),
            ]
        )

    @component.output_types(evaluation=Evaluation)
    def run(self, question: str, evidence: str):
        prompt = self.prompt_builder.run(question=question, evidence=evidence)
        response = self.generator.run(prompt=prompt["prompt"])
        raw = response["replies"][0]
        evaluation = Evaluation.model_validate_json(raw)
        return {"evaluation": evaluation}

Step 4: Final Answer Generator

A straightforward LLM call that synthesizes the answer from evidence.

# answer_generator.py
from haystack import component
from haystack.dataclasses import ChatMessage
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import ChatPromptBuilder

@component
class AnswerGenerator:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.generator = OpenAIGenerator(model=model)
        self.prompt_builder = ChatPromptBuilder(
            template=[
                ChatMessage.from_system(
                    "You are a research assistant. Answer the question using ONLY the provided evidence. "
                    "Cite sources inline with [title](link) format. If evidence is insufficient, say so."
                ),
                ChatMessage.from_user(
                    "Question: {{question}}\n\n"
                    "Evidence:\n{{evidence}}\n\n"
                    "Answer:"
                ),
            ]
        )

    @component.output_types(answer=str)
    def run(self, question: str, evidence: str):
        prompt = self.prompt_builder.run(question=question, evidence=evidence)
        response = self.generator.run(prompt=prompt["prompt"])
        return {"answer": response["replies"][0]}

Step 5: Wire the Pipeline with Branching and Loops

Now the haystack agent pipeline branching loops tutorial gets to the core: connecting these components with conditional edges. Haystack 2.0 uses Pipeline.add_edge() with a condition callable that receives the previous component’s output.

# pipeline.py
from haystack import Pipeline
from planner import Planner
from search import SerperSearch
from evaluator import Evaluator
from answer_generator import AnswerGenerator

def build_pipeline(max_iterations: int = 3) -> Pipeline:
    pipe = Pipeline()

    # Add components
    pipe.add_component("planner", Planner())
    pipe.add_component("search", SerperSearch())
    pipe.add_component("evaluator", Evaluator())
    pipe.add_component("answer_generator", AnswerGenerator())

    # Linear flow: planner -> search -> evaluator
    pipe.connect("planner.plan", "search.queries")
    pipe.connect("search.results", "evaluator.evidence")

    # Conditional edge: evaluator -> answer_generator (if sufficient)
    pipe.add_edge(
        "evaluator",
        "answer_generator",
        condition=lambda output: output["evaluation"].sufficient,
    )

    # Conditional edge: evaluator -> planner (if insufficient, loop back)
    pipe.add_edge(
        "evaluator",
        "planner",
        condition=lambda output: not output["evaluation"].sufficient,
    )

    # The loop needs state: pass the evaluation critique back to planner as evidence
    # We'll handle this by connecting evaluator's critique to planner's evidence input
    # via a custom component or by using the pipeline's data flow. Simpler: use a
    # join component that accumulates evidence across iterations.

    return pipe

The loop needs to accumulate evidence across iterations. Haystack doesn’t have a built-in “loop with accumulator” primitive, so we add a lightweight EvidenceAccumulator component.

# accumulator.py
from haystack import component
from typing import List

@component
class EvidenceAccumulator:
    def __init__(self):
        self.collected: List[dict] = []

    @component.output_types(evidence_str=str, all_results=List[dict])
    def run(self, new_results: List[dict] | None = None, reset: bool = False):
        if reset:
            self.collected = []
        if new_results:
            self.collected.extend(new_results)
        # Format for LLM consumption
        evidence_str = "\n\n".join(
            f"Source: {r['title']} ({r['link']})\nSnippet: {r['snippet']}"
            for r in self.collected
        )
        return {"evidence_str": evidence_str, "all_results": self.collected}

Update the pipeline to use it:

# pipeline.py (updated)
from haystack import Pipeline
from planner import Planner
from search import SerperSearch
from evaluator import Evaluator
from answer_generator import AnswerGenerator
from accumulator import EvidenceAccumulator

def build_pipeline(max_iterations: int = 3) -> Pipeline:
    pipe = Pipeline()

    pipe.add_component("planner", Planner())
    pipe.add_component("search", SerperSearch())
    pipe.add_component("accumulator", EvidenceAccumulator())
    pipe.add_component("evaluator", Evaluator())
    pipe.add_component("answer_generator", AnswerGenerator())

    # planner -> search
    pipe.connect("planner.plan", "search.queries")

    # search -> accumulator (accumulates across loops)
    pipe.connect("search.results", "accumulator.new_results")

    # accumulator -> evaluator
    pipe.connect("accumulator.evidence_str", "evaluator.evidence")

    # evaluator -> answer_generator (sufficient)
    pipe.add_edge(
        "evaluator",
        "answer_generator",
        condition=lambda output: output["evaluation"].sufficient,
    )

    # evaluator -> planner (insufficient) — loop back with critique
    # We need to feed the critique into the planner's evidence input.
    # The planner expects an 'evidence' string. We'll connect evaluator's
    # critique via a small adapter, but Haystack lets us use the same
    # accumulator output plus the critique. Simpler: connect evaluator
    # to planner directly with a custom condition that also passes data.
    # For clarity, add a critique-to-evidence adapter:

    from haystack import component

    @component
    class CritiqueToEvidence:
        @component.output_types(evidence=str)
        def run(self, critique: str, current_evidence: str):
            return {"evidence": f"{current_evidence}\n\n[Critique from previous iteration]: {critique}"}

    pipe.add_component("critique_adapter", CritiqueToEvidence())
    pipe.connect("evaluator.evaluation.critique", "critique_adapter.critique")
    pipe.connect("accumulator.evidence_str", "critique_adapter.current_evidence")
    pipe.connect("critique_adapter.evidence", "planner.evidence")

    # Conditional loop edge
    pipe.add_edge(
        "evaluator",
        "planner",
        condition=lambda output: not output["evaluation"].sufficient,
    )

    return pipe

Step 6: Run the Pipeline

The pipeline’s run() method accepts the initial inputs. Because we have a loop, we need to pass the question to both the planner and evaluator. Use Pipeline.run() with the data dict mapping component input names.

# main.py
import os
from dotenv import load_dotenv
from pipeline import build_pipeline

load_dotenv()

def main():
    pipe = build_pipeline(max_iterations=3)

    question = "What are the latest developments in nuclear fusion as of 2024?"

    # Initial inputs: planner needs question; evaluator needs question; accumulator starts fresh
    result = pipe.run(
        data={
            "planner": {"question": question},
            "evaluator": {"question": question},
            "accumulator": {"reset": True},
        },
        # Haystack 2.0: max_runs prevents infinite loops
        max_runs=10,
    )

    # The answer_generator output is nested under its component name
    answer = result.get("answer_generator", {}).get("answer")
    if answer:
        print("\n=== FINAL ANSWER ===\n")
        print(answer)
    else:
        print("Pipeline did not produce an answer. Check logs.")

if __name__ == "__main__":
    main()

Run it:

python main.py

Expected output (truncated):

=== FINAL ANSWER ===

As of 2024, nuclear fusion has seen several notable developments:

1. **National Ignition Facility (NIF) repeated ignition** — In 2023 and 2024, NIF achieved fusion ignition multiple times, producing more energy from fusion than the laser energy delivered to the target [NIF Achieves Fusion Ignition](https://www.llnl.gov/news/nif-achieves-fusion-ignition).

2. **ITER assembly progress** — The ITER project in France has completed the tokamak building and begun installing major components, with first plasma targeted for 2025 [ITER Project Status](https://www.iter.org/proj/inafewlines).

3. **Private sector funding surge** — Companies like Commonwealth Fusion Systems, Helion Energy, and TAE Technologies raised over $2.5B collectively in 2023-2024 [Fusion Industry Association Report](https://fusionindustryassociation.org/2024-report).

...

Step 7: Observability — Inspecting the Loop

To debug branching and loops, enable Haystack’s built-in tracing or log the pipeline’s execution graph.

# debug_run.py
from pipeline import build_pipeline
import json

pipe = build_pipeline()

# Visualize the graph
pipe.draw("pipeline_graph.png")  # Requires graphviz: pip install graphviz

# Run with a simple tracer
from haystack.tracing import tracer
from haystack.tracing.sentry import SentryTracer
# Or use the built-in logging tracer
import logging
logging.basicConfig(level=logging.DEBUG)

result = pipe.run(
    data={
        "planner": {"question": "What is the status of ITER?"},
        "evaluator": {"question": "What is the status of ITER?"},
        "accumulator": {"reset": True},
    },
    max_runs=10,
)

print(json.dumps(result, indent=2, default=str))

The max_runs parameter caps total component executions across the loop, preventing runaway iterations. Each loop iteration counts as additional runs.

Common Pitfalls

1. Forgetting to reset the accumulator between unrelated questions.
The EvidenceAccumulator holds state in self.collected. Always pass {"accumulator": {"reset": True}} at the start of each new query.

2. Infinite loops when the evaluator never returns sufficient: true.
Set max_runs conservatively (e.g., 10–15). Add a hard iteration counter in the planner prompt: “This is iteration N of 3.”

3. Type mismatches on conditional edges.
The condition callable receives the entire output dict of the source component. In our evaluator, the output is {"evaluation": Evaluation}. So output["evaluation"].sufficient works. If you change the output type, update the lambda.

4. Passing the question to every component that needs it.
Haystack 2.0 doesn’t automatically broadcast inputs. Explicitly include "question": question in the data dict for each component that declares it as an input.

Extending the Pattern

This haystack agent pipeline branching loops tutorial demonstrates the core pattern. You can extend it in several directions:

• Parallel search: Fan out to multiple search components (Serper, Bing, ArXiv) and merge results before evaluation.
• Tool use: Replace the search component with a ToolInvoker that calls arbitrary functions; the planner emits tool calls instead of queries.
• Human-in-the-loop: Add a conditional edge that pauses for human review when confidence < threshold.
• Streaming: Use OpenAIGenerator(streaming=True) and handle partial outputs in the answer generator.

When to Use This vs. a Linear Chain

Use branching and loops when:

• The task requires adaptive information gathering (you don’t know upfront how many searches you need).
• You need verification before committing to an answer.
• The problem naturally decomposes into plan → act → evaluate → repeat.

Stick to a linear Pipeline when the steps are fixed and known ahead of time — retrieval-augmented generation with a single retrieval step, for example.

The complete runnable code is available in this gist. Adjust the models, search provider, and prompts for your domain. The structure — planner, actor, evaluator, accumulator, conditional edges — transfers directly to more complex agentic workflows.