Dspy React

Build Tool-Using Agents with dspy.ReAct

Guide the user through building agents that reason step-by-step and call tools to accomplish tasks. dspy.ReAct implements the Reasoning-Action-Observation loop -- the agent thinks about what to do, calls a tool, observes the result, and repeats until it has an answer.

What is ReAct

dspy.ReAct implements the Reasoning-Action-Observation loop as an optimizable module. The agent reasons about what to do, calls a tool, observes the result, and repeats until it has enough information to answer. DSPy handles the loop mechanics and prompt construction.

When to use ReAct

Defining tools

Tools are Python functions with type hints and docstrings. DSPy uses the function signature and docstring to tell the agent what each tool does and how to call it.

def search(query: str) -> str:
    """Search the web for information about a topic."""
    # Your search implementation here
    return "search results..."

def calculate(expression: str) -> float:
    """Evaluate a mathematical expression and return the result."""
    return eval(expression)  # use a safe evaluator in production

def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    # Your weather API call here
    return f"72°F and sunny in {city}"

Tool requirements:

• Type hints on all parameters and the return type -- DSPy uses these to generate the tool schema
• Docstring explaining what the tool does -- the agent reads this to decide when to use it
• Return a string (or something that converts to string) -- the result becomes the Observation

Keep tools focused on one thing. A search tool should search, not search-and-summarize.

Basic ReAct agent

import dspy

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

def search(query: str) -> str:
    """Search for information about a topic."""
    return "DSPy is a framework for programming language models."

agent = dspy.ReAct("question -> answer", tools=[search])
result = agent(question="What is DSPy?")
print(result.answer)

That's it. The agent will: 1. Read the question 2. Decide whether to call search 3. Use the search result to formulate an answer

Constructor parameters

dspy.ReAct(
    signature,      # str | Signature -- required, defines inputs/outputs
    tools,          # list[Callable | dspy.Tool] -- required, available tools
    max_iters=20,   # int -- max reasoning-action cycles
)

The max_iters parameter

max_iters controls how many Thought-Action-Observation cycles the agent can take before it must produce an answer:

# Simple lookup -- 1-2 tool calls usually enough
agent = dspy.ReAct("question -> answer", tools=[search], max_iters=3)

# Complex research -- may need many tool calls
agent = dspy.ReAct("question -> answer", tools=[search, lookup], max_iters=8)

Guidelines:

• Default is 20 -- usually fine for most tasks
• Set it lower (2-5) for simple lookups where one or two tool calls suffice
• Keep it at 20 for complex multi-step research tasks
• If the agent hits max_iters without finishing, it returns its best answer so far

Multi-tool agents

Give the agent multiple tools and it decides which to use and when:

import dspy

def search(query: str) -> str:
    """Search the web for general information."""
    return "search results..."

def lookup_user(email: str) -> str:
    """Look up a user account by email address."""
    return '{"name": "Alice", "plan": "pro", "status": "active"}'

def check_order(order_id: str) -> str:
    """Check the status of an order by its ID."""
    return '{"order_id": "12345", "status": "shipped", "eta": "March 20"}'

agent = dspy.ReAct(
    "question -> answer",
    tools=[search, lookup_user, check_order],
    max_iters=5,
)

# The agent picks the right tool based on the question
result = agent(question="What's the status of order 12345?")
print(result.answer)  # Uses check_order

result = agent(question="What plan is alice@example.com on?")
print(result.answer)  # Uses lookup_user

The agent can also chain tools -- call lookup_user first, then use the result to call check_order.

Wrapping ReAct in a custom module

For production use, wrap dspy.ReAct inside a dspy.Module to add pre-processing, context, or post-processing:

class SupportAgent(dspy.Module):
    def __init__(self):
        self.agent = dspy.ReAct(
            "question, context -> answer",
            tools=[search, lookup_user, check_order],
            max_iters=6,
        )

    def forward(self, question):
        context = (
            "You are a customer support agent. "
            "Use lookup_user for account questions, "
            "check_order for order questions, "
            "and search for general questions."
        )
        return self.agent(question=question, context=context)

def support_reward(args, pred):
    if len(pred.answer) > 20:
        return 1.0
    return 0.0  # Response too short — not detailed enough

validated_support = dspy.Refine(
    module=SupportAgent(),
    N=3,
    reward_fn=support_reward,
    threshold=1.0,
)

