Multi-agent pipelines in Haystack 2.0: a tutorial

If you’re searching for a multi-agent pipelines haystack 2.0 tutorial that actually runs, you’ve found it. Haystack 2.0’s component-based architecture makes multi-agent workflows explicit and debuggable — no more hidden control flow. This tutorial walks through building a research assistant that delegates fact retrieval, synthesis, and verification to separate agents, with full observability at each step.

Prerequisites

You need Python 3.10+ and an OpenAI-compatible API endpoint. Install the core packages:

pip install haystack-ai==2.0.0 openai python-dotenv

Create a .env file with your credentials:

OPENAI_API_KEY=sk-your-key-here
OPENAI_BASE_URL=https://api.openai.com/v1  # or your n4n.ai endpoint

If you’re routing through n4n.ai, the base URL becomes https://api.n4n.ai/v1 and you get automatic fallback across 240+ models without changing code.

Architecture overview

We’ll build three agents connected in a pipeline:

1. Researcher — searches and retrieves relevant passages
2. Synthesizer — combines findings into a coherent answer
3. Verifier — checks claims against sources and flags hallucinations

Each agent is a Haystack component with typed inputs and outputs. The pipeline orchestrates them sequentially, passing structured data between stages.

Step 1: Define the shared data structures

Create models.py to keep types explicit:

# models.py
from dataclasses import dataclass, field
from typing import Optional
from haystack import component


@dataclass
class Source:
    content: str
    metadata: dict = field(default_factory=dict)
    score: float = 0.0


@dataclass
class ResearchResult:
    query: str
    sources: list[Source]
    raw_response: str = ""


@dataclass
class SynthesisResult:
    answer: str
    citations: list[dict] = field(default_factory=list)
    confidence: float = 0.0


@dataclass
class VerificationResult:
    verified_answer: str
    flagged_claims: list[dict] = field(default_factory=list)
    overall_confidence: float = 0.0

These dataclasses become the contract between agents. Haystack 2.0’s type system validates connections at pipeline construction time.

Step 2: Build the Researcher agent

The researcher wraps a retriever and generator. We’ll use a simple in-memory document store for demonstration — swap in Elasticsearch, Weaviate, or Pinecone for production.

# researcher.py
from haystack import component, Document
from haystack.components.generators import OpenAIGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack import Pipeline

from models import ResearchResult, Source


@component
class Researcher:
    def __init__(self, document_store: InMemoryDocumentStore, top_k: int = 5):
        self.retriever = InMemoryBM25Retriever(document_store=document_store, top_k=top_k)
        self.generator = OpenAIGenerator(model="gpt-4o-mini")
        self.top_k = top_k

    @component.output_types(result=ResearchResult)
    def run(self, query: str) -> dict:
        # Retrieve relevant documents
        retrieved = self.retriever.run(query=query)
        documents: list[Document] = retrieved["documents"]

        sources = [
            Source(
                content=doc.content,
                metadata=doc.meta,
                score=doc.score or 0.0
            )
            for doc in documents
        ]

        # Generate initial response with citations
        context = "\n\n".join([f"[{i}] {s.content}" for i, s in enumerate(sources)])
        prompt = f"""Answer the query using only the provided sources. Cite sources with [number].

Query: {query}

Sources:
{context}

Answer:"""

        response = self.generator.run(prompt=prompt)
        raw_response = response["replies"][0]

        return {"result": ResearchResult(
            query=query,
            sources=sources,
            raw_response=raw_response
        )}

Seed the document store with test data:

# seed_data.py
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack import Document

document_store = InMemoryDocumentStore()

docs = [
    Document(content="Haystack 2.0 released in March 2024 introduces component-based pipelines.", meta={"source": "release-notes"}),
    Document(content="Components in Haystack 2.0 are typed, reusable building blocks with explicit inputs/outputs.", meta={"source": "docs"}),
    Document(content="The Pipeline class connects components via directed acyclic graphs.", meta={"source": "docs"}),
    Document(content="Haystack 1.x used Nodes and Pipelines with implicit control flow.", meta={"source": "migration-guide"}),
    Document(content="Multi-agent workflows are now first-class citizens in Haystack 2.0.", meta={"source": "blog"}),
]

document_store.write_documents(docs)

Run the researcher standalone to verify:

# test_researcher.py
from researcher import Researcher
from seed_data import document_store

researcher = Researcher(document_store)
result = researcher.run(query="What changed in Haystack 2.0?")
print(result["result"].raw_response)
print(f"\nSources: {len(result['result'].sources)}")

Expected output:

Haystack 2.0 introduced component-based pipelines with typed, reusable building blocks [0][1]. The Pipeline class now connects components via directed acyclic graphs [2], replacing the implicit control flow of 1.x [3]. Multi-agent workflows are first-class citizens [4].

Sources: 5

Step 3: Build the Synthesizer agent

The synthesizer takes research results and produces a structured answer with inline citations.

# synthesizer.py
from haystack import component
from haystack.components.generators import OpenAIGenerator

from models import ResearchResult, SynthesisResult, Source


