Haystack ReAct agent tutorial: search and calculate

This haystack react agent tutorial walks through building a ReAct-style agent in Haystack 2.0 that can search the web and perform calculations. You’ll wire together a generator, tools, and the agent loop with minimal boilerplate, then see exactly what each step produces.

Prerequisites

• Python 3.10+
• An OpenAI API key (or any OpenAI-compatible endpoint)
• A SerpAPI key for web search (free tier works)
• pip install haystack-ai serpapi python-dotenv

Create a .env file in your project root:

OPENAI_API_KEY=sk-...
SERPAPI_API_KEY=...

Project structure

haystack-react-agent/
├── .env
├── main.py
└── tools.py

Define the tools

Haystack 2.0 treats tools as components. We’ll create two: a calculator and a web search wrapper around SerpAPI.

# tools.py
import os
import json
import requests
from haystack import component
from typing import Any

@component
class Calculator:
    """
    Evaluates a mathematical expression safely.
    Only supports basic arithmetic to avoid eval() risks.
    """
    @component.output_types(result=str)
    def run(self, expression: str) -> dict[str, str]:
        try:
            # Restrict to safe characters: digits, operators, parentheses, decimal, spaces
            allowed = set("0123456789+-*/(). ")
            if not set(expression).issubset(allowed):
                return {"result": "Error: invalid characters in expression"}
            result = eval(expression, {"__builtins__": {}}, {})
            return {"result": str(result)}
        except Exception as e:
            return {"result": f"Error: {e}"}