This pattern lets you:

• Pass extra context or instructions to the agent
• Add reward-based quality constraints with dspy.Refine
• Optimize the agent with DSPy optimizers (they tune the inner ReAct module)
• Save and load the optimized state

Using class-based signatures

For agents with typed inputs and outputs, use a class-based signature:

from typing import Literal

class ResearchTask(dspy.Signature):
    """Research a topic and provide a comprehensive answer with sources."""
    question: str = dspy.InputField(desc="The research question")
    answer: str = dspy.OutputField(desc="A thorough answer to the question")
    confidence: Literal["high", "medium", "low"] = dspy.OutputField(
        desc="Confidence level based on the sources found"
    )

agent = dspy.ReAct(ResearchTask, tools=[search], max_iters=5)
result = agent(question="What are the main features of DSPy?")
print(result.answer)
print(result.confidence)

ReAct vs CodeAct

Both are agent modules, but they act differently:

# ReAct -- calls tools
agent = dspy.ReAct("question -> answer", tools=[search, calculate])

# CodeAct -- writes code
agent = dspy.CodeAct("question -> answer", tools=[search, calculate])

If you're unsure, start with ReAct. Switch to CodeAct if the agent needs to do math, string manipulation, or data transformations between tool calls.

Error handling

Tools can fail. Handle errors inside your tools so the agent gets a useful message instead of a crash:

import requests

def search(query: str) -> str:
    """Search the web for information."""
    try:
        response = requests.get(
            "https://api.example.com/search",
            params={"q": query},
            timeout=5,
        )
        response.raise_for_status()
        return response.json()["results"]
    except requests.Timeout:
        return "Error: Search timed out. Try a simpler query."
    except requests.HTTPError as e:
        return f"Error: Search failed with status {e.response.status_code}."
    except Exception as e:
        return f"Error: {str(e)}"

When a tool returns an error string, the agent sees it as an Observation and can decide to retry with different arguments, try a different tool, or give a partial answer.

For module-level error handling, wrap the agent call:

class SafeAgent(dspy.Module):
    def __init__(self):
        self.agent = dspy.ReAct("question -> answer", tools=[search], max_iters=5)
        self.fallback = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        try:
            return self.agent(question=question)
        except Exception:
            # Fall back to answering without tools
            return self.fallback(question=question)

Optimizing ReAct agents

ReAct agents are optimizable like any DSPy module. The optimizer tunes the reasoning prompts so the agent makes better tool-calling decisions:

def answer_metric(example, prediction, trace=None):
    return prediction.answer.strip().lower() == example.answer.strip().lower()

# BootstrapFewShot for quick optimization
optimizer = dspy.BootstrapFewShot(metric=answer_metric, max_bootstrapped_demos=4)
optimized_agent = optimizer.compile(agent, trainset=trainset)

# MIPROv2 for better prompt optimization
optimizer = dspy.MIPROv2(metric=answer_metric, auto="medium")
optimized_agent = optimizer.compile(agent, trainset=trainset)

# Save and load
optimized_agent.save("optimized_agent.json")

Debugging

Inspect what the agent is doing:

# See the last few LM calls (thoughts, tool calls, observations)
dspy.inspect_history(n=5)

# Print the module structure
print(agent)

inspect_history shows you the full Thought-Action-Observation trace, which is invaluable for understanding why the agent called certain tools or gave a wrong answer.

Gotchas

1. Claude sets `max_iters=5` but the default is 20. Claude habitually passes max_iters=5 which cuts off complex multi-step tasks too early. The actual default is 20. Only lower it when you want to constrain simple tasks (2-3 for lookups). For complex research, the default of 20 is appropriate. 2. Tool errors are passed back as observations -- make your error messages informative so the agent can recover (e.g., "No user found with that email" not just "Error"). 3. ReAct is slow by design -- each iteration is a separate LM call. Use CodeAct for computation-heavy tasks where the agent can do work in code between tool calls. 4. Tool function docstrings become part of the prompt -- write clear, concise docstrings. Verbose docstrings waste tokens every iteration. 5. Claude ignores the `trajectory` in the return value. ReAct returns a dspy.Prediction with a .trajectory dict containing the full Thought-Action-Observation trace. Access result.trajectory to log or debug the agent's reasoning path. Claude often discards this and only uses result.answer.

