1. 从单 Agent 到多智能体为什么你的 LangChain 项目需要拆分角色如果你已经用 LangChain 写过几个 Agent大概率遇到过这样的场景一个 Agent 既要查天气、又要算日期、还要给生活建议系统提示词越写越长工具描述堆到几十个模型开始走神——该调工具的时候不调不该调的时候乱调。这不是模型不行而是你把太多职责塞进了一个上下文窗口。多智能体系统Multi-Agent System要解决的核心问题就是这个把一个大而全的 Agent 拆成若干职责单一的智能体通过 Subagents 分工、Handoffs 交接、Router 路由、Custom Workflow 编排这几种模式把它们组织起来。它适合谁适合已经跑通单 Agent、但发现提示词膨胀到难以维护、或者需要多个垂直领域协作的开发者。如果你还在写第一个 Agent建议先把单 Agent 加工具跑顺再来看这篇。我试过在一个客服场景里硬塞保修判断、问题分类、方案生成三个逻辑进一个 Agent结果提示词超过 3000 字模型经常跳过保修确认直接给方案。拆成 Handoffs 之后每一步只保留当前阶段需要的提示词和工具问题定位也清晰了。下面我把 Subagents、Handoffs、Router、Custom Workflow 四种模式的配置骨架和验证动作完整走一遍模型调用统一走 TaoToken 的 Key/API 通道省去多模型切换时反复改 base_url 的麻烦。2. TaoToken 前置统一 Key 与 API 通道多智能体系统里经常出现一个尴尬情况Supervisor 用 GPT 系列做调度子 Agent 用 Claude 做长文本分析Router 的分类节点又想用便宜的小模型。如果每个模型都单独配一套 Key 和 base_url代码里到处是环境变量调试时改一处漏一处。TaoToken 在这里的作用是提供一个统一的 API 入口你只需要一个 Key就能在同一个base_url下调用不同模型。对 LangChain 来说这意味着你可以在ChatOpenAI里直接改model参数切换模型而不用动base_url和api_key。先拿到 Key访问https://taotoken.net/api-keys带 UTM?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建一个 API Key。然后在项目里配置环境变量export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意base_url用https://taotoken.net/api不要加 UTM 参数那是给网页链接用的API 请求带上反而可能出问题。Key 的管理页面在 consolehttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite可以在这里看用量和余额。如果你打算长期跑编码类 Agent比如让子 Agent 自动改代码、跑测试可以了解一下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite它针对高频编码调用做了额度优化。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有各语言 SDK 的配置示例。3. 可复制配置四种协作模式的骨架代码3.1 统一模型客户端配置先写一个公共的模型工厂所有 Agent 都从这里拿模型实例方便统一走 TaoTokenimport os from langchain_openai import ChatOpenAI def get_model(model_name: str gpt-4o-mini, temperature: float 0): return ChatOpenAI( modelmodel_name, temperaturetemperature, api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) # 调度用稍强的模型子任务用轻量模型 supervisor_model get_model(gpt-4o) worker_model get_model(gpt-4o-mini)这样切换模型只改一个字符串不用碰 base_url。3.2 SubagentsSupervisor 把子代理包装成工具Subagents 模式的核心是主代理Supervisor通过工具调用来协调子代理子代理无状态、上下文隔离、结果返回给主代理。推荐用 Single dispatch tool 方式主代理只看到一个task工具通过agent_name参数指定具体子代理这样主代理的上下文窗口不会被一堆工具描述占满。from langchain.agents import create_agent from langchain.tools import tool from langchain_core.messages import HumanMessage # 定义两个子代理 research_agent create_agent( modelworker_model, tools[], system_prompt你是研究助手负责事实核查和信息检索。只返回结论不要寒暄。, ) writer_agent create_agent( modelworker_model, tools[], system_prompt你是写作助手负责把要点组织成通顺段落。只返回正文。, ) SUBAGENTS { research: research_agent, writer: writer_agent, } tool def task(agent_name: str, description: str) - str: 为一项任务启动一个临时的子代理。agent_name 可选 research 或 writer。 agent SUBAGENTS[agent_name] result agent.invoke({messages: [HumanMessage(contentdescription)]}) return result[messages][-1].content supervisor create_agent( modelsupervisor_model, tools[task], system_prompt( 你负责协调专业子代理。可用代理\n - research研究与事实核查\n - writer内容创作与编辑\n 请使用 task 工具分配工作最后整合结果回复用户。 ), )关键点子代理每次调用都是全新上下文不记忆历史。如果子任务需要依赖用户原始问题要在task工具里通过ToolRuntime把原始消息注入进去否则子代理会指代不明。3.3 Handoffs用 Command 驱动状态切换Handoffs 适合有明确阶段顺序的流程比如客服的保修确认 → 问题分类 → 方案生成。核心机制是工具返回Command对象更新状态变量current_step中间件根据当前步骤动态切换系统提示词和可用工具。from typing import Literal from typing_extensions import NotRequired from langchain.agents import AgentState, create_agent from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.tools import tool, ToolRuntime from langchain_core.messages import ToolMessage from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command SupportStep Literal[warranty_collector, issue_classifier, resolution_specialist] class SupportState(AgentState): current_step: NotRequired[SupportStep] warranty_status: NotRequired[Literal[in_warranty, out_of_warranty]] issue_type: NotRequired[Literal[hardware, software]] tool def record_warranty_status( status: Literal[in_warranty, out_of_warranty], runtime: ToolRuntime[None, SupportState], ) - Command: 记录保修状态并进入问题分类步骤。 return Command(update{ messages: [ToolMessage( contentf保修状态已记录{status}, tool_call_idruntime.tool_call_id, )], warranty_status: status, current_step: issue_classifier, }) tool def record_issue_type( issue_type: Literal[hardware, software], runtime: ToolRuntime[None, SupportState], ) - Command: 记录问题类型并进入方案步骤。 return Command(update{ messages: [ToolMessage( contentf问题类型已记录{issue_type}, tool_call_idruntime.tool_call_id, )], issue_type: issue_type, current_step: resolution_specialist, })这里有个容易踩的坑Command的update里必须包含ToolMessage且tool_call_id要和触发这次调用的tool_call_id匹配。少了它对话历史里工具调用没有响应下一轮模型会报格式错误。中间件负责根据current_step注入对应的提示词和工具STEP_CONFIG { warranty_collector: { prompt: 你是客服当前步骤保修验证。询问设备是否在保修期内用 record_warranty_status 记录。, tools: [record_warranty_status], }, issue_classifier: { prompt: 你是客服当前步骤问题分类。保修状态 {warranty_status}。请用户描述问题判断硬件还是软件用 record_issue_type 记录。, tools: [record_issue_type], }, resolution_specialist: { prompt: 你是客服当前步骤方案。保修 {warranty_status}问题 {issue_type}。给出具体解决方案。, tools: [], }, } wrap_model_call def apply_step_config(request: ModelRequest, handler) - ModelResponse: step request.state.get(current_step, warranty_collector) cfg STEP_CONFIG[step] system_prompt cfg[prompt].format(**request.state) request request.override(system_promptsystem_prompt, toolscfg[tools]) return handler(request) agent create_agent( modelsupervisor_model, tools[record_warranty_status, record_issue_type], state_schemaSupportState, middleware[apply_step_config], checkpointerInMemorySaver(), )checkpointer是必须的否则跨轮次的状态current_step、warranty_status不会持久化第二轮对话就丢了。3.4 Router并行分发与结果合成Router 适合一次查询需要同时问多个知识源的场景。它和 Subagents 的区别在于Router 是无状态的一次性分派流程是分类 → 并行调用 → 合成Subagents 是有状态的协调者主代理可以多轮调用子代理。用SendAPI 实现并行分发import operator from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Send from langchain_core.messages import HumanMessage class RouterState(TypedDict): query: str classifications: list results: Annotated[list, operator.add] final_answer: str def classify_query(state: RouterState) - dict: structured_llm supervisor_model.with_structured_output(ClassificationResult) result structured_llm.invoke([ {role: system, content: 分析查询确定需要咨询哪些知识源github、notion、slack。}, {role: user, content: state[query]}, ]) return {classifications: result.classifications} def route_to_agents(state: RouterState) - list[Send]: return [ Send(c[source], {query: c[query]}) for c in state[classifications] ] def synthesize_results(state: RouterState) - dict: merged \n.join(str(r) for r in state[results]) answer supervisor_model.invoke([ {role: system, content: 把以下多源结果整合成连贯回答。}, {role: user, content: merged}, ]).content return {final_answer: answer} workflow ( StateGraph(RouterState) .add_node(classify, classify_query) .add_node(github, query_github) .add_node(notion, query_notion) .add_node(slack, query_slack) .add_node(synthesize, synthesize_results) .add_edge(START, classify) .add_conditional_edges(classify, route_to_agents, [github, notion, slack]) .add_edge(github, synthesize) .add_edge(notion, synthesize) .add_edge(slack, synthesize) .add_edge(synthesize, END) .compile() )results字段用Annotated[list, operator.add]声明LangGraph 会自动把并行节点的返回值合并进同一个列表不用手动收集。3.5 Custom Workflow把 Agent 塞进自定义图当标准模式不够用时用 LangGraph 自定义执行流程在节点里调用 LangChain Agentfrom langgraph.graph import StateGraph, START, END def agent_node(state: dict) - dict: result supervisor.invoke({ messages: [{role: user, content: state[query]}] }) return {answer: result[messages][-1].content} workflow ( StateGraph(dict) .add_node(agent, agent_node) .add_edge(START, agent) .add_edge(agent, END) .compile() )这种方式的灵活性在于你可以在 Agent 节点前后插入确定性的处理逻辑比如先做关键词过滤、再调 Agent、最后做格式校验把确定性逻辑和智能体行为结合起来。4. 验证请求跑通一条完整协作链路配置写完后先验证 Subagents 链路是否通。用一个需要两个子代理协作的问题result supervisor.invoke({ messages: [HumanMessage( content帮我查一下 LangGraph 的 Send API 是做什么的然后用一段话解释清楚。 )] }) for msg in result[messages]: msg.pretty_print()预期输出里应该能看到Supervisor 先调用task(agent_nameresearch, ...)拿到研究结论后调用task(agent_namewriter, ...)最后整合成回复。如果 Supervisor 直接自己回答了没调工具检查系统提示词里是否明确写了必须使用 task 工具分配工作。再验证 Handoffs 的状态流转。用同一个thread_id连续调用四轮config {configurable: {thread_id: test_handoff_1}} r1 agent.invoke({messages: [HumanMessage(你好我的手机屏幕碎了)]}, config) r2 agent.invoke({messages: [HumanMessage(是的还在保修期内)]}, config) r3 agent.invoke({messages: [HumanMessage(屏幕是摔坏的有裂痕)]}, config) r4 agent.invoke({messages: [HumanMessage(我该怎么办)]}, config) print(r4[current_step]) # 应该是 resolution_specialist print(r4[warranty_status]) # in_warranty print(r4[issue_type]) # hardware如果current_step一直停在warranty_collector说明Command的update没生效检查工具是否真的返回了Command对象以及checkpointer是否配置。Router 的验证看并行是否真的并行在query_github、query_notion、query_slack三个节点里各加一行print加时间戳如果三个时间戳几乎相同说明Send并行调度生效了。5. 本篇常见错排查报错一InvalidUpdateError: Expected dict, got Command这个通常出现在 Handoffs 的多 Agent 子图方式里。如果你在节点函数里直接返回Command但图的状态 schema 没声明对应字段就会报这个。解决方法是确保StateGraph的状态类里声明了active_agent等字段并且Command的update里的键都在状态类里定义过。报错二子代理返回空字符串Subagents 模式下子代理的invoke返回的是完整消息列表最后一条不一定是AIMessage。如果子代理最后一步是工具调用messages[-1]可能是ToolMessage。稳妥的写法是过滤出AIMessagefrom langchain_core.messages import AIMessage ai_msgs [m for m in result[messages] if isinstance(m, AIMessage)] return ai_msgs[-1].content if ai_msgs else 报错三Handoffs 第二轮对话状态丢失九成是没配checkpointer或者每次调用传的thread_id不一样。thread_id是状态持久化的键同一个对话必须用同一个thread_id。另外InMemorySaver只存在内存里进程重启就没了生产环境要换成数据库版的 checkpointer。报错四Router 的results字段报类型错误Annotated[list, operator.add]里的list要写具体类型比如Annotated[list[AgentOutput], operator.add]否则 LangGraph 在合并时可能不知道怎么处理。另外每个并行节点返回的字典里results键对应的值必须是列表不能是单个对象。报错五模型调用返回 401 或 base_url 错误检查TAOTOKEN_BASE_URL是不是写成了带 UTM 的网页地址。API 请求的 base_url 是https://taotoken.net/api不带任何查询参数。Key 从https://taotoken.net/api-keys创建后直接复制注意不要带多余空格。6. 接入与排障把模型通道固定下来多智能体系统调试时最烦的就是模型调用不稳定导致误判——你以为是路由逻辑写错了其实是某次请求超时返回了空。把模型通道统一到 TaoToken 之后至少排除了多个 Key 轮换导致限流这类干扰因素。如果你在接入过程中遇到 401、429 或者 base_url 配置问题直接看接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有各框架的配置示例和常见错误码说明。Key 的创建和管理在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite建议给多智能体项目单独建一个 Key方便按项目看用量。想先验证某个模型在多智能体调度里的表现可以到模型对话页面直接试https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite用同一套提示词对比不同模型的工具调用准确率再决定 Supervisor 和子 Agent 分别用哪个模型。长期跑编码类 Agent 的话Coding Plan 的额度模型更适合高频调用场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite。最后给一个实操建议先把 Subagents 跑通因为它的心智模型最简单——主代理调工具子代理干活。等你能稳定控制子代理的输入输出格式了再上 Handoffs 处理有状态的多阶段流程。Router 和 Custom Workflow 留到确实需要并行查询或自定义编排时再引入不要一上来就全用上调试成本会翻倍。