從ReAct到Multi-Agent：LangGraph如何實(shí)現(xiàn)智能體間的無縫協(xié)作？原創(chuàng)

發(fā)布于 2025-9-18 10:18

瀏覽

0收藏

在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 + 工具”組合。

為什么會(huì)走向多智能體（Multi-agent）單體智能體在復(fù)雜問題上會(huì)遭遇三類瓶頸：

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)用來路由。利于快速落地。
層級(jí)（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)用；
通過識(shí)別最近一條??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?? + “安全降級(jí)”（例如改用本地規(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)到層級(jí)或混合架構(gòu)。

本文轉(zhuǎn)載自???????AI 博物院??????? 作者：longyunfeigu

?著作權(quán)歸作者所有，如需轉(zhuǎn)載，請注明出處，否則將追究法律責(zé)任

標(biāo)簽

贊

回復(fù)