Building a ReAct Agent From Scratch

Table of contents

Building a ReAct Agent: A Comprehensive Tutorial

A ReAct (Reasoning and Acting) agent is an AI system that combines language model-based reasoning with the ability to take actions using tools. The agent follows a cycle of:

1. Thinking about what to do
2. Taking actions
3. Observing results
4. Using those observations to inform its next steps

This tutorial will guide you through building a flexible, extensible ReAct agent system from scratch.

Prerequisites

Before starting, ensure you have:

• Python 3.7+
• Basic understanding of Python classes and type hints
• Familiarity with async/await concepts
• A Groq API key (or another LLM provider)

Required packages:

pip install groq python-dotenv

Part 1: Core Data Structures

The Tool Response Structure

We start by defining how tools communicate their results back to the agent:

from dataclasses import dataclass
from typing import Any, Optional

@dataclass
class ToolResponse:
    success: bool
    result: Any
    error: Optional[str] = None

This structure provides a consistent way to handle both successful and failed tool executions. The dataclass decorator automatically creates initialization, representation, and comparison methods.

The Thought State Enum

from enum import Enum

class ThoughtState(Enum):
    THINKING = "THINKING"
    ACTING = "ACTING"
    OBSERVATION = "OBSERVATION"
    FINISH = "FINISH"

This enum tracks the agent’s current state in the reasoning cycle, making it easier to monitor and debug the agent’s behavior.

Part 2: The Tool System

The Tool class provides a standardized interface for all actions the agent can take:

class Tool:
    def __init__(self, name: str, description: str, func: callable):
        self.name = name
        self.description = description
        self.func = func

    def execute(self, *args, **kwargs) -> ToolResponse:
        try:
            result = self.func(*args, **kwargs)
            return ToolResponse(success=True, result=result)
        except Exception as e:
            return ToolResponse(success=False, result=None, error=str(e))

Key features:

• Each tool has a name and description that the LLM can use to understand its purpose
• The execute method provides uniform error handling
• Tools can accept any arguments and return any type of result

Part 3: The Memory System

The Memory class gives our agent the ability to remember and learn from past interactions:

class Memory:
    def __init__(self, max_items: int = 1000):
        self.memories: List[Dict] = []
        self.max_items = max_items

    def add(self, thought: str, action: str, observation: str):
        memory = {
            "timestamp": datetime.datetime.now().isoformat(),
            "thought": thought,
            "action": action,
            "observation": observation
        }
        self.memories.append(memory)
        if len(self.memories) > self.max_items:
            self.memories.pop(0)

    def get_recent(self, n: int = 5) -> List[Dict]:
        return self.memories[-n:]

    def search(self, query: str) -> List[Dict]:
        return [m for m in self.memories if query.lower() in str(m).lower()]

The memory system:

• Maintains a fixed-size history of interactions
• Timestamps each memory for temporal context
• Provides simple search functionality
• Could be enhanced with embedding-based semantic search

Part 4: The LLM Interface

We create a wrapper for the Groq API (though this could be adapted for any LLM provider):

class GroqLLM:
    def __init__(self):
        load_dotenv()
        self.client = Groq(
            api_key=os.environ.get("GROQ_API_KEY"),
        )
        self.model = "llama-3.3-70b-versatile"

    def generate(self, prompt: str) -> str:
        try:
            chat_completion = self.client.chat.completions.create(
                messages=[
                    {
                        "role": "system",
                        "content": "You are a helpful AI assistant that responds in the exact format:\nThought: (reasoning)\nAction: tool_name[parameters] or FINISH[answer]"
                    },
                    {
                        "role": "user",
                        "content": prompt,
                    }
                ],
                model=self.model,
            )
            return chat_completion.choices[0].message.content
        except Exception as e:
            print(f"Error generating response: {e}")
            return "Error: Failed to generate response"

Key features:

• Loads API key from environment variables
• Enforces consistent response format through system message
• Provides error handling for API issues

Part 5: The ReAct Agent

The main agent class brings everything together:

