Haystack 2.x moved agents from experimental to production-ready. The Agent class now handles tool calling, memory, and multi-step reasoning out of the box. This tutorial builds a working agent that searches the web, extracts answers, and cites sources — using only components you can run locally or with free API tiers.
Prerequisites
- Python 3.10+
- An OpenAI API key (or any OpenAI-compatible endpoint)
- A Serper.dev API key for web search (free tier: 2,500 queries/month)
- Basic familiarity with Haystack 2.x components and pipelines
Install the dependencies:
pip install haystack-ai==2.7.0 \
haystack-components==0.3.0 \
python-dotenv==1.0.1 \
requests==2.32.3Create a .env file:
OPENAI_API_KEY=sk-...
SERPER_API_KEY=...Minimal agent with one search tool
Haystack’s Agent expects a list of Tool objects. Each tool wraps a component with a name, description, and the component itself. Start with a single web search tool using SerperDev.
# minimal_agent.py
import os
from dotenv import load_dotenv
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators import OpenAIGenerator
from haystack.components.tools import Tool
from haystack.components.websearch import SerperDevWebSearch
from haystack.dataclasses import ChatMessage
load_dotenv()
# 1. Configure the LLM
llm = OpenAIGenerator(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY"),
generation_kwargs={"temperature": 0.1}
)
# 2. Create the search component
search = SerperDevWebSearch(
api_key=os.getenv("SERPER_API_KEY"),
top_k=5
)
# 3. Wrap it as a tool
search_tool = Tool(
name="web_search",
description="Search the web for current information. Use for facts, news, prices, or anything not in training data.",
component=search
)
# 4. Build the agent
agent = Agent(
tools=[search_tool],
generator=llm,
system_prompt=(
"You are a research assistant. Answer questions using the web_search tool. "
"Always cite sources with URLs. If the tool returns no results, say so."
),
max_steps=5
)
# 5. Run a query
question = "What is the current price of a Tesla Model 3 in the US?"
result = agent.run(messages=[ChatMessage.from_user(question)])
# 6. Print the final answer
for msg in result["messages"]:
if msg.role == "assistant" and msg.text:
print(msg.text)Run it:
python minimal_agent.pyExpected output (prices will vary):
Based on current web search results, the Tesla Model 3 pricing in the US as of 2024:
**Model 3 RWD (Standard Range):** $38,990
**Model 3 Long Range AWD:** $45,990
**Model 3 Performance:** $52,990
These are base prices before incentives. Federal tax credit of up to $7,500 may apply depending on battery sourcing and income limits.
Sources:
- https://www.tesla.com/model3/design - Official Tesla configurator
- https://www.caranddriver.com/tesla/model-3 - Car and Driver pricing summary
- https://www.edmunds.com/tesla/model-3/ - Edmunds pricing analysisThe agent called web_search, synthesized the results, and cited URLs. Check the raw tool output by inspecting result["messages"] — you’ll see ToolCall and ToolCallResult messages in the conversation history.
Adding a second tool: content extraction
Search snippets often lack detail. Add a second tool that fetches and extracts full page content.
# agent_with_extraction.py
import os
from dotenv import load_dotenv
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators import OpenAIGenerator
from haystack.components.tools import Tool
from haystack.components.websearch import SerperDevWebSearch
from haystack.components.fetchers import LinkContentFetcher
from haystack.components.converters import HTMLToDocument
from haystack.dataclasses import ChatMessage
load_dotenv()
llm = OpenAIGenerator(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY"),
generation_kwargs={"temperature": 0.1}
)
search = SerperDevWebSearch(
api_key=os.getenv("SERPER_API_KEY"),
top_k=5
)
# Fetch + convert pipeline for extraction
fetcher = LinkContentFetcher()
converter = HTMLToDocument()
extract_pipeline = Pipeline()
extract_pipeline.add_component("fetcher", fetcher)
extract_pipeline.add_component("converter", converter)
extract_pipeline.connect("fetcher.streams", "converter.sources")
extract_tool = Tool(
name="extract_content",
description="Fetch and extract full text from a URL. Use after web_search to get details from specific sources.",
component=extract_pipeline,
parameters={
"type": "object",
"properties": {
"urls": {"type": "array", "items": {"type": "string"}}
},
"required": ["urls"]
}
)
agent = Agent(
tools=[search_tool, extract_tool],
generator=llm,
system_prompt=(
"You are a research assistant. Use web_search to find relevant URLs, then "
"use extract_content to get full details from the most promising sources. "
"Always cite sources with URLs."
),
max_steps=8
)
question = "What are the key differences between Python 3.12 and 3.13?"
result = agent.run(messages=[ChatMessage.from_user(question)])
for msg in result["messages"]:
if msg.role == "assistant" and msg.text:
print(msg.text)Expected output structure:
Python 3.13 (released October 2024) introduces several key changes over 3.12:
**Performance**
- 10-15% faster interpreter startup via tier 2 compilation (PEP 744)
- Improved `list` and `dict` operations through specialized adaptive interpreter
**New Features**
- `dbm.sqlite3` backend for `dbm` module (PEP 751)
- `random.random()` now uses ChaCha20 for better quality
- Experimental JIT compiler (disabled by default, enable with `--enable-experimental-jit`)
**Removals/Deprecations**
- `cgi` module removed (deprecated since 3.11)
- `cgitb` module removed
- `telnetlib` removed
**Typing Improvements**
- `typing.TypeIs` for type narrowing (PEP 742)
- `typing.ReadOnly` for TypedDict fields (PEP 705)
Sources:
- https://docs.python.org/3/whatsnew/3.13.html - Official changelog
- https://peps.python.org/pep-0744/ - Tier 2 compilation detailsThe agent now chains tools: search → select URLs → extract → synthesize. The max_steps=8 prevents infinite loops on complex queries.
Structured output with Pydantic
For production use, you want structured responses — not free text. Define a schema and have the agent return JSON.
# structured_agent.py
import os
import json
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import List
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators import OpenAIGenerator
from haystack.components.tools import Tool
from haystack.components.websearch import SerperDevWebSearch
from haystack.components.fetchers import LinkContentFetcher
from haystack.components.converters import HTMLToDocument
from haystack.dataclasses import ChatMessage
load_dotenv()
class Source(BaseModel):
url: str
title: str
snippet: str
class ResearchAnswer(BaseModel):
question: str
answer: str
sources: List[Source]
confidence: float = Field(ge=0.0, le=1.0)
llm = OpenAIGenerator(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY"),
generation_kwargs={"temperature": 0.0}
)
search = SerperDevWebSearch(
api_key=os.getenv("SERPER_API_KEY"),
top_k=5
)
fetcher = LinkContentFetcher()
converter = HTMLToDocument()
extract_pipeline = Pipeline()
extract_pipeline.add_component("fetcher", fetcher)
extract_pipeline.add_component("converter", converter)
extract_pipeline.connect("fetcher.streams", "converter.sources")
search_tool = Tool(
name="web_search",
description="Search the web for current information.",
component=search
)
extract_tool = Tool(
name="extract_content",
description="Fetch full content from URLs.",
component=extract_pipeline,
parameters={
"type": "object",
"properties": {"urls": {"type": "array", "items": {"type": "string"}}},
"required": ["urls"]
}
)
# System prompt enforces JSON output matching the schema
system_prompt = f"""You are a research assistant. Use tools to answer the question.
Return ONLY valid JSON matching this schema:
{ResearchAnswer.model_json_schema()}
Rules:
- Search first, then extract from top 2-3 URLs
- Include 3-5 sources with title, URL, and 1-sentence snippet
- Confidence: 0.0-1.0 based on source quality and agreement
- No markdown, no extra text — only the JSON object"""
agent = Agent(
tools=[search_tool, extract_tool],
generator=llm,
system_prompt=system_prompt,
max_steps=8
)
question = "What is the current federal funds rate in the US?"
result = agent.run(messages=[ChatMessage.from_user(question)])
# Parse and validate
for msg in result["messages"]:
if msg.role == "assistant" and msg.text:
try:
parsed = ResearchAnswer.model_validate_json(msg.text)
print(json.dumps(parsed.model_dump(), indent=2))
except Exception as e:
print(f"Parse failed: {e}")
print(f"Raw output: {msg.text}")Output:
{
"question": "What is the current federal funds rate in the US?",
"answer": "As of July 2024, the federal funds rate target range is 5.25% - 5.50%, unchanged since July 2023. The FOMC has held this rate steady through 2024 meetings.",
"sources": [
{
"url": "https://www.federalreserve.gov/monetarypolicy/fomccalendars.htm",
"title": "FOMC Meeting Calendar - Federal Reserve",
"snippet": "Official FOMC statements confirm 5.25-5.50% target range maintained through June 2024 meeting."
},
{
"url": "https://www.reuters.com/markets/us/fed-holds-interest-rates-steady-2024-06-12/",
"title": "Fed holds interest rates steady, signals one cut this year",
"snippet": "Reuters reports Fed kept rates at 5.25-5.50% at June 2024 meeting."
},
{
"url": "https://www.bloomberg.com/news/articles/2024-07-31/fed-rate-decision-july-2024",
"title": "Fed Rate Decision July 2024",
"snippet": "Bloomberg confirms no rate change at July 31, 2024 FOMC meeting."
}
],
"confidence": 0.95
}The schema forces consistent structure. Downstream consumers can rely on answer, sources, and confidence without parsing prose.
Error handling and retries
Tools fail — network timeouts, rate limits, parsing errors. Wrap tools to handle this gracefully.
# resilient_agent.py
import os
import time
from dotenv import load_dotenv
from haystack import Pipeline, component
from haystack.components.agents import Agent
from haystack.components.generators import OpenAIGenerator
from haystack.components.tools import Tool
from haystack.components.websearch import SerperDevWebSearch
from haystack.dataclasses import ChatMessage
from haystack import logging
load_dotenv()
logging.basicConfig(level=logging.INFO)
llm = OpenAIGenerator(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY"),
generation_kwargs={"temperature": 0.1}
)
# Custom component with retry logic
@component
class ResilientWebSearch:
def __init__(self, api_key: str, top_k: int = 5, max_retries: int = 3):
self.search = SerperDevWebSearch(api_key=api_key, top_k=top_k)
self.max_retries = max_retries
@component.output_types(results=list)
def run(self, query: str):
last_error = None
for attempt in range(self.max_retries):
try:
result = self.search.run(query=query)
return {"results": result["results"]}
except Exception as e:
last_error = e
wait = 2 ** attempt # exponential backoff
logging.warning(f"Search attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
time.sleep(wait)
# All retries exhausted — return empty results instead of crashing
logging.error(f"All retries exhausted for query: {query}")
return {"results": []}
resilient_search = ResilientWebSearch(
api_key=os.getenv("SERPER_API_KEY"),
top_k=5,
max_retries=3
)
search_tool = Tool(
name="web_search",
description="Search the web with automatic retries on failure.",
component=resilient_search
)
agent = Agent(
tools=[search_tool],
generator=llm,
system_prompt="Search the web and answer. If search returns no results, say you couldn't find current information.",
max_steps=5
)
# Test with a query that might hit rate limits
result = agent.run(messages=[ChatMessage.from_user("Latest SpaceX Starship launch date?")])
for msg in result["messages"]:
if msg.role == "assistant" and msg.text:
print(msg.text)The @component decorator makes ResilientWebSearch a first-class Haystack component. It returns empty results on total failure rather than raising — the agent sees a valid (empty) tool result and can respond gracefully.
Running in a pipeline for observability
For production, wrap the agent in a Pipeline to add logging, tracing, or custom components before/after.
# pipeline_agent.py
import os
from dotenv import load_dotenv
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators import OpenAIGenerator
from haystack.components.tools import Tool
from haystack.components.websearch import SerperDevWebSearch
from haystack.dataclasses import ChatMessage
from haystack import component, logging
load_dotenv()
logging.basicConfig(level=logging.INFO)
@component
class QueryLogger:
@component.output_types(query=str)
def run(self, query: str):
logging.info(f"Incoming query: {query}")
return {"query": query}
@component
class ResponseLogger:
@component.output_types(answer=str)
def run(self, messages: list):
# Extract last assistant message
for msg in reversed(messages):
if msg.role == "assistant" and msg.text:
logging.info(f"Final answer length: {len(msg.text)} chars")
return {"answer": msg.text}
return {"answer": ""}
llm = OpenAIGenerator(
model="gpt-4o-mini",
api_key=os.getenv("OPENAI_API_KEY"),
generation_kwargs={"temperature": 0.1}
)
search = SerperDevWebSearch(
api_key=os.getenv("SERPER_API_KEY"),
top_k=5
)
search_tool = Tool(
name="web_search",
description="Search the web for current information.",
component=search
)
agent = Agent(
tools=[search_tool],
generator=llm,
system_prompt="Answer using web search. Cite sources.",
max_steps=5
)
pipeline = Pipeline()
pipeline.add_component("log_query", QueryLogger())
pipeline.add_component("agent", agent)
pipeline.add_component("log_response", ResponseLogger())
pipeline.connect("log_query.query", "agent.messages") # Note: type mismatch handled by Agent
pipeline.connect("agent.messages", "log_response.messages")
# Run through pipeline
question = "Who won the 2024 Super Bowl?"
result = pipeline.run(data={"log_query": {"query": question}})
print(result["log_response"]["answer"])This pattern scales: add a TokenCounter component, a PIIRedactor, or route to different LLMs based on query classification — all without changing the agent logic.
Production checklist
Before deploying:
-
Rate limiting — SerperDev allows 2,500 free queries/month. Implement client-side throttling or switch to a paid tier. If you’re routing through a gateway like n4n.ai, you can enforce per-key quotas at the gateway level.
-
Token costs — Each agent step sends full conversation history.
max_steps=5with gpt-4o-mini typically costs $0.001–0.005 per query. Set a hard token budget in the generator’sgeneration_kwargs. -
Source freshness — SerperDev returns a
datefield in results. Filter out sources older than your threshold (e.g., 30 days for pricing, 1 year for technical specs). -
Evaluation — Build a small eval set (20–50 questions with expected answers). Run nightly to catch regressions when model versions or search indexes change.
-
Fallback — If search returns zero results, fall back to the LLM’s internal knowledge with a disclaimer: “No current sources found; this is based on training data up to 2024.”
# Fallback pattern
if not search_results:
fallback_prompt = (
"No current web results found. Answer from your training data "
"and prefix with '[Training data only]'."
)
# Re-run agent with modified system prompt or call generator directlyNext steps
- Replace
SerperDevWebSearchwithBraveWebSearchorDuckDuckGoWebSearchfor different indexes - Add a
Calculatortool for math-heavy queries - Implement
Toolsubclasses that enforce output schemas via Pydantic - Add conversation memory with
ChatMessageHistoryfor multi-turn research sessions
The agent pattern here — search → extract → synthesize → structure — handles 80% of research-style queries. The remaining 20% need domain-specific tools (SQL, API clients, code execution). Start simple, measure, then extend.