ReAct Agent Pattern — Why Your Agent Loops Forever at 3am and How to Fix It

import json
import logging
from typing import Any

logger = logging.getLogger(__name__)

class ReActLoop:
    def __init__(self, model, tools: dict[str, Any], max_iterations: int = 10):
        self.model = model
        self.tools = tools
        self.max_iterations = max_iterations
        self.history: list[dict] = []

    def run(self, user_query: str) -> str:
        self.history.append({"role": "user", "content": user_query})
        for step in range(self.max_iterations):
            response = self.model.generate(self.history)
            # Parse the model's output — expect JSON with 'thought' and 'action' or 'answer'
            try:
                parsed = json.loads(response)
            except json.JSONDecodeError as e:
                logger.warning(f"Malformed JSON at step {step}: {e}. Raw: {response[:200]}")
                # Send correction prompt
                self.history.append({"role": "assistant", "content": response})
                self.history.append({"role": "user", "content": "Your output was not valid JSON. Please output a JSON object with 'thought' and 'action' or 'answer'."})
                continue
            if "answer" in parsed:
                return parsed["answer"]
            if "action" not in parsed or "tool" not in parsed["action"]:
                logger.warning(f"Missing action/tool at step {step}: {parsed}")
                self.history.append({"role": "assistant", "content": response})
                self.history.append({"role": "user", "content": "Your output must include an 'action' with a 'tool' field. Please try again."})
                continue
            tool_name = parsed["action"]["tool"]
            tool_args = parsed["action"].get("args", {})
            if tool_name not in self.tools:
                logger.error(f"Unknown tool '{tool_name}' at step {step}")
                observation = f"Error: tool '{tool_name}' is not available."
            else:
                try:
                    observation = self.tools[tool_name](**tool_args)
                except Exception as e:
                    observation = f"Tool error: {str(e)}"
            # Append the thought, action, and observation to history
            self.history.append({"role": "assistant", "content": response})
            self.history.append({"role": "observation", "content": json.dumps(observation)})
        raise MaxIterationsExceeded(f"Agent did not produce a final answer within {self.max_iterations} steps.")

class MaxIterationsExceeded(Exception):
    pass

import json
import requests
from openai import OpenAI

# Initialize the LLM client
client = OpenAI(api_key="sk-...")  # Replace with your key

def search_web(query: str) -> str:
    """Search the web and return top result snippets."""
    try:
        response = requests.get(f"https://api.duckduckgo.com/?q={query}&format=json")
        response.raise_for_status()
        data = response.json()
        # Extract relevant snippets
        results = [item["Text"] for item in data.get("RelatedTopics", []) if "Text" in item]
        return "\n".join(results[:3]) if results else "No results found."
    except Exception as e:
        return f"Search failed: {str(e)}"

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        result = eval(expression)  # Use with caution; consider a safer eval
        return str(result)
    except Exception as e:
        return f"Calculation error: {str(e)}"

tools = {
    "search": search_web,
    "calculate": calculate
}

# Build the system prompt
system_prompt = f"""You are a helpful assistant with access to the following tools:
- search(query): Search the web for information.
- calculate(expression): Evaluate a mathematical expression.

You must output your response in JSON format:
- To use a tool: {{"thought": "...", "action": {{"tool": "tool_name", "args": {{...}}}}}}
- To answer: {{"thought": "...", "answer": "..."}}

You have a maximum of 10 steps to answer. If you cannot answer within that limit, output an answer with your best guess.
"""

def run_agent(user_query: str) -> str:
    messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_query}]
    max_iterations = 10
    for step in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            temperature=0
        )
        content = response.choices[0].message.content
        try:
            parsed = json.loads(content)
        except json.JSONDecodeError:
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": "Invalid JSON. Please output a valid JSON object."})
            continue
        if "answer" in parsed:
            return parsed["answer"]
        if "action" in parsed:
            tool_name = parsed["action"]["tool"]
            tool_args = parsed["action"].get("args", {})
            if tool_name not in tools:
                observation = f"Error: tool '{tool_name}' not found."
            else:
                observation = tools[tool_name](**tool_args)
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"Observation: {observation}"})
        else:
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": "Your output must include either 'action' or 'answer'."})
    return "I was unable to find a definitive answer within the allowed steps. Please refine your query."

# Example usage
print(run_agent("What is the capital of France?"))
print(run_agent("Calculate 123 * 456"))

import json
import redis
from openai import OpenAI
import tiktoken

r = redis.Redis(host='localhost', port=6379, db=0)
client = OpenAI(api_key="sk-...")

def get_session_history(session_id: str) -> list:
    data = r.get(f"session:{session_id}")
    return json.loads(data) if data else []

def save_session_history(session_id: str, history: list):
    r.setex(f"session:{session_id}", 3600, json.dumps(history))  # Expire after 1 hour

def prune_history(history: list, max_tokens: int = 6000):
    enc = tiktoken.encoding_for_model("gpt-4")
    # Always keep system prompt and last 3 messages
    system = history[0]
    recent = history[-3:]
    older = history[1:-3]
    total_tokens = len(enc.encode(json.dumps(system))) + sum(len(enc.encode(json.dumps(m))) for m in recent)
    # Prune older messages until under limit
    while older and total_tokens > max_tokens:
        removed = older.pop(0)
        total_tokens -= len(enc.encode(json.dumps(removed)))
    return [system] + older + recent

