当前位置：首页 > news >正文

从零手写ReAct循环：深入理解AI Agent核心架构与实现原理

news 2026/8/12 19:24:53

1. 项目概述：为什么我们要亲手搭建一个 ReAct 循环？

如果你最近在关注 AI 应用开发，尤其是智能体（Agent）领域，那么“ReAct”这个词你一定不陌生。它频繁出现在各种框架的文档、技术博客和开源项目的 README 里。但很多时候，我们只是调用langchain或llama-index的某个AgentExecutor，传入一个LLM和一堆Tools，然后看着它神奇地工作。这就像开车，会踩油门和刹车就够了，但如果你想知道引擎盖下发生了什么，或者当车子抛锚时想自己动手修理，就必须理解它的工作原理。

“从零手写 ReAct 循环”这个项目，就是一次彻底的“引擎拆解”。ReAct，即Reasoning + Acting，是驱动现代 AI Agent 进行复杂任务规划与执行的核心范式。它让 Agent 具备了“思考-行动-观察-再思考”的心跳。这个项目的目的，不是教你如何使用一个现成的 Agent 框架，而是带你用最基础的代码，从第一行开始，构建出这个循环逻辑。你将亲手实现 Agent 如何解析用户指令、如何自主调用工具、如何处理工具返回结果、以及如何基于新观察进行下一轮推理。这个过程会让你对 Agent 的稳定性、可观测性和可调试性有全新的认识。无论你是想深入 Agent 底层机制的研究者，还是希望构建更可靠、更定制化智能体的开发者，这次“手写”之旅都将为你打下不可替代的坚实基础。

2. 核心架构设计：拆解 ReAct 循环的每一个齿轮

要手写一个 ReAct 循环，我们首先得在脑子里把它拆成零件。一个完整的 ReAct 循环，远不止一个while循环那么简单，它是一套精密的协作系统。

2.1 循环状态机：理解 Agent 的“心跳节拍”

ReAct 循环的本质是一个状态机。Agent 在任何时刻都处于一个明确的状态，并根据当前状态和输入，决定下一个状态和要执行的动作。一个典型的最小状态机包含以下几个核心状态：

初始/思考（INIT/THINK）：Agent 接收到用户任务（或上一轮的行动结果），进入思考状态。在此状态下，它的核心工作是分析当前局面，决定下一步是“继续推理”还是“采取行动”。如果是推理，就生成一段逻辑链；如果是行动，则必须精确地指定调用哪个工具以及传入什么参数。
行动（ACT）：Agent 从思考状态中明确发出了一个工具调用指令。此时，循环需要解析这个指令，找到对应的工具函数，并以正确的参数格式执行它。这是循环中唯一与外部世界（数据库、API、文件系统等）产生交互的环节。
观察（OBSERVE）：工具执行完毕，返回结果（或错误）。这个结果被作为新的“观察”输入，反馈给 Agent。观察的内容质量直接决定了下一轮思考的质量。
终止（FINISH）：当 Agent 认为任务已经完成（例如，给出了最终答案），或者触发了某些终止条件（如循环次数超限、用户中断），则跳出循环，返回最终结果。

这个状态流转（THINK -> ACT -> OBSERVE -> THINK...）就是 Agent 的“心跳”。手写循环的关键，就是清晰地定义这些状态，并实现状态之间的转换逻辑。一个常见的设计误区是让 LLM 一次性输出所有内容，这会导致解析困难和控制流混乱。正确的做法是，在每次调用 LLM 时，都明确约束其输出格式，使其只能输出符合当前状态预期的内容，例如在思考状态只输出“Thought:”和“Action:”开头的行。

2.2 工具系统设计：Agent 的“手脚”如何被调用

工具（Tools）是 Agent 延伸能力的载体。手写工具系统，你需要考虑以下几个层面：

工具抽象：每个工具应该是一个统一的调用接口。我通常会定义一个Tool基类或一个Tool协议，它至少包含name（工具名）、description（功能描述，用于提示词）、parameters（参数 JSON Schema）和_run（实际执行函数）这几个属性。
工具注册与发现：你需要一个中心化的注册表（如一个 Python 字典）来管理所有可用工具。当 Agent 决定调用工具时，循环逻辑需要能根据工具名称从这个注册表中快速检索到对应的工具对象。
参数验证与解析：LLM 输出的工具调用指令（通常是一个包含tool_name和tool_input的 JSON 字符串）需要被安全地解析。你必须对参数进行验证，确保其类型和结构符合工具定义的要求，防止注入攻击或运行时错误。这里可以借助 Pydantic 这样的库来简化验证逻辑。
异步与超时：考虑到工具调用可能涉及网络 I/O（如调用 API），你的工具系统最好支持异步操作。同时，必须为每个工具调用设置超时时间，避免一个缓慢的工具拖垮整个 Agent 循环。

