python -m venv venv
source venv/bin/activate
pip install openai httpx pydantic rich tiktoken
export OPENAI_API_KEY="sk-your-key"

# project structure:
# react_agent/
#   __init__.py
#   agent.py          # Main ReAct agent class
#   tools.py          # Tool definitions and registry
#   memory.py         # Working memory management
#   parser.py         # Output parsing logic
#   prompts.py        # Prompt templates
#   config.py         # Configuration settings
#   logger.py         # Structured logging
# tests/
#   test_agent.py     # Unit tests
#   test_tools.py     # Tool tests
# main.py             # Entry point

# models.py
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from enum import Enum
from datetime import datetime
import logging

logger = logging.getLogger(__name__)


class ActionType(str, Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    ANSWER = "answer"


class ToolCall(BaseModel):
    """Represents a single tool invocation with its results."""
    name: str
    arguments: Dict[str, Any]
    result: Optional[str] = None
    error: Optional[str] = None
    timestamp: datetime = Field(default_factory=datetime.now)
    execution_time_ms: float = 0.0

    def is_successful(self) -> bool:
        return self.error is None and self.result is not None


class ReActStep(BaseModel):
    """A single step in the ReAct reasoning trace."""
    step_number: int
    action_type: ActionType
    content: str
    tool_call: Optional[ToolCall] = None
    tokens_used: int = 0
    timestamp: datetime = Field(default_factory=datetime.now)


class AgentConfig(BaseModel):
    """Configuration for the ReAct agent."""
    model: str = "gpt-4-turbo-preview"
    max_iterations: int = 10
    max_tokens: int = 4000
    temperature: float = 0.0
    timeout: float = 30.0
    max_retries: int = 3
    log_level: str = "INFO"


class AgentResult(BaseModel):
    """Result of a complete ReAct agent execution."""
    success: bool
    answer: str
    steps: List[ReActStep]
    total_tokens: int
    total_time: float
    iterations: int
    error: Optional[str] = None

    @property
    def avg_tokens_per_step(self) -> float:
        return self.total_tokens / max(self.iterations, 1)

    @property
    def cost_estimate(self) -> float:
        """Estimate cost based on GPT-4 pricing."""
        input_tokens = sum(s.tokens_used for s in self.steps if s.action_type != ActionType.OBSERVATION)
        output_tokens = sum(s.tokens_used for s in self.steps if s.action_type == ActionType.OBSERVATION)
        return (input_tokens * 0.01 + output_tokens * 0.03) / 1000

# tools.py
from typing import Callable, Any, Dict, List, Optional
from dataclasses import dataclass, field
import json
import logging
import time

logger = logging.getLogger(__name__)


@dataclass
class Tool:
    """Represents a callable tool with metadata."""
    name: str
    description: str
    function: Callable
    parameters: Dict[str, Any] = field(default_factory=dict)
    is_async: bool = False
    timeout: float = 30.0
    max_retries: int = 3
    last_error: Optional[str] = None
    error_count: int = 0

    def record_error(self, error: str) -> None:
        self.last_error = error
        self.error_count += 1
        logger.warning(f"Tool '{self.name}' error: {error}")


class ToolRegistry:
    """Registry for managing and executing tools with validation."""

    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        """Register a tool with validation."""
        if tool.name in self._tools:
            logger.warning(f"Overwriting existing tool: {tool.name}")
        self._tools[tool.name] = tool
        logger.info(f"Registered tool: {tool.name}")

    def get_tool(self, name: str) -> Tool:
        """Get a tool by name, raising ValueError if not found."""
        if name not in self._tools:
            raise ValueError(f"Tool '{name}' not found. Available: {list(self._tools.keys())}")
        return self._tools[name]

    def list_tools(self) -> List[Dict[str, Any]]:
        """List all registered tools with their schemas."""
        return [
            {
                "name": t.name,
                "description": t.description,
                "parameters": t.parameters,
                "error_count": t.error_count,
            }
            for t in self._tools.values()
        ]

    def execute(self, name: str, arguments: Dict[str, Any]) -> str:
        """Execute a tool with error handling and logging."""
        tool = self.get_tool(name)
        start_time = time.monotonic()
        try:
            result = tool.function(**arguments)
            execution_time = (time.monotonic() - start_time) * 1000
            logger.info(f"Tool '{name}' executed in {execution_time:.1f}ms")
            return str(result)
        except Exception as e:
            execution_time = (time.monotonic() - start_time) * 1000
            tool.record_error(str(e))
            logger.error(f"Tool '{name}' failed after {execution_time:.1f}ms: {e}")
            return f"Error executing {name}: {str(e)}"

    def get_tools_description(self) -> str:
        """Get formatted tool descriptions for prompt injection."""
        tools_json = json.dumps(self.list_tools(), indent=2)
        return f"Available tools:\n{tools_json}"

    def get_healthy_tools(self) -> List[str]:
        """Get tools that haven't exceeded error threshold."""
        return [name for name, tool in self._tools.items() if tool.error_count < 5]


# Built-in tools with proper error handling
def web_search(query: str, num_results: int = 5) -> str:
    """Search the web for current information."""
    if not query or len(query.strip()) < 2:
        return "Error: Query too short"
    # Placeholder - replace with actual search API
    return f"Search results for '{query}': [Result 1], [Result 2], [Result 3]"


def calculator(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        # Safe evaluation with restricted builtins
        result = eval(expression, {"__builtins__": {}}, {
            "abs": abs, "round": round, "min": min, "max": max,
            "sum": sum, "len": len, "int": int, "float": float,
        })
        return str(result)
    except ZeroDivisionError:
        return "Error: Division by zero"
    except Exception as e:
        return f"Calculation error: {str(e)}"


def read_file(file_path: str, max_chars: int = 5000) -> str:
    """Read the contents of a file with size limit."""
    try:
        from pathlib import Path
        path = Path(file_path)
        if not path.exists():
            return f"Error: File not found: {file_path}"
        if not path.is_file():
            return f"Error: Not a file: {file_path}"
        content = path.read_text(encoding="utf-8", errors="ignore")
        if len(content) > max_chars:
            return content[:max_chars] + f"\n... (truncated at {max_chars} chars)"
        return content
    except PermissionError:
        return f"Error: Permission denied: {file_path}"
    except Exception as e:
        return f"File read error: {str(e)}"


def create_default_tools() -> ToolRegistry:
    """Create a registry with default tools."""
    registry = ToolRegistry()
    registry.register(Tool(
        name="web_search",
        description="Search the web for current information",
        function=web_search,
        parameters={
            "query": {"type": "string", "description": "Search query"},
            "num_results": {"type": "integer", "description": "Number of results to return", "default": 5},
        },
    ))
    registry.register(Tool(
        name="calculator",
        description="Calculate mathematical expressions",
        function=calculator,
        parameters={
            "expression": {"type": "string", "description": "Math expression to evaluate"},
        },
    ))
    registry.register(Tool(
        name="read_file",
        description="Read file contents",
        function=read_file,
        parameters={
            "file_path": {"type": "string", "description": "Path to file"},
            "max_chars": {"type": "integer", "description": "Max characters to read", "default": 5000},
        },
    ))
    return registry

# memory.py
from typing import List, Dict, Optional
from models import ReActStep, ToolCall, ActionType
import tiktoken
import logging

logger = logging.getLogger(__name__)


class WorkingMemory:
    """Manages the agent's working memory with token budget enforcement."""

    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.steps: List[ReActStep] = []
        self.enc = tiktoken.get_encoding("cl100k_base")
        logger.info(f"WorkingMemory initialized with {max_tokens} token budget")

    def add_step(self, step: ReActStep) -> None:
        """Add a step and trim if necessary."""
        self.steps.append(step)
        initial_count = len(self.steps)
        self._trim_if_needed()
        if len(self.steps) < initial_count:
            logger.info(f"Trimmed {initial_count - len(self.steps)} old steps")

    def _trim_if_needed(self) -> None:
        """Remove oldest steps when token budget exceeded."""
        while self._count_tokens() > self.max_tokens and len(self.steps) > 1:
            self.steps.pop(0)

    def _count_tokens(self) -> int:
        """Count total tokens in memory."""
        text = self.get_context_string()
        return len(self.enc.encode(text))

    def get_context_string(self) -> str:
        """Format memory as context string for LLM."""
        parts = []
        for step in self.steps:
            if step.action_type == ActionType.THOUGHT:
                parts.append(f"Thought: {step.content}")
            elif step.action_type == ActionType.ACTION:
                tool_name = step.tool_call.name if step.tool_call else "unknown"
                args = step.tool_call.arguments if step.tool_call else {}
                parts.append(f"Action: {tool_name}({json.dumps(args, default=str)})")
            elif step.action_type == ActionType.OBSERVATION:
                parts.append(f"Observation: {step.content}")
            elif step.action_type == ActionType.ANSWER:
                parts.append(f"Answer: {step.content}")
        return "\n".join(parts)

    def get_recent_observations(self, n: int = 3) -> List[str]:
        """Get the most recent N observations."""
        observations = [
            s.content for s in self.steps
            if s.action_type == ActionType.OBSERVATION
        ]
        return observations[-n:]

    def clear(self) -> None:
        """Clear all memory."""
        self.steps.clear()
        logger.debug("Memory cleared")

    def get_step_count(self) -> int:
        return len(self.steps)

    def get_total_tokens(self) -> int:
        return sum(s.tokens_used for s in self.steps)

# parser.py
import re
import json
from typing import Optional, Tuple
from models import ActionType, ToolCall
import logging

logger = logging.getLogger(__name__)


class ReActParser:
    """Parses LLM output into structured Thought/Action/Observation/Answer."""

    THOUGHT_PATTERN = r"Thought:\s*(.+?)(?=\nAction:|\nAnswer:|$)"
    ACTION_PATTERN = r"Action:\s*(\w+)\((.+?)\)"
    ANSWER_PATTERN = r"Answer:\s*(.+)"

    def parse(self, text: str) -> Tuple[ActionType, str, Optional[ToolCall]]:
        """Parse LLM output into action type, content, and optional tool call."""
        if not text or not text.strip():
            logger.warning("Empty LLM output")
            return ActionType.THOUGHT, "I need to think about this more.", None

        thought_match = re.search(self.THOUGHT_PATTERN, text, re.DOTALL)
        if thought_match:
            thought = thought_match.group(1).strip()
            action_match = re.search(self.ACTION_PATTERN, text, re.DOTALL)
            if action_match:
                tool_name = action_match.group(1)
                try:
                    args_str = action_match.group(2)
                    args = self._parse_arguments(args_str)
                    tool_call = ToolCall(name=tool_name, arguments=args)
                    return ActionType.THOUGHT, thought, tool_call
                except Exception as e:
                    logger.warning(f"Failed to parse tool arguments: {e}")
            return ActionType.THOUGHT, thought, None

        answer_match = re.search(self.ANSWER_PATTERN, text, re.DOTALL)
        if answer_match:
            return ActionType.ANSWER, answer_match.group(1).strip(), None

        return ActionType.THOUGHT, text.strip(), None

    def _parse_arguments(self, args_str: str) -> dict:
        """Parse tool arguments from string format."""
        args_str = args_str.strip()
        try:
            return json.loads(args_str)
        except json.JSONDecodeError:
            if "=" in args_str:
                return self._parse_kwargs(args_str)
            return {"input": args_str}

    def _parse_kwargs(self, args_str: str) -> dict:
        """Parse keyword arguments from string."""
        result = {}
        for pair in args_str.split(","):
            if "=" in pair:
                key, value = pair.split("=", 1)
                value = value.strip().strip("'\"")
                result[key.strip()] = value
        return result

# prompts.py
from typing import List, Optional

SYSTEM_PROMPT = """You are a ReAct agent that solves problems by reasoning step-by-step and using tools when needed.

For each problem, follow this exact format:

Thought: [Your reasoning about what to do next]
Action: [tool_name]([arguments])
Observation: [This will be filled in after tool execution]
... (repeat Thought/Action/Observation as needed)
Thought: I now have enough information to answer
Answer: [Your final answer]

Rules:
1. Always start with a Thought
2. Use only available tools
3. Wait for Observation before continuing
4. Provide final Answer when ready
5. Be concise and accurate
6. If a tool fails, try an alternative approach
7. Never fabricate tool results"""


def build_prompt(
    query: str,
    tools_description: str,
    memory_context: str,
    step_number: int,
    max_iterations: int,
    error_context: Optional[str] = None,
) -> str:
    """Build the full prompt for the LLM."""
    prompt = f"{SYSTEM_PROMPT}\n\n"
    prompt += f"{tools_description}\n\n"
    prompt += f"Current step: {step_number}/{max_iterations}\n\n"

    if memory_context:
        prompt += f"Previous reasoning:\n{memory_context}\n\n"

    if error_context:
        prompt += f"Previous error context:\n{error_context}\n\n"

    prompt += f"User query: {query}\n\n"
    prompt += "Begin your reasoning:\n"

    return prompt

# agent.py
import time
import logging
from typing import Optional
from openai import OpenAI
from models import (
    ReActStep, AgentConfig, AgentResult,
    ActionType, ToolCall
)
from tools import ToolRegistry, create_default_tools
from memory import WorkingMemory
from parser import ReActParser
from prompts import build_prompt

logger = logging.getLogger(__name__)


class ReActAgent:
    """A production-ready ReAct agent with error handling and logging."""

    def __init__(
        self,
        config: Optional[AgentConfig] = None,
        tools: Optional[ToolRegistry] = None,
    ):
        self.config = config or AgentConfig()
        self.tools = tools or create_default_tools()
        self.memory = WorkingMemory(max_tokens=self.config.max_tokens)
        self.parser = ReActParser()
        self.client = OpenAI()
        self.step_counter = 0
        logger.info(f"ReActAgent initialized with model={self.config.model}")

    def run(self, query: str) -> AgentResult:
        """Execute the ReAct loop for a given query."""
        start_time = time.time()
        self.memory.clear()
        self.step_counter = 0
        last_error = None

        logger.info(f"Starting ReAct execution for query: {query[:100]}...")

        try:
            while self.step_counter < self.config.max_iterations:
                self.step_counter += 1
                prompt = self._build_prompt(query, last_error)

                response = self.client.chat.completions.create(
                    model=self.config.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=self.config.temperature,
                    max_tokens=self.config.max_tokens,
                )

                output = response.choices[0].message.content
                tokens_used = response.usage.total_tokens

                action_type, content, tool_call = self.parser.parse(output)

                if action_type == ActionType.ANSWER:
                    step = ReActStep(
                        step_number=self.step_counter,
                        action_type=ActionType.ANSWER,
                        content=content,
                        tokens_used=tokens_used,
                    )
                    self.memory.add_step(step)
                    break

                if tool_call:
                    step = ReActStep(
                        step_number=self.step_counter,
                        action_type=ActionType.THOUGHT,
                        content=content,
                        tool_call=tool_call,
                        tokens_used=tokens_used,
                    )
                    self.memory.add_step(step)

                    # Execute tool with retry logic
                    result = self._execute_tool_with_retry(tool_call)
                    tool_call.result = result

                    obs_step = ReActStep(
                        step_number=self.step_counter,
                        action_type=ActionType.OBSERVATION,
                        content=result,
                    )
                    self.memory.add_step(obs_step)

                    # Track errors for context
                    if result.startswith("Error"):
                        last_error = result
                    else:
                        last_error = None
                else:
                    step = ReActStep(
                        step_number=self.step_counter,
                        action_type=ActionType.THOUGHT,
                        content=content,
                        tokens_used=tokens_used,
                    )
                    self.memory.add_step(step)

        except Exception as e:
            logger.error(f"Agent execution failed: {e}")
            return AgentResult(
                success=False,
                answer=f"Agent encountered an error: {str(e)}",
                steps=self.memory.steps,
                total_tokens=self.memory.get_total_tokens(),
                total_time=time.time() - start_time,
                iterations=self.step_counter,
                error=str(e),
            )

        total_time = time.time() - start_time
        final_answer = self._get_final_answer()

        logger.info(
            f"Execution complete: {self.step_counter} steps, "
            f"{self.memory.get_total_tokens()} tokens, "
            f"{total_time:.2f}s"
        )

        return AgentResult(
            success=True,
            answer=final_answer,
            steps=self.memory.steps,
            total_tokens=self.memory.get_total_tokens(),
            total_time=total_time,
            iterations=self.step_counter,
        )

    def _execute_tool_with_retry(self, tool_call: ToolCall) -> str:
        """Execute a tool with retry logic."""
        last_error = None
        for attempt in range(self.config.max_retries):
            result = self.tools.execute(tool_call.name, tool_call.arguments)
            if not result.startswith("Error"):
                return result
            last_error = result
            logger.warning(f"Tool attempt {attempt + 1} failed: {result}")
        return last_error

    def _build_prompt(self, query: str, error_context: Optional[str] = None) -> str:
        """Build the prompt for the current step."""
        tools_desc = self.tools.get_tools_description()
        memory_context = self.memory.get_context_string()
        return build_prompt(
            query=query,
            tools_description=tools_desc,
            memory_context=memory_context,
            step_number=self.step_counter,
            max_iterations=self.config.max_iterations,
            error_context=error_context,
        )

    def _get_final_answer(self) -> str:
        """Extract the final answer from memory."""
        for step in reversed(self.memory.steps):
            if step.action_type == ActionType.ANSWER:
                return step.content
        return "Max iterations reached without final answer."

# main.py
import logging
from agent import ReActAgent
from models import AgentConfig
from rich.console import Console
from rich.markdown import Markdown
from rich.table import Table

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")


def main():
    console = Console()
    config = AgentConfig(model="gpt-4-turbo-preview", max_iterations=10)
    agent = ReActAgent(config=config)

    console.print("[bold blue]ReAct Agent Ready![/bold blue]")
    console.print("Type your query or 'quit' to exit.\n")

    while True:
        try:
            query = console.input("[bold green]You:[/bold green] ")
            if query.lower() in ("quit", "exit", "q"):
                break

            result = agent.run(query)

            console.print("\n[bold yellow]Agent Response:[/bold yellow]")
            console.print(Markdown(result.answer))

            # Show metrics table
            table = Table(title="Execution Metrics")
            table.add_column("Metric", style="cyan")
            table.add_column("Value", style="green")
            table.add_row("Steps", str(result.iterations))
            table.add_row("Total Tokens", str(result.total_tokens))
            table.add_row("Total Time", f"{result.total_time:.2f}s")
            table.add_row("Est. Cost", f"${result.cost_estimate:.4f}")
            console.print(table)
            console.print()

        except KeyboardInterrupt:
            break
        except Exception as e:
            console.print(f"[bold red]Error:[/bold red] {e}")


if __name__ == "__main__":
    main()

# CoT: Can only reason from training data
cot = "The capital of France is Paris. Paris has 2.1 million people..."

# ReAct: Can verify and gather real information
react = [
    Thought("I need to find current population of Paris"),
    Action("web_search", {"query": "Paris population 2024"}),
    Observation("Paris population: 2.1 million (2024)"),
    Answer("Paris has approximately 2.1 million people")
]

# Example error handling in the loop
Thought("I need to calculate sqrt(-1), but that's invalid")
Action("calculator", {"expression": "(-1) ** 0.5"})
Observation("Error: math domain error")
Thought("Complex numbers aren't supported. Let me explain the limitation.")
Answer("The square root of -1 is not a real number. In complex numbers, it's i.")

import asyncio
from typing import List

async def execute_tools_concurrently(tools: List[dict]) -> List[str]:
    """Execute multiple tools in parallel when they're independent."""
    tasks = []
    for tool_call in tools:
        task = asyncio.create_call(tool_call["name"], tool_call["args"])
        tasks.append(task)
    return await asyncio.gather(*tasks, return_exceptions=True)

# 1. Create a function implementing the tool logic
def get_stock_price(symbol: str) -> str:
    """Get current stock price from Yahoo Finance."""
    import yfinance as yf
    stock = yf.Ticker(symbol)
    return str(stock.info.get("currentPrice", "N/A"))

# 2. Register it with the ToolRegistry
registry.register(Tool(
    name="get_stock_price",
    description="Get current stock price for a given ticker symbol",
    function=get_stock_price,
    parameters={"symbol": {"type": "string", "description": "Stock ticker (e.g., AAPL)"}},
))

# 3. The tool is now available for the agent to use

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_web_search(query: str) -> str:
    """Cache web search results to avoid redundant API calls."""
    return web_search(query)

def check_budget(memory: WorkingMemory, max_tokens: int = 8000) -> bool:
    """Check if we're approaching token limits."""
    current = memory._count_tokens()
    if current > max_tokens * 0.8:
        logger.warning(f"Token budget at {current/max_tokens*100:.1f}%")
        return True
    return False