Planner 实战：用 LangGraph Subgraph 让 Agent 学会分解任务——从零实现 Agent Harness 的 TaoToken 配置

1. 从一次“写崩了”的任务说起Planner 到底解决什么问题如果你已经用 LangGraph 搭过一个最小可跑的 Agent Harness大概会经历这样一个阶段单轮问答、调个计算器、查个时间一切都很顺。可一旦用户丢过来一句“帮我写一份 LangGraph 学习计划包含基础概念、State、Node/Edge、工具调用、持久化、Human-in-the-Loop、Streaming、Memory、Subgraph 共 9 个主题每个主题写 200 字介绍最后再给个学习顺序表”Agent 就开始表演了——写到第三个主题 token 爆了或者干脆把前面说过的内容忘干净最后给你一段前后矛盾的文字。这不是模型不行而是单次 LLM 调用承载了太多职责它要理解需求、要规划结构、要逐段生成、还要记住自己写到哪了。上下文窗口再大也架不住这种“一次性全干完”的模式。Planner 模块要解决的就是这件事。它的核心思路非常朴素先让 LLM 只做规划把大任务拆成一组可独立执行的子任务再让 Agent 逐个执行子任务每执行完一个就回头检查进度全部完成后统一整合输出。这就是 Plan → Execute → Check 的循环也是 ReAct 模式的一个自然升级——从“边想边做”变成“先想清楚再动手做一步看一步”。我试过把同一个 9 主题写作任务分别丢给无 Planner 的 Harness 和带 Planner 的 Harness前者平均要 3 次重试才能凑出一份能看的稿子后者一次跑通而且每个子任务的输出质量明显更稳定。原因很简单每次 LLM 调用只背 200 字的 KPI它不用分心去记“我前面写了啥”。这一篇的目标很明确用 LangGraph 的 Subgraph 机制从零给 Agent Harness 加上 Planner 模块跑通“任务分解 → 子图执行 → 结果整合”的完整链路。适合已经写过基础 LangGraph 图、想进一步理解 Agent 调度骨架的开发者。全程用 TaoToken 统一 Key 和 API 通道省去多模型切换时反复改 base_url 的麻烦。2. TaoToken 前置统一 Key 与 API 通道让 Planner 和 Executor 用同一个入口在动手写 Planner 之前先把模型接入这一层理顺。Planner 和 Executor 虽然职责不同但底层都是 LLM 调用。如果 Planner 用一个模型、Executor 用另一个模型而每个模型又各自配一套 Key 和 base_url代码里会到处散落配置调试时非常痛苦。TaoToken 在这里扮演的角色是统一的 API 通道你只需要在官网注册后拿到一个 Key就可以通过同一个 base_url 调用不同模型。对于 Planner 这种“需要确定性、temperature0”的场景和 Executor 这种“需要工具绑定”的场景可以分别指定模型名但共用同一套接入配置。具体操作路径打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 完成注册进入控制台。在控制台左侧找到「API Keys」创建一个新 Key复制保存。这个 Key 后面会写进环境变量不要硬编码在代码里。如果你需要查看当前支持的模型列表和调用方式可以打开接入文档页里面有 base_url 和模型 ID 的对照说明。想先验证 Key 是否可用可以直接进模型对话页面发一条测试消息确认通道正常再写代码。这里有一个容易踩的坑Planner 和 Executor 的模型 ID 要写对。比如 Planner 用gpt-4o-mini做任务分解Executor 也用gpt-4o-mini做工具调用两者可以相同但如果你想让 Planner 用更便宜的模型、Executor 用更强的模型就要在配置类里分别指定planner_model和model两个字段。TaoToken 的好处是这两个模型名走同一个 base_url不需要为每个模型单独配一套环境变量。环境变量建议这样设置export TAOTOKEN_API_KEY你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 代码里通过init_chat_model读取。LangChain 的init_chat_model支持base_url参数把它指向 TaoToken 的 API 地址即可。这样 Planner 和 Executor 都从同一个入口走后续换模型只改模型名不动接入层。如果你打算长期跑编码类 Agent 任务可以顺带看一下 Coding Plan 页面它更适合需要持续调用、批量执行的场景而只是做本篇这种端到端验证用 API Keys 就够了。3. 可复制配置Subgraph 节点定义与状态传递的完整代码这一节是全文的核心。我会把 Planner 模块拆成三块来讲状态定义、Planner 类、Task Subgraph最后组装成 MiniHarness v2。所有代码都可以直接复制运行路径和原文一致。3.1 状态定义StandardState 与 TaskState主图的状态需要额外承载 Planner 相关的字段。子图的状态则只关心单个子任务的执行。from typing import Annotated, Literal, Optional from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langchain.chat_models import init_chat_model from langchain_core.messages import SystemMessage, ToolMessage from langchain.tools import tool from dataclasses import dataclass import json class StandardState(TypedDict): messages: Annotated[list, add_messages] iteration_count: int plan: str task_list: list[str] current_task_index: int task_results: list[str] compile_result: str class TaskState(TypedDict): subtask: str messages: Annotated[list, add_messages] tool_calls: int注意task_results用的是普通list不是add_messages。因为它存的是“已完成子任务的结果文本”不需要被 LLM 当作聊天历史消费只在 plan 和 compile 节点之间传递。3.2 Planner 类任务分解与完成检查Planner 内部调 LLM 做两件事plan()把大任务拆成子任务列表check_completion()判断是否全部完成。class Planner: def __init__(self, model_name: str gpt-4o-mini, max_subtasks: int 8): self.model init_chat_model( model_name, temperature0, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], ) self.max_subtasks max_subtasks def plan(self, task: str, context: str ) - list[str]: context_hint f上下文信息{context}\n if context else response self.model.invoke( f{context_hint} f你是一个任务规划专家。将以下任务分解为具体的子任务。\n f要求\n f- 每个子任务应该是单个可执行的步骤\n f- 子任务数量控制在 2 到 {self.max_subtasks} 之间\n f- 输出格式JSON 数组每个元素是一个字符串\n f- 只输出 JSON不要其他文字\n f\n任务{task} ) content response.content.strip() if json in content: content content.split(json)[1].split()[0].strip() elif in content: content content.split()[1].split()[0].strip() try: tasks json.loads(content) if isinstance(tasks, list) and all(isinstance(t, str) for t in tasks): return tasks[:self.max_subtasks] except json.JSONDecodeError: pass lines [l.strip(- ).strip() for l in content.split(\n) if l.strip()] return lines[:self.max_subtasks] def check_completion(self, task, task_list, task_results): if len(task_results) len(task_list): combined \n\n.join( f## {task_list[i]}\n{task_results[i]} for i in range(len(task_list)) if i len(task_results) ) return True, combined return False, def compile_answer(self, original_task, task_list, task_results): combined \n\n.join( f## 任务 {i1}: {task_list[i]}\n{task_results[i]} for i in range(len(task_list)) if i len(task_results) ) response self.model.invoke( f用户原始问题{original_task}\n\n f以下是各子任务的执行结果\n{combined}\n\n f请基于以上结果给用户一个完整的、连贯的最终回答。 ) return response.content这里有个设计细节值得说Planner 单独初始化一个模型实例而不是复用 Executor 的 model。因为 Executor 的模型绑定了 toolssystem prompt 也不一样Planner 只需要“收到任务 → 输出 JSON 列表”给它绑工具反而会干扰。另外 Planner 的 temperature 固定为 0任务分解需要确定性不需要创造力。3.3 Task Subgraph把 LLM→Tool→LLM 循环封装成子图这是本篇标题里 Subgraph 的落点。每个子任务内部都需要“LLM 判断是否调工具 → 调工具 → 再交给 LLM”这个循环。如果直接写进主图的 execute_task 节点里主图会变得非常臃肿。用 Subgraph 封装后主图只需要调用一次task_subgraph.invoke()。def build_task_subgraph(model, tools_by_name, all_tools, system_prompt, max_steps5): model_with_tools model.bind_tools(all_tools) def task_llm_node(state: TaskState) - dict: response model_with_tools.invoke([ SystemMessage(content( f{system_prompt}\n\n f当前子任务{state[subtask]}\n f专注于完成这个子任务完成后直接给出结果。 )), *state[messages], ]) return {messages: [response], tool_calls: state.get(tool_calls, 0) 1} def task_tool_node(state: TaskState) - dict: last state[messages][-1] results [] for tc in last.tool_calls: fn tools_by_name.get(tc[name]) if fn: try: results.append(ToolMessage( contentstr(fn.invoke(tc[args])), tool_call_idtc[id], )) except Exception as e: results.append(ToolMessage( contentf工具错误: {e}, tool_call_idtc[id], )) else: results.append(ToolMessage( contentf未知工具: {tc[name]}, tool_call_idtc[id], )) return {messages: results} def task_route(state: TaskState) - Literal[task_tools, END]: last state[messages][-1] if last.tool_calls and state.get(tool_calls, 0) max_steps: return task_tools return END builder StateGraph(TaskState) builder.add_node(task_llm, task_llm_node) builder.add_node(task_tools, task_tool_node) builder.add_edge(START, task_llm) builder.add_conditional_edges(task_llm, task_route, { task_tools: task_tools, END: END, }) builder.add_edge(task_tools, task_llm) return builder.compile()Subgraph 自带max_steps安全阀默认 5。这是和主图max_iterations独立的第二层保护——防止某个子任务陷入工具死循环。3.4 主图组装plan → execute_task → plan → compile主图的规划模式结构如下def _build_planning_graph(self): builder StateGraph(StandardState) builder.add_node(plan, self._plan_node) builder.add_node(execute_task, self._execute_task_node) builder.add_node(compile, self._compile_node) builder.add_edge(START, plan) builder.add_conditional_edges( plan, self._plan_route, {execute_task: execute_task, compile: compile}, ) builder.add_edge(execute_task, plan) builder.add_edge(compile, END) return builder.compile(checkpointerself.checkpointer)_plan_node首次运行时调用Planner.plan()分解任务后续运行时调用check_completion()检查进度。_plan_route根据task_results长度决定是继续执行还是进入 compile。_execute_task_node调用 Subgraph 执行当前子任务把结果追加到task_results。4. 验证请求跑通一次端到端任务分解代码写完后用一段复杂任务验证整条链路。这里用“写一份 Python 学习计划包含 6 个阶段”作为输入。if __name__ __main__: config HarnessConfig( enable_planningTrue, max_subtasks4, ) agent MiniHarness(configconfig) result agent.run( 帮我做一份 Python 学习计划包含基础语法、函数、面向对象、 文件操作、第三方库和实战项目 6 个阶段每个阶段写一段介绍。 ) final result.get(compile_result) or ( result[messages][-1].content if result.get(messages) else 无结果 ) print(f\n最终回答:\n{final[:500]}...)预期输出会分几段打印计划分解为 4 个子任务: 1. 介绍 Python 基础语法和函数 2. 介绍面向对象和文件操作 3. 介绍第三方库的使用 4. 介绍实战项目并给出学习顺序 ▶ 执行子任务 1/4: 介绍 Python 基础语法和函数... 子任务完成结果长度: 312 字符 ▶ 执行子任务 2/4: 介绍面向对象和文件操作... 子任务完成结果长度: 298 字符 ... 整合 4 个子任务结果... 整合完成如果你看到计划分解为 N 个子任务这行说明 Planner 的plan()正常返回了 JSON 数组如果看到▶ 执行子任务和子任务完成交替出现说明 Subgraph 被正确调用且结果被追加到task_results最后整合出现说明 compile 节点拿到了全部结果。验证时重点看两个地方一是子任务数量是否在 2 到 max_subtasks 之间如果返回 1 个或超过上限说明 Planner 的 prompt 需要调整二是每个子任务的结果长度是否合理如果某个子任务结果只有几十字符可能是 Subgraph 提前 END 了检查task_route里的max_steps判断。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节按真实报错来对照都是我在接入 TaoToken 和调试 LangGraph 时实际遇到过的。报错一401 Unauthorized。最常见的原因是环境变量没读到。检查TAOTOKEN_API_KEY是否在运行代码的 shell 里 export 过或者代码里是否写成了os.environ[TAOTOKEN_API_KEY]但变量名拼错。另一个原因是 Key 复制时带了空格或换行建议重新从控制台复制一次。如果用的是init_chat_model确认api_key参数传的是字符串而不是None。报错二local proxy failed / connection refused。这类报错通常出现在 base_url 写错的情况下。TaoToken 的 API 地址是https://taotoken.net/api注意结尾没有多余的斜杠也不要写成官网首页地址。如果你在本地有网络层配置确认它没有拦截这个域名。代码里base_url参数要完整写全不要只写域名。报错三reading choices / KeyError choices。这个报错说明请求发出去了但返回结构不是预期的 OpenAI 格式。常见原因是模型 ID 写错比如把gpt-4o-mini写成了gpt4o-mini或gpt-4o。TaoToken 的模型 ID 以接入文档页的列表为准复制时不要手打。另一个可能是 Planner 和 Executor 用了不同的 base_url导致其中一个走错了通道。报错四OAuth / authentication failed。如果你在 Claude Code 或类似工具里配置 TaoToken遇到 OAuth 相关报错通常是因为工具默认走了自己的登录流程。这时候需要在工具的配置文件里显式指定 Base URL、Key 和 Model ID 三件套。以 Claude Code 为例配置文件里要写清楚base_url指向 TaoToken 的 API 地址api_key填控制台生成的 Keymodel填你要用的模型 ID。三者缺一不可只填 Key 不填 base_url 会走默认通道导致认证失败。报错五Subgraph 返回空结果。如果task_results里追加的是空字符串检查_execute_task_node里task_result[messages][-1]是否真的存在。有时候 Subgraph 在task_llm节点就 END 了messages 列表可能只有一条 SystemMessage。可以在 Subgraph 的task_llm_node里加一行日志确认 LLM 是否正常返回了 content。报错六Planner 返回的不是 JSON。如果plan()走了 fallback 分支按行分割说明 LLM 没有严格输出 JSON。可以在 prompt 里加一句“不要用 markdown 代码块包裹”或者把temperature再确认一遍是 0。如果模型本身对 JSON 支持不好换一个指令遵循更强的模型 ID。6. 继续往下走从 Planner 到长期记忆与人工审批跑通这一篇之后你的 MiniHarness 已经具备了任务分解能力。v2 相比 v1 新增的核心就是 Planner 和 Subgraph 两块Planner 负责“想清楚”Subgraph 负责“把每个子任务干净地执行掉”主图的 plan → execute → plan 循环负责“做一步检查一步”。下一步可以往两个方向扩展。一个是长期记忆目前task_results只存在内存里进程结束就没了。可以把它接到 LangGraph 的 Store 或外部数据库让 Agent 跨会话记住之前完成过哪些子任务。另一个是Human-in-the-Loop在_plan_route里加一个中断点让用户在计划生成后先审批再执行适合那些子任务有副作用的场景。如果你打算把这条链路用到实际项目里建议把模型调用统一走 TaoToken 的 API 通道Planner 和 Executor 共用一套 Key 和 base_url换模型时只改模型名。需要长期跑编码类 Agent 的话可以看看 Coding Plan 的额度方案只是做验证和调试API Keys 页面生成的 Key 就够用。接入文档里有完整的 base_url 和模型 ID 对照配置时对着抄能省掉不少排查 401 和 reading choices 的时间。