┌─────────────────────────────────────────────────────────────┐

│                     ReAct Loop                              │

├─────────────────────────────────────────────────────────────┤

│                                                             │

│    ┌──────────┐    ┌──────────┐    ┌───────────┐           │

│    │ THOUGHT  │───▶│  ACTION  │───▶│OBSERVATION│           │

│    │          │    │          │    │           │           │

│    │ "I need  │    │ search(  │    │ "Results: │           │

│    │  to find │    │  'weather│    │  London   │           │

│    │  the     │    │   London'│    │  15°C,    │           │

│    │  weather"│    │ )        │    │  cloudy"  │           │

│    └──────────┘    └──────────┘    └─────┬─────┘           │

│          ▲                               │                  │

│          │                               │                  │

│          └───────────────────────────────┘                  │

│                    (repeat)                                 │

│                                                             │

│    Until: Task complete OR max steps reached                │

│                                                             │

└─────────────────────────────────────────────────────────────┘

# Silent agent (problematic)

Action: search("Python web frameworks")

Action: search("Django vs Flask")  

Action: write_report(...)  # Report may be wrong, hard to debug

# ReAct agent (transparent)

Thought: I need to find popular Python web frameworks

Action: search("Python web frameworks 2024")

Observation: Django, Flask, FastAPI are the top 3...

Thought: I should compare these frameworks. Let me focus on FastAPI 

        since it wasn't in my training data and is newest.

Action: search("FastAPI features performance")

Observation: FastAPI is async-first, automatic OpenAPI docs...

Thought: Now I have enough information to write a comparison.

Action: write_report(...)  # Each step is traceable

# Chain-of-Thought (can hallucinate)

Thought: The capital of Australia is Sydney because it's the largest city.

Answer: Sydney  # WRONG

# ReAct (grounded in facts)

Thought: I should verify the capital of Australia

Action: search("capital of Australia")

Observation: The capital of Australia is Canberra, not Sydney...

Thought: The observation confirms it's Canberra, not Sydney as I initially thought.

Answer: Canberra  # CORRECT

Thought: I'll try to get the weather from the primary API

Action: api_call("weather.primary.com/london")

Observation: Error 503 - Service unavailable

Thought: Primary API is down. I'll try the backup API instead.

Action: api_call("weather.backup.com/london")

Observation: {"temp": 15, "condition": "cloudy"}

Thought: Got the weather from backup. I can now answer the question.

import openai

import json

import re

class ReActAgent:

    def __init__(self, tools: dict):

        self.client = openai.OpenAI()

        self.tools = tools

        self.max_steps = 10

    def run(self, task: str) -> str:

        """Execute task using ReAct loop"""

        prompt = self._build_initial_prompt(task)

        for step in range(self.max_steps):

            # Get next thought and action from LLM

            response = self.client.chat.completions.create(

                model="gpt-4o",

                messages=[{"role": "user", "content": prompt}],

                stop=["Observation:"]  # Stop before observation

            )

            output = response.choices[0].message.content

            prompt += output

            # Check if agent wants to finish

            if "Final Answer:" in output:

                return self._extract_final_answer(output)

            # Extract and execute action

            action, action_input = self._parse_action(output)

            if action is None:

                prompt += "\nObservation: Could not parse action. Please use format 'Action: tool_name[input]'\n"

                continue

            # Execute the action

            observation = self._execute_action(action, action_input)

            prompt += f"\nObservation: {observation}\n"

        return "Max steps reached without finding answer."

    def _build_initial_prompt(self, task: str) -> str:

        tool_descriptions = "\n".join([

            f"- {name}: {func.__doc__ or 'No description'}"

            for name, func in self.tools.items()

        ])

        return f"""Answer the following question using the available tools.

Available tools:

{tool_descriptions}

Use this format:

Thought: [Your reasoning about what to do next]

Action: tool_name[input]

Observation: [Result of the action - will be provided]

... (repeat Thought/Action/Observation as needed)

Thought: I now have enough information to answer.

Final Answer: [Your final answer]

Question: {task}

"""

    def _parse_action(self, text: str) -> tuple:

        """Extract action and input from LLM output"""

        # Match pattern: Action: tool_name[input]

        match = re.search(r'Action:\s*(\w+)\[([^\]]*)\]', text)

        if match:

            return match.group(1), match.group(2)

        # Alternative format: Action: tool_name("input")

        match = re.search(r'Action:\s*(\w+)\("([^"]*)"\)', text)

        if match:

            return match.group(1), match.group(2)

        return None, None

    def _execute_action(self, action: str, action_input: str) -> str:

        """Execute the specified action"""

        if action not in self.tools:

            return f"Error: Unknown tool '{action}'. Available: {list(self.tools.keys())}"

        try:

            result = self.tools[action](action_input)

            return str(result)

        except Exception as e:

            return f"Error executing {action}: {str(e)}"

    def _extract_final_answer(self, text: str) -> str:

        """Extract the final answer from output"""

        match = re.search(r'Final Answer:\s*(.+)', text, re.DOTALL)

        if match:

            return match.group(1).strip()

        return text

