1. 为什么大模型需要 Plan and Execute 架构你可能已经用过大模型写代码但有没有发现一个问题它能给你一段完整的贪吃蛇实现却没法自己把文件建好、把代码写进去、再跑起来验证。这就是当前大模型最核心的局限——能思考但缺少行动力。Agent 要解决的就是这件事。把大模型和一组工具读写文件、执行命令、搜索网页组装起来让它能感知环境、改变环境这就是 Agent 的基本定义。而 Plan and Execute 是其中一种特别适合复杂多步任务的调度模式。和 React 那种“思考-行动-观察”逐步推进不同Plan and Execute 的核心思路是先让模型生成一份完整计划再逐步执行每执行完一步就重新评估并动态调整后续计划。这个“再规划”环节是它最值钱的地方。我试过用纯 React 模式做一个“查资料→整理→写报告”的任务模型经常在第三步就忘了第一步查到了什么。Plan and Execute 通过显式的计划列表和历史执行记录让模型每一步都有全局视角不容易跑偏。这篇文章面向想理解 Agent 调度原理的开发者会从零构建一个简化版 Claude Code 的 Plan and Execute 架构。你会看到完整的工具注册代码、执行循环配置以及一次从任务规划到工具调用的完整验证流程。核心检索词Agent Plan and Execute 架构、Claude Code 简化实现、React 式循环与动态再规划。适合谁看写过 Python、调过 OpenAI 或 DeepSeek API、想搞清楚 Agent 内部到底怎么调度的人。不需要 LangChain 经验但需要能看懂基本的函数调用和 JSON 解析。整个架构分四个角色Plan 模型负责生成初始计划执行 Agent 负责跑每一步Replan 模型根据执行结果调整计划主程序串联整个流程。下面从环境准备开始一步步把它搭出来。2. TaoToken 前置准备与工具注册配置在写 Agent 主循环之前先把模型调用通道和工具注册这两件事搞定。我用 TaoToken 作为模型接入层它兼容 OpenAI 的接口格式改一下 Base URL 就能用省去自己维护多模型适配的麻烦。2.1 获取 API Key 与 Base URL打开 https://taotoken.net/api-keys 创建一个 Key然后记下两个地址Base URLhttps://taotoken.net/apiAPI Keysk-开头的那串如果你用 Claude Code 或 Cline 这类工具Base URL 填https://taotoken.net/apiKey 填刚创建的Model ID 根据你选的模型填比如claude-sonnet-4-20250514或gpt-4o。这三件套Base URL Key Model ID缺一不可后面配置文件里会反复用到。2.2 工具注册把函数变成 Agent 的“手脚”Agent 的工具本质上就是普通 Python 函数关键是要让模型知道有哪些工具可用、每个工具需要什么参数。我用一个字典来注册工具结构清晰也好扩展。import os import subprocess import json # 工具注册表名称 - {描述, 参数, 函数} TOOLS {} def register_tool(name, description, parameters): def decorator(func): TOOLS[name] { name: name, description: description, parameters: parameters, function: func } return func return decorator register_tool( nameread_file, description读取指定路径的文件内容, parameters{path: 文件路径相对于项目目录} ) def read_file(path): full_path os.path.join(PROJECT_DIR, path) if not os.path.exists(full_path): return f错误文件 {path} 不存在 with open(full_path, r, encodingutf-8) as f: return f.read() register_tool( namewrite_to_file, description将内容写入指定路径的文件文件不存在则创建, parameters{path: 文件路径, content: 要写入的完整内容} ) def write_to_file(path, content): full_path os.path.join(PROJECT_DIR, path) os.makedirs(os.path.dirname(full_path), exist_okTrue) with open(full_path, w, encodingutf-8) as f: f.write(content) return f写入成功{path} register_tool( namelist_files, description列出项目目录下的所有文件, parameters{} ) def list_files(): result [] for root, dirs, files in os.walk(PROJECT_DIR): for f in files: rel os.path.relpath(os.path.join(root, f), PROJECT_DIR) result.append(rel) return \n.join(result) if result else 目录为空 register_tool( namerun_command, description在项目目录下执行终端命令, parameters{command: 要执行的命令} ) def run_command(command): result subprocess.run( command, shellTrue, cwdPROJECT_DIR, capture_outputTrue, textTrue, timeout30 ) output result.stdout result.stderr return output[:2000] if output else 命令执行完成无输出这段代码的关键点每个工具都有明确的description和parameters后面渲染系统提示词时会把这些信息拼进去模型才知道什么时候该调哪个工具、传什么参数。2.3 系统提示词模板Plan and Execute 的“剧本”Plan and Execute 模式需要两套提示词一套给 Plan 模型生成计划一套给 Replan 模型调整计划。我用占位符的方式渲染运行时填入工具列表和环境信息。PLAN_PROMPT_TEMPLATE 你是一个任务规划器。根据用户的问题生成一个分步执行计划。 可用工具 {tools_description} 当前环境 - 操作系统{os_info} - 项目目录{project_dir} - 目录下文件{file_list} 要求 1. 将任务拆解为 3-7 个具体步骤 2. 每个步骤必须能用上述工具完成 3. 输出 JSON 格式{{steps: [步骤1, 步骤2, ...]}} 4. 只输出 JSON不要有其他内容 REPLAN_PROMPT_TEMPLATE 你是一个动态规划器。根据用户问题、当前计划和已执行的历史记录决定下一步。 用户问题{question} 当前计划 {current_plan} 历史执行记录 {history} 可用工具 {tools_description} 请判断 - 如果还有未完成的步骤输出新的计划 JSON{{steps: [...]}} - 如果所有步骤已完成、可以回答用户问题输出{{final_answer: 你的回答}} 只输出 JSON。 注意{{和}}的转义因为后面要用.format()填充占位符。工具描述部分我会在运行时动态生成把TOOLS字典里的 name、description、parameters 拼成一段文本。2.4 模型调用封装统一封装一个call_model函数Plan、Replan、执行 Agent 都走这个入口。from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://taotoken.net/api ) def call_model(messages, modelclaude-sonnet-4-20250514, temperature0): response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature ) return response.choices[0].message.content把TAOTOKEN_API_KEY写到环境变量里不要硬编码在代码中。到这里前置准备就完成了工具注册表有了提示词模板有了模型调用通道也通了。接下来写核心的执行循环。3. 可复制配置Plan and Execute 执行循环完整实现这一节是整篇文章的核心。我会把 Plan and Execute 的主循环拆成三个部分计划生成、步骤执行、动态再规划。每一部分都有可复制的代码你直接拼到一起就能跑。3.1 计划生成让模型先想清楚再动手计划生成的关键是让模型输出结构化的 JSON而不是自由文本。我在提示词里明确要求只输出 JSON解析时加一层容错。import json import re def generate_plan(question, tools_desc, env_info): prompt PLAN_PROMPT_TEMPLATE.format( tools_descriptiontools_desc, os_infoenv_info[os], project_direnv_info[project_dir], file_listenv_info[file_list] ) messages [ {role: system, content: prompt}, {role: user, content: question} ] raw call_model(messages) return parse_json_safe(raw) def parse_json_safe(text): # 先尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取 json ... 代码块 match re.search(r(?:json)?\s*([\s\S]*?), text) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass # 尝试提取第一个 { 到最后一个 } start text.find({) end text.rfind(}) if start ! -1 and end ! -1: try: return json.loads(text[start:end1]) except json.JSONDecodeError: pass raise ValueError(f无法解析模型返回的 JSON{text[:200]})parse_json_safe这个函数在实际跑的时候非常有用。模型有时候会在 JSON 外面包一层解释文字或者用 markdown 代码块包起来直接json.loads会报错。三层容错基本能覆盖 95% 的情况。3.2 步骤执行用 React 式循环跑单步每个步骤的执行我用一个简化版 React 循环模型决定调哪个工具主程序执行工具把结果反馈给模型直到模型认为这一步完成。EXECUTE_PROMPT 你是一个执行器。根据当前步骤决定调用哪个工具。 当前步骤{step} 可用工具 {tools_description} 历史执行记录 {history} 输出 JSON 格式 - 调用工具{{tool: 工具名, args: {{参数名: 参数值}}}} - 步骤完成{{done: true, result: 这一步的结果}} 只输出 JSON。 def execute_step(step, tools_desc, history): messages [ {role: system, content: EXECUTE_PROMPT.format( stepstep, tools_descriptiontools_desc, historyhistory )}, {role: user, content: f请执行步骤{step}} ] max_turns 5 # 防止死循环 for _ in range(max_turns): raw call_model(messages) action parse_json_safe(raw) if action.get(done): return action.get(result, 步骤完成) tool_name action.get(tool) tool_args action.get(args, {}) if tool_name not in TOOLS: observation f错误工具 {tool_name} 不存在 else: try: observation TOOLS[tool_name][function](**tool_args) except Exception as e: observation f工具执行出错{str(e)} # 把工具结果反馈给模型 messages.append({role: assistant, content: raw}) messages.append({role: user, content: f工具执行结果{observation}}) return 达到最大轮次步骤未完成这里max_turns5是安全阀。实际跑的时候一个步骤通常 1-2 轮工具调用就完成了5 轮足够覆盖大多数情况同时防止模型陷入死循环。3.3 动态再规划Plan and Execute 的灵魂每执行完一步就把用户问题、当前计划、历史记录一起发给 Replan 模型让它决定是继续执行还是给出最终答案。def replan(question, current_plan, history, tools_desc): prompt REPLAN_PROMPT_TEMPLATE.format( questionquestion, current_planjson.dumps(current_plan, ensure_asciiFalse, indent2), historyjson.dumps(history, ensure_asciiFalse, indent2), tools_descriptiontools_desc ) messages [ {role: system, content: prompt}, {role: user, content: 请决定下一步。} ] raw call_model(messages) return parse_json_safe(raw)3.4 主循环把三个部分串起来def run_agent(question, project_dir): global PROJECT_DIR PROJECT_DIR project_dir tools_desc render_tools_description() env_info { os: os.name, project_dir: project_dir, file_list: list_files() } # 第一步生成初始计划 plan generate_plan(question, tools_desc, env_info) print(f[Plan] 初始计划{json.dumps(plan, ensure_asciiFalse, indent2)}) history [] max_steps 10 for i in range(max_steps): steps plan.get(steps, []) if not steps: print([Agent] 计划为空结束) break # 执行第一步 current_step steps[0] print(f\n[Execute] 执行步骤{current_step}) result execute_step(current_step, tools_desc, history) print(f[Observation] {result}) history.append({step: current_step, result: result}) # 动态再规划 new_plan replan(question, plan, history, tools_desc) if final_answer in new_plan: print(f\n[Final Answer] {new_plan[final_answer]}) return new_plan[final_answer] plan new_plan print(f[Replan] 新计划{json.dumps(plan, ensure_asciiFalse, indent2)}) return 达到最大步数限制任务未完成 def render_tools_description(): lines [] for name, info in TOOLS.items(): params , .join(f{k}: {v} for k, v in info[parameters].items()) lines.append(f- {name}({params}): {info[description]}) return \n.join(lines)整个主循环的逻辑很清晰生成计划 → 执行第一步 → 再规划 → 如果还有步骤就继续如果拿到最终答案就返回。max_steps10是外层安全阀防止计划一直不收敛。3.5 配置文件把参数抽出来把模型名、Base URL、最大轮次这些抽到一个config.json里方便切换。{ base_url: https://taotoken.net/api, model: claude-sonnet-4-20250514, max_steps: 10, max_turns_per_step: 5, temperature: 0, project_dir: ./workspace }如果你用 Cline 或 Claude Code 这类工具接入配置格式类似{ baseUrl: https://taotoken.net/api, apiKey: sk-你的Key, model: claude-sonnet-4-20250514 }Base URL、Key、Model ID 三件套填对工具就能正常调用模型。到这里完整的 Plan and Execute 执行循环就搭好了。下一节跑一个真实任务验证。4. 验证请求从任务规划到工具调用的完整流程代码写完了得跑一个真实任务看看效果。我用“写一个贪吃蛇游戏HTML/CSS/JS 分文件存放”这个任务来验证因为它需要多步文件操作正好能体现 Plan and Execute 的调度能力。4.1 启动 Agent 并输入任务export TAOTOKEN_API_KEYsk-你的Key mkdir -p workspace python agent.py程序启动后会提示输入任务。我输入写一个贪吃蛇游戏使用 HTML、CSS、JS 实现代码分别放在 index.html、style.css、game.js 三个文件中4.2 观察 Plan 阶段的输出模型返回的初始计划{ steps: [ 查看项目目录当前文件列表, 创建 index.html 文件包含游戏画布和引入 CSS/JS, 创建 style.css 文件定义游戏界面样式, 创建 game.js 文件实现贪吃蛇逻辑, 验证三个文件是否都已创建 ] }这个计划是合理的先看环境再逐个创建文件最后验证。注意第一步“查看文件列表”是模型自己加的因为系统提示词里告诉了它项目目录信息它知道应该先确认环境。4.3 执行第一步查看目录[Execute] 执行步骤查看项目目录当前文件列表 [Observation] 目录为空执行 Agent 调用了list_files工具返回“目录为空”。这个结果会进入历史记录Replan 时会用到。4.4 Replan 后的计划变化[Replan] 新计划 { steps: [ 创建 index.html 文件包含游戏画布和引入 CSS/JS, 创建 style.css 文件定义游戏界面样式, 创建 game.js 文件实现贪吃蛇逻辑, 验证三个文件是否都已创建 ] }可以看到“查看文件列表”这一步被移除了因为已经执行完毕。这就是 Plan and Execute 的动态再规划每执行一步计划就更新一次已完成的步骤不再出现。4.5 执行文件创建步骤接下来几轮循环执行 Agent 分别调用write_to_file创建三个文件。每次写入成功后Replan 都会更新计划移除已完成的步骤。[Execute] 执行步骤创建 index.html 文件 [Observation] 写入成功index.html [Execute] 执行步骤创建 style.css 文件 [Observation] 写入成功style.css [Execute] 执行步骤创建 game.js 文件 [Observation] 写入成功game.js4.6 最终答案与验证当所有文件创建完毕后Replan 模型判断任务完成返回最终答案[Final Answer] 贪吃蛇游戏已创建完成。三个文件均已写入项目目录 - index.html游戏主页面包含 canvas 画布 - style.css游戏样式深色背景和网格布局 - game.js游戏逻辑包含蛇的移动、食物生成、碰撞检测和分数计算 你可以直接在浏览器中打开 index.html 运行游戏。验证一下文件确实存在ls workspace/ # index.html style.css game.js打开index.html贪吃蛇能正常移动、吃食物、计分。整个流程从任务规划到工具调用再到最终验证完整跑通了。4.7 关键观察点跑完这个任务有几个地方值得注意第一Plan 模型生成的计划不是死的。初始计划里有“查看文件列表”执行完后 Replan 自动把它移除了。如果中途某个文件写入失败Replan 也会把失败的步骤重新加回计划。第二执行 Agent 每步只做一件事。创建 index.html 时不会顺便去写 CSS职责边界清晰。这样即使某一步出错也不会影响其他步骤。第三历史记录是 Replan 的关键输入。每次 Replan 都能看到之前所有步骤的执行结果所以它知道哪些做完了、哪些还没做、哪些需要重试。这套架构跑简单任务可能比纯 React 模式多几次模型调用但在多步骤、需要动态调整的场景下它的稳定性明显更好。下一节整理几个实际跑的时候容易踩的坑。5. 本篇常见错排查401、JSON 解析失败与死循环代码跑起来之后最容易卡在几个地方。我把实际踩过的坑整理出来对照报错直接定位。5.1 401 UnauthorizedKey 或 Base URL 不对最常见的报错长这样openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}排查顺序第一确认环境变量TAOTOKEN_API_KEY真的被读到了。在 Python 里加一行print(os.environ.get(TAOTOKEN_API_KEY)[:10])看看是不是sk-开头。如果打印出来是None说明环境变量没设置成功检查export命令是否在当前终端会话执行。第二确认 Base URL 是https://taotoken.net/api不要多加/v1或结尾斜杠。有些 OpenAI 兼容接口的路径规则不一样多一个字符就 404 或 401。第三确认 Key 没有多余空格。从网页复制的时候经常带上换行或空格用.strip()处理一下。如果你用的是 Claude Code 或 Cline报错可能是local proxy failed或OAuth error。这类工具通常有自己的配置文件检查~/.claude/settings.json或 Cline 的 MCP 配置里 Base URL 和 Key 是否填对。三件套Base URL Key Model ID缺一不可Model ID 写错也会报 401 或 404。5.2 JSON 解析失败模型返回了非 JSON 内容报错信息ValueError: 无法解析模型返回的 JSON好的我来帮你规划...模型有时候会“好心”加一句解释再输出 JSON。parse_json_safe的三层容错能处理大部分情况但如果模型返回的是纯文本解释、完全没有 JSON就会抛错。解决办法有两个一是把提示词里的“只输出 JSON”再强调一遍加上“不要有任何解释文字”二是把temperature设为 0减少模型自由发挥的空间。如果还是不行可以在parse_json_safe失败时打印原始返回看看模型到底输出了什么针对性调整提示词。5.3 reading choices 报错响应结构不对AttributeError: NoneType object has no attribute choices或者KeyError: choices这通常是因为 API 返回了错误响应但代码直接去取response.choices[0]。在call_model里加一层判断def call_model(messages, modelclaude-sonnet-4-20250514, temperature0): response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature ) if not response.choices: raise RuntimeError(f模型返回为空{response}) return response.choices[0].message.content如果频繁出现这个报错检查模型名是否正确。有些模型 ID 在 TaoToken 上需要用特定格式比如带日期后缀的完整名称。5.4 死循环计划一直不收敛Agent 跑了十几轮还在执行max_steps用完了也没出最终答案。常见原因一是 Replan 模型没有正确判断“任务完成”。检查REPLAN_PROMPT_TEMPLATE里是否明确告诉它“所有步骤完成后输出 final_answer”。如果提示词太模糊模型可能一直生成新计划。二是执行步骤一直失败Replan 反复把同一步加回计划。比如write_to_file因为权限问题一直报错历史记录里全是失败信息但 Replan 不知道该怎么处理。这种情况下需要在提示词里加一条“如果某步骤连续失败两次跳过该步骤并记录错误”。三是max_turns_per_step设太大单个步骤内部就循环了很多轮。建议设 3-5超过就强制返回。5.5 工具参数解析错误TypeError: write_to_file() missing 1 required positional argument: content模型返回的args里缺少必要参数。在execute_step里加参数校验required TOOLS[tool_name][parameters].keys() missing [k for k in required if k not in tool_args] if missing: observation f错误缺少参数 {missing} else: observation TOOLS[tool_name][function](**tool_args)这样模型能看到具体缺什么参数下一轮会补上。5.6 模型不调用工具直接给答案执行 Agent 拿到步骤后没有输出{tool: ...}而是直接返回了一段文字。这通常是因为提示词里没有强调“必须用 JSON 格式输出工具调用”。把EXECUTE_PROMPT里的输出格式部分加粗强调或者在系统提示词开头就写“你只能输出 JSON不能输出其他内容”。如果模型仍然不听话可以在execute_step里加一个 fallback如果解析出来的 JSON 既没有tool也没有done就当作模型想直接完成这一步把返回内容作为结果。这几个坑覆盖了实际跑的时候 90% 的报错。遇到其他问题先打印原始返回内容对照提示词和工具注册表排查基本都能定位。6. 从简化版到生产级 Agent 的接入路径跑通上面的代码后你手里已经有一个能用的 Plan and Execute Agent 了。它虽然简化但核心调度逻辑和 Claude Code 这类工具是相通的计划生成、步骤执行、动态再规划、工具调用反馈这四个环节一个不少。如果你想把它用到实际工作里有几个方向可以继续深入。第一把工具集扩展一下。目前只有读写文件、列目录、执行命令四个工具。实际编码场景还需要搜索代码、运行测试、查看 git 状态等。每加一个工具就在TOOLS字典里注册一个函数系统提示词会自动带上新工具的描述模型就能调用。第二把执行 Agent 换成更成熟的 React 实现。目前execute_step是一个简化版循环你可以把它替换成完整的 React Agent支持多轮工具调用和更复杂的观察处理。Plan and Execute 只要求执行 Agent 能完成指定步骤不关心它内部怎么跑所以替换起来很灵活。第三接入真实的编码环境。把PROJECT_DIR指向你的实际项目目录Agent 就能在你的代码库里操作。不过要注意安全边界run_command工具建议加一个用户确认环节避免模型执行危险命令。如果你想把 Agent 接到 Claude Code 或 Cline 这类工具里配置三件套就行Base URL 填https://taotoken.net/apiAPI Key 用你在 TaoToken 创建的 KeyModel ID 根据你选的模型填。这样你既可以用现成工具也可以自己写调度逻辑。需要长期跑编码任务或 Agent 工作流的话可以看看 Coding Plan它针对高频调用场景做了优化。想先验证模型效果可以直接在模型对话里试几个复杂任务看看 Plan 和 Replan 的输出质量。接入文档里有完整的配置示例和工具注册模板照着改就能用。整套代码我放在workspace目录下跑通了你可以直接复制上面的片段拼起来。遇到报错先看第 5 节的排查清单大部分问题都能解决。