ReAct Loop in Machine Learning

import openai
import json
from typing import Any

client = openai.OpenAI()

# Define tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for current information on a topic",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Evaluate a mathematical expression",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression to evaluate"}
                },
                "required": ["expression"]
            }
        }
    }
]

def execute_tool(name: str, arguments: dict) -> str:
    """Route and execute tool calls."""
    if name == "search_web":
        # Replace with real search API (SerpAPI, Tavily, etc.)
        return f"Search results for '{arguments['query']}': [simulated results]"
    elif name == "calculate":
        try:
            result = eval(arguments["expression"])  # Use safe eval in production!
            return str(result)
        except Exception as e:
            return f"Calculation error: {e}"
    return f"Unknown tool: {name}"

def react_loop(
    query: str,
    max_iterations: int = 10,
    model: str = "gpt-4o"
) -> str:
    """Execute a ReAct loop with thought-action-observation cycles."""
    
    system_prompt = """You are a helpful assistant that solves tasks step by step.
For each step, think about what you need to do, then use the available tools.
When you have enough information to answer the user's question, respond directly."""
    
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": query}
    ]
    
    for iteration in range(max_iterations):
        # Step 1: LLM generates thought + action (or final answer)
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice="auto"  # Let the model decide
        )
        
        assistant_message = response.choices[0].message
        messages.append(assistant_message)
        
        # Step 2: Check if the model wants to use a tool
        if not assistant_message.tool_calls:
            # No tool call = model is done reasoning, return answer
            print(f"[ReAct] Completed in {iteration + 1} iterations")
            return assistant_message.content
        
        # Step 3: Execute each tool call and collect observations
        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            print(f"[ReAct] Step {iteration + 1}: {func_name}({func_args})")
            
            # Execute the tool
            observation = execute_tool(func_name, func_args)
            
            # Truncate long observations to control context growth
            if len(observation) > 2000:
                observation = observation[:2000] + "\n[truncated]"
            
            # Append observation as tool response
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": observation
            })
    
    # Max iterations reached -- force a final answer
    messages.append({
        "role": "user",
        "content": "You have reached the maximum number of steps. Please provide your best answer now based on what you have learned so far."
    })
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    print(f"[ReAct] Force-terminated after {max_iterations} iterations")
    return response.choices[0].message.content

# Usage
answer = react_loop("What is the population of Bengaluru and what is 15% of it?")
print(answer)

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
import asyncio

# Define tools with proper docstrings (these become tool descriptions)
@tool
def search_database(query: str) -> str:
    """Search the product database for items matching the query.
    Use this when you need to find specific products, prices, or inventory."""
    # Replace with real DB query
    return f"Found 3 products matching '{query}': Widget A (INR 599), Widget B (INR 899), Widget C (INR 1299)"

@tool
def check_inventory(product_id: str) -> str:
    """Check real-time inventory for a specific product.
    Use this to verify stock availability before confirming orders."""
    # Replace with real inventory API
    return f"Product {product_id}: 42 units in stock at Bengaluru warehouse, 18 units at Mumbai warehouse"

@tool
def calculate_discount(price: float, discount_percent: float) -> str:
    """Calculate the discounted price given original price and discount percentage."""
    discounted = price * (1 - discount_percent / 100)
    savings = price - discounted
    return f"Original: INR {price:.2f}, Discount: {discount_percent}%, Final: INR {discounted:.2f}, Savings: INR {savings:.2f}"

# Initialize the LLM
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,       # Deterministic for agents
    max_tokens=1024,     # Cap output per step
)

# Create the ReAct agent with LangGraph
agent = create_react_agent(
    model=llm,
    tools=[search_database, check_inventory, calculate_discount],
    # Optional: customize the system prompt
    prompt="You are a helpful shopping assistant for an Indian e-commerce platform. "
           "Always check inventory before confirming availability. "
           "Quote all prices in INR.",
)

# Run the agent
async def main():
    result = await agent.ainvoke({
        "messages": [
            HumanMessage(
                content="Find me the cheapest widget and check if it's in stock. "
                        "Also tell me the price after a 20% Diwali discount."
            )
        ]
    })
    
    # Print the full trajectory
    for msg in result["messages"]:
        print(f"[{msg.type}] {msg.content[:200] if msg.content else '(tool call)'}")
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"  -> Tool: {tc['name']}({tc['args']})")

asyncio.run(main())

import time
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)

@dataclass
class ReActStep:
    """One iteration of the ReAct loop."""
    iteration: int
    thought: str
    action: str
    action_input: dict
    observation: str
    error: Optional[str] = None
    latency_ms: float = 0.0
    tokens_used: int = 0

@dataclass
class ReActResult:
    """Final result of a ReAct loop execution."""
    answer: str
    steps: list[ReActStep] = field(default_factory=list)
    total_iterations: int = 0
    total_tokens: int = 0
    total_latency_ms: float = 0.0
    terminated_by: str = "completion"  # completion | max_iterations | timeout | error

