LangChain 多智能体之 Subagents：让一个主管去指挥一队专家

当你想让一个 Agent 既懂市场、又懂财务、还懂技术时最容易想到的做法是把所有工具、所有领域知识、所有提示词一股脑塞进一个超级 Agent。结果就是——上下文越来越长、提示词互相打架、领域知识彼此干扰模型越来越糊涂。Subagents子智能体模式就是来解决这件事的让一个主智能体Supervisor当指挥把具体任务分派给各个专精的子智能体去做最后整合结果。这篇文章结合 3 个完整可运行的例子讲清这个模式怎么用。所有代码都在正文里复制即可跑。一、Subagents 是什么Subagents 是一种架构模式由一个中心主智能体Supervisor通过调用工具的方式协调多个子智能体。用户 → 主智能体(Supervisor) → 调用子智能体工具 → 子智能体执行 → 返回结果 → 主智能体整合 → 用户关键点在于每个子智能体被封装成主智能体的一个工具。主智能体并不知道子智能体内部是怎么回事它只看到有这么几个工具可以调跟调用一个普通函数没区别。为什么需要它价值说明防止上下文膨胀每个子智能体在独立、干净的上下文窗口中运行领域解耦每个专家只专注自己的领域不加载无关知识集中控制主智能体掌握全局子智能体专注执行团队并行开发不同团队可以独立开发各自的子智能体一句话分而治之。主管负责派活和汇总专家负责把活干好。共用的模型初始化下文每个例子都从.env读取模型配置MODEL_NAME/OPENAI_API_BASE/OPENAI_API_KEY。每段代码都是完整可独立运行的所以这段初始化会重复出现importosfromdotenvimportload_dotenvfromlangchain.chat_modelsimportinit_chat_model load_dotenv()modelinit_chat_model(os.getenv(MODEL_NAME,glm-5.1),model_provideropenai,base_urlos.getenv(OPENAI_API_BASE),api_keyos.getenv(OPENAI_API_KEY),streamingTrue,)二、模式一Tool per Agent每个子智能体一个工具最直观的做法给每个子智能体单独写一个工具函数。工具内部invoke对应的子智能体把结果返回给主智能体。下面有市场研究、财务分析两个专家主智能体根据用户问题自行决定调用哪个、或两个都调importosfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.toolsimporttool load_dotenv()modelinit_chat_model(os.getenv(MODEL_NAME,glm-5.1),model_provideropenai,base_urlos.getenv(OPENAI_API_BASE),api_keyos.getenv(OPENAI_API_KEY),streamingTrue,)# 1. 创建专业子智能体research_agentcreate_agent(modelmodel,system_prompt你是市场研究专家擅长行业分析、竞争格局和市场趋势判断。 必须在最后一条消息中包含摘要格式摘要[2-3 句话总结] ,)finance_agentcreate_agent(modelmodel,system_prompt你是财务分析专家擅长财务数据解读、盈利能力分析和财务预测。 必须在最后一条消息中包含摘要格式摘要[2-3 句话总结关键财务发现] ,)# 2. 把每个子智能体封装成一个工具tooldefcall_research_agent(query:str)-str:调用市场研究专家。用于行业趋势、竞争格局、市场机会分析不适合财务或技术问题。 Args: query: 研究查询例如分析中国 AI 助手市场的竞争格局 print(f\n→ 市场研究子智能体:{query})resultresearch_agent.invoke({messages:[{role:user,content:query}]})returnresult[messages][-1].contenttooldefcall_finance_agent(query:str)-str:调用财务分析专家。用于财务数据解读、盈利能力与成本分析、财务预测不适合市场或技术问题。 Args: query: 财务查询例如分析该业务的盈利能力 print(f\n→ 财务分析子智能体:{query})resultfinance_agent.invoke({messages:[{role:user,content:query}]})returnresult[messages][-1].content# 3. 创建主智能体Supervisor子智能体作为它的工具supervisorcreate_agent(modelmodel,tools[call_research_agent,call_finance_agent],system_prompt你是项目协调员Supervisor。分析用户需求判断需要哪些专家 依次调用对应工具并把各专家结果整合成一个完整回复。 实际调用工具并整合结果不要只说我需要...。,)if__name____main__:resultsupervisor.invoke({messages:[{role:user,content:我想开发一个 AI 助手产品帮我分析市场前景和盈利能力}]})print(result[messages][-1].content)这里有两个细节决定成败工具的 docstring 就是路由说明书。主智能体靠每个工具的描述来决定调谁所以用于什么 / 不适合什么要写清楚——不适合财务或技术问题这种排除性描述能显著降低误调。子智能体的输出要约束格式。子智能体的 system_prompt 里要求最后一条消息包含摘要这样主智能体result[messages][-1].content拿到的就是干货而不是一堆中间过程。适用场景子智能体数量较少 10 个、相对固定。三、模式二Single Dispatch单一调度工具当子智能体多起来比如几十个一个子智能体一个工具会把主智能体的工具列表撑爆模型选择也变困难。这时换个思路只暴露一个通用调度工具task(agent_name, query)用一个注册表做动态路由。新增子智能体只需往注册表加一项主智能体的工具列表始终只有task和list_agents两个importosfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.toolsimporttool load_dotenv()modelinit_chat_model(os.getenv(MODEL_NAME,glm-5.1),model_provideropenai,base_urlos.getenv(OPENAI_API_BASE),api_keyos.getenv(OPENAI_API_KEY),streamingTrue,)# 1. 创建子智能体并登记到注册表research_agentcreate_agent(modelmodel,system_prompt你是市场研究专家。必须在最后一条消息包含摘要格式摘要[2-3 句话总结],)writer_agentcreate_agent(modelmodel,system_prompt你是专业写作者擅长撰写文章、报告和文案。必须在最后一条消息包含你撰写的内容。,)reviewer_agentcreate_agent(modelmodel,system_prompt你是内容审查专家检查文章质量、逻辑和语法。必须在最后一条消息包含审查意见格式摘要[审查结果和改进建议],)SUBAGENTS{research:{agent:research_agent,description:市场研究、竞争分析、行业趋势},writer:{agent:writer_agent,description:撰写文章、报告、文案},reviewer:{agent:reviewer_agent,description:内容审查、质量检查、改进建议},}# 2. 定义单一调度工具 列表工具tooldeftask(agent_name:str,query:str)-str:调用指定子智能体执行任务根据名称动态路由到对应专家。 Args: agent_name: 子智能体名称research/writer/reviewer query: 任务描述或查询 print(f\n→ 调度: agent{agent_name}, query{query})ifagent_namenotinSUBAGENTS:returnf错误未知子智能体 {agent_name}。可用{list(SUBAGENTS)}agentSUBAGENTS[agent_name][agent]resultagent.invoke({messages:[{role:user,content:query}]})returnresult[messages][-1].contenttooldeflist_agents()-str:列出所有可用的子智能体及其描述。不确定有哪些专家可用时调用。return\n.join(f-{name}:{info[description]}forname,infoinSUBAGENTS.items())# 3. 创建主智能体Supervisorsupervisorcreate_agent(modelmodel,tools[task,list_agents],system_prompt你是内容创作团队的协调员Supervisor。 工具list_agents() 查看可用专家task(agent_name, query) 调用指定专家。 流程分析需求 → 必要时 list_agents() → 用 task() 调用对应专家 → 整合结果。 实际调用工具并整合结果。,)if__name____main__:resultsupervisor.invoke({messages:[{role:user,content:帮我写一篇关于 AI 发展趋势的文章先做一下市场调研}]})print(result[messages][-1].content)两种模式对比Tool per AgentSingle Dispatch工具数量每个子智能体一个固定 1~2 个新增子智能体要写新工具注册表加一项路由方式模型选工具模型填agent_name参数适用场景 10 个、固定 10 个、动态增减还多了个list_agents()工具——它让主智能体能自查有哪些专家可用配合动态注册表特别有用。四、上下文隔离与注入前面说过 Subagents 的一大价值是上下文隔离每个子智能体跑在独立、干净的上下文里。这是优点但也带来一个问题——子智能体默认看不到主对话的历史它只收到主智能体传来的那一句query。大多数时候这没问题。但有时你希望子智能体知道前因后果比如用户之前已经聊过背景现在的研究要结合上下文。怎么办主动把主对话历史注入给子智能体。做法是给工具加一个runtime: ToolRuntime参数通过runtime.state读到主智能体的当前状态含完整 messages过滤后随 query 一起传给子智能体importosfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.messagesimportAIMessage,HumanMessage,SystemMessagefromlangchain_core.toolsimporttoolfromlanggraph.prebuiltimportToolRuntime load_dotenv()modelinit_chat_model(os.getenv(MODEL_NAME,glm-5.1),model_provideropenai,base_urlos.getenv(OPENAI_API_BASE),api_keyos.getenv(OPENAI_API_KEY),streamingTrue,)research_agentcreate_agent(modelmodel,system_prompt你是市场研究专家。你会收到之前的对话历史与当前查询 请结合上下文给出更精准的研究结果。必须在最后一条消息包含摘要。,)tooldefcall_research_with_context(query:str,runtime:ToolRuntime)-str:调用市场研究专家并自动带入之前的对话历史。需基于先前讨论做研究时使用。 Args: query: 研究查询 print(f\n→ 研究子智能体带上下文:{query})# 从主智能体状态提取历史只保留纯文本的 Human/AI 消息排除工具调用相关消息history[msgformsginruntime.state.get(messages,[])ifisinstance(msg,HumanMessage)or(isinstance(msg,AIMessage)andnotmsg.tool_calls)]recenthistory[-3:]# 最近 3 条作为上下文sub_messages[SystemMessage(content以下是之前的对话历史供你参考),*recent,HumanMessage(contentquery),]print(f 注入消息数:{len(sub_messages)})resultresearch_agent.invoke({messages:sub_messages})returnresult[messages][-1].content supervisorcreate_agent(modelmodel,tools[call_research_with_context],system_prompt你是项目协调员。可调用 call_research_with_context(query) 进行研究 该工具会自动带入之前的对话历史。,)if__name____main__:resultsupervisor.invoke({messages:[HumanMessage(content我想分析一下 AI 助手市场)]})print(result[messages][-1].content)两个要点runtime.state是主智能体的当前状态runtime.state[messages]就是主对话的完整消息列表。工具签名里加上runtime: ToolRuntime框架会自动注入它这个参数对 LLM 隐藏模型不会也不用传。要过滤再注入不要把整段历史原样塞过去。这里只保留纯文本的HumanMessage和无工具调用的AIMessage排除掉ToolMessage和带tool_calls的中间消息——否则子智能体会被一堆工具调用细节干扰而且白白消耗上下文。这其实是在隔离和共享之间做权衡默认隔离保持干净需要时再精准注入一小段相关历史。五、最佳实践子智能体名称和描述要清晰它们直接决定主智能体的路由准确率。工具 docstring 要写明用于什么、不适合什么。约束子智能体输出格式在子智能体的 system_prompt 里要求固定格式如最后一条消息包含摘要确保主智能体能稳定提取关键信息。善用上下文隔离子智能体默认无状态、上下文干净这是特性不是缺陷只在确有必要时才注入历史。跨调用记忆用 checkpointer如果某个子智能体需要在多次调用间保持记忆给它加checkpointerMemorySaver()并传入固定thread_id。少量用 Tool per Agent大量用 Single Dispatch避免主智能体工具列表过长导致选择困难。六、总结主题要点示例核心模式Supervisor 把子智能体当工具调用、整合结果全部Tool per Agent每个子智能体一个工具适合少量固定模式一Single Dispatch一个调度工具 注册表适合大量动态模式二上下文隔离/注入默认隔离必要时经runtime.state注入历史模式三Subagents 的本质是“分而治之 集中调度”用主智能体保持全局视野和控制力用子智能体获得领域专精和上下文隔离。当你的 Agent 开始什么都想干、结果什么都干不好时就是该把它拆成一个主管 一队专家的时候了。