从零实现ReAct Agent：手写思考-行动循环的智能体

Simon Chen

2026-06-15 3503 words 7 minutes

Contents

什么是ReAct Agent？

大多数现代AI智能体（ChatGPT、Claude、Copilot等）底层都运行着一个核心模式：思考→行动→观察（Thought → Action → Observation）。这就是2022年Yao等人提出的ReAct框架。

与传统的"一次性回答"不同，ReAct Agent会：

思考：分析当前状态，决定下一步做什么
行动：调用外部工具（搜索、计算、读文件等）
观察：获取工具返回的结果
循环：基于观察继续思考，直到任务完成

这篇文章不依赖LangChain、AutoGen等框架，用纯Python实现一个完整的ReAct Agent。

Agent核心架构

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


┌─────────────────────────────────────┐
│            ReAct Agent              │
│                                     │
│  ┌──────────┐    ┌──────────────┐  │
│  │  LLM 大脑 │◄──►│   工具注册表  │  │
│  └────┬─────┘    └──────────────┘  │
│       │                             │
│  ┌────▼─────┐                      │
│  │ 记忆系统  │                      │
│  └──────────┘                      │
└─────────────────────────────────────┘
        │
   ┌────▼────┐
   │ 外部工具 │
   └─────────┘

完整实现

第一步：定义工具系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57


import json
import math
import datetime
from typing import Callable

class Tool:
    """工具基类：每个工具都有名称、描述和执行函数"""
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func

    def execute(self, **kwargs) -> str:
        try:
            result = self.func(**kwargs)
            return str(result)
        except Exception as e:
            return f"工具执行失败: {e}"

    def to_schema(self) -> dict:
        """生成工具的JSON Schema，供LLM理解"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self._infer_params()
        }

    def _infer_params(self) -> dict:
        """从函数签名推断参数"""
        import inspect
        sig = inspect.signature(self.func)
        params = {}
        for name, param in sig.parameters.items():
            params[name] = {"type": "string", "description": f"参数 {name}"}
        return params


class ToolRegistry:
    """工具注册表：管理所有可用工具"""
    def __init__(self):
        self.tools: dict[str, Tool] = {}

    def register(self, name: str, description: str, func: Callable):
        self.tools[name] = Tool(name, description, func)

    def get_tool(self, name: str) -> Tool | None:
        return self.tools.get(name)

    def list_tools(self) -> str:
        """生成工具列表文本，注入到prompt中"""
        lines = []
        for tool in self.tools.values():
            lines.append(f"- {tool.name}: {tool.description}")
        return "\n".join(lines)

    def get_schemas(self) -> list[dict]:
        return [t.to_schema() for t in self.tools.values()]

第二步：注册实用工具

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39


registry = ToolRegistry()

# 计算器
def calculator(expression: str) -> str:
    """安全的数学表达式计算"""
    allowed = set("0123456789+-*/().% ")
    if not all(c in allowed for c in expression):
        return "错误: 不允许的字符"
    try:
        result = eval(expression, {"__builtins__": {}}, {"math": math})
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

registry.register("calculator", "计算数学表达式，如 '2 + 3 * 4'", calculator)

# 当前时间
def get_current_time() -> str:
    now = datetime.datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")

registry.register("get_current_time", "获取当前日期和时间", get_current_time)

# 网页搜索（模拟）
def web_search(query: str) -> str:
    """模拟网络搜索 - 实际项目中接入搜索API"""
    return f"搜索结果: 关于'{query}'的信息...\n1. 相关文章A\n2. 相关文章B"

registry.register("web_search", "搜索互联网获取最新信息", web_search)

# 文件读取
def read_file(file_path: str) -> str:
    try:
        with open(file_path, 'r') as f:
            return f.read(2000)
    except FileNotFoundError:
        return f"文件不存在: {file_path}"

registry.register("read_file", "读取本地文件内容", read_file)

第三步：实现ReAct循环

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


from openai import OpenAI

class ReActAgent:
    SYSTEM_PROMPT = """你是一个智能助手，可以使用工具解决问题。

## 工具列表
{tools}

## 工作方式
每次回复必须严格遵循以下JSON格式之一：

