Haystack agent pipeline tutorial with DeepSeek V3

Haystack 2.0 moved agents from experimental to first-class citizens. The Agent component now composes cleanly with pipelines, supports structured tool calling, and streams token-by-token. This tutorial builds a working agent that uses DeepSeek V3 for reasoning and a handful of tools for retrieval and calculation. You’ll end up with a pipeline you can extend for production workloads.

Prerequisites

Python 3.10+ and an OpenRouter-compatible API key. DeepSeek V3 is available through any OpenRouter endpoint — n4n.ai exposes one at https://api.n4n.ai/v1 with automatic fallback across providers, but the code below works with any base URL that implements the OpenAI chat completions interface.

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

Create a .env file:

OPENROUTER_API_KEY=sk-or-v1-...
OPENROUTER_BASE_URL=https://api.n4n.ai/v1   # or https://openrouter.ai/api/v1

Project structure

haystack-deepseek-agent/
├── .env
├── main.py
├── tools/
│   ├── __init__.py
│   ├── calculator.py
│   └── web_search.py
└── pipeline.yaml       # optional, for serialization

The LLM wrapper

Haystack’s OpenAIGenerator works with any OpenAI-compatible endpoint. DeepSeek V3 expects the model identifier deepseek/deepseek-chat on OpenRouter.

# main.py
import os
from dotenv import load_dotenv
from haystack.components.generators import OpenAIGenerator
from haystack.components.generators.utils import print_streaming_chunk

load_dotenv()

llm = OpenAIGenerator(
    api_key=os.getenv("OPENROUTER_API_KEY"),
    api_base_url=os.getenv("OPENROUTER_BASE_URL"),
    model="deepseek/deepseek-chat",
    generation_kwargs={
        "temperature": 0.3,
        "max_tokens": 4096,
    },
    streaming_callback=print_streaming_chunk,
)

Run a quick sanity check:

if __name__ == "__main__":
    response = llm.run(prompt="Say 'hello' in three words.")
    print(response["replies"][0])

Expected output (streamed):

Hello there friend

Building tools

Haystack 2.0 tools are plain Python functions decorated with @tool. The agent sees the function signature, docstring, and type hints — so write them like you’d write a public API.

Calculator tool

# tools/calculator.py
from haystack.tools import tool
import math

@tool
def calculate(expression: str) -> str:
    """
    Evaluate a mathematical expression safely.
    
    Args:
        expression: A math expression like "2 * (3 + 4) / 5" or "sqrt(16) + log(100)"
    
    Returns:
        The numeric result as a string, or an error message.
    """
    allowed_names = {
        "abs": abs, "round": round, "min": min, "max": max,
        "sum": sum, "pow": pow,
        "sqrt": math.sqrt, "log": math.log, "log10": math.log10,
        "sin": math.sin, "cos": math.cos, "tan": math.tan,
        "pi": math.pi, "e": math.e,
    }
    try:
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

Web search tool (stubbed for reproducibility)

Replace this with a real provider (SerpAPI, Tavily, Brave) in production. The shape matters more than the implementation here.

# tools/web_search.py
from haystack.tools import tool
from typing import List, Dict
import json

@tool
def web_search(query: str, max_results: int = 5) -> str:
    """
    Search the web and return summarized results.
    
    Args:
        query: Search query string.
        max_results: Maximum number of results to return (default 5).
    
    Returns:
        JSON string with list of {title, url, snippet} objects.
    """
    # TODO: integrate real search API
    mock_results = [
        {"title": "DeepSeek V3 Release Notes", "url": "https://example.com/deepseek-v3", "snippet": "DeepSeek V3 achieves SOTA on coding benchmarks with 671B parameters."},
        {"title": "Haystack 2.0 Agent Guide", "url": "https://example.com/haystack-agents", "snippet": "Agents in Haystack 2.0 are pipeline-native components with tool calling."},
    ]
    return json.dumps(mock_results[:max_results])

Wire them into __init__.py for clean imports:

# tools/__init__.py
from .calculator import calculate
from .web_search import web_search

__all__ = ["calculate", "web_search"]

Constructing the agent

The Agent component takes the generator, a list of tools, and a system prompt that defines behavior. Haystack 2.0 uses the ToolCallingAgent pattern — the model decides when to call tools, gets results back, and continues.