Additional resources

• dspy.ReAct API docs
• For API details, see reference.md
• For worked examples, see examples.md

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Defining tools in detail -- see /dspy-tools
• CodeAct for code-based agents -- see /dspy-codeact
• Building custom modules to wrap ReAct -- see /dspy-modules
• Action-taking AI from a problem-first perspective -- see /ai-taking-actions
• Multi-agent coordination -- see /ai-coordinating-agents
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

dspy-react -- Worked Examples

Example 1: Search agent with a single tool

A simple agent that answers questions by searching a knowledge base. Demonstrates the basic ReAct pattern with one tool.

import dspy


# --- Tool ---

KNOWLEDGE_BASE = {
    "dspy": "DSPy is a framework for programming language models with optimizable modules.",
    "react": "ReAct is an agent pattern that combines reasoning with tool use in a loop.",
    "signatures": "DSPy signatures declare input/output behavior as typed specs like 'question -> answer'.",
    "optimizers": "DSPy optimizers tune prompts or weights to improve program accuracy.",
}


def search_docs(query: str) -> str:
    """Search the documentation knowledge base for information about a topic."""
    query_lower = query.lower()
    results = []
    for key, value in KNOWLEDGE_BASE.items():
        if key in query_lower or any(word in value.lower() for word in query_lower.split()):
            results.append(value)
    if results:
        return " ".join(results)
    return "No results found for that query."


# --- Agent ---

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

agent = dspy.ReAct("question -> answer", tools=[search_docs], max_iters=3)

# Test it
result = agent(question="What is DSPy and how does it use signatures?")
print(result.answer)

# Inspect the reasoning trace
dspy.inspect_history(n=3)

Key points:

• One tool is enough for many use cases -- the agent decides when to call it
• max_iters=3 keeps the agent focused since a single search is usually sufficient
• The tool returns plain strings; the agent interprets them and formulates an answer
• inspect_history shows the full Thought-Action-Observation trace for debugging

Example 2: Multi-tool customer support agent

A support agent with tools to look up users, check orders, and search a FAQ. Demonstrates routing between multiple tools and wrapping ReAct in a custom module.

import dspy
import json


# --- Tools ---

USERS_DB = {
    "alice@example.com": {"name": "Alice", "plan": "pro", "joined": "2024-01-15"},
    "bob@example.com": {"name": "Bob", "plan": "free", "joined": "2024-06-01"},
}

ORDERS_DB = {
    "ORD-001": {"user": "alice@example.com", "status": "shipped", "eta": "March 20"},
    "ORD-002": {"user": "bob@example.com", "status": "processing", "eta": "March 25"},
}

FAQ = {
    "refund": "Refunds are processed within 5-7 business days after approval.",
    "upgrade": "You can upgrade your plan at any time from Settings > Billing.",
    "cancel": "To cancel, go to Settings > Billing > Cancel Subscription.",
}


def lookup_user(email: str) -> str:
    """Look up a user account by their email address. Returns account details."""
    user = USERS_DB.get(email)
    if user:
        return json.dumps(user)
    return f"No user found with email {email}."


def check_order(order_id: str) -> str:
    """Check the status of an order by its order ID (e.g., ORD-001)."""
    order = ORDERS_DB.get(order_id.upper())
    if order:
        return json.dumps(order)
    return f"No order found with ID {order_id}."


def search_faq(topic: str) -> str:
    """Search the FAQ for help articles about a topic."""
    topic_lower = topic.lower()
    matches = []
    for key, value in FAQ.items():
        if key in topic_lower or topic_lower in key:
            matches.append(f"{key}: {value}")
    if matches:
        return "\n".join(matches)
    return "No FAQ articles found for that topic."


# --- Agent module ---

class SupportAgent(dspy.Module):
    """Customer support agent that looks up accounts, orders, and FAQ articles."""

    def __init__(self):
        self.agent = dspy.ReAct(
            "question, context -> answer",
            tools=[lookup_user, check_order, search_faq],
            max_iters=5,
        )

    def forward(self, question: str):
        context = (
            "You are a helpful customer support agent. "
            "Use lookup_user for account questions (requires email), "
            "check_order for order status (requires order ID like ORD-001), "
            "and search_faq for general help topics. "
            "Be friendly and specific in your answers."
        )
        return self.agent(question=question, context=context)


