ReAct: Reasoning + Acting AgentsΒΆ

Implementing the ReAct (Reasoning + Acting) pattern for agents that think step-by-step before acting.

# Install dependencies
# !pip install openai

Building a ReAct AgentΒΆ

import re
import json
from typing import Dict, List, Any, Optional

class ReActAgent:
    """ReAct pattern implementation"""
    
    def __init__(self):
        self.tools = {
            "search": self.search,
            "calculate": self.calculate,
            "get_weather": self.get_weather
        }
        self.max_iterations = 5
        self.trace = []  # Store reasoning trace
    
    def search(self, query: str) -> str:
        """Mock web search"""
        # Simulated search results
        knowledge_base = {
            "capital of france": "Paris is the capital of France",
            "paris population": "Paris has approximately 2.2 million people",
            "eiffel tower height": "The Eiffel Tower is 330 meters (1,083 feet) tall",
            "who invented the telephone": "Alexander Graham Bell is credited with inventing the telephone in 1876",
            "speed of light": "The speed of light is approximately 299,792,458 meters per second"
        }
        
        query_lower = query.lower()
        for key, value in knowledge_base.items():
            if key in query_lower:
                return value
        
        return f"No specific information found for '{query}'"
    
    def calculate(self, expression: str) -> float:
        """Safe calculator"""
        try:
            return eval(expression, {"__builtins__": {}}, {})
        except Exception as e:
            return f"Error: {str(e)}"
    
    def get_weather(self, location: str) -> str:
        """Mock weather API"""
        import random
        temp = random.randint(50, 85)
        conditions = ["sunny", "cloudy", "rainy"]
        return f"Weather in {location}: {temp}Β°F, {random.choice(conditions)}"
    
    def parse_action(self, thought: str) -> Optional[Dict[str, str]]:
        """Extract action from thought"""
        # Look for Action: tool_name(args) pattern
        match = re.search(r'ACTION:\s*(\w+)\(([^)]*)\)', thought, re.IGNORECASE)
        if match:
            return {
                "tool": match.group(1),
                "args": match.group(2).strip('"\'')
            }
        return None
    
    def should_stop(self, thought: str) -> bool:
        """Check if agent is ready to answer"""
        stop_words = ["answer:", "final answer:", "result:", "have all the information"]
        return any(word in thought.lower() for word in stop_words)
    
    def think(self, query: str, observations: List[str]) -> str:
        """Generate next thought (simplified - in real agent, LLM does this)"""
        # This is a simplified simulation
        # In a real agent, the LLM would generate this based on query and observations
        
        if not observations:
            # First iteration
            if "capital" in query.lower() and "population" in query.lower():
                return "THOUGHT: I need to find the capital first\nACTION: search(\"capital of France\")"
            elif "weather" in query.lower():
                location = "San Francisco"  # Extract from query
                return f"THOUGHT: I should check the weather\nACTION: get_weather(\"{location}\")"
            elif any(op in query for op in ["+", "-", "*", "/"]):
                return f"THOUGHT: I need to calculate this\nACTION: calculate(\"{query}\")"
        
        elif len(observations) == 1:
            # Second iteration - check if we need more info
            if "capital" in query.lower() and "population" in query.lower():
                if "Paris" in observations[0]:
                    return "THOUGHT: Now I need to find the population of Paris\nACTION: search(\"Paris population\")"
        
        # Ready to answer
        return "THOUGHT: I have all the information needed\nFINAL ANSWER: Based on the observations"
    
    def run(self, query: str) -> str:
        """Run ReAct loop"""
        print(f"\n{'='*70}")
        print(f"🎯 Query: {query}")
        print(f"{'='*70}\n")
        
        observations = []
        self.trace = []
        
        for iteration in range(self.max_iterations):
            print(f"--- Iteration {iteration + 1} ---\n")
            
            # 1. THINK
            thought = self.think(query, observations)
            self.trace.append({"type": "thought", "content": thought})
            print(f"πŸ’­ {thought}\n")
            
            # 2. Check if ready to answer
            if self.should_stop(thought):
                # Extract final answer
                answer_match = re.search(r'(?:FINAL |)ANSWER:\s*(.+)', thought, re.IGNORECASE | re.DOTALL)
                if answer_match:
                    final_answer = answer_match.group(1).strip()
                else:
                    final_answer = "\n".join(observations)
                
                print(f"βœ… Final Answer: {final_answer}\n")
                return final_answer
            
            # 3. ACT
            action = self.parse_action(thought)
            if not action:
                print("⚠️ No action found, stopping\n")
                break
            
            print(f"πŸ”§ ACTION: {action['tool']}({action['args']})")
            
            # Execute tool
            if action['tool'] in self.tools:
                observation = self.tools[action['tool']](action['args'])
            else:
                observation = f"Unknown tool: {action['tool']}"
            
            # 4. OBSERVE
            observations.append(str(observation))
            self.trace.append({"type": "observation", "content": observation})
            print(f"πŸ“Š OBSERVATION: {observation}\n")
        
        return "Max iterations reached without finding answer"
    
    def print_trace(self):
        """Print full reasoning trace"""
        print("\n" + "="*70)
        print("REASONING TRACE")
        print("="*70)
        for i, step in enumerate(self.trace, 1):
            print(f"\nStep {i} ({step['type'].upper()}):")
            print(step['content'])
        print("="*70 + "\n")