注意：工具的描述（description）至关重要。它是 LLM 了解工具功能的唯一途径。描述应当清晰、具体，并最好包含示例。模糊的描述会导致 LLM 错误地调用或根本想不到调用这个工具。

2.3 提示工程：为 LLM 设定清晰的“思维轨道”

LLM 是循环中的“大脑”，但大脑需要引导。手写 ReAct 时，提示词（Prompt）就是引导 LLM 按照我们设计的 ReAct 格式进行输出的“轨道模板”。一个有效的 ReAct 提示通常包含：

系统指令（System Prompt）：定义 Agent 的角色、能力和必须遵守的规则。例如，“你是一个善于使用工具解决问题的助手。你必须通过 Thought、Action、Observation 的步骤来工作。”

格式说明（Format Instructions）：以清晰、无歧义的方式，规定 LLM 每一步应该输出的格式。这是最关键的部分。例如：

请严格按照以下格式响应： Thought: 你需要描述你当前的思考过程，分析问题和可用工具。 Action: 你需要调用的工具名称，必须是以下之一：[{tool_names}] Action Input: 调用该工具所需的输入，必须是一个有效的 JSON 字符串。 Observation: 工具返回的结果会放在这里。 ...（这个 Thought/Action/Action Input/Observation 循环可以重复多次） Final Answer: 当你认为已经完成任务时，用这个字段给出最终答案。

工具描述集成：将之前定义的所有工具的name和description动态插入到提示词中，让 LLM 知道它“手头”有哪些工具可用。
对话历史/上下文管理：提示词中需要包含之前的“Thought-Action-Observation”历史记录，这是 LLM 进行多步推理的上下文。你需要设计一个结构来维护和截断这个历史，防止超出模型的上下文长度限制。

手写提示词模板时，使用f-string或Jinja2模板进行变量替换是非常方便的做法。核心目标是让 LLM 的输出易于被后续代码解析。

3. 从零开始：一步步实现核心循环引擎

理论说得再多，不如一行代码。让我们抛开所有框架，用最纯粹的 Python 来构建这个循环。假设我们使用 OpenAI 的 GPT-4 作为 LLM 后端。

3.1 第一步：定义工具与工具管理器

首先，我们定义工具的协议和几个示例工具。

from typing import Any, Dict, Optional, Callable from pydantic import BaseModel, Field import json class ToolSchema(BaseModel): """工具的参数模式定义""" name: str description: str parameters: Dict[str, Any] # 可以是JSON Schema class Tool: """工具基类""" def __init__(self, name: str, description: str, func: Callable, args_schema: Optional[ToolSchema] = None): self.name = name self.description = description self.func = func self.args_schema = args_schema def run(self, tool_input: str) -> str: """执行工具，并返回字符串格式的结果""" try: # 解析输入，可能是JSON字符串 if tool_input.strip(): parsed_input = json.loads(tool_input) else: parsed_input = {} # 这里可以添加更复杂的参数验证，基于 args_schema result = self.func(**parsed_input) return str(result) except Exception as e: return f"Error: {str(e)}" class ToolRegistry: """工具注册表""" def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, tool: Tool): self._tools[tool.name] = tool def get_tool(self, name: str) -> Optional[Tool]: return self._tools.get(name) @property def tool_names(self) -> str: return ", ".join(self._tools.keys()) @property def tool_descriptions(self) -> str: desc = [] for name, tool in self._tools.items(): desc.append(f"{name}: {tool.description}") return "\n".join(desc) # 定义几个简单的工具函数 def search_web(query: str) -> str: """模拟网络搜索""" # 这里应该是真实的搜索API调用，如Serper或Google Search API return f"关于'{query}'的搜索结果：模拟数据..." def calculator(expression: str) -> str: """计算数学表达式""" try: # 警告：实际使用中，直接eval有安全风险，此处仅作演示 result = eval(expression) return str(result) except Exception as e: return f"计算错误: {e}" # 创建工具并注册 registry = ToolRegistry() registry.register(Tool(name="search", description="用于搜索最新信息。输入应为包含'query'键的JSON，如 {\"query\": \"问题\"}", func=search_web)) registry.register(Tool(name="calculator", description="用于计算数学表达式。输入应为包含'expression'键的JSON，如 {\"expression\": \"2+2\"}", func=calculator))