@component
class WebSearch:
    """
    Thin wrapper around SerpAPI's Google search endpoint.
    Returns the top 3 organic result snippets concatenated.
    """
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or os.getenv("SERPAPI_API_KEY")
        if not self.api_key:
            raise ValueError("SERPAPI_API_KEY not set")

    @component.output_types(results=str)
    def run(self, query: str) -> dict[str, str]:
        params = {
            "engine": "google",
            "q": query,
            "api_key": self.api_key,
            "num": 3,
        }
        resp = requests.get("https://serpapi.com/search", params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        snippets = []
        for item in data.get("organic_results", [])[:3]:
            snippet = item.get("snippet") or item.get("title", "")
            snippets.append(snippet)
        return {"results": "\n---\n".join(snippets) if snippets else "No results found"}

Expected output when you import these: no output — just clean imports.

Build the agent pipeline

Haystack 2.0’s Agent component orchestrates the ReAct loop: it calls the generator, parses tool calls, executes tools, and feeds results back until the generator produces a final answer.

# main.py
import os
from dotenv import load_dotenv
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from tools import Calculator, WebSearch

load_dotenv()

# 1. Generator — the LLM that reasons and decides tool calls
generator = OpenAIGenerator(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    generation_kwargs={"temperature": 0},
)

# 2. Tools
calculator = Calculator()
search = WebSearch()

# 3. Agent with ReAct prompt template
system_prompt = """You are a helpful assistant that can use tools to answer questions.
Available tools:
- calculator: evaluates math expressions. Args: {"expression": "string"}
- web_search: searches Google via SerpAPI. Args: {"query": "string"}

Use the following format:
Thought: your reasoning
Action: tool_name
Action Input: {"arg": "value"}
Observation: tool result
... (repeat as needed)
Final Answer: your final response"""

agent = Agent(
    generator=generator,
    tools=[calculator, search],
    system_prompt=system_prompt,
    max_steps=6,  # prevent runaway loops
)

# 4. Pipeline (optional but useful for observability / serialization)
pipeline = Pipeline()
pipeline.add_component("agent", agent)

# 5. Run a query
question = "What is the current population of Tokyo multiplied by 2?"
messages = [ChatMessage.from_user(question)]

result = pipeline.run({"agent": {"messages": messages}})
final_message = result["agent"]["messages"][-1]
print(final_message.text)

Run it:

python main.py

Expected output (values will differ by search timing):

Thought: The user wants Tokyo's current population multiplied by 2. I need to search for the latest population figure, then calculate.
Action: web_search
Action Input: {"query": "Tokyo population 2024"}
Observation: Tokyo's population is estimated at 37.1 million in the greater metro area as of 2024.
---
Tokyo 2024 population estimates range from 37.1 to 37.8 million depending on metro definition.
---
Latest UN data puts Tokyo metro at 37.1 million.
Thought: I have the population figure. Now I'll multiply 37.1 million by 2.
Action: calculator
Action Input: {"expression": "37100000 * 2"}
Observation: 74200000
Final Answer: The current population of Tokyo (greater metro area) is approximately 37.1 million. Multiplied by 2, that equals 74.2 million.

Inspect the intermediate steps

The agent’s message history captures every ReAct iteration. This is invaluable for debugging.

# Add after pipeline.run()
for i, msg in enumerate(result["agent"]["messages"]):
    role = msg.role.value
    content = msg.text[:200] + ("..." if len(msg.text) > 200 else "")
    print(f"[{i}] {role}: {content}")

Expected output:

[0] user: What is the current population of Tokyo multiplied by 2?
[1] assistant: Thought: The user wants Tokyo's current population multiplied by 2...
Action: web_search
Action Input: {"query": "Tokyo population 2024"}
[2] tool: Observation: Tokyo's population is estimated at 37.1 million...
[3] assistant: Thought: I have the population figure. Now I'll multiply...
Action: calculator
Action Input: {"expression": "37100000 * 2"}
[4] tool: Observation: 74200000
[5] assistant: Final Answer: The current population of Tokyo...

Handling tool failures gracefully

Tools can fail — network errors, rate limits, bad input. The agent should see the error and decide whether to retry, use another tool, or answer with what it has.

Update Calculator.run to raise on truly invalid input instead of returning an error string, then catch it at the pipeline level:

# tools.py — revised Calculator.run
@component.output_types(result=str)
def run(self, expression: str) -> dict[str, str]:
    allowed = set("0123456789+-*/(). ")
    if not set(expression).issubset(allowed):
        raise ValueError(f"Invalid characters in expression: {expression}")
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return {"result": str(result)}
    except ZeroDivisionError:
        raise ValueError("Division by zero")
    except Exception as e:
        raise ValueError(f"Evaluation error: {e}")

Now wrap the pipeline run in a try/except and feed the error back as a tool observation:

# main.py — revised run block
from haystack.dataclasses import ChatMessage
from haystack.components.agents import AgentError

try:
    result = pipeline.run({"agent": {"messages": messages}})
except AgentError as e:
    # AgentError wraps tool exceptions; extract and feed back
    error_msg = str(e)
    messages.append(ChatMessage.from_assistant(f"Tool error: {error_msg}"))
    # Optionally retry with a corrected action
    result = pipeline.run({"agent": {"messages": messages}})

final_message = result["agent"]["messages"][-1]
print(final_message.text)

This pattern lets the agent self-correct — for example, if it passes "37.1 million * 2" to the calculator, the tool raises, the agent sees the error, and retries with "37100000 * 2".

Adding a custom tool: unit conversion

Real workloads need domain-specific tools. Here’s a unit converter that the agent can invoke like any other tool.

# tools.py — add this class
@component
class UnitConverter:
    """
    Converts between common units. Extend the map as needed.
    """
    CONVERSIONS = {
        ("celsius", "fahrenheit"): lambda c: c * 9/5 + 32,
        ("fahrenheit", "celsius"): lambda f: (f - 32) * 5/9,
        ("km", "miles"): lambda km: km * 0.621371,
        ("miles", "km"): lambda mi: mi / 0.621371,
        ("kg", "lbs"): lambda kg: kg * 2.20462,
        ("lbs", "kg"): lambda lbs: lbs / 2.20462,
    }

    @component.output_types(result=str)
    def run(self, value: float, from_unit: str, to_unit: str) -> dict[str, str]:
        key = (from_unit.lower(), to_unit.lower())
        if key not in self.CONVERSIONS:
            return {"result": f"Error: unsupported conversion {from_unit} -> {to_unit}"}
        try:
            converted = self.CONVERSIONS[key](value)
            return {"result": f"{value} {from_unit} = {converted:.4f} {to_unit}"}
        except Exception as e:
            return {"result": f"Error: {e}"}

Register it in main.py:

from tools import Calculator, WebSearch, UnitConverter

converter = UnitConverter()
agent = Agent(
    generator=generator,
    tools=[calculator, search, converter],
    system_prompt=system_prompt + """
- unit_converter: converts units. Args: {"value": number, "from_unit": "string", "to_unit": "string"}""",
    max_steps=6,
)

Test it:

question = "Convert 100 celsius to fahrenheit, then add 50."
messages = [ChatMessage.from_user(question)]
result = pipeline.run({"agent": {"messages": messages}})
print(result["agent"]["messages"][-1].text)

Expected output:

Thought: Convert 100 celsius to fahrenheit, then add 50.
Action: unit_converter
Action Input: {"value": 100, "from_unit": "celsius", "to_unit": "fahrenheit"}
Observation: 100 celsius = 212.0000 fahrenheit
Thought: Now add 50 to 212.
Action: calculator
Action Input: {"expression": "212 + 50"}
Observation: 262
Final Answer: 100 celsius = 212 fahrenheit. Adding 50 gives 262.

Streaming the final answer

For UIs, you want tokens as they arrive. Haystack 2.0 supports streaming via the generator’s streaming_callback.

# main.py — streaming variant
def print_token(token: str):
    print(token, end="", flush=True)

generator = OpenAIGenerator(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    generation_kwargs={"temperature": 0},
    streaming_callback=print_token,
)

# ... same pipeline setup ...

result = pipeline.run({"agent": {"messages": messages}})
print()  # newline after stream completes

Expected output: tokens print incrementally during the final answer phase. Tool calls and observations still appear as complete blocks (they’re not streamed).

Serializing and deploying

The pipeline is JSON-serializable — useful for version control, CI/CD, or deploying to a runtime that loads pipelines from config.

# Save
pipeline.dump("agent_pipeline.yaml")

# Load elsewhere
from haystack import Pipeline
loaded = Pipeline.load("agent_pipeline.yaml")
result = loaded.run({"agent": {"messages": [ChatMessage.from_user("2 + 2")]}})

agent_pipeline.yaml (abridged):

components:
  agent:
    type: haystack.components.agents.Agent
    init_parameters:
      generator:
        type: haystack.components.generators.OpenAIGenerator
        init_parameters:
          model: gpt-4o-mini
          generation_kwargs:
            temperature: 0
      tools:
        - type: tools.Calculator
        - type: tools.WebSearch
        - type: tools.UnitConverter
      system_prompt: "You are a helpful assistant..."
      max_steps: 6

Common pitfalls

Extending further

• Memory: Add a ChatMemory component to persist conversation history across runs.
• Structured output: Use OpenAIGenerator with response_format={"type": "json_object"} and parse the final answer into a Pydantic model.
• Observability: Hook Haystack’s tracing callbacks to log each step to Langfuse, LangSmith, or your own stack.
• Multi-agent: Compose multiple Agent components in a pipeline, each with specialized toolsets, and route via a classifier.

If you’re running this at scale behind a gateway that handles model routing, fallback, and per-token metering — n4n.ai does exactly that with a single OpenAI-compatible endpoint — you can swap OpenAIGenerator for any model the gateway exposes without changing the agent code.

Full file reference

tools.py

import os
import json
import requests
from haystack import component

@component
class Calculator:
    @component.output_types(result=str)
    def run(self, expression: str) -> dict[str, str]:
        allowed = set("0123456789+-*/(). ")
        if not set(expression).issubset(allowed):
            raise ValueError(f"Invalid characters in expression: {expression}")
        try:
            result = eval(expression, {"__builtins__": {}}, {})
            return {"result": str(result)}
        except ZeroDivisionError:
            raise ValueError("Division by zero")
        except Exception as e:
            raise ValueError(f"Evaluation error: {e}")

@component
class WebSearch:
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or os.getenv("SERPAPI_API_KEY")
        if not self.api_key:
            raise ValueError("SERPAPI_API_KEY not set")

    @component.output_types(results=str)
    def run(self, query: str) -> dict[str, str]:
        params = {"engine": "google", "q": query, "api_key": self.api_key, "num": 3}
        resp = requests.get("https://serpapi.com/search", params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        snippets = [item.get("snippet") or item.get("title", "") for item in data.get("organic_results", [])[:3]]
        return {"results": "\n---\n".join(snippets) if snippets else "No results found"}

@component
class UnitConverter:
    CONVERSIONS = {
        ("celsius", "fahrenheit"): lambda c: c * 9/5 + 32,
        ("fahrenheit", "celsius"): lambda f: (f - 32) * 5/9,
        ("km", "miles"): lambda km: km * 0.621371,
        ("miles", "km"): lambda mi: mi / 0.621371,
        ("kg", "lbs"): lambda kg: kg * 2.20462,
        ("lbs", "kg"): lambda lbs: lbs / 2.20462,
    }

    @component.output_types(result=str)
    def run(self, value: float, from_unit: str, to_unit: str) -> dict[str, str]:
        key = (from_unit.lower(), to_unit.lower())
        if key not in self.CONVERSIONS:
            return {"result": f"Error: unsupported conversion {from_unit} -> {to_unit}"}
        try:
            converted = self.CONVERSIONS[key](value)
            return {"result": f"{value} {from_unit} = {converted:.4f} {to_unit}"}
        except Exception as e:
            return {"result": f"Error: {e}"}

main.py

import os
from dotenv import load_dotenv
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from tools import Calculator, WebSearch, UnitConverter

load_dotenv()

generator = OpenAIGenerator(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    generation_kwargs={"temperature": 0},
)

calculator = Calculator()
search = WebSearch()
converter = UnitConverter()

system_prompt = """You are a helpful assistant that can use tools to answer questions.
Available tools:
- calculator: evaluates math expressions. Args: {"expression": "string"}
- web_search: searches Google via SerpAPI. Args: {"query": "string"}
- unit_converter: converts units. Args: {"value": number, "from_unit": "string", "to_unit": "string"}

Use the following format:
Thought: your reasoning
Action: tool_name
Action Input: {"arg": "value"}
Observation: tool result
... (repeat as needed)
Final Answer: your final response"""

agent = Agent(
    generator=generator,
    tools=[calculator, search, converter],
    system_prompt=system_prompt,
    max_steps=6,
)

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

# Example queries
queries = [
    "What is the current population of Tokyo multiplied by 2?",
    "Convert 100 celsius to fahrenheit, then add 50.",
    "Search for the boiling point of water at sea level in celsius, convert to fahrenheit.",
]

for q in queries:
    print(f"\n=== Query: {q} ===\n")
    messages = [ChatMessage.from_user(q)]
    result = pipeline.run({"agent": {"messages": messages}})
    print(result["agent"]["messages"][-1].text)

Run the full suite:

python main.py

You now have a working Haystack 2.0 ReAct agent that searches, calculates, converts units, handles errors, streams, and serializes. The pattern scales: add tools, tune the system prompt, and deploy the same pipeline definition anywhere Haystack runs.