从ReAct论文到代码：48小时搭建可运行的Agent推理循环——理解论文、拆解机制、落到实现

从ReAct论文到代码48小时搭建可运行的Agent推理循环——理解论文、拆解机制、落到实现一、ReAct论文解决的核心问题为什么LLM需要边想边做大语言模型LLM在纯推理模式下有一个根本性缺陷推理与行动脱节。模型在内部思考后直接输出结论但思考过程中无法获取外部信息、无法验证假设、无法修正错误判断。这导致两个典型问题一是幻觉——模型基于不完整信息做出错误推理却无法在推理中查证二是无效行动——模型决定调用工具但未充分思考调用目的导致工具结果无法被有效利用。ReActReasoningActing论文的核心贡献是将推理和行动交织在同一个循环中让模型在每一步行动前先思考行动的目的在每一步行动后反思结果对推理的影响。sequenceDiagram participant U as 用户 participant A as Agent推理循环 participant T as 工具层 participant E as 外部环境 U-A: 输入问题 A-A: Thought1: 分析问题需要什么信息 A-T: Action1: 搜索相关知识 T-E: 执行搜索请求 E-T: 返回搜索结果 T-A: Observation1: 搜索结果内容 A-A: Thought2: 结果部分回答问题还需验证 A-T: Action2: 查询数据库验证 T-E: 执行数据库查询 E-T: 返回查询结果 T-A: Observation2: 验证数据 A-A: Thought3: 信息充分可以回答 A-U: 输出最终答案上图展示ReAct的核心循环Thought思考→Action行动→Observation观察交替进行直到Thought判定信息充分后输出最终答案。与纯推理模式相比ReAct在推理过程中引入了外部信息获取能力与纯行动模式相比ReAct在每次行动前增加了目的性思考。二、ReAct机制的拆解Thought-Action-Observation三元组的循环逻辑论文核心机制ReAct论文定义了三种类型的推理步骤Thought模型对当前状态的分析和对下一步行动的规划。Thought不是自由联想而是有目标的推理当前信息是否充分如果不充分需要什么类型的信息哪种工具能获取这些信息Action模型选择并调用一个工具。Action必须与前置Thought的逻辑一致——如果Thought判定需要搜索知识则Action必须是搜索工具调用而非随机选择。Observation工具执行返回的结果。Observation注入到后续Thought中成为新的推理基础。Observation的格式化至关重要原始返回数据需压缩为模型可理解的摘要避免长文本干扰推理。循环终止条件ReAct循环何时终止论文提出两种终止方式一是Thought判定信息已充分可以给出最终答案二是达到最大步数限制论文实验中设为6步。生产环境中需要增加第三种token消耗超过预算阈值防止成本失控。Action空间设计论文实验中的Action空间包括Search[query]搜索相关知识、Lookup[keyword]在搜索结果中查找关键词、Finish[answer]输出最终答案并终止循环。生产环境中Action空间需要根据业务需求扩展但必须保持有限且明确——无限的Action空间会让模型产生不确定的行为选择。三、48小时搭建可运行ReAct推理循环的代码实现第8-16小时核心循环引擎import json import re import uuid from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any, Callable, Optional class StepType(Enum): THOUGHT thought ACTION action OBSERVATION observation FINISH finish dataclass class ReActStep: ReAct推理步骤Thought/Action/Observation三元组 step_type: StepType content: str # Thought内容或Action指令或Observation数据 tool_name: Optional[str] None # Action步骤的工具名 tool_args: Optional[dict] None # Action步骤的工具参数 timestamp: datetime field(default_factorydatetime.now) dataclass class ReActState: ReAct推理循环状态步骤历史与上下文管理 session_id: str field(default_factorylambda: str(uuid.uuid4())) question: str # 用户原始问题 steps: list[ReActStep] field(default_factorylist) max_steps: int 8 # 最大推理步数兜底保护 token_budget: int 4000 # token消耗预算防止成本失控 tokens_used: int 0 class ReActEngine: ReAct推理循环引擎实现Thought-Action-Observation交替循环 def __init__( self, llm_call: Callable, # LLM调用函数 tools: dict[str, Callable], # 注册的工具集 max_steps: int 8, token_budget: int 4000, ): self.llm_call llm_call self.tools tools self.max_steps max_steps self.token_budget token_budget def _build_prompt(self, state: ReActState) - str: 构建ReAct提示词将步骤历史格式化为LLM可理解的上下文 # ReAct的Prompt模板明确告知模型输出格式约束 prompt_parts [ f问题: {state.question}\n\n, 你必须按以下格式交替思考和行动:\n, Thought: [你对当前状态的分析和下一步规划]\n, Action: [工具调用格式为 tool_name(arg1val1, arg2val2)]\n, Observation: [工具返回的结果]\n, ... (重复Thought-Action-Observation直到信息充分)\n, Thought: [信息充分我可以给出最终答案]\n, Action: finish(最终答案内容)\n\n, ] # 将已完成的步骤注入Prompt作为推理上下文 for step in state.steps: if step.step_type StepType.THOUGHT: prompt_parts.append(fThought: {step.content}\n) elif step.step_type StepType.ACTION: if step.tool_name finish: prompt_parts.append(fAction: finish({step.content})\n) else: args_str json.dumps(step.tool_args or {}) prompt_parts.append( fAction: {step.tool_name}({args_str})\n ) elif step.step_type StepType.OBSERVATION: prompt_parts.append(fObservation: {step.content}\n) # 追加思考提示引导模型进入下一个Thought prompt_parts.append(Thought:) return .join(prompt_parts) def _parse_llm_output(self, raw_output: str) - dict: 解析LLM输出提取Thought和Action内容 result {thought: , action: None} # 提取Thought部分匹配Thought: xxx格式 thought_match re.search( rThought:\s*(.?)(?:\n|Action:), raw_output, re.DOTALL ) if thought_match: result[thought] thought_match.group(1).strip() # 提取Action部分匹配Action: tool_name(args)或Action: finish(answer) action_match re.search(rAction:\s*(.), raw_output) if action_match: action_str action_match.group(1).strip() # 解析finish动作直接提取最终答案 if action_str.startswith(finish): finish_match re.search(rfinish\((.)\), action_str) answer finish_match.group(1) if finish_match else action_str result[action] { tool: finish, args: {answer: answer.strip()}, } else: # 解析工具调用提取工具名和参数 tool_match re.search( r(\w)\((.)\), action_str ) if tool_match: tool_name tool_match.group(1) try: args json.loads(tool_match.group(2)) except json.JSONDecodeError: # JSON解析失败退化为单参数处理 args {query: tool_match.group(2)} result[action] { tool: tool_name, args: args, } return result async def run(self, question: str) - dict: 运行完整ReAct推理循环直到finish或达到终止条件 state ReActState(questionquestion) state.steps.append(ReActStep( step_typeStepType.THOUGHT, contentf我需要回答问题: {question}, )) for i in range(state.max_steps): # 检查token预算超出则强制终止 if state.tokens_used state.token_budget: return { answer: 推理资源耗尽请简化问题, steps: state.steps, reason: token_budget_exceeded, } # 构建Prompt并调用LLM prompt self._build_prompt(state) llm_output await self.llm_call(prompt) state.tokens_used len(llm_output) // 4 # 粗略token估算 # 解析LLM输出 parsed self._parse_llm_output(llm_output) # 记录Thought步骤 if parsed[thought]: state.steps.append(ReActStep( step_typeStepType.THOUGHT, contentparsed[thought], )) # 处理Action步骤 action parsed.get(action) if not action: # LLM未输出有效Action引导其重新思考 state.steps.append(ReActStep( step_typeStepType.OBSERVATION, content未检测到有效行动指令请重新思考并选择工具。, )) continue # finish动作输出最终答案并终止循环 if action[tool] finish: answer action[args].get(answer, ) state.steps.append(ReActStep( step_typeStepType.FINISH, contentanswer, )) return { answer: answer, steps: state.steps, reason: finished, } # 工具调用执行工具并记录Observation tool_name action[tool] tool_args action[args] tool_func self.tools.get(tool_name) if not tool_func: # 工具未注册告知模型选择可用工具 available list(self.tools.keys()) state.steps.append(ReActStep( step_typeStepType.OBSERVATION, content( f工具 {tool_name} 不存在。 f可用工具: {available} ), )) continue try: result await tool_func(**tool_args) # 工具结果格式化截断过长内容保留关键信息 observation self._format_observation(result) state.steps.append(ReActStep( step_typeStepType.OBSERVATION, contentobservation, )) except Exception as e: # 工具执行失败告知模型错误信息让其调整策略 state.steps.append(ReActStep( step_typeStepType.OBSERVATION, contentf工具 {tool_name} 执行失败: {str(e)}, )) # 达到最大步数仍未finish返回最后一步的推理内容 last_thought for step in reversed(state.steps): if step.step_type StepType.THOUGHT: last_thought step.content break return { answer: f推理步数耗尽。当前进展: {last_thought}, steps: state.steps, reason: max_steps_exceeded, } def _format_observation(self, result: Any) - str: 格式化工具返回结果截断过长内容避免干扰推理 if isinstance(result, str): # 字符串结果超过500字截断保留关键信息 if len(result) 500: return result[:500] ...(内容过长已截断) return result elif isinstance(result, dict): # 结构化结果提取关键字段而非输出完整JSON key_fields [title, summary, content, answer, result] summary_parts [] for k in key_fields: if k in result: val str(result[k]) if len(val) 200: val val[:200] ... summary_parts.append(f{k}: {val}) if summary_parts: return ; .join(summary_parts) # 无关键字段输出完整JSON的截断版本 json_str json.dumps(result, ensure_asciiFalse) if len(json_str) 300: return json_str[:300] ...(已截断) return json_str else: return str(result)[:300]第16-24小时工具注册与测试用例# 模拟搜索工具用于知识检索场景 async def search_tool(query: str) - dict: 搜索工具返回与查询相关的知识摘要 # 生产环境中替换为真实的搜索API调用 mock_results { Python异步编程: { title: Python asyncio编程指南, summary: asyncio是Python标准库的异步IO框架基于事件循环机制, content: 核心概念包括coroutine、task、future和事件循环..., }, ReAct论文: { title: ReAct: Synergizing Reasoning and Acting, summary: ReAct将推理和行动交织在LLM的同一个循环中, content: 论文发表于2023年核心机制是Thought-Action-Observation三元组..., }, } for key, value in mock_results.items(): if key.lower() in query.lower(): return value return {summary: f未找到与 {query} 直接相关的结果, content: } # 模拟数据库查询工具用于数据验证场景 async def query_tool(table: str, conditions: dict) - list: 数据库查询工具返回满足条件的数据记录 # 生产环境中替换为真实数据库查询 return [{id: 1, name: 示例数据, value: 100}] # 注册工具集 tools { search: search_tool, query: query_tool, } # 模拟LLM调用生产环境替换为真实LLM API async def mock_llm_call(prompt: str) - str: 模拟LLM输出返回符合ReAct格式的响应 # 此函数在48小时搭建中用于验证循环逻辑 # 生产环境替换为OpenAI/Anthropic等API调用 if 搜索 in prompt.lower() or search in prompt.lower(): return ( Thought: 我需要搜索相关知识来回答这个问题\n Action: search(queryPython异步编程) ) elif Observation in prompt: return ( Thought: 搜索结果提供了部分信息我可以据此回答\n Action: finish(基于搜索结果答案是...) ) return ( Thought: 我还不确定如何回答需要更多信息\n Action: search(query相关信息) ) # 创建引擎并运行测试 engine ReActEngine(llm_callmock_llm_call, toolstools) result await engine.run(什么是Python的异步编程) print(f答案: {result[answer]}) print(f推理步数: {len(result[steps])})四、ReAct实现中的关键权衡与边界分析LLM输出解析的鲁棒性问题ReAct实现中最大的工程挑战是LLM输出解析。模型不总是严格遵循Thought: xxx\nAction: xxx格式。可能出现格式错误、Action与Thought混在一段话中、使用了未注册的工具名。解析器的鲁棒性决定了整个循环的稳定性。flowchart TD subgraph 解析策略三层防护 P1[第一层: 正则表达式严格匹配] -- P1a[格式正确时直接提取] P2[第二层: 关键词模糊匹配] -- P2a[格式偏差时按关键词定位] P3[第三层: 整体回退引导] -- P3a[无法解析时告知模型重新输出] end subgraph 格式偏差的常见类型 E1[Thought和Action写在同一行] E2[使用非标准Action格式: 如直接写工具名而非tool_nameargs] E3[输出额外解释文本干扰格式识别] E4[未输出Action只有Thought] end P1 -- E1 -- P2 P1 -- E2 -- P2 P2 -- E3 -- P3 P2 -- E4 -- P3Observation格式化的长度控制工具返回结果可能很长搜索API返回完整网页内容。过长Observation会占用LLM上下文窗口减少可用推理步数。格式化策略是结构化结果提取关键字段文本结果截断到500字以内。截断不是丢弃而是将完整数据存储在AgentState中供后续引用仅向LLM展示摘要。Action空间的有限性约束Action空间必须有限且明确。无限的工具列表会让模型在选择时产生犹豫和错误。创业团队建议从3-5个核心工具起步搜索、查询、计算、文件操作、finish。待业务需求扩展后再增加工具但每个新工具必须在Prompt中明确描述其用途和调用格式。适用与禁用场景适用需要多步推理外部信息获取的问答场景、知识密集型任务需要搜索验证的推理、决策型任务需要数据查询支撑的判断、创业团队快速验证Agent可行性。禁用单步直接回答的简单问答ReAct循环开销不值得、纯文本生成任务无需工具调用、对延迟极度敏感的实时场景多步循环增加2-5倍延迟、工具种类超过15个的复杂场景Action选择准确率随工具数增加而下降。五、总结从ReAct论文到可运行代码的核心路径是理解Thought-Action-Observation三元组的循环逻辑拆解循环终止条件与Action空间设计实现循环引擎的三层防护LLM输出解析鲁棒性、Observation格式化长度控制、token预算与步数双重兜底。48小时搭建的关键分配前8小时理解论文机制与设计数据模型8-24小时实现核心循环引擎与工具注册24-48小时测试验证与解析鲁棒性修复。ReAct不是万能框架它最适合需要多步推理外部信息获取的场景。单步问答和纯文本生成不需要ReAct的循环开销。创业团队验证Agent可行性时ReAct是最小可用的起点3-5个工具、8步推理上限、4000token预算即可搭建一个可运行的Agent推理循环。