Agent 原理与实战：从零构建简化版 Claude Code 的 Plan and Execute 架构

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 URL：https://taotoken.net/api
• API Key：sk-开头的那串

如果你用 Claude Code 或 Cline 这类工具，Base URL 填https://taotoken.net/api，Key 填刚创建的，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( name="read_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", encoding="utf-8") as f: return f.read() @register_tool( name="write_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_ok=True) with open(full_path, "w", encoding="utf-8") as f: f.write(content) return f"写入成功：{path}" @register_tool( name="list_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( name="run_command", description="在项目目录下执行终端命令", parameters={"command": "要执行的命令"} ) def run_command(command): result = subprocess.run( command, shell=True, cwd=PROJECT_DIR, capture_output=True, text=True, timeout=30 ) 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_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api" ) def call_model(messages, model="claude-sonnet-4-20250514", temperature=0): response = client.chat.completions.create( model=model, messages=messages, temperature=temperature ) 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_description=tools_desc, os_info=env_info["os"], project_dir=env_info["project_dir"], file_list=env_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:end+1]) 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( step=step, tools_description=tools_desc, history=history )}, {"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_turns=5是安全阀。实际跑的时候，一个步骤通常 1-2 轮工具调用就完成了，5 轮足够覆盖大多数情况，同时防止模型陷入死循环。

3.3 动态再规划：Plan and Execute 的灵魂

每执行完一步，就把用户问题、当前计划、历史记录一起发给 Replan 模型，让它决定是继续执行还是给出最终答案。

def replan(question, current_plan, history, tools_desc): prompt = REPLAN_PROMPT_TEMPLATE.format( question=question, current_plan=json.dumps(current_plan, ensure_ascii=False, indent=2), history=json.dumps(history, ensure_ascii=False, indent=2), tools_description=tools_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_ascii=False, indent=2)}") 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_ascii=False, indent=2)}") 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_steps=10是外层安全阀，防止计划一直不收敛。

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_KEY="sk-你的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.js

4.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 Unauthorized：Key 或 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, model="claude-sonnet-4-20250514", temperature=0): response = client.chat.completions.create( model=model, messages=messages, temperature=temperature ) 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/api，API Key 用你在 TaoToken 创建的 Key，Model ID 根据你选的模型填。这样你既可以用现成工具，也可以自己写调度逻辑。

需要长期跑编码任务或 Agent 工作流的话，可以看看 Coding Plan，它针对高频调用场景做了优化。想先验证模型效果，可以直接在模型对话里试几个复杂任务，看看 Plan 和 Replan 的输出质量。接入文档里有完整的配置示例和工具注册模板，照着改就能用。

整套代码我放在workspace目录下跑通了，你可以直接复制上面的片段拼起来。遇到报错先看第 5 节的排查清单，大部分问题都能解决。