Implementing ReAct from Scratch

1REACT_PROMPT = '''You are an AI assistant that solves tasks by reasoning and taking actions.
2
3You have access to these tools:
4{tools_description}
5
6Use this exact format:
7
8Thought: [Your reasoning about what to do next]
9Action: [tool_name(param1="value1", param2="value2")]
10
11After each action, you will see an observation. Then continue with another thought.
12
13When you have enough information to complete the task, use:
14Thought: [Your reasoning about why you're done]
15Action: finish(answer="[Your final answer]")
16
17Important rules:
181. Always start with a Thought before taking an Action
192. Only use one Action per turn
203. Wait for the Observation before your next Thought
214. Be specific in your reasoning
225. If an action fails, think about alternatives
23
24Task: {task}
25
26Begin!
27'''
28
29def format_tools_description(tools: dict) -> str:
30    """Format tools for the prompt."""
31    lines = []
32    for name, tool in tools.items():
33        lines.append(f"- {name}: {tool['description']}")
34        if tool.get('parameters'):
35            for param, info in tool['parameters'].items():
36                lines.append(f"    - {param}: {info['description']}")
37    return "\n".join(lines)
38
39
40def build_prompt(task: str, tools: dict, history: list) -> str:
41    """Build the complete prompt with history."""
42
43    prompt = REACT_PROMPT.format(
44        tools_description=format_tools_description(tools),
45        task=task,
46    )
47
48    # Add history
49    for step in history:
50        prompt += f"\nThought: {step['thought']}"
51        prompt += f"\nAction: {step['action']}"
52        prompt += f"\nObservation: {step['observation']}"
53
54    # Prompt for next thought
55    prompt += "\nThought:"
56
57    return prompt

1TOOLS = {
2    "search": {
3        "description": "Search the web for information",
4        "parameters": {
5            "query": {"description": "The search query", "type": "string"},
6        },
7        "function": lambda query: web_search(query),
8    },
9    "read_file": {
10        "description": "Read the contents of a file",
11        "parameters": {
12            "path": {"description": "Path to the file", "type": "string"},
13        },
14        "function": lambda path: read_file(path),
15    },
16    "write_file": {
17        "description": "Write content to a file",
18        "parameters": {
19            "path": {"description": "Path to the file", "type": "string"},
20            "content": {"description": "Content to write", "type": "string"},
21        },
22        "function": lambda path, content: write_file(path, content),
23    },
24    "run_command": {
25        "description": "Run a shell command",
26        "parameters": {
27            "command": {"description": "The command to run", "type": "string"},
28        },
29        "function": lambda command: run_command(command),
30    },
31    "finish": {
32        "description": "Complete the task with a final answer",
33        "parameters": {
34            "answer": {"description": "The final answer", "type": "string"},
35        },
36        "function": lambda answer: answer,
37    },
38}

1import re
2from dataclasses import dataclass
3from typing import Optional
4
5@dataclass
6class ParsedOutput:
7    thought: str
8    action_name: str
9    action_params: dict[str, str]
10
11class OutputParser:
12    """Parse LLM output into structured format."""
13
14    def parse(self, output: str) -> ParsedOutput:
15        """Parse the LLM output."""
16
17        # Extract thought
18        thought_match = re.search(
19            r"Thought:\s*(.+?)(?=Action:|$)",
20            output,
21            re.DOTALL
22        )
23        thought = thought_match.group(1).strip() if thought_match else ""
24
25        # Extract action
26        action_match = re.search(
27            r"Action:\s*(\w+)\((.*)\)",
28            output
29        )
30
31        if not action_match:
32            raise ValueError(f"Could not parse action from: {output}")
33
34        action_name = action_match.group(1)
35        action_args_str = action_match.group(2)
36
37        # Parse action parameters
38        action_params = self._parse_params(action_args_str)
39
40        return ParsedOutput(
41            thought=thought,
42            action_name=action_name,
43            action_params=action_params,
44        )
45
46    def _parse_params(self, args_str: str) -> dict[str, str]:
47        """Parse action parameters from string."""
48        if not args_str.strip():
49            return {}
50
51        params = {}
52
53        # Match key="value" or key='value' patterns
54        pattern = r'(\w+)\s*=\s*["\'](.*?)["\'']'
55        matches = re.findall(pattern, args_str)
56
57        for key, value in matches:
58            params[key] = value
59
60        return params
61
62
63# Example usage
64parser = OutputParser()
65output = '''Thought: I need to read the config file to understand the settings.
66Action: read_file(path="config.json")'''
67
68parsed = parser.parse(output)
69print(f"Thought: {parsed.thought}")
70print(f"Action: {parsed.action_name}")
71print(f"Params: {parsed.action_params}")