3.2 第二步：构建 ReAct 提示词模板

接下来，我们构建一个动态生成提示词的函数。

def build_react_prompt(task: str, tool_descriptions: str, history: str = "") -> str: """构建ReAct格式的提示词""" prompt_template = f""" 你是一个智能助手，可以通过使用工具来解决问题。请严格按照以下格式进行回应： 你可以使用的工具： {tool_descriptions} 格式说明： Thought: 首先，你需要分析当前情况，思考下一步该做什么。 Action: 你需要调用的工具名称，必须是从以下工具中选择：{registry.tool_names} Action Input: 调用该工具所需的输入，必须是一个有效的JSON字符串。 Observation: 工具执行后的结果会放在这里。 ...（这个 Thought/Action/Action Input/Observation 循环可以重复多次） 当你确信已经得到最终答案，或者不再需要工具时，请输出： Final Answer: [你的最终答案] 开始！ 任务：{task} {history} """ return prompt_template.strip()

3.3 第三步：实现核心循环引擎

现在，是时候编写循环的核心逻辑了。我们将实现一个状态机，解析 LLM 的响应，并驱动整个流程。

import re import asyncio from openai import AsyncOpenAI # 假设使用异步客户端 class ReActAgent: def __init__(self, llm_client: AsyncOpenAI, tool_registry: ToolRegistry, max_iterations: int = 10): self.llm = llm_client self.tools = tool_registry self.max_iterations = max_iterations # 用于解析LLM响应的正则表达式 self.thought_pattern = re.compile(r'Thought:\s*(.*?)(?=\nAction:|\nFinal Answer:|$)', re.DOTALL) self.action_pattern = re.compile(r'Action:\s*(\w+)') self.action_input_pattern = re.compile(r'Action Input:\s*(.*?)(?=\nObservation:|\nFinal Answer:|$)', re.DOTALL) self.final_answer_pattern = re.compile(r'Final Answer:\s*(.*)', re.DOTALL) async def run(self, task: str) -> str: """执行任务的主循环""" history = "" for i in range(self.max_iterations): print(f"\n--- 迭代 {i+1} ---") # 1. 生成思考与行动 prompt = build_react_prompt(task, self.tools.tool_descriptions, history) print(f"Prompt sent to LLM:\n{prompt[:500]}...") # 打印部分提示词用于调试 response = await self.llm.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=500 ) llm_output = response.choices[0].message.content print(f"LLM Raw Output:\n{llm_output}") # 2. 解析输出 thought = self._extract_pattern(self.thought_pattern, llm_output) action = self._extract_pattern(self.action_pattern, llm_output) action_input = self._extract_pattern(self.action_input_pattern, llm_output) final_answer = self._extract_pattern(self.final_answer_pattern, llm_output) # 3. 检查是否结束 if final_answer: print(f"\n✅ 任务完成！最终答案：{final_answer}") return final_answer # 4. 执行动作 if action and action_input: print(f"🤔 思考：{thought}") print(f"🔧 行动：调用工具 '{action}'，输入：{action_input}") tool = self.tools.get_tool(action) if tool: observation = tool.run(action_input) print(f"👀 观察：{observation}") # 更新历史，用于下一轮 history += f"\nThought: {thought}\nAction: {action}\nAction Input: {action_input}\nObservation: {observation}\n" else: observation = f"错误：未知工具 '{action}'。可用工具：{self.tools.tool_names}" print(f"⚠️ {observation}") history += f"\nThought: {thought}\nAction: {action}\nAction Input: {action_input}\nObservation: {observation}\n" else: # 如果LLM没有输出有效的Action，可能它还在“纯思考”，或者格式错误。 # 一种策略是将整个输出作为“思考”加入历史，让LLM在下一轮继续。 observation = f"格式错误或未指定行动。请确保严格按照 Thought/Action/Action Input 格式输出。你上次的输出是：{llm_output}" print(f"⚠️ {observation}") history += f"\nObservation: {observation}\n" # 循环超过最大次数 return f"达到最大迭代次数（{self.max_iterations}）仍未完成任务。最后的历史记录：{history}" def _extract_pattern(self, pattern, text): match = pattern.search(text) return match.group(1).strip() if match else None # 使用示例 async def main(): client = AsyncOpenAI(api_key="your-api-key") # 请替换为你的API Key agent = ReActAgent(client, registry, max_iterations=5) task = "请先搜索一下‘Python的最新版本号是多少’，然后用计算器算一下这个版本号加上3.14等于多少。" result = await agent.run(task) print(f"\n最终返回结果：{result}") if __name__ == "__main__": asyncio.run(main())