# Define tools

def search(query: str) -> str:

    """Search the web for information"""

    # In production, use a real search API

    return f"Search results for '{query}': ..."

def calculate(expression: str) -> str:

    """Evaluate a mathematical expression"""

    try:

        return str(eval(expression))

    except:

        return "Error: Could not evaluate expression"

def lookup(term: str) -> str:

    """Look up a term in the knowledge base"""

    knowledge = {

        "python": "A high-level programming language",

        "react": "A JavaScript library for building UIs",

    }

    return knowledge.get(term.lower(), f"No entry found for '{term}'")

# Usage

agent = ReActAgent(tools={

    "search": search,

    "calculate": calculate,

    "lookup": lookup

})

result = agent.run("What is 25% of the population of France?")

print(result)

Thought: I need to find the population of France first, then calculate 25% of it.

Action: search[population of France 2024]

Observation: The population of France is approximately 68 million people.

Thought: Now I can calculate 25% of 68 million.

Action: calculate[68000000 * 0.25]

Observation: 17000000.0

Thought: I now have the answer.

Final Answer: 25% of France's population is 17 million people.

from hopx import Sandbox

import openai

import re

class CodeReActAgent:

    def __init__(self):

        self.client = openai.OpenAI()

        self.sandbox = None

        self.max_steps = 15

    def run(self, task: str) -> str:

        # Create sandbox for the session

        self.sandbox = Sandbox.create(template="code-interpreter")

        try:

            return self._react_loop(task)

        finally:

            self.sandbox.kill()

    def _react_loop(self, task: str) -> str:

        prompt = self._build_prompt(task)

        for step in range(self.max_steps):

            response = self.client.chat.completions.create(

                model="gpt-4o",

                messages=[{"role": "user", "content": prompt}],

                stop=["Observation:"]

            )

            output = response.choices[0].message.content

            prompt += output

            print(f"\n--- Step {step + 1} ---")

            print(output)

            if "Final Answer:" in output:

                return self._extract_answer(output)

            # Parse action

            action_type, action_content = self._parse_action(output)

            if action_type == "python":

                observation = self._run_code(action_content)

            elif action_type == "bash":

                observation = self._run_bash(action_content)

            elif action_type == "read_file":

                observation = self._read_file(action_content)

            elif action_type == "write_file":

                path, content = action_content.split("|||", 1)

                observation = self._write_file(path.strip(), content.strip())

            else:

                observation = f"Unknown action type: {action_type}"

            prompt += f"\nObservation: {observation}\n"

            print(f"Observation: {observation[:500]}...")

        return "Max steps reached"

    def _build_prompt(self, task: str) -> str:

        return f"""You are an AI assistant that solves tasks by writing and executing code.

Available actions:

- python[code]: Execute Python code

- bash[command]: Run a bash command

- read_file[path]: Read a file

- write_file[path|||content]: Write content to a file

Format:

Thought: [Your reasoning]

Action: action_type[content]

Observation: [Will be provided]

Rules:

- Always think before acting

- Check observations carefully for errors

- Use print() to see outputs in Python

- When done, provide Final Answer:

Task: {task}

"""

    def _parse_action(self, text: str) -> tuple:

        # Match: Action: python[...]

        match = re.search(r'Action:\s*(\w+)\[(.*?)\](?:\s*$|\n)', text, re.DOTALL)

        if match:

            return match.group(1), match.group(2)

        # Match multi-line code blocks

        match = re.search(r'Action:\s*(\w+)\[(.*)\]', text, re.DOTALL)

        if match:

            return match.group(1), match.group(2)

        return None, None

    def _run_code(self, code: str) -> str:

        self.sandbox.files.write("/app/script.py", code)

        result = self.sandbox.commands.run("python /app/script.py", timeout=30)

        if result.exit_code == 0:

            return result.stdout if result.stdout else "(No output)"

        else:

            return f"Error (exit code {result.exit_code}):\n{result.stderr}"

    def _run_bash(self, command: str) -> str:

        result = self.sandbox.commands.run(command, timeout=30)

        output = result.stdout + result.stderr

        return output if output else "(No output)"

    def _read_file(self, path: str) -> str:

        try:

            return self.sandbox.files.read(path)

        except Exception as e:

            return f"Error reading file: {e}"

    def _write_file(self, path: str, content: str) -> str:

        try:

            self.sandbox.files.write(path, content)

            return f"Successfully wrote to {path}"

        except Exception as e:

            return f"Error writing file: {e}"

    def _extract_answer(self, text: str) -> str:

        match = re.search(r'Final Answer:\s*(.+)', text, re.DOTALL)

        return match.group(1).strip() if match else text