1class RobustParser:
2    """Parser with error recovery."""
3
4    def __init__(self, llm):
5        self.llm = llm
6        self.parser = OutputParser()
7
8    def parse_with_retry(
9        self,
10        output: str,
11        max_retries: int = 3,
12    ) -> ParsedOutput:
13        """Parse with retry on failure."""
14
15        for attempt in range(max_retries):
16            try:
17                return self.parser.parse(output)
18            except ValueError as e:
19                if attempt == max_retries - 1:
20                    # Last attempt: try to fix the format
21                    return self._fix_and_parse(output)
22
23                # Ask LLM to reformat
24                output = self._request_reformat(output, str(e))
25
26        raise ValueError("Could not parse after max retries")
27
28    def _request_reformat(self, output: str, error: str) -> str:
29        """Ask LLM to reformat its output."""
30        prompt = f"""
31Your previous output could not be parsed:
32{output}
33
34Error: {error}
35
36Please reformat your response using exactly this format:
37Thought: [Your reasoning]
38Action: tool_name(param1="value1", param2="value2")
39
40Reformatted response:
41"""
42        return self.llm.generate(prompt)
43
44    def _fix_and_parse(self, output: str) -> ParsedOutput:
45        """Attempt to fix common issues and parse."""
46
47        # Try to extract any thought-like content
48        thought = ""
49        if ":" in output:
50            thought = output.split("\n")[0]
51
52        # Default to finish if we can't parse
53        return ParsedOutput(
54            thought=thought or "Unable to determine next step",
55            action_name="finish",
56            action_params={"answer": "Parsing error - please rephrase"},
57        )

1class ReActAgent:
2    """Complete ReAct agent implementation."""
3
4    def __init__(
5        self,
6        llm,
7        tools: dict,
8        max_steps: int = 10,
9    ):
10        self.llm = llm
11        self.tools = tools
12        self.max_steps = max_steps
13        self.parser = RobustParser(llm)
14
15    def run(self, task: str) -> str:
16        """Run the agent on a task."""
17
18        history = []
19        step = 0
20
21        while step < self.max_steps:
22            step += 1
23            print(f"\n--- Step {step} ---")
24
25            # Build prompt with current history
26            prompt = build_prompt(task, self.tools, history)
27
28            # Get LLM response
29            response = self.llm.generate(prompt)
30
31            # Parse the response
32            try:
33                parsed = self.parser.parse_with_retry(response)
34            except ValueError as e:
35                print(f"Parse error: {e}")
36                continue
37
38            print(f"Thought: {parsed.thought}")
39            print(f"Action: {parsed.action_name}({parsed.action_params})")
40
41            # Execute the action
42            observation = self._execute_action(
43                parsed.action_name,
44                parsed.action_params,
45            )
46            print(f"Observation: {observation[:200]}...")
47
48            # Add to history
49            history.append({
50                "thought": parsed.thought,
51                "action": f"{parsed.action_name}({parsed.action_params})",
52                "observation": observation,
53            })
54
55            # Check if finished
56            if parsed.action_name == "finish":
57                return parsed.action_params.get("answer", observation)
58
59        return "Max steps reached without completing task"
60
61    def _execute_action(
62        self,
63        action_name: str,
64        params: dict,
65    ) -> str:
66        """Execute an action and return observation."""
67
68        tool = self.tools.get(action_name)
69        if not tool:
70            return f"Error: Unknown tool '{action_name}'"
71
72        try:
73            result = tool["function"](**params)
74            return str(result)
75        except Exception as e:
76            return f"Error: {str(e)}"