@component
class Synthesizer:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.generator = OpenAIGenerator(model=model)

    @component.output_types(result=SynthesisResult)
    def run(self, research_result: ResearchResult) -> dict:
        sources_text = "\n\n".join(
            f"[{i}] {s.content} (source: {s.metadata.get('source', 'unknown')})"
            for i, s in enumerate(research_result.sources)
        )

        prompt = f"""Synthesize a clear, accurate answer from the research results. 
Include inline citations like [0], [1] referencing the source list.
Rate your confidence 0.0-1.0 based on source coverage.

Query: {research_result.query}

Sources:
{sources_text}

Raw research response:
{research_result.raw_response}

Respond in JSON:
{{
  "answer": "...",
  "citations": [{{"claim": "...", "source_indices": [0, 1]}}],
  "confidence": 0.0
}}"""

        response = self.generator.run(prompt=prompt)
        import json
        try:
            parsed = json.loads(response["replies"][0])
        except json.JSONDecodeError:
            # Fallback if model doesn't return valid JSON
            parsed = {
                "answer": response["replies"][0],
                "citations": [],
                "confidence": 0.5
            }

        return {"result": SynthesisResult(
            answer=parsed["answer"],
            citations=parsed.get("citations", []),
            confidence=parsed.get("confidence", 0.5)
        )}

Test it:

# test_synthesizer.py
from researcher import Researcher
from synthesizer import Synthesizer
from seed_data import document_store

researcher = Researcher(document_store)
synthesizer = Synthesizer()

research = researcher.run(query="How do Haystack 2.0 components differ from 1.x nodes?")
synthesis = synthesizer.run(research_result=research["result"])
print(synthesis["result"].answer)
print(f"Confidence: {synthesis['result'].confidence}")

Expected output:

Haystack 2.0 components are typed, reusable building blocks with explicit inputs and outputs [1], unlike 1.x nodes which relied on implicit control flow [3]. The Pipeline class now connects components via directed acyclic graphs [2], making data flow visible and debuggable.

Confidence: 0.85

Step 4: Build the Verifier agent

The verifier cross-checks each claim against the original sources — this catches hallucinations the synthesizer might introduce.

# verifier.py
from haystack import component
from haystack.components.generators import OpenAIGenerator

from models import SynthesisResult, VerificationResult, Source


@component
class Verifier:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.generator = OpenAIGenerator(model=model)

    @component.output_types(result=VerificationResult)
    def run(self, synthesis_result: SynthesisResult, sources: list[Source]) -> dict:
        sources_text = "\n\n".join(
            f"[{i}] {s.content}"
            for i, s in enumerate(sources)
        )

        claims_text = "\n".join(
            f"- {c.get('claim', '')} (cited sources: {c.get('source_indices', [])})"
            for c in synthesis_result.citations
        )

        prompt = f"""Verify each claim against the provided sources. 
Flag any claim that is unsupported, contradicted, or extrapolated beyond the source text.
Rate overall confidence 0.0-1.0.

Answer to verify:
{synthesis_result.answer}

Claims with citations:
{claims_text}

Original sources:
{sources_text}

Respond in JSON:
{{
  "verified_answer": "corrected answer if needed, otherwise original",
  "flagged_claims": [
    {{"claim": "...", "issue": "unsupported|contradicted|extrapolated", "source_indices": []}}
  ],
  "overall_confidence": 0.0
}}"""

        response = self.generator.run(prompt=prompt)
        import json
        try:
            parsed = json.loads(response["replies"][0])
        except json.JSONDecodeError:
            parsed = {
                "verified_answer": synthesis_result.answer,
                "flagged_claims": [],
                "overall_confidence": synthesis_result.confidence
            }

        return {"result": VerificationResult(
            verified_answer=parsed["verified_answer"],
            flagged_claims=parsed.get("flagged_claims", []),
            overall_confidence=parsed.get("overall_confidence", 0.5)
        )}

Step 5: Wire the pipeline

Haystack 2.0’s Pipeline class connects components by matching output names to input names. The type annotations we added (@component.output_types) enable static validation.

# pipeline.py
from haystack import Pipeline
from researcher import Researcher
from synthesizer import Synthesizer
from verifier import Verifier
from seed_data import document_store


def build_pipeline() -> Pipeline:
    pipe = Pipeline()

    pipe.add_component("researcher", Researcher(document_store))
    pipe.add_component("synthesizer", Synthesizer())
    pipe.add_component("verifier", Verifier())

    # researcher.result -> synthesizer.research_result
    pipe.connect("researcher.result", "synthesizer.research_result")
    # researcher.result.sources -> verifier.sources (need to extract)
    # synthesizer.result -> verifier.synthesis_result
    pipe.connect("synthesizer.result", "verifier.synthesis_result")

    return pipe

There’s a catch: the verifier needs both the synthesis result AND the original sources. Haystack 2.0 handles this with a small adapter component:

# adapter.py
from haystack import component
from models import ResearchResult, SynthesisResult, Source


@component
class SourceExtractor:
    @component.output_types(sources=list[Source])
    def run(self, research_result: ResearchResult) -> dict:
        return {"sources": research_result.sources}

