Module 5: ReAct, Self-Consistency, and Advanced Patterns
2. ReAct: Reasoning + Acting
Overview
ReAct (Reasoning + Acting) is a technique proposed by Yao et al. in 2022 that combines verbal reasoning (Thought) with action on external environments (Action). The core idea is simple but powerful: language models shouldn't only "think" — they should also be able to act on the world and observe the results to keep reasoning.
In a practical implementation with OpenAI, this is done through function calling: the model decides which tool to call, with which parameters, and then receives that tool's result as additional context.
The Problem ReAct Solves
Consider this scenario:
Question: "If I invest $10,000 in Tesla today, and the price goes up
8% next quarter, how much will I have?"
Pure CoT:
Thought: I need Tesla's current price
Thought: I don't know the current price → I'll assume $200 (made up/outdated)
Thought: 10000 / 200 = 50 shares
Thought: 50 * 200 * 1.08 = $10,800
Answer: $10,800
PROBLEM: The assumed price can be completely wrong.
ReAct:
Thought: I need to look up Tesla's current price
Action: search("Tesla stock price today")
Observation: Tesla (TSLA): $348.50 (March 8, 2026)
Thought: I have the real price. 10000 / 348.50 ≈ 28.7 shares
Action: calculator("10000 / 348.50 * 348.50 * 1.08")
Observation: 10800.0
Answer: With $10,000 you'd have roughly $10,800 after the 8% rise.
The Full ReAct Pattern
The ReAct loop is made of:
Thought: [Reasoning about what to do next]
Action: [tool_name(parameters)]
Observation: [Tool result]
... (repeat until you have an answer)
Answer: [Final answer]
Rules of the pattern
- Thought always precedes Action: The model has to reason before it acts
- One action at a time: Don't run multiple actions without processing the observations
- Observation informs the next Thought: The tool's result has to influence the reasoning
- Stop when you have enough information: Don't call tools unnecessarily
Full Implementation with OpenAI
from openai import OpenAI
import json
import math
client = OpenAI()
# --- Tool definitions ---
def calculator(expression: str) -> str:
"""Safely evaluates mathematical expressions."""
# In production: use a safe evaluation library
# Here we only allow basic math operations
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return "Error: Expression not allowed"
try:
result = eval(expression, {"__builtins__": {}}, {"math": math})
return str(round(float(result), 4))
except Exception as e:
return f"Error: {e}"
def search_info(query: str) -> str:
"""Simulates an information search."""
# In production: call a real API (Google, Bing, etc.)
database = {
"tesla stock price": "Tesla (TSLA): $348.50 (March 8, 2026)",
"apple stock": "Apple (AAPL): $218.30 (March 8, 2026)",
"gdp spain 2024": "Spain GDP 2024: ~1.47 trillion EUR (Bank of Spain)",
"population mexico": "Mexico population 2024: ~129.9 million (INEGI)",
"exchange rate eur usd": "EUR/USD: 1.0847 (March 8, 2026)",
}
query_lower = query.lower()
for key, value in database.items():
if any(word in query_lower for word in key.split()):
return value
return f"No specific information found about: {query}"
def convert_units(value: float, from_unit: str, to_unit: str) -> str:
"""Converts between common units."""
conversions = {
("km", "miles"): 0.621371,
("miles", "km"): 1.60934,
("kg", "lb"): 2.20462,
("lb", "kg"): 0.453592,
("celsius", "fahrenheit"): lambda x: x * 9/5 + 32,
("fahrenheit", "celsius"): lambda x: (x - 32) * 5/9,
}
key = (from_unit.lower(), to_unit.lower())
if key in conversions:
conv = conversions[key]
result = conv(value) if callable(conv) else value * conv
return f"{value} {from_unit} = {round(result, 4)} {to_unit}"
return f"Conversion {from_unit} → {to_unit} not available"
# --- Tool schema definition for OpenAI ---
TOOLS = [
{
"type": "function",
"function": {
"name": "calculator",
"description": "Evaluates mathematical expressions. Use it whenever you need to compute numbers, percentages, arithmetic operations.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate. Example: '100 * 1.08' or '(500 - 200) / 300'"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "search_info",
"description": "Searches for current factual information about prices, economic data, statistics. Use it when you need real-world data.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Be specific. Example: 'Tesla stock price today' or 'Spain GDP 2024'"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "convert_units",
"description": "Converts between units of measurement: distance, weight, temperature.",
"parameters": {
"type": "object",
"properties": {
"value": {"type": "number", "description": "The value to convert"},
"from_unit": {"type": "string", "description": "Source unit (km, miles, kg, lb, celsius, fahrenheit)"},
"to_unit": {"type": "string", "description": "Target unit"}
},
"required": ["value", "from_unit", "to_unit"]
}
}
}
]
# --- Tool dispatcher ---
TOOL_FUNCTIONS = {
"calculator": lambda args: calculator(args["expression"]),
"search_info": lambda args: search_info(args["query"]),
"convert_units": lambda args: convert_units(
args["value"], args["from_unit"], args["to_unit"]
)
}
# --- Main ReAct engine ---
def run_react(
problem: str,
max_steps: int = 8,
verbose: bool = True
) -> dict:
"""
Runs the ReAct loop to solve a problem.
Args:
problem: The question or problem to solve
max_steps: Maximum number of Thought→Action→Observation cycles
verbose: If True, prints the process step by step
Returns:
dict with 'answer', 'steps', 'tools_used'
"""
system_prompt = """You are an assistant that solves problems using tools.
For each problem:
1. Think (Thought) about what information you need or what to compute
2. Use the available tools (Action) when you need data or calculations
3. Observe the result (you'll receive it as an Observation)
4. Repeat until you have enough information
5. Give a clear and complete final answer
IMPORTANT:
- Use the calculator tool for ALL numeric calculations
- Use search_info when you need current data or specific facts
- Don't make up data you could look up with the tools
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": problem}
]
steps = []
tools_used = []
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=TOOLS,
tool_choice="auto",
temperature=0
)
msg = response.choices[0].message
# Record the thought
if msg.content:
steps.append({"type": "thought", "content": msg.content})
if verbose:
print(f"\n[Thought {step+1}]: {msg.content}")
# If there are no tool calls, the model has the answer
if not msg.tool_calls:
answer = msg.content or "No answer"
if verbose:
print(f"\n[Answer]: {answer}")
return {
"answer": answer,
"steps": steps,
"tools_used": tools_used,
"steps_used": step + 1
}
# Process tool calls
messages.append({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]
})
for tc in msg.tool_calls:
tool_name = tc.function.name
args = json.loads(tc.function.arguments)
if verbose:
print(f"\n[Action]: {tool_name}({args})")
# Run the tool
if tool_name in TOOL_FUNCTIONS:
result = TOOL_FUNCTIONS[tool_name](args)
else:
result = f"Error: Tool '{tool_name}' not available"
tools_used.append(tool_name)
steps.append({
"type": "action",
"tool": tool_name,
"args": args,
"result": result
})
if verbose:
print(f"[Observation]: {result}")
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
# If we hit the step limit
return {
"answer": "Step limit reached. Last result: " + str(steps[-1] if steps else "none"),
"steps": steps,
"tools_used": tools_used,
"steps_used": max_steps
}
# --- Usage examples ---
if __name__ == "__main__":
print("=" * 60)
print("Example 1: Calculation with real data")
print("=" * 60)
result = run_react(
"If I invest $5,000 in Apple shares today and the price rises 12%, "
"how much will I have? How many shares would I buy?",
verbose=True
)
print(f"\nFinal answer: {result['answer']}")
print(f"Tools used: {result['tools_used']}")
print("\n" + "=" * 60)
print("Example 2: Conversion with a calculation")
print("=" * 60)
result2 = run_react(
"A runner runs a marathon (42.195 km). "
"How many miles is that? And how many meters?",
verbose=True
)
print(f"\nFinal answer: {result2['answer']}")
ReAct vs Pure CoT: Detailed Comparison
| Aspect | Pure CoT | ReAct |
|---|---|---|
| External data | No (uses the model's knowledge) | Yes (APIs, databases) |
| Verification | Mental only | With real tools |
| Freshness | Limited by training cutoff | Real time (if you have the tools) |
| Number of API calls | 1 | 1 + N (one per action) |
| Cost | 1x | 3-10x depending on steps |
| Latency | Low | Medium-High |
| Math accuracy | Medium | High (with a real calculator) |
| When to prefer it | Pure math/logic, no external data | Whenever you need real data |
Advanced ReAct Patterns
Pattern 1: ReAct with Validation
def react_with_validation(problem: str) -> dict:
"""ReAct that validates the result before returning it."""
result = run_react(problem, verbose=False)
# Extra validation step
validation_prompt = f"""
Original problem: {problem}
Proposed answer: {result['answer']}
Steps taken: {json.dumps(result['steps'], indent=2)}
Is the answer correct and complete? If there are errors, fix them.
If it's correct, say "VALID: " followed by the final answer.
"""
validation = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": validation_prompt}],
temperature=0
)
validated_answer = validation.choices[0].message.content
return {
**result,
"validated_answer": validated_answer
}
Pattern 2: ReAct with Observation Cache
from functools import lru_cache
@lru_cache(maxsize=100)
def search_with_cache(query: str) -> str:
"""Cache search results to avoid repeated calls."""
return search_info(query)
def react_with_cache(problem: str) -> dict:
"""ReAct that avoids repeating identical searches."""
# Replace the search function with the cached version
TOOL_FUNCTIONS["search_info"] = lambda args: search_with_cache(args["query"])
return run_react(problem, verbose=False)
Pattern 3: Multi-Agent ReAct
def react_multi_agent(problem: str) -> dict:
"""
Splits the problem into sub-problems, solves each one with ReAct,
then synthesizes the results.
"""
# Step 1: Decompose the problem
decomposition = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"Break this problem into 2-3 independent sub-questions: {problem}\nAnswer in this format: 1. ... 2. ... 3. ..."
}],
temperature=0
).choices[0].message.content
# Extract sub-questions (simplified)
sub_questions = [line.strip() for line in decomposition.split("\n") if line.strip() and line[0].isdigit()]
# Step 2: Solve each sub-question with ReAct
partial_results = []
for sub_question in sub_questions[:3]: # Max 3
result = run_react(sub_question, verbose=False)
partial_results.append({
"question": sub_question,
"answer": result["answer"]
})
# Step 3: Synthesize
synthesis_prompt = f"""
Original problem: {problem}
Sub-answers found:
{json.dumps(partial_results, ensure_ascii=False, indent=2)}
Synthesize a complete final answer to the original problem.
"""
synthesis = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": synthesis_prompt}],
temperature=0
).choices[0].message.content
return {
"final_answer": synthesis,
"sub_results": partial_results
}
Integration with Anthropic
import anthropic
client_anthropic = anthropic.Anthropic()
tools_anthropic = [
{
"name": "calculator",
"description": "Evaluates mathematical expressions. Use it to compute numbers.",
"input_schema": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "Mathematical expression"}
},
"required": ["expression"]
}
},
{
"name": "search_info",
"description": "Searches for current factual information.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
}
]
def run_react_anthropic(problem: str, max_steps: int = 6) -> str:
"""
ReAct implementation using Claude (Anthropic).
The Anthropic API handles tool_use differently from OpenAI.
"""
messages = [{"role": "user", "content": problem}]
for step in range(max_steps):
response = client_anthropic.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=1000,
tools=tools_anthropic,
messages=messages,
system="Solve problems using tools. Think before you act. Use the tools whenever you need real data or calculations."
)
# Anthropic uses stop_reason to signal tool_use
if response.stop_reason == "end_turn":
# Final answer
for block in response.content:
if hasattr(block, "text"):
return block.text
# Process tool_use
tool_results = []
messages.append({"role": "assistant", "content": response.content})
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
args = block.input
if tool_name in TOOL_FUNCTIONS:
result = TOOL_FUNCTIONS[tool_name](args)
else:
result = f"Tool not available: {tool_name}"
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
if tool_results:
messages.append({"role": "user", "content": tool_results})
return "Step limit reached"
Real Use Cases
Case 1: Personal Finance Assistant
financial_problems = [
"I have €15,000 in savings. If I put them in a deposit at 2.5% a year for 3 years, how much will I have?",
"What's my net salary if I earn €35,000 gross a year in Spain?",
"If I send 500 USD to Mexico today, how many Mexican pesos will my family receive?"
]
for problem in financial_problems:
print(f"\nPROBLEM: {problem}")
result = run_react(problem, verbose=False)
print(f"ANSWER: {result['answer']}")
print(f"TOOLS: {result['tools_used']}")
Case 2: Automated Research
def research_react(topic: str) -> dict:
"""Research a topic using multiple chained searches."""
prompt = f"""
Research: {topic}
Use the search tools to gather:
1. Current statistical data
2. Historical context if relevant
3. Current trends
At the end, produce a structured summary with the data you found.
"""
return run_react(prompt, max_steps=10, verbose=True)
result = research_react("current state of the semiconductor industry in 2026")
Troubleshooting
Problem 1: The model gets stuck in a loop
Symptom: The model calls the same tool over and over without making progress.
Causes:
- The observation doesn't satisfy what the model expected
- The model doesn't process the tool's result properly
Solution:
def detect_loop(steps: list, window: int = 3) -> bool:
"""Detect whether the model is looping."""
if len(steps) < window * 2:
return False
last_actions = [s.get("tool") for s in steps[-window*2:] if s.get("type") == "action"]
# If the last window*2 actions are all the same, it's a loop
if len(set(last_actions)) == 1 and len(last_actions) >= window:
return True
return False
# In run_react, add:
if detect_loop(steps):
messages.append({
"role": "user",
"content": "It looks like you're repeating the same actions. Do you have enough information to answer? If so, give your best answer with what you have."
})
Problem 2: Tool call with wrong arguments
Symptom: json.loads() fails or the arguments don't have the expected shape.
Solution:
def run_tool_safe(tool_name: str, args_str: str) -> str:
"""Run a tool with robust error handling."""
try:
args = json.loads(args_str)
except json.JSONDecodeError as e:
return f"Error parsing arguments: {e}. Arguments received: {args_str}"
if tool_name not in TOOL_FUNCTIONS:
return f"Tool '{tool_name}' does not exist. Available tools: {list(TOOL_FUNCTIONS.keys())}"
try:
return TOOL_FUNCTIONS[tool_name](args)
except KeyError as e:
return f"Missing argument: {e}"
except Exception as e:
return f"Error running {tool_name}: {e}"
Problem 3: Observation not used in the next Thought
Symptom: The model answers as if it had never received the observation.
Solution: Be more explicit in the system prompt:
improved_system_prompt = """You are an assistant that MUST use tools.
STRICT RULES:
1. NEVER make up numeric data — use calculator
2. NEVER assume prices or current data — look them up with search_info
3. ALWAYS mention in your Thought the data you just received from the Observation
4. If the Observation has an error, try again with a different query
Example of correct usage:
Thought: I need Apple's price to compute this.
Action: search_info("Apple AAPL stock price")
Observation: Apple (AAPL): $218.30
Thought: Apple's current price is $218.30. Now I compute how many shares I can buy with $5000.
Action: calculator("5000 / 218.30")
...
"""
Exercises
Exercise 1: Add a new tool
Implement a tool get_date(timezone: str) -> str that returns the current date and time for a given timezone. Wire this tool into the ReAct system and test it with the question: "What day of the week is it in Tokyo right now?"
See solution
from datetime import datetime
import pytz # pip install pytz
def get_date(timezone: str) -> str:
"""Gets the current date and time for a timezone."""
timezone_map = {
"tokyo": "Asia/Tokyo",
"new_york": "America/New_York",
"madrid": "Europe/Madrid",
"mexico": "America/Mexico_City",
"london": "Europe/London"
}
tz_name = timezone_map.get(timezone.lower(), timezone)
try:
tz = pytz.timezone(tz_name)
now = datetime.now(tz)
return f"Date/time in {timezone}: {now.strftime('%A %d/%m/%Y %H:%M:%S')} ({tz_name})"
except Exception:
return f"Timezone not recognized: {timezone}"
# Add it to the tools schema:
new_tool = {
"type": "function",
"function": {
"name": "get_date",
"description": "Gets the current date and time in a specific timezone.",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Timezone (tokyo, new_york, madrid, mexico, london)"
}
},
"required": ["timezone"]
}
}
}
TOOLS.append(new_tool)
TOOL_FUNCTIONS["get_date"] = lambda args: get_date(args["timezone"])
# Try it:
result = run_react("What day of the week is it in Tokyo right now?")
Exercise 2: ReAct with history
Implement a version of ReAct that remembers previous conversations. The user should be able to ask follow-up questions.
User: "How many Tesla shares can I buy with $10,000?"
Bot: [answers with ReAct]
User: "And if the price drops 20%?" ← Must remember the previous conversation
Bot: [answers taking the previous context into account]
See solution
class ReActConversation:
def __init__(self):
self.history = []
self.system_prompt = """You are a financial assistant.
You remember the previous conversation and use tools for current data."""
def ask(self, question: str) -> str:
# Add the question to the history
self.history.append({"role": "user", "content": question})
# Build messages with the history
messages = [{"role": "system", "content": self.system_prompt}] + self.history.copy()
for _ in range(8):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=TOOLS,
tool_choice="auto",
temperature=0
)
msg = response.choices[0].message
if not msg.tool_calls:
answer = msg.content
self.history.append({"role": "assistant", "content": answer})
return answer
messages.append({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]
})
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
result = TOOL_FUNCTIONS.get(tc.function.name, lambda _: "Error")(args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
return "Step limit reached"
# Usage:
bot = ReActConversation()
print(bot.ask("How many Tesla shares can I buy with $10,000?"))
print(bot.ask("And if the price drops 20%?"))
Exercise 3: Compare ReAct vs CoT
Design an experiment that compares the accuracy of ReAct vs pure CoT for 5 questions that require mathematical calculations with real data. Which ones does ReAct win? Which ones are a tie?
See solution
from collections import namedtuple
Question = namedtuple("Question", ["text", "correct_answer", "question_type"])
questions = [
Question(
"What is 15% of 2,840?",
"426",
"pure_math" # CoT and ReAct should tie
),
Question(
"If Apple is at $218.30, how many shares do I buy with $5,000?",
"22", # 5000/218.30 ≈ 22.9
"external_data" # ReAct should win
),
Question(
"How many kilometers are 100 miles?",
"160.93",
"conversion" # CoT may know it from training; ReAct confirms it
),
]
def pure_cot(question: str) -> str:
"""CoT without tools."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"{question}\n\nThink step by step and give the numeric answer."
}],
temperature=0
)
return response.choices[0].message.content
results = []
for q in questions:
cot_resp = pure_cot(q.text)
react_resp = run_react(q.text, verbose=False)["answer"]
results.append({
"question": q.text[:50],
"question_type": q.question_type,
"cot": cot_resp[:100],
"react": react_resp[:100],
"correct": q.correct_answer
})
print(f"\n{q.text[:50]}...")
print(f" CoT: {cot_resp[:80]}...")
print(f" ReAct: {react_resp[:80]}...")
Expected result: For pure_math, both are similar. For external_data, ReAct is more accurate because it uses the real price.
Exercise 4: ReAct with timeout
Implement a decorator that caps the total execution time of a ReAct loop at N seconds. If time runs out, return the best partial answer.
See solution
import signal
import time
class TimeoutError(Exception):
pass
def with_timeout(seconds: int):
"""Decorator that caps execution time."""
def decorator(func):
def handler(signum, frame):
raise TimeoutError(f"Timeout: {seconds}s exceeded")
def wrapper(*args, **kwargs):
signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
signal.alarm(0) # Cancel the alarm
return result
except TimeoutError:
return {"answer": "Timeout: partial answer not available",
"steps": [], "error": "timeout"}
return wrapper
return decorator
@with_timeout(15) # 15 seconds max
def react_with_timeout(problem: str) -> dict:
return run_react(problem, verbose=False)
# Usage:
result = react_with_timeout("Very complex problem that could take a long time...")
Note: signal.SIGALRM only works on Unix/macOS. On Windows you need threading.Timer.
Summary
- ReAct = Thought → Action → Observation → repeat until you have an answer
- When to use it: Whenever you need external data (prices, dates, searches, precise calculations)
- When NOT to use it: Pure math with no external data (CoT is faster and cheaper)
- Implementation: OpenAI function calling + a tool dispatcher
- Precautions: Cap max_steps, detect loops, handle errors per tool
- Cost: 3-10x more than plain CoT, justified when the data is critical