class ReActAgent:
    """Production ReAct agent with self-correction, cost tracking, and observability."""
    
    def __init__(
        self,
        llm_client,
        tools: dict,
        system_prompt: str,
        max_iterations: int = 10,
        max_retries_per_tool: int = 2,
        timeout_seconds: float = 120.0,
        max_observation_chars: int = 3000,
        token_budget: int = 50000,
    ):
        self.llm = llm_client
        self.tools = tools  # {name: callable}
        self.system_prompt = system_prompt
        self.max_iterations = max_iterations
        self.max_retries = max_retries_per_tool
        self.timeout = timeout_seconds
        self.max_obs_chars = max_observation_chars
        self.token_budget = token_budget
    
    def _truncate_observation(self, obs: str) -> str:
        if len(obs) <= self.max_obs_chars:
            return obs
        half = self.max_obs_chars // 2
        return obs[:half] + f"\n\n[... truncated {len(obs) - self.max_obs_chars} chars ...]\n\n" + obs[-half:]
    
    def _execute_tool_with_retry(self, name: str, args: dict) -> tuple[str, Optional[str]]:
        """Execute a tool with retry logic. Returns (observation, error)."""
        for attempt in range(self.max_retries + 1):
            try:
                result = self.tools[name](**args)
                return self._truncate_observation(str(result)), None
            except Exception as e:
                error_msg = f"Tool '{name}' failed (attempt {attempt + 1}): {type(e).__name__}: {e}"
                logger.warning(error_msg)
                if attempt == self.max_retries:
                    return error_msg, str(e)
                time.sleep(0.5 * (attempt + 1))  # Backoff
        return "Tool execution failed after all retries", "max_retries_exceeded"
    
    def run(self, query: str) -> ReActResult:
        """Execute the ReAct loop."""
        result = ReActResult(answer="")
        start_time = time.time()
        history = []
        total_tokens = 0
        
        for iteration in range(self.max_iterations):
            # Check timeout
            elapsed = (time.time() - start_time) * 1000
            if elapsed > self.timeout * 1000:
                result.terminated_by = "timeout"
                result.answer = self._force_answer(query, history)
                break
            
            # Check token budget
            if total_tokens > self.token_budget:
                result.terminated_by = "token_budget"
                result.answer = self._force_answer(query, history)
                break
            
            # Generate thought + action
            step_start = time.time()
            llm_response = self._call_llm(query, history)
            step_latency = (time.time() - step_start) * 1000
            total_tokens += llm_response.get("tokens", 0)
            
            thought = llm_response["thought"]
            action = llm_response["action"]
            action_input = llm_response["action_input"]
            
            # Check for completion
            if action == "FINISH":
                step = ReActStep(
                    iteration=iteration + 1,
                    thought=thought,
                    action="FINISH",
                    action_input=action_input,
                    observation="",
                    latency_ms=step_latency,
                    tokens_used=llm_response.get("tokens", 0)
                )
                result.steps.append(step)
                result.answer = action_input.get("answer", thought)
                result.terminated_by = "completion"
                break
            
            # Execute tool
            if action not in self.tools:
                observation = f"Error: Tool '{action}' not found. Available tools: {list(self.tools.keys())}"
                error = "tool_not_found"
            else:
                observation, error = self._execute_tool_with_retry(action, action_input)
            
            step = ReActStep(
                iteration=iteration + 1,
                thought=thought,
                action=action,
                action_input=action_input,
                observation=observation,
                error=error,
                latency_ms=step_latency,
                tokens_used=llm_response.get("tokens", 0)
            )
            result.steps.append(step)
            history.append(step)
            
            logger.info(f"Step {iteration + 1}: {action}({action_input}) -> {observation[:100]}...")
        
        else:
            # Max iterations reached
            result.terminated_by = "max_iterations"
            result.answer = self._force_answer(query, history)
        
        result.total_iterations = len(result.steps)
        result.total_tokens = total_tokens
        result.total_latency_ms = (time.time() - start_time) * 1000
        return result
    
    def _call_llm(self, query, history):
        """Call LLM with structured output parsing. Implementation depends on provider."""
        # Placeholder -- implement with your LLM provider
        raise NotImplementedError("Implement with OpenAI, Anthropic, etc.")
    
    def _force_answer(self, query, history):
        """Force a best-effort answer when termination is triggered."""
        # Call LLM one final time without tools
        return "Based on the information gathered so far..."

# ReAct Agent Configuration (YAML)
agent:
  name: customer-support-agent
  model: gpt-4o
  temperature: 0
  max_tokens_per_step: 1024

loop:
  max_iterations: 10
  timeout_seconds: 120
  token_budget: 50000
  observation_max_chars: 3000
  force_answer_on_termination: true

tools:
  - name: search_knowledge_base
    description: "Search internal knowledge base for support articles"
    timeout_ms: 5000
    retry_count: 2
  - name: lookup_order
    description: "Look up order details by order ID or customer email"
    timeout_ms: 3000
    retry_count: 1
  - name: check_refund_policy
    description: "Check refund eligibility for a given order"
    timeout_ms: 2000
    retry_count: 1
  - name: escalate_to_human
    description: "Escalate the conversation to a human agent when the issue is too complex"
    timeout_ms: 1000
    retry_count: 0

observability:
  log_full_trajectory: true
  track_token_usage: true
  alert_on_max_iterations: true
  export_to: langsmith