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

  1. Thought always precedes Action: The model has to reason before it acts
  2. One action at a time: Don't run multiple actions without processing the observations
  3. Observation informs the next Thought: The tool's result has to influence the reasoning
  4. 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

AspectPure CoTReAct
External dataNo (uses the model's knowledge)Yes (APIs, databases)
VerificationMental onlyWith real tools
FreshnessLimited by training cutoffReal time (if you have the tools)
Number of API calls11 + N (one per action)
Cost1x3-10x depending on steps
LatencyLowMedium-High
Math accuracyMediumHigh (with a real calculator)
When to prefer itPure math/logic, no external dataWhenever 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

Additional resources

  1. ReAct: Synergizing Reasoning and Acting in Language Models - Original paper
  2. OpenAI Function Calling Documentation
  3. Anthropic Tool Use Documentation
  4. OpenAI Cookbook: How to call functions with chat models
  5. ReAct implementation examples
  6. LangChain Agents (similar implementation)