# main.py (continued)
from haystack.agents import Agent
from haystack.agents import ToolCallingAgent
from tools import calculate, web_search

system_prompt = """
You are a precise technical assistant. You have access to tools for calculation and web search.
- Use calculate for any math, unit conversions, or numerical reasoning.
- Use web_search for current facts, documentation, or information beyond your training cutoff.
- Always show your reasoning before calling a tool.
- Cite sources from web_search results by title and URL.
- If a tool fails, acknowledge the failure and try an alternative approach.
"""

agent = ToolCallingAgent(
    generator=llm,
    tools=[calculate, web_search],
    system_prompt=system_prompt,
    max_tool_calls=5,           # prevent infinite loops
    raise_on_tool_failure=False,  # let the agent handle errors
)

Wrapping in a pipeline

Haystack pipelines give you observability, serialization, and component reuse. The agent is just another component.

# main.py (continued)
from haystack import Pipeline
from haystack.components.joiners import DocumentJoiner  # if you add retrieval later

pipe = Pipeline()
pipe.add_component("agent", agent)

# Optional: add a simple input/output adapter for cleaner API
from haystack.components.others import Multiplexer

pipe.add_component("query_mux", Multiplexer(str))
pipe.connect("query_mux", "agent")

Run it:

if __name__ == "__main__":
    questions = [
        "What's the square root of 671 billion? Use the calculator.",
        "Search for 'Haystack 2.0 agent pipeline' and summarize the key points.",
        "If DeepSeek V3 has 671B parameters and each parameter is 2 bytes (FP16), how many GB of VRAM for weights alone?",
    ]

    for q in questions:
        print(f"\n{'='*60}")
        print(f"Q: {q}")
        print(f"{'='*60}")
        result = pipe.run({"query_mux": {"value": q}})
        print(result["agent"]["replies"][0])

Expected output (checkpoint 1)

============================================================
Q: What's the square root of 671 billion? Use the calculator.
============================================================
I'll calculate the square root of 671 billion for you.

[Tool call: calculate(expression='sqrt(671_000_000_000)')]
[Tool result: 819145.894...]

The square root of 671 billion is approximately **819,145.89**.

Expected output (checkpoint 2)

============================================================
Q: Search for 'Haystack 2.0 agent pipeline' and summarize the key points.
============================================================
[Tool call: web_search(query='Haystack 2.0 agent pipeline', max_results=5)]
[Tool result: [{"title": "Haystack 2.0 Agent Guide", "url": "https://example.com/haystack-agents", "snippet": "Agents in Haystack 2.0 are pipeline-native components with tool calling."}]]

Based on the search results:

**Key points about Haystack 2.0 agent pipelines:**
- Agents are now first-class pipeline components (`ToolCallingAgent`)
- Native tool calling with automatic schema generation from function signatures
- Support for streaming responses token-by-token
- Configurable max tool calls to prevent loops
- Serializable to YAML for deployment