# Usage

agent = CodeReActAgent()

result = agent.run("""

Download the iris dataset, train a simple classifier, 

and report the accuracy.

""")

print(f"\nFinal Result:\n{result}")

--- Step 1 ---

Thought: I need to download the iris dataset and train a classifier. 

I'll use scikit-learn which has the iris dataset built-in.

Action: python[

from sklearn.datasets import load_iris

from sklearn.model_selection import train_test_split

from sklearn.ensemble import RandomForestClassifier

from sklearn.metrics import accuracy_score

# Load data

iris = load_iris()

X, y = iris.data, iris.target

# Split

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Train

clf = RandomForestClassifier(n_estimators=100)

clf.fit(X_train, y_train)

# Evaluate

predictions = clf.predict(X_test)

accuracy = accuracy_score(y_test, predictions)

print(f"Accuracy: {accuracy:.2%}")

]

Observation: Accuracy: 96.67%

--- Step 2 ---

Thought: The classifier trained successfully with 96.67% accuracy. 

I have enough information to provide the final answer.

Final Answer: I trained a Random Forest classifier on the Iris dataset 

and achieved 96.67% accuracy on the test set.

def _build_prompt_with_reflection(self, task: str) -> str:

    return f"""...

Format:

Thought: [Your reasoning about what to do]

Action: action_type[content]

Observation: [Result - will be provided]

Reflection: [What did I learn? Did it work? What should I do next?]

..."""

class CriticalReActAgent(ReActAgent):

    def _react_step(self, thought: str, action: str) -> str:

        # First, critique the proposed action

        critique = self._critique_action(thought, action)

        if "PROBLEM:" in critique:

            # Revise action based on critique

            revised = self._revise_action(thought, action, critique)

            return self._execute_action(*self._parse_action(revised))

        return self._execute_action(*self._parse_action(action))

    def _critique_action(self, thought: str, action: str) -> str:

        response = self.client.chat.completions.create(

            model="gpt-4o",

            messages=[{

                "role": "user",

                "content": f"""Critique this action. Is it correct and safe?

Thought: {thought}

Action: {action}

If there's a problem, start with "PROBLEM:" and explain.

If it looks good, say "APPROVED"."""

            }]

        )

        return response.choices[0].message.content

import asyncio

