拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从“暴力调用”到“精细编排”:解构 AI Agent 的大脑核心——Planner 与 Router

从“暴力调用”到“精细编排”:解构 AI Agent 的大脑核心——Planner 与 Router

1. 当 Agent 挂载 100 个 MCP Server 时,为什么“全量加载”必然崩

先说一个我踩过的坑。早期做内部工具助手时,我把十几个 MCP Server 的 Tool 定义一股脑塞进 System Prompt,刚开始只有 5 个工具时一切正常,等到接入第 20 个 Server、工具数逼近 80 个时,模型开始出现诡异行为:明明让它查数据库,它却去调 GitHub 的搜索接口;让它发 Slack 消息,它把 Jira 的参数格式套了上去。排查了半天才发现,问题不在模型本身,而在于上下文里塞了太多无关的工具描述。

这就是 AI Agent 从“暴力调用”走向“精细编排”的分水岭。所谓暴力调用,就是把所有可用工具一次性喂给大模型,让它自己挑;所谓精细编排,则是引入 Planner(规划者)和 Router(路由者)两层抽象,让“想清楚做什么”和“找准确用谁做”各司其职。本文聚焦 AI Agent 中 Planner 与 Router 的协作机制,以 LangGraph 与 MCP 为技术底座,拆解任务规划与路由分发的编排逻辑,并给出可复制的 LangGraph 节点配置与 MCP 工具注册示例。

先说清楚这套架构适合谁:如果你正在用 LangGraph、Cline、Claude Code 这类工具构建多工具 Agent,或者你的 MCP Server 数量已经超过 10 个、开始感受到 Token 成本和幻觉压力,那这篇文章就是写给你的。如果你只是单工具调用,暂时用不上这么重的编排,但理解这套思路对后续扩展有好处。

全量加载为什么行不通,核心是三个物理约束。第一是上下文窗口的有效推理质量。虽然现在模型动辄 128K、200K 窗口,但“能塞进去”和“能推理对”是两回事。工具定义越多,注意力越涣散,模型在长 Prompt 中会丢失重点,这就是常说的 Lost in the middle。第二是推理成本。每一轮对话都要重复传输巨大的工具集定义,100 个 Server 可能意味着数百个函数定义、数万个 Token,API 账单会指数级增长。第三是动态生态。企业里的 MCP Server 是动态增减的,要求模型每一刻都“记住”所有端点,既不科学也不可扩展。

结论很明确:Agent 架构必须从“全量加载”演进为“按需加载”。而实现按需加载的关键,就是把 Planner 和 Router 拆开。Planner 决定“做什么”,负责把模糊指令拆成清晰步骤;Router 决定“用谁做”,负责在成百上千个工具里筛出当前步骤真正需要的 Top-K 个候选项。两者配合,才能让 Agent 在工具丛林里游刃有余。

2. TaoToken 前置准备:给 Planner 和 Router 配一个稳定的模型入口

在动手写 LangGraph 节点之前,得先解决模型调用的问题。Planner 需要强推理模型来拆解复杂任务,Router 需要快速模型来做语义筛选,这两类调用如果各自去对接不同厂商,配置会非常碎。我的做法是统一走一个兼容 OpenAI 协议的入口,TaoToken 就是这样一个选择,它提供模型对话和 API 调用能力,Base URL 和 Key 的配置方式和标准 OpenAI SDK 一致,省去了多厂商适配的麻烦。

先说明一点:TaoToken 在这里扮演的是模型调用入口的角色,不是替代你的编辑器或 Agent 框架。LangGraph 负责编排逻辑,MCP 负责工具协议,TaoToken 负责把模型请求稳定地送出去。三者是配合关系。

你需要准备两样东西:一个 API Key,以及确认要用的 Model ID。获取 Key 的入口在控制台的 API Keys 页面,地址是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。拿到 Key 之后,Base URL 统一用 https://taotoken.net/api ,注意这个地址不带任何查询参数,直接作为 OpenAI 兼容端点使用。