# Create agent
agent = ReActAgent()

# Test 1: Multi-step query
result = agent.run("What's the population of the capital of France?")
agent.print_trace()

ReAct with Real LLMΒΆ

Here’s how to implement ReAct with a real LLM:

# Example with OpenAI (requires API key)
'''
from openai import OpenAI

class ReActAgentLLM:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.tools = {...}  # Same tools as above
        
        self.system_prompt = """
You are a helpful assistant that uses the ReAct pattern.

For each query, follow this format:
THOUGHT: [Your reasoning about what to do next]
ACTION: [tool_name(arguments)]

After seeing the observation, think again:
THOUGHT: [Your reasoning about the observation]

When you have enough information:
THOUGHT: I have all the information needed
FINAL ANSWER: [Your complete answer]

Available tools:
- search(query): Search for information
- calculate(expression): Perform calculations
- get_weather(location): Get weather information
"""
    
    def think(self, query: str, observations: List[str]) -> str:
        """Use LLM to generate next thought"""
        messages = [{"role": "system", "content": self.system_prompt}]
        messages.append({"role": "user", "content": query})
        
        # Add previous observations
        for obs in observations:
            messages.append({"role": "assistant", "content": f"OBSERVATION: {obs}"})
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            temperature=0
        )
        
        return response.choices[0].message.content
    
    def run(self, query: str, max_iterations: int = 5) -> str:
        observations = []
        
        for i in range(max_iterations):
            # Get LLM's thought
            thought = self.think(query, observations)
            print(f"πŸ’­ {thought}\n")
            
            # Check if done
            if "FINAL ANSWER" in thought:
                return thought.split("FINAL ANSWER:")[1].strip()
            
            # Parse and execute action
            action = self.parse_action(thought)
            if action:
                result = self.tools[action['tool']](action['args'])
                observations.append(str(result))
                print(f"πŸ“Š OBSERVATION: {result}\n")
        
        return "Max iterations reached"

# Usage:
# agent = ReActAgentLLM(api_key="your-key")
# result = agent.run("What's the population of the capital of France?")
'''

print("ReAct with LLM example (commented - requires API key)")

Advanced: Self-ReflectionΒΆ

Agents can critique their own reasoning:

class SelfReflectiveAgent(ReActAgent):
    """Agent that can reflect on its reasoning"""
    
    def reflect(self, thought: str, observation: str) -> str:
        """Analyze if the action was helpful"""
        # Simplified reflection
        if "error" in observation.lower() or "not found" in observation.lower():
            return "❌ REFLECTION: This action didn't help. I should try a different approach."
        else:
            return "βœ“ REFLECTION: This information is useful for answering the query."
    
    def run(self, query: str) -> str:
        """Run with reflection"""
        print(f"\n{'='*70}")
        print(f"🎯 Query: {query}")
        print(f"{'='*70}\n")
        
        observations = []
        
        for iteration in range(self.max_iterations):
            print(f"--- Iteration {iteration + 1} ---\n")
            
            thought = self.think(query, observations)
            print(f"πŸ’­ {thought}\n")
            
            if self.should_stop(thought):
                return "Query completed"
            
            action = self.parse_action(thought)
            if not action:
                break
            
            print(f"πŸ”§ ACTION: {action['tool']}({action['args']})")
            observation = self.tools[action['tool']](action['args'])
            print(f"πŸ“Š OBSERVATION: {observation}\n")
            
            # NEW: Reflect on the observation
            reflection = self.reflect(thought, str(observation))
            print(f"πŸ” {reflection}\n")
            
            observations.append(str(observation))
        
        return "Completed"

# Test self-reflective agent
reflective_agent = SelfReflectiveAgent()
reflective_agent.run("What's the height of the Eiffel Tower?")

Best PracticesΒΆ

1. Clear Thought FormatΒΆ

THOUGHT: [Clear reasoning]
ACTION: [Specific tool and arguments]

2. Limit IterationsΒΆ

  • Prevent infinite loops

  • Typically 5-10 iterations max

  • Track cost/time

3. Error RecoveryΒΆ

  • Handle tool failures gracefully

  • Allow agent to try alternative approaches

  • Provide helpful error messages

4. DebuggingΒΆ

  • Log all thoughts and observations

  • Visualize reasoning trace

  • Track which tools are used

Key TakeawaysΒΆ

βœ… ReAct combines reasoning and acting in a loop

βœ… Agents think β†’ act β†’ observe β†’ repeat

βœ… Enables multi-step problem solving

βœ… Self-reflection improves agent performance

βœ… Always limit iterations and handle errors