async def parallel_react(task: str, perspectives: list[str]) -> str:

    """Run multiple ReAct agents with different perspectives"""

    async def run_with_perspective(perspective: str):

        prompt = f"You are a {perspective}. {task}"

        return await react_agent.run_async(prompt)

    results = await asyncio.gather(*[

        run_with_perspective(p) for p in perspectives

    ])

    # Merge results

    return merge_answers(results)

# Usage

answer = await parallel_react(

    "What are the pros and cons of microservices?",

    perspectives=["software architect", "DevOps engineer", "developer"]

)

def _build_prompt_structured(self, task: str) -> str:

    return f"""...

Respond in JSON format:

{{

    "thought": "your reasoning",

    "action": {{

        "tool": "tool_name",

        "input": "tool input"

    }}

}}

Or when finished:

{{

    "thought": "final reasoning",

    "final_answer": "your answer"

}}

..."""

class TokenAwareReActAgent:

    def __init__(self, max_context_tokens: int = 8000):

        self.max_tokens = max_context_tokens

        self.history = []

    def _manage_context(self, prompt: str) -> str:

        """Trim history if context too long"""

        estimated_tokens = len(prompt) // 4

        if estimated_tokens > self.max_tokens:

            # Keep first (task) and last N steps

            self.history = self.history[:1] + self.history[-5:]

            prompt = self._rebuild_prompt()

        return prompt

def _execute_with_recovery(self, action: str, input: str, max_retries: int = 3):

    for attempt in range(max_retries):

        try:

            result = self.tools[action](input)

            return result

        except Exception as e:

            if attempt == max_retries - 1:

                return f"Failed after {max_retries} attempts: {e}"

            # Let agent know about failure

            return f"Attempt {attempt + 1} failed: {e}. You can retry."

def _truncate_observation(self, obs: str, max_length: int = 2000) -> str:

    if len(obs) <= max_length:

        return obs

    return obs[:max_length] + f"\n... (truncated, {len(obs) - max_length} chars omitted)"

from hopx import Sandbox

import openai

import json

from datetime import datetime

class ProductionReActAgent:

    def __init__(self):

        self.client = openai.OpenAI()

        self.max_steps = 20

        self.trace = []

    def run(self, task: str, tools: dict) -> dict:

        """Run ReAct loop and return structured result"""

        self.trace = []

        start_time = datetime.now()

        # Create sandbox if code execution needed

        sandbox = None

        if "run_code" in tools:

            sandbox = Sandbox.create(template="code-interpreter")

            tools["run_code"] = lambda code: self._safe_execute(sandbox, code)

        try:

            messages = [

                {"role": "system", "content": self._system_prompt(tools)},

                {"role": "user", "content": task}

            ]

            for step in range(self.max_steps):

                # Get next action

                response = self.client.chat.completions.create(

                    model="gpt-4o",

                    messages=messages,

                    response_format={"type": "json_object"}

                )

                output = json.loads(response.choices[0].message.content)

                self.trace.append({

                    "step": step + 1,

                    "thought": output.get("thought"),

                    "action": output.get("action"),

                    "timestamp": datetime.now().isoformat()

                })

                # Check for completion

                if "final_answer" in output:

                    return {

                        "success": True,

                        "answer": output["final_answer"],

                        "steps": step + 1,

                        "duration": (datetime.now() - start_time).seconds,

                        "trace": self.trace

                    }

                # Execute action

                action = output.get("action", {})

                tool_name = action.get("tool")

                tool_input = action.get("input")

                if tool_name not in tools:

                    observation = f"Error: Unknown tool '{tool_name}'"

                else:

                    try:

                        observation = str(tools[tool_name](tool_input))

                    except Exception as e:

                        observation = f"Error: {e}"

                # Truncate long observations

                if len(observation) > 3000:

                    observation = observation[:3000] + "... (truncated)"

                self.trace[-1]["observation"] = observation

                # Add to messages

                messages.append({

                    "role": "assistant",

                    "content": json.dumps(output)

                })

                messages.append({

                    "role": "user",