Source: "Haystack 2.0 Agent Guide" (https://example.com/haystack-agents)

Expected output (checkpoint 3)

============================================================
Q: If DeepSeek V3 has 671B parameters and each parameter is 2 bytes (FP16), how many GB of VRAM for weights alone?
============================================================
I'll calculate the VRAM requirement for DeepSeek V3 weights in FP16.

[Tool call: calculate(expression='671_000_000_000 * 2 / (1024**3)')]
[Tool result: 1245.8...]

At FP16 (2 bytes per parameter), 671 billion parameters require approximately **1,246 GB** of VRAM for model weights alone.

This exceeds single-GPU capacity — practical deployment requires tensor parallelism across multiple GPUs (e.g., 8× H100 80GB = 640GB, still insufficient; you'd need 16+ GPUs or quantization to 4-bit/8-bit).

Streaming responses

The streaming_callback on the generator handles token-level streaming. For pipeline-level streaming, use pipe.run_async() with an async generator:

# main.py (add to bottom)
import asyncio
from haystack import AsyncPipeline

async_pipe = AsyncPipeline()
async_pipe.add_component("agent", agent)
async_pipe.add_component("query_mux", Multiplexer(str))
async_pipe.connect("query_mux", "agent")

async def stream_query(question: str):
    print(f"Q: {question}\nA: ", end="", flush=True)
    async for chunk in async_pipe.run_async({"query_mux": {"value": question}}):
        # chunk contains partial replies during streaming
        if "replies" in chunk.get("agent", {}):
            for reply in chunk["agent"]["replies"]:
                print(reply, end="", flush=True)
    print()

if __name__ == "__main__":
    asyncio.run(stream_query("Explain the difference between tool calling and function calling in one paragraph."))

Serializing to YAML

Haystack pipelines serialize to YAML for version control and deployment. The agent serializes its tool schemas automatically.

# main.py (add)
pipe.dump("pipeline.yaml")

Resulting pipeline.yaml (truncated):

components:
  agent:
    type: haystack.agents.ToolCallingAgent
    init_parameters:
      generator:
        type: haystack.components.generators.OpenAIGenerator
        init_parameters:
          model: deepseek/deepseek-chat
          api_base_url: https://api.n4n.ai/v1
          generation_kwargs:
            temperature: 0.3
            max_tokens: 4096
      tools:
        - name: calculate
          parameters:
            type: object
            properties:
              expression:
                type: string
            required: [expression]
        - name: web_search
          parameters:
            type: object
            properties:
              query:
                type: string
              max_results:
                type: integer
                default: 5
            required: [query]
      system_prompt: |
        You are a precise technical assistant...
      max_tool_calls: 5
      raise_on_tool_failure: false
  query_mux:
    type: haystack.components.others.Multiplexer
    init_parameters:
      type: str
connections:
  - sender: query_mux
    receiver: agent

Load it back:

from haystack import Pipeline
loaded = Pipeline.load("pipeline.yaml")
result = loaded.run({"query_mux": {"value": "What is 2^10?"}})

Production hardening

Structured output with Pydantic

For downstream consumers, constrain the final answer format:

from pydantic import BaseModel, Field
from typing import List, Optional

class AgentResponse(BaseModel):
    answer: str
    tools_used: List[str] = Field(default_factory=list)
    sources: List[dict] = Field(default_factory=list)
    confidence: float = Field(ge=0.0, le=1.0)

Add a second generator step that formats the agent’s raw output into this schema, or use Haystack’s StructuredOutputGenerator (available in 2.6+).

Error handling and retries

Wrap tool calls with tenacity for transient failures:

# tools/calculator.py (enhanced)
from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(wait=wait_exponential(multiplier=1, min=1, max=10), stop=stop_after_attempt(3))
def calculate(expression: str) -> str:
    # ... same implementation

Observability

Hook Haystack’s logging or integrate with OpenTelemetry:

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("haystack")
logger.setLevel(logging.DEBUG)  # shows tool calls, prompt templates, latency

Rate limits and fallback

If you route through n4n.ai, the gateway handles provider-level fallbacks automatically — DeepSeek V3 on one provider fails, it retries on another. You still need application-level handling for 429s:

from haystack.components.generators import OpenAIGenerator
from openai import RateLimitError

class ResilientGenerator(OpenAIGenerator):
    def run(self, *args, **kwargs):
        try:
            return super().run(*args, **kwargs)
        except RateLimitError:
            # Could trigger a model switch here
            raise

Extending: adding retrieval

The agent pattern scales. Add a DocumentRetriever tool that queries your vector store:

# tools/retriever.py
from haystack.tools import tool
from haystack import Document
from typing import List

@tool
def search_docs(query: str, top_k: int = 5) -> str:
    """
    Search internal knowledge base.
    """
    # In reality: embed query, hit Qdrant/Weaviate/Pinecone, return Document objects
    docs = [
        Document(content="DeepSeek V3 uses MoE with 256 experts...", meta={"source": "tech-report.pdf"}),
        Document(content="Haystack 2.0 released March 2024...", meta={"source": "changelog.md"}),
    ]
    return "\n\n".join(f"[{d.meta['source']}] {d.content}" for d in docs[:top_k])

Register it in the agent’s tool list and the model will call it when relevant.

Summary

You now have a Haystack 2.0 agent pipeline that:

• Runs DeepSeek V3 via any OpenRouter-compatible endpoint
• Exposes typed tools for calculation and search
• Streams token-by-token
• Serializes to YAML for deployment
• Extends cleanly with retrieval, structured output, and observability

The agent is a pipeline component — compose it with classifiers, routers, evaluators, or other agents. That’s the point of Haystack 2.0: agents aren’t special snowflakes anymore. They’re just components that happen to reason.