def support_quality_reward(args, pred):
    """Reward detailed, helpful responses."""
    if len(pred.answer.strip()) <= 30:
        return 0.5
    return 1.0


support = dspy.Refine(module=SupportAgent(), N=3, reward_fn=support_quality_reward, threshold=1.0)


# --- Usage ---

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

# support is defined above as dspy.Refine(module=SupportAgent(), ...)

# Test with different question types
questions = [
    "What plan is alice@example.com on?",
    "Where is my order ORD-002?",
    "How do I get a refund?",
    "Can you check if bob@example.com has any orders?",  # requires chaining tools
]

for q in questions:
    result = support(question=q)
    print(f"\nQ: {q}")
    print(f"A: {result.answer}")


# --- Optimization ---

def support_metric(example, prediction, trace=None):
    """Check if the answer contains the expected key information."""
    answer = prediction.answer.lower()
    # Check that expected keywords appear in the answer
    return all(kw.lower() in answer for kw in example.expected_keywords)


trainset = [
    dspy.Example(
        question="What plan is alice@example.com on?",
        expected_keywords=["pro"],
    ).with_inputs("question"),
    dspy.Example(
        question="Where is my order ORD-001?",
        expected_keywords=["shipped", "march 20"],
    ).with_inputs("question"),
    dspy.Example(
        question="How do I cancel my subscription?",
        expected_keywords=["settings", "billing"],
    ).with_inputs("question"),
]

# optimizer = dspy.BootstrapFewShot(metric=support_metric, max_bootstrapped_demos=3)
# optimized_support = optimizer.compile(support, trainset=trainset)
# optimized_support.save("support_agent.json")

Key points:

• Three tools covering different domains -- the agent picks the right one based on the question
• The agent can chain tools: look up a user, then check their orders
• Context string guides the agent on when to use each tool
• dspy.Refine with a reward function enforces minimum response quality by retrying if the answer is too short
• The metric checks for expected keywords rather than exact match, which works well for open-ended agent responses

Example 3: Data lookup agent with API calls

An agent that fetches data from REST APIs to answer questions. Demonstrates real HTTP calls, error handling in tools, and structured output.

import dspy
import requests
from typing import Literal


# --- Tools ---

def get_github_repo(repo: str) -> str:
    """Get information about a GitHub repository. Pass the full name like 'stanfordnlp/dspy'."""
    try:
        response = requests.get(
            f"https://api.github.com/repos/{repo}",
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=10,
        )
        if response.status_code == 404:
            return f"Repository '{repo}' not found."
        response.raise_for_status()
        data = response.json()
        return (
            f"Name: {data['full_name']}\n"
            f"Description: {data['description']}\n"
            f"Stars: {data['stargazers_count']}\n"
            f"Language: {data['language']}\n"
            f"Open issues: {data['open_issues_count']}\n"
            f"Last updated: {data['updated_at']}"
        )
    except requests.Timeout:
        return "Error: GitHub API request timed out. Try again."
    except requests.RequestException as e:
        return f"Error fetching repository info: {str(e)}"


def get_github_issues(repo: str, state: str = "open") -> str:
    """Get recent issues for a GitHub repository. Pass repo as 'owner/name' and state as 'open' or 'closed'."""
    try:
        response = requests.get(
            f"https://api.github.com/repos/{repo}/issues",
            params={"state": state, "per_page": 5, "sort": "updated"},
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=10,
        )
        response.raise_for_status()
        issues = response.json()
        if not issues:
            return f"No {state} issues found for {repo}."
        lines = []
        for issue in issues:
            lines.append(f"#{issue['number']}: {issue['title']} ({issue['state']})")
        return "\n".join(lines)
    except requests.Timeout:
        return "Error: GitHub API request timed out. Try again."
    except requests.RequestException as e:
        return f"Error fetching issues: {str(e)}"