模型选择上,我的建议是分角色配置。Planner 节点用推理能力强的模型,比如 DeepSeek-V3 或 Claude 系列,它负责逻辑拆解,慢一点没关系,正确性优先。Router 节点用响应快的轻量模型,比如 GPT-4o-mini 或 Claude Haiku 这类,它只做语义匹配,精度够用就行,延迟越低越好。这种“大小模型协同”的策略,能在保证规划质量的同时把整体延迟压下来。

如果你还没想好具体用哪个模型,可以先去模型对话页面试一下不同模型对同一段规划指令的响应差异,地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。试的时候给一段稍微复杂的任务描述,比如“分析上周销售数据并生成报告发到指定频道”,看哪个模型拆解出的步骤更合理、更少遗漏依赖关系。

配置层面,我习惯把模型参数写进环境变量,避免硬编码。下面是一个 .env 的示例结构,你可以直接照着改:

# .env TAOTOKEN_API_KEY=sk-your-key-here TAOTOKEN_BASE_URL=https://taotoken.net/api PLANNER_MODEL=deepseek-ai/DeepSeek-V3 ROUTER_MODEL=gpt-4o-mini

这里有个细节要注意:Base URL 后面不要手动加 /v1,OpenAI SDK 会自动拼接路径。如果你用的是 LangChain 的 ChatOpenAI,配置方式如下:

from langchain_openai import ChatOpenAI import os planner_llm = ChatOpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), model=os.getenv("PLANNER_MODEL"), temperature=0 ) router_llm = ChatOpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), model=os.getenv("ROUTER_MODEL"), temperature=0 )

把 Planner 和 Router 的模型实例分开创建,后面在 LangGraph 节点里各用各的,互不干扰。这一步做完,模型入口就通了,接下来进入真正的编排配置。

3. 可复制的 LangGraph 节点配置与 MCP 工具注册示例

这一节是全文的核心,我会给出完整的 LangGraph 状态定义、Router 节点、Planner 节点和 Executor 节点的配置,以及 MCP 工具的注册方式。代码可以直接跑,你只需要替换 Key 和工具注册表。

先定义 Agent 的状态结构。LangGraph 用 TypedDict 来描述状态在节点之间如何流转:

from typing import Annotated, TypedDict, List import operator class AgentState(TypedDict): input: str # 用户原始指令 relevant_tools: List[str] # Router 筛选出的工具名 plan: List[str] # Planner 生成的步骤 observations: Annotated[List[str], operator.add] # 执行反馈累积 final_response: str # 最终输出

这里 observations 用了 operator.add 作为 reducer,意味着每次 Executor 返回的结果会追加而不是覆盖,方便 Planner 在 Re-Act 循环里看到完整历史。

接下来是 MCP 工具注册表。真实场景里这些描述来自 MCP Server 的 list_tools 接口,这里先用一个字典模拟,重点是 description 字段的写法——它是 Router 做语义匹配的依据,必须写清楚“这个工具做什么、什么时候用”:

MCP_TOOLS_REGISTRY = { "query_database": { "description": "执行 SQL 查询以获取财务数据库中的报表数据,仅在需要分析收入、支出或利润时使用。", "server": "postgres-finance" }, "generate_pdf": { "description": "将文本或 Markdown 内容转换为 PDF 文件并保存到指定路径。", "server": "document-tools" }, "send_slack_message": { "description": "向 Slack 指定频道发送消息,需要提供 channel 和 text 参数。", "server": "slack-connector" }, "search_github": { "description": "搜索 GitHub 仓库、Issue 或 Pull Request。", "server": "github-mcp" }, "calculator": { "description": "执行复杂数学计算,支持四则运算和百分比。", "server": "local-tools" } }

Router 节点的逻辑是:把工具注册表里的 name 和 description 拼成候选清单,让轻量模型从中选出与当前任务相关的工具名。注意这里只返回工具名,不返回完整 Schema,目的是压缩上下文:

