尧图网站建设 
1. 项目概述为什么我们要“手搓”一个ReAct循环如果你最近在关注AI应用开发尤其是智能体Agent领域那么“ReAct”这个词一定高频出现在你的视野里。它不是什么新出的前端框架而是驱动现代AI智能体进行“思考”和“行动”的核心范式。简单来说ReAct就是Reason推理 Act行动的循环。一个智能体接收到任务后不是直接给出最终答案而是先“想一想”Reason决定下一步该做什么比如调用某个工具、查询知识库然后执行这个动作Act根据执行结果再进入下一轮的“思考”如此循环直到任务完成。你可能会问现在LangChain、LlamaIndex这些框架不是已经把ReAct封装好了吗直接用不就行了没错用框架可以快速搭建原型。但这就好比学开车你只知道踩油门和刹车却不了解发动机和变速箱是怎么协同工作的。当你的Agent在复杂任务中“卡壳”、陷入死循环或者做出匪夷所思的决策时如果你对底层机制一无所知调试将变得异常痛苦。“从零手写”的价值就在于此。这不是为了重复造轮子而是为了亲手摸清这个“轮子”的每一个辐条是如何咬合的。通过剥离框架的抽象层我们能够透彻理解核心机制真正搞懂“思考-行动”循环是如何流转的状态如何管理工具如何被选择和调用。获得终极的调试能力当智能体行为异常时你能像外科医生一样精准定位是推理环节的提示词出了问题还是工具返回的结果解析有误或是状态管理出现了混乱。实现高度定制化框架提供的是通用解决方案。当你需要为特定领域如金融分析、游戏NPC、自动化运维设计独特的推理逻辑或行动策略时手写的架构给你最大的自由度。所以这个项目标题“从零手写 ReAct 循环”瞄准的正是那些不满足于当“调包侠”渴望掌握智能体最核心“心跳”机制的开发者。我们将一起用代码模拟出这个让AI能够自主规划、使用工具、逐步逼近目标的思考过程。2. ReAct循环的核心原理与架构设计在动手写代码之前我们必须像建筑师审视蓝图一样把ReAct循环的每一个组件和它们之间的数据流彻底想清楚。一个最简化的ReAct循环可以抽象为以下几个核心部分2.1 核心组件拆解智能体Agent这是循环的“大脑”。它持有当前的任务目标、与环境的交互历史记忆以及可供使用的工具列表。它的核心职责是发起“推理”。推理器Reasoner通常是基于大语言模型LLM的模块。它接收当前的“状态”包括任务描述、历史记录、可用工具然后输出一个结构化的“思考”和“决策”。这个决策通常表现为“我需要先做A那么我就调用工具X。”动作执行器Actuator负责执行推理器做出的决策。如果决策是调用工具它就找到对应的工具函数传入参数并执行然后将执行结果收集回来。观察与状态更新器Observer将动作执行的结果成功或失败附带返回数据进行格式化形成一条新的“观察”记录并将其追加到交互历史中更新整个循环的状态。循环控制器Loop Controller判断循环是否应该继续。判断条件可以是任务是否已达成如答案已找到、是否陷入了死循环如重复调用同一工具、是否超过了最大步数限制。2.2 数据流心跳的节拍一次完整的心跳循环迭代遵循以下节拍状态输入将当前任务描述、完整的历史交互记录格式为Thought: ... Action: ... Observation: ...、可用工具描述组合成一个提示Prompt输入给推理器LLM。推理与决策LLM根据提示生成一段文本。我们需要用解析器Parser从这段文本中提取出关键结构thought本次推理过程、action要执行的动作名称、action_input动作的输入参数。动作执行根据解析出的action在工具注册表中找到对应的函数将action_input作为参数传入并执行。观察生成捕获工具函数的执行结果或异常将其格式化为一个observation字符串。历史更新将本次循环产生的thought,action,observation三元组追加到历史记录列表中。这就构成了智能体的“短期记忆”。循环判断检查observation是否已经包含了最终答案或者是否触发了终止条件如最大步数。如果是则跳出循环返回结果如果不是则回到第1步。注意这里有一个关键设计点。我们要求LLM输出的格式必须是可解析的比如固定输出“Thought: ... Action: ... Action Input: ...”。这是ReAct能稳定运行的前提也是手写时需要精心设计提示词和解析逻辑的地方。2.3 与框架的对比我们手写的重点像LangChain的Agent它把这些组件都封装好了提供了AgentExecutor来驱动循环。我们手写就是要实现一个自己的、轻量级的AgentExecutor。我们的关注点将集中在提示工程如何构建最有效的提示让LLM稳定地输出我们想要的格式。解析逻辑如何鲁棒地从LLM的非确定性输出中准确提取出动作指令。工具管理如何优雅地注册、描述和调用工具函数。状态管理如何设计历史记录的数据结构使其既能包含完整信息又不会因为过长而影响后续推理未来可以考虑摘要或向量化存储但本项目先从简单列表开始。错误处理与循环终止当工具调用失败、LLM输出无法解析时系统该如何降级处理如何设计合理的终止条件避免无限循环3. 从零开始构建手写ReAct循环的代码骨架理论清晰了现在打开你的代码编辑器我们开始一行行构建这个循环。我们将使用Python作为实现语言并假设你已经配置好了OpenAI或类似LLM的API环境。3.1 环境准备与基础定义首先定义最基础的数据结构。我们将使用Pydantic来创建数据模型这能让我们的代码更清晰、类型更安全。from typing import Any, Callable, Dict, List, Optional, Tuple from pydantic import BaseModel, Field import openai # 或其他LLM客户端 # 定义工具一个可调用函数及其描述 class Tool(BaseModel): name: str func: Callable[[str], str] # 简化起见输入输出均为字符串 description: str # 这个描述至关重要LLM靠它来决定是否使用该工具 # 定义单步交互记录 class StepRecord(BaseModel): thought: str action: str action_input: str observation: str # 定义智能体状态 class AgentState(BaseModel): task: str history: List[StepRecord] Field(default_factorylist) max_steps: int 10 current_step: int 03.2 核心引擎ReAct循环控制器接下来我们实现循环控制器也就是整个系统的心脏。class ReActEngine: def __init__(self, llm_client, tools: List[Tool]): self.llm llm_client self.tools {tool.name: tool for tool in tools} # 工具字典便于按名查找 def run(self, task: str, max_steps: int 10) - str: 执行ReAct循环的主函数 state AgentState(tasktask, max_stepsmax_steps) while state.current_step state.max_steps: print(f\n Step {state.current_step 1} ) # 1. 构建提示 prompt self._build_prompt(state) # print(fPrompt:\n{prompt}\n) # 调试时打开 # 2. 调用LLM进行推理 llm_response self._call_llm(prompt) # print(fLLM Raw Output:\n{llm_response}\n) # 调试时打开 # 3. 解析LLM输出 thought, action, action_input self._parse_llm_output(llm_response) if thought is None: # 解析失败 observation fError: Failed to parse LLM output: {llm_response} self._update_history(state, thought, actionParseError, action_input, observationobservation) state.current_step 1 continue print(fThought: {thought}) print(fAction: {action}) print(fAction Input: {action_input}) # 4. 检查是否为最终答案 if action.lower() final: print(fFinal Answer: {action_input}) return action_input # 5. 执行动作 observation self._execute_action(action, action_input) print(fObservation: {observation}) # 6. 更新历史与状态 self._update_history(state, thought, action, action_input, observation) state.current_step 1 # 7. 检查是否可以从Observation中提前终止可选 if self._is_task_completed(observation, state.task): print(Task completed based on observation.) return observation return fReached max steps ({state.max_steps}) without completing the task. Latest state: {state.history[-1] if state.history else No history} def _build_prompt(self, state: AgentState) - str: 构建给LLM的提示。这是ReAct的核心之一。 # 工具描述部分 tools_text \n.join([f- {tool.name}: {tool.description} for tool in self.tools.values()]) # 历史记录部分 history_text for record in state.history: history_text fThought: {record.thought}\nAction: {record.action}\nAction Input: {record.action_input}\nObservation: {record.observation}\n\n prompt fYou are a helpful AI assistant. Your task is: {state.task} You have access to the following tools: {tools_text} You must respond in the following format: Thought: [Your reasoning process here. Think step by step about what to do next.] Action: [The name of the tool to use, or Final if you have the final answer.] Action Input: [The input to the tool, or the final answer if Action is Final.] Begin! {history_text}Thought: return prompt def _call_llm(self, prompt: str) - str: 调用LLM API。这里以OpenAI为例。 # 注意实际使用时请处理异常和速率限制 response openai.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messages[{role: user, content: prompt}], temperature0, # 对于结构化输出温度通常设为0以保证稳定性 max_tokens500 ) return response.choices[0].message.content.strip() def _parse_llm_output(self, text: str) - Tuple[Optional[str], Optional[str], Optional[str]]: 从LLM输出中解析出Thought, Action, Action Input。 这是一个脆弱的环节需要健壮的解析逻辑。 lines text.split(\n) thought, action, action_input None, None, None for i, line in enumerate(lines): if line.startswith(Thought:): thought line[8:].strip() # 可能Thought有多行我们取到下一个Action:之前的所有内容 thought_lines [thought] for next_line in lines[i1:]: if next_line.startswith((Action:, Action Input:)): break thought_lines.append(next_line.strip()) thought .join(thought_lines).strip() elif line.startswith(Action:): action line[7:].strip() elif line.startswith(Action Input:): action_input line[13:].strip() # 如果Action Input包含多行或引号这里需要更复杂的处理 return thought, action, action_input def _execute_action(self, action_name: str, action_input: str) - str: 执行工具调用。 if action_name not in self.tools: return fError: Unknown action {action_name}. Available actions: {list(self.tools.keys())} tool self.tools[action_name] try: result tool.func(action_input) return str(result) except Exception as e: return fError executing action {action_name}: {e} def _update_history(self, state: AgentState, thought: str, action: str, action_input: str, observation: str): 将当前步骤记录添加到历史中。 state.history.append(StepRecord( thoughtthought, actionaction, action_inputaction_input, observationobservation )) def _is_task_completed(self, observation: str, task: str) - bool: 一个简单的启发式方法判断任务是否完成。 在实际应用中这可能需要更复杂的逻辑甚至另一个LLM调用来判断。 # 例如如果任务是关于计算且observation看起来像数字答案 # 或者如果observation包含“答案”、“结果是”等关键词 # 这里只是一个非常简单的示例 lower_obs observation.lower() if error not in lower_obs and (len(lower_obs) 100): # 简单假设短的非错误输出可能是答案 # 可以加入更多基于具体任务类型的逻辑 return True return False3.3 定义并注册工具没有工具的Agent就像没有手脚的大脑。我们来定义几个简单的工具。# 定义几个示例工具 def search_wikipedia(query: str) - str: 模拟搜索维基百科。在实际中这里会调用真实的API。 # 模拟返回 mock_data { Python: Python is a high-level, interpreted programming language., GPT: Generative Pre-trained Transformer is a family of large language models., ReAct: ReAct is a prompting paradigm that combines Reasoning and Acting. } return mock_data.get(query, fNo information found for {query}.) def calculator(expression: str) - str: 计算数学表达式。警告直接用eval有安全风险仅用于演示。 try: # 严重警告在生产环境中绝对不要用eval直接执行用户或LLM提供的字符串。 # 这里仅作演示应使用安全的表达式求值库如ast.literal_eval配合自定义解析。 result eval(expression, {__builtins__: None}, {}) return str(result) except Exception as e: return fCalculation error: {e} def get_current_time(_: str) - str: 返回当前时间。 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) # 创建工具列表 tools [ Tool(nameSearchWikipedia, funcsearch_wikipedia, descriptionUseful for when you need to answer questions about general knowledge, history, or concepts. Input should be a clear search query.), Tool(nameCalculator, funccalculator, descriptionUseful for performing arithmetic calculations. Input should be a valid mathematical expression, e.g., (12 5) * 3.), Tool(nameGetCurrentTime, funcget_current_time, descriptionUseful for when you need to know the current date and time. Input is ignored.), ]4. 让心跳转起来运行与调试你的第一个ReAct Agent现在让我们点燃引擎看看这个手写的“心脏”是否能跳动起来。4.1 首次运行测试# 初始化引擎 client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key engine ReActEngine(llm_clientclient, toolstools) # 执行一个简单任务 task1 What is Python, and what is 15 raised to the power of 2? result1 engine.run(task1, max_steps5) print(f\n Final Result for Task 1 \n{result1})预期的控制台输出可能类似于 Step 1 Thought: The user is asking two questions: what Python is, and a math calculation. I should first search for information about Python, then calculate the power. Action: SearchWikipedia Action Input: Python Observation: Python is a high-level, interpreted programming language. Step 2 Thought: I have the answer about Python. Now I need to calculate 15 raised to the power of 2, which is 15^2. Action: Calculator Action Input: 15 ** 2 Observation: 225 Step 3 Thought: I now have both pieces of information. I can provide the final answer. Action: Final Action Input: Python is a high-level, interpreted programming language. 15 raised to the power of 2 is 225. Final Result for Task 1 Python is a high-level, interpreted programming language. 15 raised to the power of 2 is 225.看它工作了Agent成功地将复杂问题分解先调用搜索工具获取信息再调用计算器进行计算最后整合信息给出最终答案。这就是ReAct循环的力量。4.2 处理更复杂与模糊的任务让我们挑战一个需要多步推理和工具选择的模糊任务。task2 Who invented the light bulb, and how old would they be today if they were alive? result2 engine.run(task2, max_steps6) print(f\n Final Result for Task 2 \n{result2})可能的输出流程Step 1: Thought: 需要先找出电灯泡的发明者。Action:SearchWikipedia, Input:inventor of light bulb。Step 2: Observation: 返回“托马斯·爱迪生”等信息。Step 3: Thought: 知道了发明者是托马斯·爱迪生生于1847年2月11日。现在需要计算如果他今天还活着是多少岁。这需要知道今天的日期然后计算年龄。Action:GetCurrentTime。Step 4: Observation: 返回当前日期时间例如“2023-10-27”。Step 5: Thought: 现在有了出生日期和当前日期需要计算年龄差。这涉及到日期计算我的计算器工具可能只能处理算术。我需要手动推理或尝试计算年份差。先试试计算器。Action:Calculator, Input:2023 - 1847。Step 6: Observation:176。Step 7: Thought: 计算得出176年。我可以给出最终答案了。Action:Final...这个例子展示了循环如何应对需要组合多个工具搜索、获取时间、计算和信息串联的任务。同时它也暴露了我们当前简单设计的局限性日期计算并不精确没有考虑月份和日期计算器工具的功能也很基础。5. 手写实践中遇到的典型问题与深度优化方案当你亲手运行起这个循环很快就会发现一堆“坑”。下面是我在反复调试中总结的核心问题和进阶优化方案。5.1 LLM输出格式不稳定解析器崩溃问题LLM可能不严格按照“Thought:... Action:... Action Input:...”的格式输出。它可能漏掉前缀可能把“Action Input”写成“Input”可能在一行内输出多个字段或者在“Thought”部分包含类似“Action:”的词语。解决方案强化你的_parse_llm_output函数。正则表达式增强使用更灵活的正则表达式进行匹配允许字段间有空白行允许字段顺序略有变化。import re def _parse_llm_output_v2(self, text: str) - Tuple[Optional[str], Optional[str], Optional[str]]: thought_pattern rThought:\s*(.*?)(?\n\s*(?:Action:|Action Input:|$)) action_pattern rAction:\s*(\w) action_input_pattern rAction Input:\s*(.*?)(?\n\s*(?:Thought:|Action:|$)) thought re.search(thought_pattern, text, re.DOTALL) action re.search(action_pattern, text) action_input re.search(action_input_pattern, text, re.DOTALL) thought thought.group(1).strip() if thought else None action action.group(1).strip() if action else None action_input action_input.group(1).strip() if action_input else None return thought, action, action_inputJSON格式强制在提示词中明确要求LLM输出JSON格式。这通常能获得更稳定的解析结果。将提示词末尾的格式要求改为You must respond in a strict JSON format: {{ thought: Your reasoning here, action: ToolName or Final, action_input: input string }}然后在解析时使用json.loads()。 3.后处理与兜底如果解析失败不要直接崩溃。可以将解析失败的原始文本作为observation反馈给下一轮循环并在提示词中加入“你上次的回复格式有误请严格遵守格式”之类的提醒让LLM自我纠正。5.2 工具调用失败与错误处理问题工具执行可能抛出异常如网络错误、无效输入或者返回的结果格式不符合预期。解决方案完善的工具包装在每个工具函数内部进行充分的输入验证和异常捕获返回结构化的错误信息而不是抛出异常。引擎层的错误处理在_execute_action方法中除了捕获异常还可以根据工具返回的字符串内容判断是否成功。例如约定工具返回以“Error:”开头的字符串表示失败。将错误作为观察无论工具调用成功还是失败都将结果格式化为observation并入历史。LLM可以从错误中学习调整后续策略。例如如果计算器返回“Calculation error: invalid syntax”下一轮LLM的Thought可能会是“上次计算表达式有误我需要检查一下表达式格式。”5.3 循环失控无限循环与重复动作问题Agent可能陷入死循环反复执行同一个无意义的动作或者在一个问题上打转。解决方案最大步数限制我们已经实现了这是最后的安全网。重复动作检测在_update_history中检查最近N步的历史记录。如果发现完全相同的(action, action_input)组合重复出现可以中断循环并返回错误或者在observation中插入警告“你正在重复之前的操作请尝试新的策略。”思维状态跟踪在State中增加一个context或summary字段每几步就对历史进行一次摘要防止历史记录过长导致提示词爆炸也帮助LLM保持对整体任务的宏观认知。更智能的终止判断_is_task_completed函数可以升级。可以训练一个小的分类器或者使用另一个LLM调用来判断当前observation是否已经满意地回答了原始task。5.4 提示工程优化让LLM更“听话”问题LLM有时会忽略工具直接给出答案或者选择的工具不合理。解决方案少样本示例Few-shot在提示词中提供1-2个完整的、格式正确的ReAct循环示例。这是大幅提升LLM格式遵从性和推理逻辑的最有效方法之一。工具描述精细化工具的描述 (description) 至关重要。要清晰说明工具的用途、输入格式和输出示例。例如将“做计算”改为“用于计算数学表达式。输入必须是一个只包含数字和运算符 - * / ** ( )的字符串例如‘(34)*5’。”系统指令强化在消息列表的开头使用system角色给出更强烈的指令强调必须使用工具、必须遵循格式。5.5 性能与成本考量问题每一步都需要调用LLM成本高、速度慢。解决方案历史截断与摘要当历史记录超过一定长度token数时不再全部放入提示词。可以保留最近几步的完整记录并对更早的记录进行摘要用LLM生成摘要或提取关键事实。并行工具调用如果多个工具调用之间没有依赖关系理论上可以并行执行。但这需要更复杂的动作解析和状态管理逻辑。使用更小/更快的模型对于推理步骤可以尝试使用更经济的小模型如GPT-3.5-turbo只在必要时才调用大模型。6. 超越基础将手写引擎升级为生产级框架的思考当我们成功让基础循环跑通后就可以思考如何将它打磨得更具实用性、鲁棒性和扩展性。6.1 状态管理的进化基础的List[StepRecord]会随着循环进行不断膨胀导致两个问题提示词token数超标和LLM难以从冗长历史中提取关键信息。向量化记忆将每一步的thought和observation编码成向量存储到向量数据库如Chroma, FAISS。在每一轮不是传入全部历史文本而是根据当前任务描述进行语义检索只召回最相关的几步历史。这模仿了人类的“选择性记忆”。摘要记忆每进行N步或者当历史token数达到阈值时启动一个“摘要智能体”将过去一段时间的交互浓缩成一段简洁的摘要替换掉原来的详细记录。摘要作为新的“长期记忆”进入后续循环。6.2 工具生态的扩展动态工具注册允许在运行时动态添加或移除工具而不是在引擎初始化时固定死。工具组合与工作流定义更高级的“复合工具”或“子工作流”。例如一个“数据分析”工具内部可能封装了“查询数据库-清洗数据-生成图表”的固定流程。工具验证与授权为工具增加权限控制某些敏感工具如发送邮件、操作数据库需要额外的验证或确认步骤才能执行。6.3 引入规划与反思机制基础的ReAct是“走一步看一步”的。更强大的Agent需要“走一步看三步”甚至“事后复盘”。规划阶段在循环开始前或遇到复杂任务时先让LLM生成一个高层次计划Plan例如“第一步搜索A第二步根据A的结果搜索B第三步计算C”。后续的ReAct循环则遵循这个计划大纲。反思阶段在每一步或任务结束后增加一个“反思”步骤。让LLM评估刚才的行动是否有效是否偏离目标从中可以学到什么并据此调整后续策略。这相当于给循环加了一个“元认知”层。6.4 可观测性与调试工具对于开发者而言一个“黑盒”Agent是可怕的。详细日志记录每一步的完整Prompt、LLM原始响应、解析结果、工具输入输出、耗时等信息。可视化追踪可以生成一个交互式的轨迹图展示Agent的思考路径、工具调用序列和状态演变类似于LangChain的LangSmith提供的功能。断点与干预允许开发者在特定步骤暂停循环手动修改或注入下一步的Action和Observation用于测试和调试。手写ReAct循环的过程是一个从“知其然”到“知其所以然”的深度旅程。你遇到的每一个报错、每一次循环失控都会让你对智能体内部运作机制的理解加深一分。当你终于能流畅地驾驭这个“心跳”并开始根据自己的需求去改造和强化它时你就从一个框架的使用者真正变成了智能体架构的创造者。这就是动手的价值。 从零手写ReAct循环：深入理解AI智能体的思考与行动机制
1. 项目概述为什么我们要“手搓”一个ReAct循环如果你最近在关注AI应用开发尤其是智能体Agent领域那么“ReAct”这个词一定高频出现在你的视野里。它不是什么新出的前端框架而是驱动现代AI智能体进行“思考”和“行动”的核心范式。简单来说ReAct就是Reason推理 Act行动的循环。一个智能体接收到任务后不是直接给出最终答案而是先“想一想”Reason决定下一步该做什么比如调用某个工具、查询知识库然后执行这个动作Act根据执行结果再进入下一轮的“思考”如此循环直到任务完成。你可能会问现在LangChain、LlamaIndex这些框架不是已经把ReAct封装好了吗直接用不就行了没错用框架可以快速搭建原型。但这就好比学开车你只知道踩油门和刹车却不了解发动机和变速箱是怎么协同工作的。当你的Agent在复杂任务中“卡壳”、陷入死循环或者做出匪夷所思的决策时如果你对底层机制一无所知调试将变得异常痛苦。“从零手写”的价值就在于此。这不是为了重复造轮子而是为了亲手摸清这个“轮子”的每一个辐条是如何咬合的。通过剥离框架的抽象层我们能够透彻理解核心机制真正搞懂“思考-行动”循环是如何流转的状态如何管理工具如何被选择和调用。获得终极的调试能力当智能体行为异常时你能像外科医生一样精准定位是推理环节的提示词出了问题还是工具返回的结果解析有误或是状态管理出现了混乱。实现高度定制化框架提供的是通用解决方案。当你需要为特定领域如金融分析、游戏NPC、自动化运维设计独特的推理逻辑或行动策略时手写的架构给你最大的自由度。所以这个项目标题“从零手写 ReAct 循环”瞄准的正是那些不满足于当“调包侠”渴望掌握智能体最核心“心跳”机制的开发者。我们将一起用代码模拟出这个让AI能够自主规划、使用工具、逐步逼近目标的思考过程。2. ReAct循环的核心原理与架构设计在动手写代码之前我们必须像建筑师审视蓝图一样把ReAct循环的每一个组件和它们之间的数据流彻底想清楚。一个最简化的ReAct循环可以抽象为以下几个核心部分2.1 核心组件拆解智能体Agent这是循环的“大脑”。它持有当前的任务目标、与环境的交互历史记忆以及可供使用的工具列表。它的核心职责是发起“推理”。推理器Reasoner通常是基于大语言模型LLM的模块。它接收当前的“状态”包括任务描述、历史记录、可用工具然后输出一个结构化的“思考”和“决策”。这个决策通常表现为“我需要先做A那么我就调用工具X。”动作执行器Actuator负责执行推理器做出的决策。如果决策是调用工具它就找到对应的工具函数传入参数并执行然后将执行结果收集回来。观察与状态更新器Observer将动作执行的结果成功或失败附带返回数据进行格式化形成一条新的“观察”记录并将其追加到交互历史中更新整个循环的状态。循环控制器Loop Controller判断循环是否应该继续。判断条件可以是任务是否已达成如答案已找到、是否陷入了死循环如重复调用同一工具、是否超过了最大步数限制。2.2 数据流心跳的节拍一次完整的心跳循环迭代遵循以下节拍状态输入将当前任务描述、完整的历史交互记录格式为Thought: ... Action: ... Observation: ...、可用工具描述组合成一个提示Prompt输入给推理器LLM。推理与决策LLM根据提示生成一段文本。我们需要用解析器Parser从这段文本中提取出关键结构thought本次推理过程、action要执行的动作名称、action_input动作的输入参数。动作执行根据解析出的action在工具注册表中找到对应的函数将action_input作为参数传入并执行。观察生成捕获工具函数的执行结果或异常将其格式化为一个observation字符串。历史更新将本次循环产生的thought,action,observation三元组追加到历史记录列表中。这就构成了智能体的“短期记忆”。循环判断检查observation是否已经包含了最终答案或者是否触发了终止条件如最大步数。如果是则跳出循环返回结果如果不是则回到第1步。注意这里有一个关键设计点。我们要求LLM输出的格式必须是可解析的比如固定输出“Thought: ... Action: ... Action Input: ...”。这是ReAct能稳定运行的前提也是手写时需要精心设计提示词和解析逻辑的地方。2.3 与框架的对比我们手写的重点像LangChain的Agent它把这些组件都封装好了提供了AgentExecutor来驱动循环。我们手写就是要实现一个自己的、轻量级的AgentExecutor。我们的关注点将集中在提示工程如何构建最有效的提示让LLM稳定地输出我们想要的格式。解析逻辑如何鲁棒地从LLM的非确定性输出中准确提取出动作指令。工具管理如何优雅地注册、描述和调用工具函数。状态管理如何设计历史记录的数据结构使其既能包含完整信息又不会因为过长而影响后续推理未来可以考虑摘要或向量化存储但本项目先从简单列表开始。错误处理与循环终止当工具调用失败、LLM输出无法解析时系统该如何降级处理如何设计合理的终止条件避免无限循环3. 从零开始构建手写ReAct循环的代码骨架理论清晰了现在打开你的代码编辑器我们开始一行行构建这个循环。我们将使用Python作为实现语言并假设你已经配置好了OpenAI或类似LLM的API环境。3.1 环境准备与基础定义首先定义最基础的数据结构。我们将使用Pydantic来创建数据模型这能让我们的代码更清晰、类型更安全。from typing import Any, Callable, Dict, List, Optional, Tuple from pydantic import BaseModel, Field import openai # 或其他LLM客户端 # 定义工具一个可调用函数及其描述 class Tool(BaseModel): name: str func: Callable[[str], str] # 简化起见输入输出均为字符串 description: str # 这个描述至关重要LLM靠它来决定是否使用该工具 # 定义单步交互记录 class StepRecord(BaseModel): thought: str action: str action_input: str observation: str # 定义智能体状态 class AgentState(BaseModel): task: str history: List[StepRecord] Field(default_factorylist) max_steps: int 10 current_step: int 03.2 核心引擎ReAct循环控制器接下来我们实现循环控制器也就是整个系统的心脏。class ReActEngine: def __init__(self, llm_client, tools: List[Tool]): self.llm llm_client self.tools {tool.name: tool for tool in tools} # 工具字典便于按名查找 def run(self, task: str, max_steps: int 10) - str: 执行ReAct循环的主函数 state AgentState(tasktask, max_stepsmax_steps) while state.current_step state.max_steps: print(f\n Step {state.current_step 1} ) # 1. 构建提示 prompt self._build_prompt(state) # print(fPrompt:\n{prompt}\n) # 调试时打开 # 2. 调用LLM进行推理 llm_response self._call_llm(prompt) # print(fLLM Raw Output:\n{llm_response}\n) # 调试时打开 # 3. 解析LLM输出 thought, action, action_input self._parse_llm_output(llm_response) if thought is None: # 解析失败 observation fError: Failed to parse LLM output: {llm_response} self._update_history(state, thought, actionParseError, action_input, observationobservation) state.current_step 1 continue print(fThought: {thought}) print(fAction: {action}) print(fAction Input: {action_input}) # 4. 检查是否为最终答案 if action.lower() final: print(fFinal Answer: {action_input}) return action_input # 5. 执行动作 observation self._execute_action(action, action_input) print(fObservation: {observation}) # 6. 更新历史与状态 self._update_history(state, thought, action, action_input, observation) state.current_step 1 # 7. 检查是否可以从Observation中提前终止可选 if self._is_task_completed(observation, state.task): print(Task completed based on observation.) return observation return fReached max steps ({state.max_steps}) without completing the task. Latest state: {state.history[-1] if state.history else No history} def _build_prompt(self, state: AgentState) - str: 构建给LLM的提示。这是ReAct的核心之一。 # 工具描述部分 tools_text \n.join([f- {tool.name}: {tool.description} for tool in self.tools.values()]) # 历史记录部分 history_text for record in state.history: history_text fThought: {record.thought}\nAction: {record.action}\nAction Input: {record.action_input}\nObservation: {record.observation}\n\n prompt fYou are a helpful AI assistant. Your task is: {state.task} You have access to the following tools: {tools_text} You must respond in the following format: Thought: [Your reasoning process here. Think step by step about what to do next.] Action: [The name of the tool to use, or Final if you have the final answer.] Action Input: [The input to the tool, or the final answer if Action is Final.] Begin! {history_text}Thought: return prompt def _call_llm(self, prompt: str) - str: 调用LLM API。这里以OpenAI为例。 # 注意实际使用时请处理异常和速率限制 response openai.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messages[{role: user, content: prompt}], temperature0, # 对于结构化输出温度通常设为0以保证稳定性 max_tokens500 ) return response.choices[0].message.content.strip() def _parse_llm_output(self, text: str) - Tuple[Optional[str], Optional[str], Optional[str]]: 从LLM输出中解析出Thought, Action, Action Input。 这是一个脆弱的环节需要健壮的解析逻辑。 lines text.split(\n) thought, action, action_input None, None, None for i, line in enumerate(lines): if line.startswith(Thought:): thought line[8:].strip() # 可能Thought有多行我们取到下一个Action:之前的所有内容 thought_lines [thought] for next_line in lines[i1:]: if next_line.startswith((Action:, Action Input:)): break thought_lines.append(next_line.strip()) thought .join(thought_lines).strip() elif line.startswith(Action:): action line[7:].strip() elif line.startswith(Action Input:): action_input line[13:].strip() # 如果Action Input包含多行或引号这里需要更复杂的处理 return thought, action, action_input def _execute_action(self, action_name: str, action_input: str) - str: 执行工具调用。 if action_name not in self.tools: return fError: Unknown action {action_name}. Available actions: {list(self.tools.keys())} tool self.tools[action_name] try: result tool.func(action_input) return str(result) except Exception as e: return fError executing action {action_name}: {e} def _update_history(self, state: AgentState, thought: str, action: str, action_input: str, observation: str): 将当前步骤记录添加到历史中。 state.history.append(StepRecord( thoughtthought, actionaction, action_inputaction_input, observationobservation )) def _is_task_completed(self, observation: str, task: str) - bool: 一个简单的启发式方法判断任务是否完成。 在实际应用中这可能需要更复杂的逻辑甚至另一个LLM调用来判断。 # 例如如果任务是关于计算且observation看起来像数字答案 # 或者如果observation包含“答案”、“结果是”等关键词 # 这里只是一个非常简单的示例 lower_obs observation.lower() if error not in lower_obs and (len(lower_obs) 100): # 简单假设短的非错误输出可能是答案 # 可以加入更多基于具体任务类型的逻辑 return True return False3.3 定义并注册工具没有工具的Agent就像没有手脚的大脑。我们来定义几个简单的工具。# 定义几个示例工具 def search_wikipedia(query: str) - str: 模拟搜索维基百科。在实际中这里会调用真实的API。 # 模拟返回 mock_data { Python: Python is a high-level, interpreted programming language., GPT: Generative Pre-trained Transformer is a family of large language models., ReAct: ReAct is a prompting paradigm that combines Reasoning and Acting. } return mock_data.get(query, fNo information found for {query}.) def calculator(expression: str) - str: 计算数学表达式。警告直接用eval有安全风险仅用于演示。 try: # 严重警告在生产环境中绝对不要用eval直接执行用户或LLM提供的字符串。 # 这里仅作演示应使用安全的表达式求值库如ast.literal_eval配合自定义解析。 result eval(expression, {__builtins__: None}, {}) return str(result) except Exception as e: return fCalculation error: {e} def get_current_time(_: str) - str: 返回当前时间。 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) # 创建工具列表 tools [ Tool(nameSearchWikipedia, funcsearch_wikipedia, descriptionUseful for when you need to answer questions about general knowledge, history, or concepts. Input should be a clear search query.), Tool(nameCalculator, funccalculator, descriptionUseful for performing arithmetic calculations. Input should be a valid mathematical expression, e.g., (12 5) * 3.), Tool(nameGetCurrentTime, funcget_current_time, descriptionUseful for when you need to know the current date and time. Input is ignored.), ]4. 让心跳转起来运行与调试你的第一个ReAct Agent现在让我们点燃引擎看看这个手写的“心脏”是否能跳动起来。4.1 首次运行测试# 初始化引擎 client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key engine ReActEngine(llm_clientclient, toolstools) # 执行一个简单任务 task1 What is Python, and what is 15 raised to the power of 2? result1 engine.run(task1, max_steps5) print(f\n Final Result for Task 1 \n{result1})预期的控制台输出可能类似于 Step 1 Thought: The user is asking two questions: what Python is, and a math calculation. I should first search for information about Python, then calculate the power. Action: SearchWikipedia Action Input: Python Observation: Python is a high-level, interpreted programming language. Step 2 Thought: I have the answer about Python. Now I need to calculate 15 raised to the power of 2, which is 15^2. Action: Calculator Action Input: 15 ** 2 Observation: 225 Step 3 Thought: I now have both pieces of information. I can provide the final answer. Action: Final Action Input: Python is a high-level, interpreted programming language. 15 raised to the power of 2 is 225. Final Result for Task 1 Python is a high-level, interpreted programming language. 15 raised to the power of 2 is 225.看它工作了Agent成功地将复杂问题分解先调用搜索工具获取信息再调用计算器进行计算最后整合信息给出最终答案。这就是ReAct循环的力量。4.2 处理更复杂与模糊的任务让我们挑战一个需要多步推理和工具选择的模糊任务。task2 Who invented the light bulb, and how old would they be today if they were alive? result2 engine.run(task2, max_steps6) print(f\n Final Result for Task 2 \n{result2})可能的输出流程Step 1: Thought: 需要先找出电灯泡的发明者。Action:SearchWikipedia, Input:inventor of light bulb。Step 2: Observation: 返回“托马斯·爱迪生”等信息。Step 3: Thought: 知道了发明者是托马斯·爱迪生生于1847年2月11日。现在需要计算如果他今天还活着是多少岁。这需要知道今天的日期然后计算年龄。Action:GetCurrentTime。Step 4: Observation: 返回当前日期时间例如“2023-10-27”。Step 5: Thought: 现在有了出生日期和当前日期需要计算年龄差。这涉及到日期计算我的计算器工具可能只能处理算术。我需要手动推理或尝试计算年份差。先试试计算器。Action:Calculator, Input:2023 - 1847。Step 6: Observation:176。Step 7: Thought: 计算得出176年。我可以给出最终答案了。Action:Final...这个例子展示了循环如何应对需要组合多个工具搜索、获取时间、计算和信息串联的任务。同时它也暴露了我们当前简单设计的局限性日期计算并不精确没有考虑月份和日期计算器工具的功能也很基础。5. 手写实践中遇到的典型问题与深度优化方案当你亲手运行起这个循环很快就会发现一堆“坑”。下面是我在反复调试中总结的核心问题和进阶优化方案。5.1 LLM输出格式不稳定解析器崩溃问题LLM可能不严格按照“Thought:... Action:... Action Input:...”的格式输出。它可能漏掉前缀可能把“Action Input”写成“Input”可能在一行内输出多个字段或者在“Thought”部分包含类似“Action:”的词语。解决方案强化你的_parse_llm_output函数。正则表达式增强使用更灵活的正则表达式进行匹配允许字段间有空白行允许字段顺序略有变化。import re def _parse_llm_output_v2(self, text: str) - Tuple[Optional[str], Optional[str], Optional[str]]: thought_pattern rThought:\s*(.*?)(?\n\s*(?:Action:|Action Input:|$)) action_pattern rAction:\s*(\w) action_input_pattern rAction Input:\s*(.*?)(?\n\s*(?:Thought:|Action:|$)) thought re.search(thought_pattern, text, re.DOTALL) action re.search(action_pattern, text) action_input re.search(action_input_pattern, text, re.DOTALL) thought thought.group(1).strip() if thought else None action action.group(1).strip() if action else None action_input action_input.group(1).strip() if action_input else None return thought, action, action_inputJSON格式强制在提示词中明确要求LLM输出JSON格式。这通常能获得更稳定的解析结果。将提示词末尾的格式要求改为You must respond in a strict JSON format: {{ thought: Your reasoning here, action: ToolName or Final, action_input: input string }}然后在解析时使用json.loads()。 3.后处理与兜底如果解析失败不要直接崩溃。可以将解析失败的原始文本作为observation反馈给下一轮循环并在提示词中加入“你上次的回复格式有误请严格遵守格式”之类的提醒让LLM自我纠正。5.2 工具调用失败与错误处理问题工具执行可能抛出异常如网络错误、无效输入或者返回的结果格式不符合预期。解决方案完善的工具包装在每个工具函数内部进行充分的输入验证和异常捕获返回结构化的错误信息而不是抛出异常。引擎层的错误处理在_execute_action方法中除了捕获异常还可以根据工具返回的字符串内容判断是否成功。例如约定工具返回以“Error:”开头的字符串表示失败。将错误作为观察无论工具调用成功还是失败都将结果格式化为observation并入历史。LLM可以从错误中学习调整后续策略。例如如果计算器返回“Calculation error: invalid syntax”下一轮LLM的Thought可能会是“上次计算表达式有误我需要检查一下表达式格式。”5.3 循环失控无限循环与重复动作问题Agent可能陷入死循环反复执行同一个无意义的动作或者在一个问题上打转。解决方案最大步数限制我们已经实现了这是最后的安全网。重复动作检测在_update_history中检查最近N步的历史记录。如果发现完全相同的(action, action_input)组合重复出现可以中断循环并返回错误或者在observation中插入警告“你正在重复之前的操作请尝试新的策略。”思维状态跟踪在State中增加一个context或summary字段每几步就对历史进行一次摘要防止历史记录过长导致提示词爆炸也帮助LLM保持对整体任务的宏观认知。更智能的终止判断_is_task_completed函数可以升级。可以训练一个小的分类器或者使用另一个LLM调用来判断当前observation是否已经满意地回答了原始task。5.4 提示工程优化让LLM更“听话”问题LLM有时会忽略工具直接给出答案或者选择的工具不合理。解决方案少样本示例Few-shot在提示词中提供1-2个完整的、格式正确的ReAct循环示例。这是大幅提升LLM格式遵从性和推理逻辑的最有效方法之一。工具描述精细化工具的描述 (description) 至关重要。要清晰说明工具的用途、输入格式和输出示例。例如将“做计算”改为“用于计算数学表达式。输入必须是一个只包含数字和运算符 - * / ** ( )的字符串例如‘(34)*5’。”系统指令强化在消息列表的开头使用system角色给出更强烈的指令强调必须使用工具、必须遵循格式。5.5 性能与成本考量问题每一步都需要调用LLM成本高、速度慢。解决方案历史截断与摘要当历史记录超过一定长度token数时不再全部放入提示词。可以保留最近几步的完整记录并对更早的记录进行摘要用LLM生成摘要或提取关键事实。并行工具调用如果多个工具调用之间没有依赖关系理论上可以并行执行。但这需要更复杂的动作解析和状态管理逻辑。使用更小/更快的模型对于推理步骤可以尝试使用更经济的小模型如GPT-3.5-turbo只在必要时才调用大模型。6. 超越基础将手写引擎升级为生产级框架的思考当我们成功让基础循环跑通后就可以思考如何将它打磨得更具实用性、鲁棒性和扩展性。6.1 状态管理的进化基础的List[StepRecord]会随着循环进行不断膨胀导致两个问题提示词token数超标和LLM难以从冗长历史中提取关键信息。向量化记忆将每一步的thought和observation编码成向量存储到向量数据库如Chroma, FAISS。在每一轮不是传入全部历史文本而是根据当前任务描述进行语义检索只召回最相关的几步历史。这模仿了人类的“选择性记忆”。摘要记忆每进行N步或者当历史token数达到阈值时启动一个“摘要智能体”将过去一段时间的交互浓缩成一段简洁的摘要替换掉原来的详细记录。摘要作为新的“长期记忆”进入后续循环。6.2 工具生态的扩展动态工具注册允许在运行时动态添加或移除工具而不是在引擎初始化时固定死。工具组合与工作流定义更高级的“复合工具”或“子工作流”。例如一个“数据分析”工具内部可能封装了“查询数据库-清洗数据-生成图表”的固定流程。工具验证与授权为工具增加权限控制某些敏感工具如发送邮件、操作数据库需要额外的验证或确认步骤才能执行。6.3 引入规划与反思机制基础的ReAct是“走一步看一步”的。更强大的Agent需要“走一步看三步”甚至“事后复盘”。规划阶段在循环开始前或遇到复杂任务时先让LLM生成一个高层次计划Plan例如“第一步搜索A第二步根据A的结果搜索B第三步计算C”。后续的ReAct循环则遵循这个计划大纲。反思阶段在每一步或任务结束后增加一个“反思”步骤。让LLM评估刚才的行动是否有效是否偏离目标从中可以学到什么并据此调整后续策略。这相当于给循环加了一个“元认知”层。6.4 可观测性与调试工具对于开发者而言一个“黑盒”Agent是可怕的。详细日志记录每一步的完整Prompt、LLM原始响应、解析结果、工具输入输出、耗时等信息。可视化追踪可以生成一个交互式的轨迹图展示Agent的思考路径、工具调用序列和状态演变类似于LangChain的LangSmith提供的功能。断点与干预允许开发者在特定步骤暂停循环手动修改或注入下一步的Action和Observation用于测试和调试。手写ReAct循环的过程是一个从“知其然”到“知其所以然”的深度旅程。你遇到的每一个报错、每一次循环失控都会让你对智能体内部运作机制的理解加深一分。当你终于能流畅地驾驭这个“心跳”并开始根据自己的需求去改造和强化它时你就从一个框架的使用者真正变成了智能体架构的创造者。这就是动手的价值。