LangGraph 完整指南 2026:構建多智慧體協作系統

LangGraph 完整指南 2026:構建多智慧體協作系統

為什麼選擇 LangGraph?

在 2026 年的 AI 代理生態中,LangGraph 已成為構建複雜多智慧體系統的首選框架。作為 LangChain 生態的擴充套件,它引入了基於狀態圖的編排模型,讓開發者能夠精確控制智慧體的執行流程、條件分支和迴圈邏輯。

根據 Towards AI 的 2026 年框架評測,LangGraph 在可擴充套件性狀態管理生產就緒度三個維度均位列前三。相比 CrewAI 的簡單鏈式呼叫,LangGraph 提供了更細粒度的控制能力。

核心概念解析

狀態圖(State Graph)

LangGraph 的核心是狀態圖模型。每個智慧體執行都可以看作狀態轉換:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定義狀態結構
class AgentState(TypedDict):
    messages: list
    current_step: str
    results: Annotated[list, operator.add]
    metadata: dict

節點(Nodes)與邊(Edges)

節點代表執行單元,邊定義流程控制:

from langchain_core.messages import HumanMessage, AIMessage

def research_node(state: AgentState):
    """研究節點:執行網路搜尋"""
    query = state["messages"][-1].content
    # 呼叫搜尋工具
    results = search_web(query)
    return {"results": [results], "current_step": "research"}

def analysis_node(state: AgentState):
    """分析節點:處理研究結果"""
    context = state["results"][-1]
    analysis = llm.invoke(f"分析以下內容:{context}")
    return {"messages": [AIMessage(content=analysis)], "current_step": "analysis"}

# 構建圖
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)

# 定義邊
workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", END)

條件邊與迴圈

LangGraph 的強大之處在於支援條件分支迴圈執行

from langgraph.graph import ConditionalEdges

def should_continue(state: AgentState) -> str:
    """根據結果質量決定下一步"""
    if len(state["results"]) < 3:
        return "research_more"  # 需要更多研究
    return "finalize"  # 可以結束

# 新增條件邊
workflow.add_conditional_edges(
    "research",
    should_continue,
    {
        "research_more": "research",  # 迴圈回研究節點
        "finalize": "analysis"  # 進入分析
    }
)

這種模式非常適合需要迭代最佳化的場景,比如程式碼生成、內容創作或資料分析。

多智慧體協作實戰

下面是一個完整的多智慧體協作範例,包含研究員、分析師和稽核員三個角色:

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
import operator

class TeamState(TypedDict):
    task: str
    research_findings: Annotated[List[str], operator.add]
    analysis: str
    review_comments: List[str]
    final_output: str
    iteration_count: int

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 研究員智慧體
def researcher(state: TeamState):
    prompt = f"""作為研究員,請針對以下任務進行資訊蒐集:
    任務:{state['task']}
    當前發現:{state['research_findings']}
    
    請提供 3-5 個關鍵資訊點。"""
    
    response = llm.invoke(prompt)
    findings = response.content.split('\n')
    return {"research_findings": findings, "iteration_count": state['iteration_count'] + 1}

# 分析師智慧體
def analyst(state: TeamState):
    prompt = f"""作為分析師,請基於以下研究發現進行深度分析:
    {chr(10).join(state['research_findings'])}
    
    請提供結構化分析報告。"""
    
    response = llm.invoke(prompt)
    return {"analysis": response.content}

# 稽核員智慧體
def reviewer(state: TeamState):
    prompt = f"""作為稽核員,請評估以下分析的質量:
    {state['analysis']}
    
    如果質量達標,回覆"APPROVED"。否則列出需要改進的點。"""
    
    response = llm.invoke(prompt)
    if "APPROVED" in response.content:
        return {"final_output": state['analysis'], "review_comments": ["稽核透過"]}
    return {"review_comments": [response.content]}

# 決定是否繼續迭代
def should_iterate(state: TeamState) -> str:
    if state['iteration_count'] >= 3:
        return "finalize"
    if any("需要改進" in comment for comment in state['review_comments']):
        return "revise"
    return "finalize"

# 構建協作圖
team_workflow = StateGraph(TeamState)

team_workflow.add_node("researcher", researcher)
team_workflow.add_node("analyst", analyst)
team_workflow.add_node("reviewer", reviewer)

team_workflow.set_entry_point("researcher")
team_workflow.add_edge("researcher", "analyst")
team_workflow.add_edge("analyst", "reviewer")

team_workflow.add_conditional_edges(
    "reviewer",
    should_iterate,
    {
        "revise": "researcher",  # 返回重新研究
        "finalize": END
    }
)

app = team_workflow.compile()

# 執行
result = app.invoke({
    "task": "分析 2026 年 AI 代理框架的技術趨勢",
    "research_findings": [],
    "analysis": "",
    "review_comments": [],
    "final_output": "",
    "iteration_count": 0
})

print(result['final_output'])

記憶與檢查點

LangGraph 內建了檢查點系統,支援長對話記憶和斷點續跑:

from langgraph.checkpoint.memory import MemorySaver

# 啟用記憶儲存
memory = MemorySaver()
app = team_workflow.compile(checkpointer=memory)

# 使用執行緒 ID 保持對話歷史
config = {"configurable": {"thread_id": "conversation-123"}}

# 第一次呼叫
result1 = app.invoke({"task": "研究量子計算", ...}, config)

# 後續呼叫會保留之前的狀態
result2 = app.invoke({"task": "基於上述研究進行擴充套件", ...}, config)

生產環境最佳實踐

1. 錯誤處理與重試

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def safe_node_execution(state: AgentState):
    try:
        # 節點邏輯
        return result
    except Exception as e:
        return {"error": str(e), "retry_count": state.get('retry_count', 0) + 1}

2. 流式輸出

for event in app.stream(inputs, config, stream_mode="values"):
    for node, value in event.items():
        print(f"節點 {node} 輸出:{value}")

3. 超時控制

from langgraph.pregel import Pregel

app = team_workflow.compile(
    checkpointer=memory,
    interrupt_after=["reviewer"],  # 在稽核節點後暫停
)

# 設定超時
config = {"recursion_limit": 50}  # 最大遞迴深度

與其他框架對比

特性LangGraphCrewAIAutoGen
狀態管理✅ 完整狀態圖⚠️ 簡單鏈式⚠️ 會話式
條件分支✅ 原生支援❌ 有限⚠️ 需要自訂
迴圈執行✅ 原生支援❌ 不支援✅ 支援
記憶系統✅ 檢查點⚠️ 基礎✅ 完整
學習曲線中等
生產就緒✅ 高

實際應用場景

內容創作流水線

研究 → 大綱 → 初稿 → 稽核 → 修改 → 發佈

程式碼開發助手

需求分析 → 架構設計 → 程式碼生成 → 測試 → 程式碼審查 → 修復

資料分析管道

資料收集 → 清洗 → 分析 → 視覺化 → 報告生成 → 稽核

總結

LangGraph 為 2026 年的多智慧體系統提供了最靈活最可靠的編排方案。它的狀態圖模型讓複雜流程變得可預測、可除錯,而檢查點系統確保了長對話的連續性。

如果你正在構建需要多輪迭代條件分支多人協作的 AI 應用,LangGraph 是當前的最佳選擇。

參考資源


作者: Kevin Peng
發佈日期: 2026-03-31
分類: AI 助手 / 多智慧體系統
閱讀時間: 約 8 分鐘