这段代码实现了一个最基础的、但完全可运行的 ReAct 循环。它清晰地展示了状态流转：构建提示 -> 调用 LLM -> 解析输出 -> 判断终止 -> 执行工具 -> 更新历史 -> 进入下一轮。

4. 核心环节的深度优化与实战技巧

上面的基础版本能跑起来，但离一个健壮、实用的 Agent 还有距离。接下来，我们深入几个核心环节，进行工业级的优化。

4.1 输出解析的鲁棒性增强

依赖正则表达式进行解析是脆弱的。LLM 的输出可能会有轻微的格式偏差（比如多一个空格，换行符不同）。更健壮的做法是：

结构化输出引导：利用现代 LLM 支持 JSON 格式输出的能力（如 OpenAI 的response_format={ "type": "json_object" }），直接要求 LLM 返回一个结构化的 JSON 对象，包含thought,action,action_input等字段。这能从根本上避免解析错误。
后备解析策略：如果使用文本格式，可以结合多种解析方法。例如，先尝试用正则，如果失败，可以尝试用关键词查找（如查找“Action:”的位置），或者甚至将解析失败的文本连同错误信息一起喂回给 LLM，让它自己纠正格式。
输出清洗：在解析前，对 LLM 的输出进行简单的清洗，如去除首尾空白、合并连续的空行等。

# 增强版解析函数示例 def parse_llm_output_robust(text: str) -> Dict[str, str]: """更鲁棒的解析函数，尝试多种方法""" result = {"thought": "", "action": "", "action_input": "", "final_answer": ""} # 方法1：尝试正则 patterns = {...} # 同上文的模式 for key, pattern in patterns.items(): match = pattern.search(text) if match: result[key] = match.group(1).strip() # 方法2：如果正则没找到关键字段，尝试基于关键词的简单分割 if not result["action"] and "Action:" in text: # 简单的行分割逻辑 lines = text.split('\n') for i, line in enumerate(lines): if line.startswith('Action:'): result['action'] = line.replace('Action:', '').strip() # 尝试找下一行的Action Input if i+1 < len(lines) and lines[i+1].startswith('Action Input:'): result['action_input'] = lines[i+1].replace('Action Input:', '').strip() break # 方法3：如果还是不行，可以返回一个特殊标记，让主循环处理 return result

4.2 工具调用的错误处理与超时控制

工具调用是循环中最可能出错的地方。我们必须做好防御。

异常捕获：每个工具的run方法都应该有完善的try...except，将任何异常转换为对 Agent 友好的字符串信息（如“调用搜索API时网络超时”），而不是让整个程序崩溃。
超时控制：使用asyncio.wait_for为工具调用设置超时。超时后，返回“工具调用超时”的观察。
输入验证：在tool.run()内部，使用args_schema（如果定义了）对输入参数进行严格的验证，类型不匹配或缺少必需参数时，直接返回错误信息，而不是尝试执行。

import asyncio from functools import partial async def run_tool_with_timeout(tool: Tool, tool_input: str, timeout: float = 30.0) -> str: """带超时和异常处理的工具运行""" try: # 将同步函数转换为异步执行 loop = asyncio.get_event_loop() # 注意：如果tool.func是CPU密集型，考虑用run_in_executor result = await asyncio.wait_for( loop.run_in_executor(None, partial(tool.run, tool_input)), timeout=timeout ) return result except asyncio.TimeoutError: return f"错误：调用工具 '{tool.name}' 超时（>{timeout}秒）。" except json.JSONDecodeError: return f"错误：工具输入不是有效的JSON格式：{tool_input}" except Exception as e: return f"错误：调用工具 '{tool.name}' 时发生异常：{str(e)}"

4.3 上下文管理与历史截断策略

随着循环进行，提示词会越来越长。我们必须管理上下文。

