Agent & ReAct Pattern
Build AI systems that can reason about tasks, use tools, plan multi-step solutions, and collaborate with other agents - from simple ReAct loops to sophisticated multi-agent architectures.
What Is an AI Agent?
An AI agent is a system that uses an LLM as its reasoning engine to autonomously decide what actions to take, execute those actions using tools, observe the results, and iterate until the task is complete. Unlike simple prompt-in/response-out LLM calls, agents operate in a loop.
The Observe-Think-Act Loop
THE AGENT LOOP
==============
+---> OBSERVE <---+
| (read tool |
| results, |
| user input) |
| | |
| v |
| THINK |
| (LLM reasons |
| about next |
| step) |
| | |
| v |
+------ ACT |
(call tool, |
respond to |
user, or |
finish) |
| |
+--------+
(loop until done)
This loop is the fundamental building block of all agent patterns. The key innovation is that the LLM decides which action to take and when to stop - the control flow is dynamic, not hardcoded.
The ReAct Pattern (Reason + Act)
ReAct is the most influential agent pattern. Introduced in the 2022 paper "ReAct: Synergizing Reasoning and Acting in Language Models," it interleaves chain-of-thought reasoning with tool actions. The LLM explicitly writes out its reasoning before each action, making the agent's decision process transparent and debuggable.
ReAct Trace Example
User: What is the current stock price of NVIDIA and how does it
compare to its 52-week high?
Thought 1: I need to find NVIDIA's current stock price and its
52-week high. Let me search for the current price first.
Action 1: search_stock(symbol="NVDA")
Observation 1: NVDA current price: $892.45, open: $885.20,
volume: 42.3M
Thought 2: I have the current price ($892.45). Now I need the
52-week high to make the comparison.
Action 2: get_stock_stats(symbol="NVDA", stat="52_week_high")
Observation 2: NVDA 52-week high: $974.00 (reached 2025-01-06)
Thought 3: I now have both pieces of information.
Current price: $892.45, 52-week high: $974.00.
The current price is $81.55 below the high, which is about 8.4%.
I can now give a complete answer.
Action 3: respond(
"NVIDIA (NVDA) is currently trading at $892.45, which is
$81.55 (8.4%) below its 52-week high of $974.00 reached
on January 6, 2025."
)
Tool Use Pattern
Tools are functions that the agent can call to interact with the external world. Modern LLM APIs (Claude, GPT-4) support native tool/function calling, where the model returns structured JSON specifying which tool to call with what arguments.
ReAct Agent with Claude Tool Calling
import anthropic
import json
client = anthropic.Anthropic()
# Define tools the agent can use
tools = [
{
"name": "search_web",
"description": "Search the web for current information",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
}
},
"required": ["query"]
}
},
{
"name": "calculate",
"description": "Evaluate a mathematical expression",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Math expression to evaluate, e.g. '2 + 2 * 3'"
}
},
"required": ["expression"]
}
},
{
"name": "read_file",
"description": "Read the contents of a file",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to read"
}
},
"required": ["path"]
}
}
]
def execute_tool(name: str, args: dict) -> str:
"""Execute a tool and return the result as a string."""
if name == "search_web":
# In production, call a real search API
return f"Search results for '{args['query']}': ..."
elif name == "calculate":
try:
result = eval(args["expression"]) # Use a safe evaluator in prod
return str(result)
except Exception as e:
return f"Error: {e}"
elif name == "read_file":
try:
with open(args["path"]) as f:
return f.read()[:5000] # Limit size
except Exception as e:
return f"Error reading file: {e}"
return f"Unknown tool: {name}"
def react_agent(user_message: str, max_iterations: int = 10) -> str:
"""Run a ReAct agent loop with Claude's native tool calling."""
messages = [{"role": "user", "content": user_message}]
system = """You are a helpful assistant with access to tools.
Think step by step. Use tools when you need external information
or calculations. When you have enough information, provide your
final answer directly without using any tools."""
for iteration in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system=system,
tools=tools,
messages=messages,
)
# Check if the model wants to use a tool
if response.stop_reason == "tool_use":
# Process all tool calls in this response
tool_results = []
assistant_content = response.content
for block in response.content:
if block.type == "tool_use":
print(f" [Tool Call] {block.name}({json.dumps(block.input)})")
result = execute_tool(block.name, block.input)
print(f" [Result] {result[:200]}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# Add assistant response and tool results to conversation
messages.append({"role": "assistant", "content": assistant_content})
messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
# Agent is done - return final text response
for block in response.content:
if hasattr(block, "text"):
return block.text
return "Agent completed without a text response."
return "Agent reached maximum iterations without completing."
# Usage
result = react_agent("What is 15% of the GDP of France in 2024?")
print(result)
Planning Patterns
For complex tasks, agents need explicit planning before acting. Here are the major planning approaches:
Plan-and-Execute
The agent creates a complete plan first, then executes each step. Good for tasks with clear subtask boundaries.
# Plan-and-Execute: Create plan first, then execute steps
def plan_and_execute(task: str) -> str:
# Step 1: Create a plan
plan_response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""Create a step-by-step plan to accomplish this task.
Return a numbered list of concrete steps.
Task: {task}"""
}]
)
plan = plan_response.content[0].text
steps = [s.strip() for s in plan.strip().split("\n") if s.strip()]
print(f"Plan ({len(steps)} steps):\n{plan}\n")
# Step 2: Execute each step
results = []
for i, step in enumerate(steps):
print(f"Executing step {i+1}: {step}")
step_result = react_agent(
f"""You are executing step {i+1} of a plan.
Previous results: {json.dumps(results[-3:])}
Current step: {step}
Original task: {task}
Execute this step and return the result."""
)
results.append({"step": step, "result": step_result})
# Step 3: Synthesize final answer
synthesis = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""Synthesize a final answer from these step results.
Original task: {task}
Step results: {json.dumps(results, indent=2)}
Final answer:"""
}]
)
return synthesis.content[0].text
Tree of Thought
The agent explores multiple reasoning paths in parallel and evaluates which path is most promising before committing.
# Tree of Thought: Explore multiple reasoning paths
def tree_of_thought(problem: str, n_thoughts: int = 3) -> str:
# Step 1: Generate multiple initial approaches
thoughts = []
for i in range(n_thoughts):
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=300,
temperature=0.8, # Higher temp for diversity
messages=[{
"role": "user",
"content": f"""Propose approach #{i+1} to solve this problem.
Be creative and consider different angles.
Problem: {problem}
Approach:"""
}]
)
thoughts.append(response.content[0].text)
# Step 2: Evaluate each approach
evaluation = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""Evaluate these approaches to solving the problem.
Rate each 1-10 and explain why. Select the best one.
Problem: {problem}
Approaches:
{chr(10).join(f'Approach {i+1}: {t}' for i, t in enumerate(thoughts))}
Evaluation:"""
}]
)
# Step 3: Execute the best approach using an agent
return react_agent(
f"""Solve this problem using the best approach identified.
Problem: {problem}
Best approach analysis: {evaluation.content[0].text}"""
)
Multi-Agent Patterns
Complex systems often benefit from multiple specialized agents working together. Here are the three dominant multi-agent architectures:
Architecture Comparison
SUPERVISOR DEBATE SWARM
========== ====== =====
+----------+ +---------+ Agent A <--> Agent B
|Supervisor| | Agent 1 | ^ ^
+----+-----+ +----+----+ | |
| | v v
+----+----+ +----+----+ Agent C <--> Agent D
| | | | Agent 2 |
v v v +----+----+ (peer-to-peer,
Agent Agent Agent | self-organizing)
1 2 3 +----+----+
| Agent 3 |
(hierarchical, +---------+
coordinated)
(adversarial,
consensus)
Multi-Agent Debate System
# Multi-agent debate: Agents argue different positions, then reach consensus
import anthropic
client = anthropic.Anthropic()
def agent_debate(question: str, n_rounds: int = 3) -> str:
"""Two agents debate a question, then a judge synthesizes."""
agent_a_history = []
agent_b_history = []
# Initial positions
for agent_name, history, persona in [
("Agent A", agent_a_history, "You argue FOR the proposition. Be thorough and cite evidence."),
("Agent B", agent_b_history, "You argue AGAINST the proposition. Challenge assumptions and find weaknesses.")
]:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=400,
system=persona,
messages=[{
"role": "user",
"content": f"Present your initial position on: {question}"
}]
)
history.append(response.content[0].text)
print(f"\n{agent_name} (Round 1):\n{history[-1][:300]}...")
# Debate rounds
for round_num in range(2, n_rounds + 1):
# Agent A responds to Agent B's latest argument
response_a = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=400,
system="You argue FOR the proposition. Address the counterarguments.",
messages=[{
"role": "user",
"content": f"""Question: {question}
Your previous argument: {agent_a_history[-1]}
Opponent's counterargument: {agent_b_history[-1]}
Respond to their points and strengthen your position:"""
}]
)
agent_a_history.append(response_a.content[0].text)
# Agent B responds to Agent A's latest argument
response_b = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=400,
system="You argue AGAINST the proposition. Find flaws in the reasoning.",
messages=[{
"role": "user",
"content": f"""Question: {question}
Your previous argument: {agent_b_history[-1]}
Opponent's counterargument: {agent_a_history[-1]}
Respond to their points and strengthen your position:"""
}]
)
agent_b_history.append(response_b.content[0].text)
print(f"\n--- Round {round_num} ---")
print(f"Agent A: {agent_a_history[-1][:200]}...")
print(f"Agent B: {agent_b_history[-1][:200]}...")
# Judge synthesizes
judge_response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=800,
system="You are an impartial judge. Synthesize the strongest arguments from both sides into a balanced, nuanced conclusion.",
messages=[{
"role": "user",
"content": f"""Question: {question}
Agent A's arguments (FOR):
{chr(10).join(f'Round {i+1}: {a}' for i, a in enumerate(agent_a_history))}
Agent B's arguments (AGAINST):
{chr(10).join(f'Round {i+1}: {b}' for i, b in enumerate(agent_b_history))}
Provide a balanced synthesis and final verdict:"""
}]
)
return judge_response.content[0].text
# Usage
result = agent_debate(
"Should companies build their own LLMs or use API providers?",
n_rounds=3
)
print(f"\nFinal Verdict:\n{result}")
Agent Memory Systems
Agents need memory to maintain context across interactions and learn from past experiences. There are three types of agent memory:
| Memory Type | What It Stores | Duration | Implementation |
|---|---|---|---|
| Short-term (Working) | Current conversation, recent tool results | Single session | LLM context window, message history |
| Long-term (Semantic) | Facts, knowledge, user preferences | Persistent | Vector database, key-value store |
| Episodic | Past interactions, successful strategies | Persistent | Structured logs, summarized episodes |
# Agent memory implementation
import chromadb
from datetime import datetime
import json
class AgentMemory:
"""Three-tier memory system for AI agents."""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.chroma = chromadb.PersistentClient(path="./agent_memory")
# Short-term: current conversation
self.short_term = []
# Long-term: persistent facts and preferences
self.long_term = self.chroma.get_or_create_collection(
f"{agent_id}_long_term"
)
# Episodic: past interaction summaries
self.episodic = self.chroma.get_or_create_collection(
f"{agent_id}_episodic"
)
def add_to_short_term(self, role: str, content: str):
"""Add a message to the current conversation."""
self.short_term.append({"role": role, "content": content})
# Trim if too long (keep last 20 messages)
if len(self.short_term) > 20:
self.short_term = self.short_term[-20:]
def store_fact(self, fact: str, metadata: dict = None):
"""Store a long-term fact (user preference, learned info)."""
self.long_term.add(
documents=[fact],
ids=[f"fact_{datetime.now().isoformat()}"],
metadatas=[metadata or {}]
)
def recall_facts(self, query: str, n: int = 5) -> list[str]:
"""Retrieve relevant long-term memories."""
results = self.long_term.query(query_texts=[query], n_results=n)
return results["documents"][0] if results["documents"] else []
def save_episode(self, summary: str, outcome: str):
"""Save a summarized past interaction for future reference."""
self.episodic.add(
documents=[summary],
ids=[f"episode_{datetime.now().isoformat()}"],
metadatas=[{"outcome": outcome, "timestamp": datetime.now().isoformat()}]
)
def recall_episodes(self, situation: str, n: int = 3) -> list[str]:
"""Find similar past episodes for guidance."""
results = self.episodic.query(query_texts=[situation], n_results=n)
return results["documents"][0] if results["documents"] else []
def get_context_prompt(self, current_query: str) -> str:
"""Build a context string from all memory types."""
# Relevant long-term memories
facts = self.recall_facts(current_query, n=3)
facts_str = "\n".join(f"- {f}" for f in facts) if facts else "None"
# Similar past episodes
episodes = self.recall_episodes(current_query, n=2)
episodes_str = "\n".join(f"- {e}" for e in episodes) if episodes else "None"
return f"""AGENT MEMORY CONTEXT:
Relevant facts about this user/topic:
{facts_str}
Similar past interactions:
{episodes_str}
Current conversation:
{json.dumps(self.short_term[-5:], indent=2)}"""
When Agents Fail
Agents are powerful but fragile. Understanding failure modes is critical for building reliable agent systems.
Common agent failure modes:
- Infinite loops: The agent keeps calling the same tool or alternating between two tools without making progress. ALWAYS implement a maximum iteration limit.
- Tool misuse: The agent calls a tool with wrong arguments, gets an error, and retries with the same wrong arguments. Implement clear error messages and argument validation.
- Goal drift: The agent gets distracted by intermediate results and pursues a tangent instead of the original task. Include the original task in every iteration's prompt.
- Hallucinated tools: The agent tries to call a tool that does not exist. Only provide tools the agent actually has access to, and validate tool names before execution.
- Context window exhaustion: Long agent runs accumulate so many messages that the context window fills up, causing the agent to lose track of early information. Implement conversation summarization.
- Cost explosion: Each iteration costs an API call. A 50-iteration agent run with GPT-4 can cost $5-10 per request. Set cost budgets and iteration limits.
Agent Guardrails Implementation
# Production agent with guardrails
class SafeAgent:
def __init__(self, max_iterations=15, max_cost_usd=1.0):
self.max_iterations = max_iterations
self.max_cost_usd = max_cost_usd
self.total_tokens = 0
self.iteration_count = 0
self.tool_call_history = []
def detect_loop(self, tool_name: str, tool_args: dict) -> bool:
"""Detect if the agent is stuck in a loop."""
current_call = f"{tool_name}:{json.dumps(tool_args, sort_keys=True)}"
# Check if same exact call was made in last 3 iterations
recent = self.tool_call_history[-3:]
if recent.count(current_call) >= 2:
return True
self.tool_call_history.append(current_call)
return False
def estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
"""Estimate cost of this API call."""
# Claude Sonnet pricing (approximate)
cost = (input_tokens * 0.003 + output_tokens * 0.015) / 1000
return cost
def run(self, task: str) -> str:
total_cost = 0.0
for i in range(self.max_iterations):
self.iteration_count = i + 1
# ... make API call ...
# response = client.messages.create(...)
# Track costs
# cost = self.estimate_cost(response.usage.input_tokens,
# response.usage.output_tokens)
# total_cost += cost
# Check cost budget
if total_cost > self.max_cost_usd:
return f"Agent stopped: cost budget exceeded (${total_cost:.2f})"
# Check for loops before executing tool
# if self.detect_loop(tool_name, tool_args):
# return "Agent stopped: loop detected"
return f"Agent stopped: max iterations ({self.max_iterations}) reached"
Comparison of Agent Frameworks
Several frameworks implement agent patterns. Choose based on your needs:
| Framework | Best For | Agent Type | Learning Curve | Production Ready |
|---|---|---|---|---|
| LangChain / LangGraph | Flexible agent graphs, custom workflows | Single & Multi-agent | Medium | Yes |
| CrewAI | Role-based multi-agent teams | Multi-agent (supervisor) | Low | Growing |
| AutoGen | Conversational multi-agent systems | Multi-agent (debate/chat) | Medium | Growing |
| Claude Tool Use (native) | Simple, reliable single agents | Single agent (ReAct) | Low | Yes |
| OpenAI Assistants API | Managed agent infrastructure | Single agent | Low | Yes |
| Haystack | RAG-focused agents with pipelines | Single agent | Medium | Yes |
| Swarm (OpenAI) | Lightweight agent handoffs | Multi-agent (swarm) | Low | Experimental |
| DSPy | Optimized prompt pipelines | Pipeline (not agent) | High | Research |
When to Use Agents vs Simpler Patterns
Use Agents When
- Steps are not known in advance
- Task requires dynamic tool selection
- Multiple information sources needed
- Problem requires iterative refinement
- User request is open-ended
Use Prompt Chaining Instead When
- Steps are predictable and fixed
- No external tools needed
- Deterministic output required
- Cost control is critical
- Latency must be predictable
Use RAG Instead When
- Task is just question-answering over documents
- No actions need to be taken
- Single retrieval + generation is sufficient
- Low latency is required
- Cost per query must stay under $0.01
What's Next
In the next lesson, we explore the Prompt Chaining pattern - the simpler, more predictable alternative to agents for multi-step tasks. You will learn how to decompose complex prompts into sequential chains, add validation gates, implement map-reduce for documents, and handle errors gracefully in prompt pipelines.
Ready to Go Deeper?
Live instructor-led courses from our partners. Affiliate disclosure.
AI & ML Courses - 30% Off
Live instructor-led AI, machine learning, data science, and cloud courses for working professionals. Use code Limited30 at checkout.
EdurekaDataCamp - AI & Data Science
Hands-on Python, machine learning, and AI courses with interactive exercises and real projects.
DataCampedX - Top AI Courses
University-level AI courses from MIT, Harvard, Stanford. Earn certificates that employers recognize.
edX