Update the pipeline:

# pipeline.py (updated)
from haystack import Pipeline
from researcher import Researcher
from synthesizer import Synthesizer
from verifier import Verifier
from adapter import SourceExtractor
from seed_data import document_store


def build_pipeline() -> Pipeline:
    pipe = Pipeline()

    pipe.add_component("researcher", Researcher(document_store))
    pipe.add_component("extractor", SourceExtractor())
    pipe.add_component("synthesizer", Synthesizer())
    pipe.add_component("verifier", Verifier())

    pipe.connect("researcher.result", "synthesizer.research_result")
    pipe.connect("researcher.result", "extractor.research_result")
    pipe.connect("extractor.sources", "verifier.sources")
    pipe.connect("synthesizer.result", "verifier.synthesis_result")

    return pipe

Step 6: Run the full pipeline

# main.py
from pipeline import build_pipeline

pipe = build_pipeline()

query = "What are the key differences between Haystack 1.x and 2.0?"
result = pipe.run({"researcher": {"query": query}})

verification = result["verifier"]["result"]
print("=== FINAL ANSWER ===")
print(verification.verified_answer)
print(f"\nOverall confidence: {verification.overall_confidence:.2f}")

if verification.flagged_claims:
    print("\n=== FLAGGED CLAIMS ===")
    for flag in verification.flagged_claims:
        print(f"  - {flag['claim']}: {flag['issue']}")

Expected output:

=== FINAL ANSWER ===
Haystack 2.0 introduces component-based pipelines with typed, reusable building blocks that have explicit inputs and outputs. The Pipeline class connects components via directed acyclic graphs, replacing the implicit control flow of 1.x nodes. Multi-agent workflows are now first-class citizens.

Overall confidence: 0.92

=== FLAGGED CLAIMS ===

No flagged claims — the verifier confirmed all citations. Try a query that triggers hallucination:

# test_hallucination.py
from pipeline import build_pipeline

pipe = build_pipeline()

# This query has no support in our seed documents
query = "What is Haystack's pricing model for enterprise?"
result = pipe.run({"researcher": {"query": query}})

verification = result["verifier"]["result"]
print(verification.verified_answer)
print(f"Confidence: {verification.overall_confidence}")
for flag in verification.flagged_claims:
    print(f"FLAGGED: {flag}")

Expected output:

The provided sources do not contain information about Haystack's enterprise pricing model.

Confidence: 0.15
FLAGGED: {'claim': 'Haystack offers tiered enterprise pricing starting at $2000/month', 'issue': 'unsupported', 'source_indices': []}

The verifier caught the hallucination and forced a correction.

Step 7: Add observability

Haystack 2.0 emits structured logs at each component. Enable debug logging to trace execution:

# observability.py
import logging
import json
from haystack import Pipeline
from haystack.core.pipeline import PipelineWrapper

logging.basicConfig(level=logging.DEBUG)

# Or use the built-in tracer
from haystack.tracing import tracer

class JSONTracer:
    def trace(self, component_name: str, inputs: dict, outputs: dict):
        print(json.dumps({
            "component": component_name,
            "inputs": {k: str(v)[:200] for k, v in inputs.items()},
            "outputs": {k: str(v)[:200] for k, v in outputs.items()}
        }))

tracer.add_tracer(JSONTracer())

Run the pipeline again — you’ll see each component’s inputs and outputs as JSON lines, making debugging trivial.

Production considerations

Error handling

Wrap component run methods in try/except and return structured error objects:

@component
class Researcher:
    @component.output_types(result=ResearchResult, error=str)
    def run(self, query: str) -> dict:
        try:
            # ... existing logic
            return {"result": research_result}
        except Exception as e:
            return {"error": f"Researcher failed: {e}"}

The pipeline continues executing downstream components; check for error in results.

Parallel execution

Independent agents can run concurrently. The researcher and a hypothetical Summarizer agent could both consume the query simultaneously:

pipe.connect("query", "researcher.query")
pipe.connect("query", "summarizer.query")  # runs in parallel

Haystack 2.0 executes disconnected branches in parallel automatically.

Streaming responses

For long-running agents, implement run_async and stream tokens via callbacks. The OpenAIGenerator supports streaming natively — pass streaming_callback to the constructor.

Routing directives

When running behind a gateway that supports per-request routing (like n4n.ai), pass model preferences via the generator’s generation_kwargs:

generator = OpenAIGenerator(
    model="gpt-4o",
    generation_kwargs={"metadata": {"routing": {"prefer": "low-latency"}}}
)

The gateway honors these hints without code changes on your side.

Summary

You’ve built a three-agent pipeline with:

• Explicit typed contracts between agents
• Verification that catches hallucinations
• Full observability at each step
• Production-ready error handling patterns

Haystack 2.0’s component model makes multi-agent systems maintainable — each agent is independently testable, swappable, and debuggable. The pipeline definition is declarative and validated at construction time, not runtime.

Next steps: add a router agent that classifies queries and dispatches to specialized researcher agents (code, docs, API), implement persistent conversation memory, and wire evaluation metrics to track verification catch rates over time.