from langchain_core.messages import HumanMessage def router_node(state: AgentState): tools_desc = "\n".join( [f"- {name}: {info['description']}" for name, info in MCP_TOOLS_REGISTRY.items()] ) prompt = ( f"你是一个工具路由者。请从以下工具列表中挑选出完成该任务必需的工具名," f"用英文逗号分隔,不要解释。\n\n工具列表:\n{tools_desc}\n\n" f"任务:{state['input']}" ) response = router_llm.invoke([HumanMessage(content=prompt)]) selected = [ t.strip() for t in response.content.split(",") if t.strip() in MCP_TOOLS_REGISTRY ] return {"relevant_tools": selected}

Planner 节点拿到 Router 筛选后的工具集,生成结构化步骤。这里的关键是让 Planner 只看到相关工具,屏蔽无关干扰:

def planner_node(state: AgentState): tools_info = "\n".join( [f"- {t}: {MCP_TOOLS_REGISTRY[t]['description']}" for t in state["relevant_tools"]] ) prompt = ( f"你是一个任务规划者。基于以下可用工具,将任务拆解为有序的执行步骤," f"每行一个步骤,不要编号。\n\n可用工具:\n{tools_info}\n\n" f"任务:{state['input']}" ) response = planner_llm.invoke([HumanMessage(content=prompt)]) steps = [line.strip() for line in response.content.split("\n") if line.strip()] return {"plan": steps}

Executor 节点在真实场景里会去调用 MCP Server,这里先模拟返回,重点是把执行结果写回 observations,供 Planner 下一轮参考:

def executor_node(state: AgentState): results = [] for step in state["plan"]: results.append(f"已执行:{step}") return { "observations": results, "final_response": "任务完成,执行步骤:\n" + "\n".join(state["plan"]) }

最后用 StateGraph 把节点串起来,Router 在前、Planner 在后,这是我在实践中验证过的顺序:

from langgraph.graph import StateGraph, START, END workflow = StateGraph(AgentState) workflow.add_node("router", router_node) workflow.add_node("planner", planner_node) workflow.add_node("executor", executor_node) workflow.add_edge(START, "router") workflow.add_edge("router", "planner") workflow.add_edge("planner", "executor") workflow.add_edge("executor", END) app = workflow.compile()

为什么 Router 要放在 Planner 前面?打个比方:你去一家有 1000 道菜的饭店,如果服务员先让你看整本菜单,你会看晕;高效的做法是你先说“我想吃海鲜”,服务员把菜单翻到海鲜那页,你再从这缩减后的几道菜里组合晚餐。Router 先行就是先翻到那一页,Planner 随后才是组合晚餐。这个顺序能有效解决 Token 爆炸和注意力崩溃。

如果你用的是 Cline 或 Claude Code 这类工具,MCP 工具注册的配置方式略有不同。以 Cline 的 MCP 配置为例,需要在 settings 里声明 Server 连接信息,三件套是 Base URL、Key 和 Model ID:

