LangGraph:如何用“圖思維”輕松管理多Agent協(xié)作?
引言
當(dāng)AI任務(wù)變得復(fù)雜時,我們需要更好的“調(diào)度員”。
隨著智能應(yīng)用場景的深化,單一 Agent 在處理復(fù)雜任務(wù)(如電商智能客服、金融數(shù)據(jù)分析流水線)時顯現(xiàn)出明顯局限性。傳統(tǒng)鏈?zhǔn)秸{(diào)用框架(如 LangChain)依賴開發(fā)者手動編排流程,在面對任務(wù)分支、動態(tài)決策和資源復(fù)用等場景時,往往陷入維護(hù)成本高、擴(kuò)展性差的困境。
圖片
本文將深入解析:
1. LangGraph的核心原理:圖結(jié)構(gòu)如何模擬復(fù)雜任務(wù)流?
2. 多 Agent 應(yīng)用實(shí)戰(zhàn):如何實(shí)現(xiàn)異構(gòu)Agent的協(xié)作與通信?
3. 對比LangChain:何時該用LangGraph?
1、LangGraph核心原理
LangGraph 的核心原理,簡單說就是:用“圖”來重構(gòu)任務(wù)流。
1.1. 圖結(jié)構(gòu)(Graph)的三大要素
- 節(jié)點(diǎn)(Node):代表一個獨(dú)立單元,可以是:
Agent 節(jié)點(diǎn):封裝獨(dú)立 Agent 能力(如調(diào)用GPT-4處理問題、知識檢索)
Tool 節(jié)點(diǎn):調(diào)用具體工具(如搜索API、數(shù)據(jù)庫查詢),用于執(zhí)行特定計(jì)算或處理任務(wù)
END 節(jié)點(diǎn):流程結(jié)束的標(biāo)識
- 邊(Edge):標(biāo)注 “Where to go”,表示狀態(tài)流轉(zhuǎn)的決策路徑,決定下一步跳轉(zhuǎn)到哪個節(jié)點(diǎn)
- 順序執(zhí)行(線性流程)
- 條件跳轉(zhuǎn)(如“若用戶投訴,轉(zhuǎn)人工審核”)
- 狀態(tài)(State):貫穿整個流程,記錄數(shù)據(jù)或交互狀態(tài)(如用戶輸入的消息),驅(qū)動節(jié)點(diǎn)間的流轉(zhuǎn)
圖片
針對如上圖例,流程說明如下:
1)用戶輸入觸發(fā)初始狀態(tài),進(jìn)入 Agent 節(jié)點(diǎn) 處理交互;
2)處理后通過 Edge 決策流向,可能跳轉(zhuǎn)至 Tool 節(jié)點(diǎn) 調(diào)用工具,或直接到達(dá) END 節(jié)點(diǎn) 結(jié)束流程;
3)若進(jìn)入 Tool 節(jié)點(diǎn),處理完成后狀態(tài)可能再次流轉(zhuǎn),形成循環(huán)或推進(jìn)至結(jié)束。 該模型通過節(jié)點(diǎn)、邊、狀態(tài)的協(xié)作,實(shí)現(xiàn)交互任務(wù)的流程化處理。
1.2 動態(tài)編排:像流程圖一樣設(shè)計(jì)AI流程
LangGraph支持循環(huán)、分支、并行等控制流,例如:
# 示例:根據(jù)用戶問題動態(tài)路由 def route(state): if "投訴" in state["input"]: return "human_review" # 轉(zhuǎn)人工 else: return "ai_agent" # 由AI處理
# 示例:根據(jù)用戶問題動態(tài)路由
def route(state):
if "投訴" in state["input"]:
return "human_review" # 轉(zhuǎn)人工
else:
return "ai_agent" # 由AI處理
2、多 Agent 協(xié)作實(shí)戰(zhàn)
為了更好讓大家理解,我們應(yīng)用一個「客服工單處理案例」來進(jìn)行介紹。
假設(shè)我們需要處理用戶工單,主要流程包括:意圖識別 → 專家Agent處理 → 結(jié)果審核
2.1 定義 Agent 節(jié)點(diǎn)
from langgraph.graph import StateGraph, MessagesState
# Create the workflow
workflow = StateGraph(MessagesState)
# 意圖識別
def identify_intent(state):
last_message = state["messages"][-1].content
if "賬號" in last_message or "登錄" in last_message:
intent = "technical"
else:
intent = "general"
return {
"messages": [AIMessage(cnotallow="意圖識別完成")],
"intent": intent
}
# 專家處理
def call_expert(state):
return {"messages": [AIMessage(cnotallow="專家處理結(jié)果:您的問題需要專業(yè)技術(shù)支持")]}
# 人工審核
def human_review(state):
return {"messages": [AIMessage(cnotallow="人工審核完成:您的問題已處理")]}
# 添加三個Agent節(jié)點(diǎn)
workflow.add_node("intent_agent", identify_intent) # 意圖識別
workflow.add_node("expert_agent", call_expert) # 專家處理
workflow.add_node("review_agent", human_review) # 人工審核
2.2 設(shè)計(jì) workflow 邏輯
# 設(shè)置入口點(diǎn)
workflow.set_entry_point("intent_agent")
# 條件邊:根據(jù)意圖選擇路徑
def router(state):
if state["intent"] == "technical":
return "expert_agent"
else:
return "review_agent"
workflow.add_conditional_edges("intent_agent", router)
# 專家處理后必須審核
workflow.add_edge("expert_agent", "review_agent")
# 設(shè)置結(jié)束點(diǎn)
workflow.set_finish_point("review_agent")
2.3 運(yùn)行 workflow
# Compile the workflow
app = workflow.compile()
output = app.invoke({"messages": [HumanMessage(cnotallow="我的賬號無法登錄")]})
print("執(zhí)行結(jié)果:\n\n", final_state["messages"][-1].content)
執(zhí)行結(jié)果:
圖片
程序執(zhí)行流程為:用戶輸入 → 意圖識別 → 技術(shù)問題 → 專家Agent → 人工審核 → 返回結(jié)果
LangGraph 圖例如下:
圖片
3、LangChain vs LangGraph
當(dāng)LangChain 遇上 LangGraph,我們應(yīng)該如何選擇呢?
圖片
簡單下個結(jié)論:
- 如果是固定流程(如“搜索→總結(jié)”),用LangChain更輕量。
- 如果是動態(tài)、多角色協(xié)作(如客服、數(shù)據(jù)分析流水線),LangGraph是更優(yōu)解。
結(jié)語
LangGraph的圖思維將復(fù)雜任務(wù)分解為可編排的節(jié)點(diǎn),通過動態(tài)路由和共享狀態(tài)實(shí)現(xiàn)高效協(xié)作。如果你正面臨多Agent的“spaghetti code”難題,不妨用LangGraph重構(gòu)你的流程!
圖片