从零构建ReAct Agent

说实话，当我第一次看到 ReAct 这个名词的时候，还以为是某个新出的前端框架。直到折腾了半天才发现，这玩意儿是解决 LLM “一本正经胡说八道” 的神器。

作为一个在 LLM 应用开发里踩过无数坑的人，我可以负责任地说：ReAct 可能是目前最实用的 Agent 设计模式之一。今天就从零开始，用 200 行左右的代码，带你完整实现一个能思考、能行动、会纠错的 ReAct Agent。

---

为什么需要 ReAct？LLM 的致命缺陷

先说说问题。

传统的 LLM 调用方式简单粗暴：输入 prompt，输出答案。这在写诗、翻译、闲聊这类任务上表现不错。但一旦涉及需要多步推理、实时数据、或工具调用的场景，LLM 就原形毕露了。

比如我问 GPT-4：“苹果公司现在的 CEO 是谁？他今年多大？”

如果知识库截止到训练日期，它可能会给我一个过时的答案。更糟的是，它可能会"幻觉"出一个看似合理的错误答案——比如告诉你乔布斯还在世。

这就是 LLM 的致命缺陷：无法自主获取最新信息，也无法在多步推理中自我纠错。

ReAct（Reasoning + Acting）就是为了解决这个问题而生的。

---

ReAct 是什么？一个直觉理解

ReAct = Reasoning（推理）+ Acting（行动）。

核心思想很简单：让 LLM 像人类一样思考——先想想，再行动，观察结果，再调整思路。

传统方式：

用户提问 → LLM 直接回答 → 可能答错

ReAct 方式：

用户提问 → LLM 思考："我需要什么信息？" → 调用工具 → 观察结果 → 
继续思考 → 再行动 → 直到得到答案 → 最终回答

这个过程形成了一个"思考-行动-观察"的循环，直到 LLM 觉得自己掌握了足够信息，才会给出最终答案。

---

ReAct 适合什么场景？

不是所有任务都需要 ReAct。它最适合以下场景：

1. 需要实时信息的任务

比如查询天气、股价、新闻。这些信息每天都在变，LLM 的训练数据无法覆盖。

2. 需要多步推理的复杂问题

比如"找出 2024 年诺贝尔奖得主中，年龄最大的那位的主要成就"。这需要：

• 先查 2024 年诺贝尔奖得主名单
• 然后查每个人的年龄
• 比较找出年龄最大的
• 最后查那位得主的主要成就

3. 需要调用外部工具的场景

比如计算复杂公式、执行代码、操作数据库。

4. 容错率低的任务

医疗诊断、法律咨询、金融分析——这些场景下，一个错误的答案可能代价高昂。ReAct 的自我纠错机制能大大降低出错概率。

---

完整代码实现：一个能用的 ReAct Agent

好了，理论说够了，直接上代码。

下面是完整的 ReAct Agent 实现，包含核心逻辑、工具定义和运行示例。代码基于 Python，不依赖任何 Agent 框架，你可以直接复制粘贴使用。

import json
import re
from typing import Dict, List, Callable, Any
from datetime import datetime