{ "mcpServers": { "postgres-finance": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-postgres"], "env": { "DATABASE_URL": "postgresql://user:pass@localhost:5432/finance" } } } }

而模型入口的配置,在 Cline 里对应的是 API Provider 设置,Base URL 填 https://taotoken.net/api ,Key 填你的 TaoToken Key,Model ID 填你选的模型名。这三项缺一不可,配错任何一项都会导致请求失败。

4. 验证请求与成功结果:路由命中率和规划步数怎么看

配置写完不代表能跑通,得有一套验证动作来确认 Router 筛得准、Planner 拆得对。我通常从三个维度验证:路由命中率、规划步数合理性、端到端执行结果。

先跑一个最小验证请求。用一段包含明确工具需求的指令,观察 Router 返回的 relevant_tools 是否命中预期:

test_input = "查询过去一周的财务数据库报表,生成一份 PDF 总结,并发送到 Slack 的 finance 频道。" result = app.invoke({"input": test_input}) print("Router 筛选结果:", result["relevant_tools"]) print("Planner 规划步骤:", result["plan"]) print("最终输出:", result["final_response"])

预期输出应该是 Router 命中 query_database、generate_pdf、send_slack_message 三个工具,而 search_github 和 calculator 被过滤掉。如果 Router 把 search_github 也选进来了,说明工具 description 写得不够区分,或者 Router 模型能力不足,需要调整。

路由命中率怎么量化?我的做法是准备一组测试用例,每条用例标注“期望命中的工具集”,然后批量跑 Router 节点,统计命中率:

test_cases = [ {"input": "查一下上周的销售数据", "expected": ["query_database"]}, {"input": "把这份报告转成 PDF", "expected": ["generate_pdf"]}, {"input": "在 Slack 上通知团队", "expected": ["send_slack_message"]}, {"input": "算一下 15% 的增长率", "expected": ["calculator"]}, ] hit = 0 for case in test_cases: res = router_node({"input": case["input"]}) if set(res["relevant_tools"]) == set(case["expected"]): hit += 1 print(f"路由命中率:{hit}/{len(test_cases)} = {hit/len(test_cases)*100}%")

实测下来,工具 description 写得越具体,命中率越高。比如把“查询数据库”改成“执行 SQL 查询以获取财务数据库中的报表数据,仅在需要分析收入、支出或利润时使用”,命中率能从 70% 左右提升到 90% 以上。这个细节值得花时间打磨。

规划步数合理性怎么判断?看 Planner 输出的步骤数是否与任务复杂度匹配。简单任务 1-2 步,中等任务 3-5 步,复杂任务 5-8 步。如果简单任务被拆成 10 步,说明 Planner 过度规划,可能是模型 temperature 太高或者 Prompt 里没限制步数。如果复杂任务只拆出 1 步,说明规划不足,需要检查工具描述是否让 Planner 理解了任务依赖。

端到端验证时,我建议打开可观测性工具看完整链路。Phoenix 或 LangSmith 都能追踪每个节点的输入输出,你能清楚看到 Router 选了什么、Planner 想了什么、Executor 返回了什么。这种透明度对调试至关重要,尤其是当最终结果不符合预期时,能快速定位是路由错了还是规划错了。

一个成功的验证结果长这样:Router 精准筛出 3 个工具,Planner 生成 4 个有序步骤(查询→分析→生成 PDF→发送 Slack),Executor 按序执行并返回完整反馈。整个过程 Token 消耗比全量加载降低 80% 以上,延迟也在可接受范围内。

5. 本篇常见错误排查:401、local proxy failed、reading choices 怎么解

配置和验证过程中,最容易卡住的就是各种报错。这一节我把常见错误和排查路径整理出来,对照着看能省不少时间。

401 Unauthorized是最常见的。原因通常是 Key 没配、Key 过期、或者 Base URL 写错了。排查步骤:先确认环境变量里 TAOTOKEN_API_KEY 确实有值,再确认 Base URL 是 https://taotoken.net/api 而不是别的地址。如果用的是 Cline 或 Claude Code,检查 settings 里的 API Provider 配置,Base URL、Key、Model ID 三件套是否齐全。特别注意 Base URL 后面不要手动加 /v1,SDK 会自动拼。如果还是 401,去控制台重新生成一个 Key 试试,地址是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。

local proxy failed这个报错通常出现在网络层。它意味着请求没能到达目标端点。排查方向:确认你的运行环境能正常访问外网,检查是否有本地代理配置冲突。如果你在代码里设置了 HTTP_PROXY 或 HTTPS_PROXY 环境变量,尝试临时清掉再跑。另外确认 Base URL 拼写正确,少一个字符都会导致连接失败。

reading choices 相关报错,比如 “Error reading choices” 或返回结构解析失败,多半是响应格式不符合 OpenAI 兼容规范。这种情况先确认你用的模型是否支持 OpenAI 协议,再检查 SDK 版本是否过旧。如果是 LangChain 的 ChatOpenAI,升级到最新版通常能解决。还有一种可能是 Router 节点里 response.content 为空,导致后续 split 报错,加一个空值判断就能规避:

content = response.content or "" selected = [t.strip() for t in content.split(",") if t.strip() in MCP_TOOLS_REGISTRY]

OAuth 相关报错,如果你在 MCP Server 配置里用了需要 OAuth 的服务,报错通常提示 token 无效或授权过期。排查步骤:确认 OAuth token 是否还在有效期,检查回调地址配置是否正确。对于本地开发的 MCP Server,建议先用不需要 OAuth 的工具做验证,跑通链路后再逐个接入需要授权的服务。

模型返回空计划或计划格式混乱,这不是报错但很常见。原因通常是 Planner 的 Prompt 约束不够强。解决办法是在 Prompt 里明确要求“每行一个步骤,不要编号,不要解释”,并在解析时做容错处理。如果模型仍然不听话,降低 temperature 到 0,或者换一个指令遵循能力更强的模型。

Router 筛选结果为空,说明没有工具名匹配上。检查工具注册表里的 name 是否和模型返回的一致,有时候模型会返回带引号或带空格的名字,需要 strip 处理。另外确认 Router 的 Prompt 里明确说了“用英文逗号分隔”,否则模型可能用中文逗号或换行分隔。

排查的核心思路是:先确认模型入口通不通(401 类),再确认网络通不通(proxy 类),最后确认数据格式对不对(reading choices 类)。按这个顺序排查,大部分问题都能快速定位。

6. 从能跑到好用:Planner 与 Router 的持续调优方向

跑通最小闭环只是起点,真正让这套架构在生产环境稳定运行,还需要在几个方向上持续调优。

第一是工具描述的打磨。Router 的命中率直接取决于 description 的质量。我的经验是,好的 description 要回答三个问题:这个工具做什么、什么时候用、什么时候不用。比如“查询数据库”这种描述太泛,改成“执行 SQL 查询以获取财务数据库中的报表数据,仅在需要分析收入、支出或利润时使用”就具体多了。花在 description 上的时间,会直接转化为路由准确率的提升。

第二是缓存高频路由路径。对于反复出现的指令模式,没必要每次都让 Router 跑一遍语义匹配。可以建一个 Query 到 Selected Tools 的缓存层,相似度极高时直接命中缓存,跳过 Router 推理。这能显著降低延迟和成本。

第三是引入 Skill 抽象层。当 MCP 工具数量从 10 增长到 100,即便 Router 筛得再准,Planner 面对的全是原子级工具也会陷入步骤过多的泥潭。这时候把多个工具封装成高阶 Skill,比如把“搜索→爬取→摘要→导出”封装成一个 Research_Skill,Planner 的思考负载能从 6 步降到 2 步,出错概率大幅下降。

第四是增量式规划。不要每次执行完一步都让 Planner 重头写一遍完整计划,让它只输出“当前状态”和“下一步动作”,上下文保持在最小限度。这能避免 Token 浪费,也能减少长链路中的逻辑漂移。

如果你打算把这套架构用于长期编码或 Agent 场景,可以考虑 Coding Plan,地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ,它针对持续性的编码任务做了优化。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有更详细的配置说明和示例。

最后说一个我踩过的坑:不要一上来就追求完美架构。先用 3-5 个工具跑通 Router 加 Planner 的最小闭环,确认链路通了、结果对了,再逐步增加工具数量和 Skill 抽象。架构的复杂度应该跟着业务需求走,而不是反过来。从暴力调用到精细编排,本质上是让每一层只做自己最擅长的事,Router 管广度,Planner 管深度,Executor 管执行,各司其职,系统才能稳。

返回列表