def run_agent_production(session_id: str, user_query: str) -> str:
    history = get_session_history(session_id)
    if not history:
        history = [{"role": "system", "content": system_prompt}]  # Assume system_prompt defined
    history.append({"role": "user", "content": user_query})
    max_iterations = 10
    for step in range(max_iterations):
        # Prune history before sending to LLM
        pruned_history = prune_history(history)
        response = client.chat.completions.create(
            model="gpt-4",
            messages=pruned_history,
            temperature=0
        )
        content = response.choices[0].message.content
        # Parse and handle as before
        # ... (same as previous implementation)
        # Save history after each iteration
        save_session_history(session_id, history)
    return "Fallback: Could not answer."

# Mistake 1: No input validation
def get_weather(city: str) -> str:
    if not city.strip():
        return "Error: city name cannot be empty. Please provide a valid city."
    # ... actual API call

# Mistake 2: Ignoring tool errors
# Instead of:
# except Exception as e:
#     observation = "Error occurred"
# Do:
except Exception as e:
    observation = f"Tool failed: {str(e)}. You can try a different approach or rephrase the input."

# Mistake 3: Not truncating observations
def truncate_observation(obs: str, max_chars: int = 1000) -> str:
    if len(obs) > max_chars:
        return obs[:max_chars] + "... [truncated]"
    return obs

# Mistake 4: No max iterations in prompt
system_prompt = """You have access to tools. You must answer within 10 steps.
If you cannot answer after 10 steps, output an answer with your best guess."""

from openai import OpenAI

client = OpenAI(api_key="sk-...")

def classify_query(query: str) -> str:
    """Classify the query into a pattern type."""
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Classify the user query into one of: 'react', 'plan_solve', 'reflexion', 'rewoo'. Respond with only the label."},
            {"role": "user", "content": query}
        ],
        temperature=0
    )
    return response.choices[0].message.content.strip()

# Example usage
query = "Find the latest news about AI and summarize it."
pattern = classify_query(query)
if pattern == "react":
    run_react_agent(query)
elif pattern == "plan_solve":
    run_plan_solve_agent(query)
# ... etc.

import logging
import time
from opentelemetry import trace

tracer = trace.get_tracer(__name__)
logger = logging.getLogger(__name__)

def monitored_run_agent(session_id: str, user_query: str) -> str:
    with tracer.start_as_current_span("agent_run") as span:
        span.set_attribute("session_id", session_id)
        span.set_attribute("query", user_query)
        history = get_session_history(session_id)
        max_iterations = 10
        for step in range(max_iterations):
            with tracer.start_as_current_span(f"iteration_{step}") as iter_span:
                start_time = time.time()
                # ... agent logic ...
                latency = time.time() - start_time
                iter_span.set_attribute("latency", latency)
                iter_span.set_attribute("step", step)
                logger.info({
                    "event": "agent_iteration",
                    "session_id": session_id,
                    "step": step,
                    "latency": latency,
                    "tool_called": tool_name if 'tool_name' in locals() else None,
                    "observation_length": len(observation) if 'observation' in locals() else 0
                })
                # Check for alerts
                if step > 5:
                    logger.warning(f"High iteration count: {step} for session {session_id}")
                if latency > 10:
                    logger.warning(f"High latency: {latency}s for session {session_id} step {step}")
        return "Fallback"

# io.thecodeforge.agent.observation
import json

def query_database(sql: str) -> dict:
    try:
        result = execute_sql(sql)  # assume this exists
        return {
            "status": "success",
            "data": result,
            "error": None
        }
    except Exception as e:
        return {
            "status": "error",
            "data": None,
            "error": repr(e)  # full traceback, not a summary
        }

# Execution in ReAct loop
observation = query_database("SELECT * FROM users")
print(json.dumps(observation, indent=2))
# Output:
# {
#   "status": "error",
#   "data": null,
#   "error": "ProgrammingError: (MySQLdb._exceptions.ProgrammingError) (1146, "Table 'test.users' doesn't exist")"
# }

# io.thecodeforge.agent.fixed_graph
from langgraph.graph import StateGraph, END

def build_fixed_agent():
    graph = StateGraph(AgentState)
    
    # Fixed nodes - LLM never decides order
    graph.add_node("classify", lambda state: {"category": llm_classify(state)})
    graph.add_node("search", lambda state: {"search_results": tool_search(state["category"])})
    graph.add_node("validate", lambda state: {"validated": tool_validate(state["search_results"])})
    graph.add_node("synthesize", lambda state: {"response": llm_synthesize(state["validated"])})
    graph.add_node("respond", lambda state: {"output": state["response"]})

    # Deterministic edges - no conditional branching
    graph.add_edge("classify", "search")
    graph.add_edge("search", "validate")
    graph.add_edge("validate", "synthesize")
    graph.add_edge("synthesize", END)
    
    return graph.compile()

agent = build_fixed_agent()
result = agent.invoke({"query": "latest AWS prices"})
print("Execution took", len(result), "steps - always 5")