LangGraph：如何用“圖思維”輕松管理多Agent協(xié)作？

作者：張張 2025-04-01 08:48:34

LangGraph的圖思維將復雜任務分解為可編排的節(jié)點，通過動態(tài)路由和共享狀態(tài)實現(xiàn)高效協(xié)作。如果你正面臨多Agent的“spaghetti code”難題，不妨用LangGraph重構(gòu)你的流程！

引言

當AI任務變得復雜時，我們需要更好的“調(diào)度員”。

隨著智能應用場景的深化，單一 Agent 在處理復雜任務（如電商智能客服、金融數(shù)據(jù)分析流水線）時顯現(xiàn)出明顯局限性。傳統(tǒng)鏈式調(diào)用框架（如 LangChain）依賴開發(fā)者手動編排流程，在面對任務分支、動態(tài)決策和資源復用等場景時，往往陷入維護成本高、擴展性差的困境。

圖片

本文將深入解析：

1. LangGraph的核心原理：圖結(jié)構(gòu)如何模擬復雜任務流？

2. 多 Agent 應用實戰(zhàn)：如何實現(xiàn)異構(gòu)Agent的協(xié)作與通信？

3. 對比LangChain：何時該用LangGraph？

1、LangGraph核心原理

LangGraph 的核心原理，簡單說就是：用“圖”來重構(gòu)任務流。

1.1. 圖結(jié)構(gòu)（Graph）的三大要素

節(jié)點（Node）：代表一個獨立單元，可以是：

Agent 節(jié)點：封裝獨立 Agent 能力（如調(diào)用GPT-4處理問題、知識檢索）

Tool 節(jié)點：調(diào)用具體工具（如搜索API、數(shù)據(jù)庫查詢），用于執(zhí)行特定計算或處理任務

END 節(jié)點：流程結(jié)束的標識

邊（Edge）：標注 “Where to go”，表示狀態(tài)流轉(zhuǎn)的決策路徑，決定下一步跳轉(zhuǎn)到哪個節(jié)點
順序執(zhí)行（線性流程）
條件跳轉(zhuǎn)（如“若用戶投訴，轉(zhuǎn)人工審核”）
狀態(tài)（State）：貫穿整個流程，記錄數(shù)據(jù)或交互狀態(tài)（如用戶輸入的消息），驅(qū)動節(jié)點間的流轉(zhuǎn)

圖片

針對如上圖例，流程說明如下：

1）用戶輸入觸發(fā)初始狀態(tài)，進入 Agent 節(jié)點處理交互；

2）處理后通過 Edge 決策流向，可能跳轉(zhuǎn)至 Tool 節(jié)點調(diào)用工具，或直接到達 END 節(jié)點結(jié)束流程；

3）若進入 Tool 節(jié)點，處理完成后狀態(tài)可能再次流轉(zhuǎn)，形成循環(huán)或推進至結(jié)束。該模型通過節(jié)點、邊、狀態(tài)的協(xié)作，實現(xiàn)交互任務的流程化處理。

1.2 動態(tài)編排：像流程圖一樣設計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é)作實戰(zhàn)

為了更好讓大家理解，我們應用一個「客服工單處理案例」來進行介紹。

假設我們需要處理用戶工單，主要流程包括：意圖識別 → 專家Agent處理 → 結(jié)果審核

2.1 定義 Agent 節(jié)點

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è)技術支持")]}
# 人工審核
def human_review(state):
    return {"messages": [AIMessage(cnotallow="人工審核完成：您的問題已處理")]}
    
    
# 添加三個Agent節(jié)點  
workflow.add_node("intent_agent", identify_intent)  # 意圖識別  
workflow.add_node("expert_agent", call_expert)      # 專家處理  
workflow.add_node("review_agent", human_review)     # 人工審核

2.2 設計 workflow 邏輯

# 設置入口點
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")  
# 設置結(jié)束點
workflow.set_finish_point("review_agent")

2.3 運行 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é)果：

圖片