class ReActAgent:
    """
    ReAct Agent 核心实现
    支持多轮思考-行动-观察循环
    """
    
    def __init__(self, llm_client, tools: Dict[str, Callable], max_iterations: int = 10):
        """
        初始化 Agent
        
        Args:
            llm_client: LLM 客户端，需要有 generate() 方法
            tools: 工具函数字典，key 为工具名，value 为工具函数
            max_iterations: 最大迭代次数，防止死循环
        """
        self.llm = llm_client
        self.tools = tools
        self.max_iterations = max_iterations
        
    def run(self, query: str) -> str:
        """
        运行 Agent 处理用户查询
        
        Args:
            query: 用户问题
            
        Returns:
            最终答案
        """
        # 初始化思考轨迹
        trajectory = []
        
        # 构建系统提示词，告诉 LLM 怎么玩 ReAct
        system_prompt = self._build_system_prompt()
        
        # 构建初始用户消息
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Question: {query}\n\nLet's solve this step by step."}
        ]
        
        # 开始 ReAct 循环
        for i in range(self.max_iterations):
            # 调用 LLM 获取下一步思考
            response = self.llm.generate(messages)
            
            # 解析 LLM 的回复
            thought, action, action_input = self._parse_response(response)
            
            print(f"\n=== Step {i+1} ===")
            print(f"Thought: {thought}")
            print(f"Action: {action}")
            print(f"Action Input: {action_input}")
            
            # 记录思考过程
            trajectory.append({
                "thought": thought,
                "action": action,
                "action_input": action_input
            })
            
            # 检查是否得到了最终答案
            if action.lower() == "finish":
                print(f"\n✅ 任务完成！")
                return action_input
            
            # 执行工具调用
            if action in self.tools:
                try:
                    observation = self.tools[action](action_input)
                    print(f"Observation: {observation}")
                except Exception as e:
                    observation = f"Error: {str(e)}"
                    print(f"Observation: {observation}")
            else:
                observation = f"Error: Unknown action '{action}'. Available actions: {list(self.tools.keys())}"
                print(f"Observation: {observation}")
            
            # 更新消息历史，加入观察和下一步指示
            messages.append({"role": "assistant", "content": response})
            messages.append({
                "role": "user", 
                "content": f"Observation: {observation}\n\nContinue with the next thought and action."
            })
        
        # 超过最大迭代次数，返回已收集的信息
        return f"Reached maximum iterations. Last thought: {thought}"
    
    def _build_system_prompt(self) -> str:
        """
        构建系统提示词，定义 ReAct 的输出格式
        """
        tool_descriptions = "\n".join([
            f"- {name}: {func.__doc__ or 'No description'}" 
            for name, func in self.tools.items()
        ])
        
        return f"""You are a helpful AI assistant that solves problems step by step.
You have access to the following tools:

{tool_descriptions}

Use the following format:

Thought: [your reasoning about what to do next]
Action: [the tool name to use, or "finish" if you have the answer]
Action Input: [the input to the tool, or the final answer if finishing]

Rules:
1. Always start with "Thought:"
2. After thought, provide "Action:" and "Action Input:"
3. Use "finish" as the action when you have the complete answer
4. Be concise in your thoughts and actions
5. If a tool returns an error, try a different approach
"""
    
    def _parse_response(self, response: str) -> tuple:
        """
        解析 LLM 的回复，提取 thought, action, action_input
        """
        # 使用正则表达式提取各个部分
        thought_match = re.search(r'Thought:\s*(.+?)(?=\nAction:|$)', response, re.DOTALL)
        action_match = re.search(r'Action:\s*(.+?)(?=\nAction Input:|$)', response, re.DOTALL)
        input_match = re.search(r'Action Input:\s*(.+?)(?=\n|$)', response, re.DOTALL)
        
        thought = thought_match.group(1).strip() if thought_match else ""
        action = action_match.group(1).strip() if action_match else "finish"
        action_input = input_match.group(1).strip() if input_match else response.strip()
        
        return thought, action, action_input


# ==================== 工具函数定义 ====================

def search_web(query: str) -> str:
    """
    模拟网络搜索工具
    实际使用时可以替换为真实的搜索引擎 API
    """
    # 这里用简单的字典模拟搜索结果
    knowledge_base = {
        "tim cook age": "Tim Cook was born on November 1, 1960, making him 64 years old as of 2024.",
        "apple ceo": "Tim Cook has been the CEO of Apple Inc. since August 2011.",
        "python programming": "Python is a high-level programming language known for its simplicity.",
        "current date": datetime.now().strftime("%Y-%m-%d"),
    }
    
    query_lower = query.lower()
    for key, value in knowledge_base.items():
        if key in query_lower:
            return value
    
    return f"No specific information found for: {query}. Try a different search term."


def calculate(expression: str) -> str:
    """
    计算数学表达式
    安全地执行简单的数学计算
    """
    try:
        # 只允许数字和基本运算符，防止代码注入
        allowed_chars = set('0123456789+-*/.() ')
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression"
        
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"


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


# ==================== 模拟 LLM 客户端 ====================

class SimpleLLM:
    """
    简单的 LLM 客户端模拟
    实际使用时替换为 OpenAI、Claude 或其他 LLM API
    """
    
    def generate(self, messages: List[Dict]) -> str:
        """
        模拟 LLM 生成回复
        这里用简单的规则模拟 ReAct 行为
        """
        last_message = messages[-1]["content"]
        
        # 模拟 ReAct 推理过程
        if "apple" in last_message.lower() and "ceo" in last_message.lower():
            if "observation" not in last_message.lower():
                return """Thought: I need to find out who is the current CEO of Apple Inc.
Action: search_web
Action Input: Apple CEO current"""
            else:
                return """Thought: Now I know Tim Cook is the CEO. Let me find out his age.
Action: search_web
Action Input: Tim Cook age"""
        
        elif "tim cook" in last_message.lower() and "64" in last_message.lower():
            return """Thought: I now have all the information needed. Tim Cook is the CEO of Apple and he is 64 years old (born in 1960).
Action: finish
Action Input: Tim Cook is the current CEO of Apple Inc. He was born on November 1, 1960, making him 64 years old as of 2024. He has been leading Apple since August 2011, succeeding Steve Jobs."""
        
        elif "calculate" in last_message.lower() or any(op in last_message for op in ['+', '-', '*', '/']):
            # 提取数学表达式
            numbers = re.findall(r'\d+', last_message)
            if len(numbers) >= 2:
                return f"""Thought: The user wants me to calculate {numbers[0]} * {numbers[1]}.
Action: calculate
Action Input: {numbers[0]} * {numbers[1]}"""
        
        # 默认回复
        return """Thought: I have gathered enough information to provide an answer.
Action: finish
Action Input: Based on the available information, I can provide the following answer..."""