class ReActAgent:
    def __init__(self, llm_client, tools: List[Tool]):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
        self.memory = Memory()

    def _create_prompt(self, query: str, context: Optional[str] = None) -> str:
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])

        recent_memory = self.memory.get_recent(3)
        memory_str = "\n".join([
            f"Previous interaction {i+1}:\n"
            f"Thought: {m['thought']}\n"
            f"Action: {m['action']}\n"
            f"Observation: {m['observation']}"
            for i, m in enumerate(recent_memory)
        ])

        return f"""You are a ReAct agent that can use tools to help solve tasks.
Available tools:
{tool_descriptions}

Format your response as:
Thought: (your reasoning)
Action: tool_name[parameters] or FINISH[final answer]
Observation: (result from tool or final answer)

Recent memory:
{memory_str}

Current context: {context if context else 'None'}
Current query: {query}

Let's approach this step by step:
"""

The agent combines:

• Tool management
• Memory system
• Prompt engineering
• Step-by-step execution

Part 6: Action Parsing and Execution

The agent includes sophisticated action parsing:

def _parse_action(self, action: str) -> Tuple[str, Dict[str, Any]]:
    action = action.strip()
    
    if action.startswith("FINISH"):
        match = re.search(r"FINISH\[(.*)\]", action)
        if match:
            return "FINISH", {"answer": match.group(1).strip()}
        return "FINISH", {"answer": "Task completed"}

    match = re.match(r"(\w+)\[(.*)\]", action)
    if not match:
        raise ValueError(f"Invalid action format: {action}")

    tool_name, params_str = match.groups()
    params = {}
    
    if params_str:
        try:
            if "=" in params_str:
                param_pairs = [p.strip() for p in params_str.split(",")]
                for pair in param_pairs:
                    if "=" in pair:
                        key, value = pair.split("=", 1)
                        params[key.strip()] = eval(value.strip())
            else:
                params = {"input": params_str}
        except Exception as e:
            raise ValueError(f"Invalid parameters format: {params_str}") from e

    return tool_name, params

This parser:

• Handles both FINISH actions and tool calls
• Supports key=value parameter pairs
• Provides flexible parameter parsing
• Includes robust error handling

Part 7: Running the Agent

Finally, we implement the main execution loop:

def run(self, query: str, max_steps: int = 10) -> str:
    context = None
    step = 0

    while step < max_steps:
        thought, action, observation = self._execute_step(query, context)
        self.memory.add(thought, action, observation)

        print(f"\nStep {step + 1}:")
        print(f"Thought: {thought}")
        print(f"Action: {action}")
        print(f"Observation: {observation}")

        if "FINISH" in action:
            return observation

        context = f"Previous observation: {observation}"
        step += 1

    return "Maximum steps reached without completion"

The execution loop:

• Maintains context between steps
• Enforces a maximum step limit
• Provides detailed logging
• Stores execution history in memory

Part 8: Creating and Using the Agent

Here’s how to create and use the agent:

def create_agent():
    tools = [
        Tool("search", "Search the web for information", search_web),
        Tool("calculate", "Evaluate a mathematical expression", calculate),
        Tool("current_time", "Get the current time", current_time)
    ]

    llm_client = GroqLLM()
    agent = ReActAgent(llm_client, tools)
    return agent

# Usage example
agent = create_agent()
result = agent.run("What is the current time plus 2 hours?")
print(f"\nFinal result: {result}")

Extending the Agent

The agent can be extended in several ways:

1. Add new tools by creating new Tool instances
2. Enhance the memory system with embedding-based search
3. Implement async execution for better performance
4. Add structured output validation
5. Implement tool-specific parameter validation
6. Add conversation history management
7. Implement better error recovery strategies

Best Practices

When working with this agent:

1. Always provide clear tool descriptions
2. Implement proper error handling in tools
3. Monitor memory usage for long-running sessions
4. Test tools independently before adding them to the agent
5. Use type hints for better code maintainability
6. Document tool parameters and expected outputs
7. Implement logging for debugging
8. Consider rate limiting for external API calls
9. Add timeout mechanisms for long-running operations
10. Implement proper security measures for sensitive operations

Conclusion

This ReAct agent provides a flexible foundation for building AI systems that can reason about and interact with their environment. By following this tutorial and understanding each component, you can create and customize your own ReAct agent for various applications.

Remember to:

• Start with simple tools and gradually add complexity
• Test thoroughly, especially error handling
• Monitor the agent’s performance and memory usage
• Keep security in mind when implementing new tools
• Document your customizations and extensions

The modular design allows for easy expansion and modification to suit your specific needs.