文章目录
- LangChain 多工具协作:一个 Agent 不够用时怎么办
- 一、先理解问题根源:工具越多越不准
- 二、解法一:工具分组 + 路由(先试这个)
- 2.1 用 RunnableBranch 做路由
- 2.2 更省的做法:用关键词/规则先兜一层
- 三、解法二:多 Agent 分工(Supervisor 模式)
- 3.1 手写 Supervisor
- 3.2 官方的 langgraph-supervisor
- 四、两种方案怎么选
- 五、多 Agent 的三个坑
- 5.1 消息爆炸
- 5.2 无限来回
- 5.3 子 Agent 抢活
- 六、小结
LangChain 多工具协作:一个 Agent 不够用时怎么办
给一个 Agent 挂十几个工具,很快就会发现它开始乱选:
该查天气的去查了数据库,该下单的去搜了网页。这篇讲两种解法——工具分组路由和多 Agent 分工,
以及什么时候该用哪个。
一、先理解问题根源:工具越多越不准
这不是玄学,有三个实际原因:
- schema 占用上下文:每个工具的名称、描述、参数都要送进 prompt,
20 个工具能吃掉几千 token,挤压真正的任务信息; - 选择空间变大:模型要在 20 个选项里挑一个,
描述稍微有点重叠(比如search_doc和search_wiki)就会选错; - 注意力稀释:长 schema 列表里,排在中间的工具容易被忽略。
经验阈值:单个 Agent 的工具数控制在 10 个以内,超过就要拆分。
二、解法一:工具分组 + 路由(先试这个)
思路:不把所有工具都给一个 Agent,而是先判断任务类型,
再把对应的一组工具注入进去。
用户提问 ↓ 路由(一次轻量 LLM 调用,只做分类) ↓ ┌────────┬────────┬────────┐ 订单组 物流组 售后组 (3工具) (2工具) (4工具) ↓ 对应 Agent 执行2.1 用 RunnableBranch 做路由
fromlangchain_core.runnablesimportRunnableBranch,RunnableLambdafromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser route_prompt=ChatPromptTemplate.from_template(""" 判断下面的用户问题属于哪一类,只输出一个词:order / logistics / service / other 问题:{question} """)router=route_prompt|llm|StrOutputParser()defbuild_agent(category:str):tool_map={"order":[search_order,list_orders,cancel_order],"logistics":[query_logistics,change_address],"service":[create_ticket,refund,send_email],}tools=tool_map.get(category,[])ifnottools:returnNonereturncreate_react_agent(llm,tools)defhandle(question:str):category=router.invoke({"question":question}).strip()agent=build_agent(category)ifagentisNone:returngeneral_chain.invoke(question)returnagent.invoke({"messages":[("user",question)]})路由本身只是一次极轻的 LLM 调用(输出一个词),
代价远小于"给 20 个工具让模型慢慢挑"。
2.2 更省的做法:用关键词/规则先兜一层
能不调 LLM 就别调:
importre RULES=[(r"物流|快递|发货|签收|单号","logistics"),(r"退款|退货|投诉|发票","service"),(r"订单|下单|购买|支付","order"),]defcheap_route(question:str):forpattern,catinRULES:ifre.search(pattern,question):returncatreturnNonedefhandle(question:str):cat=cheap_route(question)orrouter.invoke({"question":question}).strip()...规则命中率高的话,能省掉大部分路由调用。
规则没命中再降级到 LLM 分类。这是典型的"快慢结合"。
三、解法二:多 Agent 分工(Supervisor 模式)
当任务真的复杂到需要多个角色各自推理、再汇总时,才上多 Agent。
典型结构:
┌────────────┐ │ supervisor │ ← 负责分派和汇总 └──────┬─────┘ ┌──────────┼──────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │research │ │ code │ │ writer │ │ Agent │ │ Agent │ │ Agent │ └─────────┘ └─────────┘ └─────────┘- supervisor:不干活,只判断"下一步交给谁"和"是不是可以结束了";
- 子 Agent:各自有专门的工具和 prompt,只负责自己那一摊。
3.1 手写 Supervisor
fromtypingimportLiteralfrompydanticimportBaseModelfromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlanggraph.prebuiltimportcreate_react_agentfromlanggraph.typesimportCommand# --- 子 Agent ---research_agent=create_react_agent(llm,[web_search,read_page],prompt="你负责搜集资料,只做检索,不写结论。")code_agent=create_react_agent(llm,[run_python,read_file],prompt="你负责写代码并运行验证,只做计算,不写文章。")# --- Supervisor 的决策 schema ---classRoute(BaseModel):next:Literal["research","coder","FINISH"]supervisor_llm=llm.with_structured_output(Route)defsupervisor(state:MessagesState):resp=supervisor_llm.invoke([("system","你是团队主管。根据对话进展,决定下一步交给谁。""research=需要查资料,coder=需要算数据,FINISH=信息够了可以结束。"),*state["messages"],])return{"next":resp.next}defcall_research(state:MessagesState):result=research_agent.invoke({"messages":state["messages"]})return{"messages":result["messages"][-1:]}defcall_coder(state:MessagesState):result=code_agent.invoke({"messages":state["messages"]})return{"messages":result["messages"][-1:]}# --- 组装图 ---graph=StateGraph(MessagesState)graph.add_node("supervisor",supervisor)graph.add_node("research",call_research)graph.add_node("coder",call_coder)graph.add_edge(START,"supervisor")graph.add_conditional_edges("supervisor",lambdas:s["next"],{"research":"research","coder":"coder","FINISH":END,})graph.add_edge("research","supervisor")graph.add_edge("coder","supervisor")team=graph.compile()要点:
- 子 Agent 的返回只取最后一条消息塞回主 state,
否则主对话会被子 Agent 的中间过程撑爆; - supervisor 用
with_structured_output保证输出可控; - 每个子 Agent 的 prompt 要明确边界(“只做检索,不写结论”),
否则它们会互相抢活。
3.2 官方的 langgraph-supervisor
不想手写可以用官方包:
pipinstalllanggraph-supervisorfromlanggraph_supervisorimportcreate_supervisor workflow=create_supervisor([research_agent,code_agent,writer_agent],model=llm,prompt="你是一个三人团队的主管,合理分派任务,最后汇总成一个完整报告。",)app=workflow.compile()它帮你处理了消息传递、路由、终止条件这些样板代码。
但建议先手写一遍理解原理,再用官方包,不然出了问题不知道怎么调。
四、两种方案怎么选
| 维度 | 工具分组路由 | 多 Agent 分工 |
|---|---|---|
| 复杂度 | 低 | 高 |
| LLM 调用次数 | 少(1 次路由 + 1 个 Agent) | 多(每轮 supervisor + 子 Agent) |
| 延迟 | 低 | 高 |
| 适用场景 | 任务一次就能归到某一类 | 任务需要多角色接力 |
| 调试难度 | 低 | 高 |
决策依据:任务是否需要"接力"。
- “帮我查一下这个订单的物流” → 一类工具搞定 →分组路由;
- “调研一下竞品,写份分析报告” → 要先检索、再分析、再写作 →多 Agent。
能用分组路由解决的,别上多 Agent。
多 Agent 的成本是数倍增长的,而且调试非常折磨人。
五、多 Agent 的三个坑
5.1 消息爆炸
子 Agent 跑 5 步就有 10 条消息,全塞回主 state,
几轮下来上下文就满了。
解法:只回传结论。
defcall_research(state:MessagesState):result=research_agent.invoke({"messages":state["messages"]})# 只取最终那条 AI 消息,并重标记来源final=result["messages"][-1]return{"messages":[AIMessage(content=f"[调研结果]{final.content}",name="research")]}用name字段标记来源,supervisor 能分得清是谁说的。
5.2 无限来回
supervisor 可能让 research 和 coder 反复横跳。
解法:限制轮数。
classState(MessagesState):rounds:intdefsupervisor(state:State):ifstate.get("rounds",0)>=6:return{"next":"FINISH"}...return{"next":resp.next,"rounds":state.get("rounds",0)+1}5.3 子 Agent 抢活
没有明确边界时,research 也会写结论,writer 又去检索一遍。
解法:prompt 里写死"不做什么",比写"要做什么"更重要:
你是资料检索员。 只做:使用搜索工具收集信息,返回原文要点。 不做:不要分析、不要下结论、不要撰写报告、不要调用其他角色的工具。六、小结
- 工具超过10 个就要拆,否则模型选择准确率明显下降;
- 先试工具分组路由:一次轻量分类 → 注入对应工具组,代价小效果好;
- 路由可以规则优先、LLM 兜底,省调用;
- 多 Agent 只在任务需要多角色接力时才上;
- Supervisor 模式:子 Agent 只回传结论(用
name标记来源)、
限制轮数、prompt 里写死"不做什么"; - 官方
langgraph-supervisor能省样板代码,但建议先手写一遍理解原理。
下一篇讲 LangGraph 的 State 设计——多节点之间怎么安全共享数据。