# ==================== 运行示例 ====================

if __name__ == "__main__":
    # 初始化工具
    tools = {
        "search_web": search_web,
        "calculate": calculate,
        "get_current_time": get_current_time,
    }
    
    # 初始化 LLM 客户端（实际使用时替换为真实的 API 客户端）
    llm = SimpleLLM()
    
    # 创建 Agent
    agent = ReActAgent(llm, tools, max_iterations=5)
    
    # 测试用例 1：需要搜索的多步问题
    print("=" * 50)
    print("测试用例 1：查询 Apple CEO 及其年龄")
    print("=" * 50)
    query1 = "Who is the current CEO of Apple and how old is he?"
    answer1 = agent.run(query1)
    print(f"\n最终答案: {answer1}")
    
    # 测试用例 2：数学计算
    print("\n" + "=" * 50)
    print("测试用例 2：数学计算")
    print("=" * 50)
    
    # 重置 Agent 状态
    agent2 = ReActAgent(llm, tools, max_iterations=5)
    query2 = "Calculate 123 * 456"
    answer2 = agent2.run(query2)
    print(f"\n最终答案: {answer2}")

---

代码解析：关键设计点

这段代码虽然不长，但包含了一个完整 ReAct Agent 的核心逻辑。让我解释几个关键设计点：

1. 严格的输出格式控制

通过系统提示词强制 LLM 按照固定格式输出：

Thought: ...
Action: ...
Action Input: ...

这是 ReAct 工作的基础。如果 LLM 不遵守这个格式，整个流程就会崩溃。实际项目中，你可能需要：

• 使用 few-shot prompting 给出示例
• 对不遵守格式的回复进行重试
• 使用更强大的模型（GPT-4、Claude 3 Opus 等）

2. 轨迹追踪

trajectory 列表记录了完整的"思考-行动-观察"链条。这不仅用于调试，还可以：

• 作为上下文喂给 LLM，让它了解之前做了什么
• 保存到数据库，用于后续分析
• 展示给用户，增加透明度

3. 工具解耦

工具函数完全独立于 Agent 核心逻辑。你可以轻松添加新工具：

def my_new_tool(input: str) -> str:
    """工具描述"""
    # 实现逻辑
    return result

tools["my_new_tool"] = my_new_tool

4. 安全考虑

calculate 函数里限制了允许的字符，防止代码注入攻击。这在生产环境是必须的——你永远不知道 LLM 会输出什么。

---

连接到真实 LLM：OpenAI 示例

上面的代码用了模拟的 LLM。要连接到真实的 OpenAI API，只需要简单的适配器：

from openai import OpenAI

class OpenAIAdapter:
    def __init__(self, api_key: str, model: str = "gpt-4"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    def generate(self, messages: List[Dict]) -> str:
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7,
            max_tokens=500
        )
        return response.choices[0].message.content

# 使用方式
llm = OpenAIAdapter(api_key="your-api-key")
agent = ReActAgent(llm, tools)
answer = agent.run("你的问题")

同样的模式也适用于 Claude、Gemini、或其他任何 LLM API。

---

生产环境的改进建议

上面的代码是教学用的简化版。实际部署时，你还需要考虑：

1. 错误处理

• 工具调用失败的重试机制
• LLM 输出格式错误的恢复策略
• 超时处理

2. 上下文管理

• 长对话的上下文压缩
• 关键信息的记忆机制
• 多轮对话的状态保持

3. 工具增强

• 工具参数的自动校验
• 工具调用的并发执行
• 工具返回结果的后处理

4. 可观测性

• 完整的执行日志
• 性能指标收集
• 用户反馈循环

---

写在最后

ReAct 并不是什么高深的技术，但它的实用价值很高。

在这个 LLM 应用爆炸的时代，单纯调用 API 已经不够了。用户要的不是"生成一段文字"，而是"解决一个实际问题"。ReAct 提供了一种结构化的方式，让 LLM 从"答题者"变成"问题解决者"。

我在这篇文章里提供的代码，你可以基于它：

• 添加更多工具，对接你的业务系统
• 接入真实的 LLM API
• 构建更复杂的 Agent 工作流
• 或者直接用于生产环境（记得加上错误处理和日志）