從ReAct到Multi-Agent:LangGraph如何實(shí)現(xiàn)智能體間的無縫協(xié)作? 原創(chuàng)
在LangChain生態(tài)日益成熟的今天,我們構(gòu)建的AI應(yīng)用正從簡單的“問答機(jī)器人”向復(fù)雜的“智能體協(xié)作系統(tǒng)”演進(jìn)。而LangGraph,作為LangChain生態(tài)中用于構(gòu)建有狀態(tài)、多步驟AI應(yīng)用的利器,其核心價(jià)值之一就在于對多智能體(Multi-Agent) 系統(tǒng)的優(yōu)雅支持。
為什么要“多智能體”?
- 智能體(Agent)是什么本質(zhì)是一個(gè)能感知環(huán)境、基于策略行動(dòng)以實(shí)現(xiàn)目標(biāo)的“自主體”。在 LangChain/LangGraph 中,哪怕最簡單的對話循環(huán)也可視為一個(gè)智能體。
a.具備感知(輸入)、決策(策略/推理)、行動(dòng)(工具/調(diào)用)、學(xué)習(xí)(記憶/更新)能力。
b.形式上可以是軟件角色、機(jī)器人、業(yè)務(wù)微服務(wù)、甚至一個(gè)“LLM + 工具”組合。
- 為什么會走向多智能體(Multi-agent)單體智能體在復(fù)雜問題上會遭遇三類瓶頸:
a.工具過載:工具太多導(dǎo)致“調(diào)用哪個(gè)”的決策困難。
b.上下文負(fù)擔(dān):長對話導(dǎo)致推理退化。
c.領(lǐng)域?qū)>珱_突:規(guī)劃、檢索、計(jì)算、執(zhí)行等專業(yè)能力難以在單一提示詞里兼顧。解決方法是“模塊化 + 專精化”:將系統(tǒng)拆分為多個(gè)職責(zé)單一的小智能體,組合成中/大型系統(tǒng)。
- 多智能體的核心收益
a.模塊化:開發(fā)、測試、維護(hù)成本更低。
b.專精化:專家智能體的魯棒性更強(qiáng)。
c.可控性:通信/交接路徑與策略可顯式定義,而不是完全交給 LLM 即興發(fā)揮。
五種典型架構(gòu)
- 網(wǎng)絡(luò)(Fully-connected Network)任意智能體都可與其他智能體通信。靈活但易“泛濫”,適合探索性、弱流程約束場景。
- 監(jiān)督者(Supervisor)引入一個(gè)“調(diào)度/路由”智能體,由它決定調(diào)用哪個(gè)專家。結(jié)構(gòu)清晰,利于審計(jì)與限流。
- 監(jiān)督者(工具調(diào)用變體)把專家抽象為“工具”,監(jiān)督者是 ReAct 智能體,通過工具調(diào)用來路由。利于快速落地。
- 層級(Hierarchical)多個(gè)團(tuán)隊(duì)各自有監(jiān)督者,頂層再由總監(jiān)督者統(tǒng)籌。適合大規(guī)模系統(tǒng)或多產(chǎn)品線場景。
- 自定義工作流(Deterministic/Dynamic Mix)部分邊是固定順序,部分由 LLM 通過?
?Command?? 動(dòng)態(tài)路由。工程中較常見的折中方案。
如果你不確定選哪種架構(gòu),先用“監(jiān)督者(工具調(diào)用)”起步,隨后按復(fù)雜性迭代。
Handoffs & Command
- 交接概念智能體執(zhí)行完當(dāng)前職責(zé)后,決定“結(jié)束/繼續(xù)/轉(zhuǎn)交給他人”。交接的關(guān)鍵是顯式描述:
a.目標(biāo)智能體:??goto??
b.攜帶負(fù)載(狀態(tài)更新):??update??
c.圖域(在哪個(gè)圖生效):??graph??,常見為當(dāng)前圖或 ??Command.PARENT??(從子圖跳回父圖)
- 最小 Command 模式
a.智能體節(jié)點(diǎn)函數(shù)返回 ??Command??,用于控制下一步路由;否則返回狀態(tài)更新結(jié)束當(dāng)前輪次。
b.在工具調(diào)用場景中,務(wù)必插入配對的“工具結(jié)果消息”,以滿足多數(shù) LLM 提供商的協(xié)議約束(每個(gè) ??ai_msg??? 的工具調(diào)用,必須跟一個(gè) ??tool?? 消息)。
- 常見交接手法
a.直接在智能體節(jié)點(diǎn)里決策,返回 ??Command??。
b.把交接包裝成一個(gè)“工具”,由 LLM 以工具調(diào)用的方式觸發(fā)交接(顯著提升統(tǒng)一性)。
c.子圖內(nèi)要交接到父圖的其他智能體時(shí),設(shè)置 ??graph=Command.PARENT??。
智能體通信與狀態(tài)設(shè)計(jì)
- 統(tǒng)一 state(共享消息)圖中的每一步都接收并產(chǎn)出?
?state??,通常包含??messages??。共享完整“草稿”(推理過程)能提升整體推理能力,但要提防上下文爆炸。 - 異構(gòu) state(私有草稿 + 共享摘要)各智能體維護(hù)自己格式的內(nèi)部狀態(tài),借助輸入/輸出轉(zhuǎn)換與父圖 state 對接。
a.優(yōu)點(diǎn):清晰邊界、可做“信息最小化共享”。
b.技巧:交接時(shí)只共享“上一條 AI 回復(fù) + 工具回執(zhí)”,而非全部草稿。
- 工具調(diào)用與負(fù)載監(jiān)督者作為 ReAct 節(jié)點(diǎn)時(shí),工具的參數(shù)就是負(fù)載。LangGraph 支持將父圖 state 注入到工具(例如?
?InjectedState??),實(shí)現(xiàn)“帶記憶的交接”。
三種Handoff模式對比與最小代碼片段
1. 直接 Command 交接(節(jié)點(diǎn)內(nèi)路由)
- 適用:兩個(gè)或少數(shù)幾個(gè)智能體的網(wǎng)絡(luò)架構(gòu),邏輯簡單、路由清晰。
- 關(guān)鍵點(diǎn):當(dāng)?
?ai_msg.tool_calls?? 非空,插入工具結(jié)果消息,再??goto?? 下一個(gè)智能體。
from typing_extensions import Literal
from langgraph.types import Command
from langgraph.graph import MessagesState
from langchain_core.tools import tool
@tool
def transfer_to_multiplication_expert():
"""向乘法智能體尋求幫助(只用于表明交接意圖)"""
return
def addition_expert(state: MessagesState) -> Command[Literal["multiplication_expert", "__end__"]]:
system_prompt = "你是加法專家。若需乘法,請先完成加法,再交接。"
messages = [{"role": "system", "content": system_prompt}] + state["messages"]
ai_msg = model.bind_tools([transfer_to_multiplication_expert]).invoke(messages)
if ai_msg.tool_calls:
tool_call_id = ai_msg.tool_calls[-1]["id"]
tool_msg = {"role": "tool", "content": "成功交接", "tool_call_id": tool_call_id}
return Command(goto="multiplication_expert", update={"messages": [ai_msg, tool_msg]})
return {"messages": [ai_msg]}- 常見坑
a.遺漏工具結(jié)果消息:導(dǎo)致提供商報(bào)錯(cuò)或上下文不同步。
b.無限交接:加步數(shù)上限或終止條件。
- 建議引入步數(shù)預(yù)算:
MAX_STEPS = 8
def guard_and_return(cmd_or_update, state):
steps = state.get("steps", 0) + 1
if steps > MAX_STEPS:
return {"messages": [{"role": "assistant", "content": "超出步數(shù)預(yù)算,結(jié)束。"}]}
if isinstance(cmd_or_update, Command):
cmd_or_update.update = {**cmd_or_update.update, "steps": steps}
return cmd_or_update
return {"messages": cmd_or_update["messages"], "steps": steps}2. 交接工具(handoff tool)
- 適用:每個(gè)智能體是一個(gè)“子圖”,通過工具將控制權(quán)交還父圖并路由到目標(biāo)智能體。
- 關(guān)鍵點(diǎn):在工具中返回?
?Command(goto=..., graph=Command.PARENT, update=...)??。
from typing import Annotated
from langchain_core.tools.base import InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langchain_core.tools import tool
from langgraph.types import Command
def make_handoff_tool(*, agent_name: str):
tool_name = f"transfer_to_{agent_name}"
@tool(tool_name)
def handoff_to_agent(
state: Annotated[dict, InjectedState],
tool_call_id: Annotated[str, InjectedToolCallId],
):
tool_message = {
"role": "tool",
"content": f"成功交接到 {agent_name}",
"name": tool_name,
"tool_call_id": tool_call_id,
}
return Command(
goto=agent_name,
graph=Command.PARENT,
update={"messages": state["messages"] + [tool_message]},
)
return handoff_to_agent- 優(yōu)點(diǎn):
a.路由邏輯統(tǒng)一沉淀在“工具層”,智能體代碼更干凈。
b.便于在團(tuán)隊(duì)內(nèi)部約定交接規(guī)范與審計(jì)。
3. 預(yù)構(gòu)建 ReAct 智能體(create_react_agent)
- 適用:不需要自定義 ToolNode/循環(huán)的場景,快速搭建。
- 要點(diǎn):把交接工具加入工具列表即可。
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
@tool
def add(a: int, b: int) -> int:
return a + b
@tool
def multiply(a: int, b: int) -> int:
return a * b
addition_expert = create_react_agent(
model,
[add, make_handoff_tool(agent_name="multiplication_expert")],
prompt="你是加法專家。必要時(shí)交接給乘法專家。",
)
multiplication_expert = create_react_agent(
model,
[multiply, make_handoff_tool(agent_name="addition_expert")],
prompt="你是乘法專家。必要時(shí)交接給加法專家。",
)函數(shù) API
當(dāng)你希望“每個(gè)智能體能與所有其他智能體通信”,但又想暫時(shí)避開圖的節(jié)點(diǎn)/邊定義時(shí),可以用 LangGraph 函數(shù) API:
- 用?
?@entrypoint()?? 定義主流程循環(huán); - 用?
?@task?? 封裝對各智能體的調(diào)用; - 通過識別最近一條?
?AIMessage.tool_calls?? 決定交接目標(biāo); - 在工具上用?
?@tool(return_direct=True)??,使得代理一旦調(diào)用交接工具就立刻“跳出”自身循環(huán),將控制權(quán)交回主流程。
工程上建議在主循環(huán)里加入:
- 步數(shù)上限、異常重試、交接工具白名單校驗(yàn)(防止 LLM“幻想”出未定義的工具名)。
工程化落地的關(guān)鍵要點(diǎn)
- 提示詞
a.“先做自己職責(zé),再交接”應(yīng)寫入 system prompt。
b.明確輸出格式、是否中文、是否需要結(jié)構(gòu)化信息(JSON/Markdown)。
- 終止條件與預(yù)算控制
a.設(shè)定步數(shù)預(yù)算、調(diào)用預(yù)算、時(shí)間預(yù)算;超限即 ??__end__??。
b.在狀態(tài)中追蹤 ??steps/cost??,并在每輪更新。
- 狀態(tài)設(shè)計(jì)與內(nèi)存治理
a.區(qū)分“共享消息”和“私有草稿”;用轉(zhuǎn)換器只暴露必要上下文。
b.引入“摘要工具/裁剪策略”(例如只保留近 N 輪 + 摘要)。
- 工具安全
a.工具入?yún)?qiáng)類型化(Pydantic 校驗(yàn));
b.重要操作前加“確認(rèn)工具”(雙人復(fù)核型);
c.對“交接工具名”設(shè)白名單,拒絕未知路由。
- 可觀測性與調(diào)試
a.使用 ??graph.stream(..., subgraphs=True)?? 訂閱事件;
b.統(tǒng)一 ??pretty_print_messages???,將 ??ai/tool?? 消息流對齊打??;
c.日志里打出 ??tool_call_id?? 對應(yīng)關(guān)系,便于問題復(fù)盤。
- 性能與并行
a.能“分而治之”的子任務(wù)用并行 map,再匯總 reduce;
b.對“冷工具”(慢 IO)啟用異步;
c.壓測不同架構(gòu)的延遲/費(fèi)用分布,形成“選型矩陣”。
- 穩(wěn)定性與測試
a.構(gòu)造“金樣本對話流”,回歸測試交接路徑與最終答案;
b.在異常熱點(diǎn)處增加 ??try/except?? + “安全降級”(例如改用本地規(guī)則/緩存)。
常見錯(cuò)誤與快速排查
- 少了工具結(jié)果消息:每個(gè)?
?ai_msg?? 的 tool call 后必須追加 1 條??tool?? 消息。 - 子圖交接沒指定 ?
??Command.PARENT???:導(dǎo)致交接“困在子圖”。 - ?
??update??? 寫錯(cuò)鍵名或結(jié)構(gòu):??{"messages": [...]}??不可缺。 - ?
??tool_call_id??? 不匹配:工具回執(zhí)里的 ID 必須對應(yīng)剛觸發(fā)的 call。 - 未設(shè) START 邊或錯(cuò)誤路由名:?
?builder.add_edge(START, "node")?? 別漏。 - 無限循環(huán):缺少步數(shù)控制,或提示詞沒強(qiáng)制“先做自己,再交接”。
總結(jié)
多智能體不是“把大模型套上更多提示詞”,而是“把復(fù)雜問題拆解成可治理的自治單元”,并通過可觀測、可控的交接協(xié)議把它們編排起來。Command 是編排的內(nèi)核;state 是溝通的載體;工具是行動(dòng)的接口;層次/網(wǎng)絡(luò)/監(jiān)督者/自定義工作流是架構(gòu)的選項(xiàng)。 真正跑代碼的時(shí)候,建議先用最簡單的模式跑通端到端,再引入交接工具統(tǒng)一路由,最后按規(guī)模演進(jìn)到層級或混合架構(gòu)。
本文轉(zhuǎn)載自???????AI 博物院??????? 作者:longyunfeigu

