Token 计数：使用tiktoken等库估算每次添加历史记录后的总 token 数。
滑动窗口：当 token 数接近模型上限（如 GPT-4 的 8k 或 32k）时，采用滑动窗口策略。保留最重要的部分：最新的几次“Thought-Action-Observation”循环（因为它们包含最新进展）和最早的系统指令/任务描述。可以丢弃中间的一些旧循环。
总结压缩：一种更高级的策略是，当历史过长时，调用 LLM 本身对之前的对话历史进行总结，用一段简短的摘要替换掉大段旧历史。这需要额外的 LLM 调用，但能保留更多语义信息。
关键信息提取：对于某些任务，可以从历史观察中提取关键事实或数据，只将这些提取出的信息保留在上下文中，丢弃原始的冗长观察文本。

5. 常见问题排查与调试心法

手写 Agent 时，你会遇到各种诡异的问题。下面是我踩过坑后总结的排查清单。

5.1 Agent 陷入死循环或无效循环

症状：Agent 反复调用同一个工具，或者在不同工具间来回切换，始终无法输出Final Answer。
排查：
1. 检查工具描述：工具描述是否清晰、无歧义？LLM 是否真的理解每个工具的作用？尝试简化描述或添加更具体的例子。
2. 检查观察反馈：工具的返回结果（Observation）是否清晰、有用？一个返回“查询成功”但无实质内容的观察，无法推动 Agent 前进。确保工具返回的是信息量充足的数据。
3. 增强思考引导：在系统提示词中，更加强调“在给出 Final Answer 前，你必须确认问题已完全解决”。可以要求 LLM 在 Thought 部分更明确地评估当前进度。
4. 设置循环上限：就像我们代码中的max_iterations，这是最后的安全网。达到上限后，强制终止并返回当前历史，便于分析卡住的原因。

5.2 LLM 不按格式输出

症状：解析器无法提取出Action或Action Input，导致流程中断。
排查：
1. 强化格式指令：在提示词中用非常醒目的方式（如三个引号括起来的代码块）展示格式，并强调“必须严格遵守”。
2. 使用结构化输出：如前所述，切换到 JSON 格式输出，这是最彻底的解决方案。
3. 实施解析后备方案：像上面优化部分写的，当正则解析失败时，尝试其他方法提取信息，或者将错误信息反馈给 LLM 让其重试。
4. 检查 Temperature 参数：确保生成时temperature=0以获得最大确定性的输出。非零的 temperature 会增加格式错误的概率。

5.3 工具调用结果不佳导致决策错误

症状：Agent 基于错误的工具返回结果做出了错误的后续决策。
排查：
1. 工具结果验证：对于关键工具，可以在其返回结果后，添加一个验证步骤。例如，搜索工具返回空结果时，Observation 可以不是“[]”，而是“未找到相关信息，请尝试更换关键词”。
2. 让 LLM 评估观察：在提示词模板中，可以要求 LLM 在 Thought 部分对上一个 Observation 进行简短评估，如“这个结果是否回答了当前子问题？如果否，我们可能需要换种方式。”
3. 人工审核链路：在开发阶段，将每一步的 Thought、Action、Observation 都打印出来（就像我们示例代码中的print语句），这是调试的黄金手段。你能清晰地看到 Agent 的“思维链”在哪里断了。

5.4 性能与成本优化

问题：复杂任务循环次数多，LLM 调用次数也多，导致响应慢、成本高。
优化：
1. 选择合适模型：对于推理步骤，可以使用能力强的大模型（如 GPT-4）；对于简单的格式校验或文本提取，可以尝试小模型（如 GPT-3.5-Turbo），甚至用规则代替。
2. 并行化工具调用：如果 Agent 的思考步骤中涉及多个可以并行执行且互不依赖的工具调用，可以设计支持并行 Action 的格式，然后同时执行它们，最后合并 Observations。这能显著减少循环轮数。
3. 缓存：对于内容不变的查询（如“今天的日期”），工具结果可以进行缓存，避免重复计算或 API 调用。
4. 提前终止：如果某一步的 Observation 已经包含了明确的最终答案，可以设计逻辑让 Agent 提前结束，而不必非要输出“Final Answer”格式。

手写 ReAct 循环的过程，是一个不断与 LLM 的“非确定性”和现实世界的“复杂性”作斗争的过程。每一次调试，每一次对提示词的微调，每一次对工具返回格式的优化，都让你对 Agent 的内在机制理解更深一层。当你终于看到一个由自己亲手编写的 Agent 流畅地完成一个多步骤任务时，那种对系统全局的掌控感和成就感，是使用任何高级框架都无法替代的。这，就是“从零手写”的价值所在。

查看全文

http://www.cnnetsun.cn/news/3986551.html