1. 项目概述从“代码生成”到“代码执行”的范式跃迁如果你用过 Claude Code 或者 GitHub Copilot肯定对它们“写代码”的能力印象深刻。你写个注释它就能给你生成一段看起来不错的函数。但不知道你有没有遇到过这种情况生成的代码逻辑上没问题但一运行就报错要么是导入了不存在的包要么是调用了错误的 API。这背后的根本原因在于传统的代码生成模型本质上是一个“文本预测”模型——它根据你给的上下文预测下一个最可能出现的“词元”。它并不真正理解代码的“运行环境”更无法验证代码的“执行结果”。而 Claude Code 源码中揭示的ReAct 主循环正是为了解决这个问题而生的核心架构。它不是一个简单的代码补全工具而是一个具备“思考-行动-观察”能力的智能体。简单来说它让 AI 不再只是“写”代码而是开始“执行”代码并根据执行结果进行“修正”和“迭代”。这就像是从一个只会纸上谈兵的参谋变成了一个能亲自上阵、根据战场反馈调整战术的指挥官。这个主循环机制是当前 AI 编程助手从“辅助生成”迈向“自主执行”的关键一步。它不仅仅是 Claude Code 的技术内核更代表了下一代开发工具的发展方向。无论是想深入理解 AI 如何与真实开发环境交互还是希望在自己的项目中引入类似的智能体能力拆解 ReAct 主循环都是绝佳的切入点。接下来我将带你深入源码看看这个循环是如何一步步运转起来的。2. 核心架构与设计哲学为什么是 ReAct在深入代码之前我们必须先理解 Claude Code 选择 ReAct 框架的底层逻辑。ReAct 是 “Reasoning Acting” 的缩写其核心思想是让智能体交替进行“推理”和“行动”。在代码生成的场景下这被具体化为推理分析当前任务、已有代码、错误信息规划下一步要做什么。行动执行一个具体操作比如运行一段代码、读取一个文件、安装一个依赖。观察获取行动的结果成功输出、错误信息、文件内容变化。这个循环会一直持续直到任务被完成或达到终止条件。那么为什么这种模式特别适合代码生成呢2.1 传统代码生成的局限性传统的端到端代码生成模型如基于 Transformer 的大模型存在几个固有缺陷环境盲区模型不知道你项目的package.json里具体有哪些依赖和版本也不知道当前工作目录的文件结构。它只能基于训练数据中的“常见模式”进行猜测。静态幻觉模型可能会“幻想”出一些不存在的 API 或函数签名因为它学到的知识可能过时或者与你的特定项目环境不符。缺乏验证生成的代码没有经过“运行”这一终极测试。语法正确不代表逻辑正确更不代表能在你的环境中运行。2.2 ReAct 带来的范式优势Claude Code 的 ReAct 主循环直接针对了上述痛点动态环境感知通过“行动”如ls,cat智能体可以实时读取项目文件、查看日志、检查依赖从而获得最新的、准确的上下文信息。这解决了“环境盲区”问题。执行反馈驱动通过“观察”代码运行的结果成功或报错智能体获得了最直接的反馈。如果代码报错错误堆栈信息会成为下一轮“推理”的关键输入指导它进行修正。这实现了闭环验证。任务分解与规划复杂的编程任务如“实现一个用户登录API”可以被拆解成一系列子步骤检查现有路由、创建模型、编写控制器、测试端点。ReAct 的推理步骤天然适合这种逐步推进的任务解决方式。从源码的设计中可以看出Claude Code 并不是简单套用 ReAct 论文中的模板而是对其进行了深度定制使其完全服务于“在真实开发环境中解决真实编程问题”这一核心目标。其架构设计紧密围绕工具调用、状态管理和循环控制这三个支柱展开。3. 主循环核心模块深度解析让我们进入正题拆解 Claude Code 源码中 ReAct 主循环的核心模块。整个循环可以看作一个状态机其核心类通常被命名为Agent或ReActLoop。以下是其关键组成部分的解析。3.1 状态管理AgentState一切始于状态。主循环需要一个数据结构来保存当前任务执行的所有上下文。在 Claude Code 中这个结构通常被定义为AgentState或类似的类。# 示例性代码展示状态管理的核心字段 class AgentState: def __init__(self): # 核心输入 self.task: str # 用户提出的原始任务如“修复这个bug” self.working_dir: Path Path(.) # 智能体操作的工作目录 # 会话历史与上下文 self.conversation_history: List[Message] [] # 包含用户、助手、工具消息的完整对话 self.code_context: Dict[str, str] {} # 当前关注的相关代码文件片段key为文件路径 # 执行环境 self.interpreter: CodeInterpreter None # 代码解释器实例用于执行代码 self.file_system: FileSystemTool None # 文件系统工具实例 # 循环控制状态 self.max_steps: int 20 # 最大循环步数防止无限循环 self.current_step: int 0 self.is_task_complete: bool False # 任务完成标志 self.last_error: Optional[str] None # 上一步行动产生的错误信息设计要点解析分离会话与代码上下文conversation_history保存了自然语言对话供模型理解任务脉络code_context则专门存储提取的代码供模型分析引用。这种分离提高了信息检索的效率。工具实例化状态中持有interpreter和file_system等工具实例意味着这些工具是有状态的。例如代码解释器会维护一个持久的运行时会话如一个 Jupyter Kernel使得前后执行的代码可以共享变量这对于调试和迭代至关重要。轻量级与可序列化AgentState的设计需要考虑可能的中断与恢复。虽然持有工具实例但核心数据如任务、历史、当前步骤应易于序列化保存。实操心得在实际模仿实现时code_context的管理策略是性能关键。不要每次都把整个文件内容塞进去。更佳实践是使用一个ContextManager它根据模型当前关注的焦点如错误行号、提到的函数名动态地从工作目录中读取相关的代码块比如函数定义及其上下几行这样可以有效控制输入给模型的令牌数量节省成本并提升推理速度。3.2 推理引擎ReasoningModule这是智能体的“大脑”负责根据当前状态推理出下一步该做什么。在 Claude Code 的实现中这通常不是一个独立的复杂模块而是通过精心设计的提示词引导大模型如 Claude 3输出结构化的“思考”和“行动计划”。核心流程如下组装提示词将AgentState中的任务、对话历史、相关代码、最近错误等信息按照预定模板格式化成一段给模型的提示。调用模型请求大模型生成回复。解析输出模型的回复需要被解析成结构化的数据。Claude Code 极大可能利用了 Claude 模型对结构化输出如 JSON的良好支持或者使用正则表达式进行抽取。# 提示词模板示例简化 REASONING_PROMPT_TEMPLATE 你是一个在{working_dir}目录下工作的资深软件工程师。 你的最终任务是{task} ## 当前上下文 {conversation_history_snippet} {code_context_snippet} {error_context_if_any} ## 可用工具 1. 执行代码可以运行Python/Shell代码片段并返回结果或错误。 2. 读写文件可以读取、创建、编辑、删除文件。 3. 搜索代码可以在项目中搜索特定模式。 ## 要求 请你逐步思考 1. 首先分析当前状况和任务目标。 2. 然后决定下一步是使用工具还是认为任务已完成。 3. 如果使用工具请严格按照以下JSON格式输出 json {{ thought: 你的详细推理过程解释为什么这么做。, action: 工具名称如 run_code 或 read_file, action_input: {{ // 工具参数 code: 要执行的代码, path: 文件路径 // ... 其他参数 }} }}如果任务完成请输出{{ thought: 解释任务如何完成。, action: finalize, action_input: {{summary: 任务完成总结}} }}现在开始你的思考 **关键设计解析** * **思维链强制**提示词中明确要求模型输出“thought”这不仅仅是给用户看的更重要的是让模型将自己的推理过程“外化”这通常能显著提升行动决策的质量和一致性。 * **结构化输出**要求模型输出严格的 JSON 格式这是实现程序自动化解析的关键。Claude 模型在这方面非常可靠大大降低了后处理的复杂度。 * **上下文精选**提示词中插入的 {code_context_snippet} 等需要由上游模块从 AgentState 中提取最相关的部分而不是全部倒入这是控制成本的核心。 **注意事项**模型有时会“不听话”输出格式不正确的 JSON 或忘记输出“thought”。在源码中你一定会看到健壮的**输出解析和错误处理逻辑**。例如使用 json.loads() 配合 try-catch并在解析失败时将模型的错误输出连同解析失败信息一起作为新的上下文喂回给模型要求它纠正。这是一个典型的自我修正循环。 ### 3.3 工具执行器ToolExecutor 推理模块决定了“做什么”工具执行器则负责“怎么做”。它是连接智能体“思考”和真实世界“影响”的桥梁。Claude Code 的工具集是其强大能力的直接体现。 python class ToolExecutor: def __init__(self, state: AgentState): self.state state self._tools { run_code: self._execute_code, read_file: self._read_file, write_file: self._write_file, search_files: self._grep, run_shell: self._run_shell_command, } def execute(self, action: str, action_input: Dict) - Dict: 执行指定工具并返回标准化结果 if action not in self._tools: return {error: f未知工具: {action}} try: # 调用对应的工具函数 result self._tools[action](**action_input) # 标准化成功返回 return { status: success, output: result, observation: str(result)[:500] # 截断过长的输出 } except Exception as e: # 标准化错误返回错误信息对后续推理至关重要 return { status: error, output: None, observation: f工具执行失败: {type(e).__name__}: {e} } def _execute_code(self, code: str, language: str python) - str: 核心在持久化解释器中执行代码 # 使用 state 中维护的解释器实例 return self.state.interpreter.run(code, language) def _read_file(self, path: str) - str: 读取文件内容路径相对于 working_dir full_path self.state.working_dir / path return full_path.read_text() def _write_file(self, path: str, content: str, append: bool False) - str: 写入文件。注意这是改变真实环境的操作 full_path self.state.working_dir / path if append: full_path.write_text(full_path.read_text() content) else: full_path.write_text(content) return f文件 {path} 写入成功。安全与设计解析沙箱化执行_execute_code是重中之重。Claude Code 的源码中interpreter.run()的背后一定是一个严格的沙箱环境。它可能基于 Docker 容器、安全的进程隔离或专门的代码沙箱库如pysandbox。这是防止恶意或错误代码破坏宿主机的生命线。路径隔离所有文件操作都基于self.state.working_dir。这意味着智能体被“禁锢”在指定的项目目录内无法访问或修改系统其他文件这是最基本的安全边界。标准化返回无论工具执行成功与否execute方法都返回结构一致的字典。observation字段的内容会直接进入下一轮推理的提示词。因此将错误信息清晰、友好地格式化在这里非常重要要便于模型理解。工具设计的原子性工具应该足够“原子”每个工具只做一件事。比如不设计一个“运行测试并修复”的复杂工具而是拆成“运行测试”、“读取测试输出”、“编辑文件”等多个工具由模型通过多次调用来组合完成复杂任务。这提高了系统的灵活性和可解释性。3.4 循环控制器ReActLoop.run()这是将所有模块串联起来的“总指挥”。它的逻辑清晰而严谨体现了 ReAct 模式的精髓。class ReActLoop: def __init__(self, llm_client, initial_state: AgentState): self.llm llm_client self.state initial_state self.tool_executor ToolExecutor(initial_state) def run(self) - AgentState: 主循环入口 while not self.state.is_task_complete and self.state.current_step self.state.max_steps: self.state.current_step 1 print(f\n 步骤 {self.state.current_step} ) # 1. 推理阶段决定下一步行动 reasoning_result self._reasoning_step() if reasoning_result.get(action) finalize: self.state.is_task_complete True print(f任务完成: {reasoning_result.get(summary)}) break # 2. 行动阶段执行工具 action reasoning_result[action] action_input reasoning_result[action_input] print(f执行动作: {action}, 输入: {action_input}) tool_result self.tool_executor.execute(action, action_input) # 3. 观察阶段记录结果到状态 self._update_state_with_observation(tool_result, reasoning_result) # 4. 检查终止条件如连续失败 if self._should_early_stop(): print(提前终止连续失败次数过多。) break return self.state def _reasoning_step(self) - Dict: 封装提示词构建、调用LLM、解析输出的全过程 prompt self._construct_prompt() # 使用状态信息构建提示词 llm_response self.llm.complete(prompt) return self._parse_llm_response(llm_response) # 解析出 thought, action, action_input def _update_state_with_observation(self, tool_result: Dict, reasoning_result: Dict): 将本轮结果记录到对话历史中供下一轮推理使用 # 添加助手的“思考”消息 self.state.conversation_history.append({ role: assistant, content: reasoning_result[thought] }) # 添加“工具调用”消息可选取决于如何设计消息角色 # 添加“工具结果”观察消息 self.state.conversation_history.append({ role: user, # 通常将工具返回视为“环境”的输入用 user 角色模拟 content: f工具执行结果: {tool_result[observation]} }) # 如果执行出错更新错误状态 if tool_result[status] error: self.state.last_error tool_result[observation] else: self.state.last_error None # 根据工具结果可能更新 code_context # 例如如果刚读取了一个新文件将其内容提取关键部分加入上下文 self._refresh_code_context()循环控制逻辑解析显式状态更新每一轮循环后都必须显式地将“思考”、“行动结果”更新到state.conversation_history中。这是模型拥有“记忆”的关键。错误处理与早期终止_should_early_stop()函数是一个安全阀。常见的策略是如果连续 N 步比如5步工具执行都失败了或者模型陷入了重复的动作循环就主动终止任务避免浪费资源。这需要设计一些启发式规则来检测循环。上下文刷新策略_refresh_code_context()是一个智能函数。它不能简单地把所有文件内容都加进去。它的策略可能是关注最近被读取或修改的文件关注错误堆栈中提到的文件和行号根据模型“思考”中提到的函数名或类名去项目中进行轻量级检索如使用ripgrep并加载相关片段。4. 实战推演一个完整的调试循环让我们通过一个虚构但非常典型的例子看看主循环是如何实际运作的。假设任务是在一个 Flask 项目中“修复/login端点的 500 错误”。步骤 0 (初始状态):state.task: “修复/login端点的 500 错误”state.conversation_history: [用户消息]state.code_context: {} (初始为空)state.last_error: None步骤 1 (推理):模型思考: “我需要先复现这个错误。应该检查一下app.py或相关路由文件找到/login路由的定义然后尝试运行一个简单的测试请求来获取具体的错误信息。”模型输出:{ thought: 首先我需要定位到 /login 路由的处理函数。让我先查看项目的主要应用文件。, action: read_file, action_input: {path: app.py} }步骤 1 (行动与观察):ToolExecutor读取app.py。观察结果: “文件内容显示/login路由在auth.py文件中。app.py导入了auth蓝图。”步骤 2 (推理):新的提示词包含了上一步的思考和观察结果。模型思考: “好的路由在auth.py。现在读取这个文件查看login函数的具体实现。”模型输出:{ thought: 现在读取 auth.py 文件以检查 login 函数。, action: read_file, action_input: {path: routes/auth.py} }步骤 2 (行动与观察):读取auth.py。观察结果: “文件内容显示login函数尝试从request.form获取username和password然后调用User.query.filter_by(...).first()。”步骤 3 (推理):模型思考: “我需要运行这个 Flask 应用并模拟一个对/login的 POST 请求来触发 500 错误并查看堆栈跟踪。首先确保依赖已安装然后启动一个测试服务器。”模型输出:{ thought: 我将启动一个简单的 Flask 测试服务器并向 /login 发送一个请求来捕获错误。, action: run_code, action_input: { code: import sys\nsys.path.insert(0, .)\nfrom app import app\napp.testing True\nwith app.test_client() as client:\n resp client.post(/login, data{username:test,password:test})\n print(fStatus: {resp.status_code})\n print(fResponse: {resp.data.decode()})\n, language: python } }步骤 3 (行动与观察):在沙箱中执行上述代码。观察结果 (错误): “工具执行失败:AttributeError: NoneType object has no attribute query。错误发生在auth.py第 15 行User.query.filter_by(...)。”步骤 4 (推理):模型思考: “User对象是None这意味着 SQLAlchemy 的User模型可能没有正确导入或定义。让我检查一下models.py文件并确认User类是否存在以及是否导入了db对象。”模型输出:{ thought: 错误表明 User 模型未定义或未与数据库绑定。检查 models.py 和数据库初始化代码。, action: read_file, action_input: {path: models.py} }... 循环继续。最终模型可能会发现是models.py中缺少db SQLAlchemy(app)的初始化或者User类没有继承db.Model。它会通过“写文件”工具进行修复然后再次运行测试进行验证直到错误消失最终输出finalize动作。这个推演清晰地展示了 ReAct 主循环如何将复杂的调试任务分解为一系列可执行的“读文件”、“运行代码”等原子操作并通过执行反馈逐步逼近问题根源。5. 性能优化与工程实践要点阅读源码时你会发现 Claude Code 的工程实现远不止上述基础循环。以下是一些关键的优化点和工程实践它们决定了系统的可用性和效率。5.1 上下文长度管理与优化这是成本与效果平衡的核心。大模型的上下文窗口是宝贵资源如 128K tokens必须高效利用。对话历史摘要不可能把所有历史对话都塞进提示词。Claude Code 很可能实现了“滑动窗口”或“摘要”策略。例如只保留最近 10 轮交互的完整消息对于更早的消息则使用另一个 LLM 调用将其总结成一段简短的背景描述。代码上下文的动态加载code_context不是静态的。一个高效的ContextManager会做以下工作监听焦点从模型的“思考”和错误信息中提取出它正在关注的文件名、函数名、类名、行号。智能检索使用轻量级代码分析如 AST 解析或文本搜索只加载与焦点相关的代码片段。例如当模型提到handle_login函数时不仅加载这个函数还加载它的调用者、它调用的函数、以及同文件中的相关类和导入语句。去除冗余避免在不同轮次中重复加载相同的长代码块。5.2 工具执行的稳定性保障工具调用是故障高发区必须有鲁棒的设计。超时与资源限制run_code工具必须设置严格的执行超时如30秒和资源限制内存、CPU。防止一段死循环代码拖垮整个服务。副作用管理与回滚write_file是危险操作。高级的实现会考虑版本备份在修改文件前自动备份原文件。如果后续步骤导致更严重错误可以提示用户或智能体回滚。沙箱文件系统更彻底的方案是让智能体在一个虚拟文件系统如内存文件系统或 Docker 容器内的独立目录中操作所有修改在任务最终确认成功后才一次性提交到真实文件系统。工具结果的后处理代码执行的输出可能非常冗长如pip install的日志。直接塞给模型会浪费 tokens 且干扰注意力。需要有一个OutputProcessor来截断过长的输出只保留头部和尾部关键行。提取关键信息例如从pytest的输出中提取出失败测试的名称和错误断言从ImportError中提取缺失的模块名。5.3 提示词工程与少样本学习Claude Code 的强大很大程度上源于其精心设计的提示词。在源码中你可能会发现一个“提示词模板库”。角色扮演与约束开头的系统提示词如“你是一个资深软件工程师...”设定了角色的行为边界和专业性。少样本示例提示词中很可能包含了几个精心构造的“示例对话”Few-shot Examples。例如用户帮我写一个计算斐波那契数列的函数。 助手思考我需要创建一个 Python 函数。首先检查当前目录是否有相关文件...思考过程 助手行动{action: write_file, action_input: {path: fib.py, content: def fib(n): ...}} 工具结果文件写入成功。 用户现在测试一下 fib(5)。 助手思考我需要运行刚创建的文件中的函数... 助手行动{action: run_code, action_input: {code: from fib import fib; print(fib(5)), language: python}} 工具结果5这些示例教会了模型 ReAct 的格式和思考方式。动态提示词调整根据任务类型调试、重构、新功能开发或当前状态是否连续出错系统可能会切换不同的提示词模板以更好地引导模型。5.4 循环终止与异常处理策略一个健壮的循环必须知道何时停止。成功终止模型主动输出finalize动作。失败终止步数限制max_steps是硬限制。循环检测如果最近几步的“行动”序列出现重复模式如反复读写同一个文件但状态不变则触发终止。用户中断实现一个信号监听允许用户手动停止。优雅降级当循环异常终止时系统应能保存当前AgentState至少是对话历史和代码修改并生成一份报告说明已完成的步骤和遇到的关键问题方便用户接手。6. 从源码到实践构建你自己的简易 ReAct 智能体理解了 Claude Code 的设计精髓后我们可以尝试用最简化的方式构建一个自己的 ReAct 代码智能体。这里提供一个高度浓缩但可运行的骨架。import json import subprocess from pathlib import Path from typing import Dict, List # 假设使用 OpenAI API你需要安装 openai 库并设置 API_KEY import openai class SimpleCodeAgent: def __init__(self, task: str, work_dir: str ., model: str gpt-4): self.task task self.work_dir Path(work_dir) self.model model self.client openai.OpenAI(api_keyyour-api-key) self.conversation [{role: user, content: task}] self.max_steps 15 def run(self): for step in range(self.max_steps): print(f\n--- 步骤 {step1} ---) # 1. 推理 response self._call_llm() try: action_data self._extract_action(response) except ValueError as e: print(f解析模型输出失败: {e}) self.conversation.append({role: user, content: f你的输出格式不正确。请严格按照JSON格式输出。错误{e}}) continue if action_data.get(action) finalize: print(f任务完成: {action_data.get(summary, )}) break # 2. 行动 result self._execute_tool(action_data[action], action_data.get(action_input, {})) observation result[observation] # 3. 观察与记录 self.conversation.append({role: assistant, content: action_data.get(thought, )}) self.conversation.append({role: user, content: f结果: {observation}}) print(f观察: {observation[:200]}...) def _call_llm(self) - str: prompt self._build_prompt() response self.client.chat.completions.create( modelself.model, messages[{role: system, content: 你是一个Python编程助手。请用JSON格式输出你的思考和行动。}] self.conversation[-10:], # 滑动窗口 temperature0.2, ) return response.choices[0].message.content def _build_prompt(self): # 简化的提示词 return self.conversation def _extract_action(self, response: str) - Dict: # 尝试从响应中提取JSON代码块 import re json_match re.search(rjson\n(.*?)\n, response, re.DOTALL) if json_match: json_str json_match.group(1) else: json_str response.strip() return json.loads(json_str) def _execute_tool(self, action: str, inputs: Dict) - Dict: if action run_python: code inputs.get(code, ) return self._run_python_code(code) elif action read_file: path inputs.get(path, ) return self._read_file(path) elif action write_file: path inputs.get(path, ) content inputs.get(content, ) return self._write_file(path, content) else: return {observation: f错误未知动作 {action}} def _run_python_code(self, code: str) - Dict: try: # 警告简易实现无沙箱仅用于演示生产环境必须使用Docker等隔离。 result subprocess.run( [python3, -c, code], cwdself.work_dir, capture_outputTrue, textTrue, timeout10 ) if result.returncode 0: return {observation: f执行成功:\n{result.stdout}} else: return {observation: f执行失败:\n{result.stderr}} except subprocess.TimeoutExpired: return {observation: 错误代码执行超时10秒。} except Exception as e: return {observation: f错误{e}} def _read_file(self, path: str) - Dict: try: full_path self.work_dir / path content full_path.read_text() return {observation: f文件内容:\n{content}} except Exception as e: return {observation: f读取文件失败: {e}} def _write_file(self, path: str, content: str) - Dict: try: full_path self.work_dir / path full_path.write_text(content) return {observation: f文件 {path} 已写入。} except Exception as e: return {observation: f写入文件失败: {e}} # 使用示例 if __name__ __main__: agent SimpleCodeAgent( task请检查当前目录下是否有 requirements.txt 文件如果有请列出其中的前三个包。, work_dir./sample_project ) agent.run()这个简易实现的重点与警告核心流程完整它包含了 ReAct 的推理、行动、观察循环以及状态对话历史管理。极度简化没有代码上下文管理没有复杂的提示词工程工具很少。安全警告_run_python_code直接使用subprocess存在严重安全风险绝对不可用于生产环境或处理不可信代码。真实系统必须使用 Docker 或专用沙箱。健壮性不足错误处理、循环检测、上下文长度控制都非常基础。这个例子旨在帮助你将前面解析的理论映射到具体的代码结构上。要构建一个可用的系统你需要在此基础上逐一实现我们前面讨论的各个优化模块。7. 常见陷阱与排查指南在实际运行 ReAct 智能体时你会遇到各种各样的问题。以下是一些典型陷阱及其排查思路。7.1 模型不按格式输出现象模型回复是自然语言而不是你要求的 JSON 格式。原因提示词不够清晰少样本示例不足模型温度temperature参数过高。解决强化格式指令在系统提示词和用户提示词中用非常明确、加重的语言描述格式要求。例如“你必须且只能输出一个 JSON 对象包含 thought, action, action_input 字段。”提供更清晰的示例在少样本示例中展示格式正确的完整交互。降低温度将temperature设为 0.1 或 0.2降低随机性使输出更确定。使用模型的结构化输出功能如果使用的 API 支持如 OpenAI 的response_format或 Anthropic 的tool_use优先使用这些功能它们比文本指令更可靠。7.2 智能体陷入死循环现象智能体反复执行相似操作如反复读取同一个文件无法推进任务。原因模型未能从观察结果中提取有效信息来更新计划任务本身模糊或不可实现。解决增强观察结果的信息量确保工具返回的observation不仅包含“成功/失败”还包含对下一步推理有用的关键数据。例如运行测试失败时不要只返回“测试失败”而要提取具体的失败断言和行号。在提示词中加入“防呆”引导例如“如果你连续三次尝试相同或类似的操作都未能取得进展请重新评估你的计划或者尝试一个完全不同的方法。”实现循环检测器在AgentState中记录最近 N 步的行动和观察摘要。如果检测到重复模式则在提示词中明确告知模型“注意你似乎陷入了循环。你最近三步都在做 X。请尝试一个不同的策略。”设置硬性步数限制这是最后的安全网。7.3 工具执行结果太长干扰模型现象运行pip install或npm start产生了数百行日志模型被淹没在无关信息中。原因未对工具输出进行后处理。解决为每个工具设计输出过滤器run_code捕获最后 20 行输出如果遇到错误非零退出码则优先捕获错误信息。search_files只返回匹配的行和前后几行上下文而不是整个文件。总结冗长输出对于特别长的输出可以调用另一个轻量级模型如 GPT-3.5-turbo进行摘要再将摘要放入观察中。例如“安装过程成功生成了大量日志。摘要成功安装了 package-a (v1.2), package-b (v3.4)无错误。”提供“查看更多”选项在观察结果末尾注明“输出已截断完整日志共XXX行”。如果模型在后续思考中明确要求查看完整日志再通过一个新工具调用提供。7.4 代码修改引入新错误现象智能体为了修复一个错误而修改了代码但修改后产生了新的、更隐蔽的错误。原因模型缺乏对代码库的全局理解修改是局部的、基于试探的。解决强制运行测试在智能体完成一组修改后在提示词中要求它必须运行相关的单元测试或集成测试并将测试结果作为下一轮观察的核心依据。实施“小步快跑”策略鼓励模型进行更小、更增量的修改。每修改一个文件就运行一次相关的检查如语法检查python -m py_compile或导入检查。引入代码分析工具将lint如flake8、type check如mypy作为工具集成进来。在模型提交修改前自动运行这些检查并将警告和错误作为观察的一部分反馈给模型。调试一个 ReAct 智能体本身就是一个递归过程观察它的失败模式然后修改你的提示词、工具设计或状态管理逻辑再观察效果。这个过程本身就充满了“智能体”的味道。
Claude Code ReAct主循环源码解析：从代码生成到执行的AI编程范式跃迁
1. 项目概述从“代码生成”到“代码执行”的范式跃迁如果你用过 Claude Code 或者 GitHub Copilot肯定对它们“写代码”的能力印象深刻。你写个注释它就能给你生成一段看起来不错的函数。但不知道你有没有遇到过这种情况生成的代码逻辑上没问题但一运行就报错要么是导入了不存在的包要么是调用了错误的 API。这背后的根本原因在于传统的代码生成模型本质上是一个“文本预测”模型——它根据你给的上下文预测下一个最可能出现的“词元”。它并不真正理解代码的“运行环境”更无法验证代码的“执行结果”。而 Claude Code 源码中揭示的ReAct 主循环正是为了解决这个问题而生的核心架构。它不是一个简单的代码补全工具而是一个具备“思考-行动-观察”能力的智能体。简单来说它让 AI 不再只是“写”代码而是开始“执行”代码并根据执行结果进行“修正”和“迭代”。这就像是从一个只会纸上谈兵的参谋变成了一个能亲自上阵、根据战场反馈调整战术的指挥官。这个主循环机制是当前 AI 编程助手从“辅助生成”迈向“自主执行”的关键一步。它不仅仅是 Claude Code 的技术内核更代表了下一代开发工具的发展方向。无论是想深入理解 AI 如何与真实开发环境交互还是希望在自己的项目中引入类似的智能体能力拆解 ReAct 主循环都是绝佳的切入点。接下来我将带你深入源码看看这个循环是如何一步步运转起来的。2. 核心架构与设计哲学为什么是 ReAct在深入代码之前我们必须先理解 Claude Code 选择 ReAct 框架的底层逻辑。ReAct 是 “Reasoning Acting” 的缩写其核心思想是让智能体交替进行“推理”和“行动”。在代码生成的场景下这被具体化为推理分析当前任务、已有代码、错误信息规划下一步要做什么。行动执行一个具体操作比如运行一段代码、读取一个文件、安装一个依赖。观察获取行动的结果成功输出、错误信息、文件内容变化。这个循环会一直持续直到任务被完成或达到终止条件。那么为什么这种模式特别适合代码生成呢2.1 传统代码生成的局限性传统的端到端代码生成模型如基于 Transformer 的大模型存在几个固有缺陷环境盲区模型不知道你项目的package.json里具体有哪些依赖和版本也不知道当前工作目录的文件结构。它只能基于训练数据中的“常见模式”进行猜测。静态幻觉模型可能会“幻想”出一些不存在的 API 或函数签名因为它学到的知识可能过时或者与你的特定项目环境不符。缺乏验证生成的代码没有经过“运行”这一终极测试。语法正确不代表逻辑正确更不代表能在你的环境中运行。2.2 ReAct 带来的范式优势Claude Code 的 ReAct 主循环直接针对了上述痛点动态环境感知通过“行动”如ls,cat智能体可以实时读取项目文件、查看日志、检查依赖从而获得最新的、准确的上下文信息。这解决了“环境盲区”问题。执行反馈驱动通过“观察”代码运行的结果成功或报错智能体获得了最直接的反馈。如果代码报错错误堆栈信息会成为下一轮“推理”的关键输入指导它进行修正。这实现了闭环验证。任务分解与规划复杂的编程任务如“实现一个用户登录API”可以被拆解成一系列子步骤检查现有路由、创建模型、编写控制器、测试端点。ReAct 的推理步骤天然适合这种逐步推进的任务解决方式。从源码的设计中可以看出Claude Code 并不是简单套用 ReAct 论文中的模板而是对其进行了深度定制使其完全服务于“在真实开发环境中解决真实编程问题”这一核心目标。其架构设计紧密围绕工具调用、状态管理和循环控制这三个支柱展开。3. 主循环核心模块深度解析让我们进入正题拆解 Claude Code 源码中 ReAct 主循环的核心模块。整个循环可以看作一个状态机其核心类通常被命名为Agent或ReActLoop。以下是其关键组成部分的解析。3.1 状态管理AgentState一切始于状态。主循环需要一个数据结构来保存当前任务执行的所有上下文。在 Claude Code 中这个结构通常被定义为AgentState或类似的类。# 示例性代码展示状态管理的核心字段 class AgentState: def __init__(self): # 核心输入 self.task: str # 用户提出的原始任务如“修复这个bug” self.working_dir: Path Path(.) # 智能体操作的工作目录 # 会话历史与上下文 self.conversation_history: List[Message] [] # 包含用户、助手、工具消息的完整对话 self.code_context: Dict[str, str] {} # 当前关注的相关代码文件片段key为文件路径 # 执行环境 self.interpreter: CodeInterpreter None # 代码解释器实例用于执行代码 self.file_system: FileSystemTool None # 文件系统工具实例 # 循环控制状态 self.max_steps: int 20 # 最大循环步数防止无限循环 self.current_step: int 0 self.is_task_complete: bool False # 任务完成标志 self.last_error: Optional[str] None # 上一步行动产生的错误信息设计要点解析分离会话与代码上下文conversation_history保存了自然语言对话供模型理解任务脉络code_context则专门存储提取的代码供模型分析引用。这种分离提高了信息检索的效率。工具实例化状态中持有interpreter和file_system等工具实例意味着这些工具是有状态的。例如代码解释器会维护一个持久的运行时会话如一个 Jupyter Kernel使得前后执行的代码可以共享变量这对于调试和迭代至关重要。轻量级与可序列化AgentState的设计需要考虑可能的中断与恢复。虽然持有工具实例但核心数据如任务、历史、当前步骤应易于序列化保存。实操心得在实际模仿实现时code_context的管理策略是性能关键。不要每次都把整个文件内容塞进去。更佳实践是使用一个ContextManager它根据模型当前关注的焦点如错误行号、提到的函数名动态地从工作目录中读取相关的代码块比如函数定义及其上下几行这样可以有效控制输入给模型的令牌数量节省成本并提升推理速度。3.2 推理引擎ReasoningModule这是智能体的“大脑”负责根据当前状态推理出下一步该做什么。在 Claude Code 的实现中这通常不是一个独立的复杂模块而是通过精心设计的提示词引导大模型如 Claude 3输出结构化的“思考”和“行动计划”。核心流程如下组装提示词将AgentState中的任务、对话历史、相关代码、最近错误等信息按照预定模板格式化成一段给模型的提示。调用模型请求大模型生成回复。解析输出模型的回复需要被解析成结构化的数据。Claude Code 极大可能利用了 Claude 模型对结构化输出如 JSON的良好支持或者使用正则表达式进行抽取。# 提示词模板示例简化 REASONING_PROMPT_TEMPLATE 你是一个在{working_dir}目录下工作的资深软件工程师。 你的最终任务是{task} ## 当前上下文 {conversation_history_snippet} {code_context_snippet} {error_context_if_any} ## 可用工具 1. 执行代码可以运行Python/Shell代码片段并返回结果或错误。 2. 读写文件可以读取、创建、编辑、删除文件。 3. 搜索代码可以在项目中搜索特定模式。 ## 要求 请你逐步思考 1. 首先分析当前状况和任务目标。 2. 然后决定下一步是使用工具还是认为任务已完成。 3. 如果使用工具请严格按照以下JSON格式输出 json {{ thought: 你的详细推理过程解释为什么这么做。, action: 工具名称如 run_code 或 read_file, action_input: {{ // 工具参数 code: 要执行的代码, path: 文件路径 // ... 其他参数 }} }}如果任务完成请输出{{ thought: 解释任务如何完成。, action: finalize, action_input: {{summary: 任务完成总结}} }}现在开始你的思考 **关键设计解析** * **思维链强制**提示词中明确要求模型输出“thought”这不仅仅是给用户看的更重要的是让模型将自己的推理过程“外化”这通常能显著提升行动决策的质量和一致性。 * **结构化输出**要求模型输出严格的 JSON 格式这是实现程序自动化解析的关键。Claude 模型在这方面非常可靠大大降低了后处理的复杂度。 * **上下文精选**提示词中插入的 {code_context_snippet} 等需要由上游模块从 AgentState 中提取最相关的部分而不是全部倒入这是控制成本的核心。 **注意事项**模型有时会“不听话”输出格式不正确的 JSON 或忘记输出“thought”。在源码中你一定会看到健壮的**输出解析和错误处理逻辑**。例如使用 json.loads() 配合 try-catch并在解析失败时将模型的错误输出连同解析失败信息一起作为新的上下文喂回给模型要求它纠正。这是一个典型的自我修正循环。 ### 3.3 工具执行器ToolExecutor 推理模块决定了“做什么”工具执行器则负责“怎么做”。它是连接智能体“思考”和真实世界“影响”的桥梁。Claude Code 的工具集是其强大能力的直接体现。 python class ToolExecutor: def __init__(self, state: AgentState): self.state state self._tools { run_code: self._execute_code, read_file: self._read_file, write_file: self._write_file, search_files: self._grep, run_shell: self._run_shell_command, } def execute(self, action: str, action_input: Dict) - Dict: 执行指定工具并返回标准化结果 if action not in self._tools: return {error: f未知工具: {action}} try: # 调用对应的工具函数 result self._tools[action](**action_input) # 标准化成功返回 return { status: success, output: result, observation: str(result)[:500] # 截断过长的输出 } except Exception as e: # 标准化错误返回错误信息对后续推理至关重要 return { status: error, output: None, observation: f工具执行失败: {type(e).__name__}: {e} } def _execute_code(self, code: str, language: str python) - str: 核心在持久化解释器中执行代码 # 使用 state 中维护的解释器实例 return self.state.interpreter.run(code, language) def _read_file(self, path: str) - str: 读取文件内容路径相对于 working_dir full_path self.state.working_dir / path return full_path.read_text() def _write_file(self, path: str, content: str, append: bool False) - str: 写入文件。注意这是改变真实环境的操作 full_path self.state.working_dir / path if append: full_path.write_text(full_path.read_text() content) else: full_path.write_text(content) return f文件 {path} 写入成功。安全与设计解析沙箱化执行_execute_code是重中之重。Claude Code 的源码中interpreter.run()的背后一定是一个严格的沙箱环境。它可能基于 Docker 容器、安全的进程隔离或专门的代码沙箱库如pysandbox。这是防止恶意或错误代码破坏宿主机的生命线。路径隔离所有文件操作都基于self.state.working_dir。这意味着智能体被“禁锢”在指定的项目目录内无法访问或修改系统其他文件这是最基本的安全边界。标准化返回无论工具执行成功与否execute方法都返回结构一致的字典。observation字段的内容会直接进入下一轮推理的提示词。因此将错误信息清晰、友好地格式化在这里非常重要要便于模型理解。工具设计的原子性工具应该足够“原子”每个工具只做一件事。比如不设计一个“运行测试并修复”的复杂工具而是拆成“运行测试”、“读取测试输出”、“编辑文件”等多个工具由模型通过多次调用来组合完成复杂任务。这提高了系统的灵活性和可解释性。3.4 循环控制器ReActLoop.run()这是将所有模块串联起来的“总指挥”。它的逻辑清晰而严谨体现了 ReAct 模式的精髓。class ReActLoop: def __init__(self, llm_client, initial_state: AgentState): self.llm llm_client self.state initial_state self.tool_executor ToolExecutor(initial_state) def run(self) - AgentState: 主循环入口 while not self.state.is_task_complete and self.state.current_step self.state.max_steps: self.state.current_step 1 print(f\n 步骤 {self.state.current_step} ) # 1. 推理阶段决定下一步行动 reasoning_result self._reasoning_step() if reasoning_result.get(action) finalize: self.state.is_task_complete True print(f任务完成: {reasoning_result.get(summary)}) break # 2. 行动阶段执行工具 action reasoning_result[action] action_input reasoning_result[action_input] print(f执行动作: {action}, 输入: {action_input}) tool_result self.tool_executor.execute(action, action_input) # 3. 观察阶段记录结果到状态 self._update_state_with_observation(tool_result, reasoning_result) # 4. 检查终止条件如连续失败 if self._should_early_stop(): print(提前终止连续失败次数过多。) break return self.state def _reasoning_step(self) - Dict: 封装提示词构建、调用LLM、解析输出的全过程 prompt self._construct_prompt() # 使用状态信息构建提示词 llm_response self.llm.complete(prompt) return self._parse_llm_response(llm_response) # 解析出 thought, action, action_input def _update_state_with_observation(self, tool_result: Dict, reasoning_result: Dict): 将本轮结果记录到对话历史中供下一轮推理使用 # 添加助手的“思考”消息 self.state.conversation_history.append({ role: assistant, content: reasoning_result[thought] }) # 添加“工具调用”消息可选取决于如何设计消息角色 # 添加“工具结果”观察消息 self.state.conversation_history.append({ role: user, # 通常将工具返回视为“环境”的输入用 user 角色模拟 content: f工具执行结果: {tool_result[observation]} }) # 如果执行出错更新错误状态 if tool_result[status] error: self.state.last_error tool_result[observation] else: self.state.last_error None # 根据工具结果可能更新 code_context # 例如如果刚读取了一个新文件将其内容提取关键部分加入上下文 self._refresh_code_context()循环控制逻辑解析显式状态更新每一轮循环后都必须显式地将“思考”、“行动结果”更新到state.conversation_history中。这是模型拥有“记忆”的关键。错误处理与早期终止_should_early_stop()函数是一个安全阀。常见的策略是如果连续 N 步比如5步工具执行都失败了或者模型陷入了重复的动作循环就主动终止任务避免浪费资源。这需要设计一些启发式规则来检测循环。上下文刷新策略_refresh_code_context()是一个智能函数。它不能简单地把所有文件内容都加进去。它的策略可能是关注最近被读取或修改的文件关注错误堆栈中提到的文件和行号根据模型“思考”中提到的函数名或类名去项目中进行轻量级检索如使用ripgrep并加载相关片段。4. 实战推演一个完整的调试循环让我们通过一个虚构但非常典型的例子看看主循环是如何实际运作的。假设任务是在一个 Flask 项目中“修复/login端点的 500 错误”。步骤 0 (初始状态):state.task: “修复/login端点的 500 错误”state.conversation_history: [用户消息]state.code_context: {} (初始为空)state.last_error: None步骤 1 (推理):模型思考: “我需要先复现这个错误。应该检查一下app.py或相关路由文件找到/login路由的定义然后尝试运行一个简单的测试请求来获取具体的错误信息。”模型输出:{ thought: 首先我需要定位到 /login 路由的处理函数。让我先查看项目的主要应用文件。, action: read_file, action_input: {path: app.py} }步骤 1 (行动与观察):ToolExecutor读取app.py。观察结果: “文件内容显示/login路由在auth.py文件中。app.py导入了auth蓝图。”步骤 2 (推理):新的提示词包含了上一步的思考和观察结果。模型思考: “好的路由在auth.py。现在读取这个文件查看login函数的具体实现。”模型输出:{ thought: 现在读取 auth.py 文件以检查 login 函数。, action: read_file, action_input: {path: routes/auth.py} }步骤 2 (行动与观察):读取auth.py。观察结果: “文件内容显示login函数尝试从request.form获取username和password然后调用User.query.filter_by(...).first()。”步骤 3 (推理):模型思考: “我需要运行这个 Flask 应用并模拟一个对/login的 POST 请求来触发 500 错误并查看堆栈跟踪。首先确保依赖已安装然后启动一个测试服务器。”模型输出:{ thought: 我将启动一个简单的 Flask 测试服务器并向 /login 发送一个请求来捕获错误。, action: run_code, action_input: { code: import sys\nsys.path.insert(0, .)\nfrom app import app\napp.testing True\nwith app.test_client() as client:\n resp client.post(/login, data{username:test,password:test})\n print(fStatus: {resp.status_code})\n print(fResponse: {resp.data.decode()})\n, language: python } }步骤 3 (行动与观察):在沙箱中执行上述代码。观察结果 (错误): “工具执行失败:AttributeError: NoneType object has no attribute query。错误发生在auth.py第 15 行User.query.filter_by(...)。”步骤 4 (推理):模型思考: “User对象是None这意味着 SQLAlchemy 的User模型可能没有正确导入或定义。让我检查一下models.py文件并确认User类是否存在以及是否导入了db对象。”模型输出:{ thought: 错误表明 User 模型未定义或未与数据库绑定。检查 models.py 和数据库初始化代码。, action: read_file, action_input: {path: models.py} }... 循环继续。最终模型可能会发现是models.py中缺少db SQLAlchemy(app)的初始化或者User类没有继承db.Model。它会通过“写文件”工具进行修复然后再次运行测试进行验证直到错误消失最终输出finalize动作。这个推演清晰地展示了 ReAct 主循环如何将复杂的调试任务分解为一系列可执行的“读文件”、“运行代码”等原子操作并通过执行反馈逐步逼近问题根源。5. 性能优化与工程实践要点阅读源码时你会发现 Claude Code 的工程实现远不止上述基础循环。以下是一些关键的优化点和工程实践它们决定了系统的可用性和效率。5.1 上下文长度管理与优化这是成本与效果平衡的核心。大模型的上下文窗口是宝贵资源如 128K tokens必须高效利用。对话历史摘要不可能把所有历史对话都塞进提示词。Claude Code 很可能实现了“滑动窗口”或“摘要”策略。例如只保留最近 10 轮交互的完整消息对于更早的消息则使用另一个 LLM 调用将其总结成一段简短的背景描述。代码上下文的动态加载code_context不是静态的。一个高效的ContextManager会做以下工作监听焦点从模型的“思考”和错误信息中提取出它正在关注的文件名、函数名、类名、行号。智能检索使用轻量级代码分析如 AST 解析或文本搜索只加载与焦点相关的代码片段。例如当模型提到handle_login函数时不仅加载这个函数还加载它的调用者、它调用的函数、以及同文件中的相关类和导入语句。去除冗余避免在不同轮次中重复加载相同的长代码块。5.2 工具执行的稳定性保障工具调用是故障高发区必须有鲁棒的设计。超时与资源限制run_code工具必须设置严格的执行超时如30秒和资源限制内存、CPU。防止一段死循环代码拖垮整个服务。副作用管理与回滚write_file是危险操作。高级的实现会考虑版本备份在修改文件前自动备份原文件。如果后续步骤导致更严重错误可以提示用户或智能体回滚。沙箱文件系统更彻底的方案是让智能体在一个虚拟文件系统如内存文件系统或 Docker 容器内的独立目录中操作所有修改在任务最终确认成功后才一次性提交到真实文件系统。工具结果的后处理代码执行的输出可能非常冗长如pip install的日志。直接塞给模型会浪费 tokens 且干扰注意力。需要有一个OutputProcessor来截断过长的输出只保留头部和尾部关键行。提取关键信息例如从pytest的输出中提取出失败测试的名称和错误断言从ImportError中提取缺失的模块名。5.3 提示词工程与少样本学习Claude Code 的强大很大程度上源于其精心设计的提示词。在源码中你可能会发现一个“提示词模板库”。角色扮演与约束开头的系统提示词如“你是一个资深软件工程师...”设定了角色的行为边界和专业性。少样本示例提示词中很可能包含了几个精心构造的“示例对话”Few-shot Examples。例如用户帮我写一个计算斐波那契数列的函数。 助手思考我需要创建一个 Python 函数。首先检查当前目录是否有相关文件...思考过程 助手行动{action: write_file, action_input: {path: fib.py, content: def fib(n): ...}} 工具结果文件写入成功。 用户现在测试一下 fib(5)。 助手思考我需要运行刚创建的文件中的函数... 助手行动{action: run_code, action_input: {code: from fib import fib; print(fib(5)), language: python}} 工具结果5这些示例教会了模型 ReAct 的格式和思考方式。动态提示词调整根据任务类型调试、重构、新功能开发或当前状态是否连续出错系统可能会切换不同的提示词模板以更好地引导模型。5.4 循环终止与异常处理策略一个健壮的循环必须知道何时停止。成功终止模型主动输出finalize动作。失败终止步数限制max_steps是硬限制。循环检测如果最近几步的“行动”序列出现重复模式如反复读写同一个文件但状态不变则触发终止。用户中断实现一个信号监听允许用户手动停止。优雅降级当循环异常终止时系统应能保存当前AgentState至少是对话历史和代码修改并生成一份报告说明已完成的步骤和遇到的关键问题方便用户接手。6. 从源码到实践构建你自己的简易 ReAct 智能体理解了 Claude Code 的设计精髓后我们可以尝试用最简化的方式构建一个自己的 ReAct 代码智能体。这里提供一个高度浓缩但可运行的骨架。import json import subprocess from pathlib import Path from typing import Dict, List # 假设使用 OpenAI API你需要安装 openai 库并设置 API_KEY import openai class SimpleCodeAgent: def __init__(self, task: str, work_dir: str ., model: str gpt-4): self.task task self.work_dir Path(work_dir) self.model model self.client openai.OpenAI(api_keyyour-api-key) self.conversation [{role: user, content: task}] self.max_steps 15 def run(self): for step in range(self.max_steps): print(f\n--- 步骤 {step1} ---) # 1. 推理 response self._call_llm() try: action_data self._extract_action(response) except ValueError as e: print(f解析模型输出失败: {e}) self.conversation.append({role: user, content: f你的输出格式不正确。请严格按照JSON格式输出。错误{e}}) continue if action_data.get(action) finalize: print(f任务完成: {action_data.get(summary, )}) break # 2. 行动 result self._execute_tool(action_data[action], action_data.get(action_input, {})) observation result[observation] # 3. 观察与记录 self.conversation.append({role: assistant, content: action_data.get(thought, )}) self.conversation.append({role: user, content: f结果: {observation}}) print(f观察: {observation[:200]}...) def _call_llm(self) - str: prompt self._build_prompt() response self.client.chat.completions.create( modelself.model, messages[{role: system, content: 你是一个Python编程助手。请用JSON格式输出你的思考和行动。}] self.conversation[-10:], # 滑动窗口 temperature0.2, ) return response.choices[0].message.content def _build_prompt(self): # 简化的提示词 return self.conversation def _extract_action(self, response: str) - Dict: # 尝试从响应中提取JSON代码块 import re json_match re.search(rjson\n(.*?)\n, response, re.DOTALL) if json_match: json_str json_match.group(1) else: json_str response.strip() return json.loads(json_str) def _execute_tool(self, action: str, inputs: Dict) - Dict: if action run_python: code inputs.get(code, ) return self._run_python_code(code) elif action read_file: path inputs.get(path, ) return self._read_file(path) elif action write_file: path inputs.get(path, ) content inputs.get(content, ) return self._write_file(path, content) else: return {observation: f错误未知动作 {action}} def _run_python_code(self, code: str) - Dict: try: # 警告简易实现无沙箱仅用于演示生产环境必须使用Docker等隔离。 result subprocess.run( [python3, -c, code], cwdself.work_dir, capture_outputTrue, textTrue, timeout10 ) if result.returncode 0: return {observation: f执行成功:\n{result.stdout}} else: return {observation: f执行失败:\n{result.stderr}} except subprocess.TimeoutExpired: return {observation: 错误代码执行超时10秒。} except Exception as e: return {observation: f错误{e}} def _read_file(self, path: str) - Dict: try: full_path self.work_dir / path content full_path.read_text() return {observation: f文件内容:\n{content}} except Exception as e: return {observation: f读取文件失败: {e}} def _write_file(self, path: str, content: str) - Dict: try: full_path self.work_dir / path full_path.write_text(content) return {observation: f文件 {path} 已写入。} except Exception as e: return {observation: f写入文件失败: {e}} # 使用示例 if __name__ __main__: agent SimpleCodeAgent( task请检查当前目录下是否有 requirements.txt 文件如果有请列出其中的前三个包。, work_dir./sample_project ) agent.run()这个简易实现的重点与警告核心流程完整它包含了 ReAct 的推理、行动、观察循环以及状态对话历史管理。极度简化没有代码上下文管理没有复杂的提示词工程工具很少。安全警告_run_python_code直接使用subprocess存在严重安全风险绝对不可用于生产环境或处理不可信代码。真实系统必须使用 Docker 或专用沙箱。健壮性不足错误处理、循环检测、上下文长度控制都非常基础。这个例子旨在帮助你将前面解析的理论映射到具体的代码结构上。要构建一个可用的系统你需要在此基础上逐一实现我们前面讨论的各个优化模块。7. 常见陷阱与排查指南在实际运行 ReAct 智能体时你会遇到各种各样的问题。以下是一些典型陷阱及其排查思路。7.1 模型不按格式输出现象模型回复是自然语言而不是你要求的 JSON 格式。原因提示词不够清晰少样本示例不足模型温度temperature参数过高。解决强化格式指令在系统提示词和用户提示词中用非常明确、加重的语言描述格式要求。例如“你必须且只能输出一个 JSON 对象包含 thought, action, action_input 字段。”提供更清晰的示例在少样本示例中展示格式正确的完整交互。降低温度将temperature设为 0.1 或 0.2降低随机性使输出更确定。使用模型的结构化输出功能如果使用的 API 支持如 OpenAI 的response_format或 Anthropic 的tool_use优先使用这些功能它们比文本指令更可靠。7.2 智能体陷入死循环现象智能体反复执行相似操作如反复读取同一个文件无法推进任务。原因模型未能从观察结果中提取有效信息来更新计划任务本身模糊或不可实现。解决增强观察结果的信息量确保工具返回的observation不仅包含“成功/失败”还包含对下一步推理有用的关键数据。例如运行测试失败时不要只返回“测试失败”而要提取具体的失败断言和行号。在提示词中加入“防呆”引导例如“如果你连续三次尝试相同或类似的操作都未能取得进展请重新评估你的计划或者尝试一个完全不同的方法。”实现循环检测器在AgentState中记录最近 N 步的行动和观察摘要。如果检测到重复模式则在提示词中明确告知模型“注意你似乎陷入了循环。你最近三步都在做 X。请尝试一个不同的策略。”设置硬性步数限制这是最后的安全网。7.3 工具执行结果太长干扰模型现象运行pip install或npm start产生了数百行日志模型被淹没在无关信息中。原因未对工具输出进行后处理。解决为每个工具设计输出过滤器run_code捕获最后 20 行输出如果遇到错误非零退出码则优先捕获错误信息。search_files只返回匹配的行和前后几行上下文而不是整个文件。总结冗长输出对于特别长的输出可以调用另一个轻量级模型如 GPT-3.5-turbo进行摘要再将摘要放入观察中。例如“安装过程成功生成了大量日志。摘要成功安装了 package-a (v1.2), package-b (v3.4)无错误。”提供“查看更多”选项在观察结果末尾注明“输出已截断完整日志共XXX行”。如果模型在后续思考中明确要求查看完整日志再通过一个新工具调用提供。7.4 代码修改引入新错误现象智能体为了修复一个错误而修改了代码但修改后产生了新的、更隐蔽的错误。原因模型缺乏对代码库的全局理解修改是局部的、基于试探的。解决强制运行测试在智能体完成一组修改后在提示词中要求它必须运行相关的单元测试或集成测试并将测试结果作为下一轮观察的核心依据。实施“小步快跑”策略鼓励模型进行更小、更增量的修改。每修改一个文件就运行一次相关的检查如语法检查python -m py_compile或导入检查。引入代码分析工具将lint如flake8、type check如mypy作为工具集成进来。在模型提交修改前自动运行这些检查并将警告和错误作为观察的一部分反馈给模型。调试一个 ReAct 智能体本身就是一个递归过程观察它的失败模式然后修改你的提示词、工具设计或状态管理逻辑再观察效果。这个过程本身就充满了“智能体”的味道。