1"""
2Complete ReAct Agent Implementation
3Run with: python react_agent.py
4"""
5
6import re
7import os
8import subprocess
9from dataclasses import dataclass
10import anthropic
11
12# Initialize Anthropic client
13client = anthropic.Anthropic()
14
15
16@dataclass
17class ParsedOutput:
18    thought: str
19    action_name: str
20    action_params: dict[str, str]
21
22
23class OutputParser:
24    def parse(self, output: str) -> ParsedOutput:
25        thought_match = re.search(
26            r"Thought:\s*(.+?)(?=Action:|$)",
27            output,
28            re.DOTALL
29        )
30        thought = thought_match.group(1).strip() if thought_match else ""
31
32        action_match = re.search(r"Action:\s*(\w+)\((.*)\)", output)
33        if not action_match:
34            raise ValueError(f"Could not parse action from: {output}")
35
36        action_name = action_match.group(1)
37        action_args_str = action_match.group(2)
38
39        params = {}
40        pattern = r'(\w+)\s*=\s*["\'](.*?)["\'"]'
41        matches = re.findall(pattern, action_args_str)
42        for key, value in matches:
43            params[key] = value
44
45        return ParsedOutput(thought, action_name, params)
46
47
48# Tool implementations
49def search(query: str) -> str:
50    """Simulate web search."""
51    return f"Search results for '{query}': [Simulated results]"
52
53
54def read_file(path: str) -> str:
55    """Read a file."""
56    try:
57        with open(path, "r") as f:
58            return f.read()
59    except Exception as e:
60        return f"Error reading file: {e}"
61
62
63def write_file(path: str, content: str) -> str:
64    """Write to a file."""
65    try:
66        with open(path, "w") as f:
67            f.write(content)
68        return f"Successfully wrote to {path}"
69    except Exception as e:
70        return f"Error writing file: {e}"
71
72
73def run_command(command: str) -> str:
74    """Run a shell command."""
75    try:
76        result = subprocess.run(
77            command,
78            shell=True,
79            capture_output=True,
80            text=True,
81            timeout=30,
82        )
83        return result.stdout + result.stderr
84    except Exception as e:
85        return f"Error running command: {e}"
86
87
88def list_files(directory: str) -> str:
89    """List files in a directory."""
90    try:
91        files = os.listdir(directory)
92        return "\n".join(files)
93    except Exception as e:
94        return f"Error listing files: {e}"
95
96
97# Tool definitions
98TOOLS = {
99    "search": {
100        "description": "Search the web for information",
101        "function": search,
102    },
103    "read_file": {
104        "description": "Read the contents of a file",
105        "function": read_file,
106    },
107    "write_file": {
108        "description": "Write content to a file",
109        "function": write_file,
110    },
111    "run_command": {
112        "description": "Run a shell command",
113        "function": run_command,
114    },
115    "list_files": {
116        "description": "List files in a directory",
117        "function": list_files,
118    },
119    "finish": {
120        "description": "Complete the task with a final answer",
121        "function": lambda answer: answer,
122    },
123}
124
125
126REACT_PROMPT = '''You are an AI assistant that solves tasks by reasoning and taking actions.
127
128You have access to these tools:
129- search: Search the web for information
130- read_file: Read the contents of a file (path)
131- write_file: Write content to a file (path, content)
132- run_command: Run a shell command (command)
133- list_files: List files in a directory (directory)
134- finish: Complete the task with a final answer (answer)
135
136Use this exact format:
137
138Thought: [Your reasoning about what to do next]
139Action: tool_name(param="value")
140
141Important:
1421. Always start with a Thought
1432. Use exactly one Action per turn
1443. Wait for Observation before continuing
1454. Use finish() when done
146
147Task: {task}
148
149{history}Thought:'''
150
151
152class ReActAgent:
153    def __init__(self, max_steps: int = 10):
154        self.max_steps = max_steps
155        self.parser = OutputParser()
156
157    def run(self, task: str) -> str:
158        history = []
159        step = 0
160
161        while step < self.max_steps:
162            step += 1
163            print(f"\n{'='*50}")
164            print(f"Step {step}")
165            print('='*50)
166
167            # Build prompt
168            history_str = ""
169            for h in history:
170                history_str += f"Thought: {h['thought']}\n"
171                history_str += f"Action: {h['action']}\n"
172                history_str += f"Observation: {h['observation']}\n\n"
173
174            prompt = REACT_PROMPT.format(task=task, history=history_str)
175
176            # Get LLM response
177            response = client.messages.create(
178                model="claude-sonnet-4-20250514",
179                max_tokens=1024,
180                messages=[{"role": "user", "content": prompt}],
181            )
182            output = response.content[0].text
183
184            # Parse response
185            try:
186                parsed = self.parser.parse(output)
187            except ValueError as e:
188                print(f"Parse error: {e}")
189                print(f"Raw output: {output}")
190                continue
191
192            print(f"Thought: {parsed.thought}")
193            print(f"Action: {parsed.action_name}({parsed.action_params})")
194
195            # Execute action
196            tool = TOOLS.get(parsed.action_name)
197            if not tool:
198                observation = f"Unknown tool: {parsed.action_name}"
199            else:
200                try:
201                    observation = str(tool["function"](**parsed.action_params))
202                except Exception as e:
203                    observation = f"Error: {e}"
204
205            print(f"Observation: {observation[:300]}...")
206
207            # Update history
208            history.append({
209                "thought": parsed.thought,
210                "action": f"{parsed.action_name}({parsed.action_params})",
211                "observation": observation,
212            })
213
214            # Check if done
215            if parsed.action_name == "finish":
216                return parsed.action_params.get("answer", observation)
217
218        return "Max steps reached"
219
220
221if __name__ == "__main__":
222    agent = ReActAgent(max_steps=10)
223
224    # Example task
225    result = agent.run(
226        "List the Python files in the current directory and "
227        "tell me how many there are."
228    )
229
230    print(f"\n{'='*50}")
231    print(f"Final Result: {result}")