def search_pypi(package_name: str) -> str:
    """Search for a Python package on PyPI and return its details."""
    try:
        response = requests.get(
            f"https://pypi.org/pypi/{package_name}/json",
            timeout=10,
        )
        if response.status_code == 404:
            return f"Package '{package_name}' not found on PyPI."
        response.raise_for_status()
        data = response.json()
        info = data["info"]
        return (
            f"Name: {info['name']}\n"
            f"Version: {info['version']}\n"
            f"Summary: {info['summary']}\n"
            f"Author: {info['author']}\n"
            f"License: {info['license']}\n"
            f"Home page: {info['home_page'] or info.get('project_url', 'N/A')}"
        )
    except requests.Timeout:
        return "Error: PyPI request timed out. Try again."
    except requests.RequestException as e:
        return f"Error searching PyPI: {str(e)}"


# --- Signature ---

class ResearchAnswer(dspy.Signature):
    """Research a question about open-source software using available tools."""
    question: str = dspy.InputField(desc="A question about open-source packages or repositories")
    answer: str = dspy.OutputField(desc="A detailed answer with specific data from the tools")
    confidence: Literal["high", "medium", "low"] = dspy.OutputField(
        desc="Confidence based on whether the tools returned useful data"
    )


# --- Agent module ---

class OSSResearcher(dspy.Module):
    """Agent that researches open-source software using GitHub and PyPI APIs."""

    def __init__(self):
        self.agent = dspy.ReAct(
            ResearchAnswer,
            tools=[get_github_repo, get_github_issues, search_pypi],
            max_iters=6,
        )

    def forward(self, question: str):
        return self.agent(question=question)


def research_confidence_reward(args, pred):
    """Reward high-confidence answers backed by real tool data."""
    if pred.confidence == "low":
        return 0.5
    return 1.0


# --- Usage ---

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

researcher = dspy.Refine(
    module=OSSResearcher(),
    N=3,
    reward_fn=research_confidence_reward,
    threshold=1.0,
)

questions = [
    "How many stars does stanfordnlp/dspy have and what is the latest PyPI version?",
    "What are the most recent open issues in the stanfordnlp/dspy repo?",
    "Compare the dspy and langchain packages on PyPI.",
]

for q in questions:
    print(f"\nQ: {q}")
    result = researcher(question=q)
    print(f"A: {result.answer}")
    print(f"Confidence: {result.confidence}")


# --- Optimization ---

def research_metric(example, prediction, trace=None):
    """Score based on confidence and whether key facts are in the answer."""
    has_data = prediction.confidence in ("high", "medium")
    has_answer = len(prediction.answer.strip()) > 50
    return has_data + 0.5 * has_answer

# trainset = [
#     dspy.Example(question="How many stars does stanfordnlp/dspy have?").with_inputs("question"),
#     dspy.Example(question="What version is dspy on PyPI?").with_inputs("question"),
# ]
# optimizer = dspy.MIPROv2(metric=research_metric, auto="light")
# optimized = optimizer.compile(researcher, trainset=trainset)
# optimized.save("oss_researcher.json")

Key points:

• Real API calls with proper error handling -- every tool catches timeouts and HTTP errors, returning a useful message instead of crashing
• Class-based signature with typed output (confidence: Literal[...]) gives structured results
• The agent chains tools naturally: fetch repo info from GitHub, then check PyPI for the package version
• dspy.Refine retries when confidence is low, nudging the agent to gather more data before answering
• MIPROv2 is a good optimizer choice for agents because it tunes the reasoning instructions

ReAct API Reference

Condensed from dspy.ai/api/modules/ReAct. Verify against upstream for latest.

Constructor

dspy.ReAct(
    signature,      # str | type[Signature] -- required
    tools,          # list[Callable | dspy.Tool] -- required
    max_iters=20,   # int -- max reasoning-action cycles
)

A "finish" tool is added automatically -- the agent calls it to signal task completion.

Key Methods

• forward(**input_args) -> dspy.Prediction -- run the agent loop synchronously
• aforward(**input_args) -> dspy.Prediction -- async variant
• truncate_trajectory(trajectory) -- removes oldest tool calls when context exceeds limits; override for custom truncation logic

Inherited Module Methods

Return Value

dspy.Prediction with:

• Output fields matching your signature (e.g., .answer)
• .trajectory -- dict mapping the full Thought-Action-Observation trace

Internal Behavior

1. Constructs an internal react_signature extending your signature with trajectory tracking and tool selection fields 2. Each iteration: agent produces a Thought, selects a tool (or "finish"), executes the tool, records the Observation 3. When "finish" is called, a fallback ChainOfThought extraction step produces the final output fields 4. All iterations run at the configured LM temperature