如果需要使用工具：
```json
{{"thought": "你的思考过程", "action": "工具名称", "action_input": {{"参数名": "参数值"}}}}

如果任务完成，给出最终答案：

1

{{"thought": "总结思考", "action": "finish", "final_answer": "最终答案"}}

规则

每次只执行一个动作
必须先思考再行动
基于观察结果继续思考
最多循环10次，必须给出最终答案 """

def init(self, model: str = “gpt-4”, api_key: str = None, base_url: str = None, max_iterations: int = 10): self.client = OpenAI(api_key=api_key, base_url=base_url) self.model = model self.registry = ToolRegistry() self.max_iterations = max_iterations self.history: list[dict] = []

def register_tool(self, name: str, description: str, func: Callable): self.registry.register(name, description, func)

def _build_system_prompt(self) -> str: return self.SYSTEM_PROMPT.format(tools=self.registry.list_tools())

def _call_llm(self, messages: list[dict]) -> str: “““调用LLM获取响应””” response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=0, ) return response.choices[0].message.content

def _parse_action(self, text: str) -> dict: “““从LLM输出中解析JSON动作””” # 尝试提取JSON块 import re json_match = re.search(r’json\s*(.*?)\s*’, text, re.DOTALL) if json_match: return json.loads(json_match.group(1)) # 尝试直接解析 try: return json.loads(text) except: return {“action”: “finish”, “final_answer”: text}

def run(self, task: str) -> str: “““执行任务的主循环””” system_prompt = self._build_system_prompt() messages = [{“role”: “system”, “content”: system_prompt}] messages.append({“role”: “user”, “content”: f"任务: {task}"})
```
 for i in range(self.max_iterations):
     print(f"\n--- 循环 {i+1}/{self.max_iterations} ---")

     # 1. 调用LLM思考
     response = self._call_llm(messages)
     print(f"LLM: {response[:200]}...")

     # 2. 解析动作
     action = self._parse_action(response)
     print(f"动作: {action.get('action')}")

     # 3. 如果是finish，返回结果
     if action.get("action") == "finish":
         return action.get("final_answer", "任务完成")

     # 4. 执行工具
     tool_name = action.get("action")
     tool_input = action.get("action_input", {})
     tool = self.registry.get_tool(tool_name)

     if tool:
         observation = tool.execute(**tool_input)
     else:
         observation = f"工具 '{tool_name}' 不存在"

     print(f"观察: {observation[:200]}")

     # 5. 将对话历史加入messages
     messages.append({"role": "assistant", "content": response})
     messages.append({
         "role": "user",
         "content": f"工具返回结果:\n{observation}\n\n请继续思考和行动。"
     })

 return "达到最大循环次数，任务未完成"
```

使用示例

if name == “main”: agent = ReActAgent( model=“gpt-4”, api_key=“your-api-key” )

# 注册工具
agent.register_tool("calculator", "计算数学表达式", calculator)
agent.register_tool("get_current_time", "获取当前时间", get_current_time)
agent.register_tool("web_search", "搜索互联网", web_search)

# 运行任务
result = agent.run("现在几点了？帮我算一下从现在到2027年元旦还有多少天")
print(f"\n最终结果: {result}")

1
2

## 运行效果

— 循环 1/10 — LLM: {“thought”: “用户想知道当前时间和到2027年元旦的天数差…”, “action”: “get_current_time”, “action_input”: {}} 动作: get_current_time 观察: 2026-06-15 14:30:22

— 循环 2/10 — LLM: {“thought”: “现在是2026-06-15，我需要计算到2027-01-01的天数…”, “action”: “calculator”, “action_input”: {“expression”: “200”}} 动作: calculator 观察: 200 = 200

最终结果: 现在是2026年6月15日14:30，距离2027年元旦还有约200天。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46



## 进阶：添加记忆系统

一个真正的Agent需要记住之前的对话和学到的知识：

```python
from dataclasses import dataclass, field

@dataclass
class AgentMemory:
    """Agent的记忆系统"""
    short_term: list[dict] = field(default_factory=list)  # 当前对话
    long_term: dict = field(default_factory=dict)          # 持久记忆
    facts: list[str] = field(default_factory=list)         # 提取的事实

    def add_interaction(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})

    def extract_facts(self, llm_client, model: str):
        """用LLM从对话中提取关键事实"""
        if len(self.short_term) < 3:
            return
        conversation = "\n".join(
            f"{m['role']}: {m['content'][:200]}" for m in self.short_term[-10:]
        )
        prompt = f"""从以下对话中提取关于用户的关键事实，每行一个：
{conversation}
"""
        response = llm_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0
        )
        new_facts = [f.strip() for f in response.choices[0].message.content.split("\n") if f.strip()]
        self.facts.extend(new_facts)

    def get_context(self) -> str:
        """获取记忆上下文"""
        parts = []
        if self.facts:
            parts.append("已知事实:\n" + "\n".join(f"- {f}" for f in self.facts[-20:]))
        if self.short_term:
            parts.append("近期对话:\n" + "\n".join(
                f"{m['role']}: {m['content'][:100]}" for m in self.short_term[-6:]
            ))
        return "\n\n".join(parts) if parts else "暂无记忆"

踩坑记录

1. JSON解析失败

LLM输出的JSON格式不稳定，经常有多余的换行或注释。解决方案是用正则提取JSON块，而非直接json.loads。

2. 工具幻觉

LLM可能编造不存在的工具名。在执行前必须校验工具是否存在，并将"可用工具列表"注入system prompt。

3. 无限循环

某些任务LLM会反复调用同一个工具但得不到想要的结果。需要设置最大循环次数（建议10-15次）。

4. 成本控制

每轮循环都会调用一次LLM，token消耗是普通对话的5-10倍。生产环境建议用gpt-5.4-mini或claude-haiku控制成本。

总结

ReAct Agent的核心就是思考→行动→观察的循环。理解了这个模式，你就能看懂LangChain、AutoGen、CrewAI等框架的底层逻辑——它们本质上都是在帮你管理这个循环，加上了工具调用、错误处理、并行执行等工程化能力。

下次使用Agent框架时，不妨想想：框架帮我省了什么？哪些是它做不到、需要我自己处理的？这就是从"会用框架"到"理解原理"的跨越。