LangGraph多Agent編排實戰：用Python構建生產級AI工作流的6個核心模式

AI与大数据

LangGraph多Agent編排：為什麼單Agent永遠不夠

單個AI Agent只能處理簡單任務，一旦涉及多步驟決策、條件分支、人工審批、狀態回滾，程式碼就變成一團義大利麵。LangGraph基於有向圖（DAG）模型，讓你用宣告式方式定義Agent之間的協作流程——每個節點是一個Agent或函式，每條邊是狀態傳遞路徑。2026年，LangGraph已支援StateGraph→條件路由→人機協作→檢查點持久化→子圖巢狀→串流輸出全鏈路能力。

本文將從6個核心模式出發，帶你完成圖定義→條件路由→人工審批→狀態持久化→子圖組合→串流輸出的全鏈路實戰。

核心概念

概念 說明
StateGraph 基於狀態的有向圖，節點處理狀態、邊傳遞狀態
Node 圖中的處理單元，接收狀態、傳回狀態更新
Edge 節點間的連線，分為普通邊和條件邊
Conditional Edge 根據狀態動態選擇下一個節點的邊
Checkpoint 狀態快照，支援暫停/恢復/回滾
Human-in-the-Loop 人工審批節點，暫停執行等待人工輸入
Subgraph 巢狀的子圖，支援模組化組合
Stream Mode 串流輸出模式，即時傳回中間結果

問題分析：多Agent編排解決的5類痛點

  1. 流程不可控：單Agent的ReAct迴圈無法表達複雜業務流程
  2. 狀態丟失：長對話中Agent無法記住之前的關鍵決策
  3. 人工審批缺失：關鍵操作無法暫停等待人工確認
  4. 錯誤不可恢復：執行失敗後無法回滾到上一個穩定狀態
  5. 協作混亂：多個Agent之間的職責邊界模糊、呼叫順序不可控

分步實操：6個LangGraph多Agent編排核心模式

模式1：StateGraph基礎與圖定義

pip install langgraph==0.4.0 langchain-openai==0.3.0

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_intent: str
    search_results: list[str]
    final_answer: str

def classify_intent(state: AgentState) -> dict:
    last_message = state["messages"][-1].content
    if any(kw in last_message for kw in ["搜尋", "查找", "查詢", "search"]):
        return {"user_intent": "search"}
    elif any(kw in last_message for kw in ["計算", "分析", "統計", "calculate"]):
        return {"user_intent": "calculate"}
    else:
        return {"user_intent": "chat"}

def search_agent(state: AgentState) -> dict:
    query = state["messages"][-1].content
    results = [f"搜尋結果1: {query}的相關資訊", f"搜尋結果2: {query}的深度分析"]
    return {"search_results": results}

def calculate_agent(state: AgentState) -> dict:
    return {"final_answer": "計算結果: 42"}

def chat_agent(state: AgentState) -> dict:
    return {"final_answer": "你好！我是AI助手，有什麼可以幫你？"}

def route_intent(state: AgentState) -> str:
    return state["user_intent"]

graph = StateGraph(AgentState)

graph.add_node("classify", classify_intent)
graph.add_node("search", search_agent)
graph.add_node("calculate", calculate_agent)
graph.add_node("chat", chat_agent)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
    "search": "search",
    "calculate": "calculate",
    "chat": "chat",
})
graph.add_edge("search", END)
graph.add_edge("calculate", END)
graph.add_edge("chat", END)

app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "幫我搜尋Python最新特性"}],
    "user_intent": "",
    "search_results": [],
    "final_answer": "",
})
print(result["search_results"])

模式2：條件路由與多輪對話

from typing import Literal

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    topic: str
    outline: str
    draft: str
    review_feedback: str
    revision_count: int

def researcher(state: ResearchState) -> dict:
    topic = state.get("topic", state["messages"][-1].content)
    outline = f"關於{topic}的研究大綱:\n1. 背景介紹\n2. 核心概念\n3. 實踐案例\n4. 總結展望"
    return {"topic": topic, "outline": outline}

def writer(state: ResearchState) -> dict:
    draft = f"基於大綱《{state['outline']}》撰寫的初稿..."
    return {"draft": draft}

def reviewer(state: ResearchState) -> dict:
    if state.get("revision_count", 0) >= 2:
        return {"review_feedback": "approved"}
    return {"review_feedback": "需要修改：增加更多實例和資料支撐"}

def should_revise(state: ResearchState) -> Literal["revise", "publish"]:
    if state.get("review_feedback") == "approved":
        return "publish"
    return "revise"

def reviser(state: ResearchState) -> dict:
    revised = f"修訂版（第{state.get('revision_count', 0) + 1}次修改）: {state['draft']}\n已增加實例和資料。"
    return {"draft": revised, "revision_count": state.get("revision_count", 0) + 1}

def publisher(state: ResearchState) -> dict:
    return {"final_answer": f"發布文章: {state['draft']}"}

research_graph = StateGraph(ResearchState)

research_graph.add_node("research", researcher)
research_graph.add_node("write", writer)
research_graph.add_node("review", reviewer)
research_graph.add_node("revise", reviser)
research_graph.add_node("publish", publisher)

research_graph.add_edge(START, "research")
research_graph.add_edge("research", "write")
research_graph.add_edge("write", "review")
research_graph.add_conditional_edges("review", should_revise, {
    "revise": "revise",
    "publish": "publish",
})
research_graph.add_edge("revise", "review")
research_graph.add_edge("publish", END)

research_app = research_graph.compile()

模式3：人機協作（Human-in-the-Loop）

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    messages: Annotated[list, add_messages]
    request: str
    risk_level: str
    approved: bool
    result: str

def risk_assessor(state: ApprovalState) -> dict:
    request = state["request"]
    if any(kw in request for kw in ["刪除", "重置", "清空", "delete"]):
        return {"risk_level": "high"}
    elif any(kw in request for kw in ["修改", "更新", "update"]):
        return {"risk_level": "medium"}
    return {"risk_level": "low"}

def human_approval(state: ApprovalState) -> dict:
    if state["risk_level"] == "high":
        decision = interrupt(f"高風險操作需要審批: {state['request']}")
        return {"approved": decision.get("approved", False)}
    return {"approved": True}

def executor(state: ApprovalState) -> dict:
    if state["approved"]:
        return {"result": f"執行成功: {state['request']}"}
    return {"result": f"操作被拒絕: {state['request']}"}

approval_graph = StateGraph(ApprovalState)
approval_graph.add_node("assess", risk_assessor)
approval_graph.add_node("approve", human_approval)
approval_graph.add_node("execute", executor)

approval_graph.add_edge(START, "assess")
approval_graph.add_edge("assess", "approve")
approval_graph.add_edge("approve", "execute")
approval_graph.add_edge("execute", END)

checkpointer = MemorySaver()
approval_app = approval_graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-001"}}
result = approval_app.invoke(
    {"request": "刪除生產資料庫的所有測試資料", "messages": [], "risk_level": "", "approved": False, "result": ""},
    config=config,
)

for state in approval_app.get_state_history(config):
    if state.next:
        approval_app.invoke(
            Command(resume={"approved": True}),
            config=config,
        )
        break

模式4：檢查點持久化與狀態恢復

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

class LongTaskState(TypedDict):
    messages: Annotated[list, add_messages]
    task_id: str
    steps_completed: list[str]
    current_step: str
    error: str

def step_one(state: LongTaskState) -> dict:
    return {"current_step": "step_one", "steps_completed": state.get("steps_completed", []) + ["step_one"]}

def step_two(state: LongTaskState) -> dict:
    return {"current_step": "step_two", "steps_completed": state.get("steps_completed", []) + ["step_two"]}

def step_three(state: LongTaskState) -> dict:
    return {"current_step": "step_three", "steps_completed": state.get("steps_completed", []) + ["step_three"]}

long_task_graph = StateGraph(LongTaskState)
long_task_graph.add_node("one", step_one)
long_task_graph.add_node("two", step_two)
long_task_graph.add_node("three", step_three)

long_task_graph.add_edge(START, "one")
long_task_graph.add_edge("one", "two")
long_task_graph.add_edge("two", "three")
long_task_graph.add_edge("three", END)

conn = sqlite3.connect(":memory:", check_same_thread=False)
sqlite_checkpointer = SqliteSaver(conn)
long_task_app = long_task_graph.compile(checkpointer=sqlite_checkpointer)

task_config = {"configurable": {"thread_id": "long-task-001"}}
long_task_app.invoke({"task_id": "task-1", "messages": [], "steps_completed": [], "current_step": "", "error": ""}, config=task_config)

state_snapshot = long_task_app.get_state(task_config)
print(f"已完成步驟: {state_snapshot.values.get('steps_completed', [])}")

for history_state in long_task_app.get_state_history(task_config):
    print(f"步驟: {history_state.values.get('current_step', 'N/A')}, 時間: {history_state.created_at}")

模式5：子圖巢狀與模組化組合

class CodeReviewState(TypedDict):
    messages: Annotated[list, add_messages]
    code: str
    lint_result: str
    security_result: str
    style_result: str
    final_review: str

def linter(state: CodeReviewState) -> dict:
    return {"lint_result": f"Lint檢查通過: {state['code'][:50]}..."}

def security_scanner(state: CodeReviewState) -> dict:
    return {"security_result": "安全掃描: 未發現漏洞"}

def style_checker(state: CodeReviewState) -> dict:
    return {"style_result": "程式碼風格: 符合PEP8規範"}

review_subgraph = StateGraph(CodeReviewState)
review_subgraph.add_node("lint", linter)
review_subgraph.add_node("security", security_scanner)
review_subgraph.add_node("style", style_checker)

review_subgraph.add_edge(START, "lint")
review_subgraph.add_edge(START, "security")
review_subgraph.add_edge(START, "style")
review_subgraph.add_edge("lint", END)
review_subgraph.add_edge("security", END)
review_subgraph.add_edge("style", END)

review_sub_app = review_subgraph.compile()

class DevPipelineState(TypedDict):
    messages: Annotated[list, add_messages]
    code: str
    review_result: str
    test_result: str
    deploy_result: str

def code_review_node(state: DevPipelineState) -> dict:
    review_output = review_sub_app.invoke({
        "code": state["code"],
        "messages": [],
        "lint_result": "",
        "security_result": "",
        "style_result": "",
        "final_review": "",
    })
    return {"review_result": f"Lint: {review_output['lint_result']} | Security: {review_output['security_result']} | Style: {review_output['style_result']}"}

def test_runner(state: DevPipelineState) -> dict:
    return {"test_result": "所有測試通過 (42/42)"}

def deployer(state: DevPipelineState) -> dict:
    return {"deploy_result": "部署成功: v1.0.0 已上線"}

pipeline_graph = StateGraph(DevPipelineState)
pipeline_graph.add_node("review", code_review_node)
pipeline_graph.add_node("test", test_runner)
pipeline_graph.add_node("deploy", deployer)

pipeline_graph.add_edge(START, "review")
pipeline_graph.add_edge("review", "test")
pipeline_graph.add_edge("test", "deploy")
pipeline_graph.add_edge("deploy", END)

pipeline_app = pipeline_graph.compile()

模式6：串流輸出與即時回饋

class StreamChatState(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    thinking: str
    response: str

def thinker(state: StreamChatState) -> dict:
    return {"thinking": f"正在分析: {state['query']}"}

def responder(state: StreamChatState) -> dict:
    return {"response": f"關於'{state['query']}'的詳細回答: LangGraph支援串流輸出，可以即時傳回中間結果..."}

stream_graph = StateGraph(StreamChatState)
stream_graph.add_node("think", thinker)
stream_graph.add_node("respond", responder)

stream_graph.add_edge(START, "think")
stream_graph.add_edge("think", "respond")
stream_graph.add_edge("respond", END)

stream_app = stream_graph.compile()

for event in stream_app.stream({"query": "LangGraph怎麼用？", "messages": [], "thinking": "", "response": ""}):
    for node_name, node_output in event.items():
        print(f"[{node_name}] {node_output}")

避坑指南

坑1：狀態Schema定義不完整

# ❌ 錯誤：缺少欄位預設值
class BadState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str  # 呼叫時必須提供，否則KeyError

# ✅ 正確：使用Optional或提供預設值
from typing import Optional

class GoodState(TypedDict):
    messages: Annotated[list, add_messages]
    result: str
    metadata: Optional[dict]  # 可選欄位

坑2：條件路由傳回值不匹配

# ❌ 錯誤：路由函式傳回的值不在映射表中
def bad_router(state) -> str:
    return "unknown_node"  # 映射表中沒有這個key

# ✅ 正確：確保所有可能的傳回值都在映射表中
def good_router(state) -> Literal["search", "calculate", "chat"]:
    if "搜尋" in state["messages"][-1].content:
        return "search"
    elif "計算" in state["messages"][-1].content:
        return "calculate"
    return "chat"

坑3：忘記新增Checkpointer導致狀態丟失

# ❌ 錯誤：不使用checkpointer，中斷後無法恢復
app = graph.compile()  # 無狀態

# ✅ 正確：使用checkpointer持久化狀態
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

坑4：interrupt使用不當

# ❌ 錯誤：在非checkpointer圖中使用interrupt
app = graph.compile()  # 無checkpointer
# interrupt()會拋出異常

# ✅ 正確：必須搭配checkpointer使用
app = graph.compile(checkpointer=MemorySaver())

坑5：子圖狀態與父圖狀態不匹配

# ❌ 錯誤：子圖狀態欄位與父圖完全不同
class ParentState(TypedDict):
    code: str
class ChildState(TypedDict):
    data: bytes  # 欄位不相容

# ✅ 正確：子圖狀態是父圖狀態的子集或相容擴充
class ChildState(TypedDict):
    code: str  # 與父圖欄位對應
    extra_field: str

報錯排查

序號 報錯資訊 原因 解決方法
1 KeyError: 'field_name' 狀態Schema缺少欄位 確保invoke時提供所有必填欄位
2 InvalidEdgeError 邊指向不存在的節點 檢查add_edge的節點名是否已add_node
3 GraphRecursionError 圖中存在無限迴圈 新增最大迭代次數或終止條件
4 interrupt() called without checkpointer 未配置checkpointer 編譯時傳入checkpointer參數
5 StateUpdateError: incompatible types 狀態更新型別不匹配 檢查節點傳回的dict鍵值型別
6 NodeNotFoundError 條件路由傳回未註冊的節點 確保路由傳回值在映射表中
7 CheckpointError: thread_id required 未提供thread_id invoke時傳入configurable.thread_id
8 SubgraphStateError 子圖與父圖狀態不相容 統一狀態Schema或使用轉換層
9 StreamTimeoutError 串流輸出逾時 檢查節點是否有死迴圈或長時間阻塞
10 SerializationError 狀態包含不可序列化物件 確保狀態中只有JSON可序列化型別

進階最佳化

  1. 自定義Reducer：使用Annotated[type, custom_reducer]實作複雜的狀態合併邏輯
  2. 並行節點：使用add_edge(START, ["node_a", "node_b"])實作節點並行執行
  3. 圖視覺化：使用app.get_graph().draw_mermaid()生成Mermaid流程圖
  4. 生產級Checkpointer：使用PostgreSQL或Redis替代MemorySaver
  5. 非同步執行：使用AsyncStateGraph和ainvoke實作非同步Agent

對比分析

維度 LangGraph CrewAI AutoGen LangChain Agent
圖編排能力 ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐ ⭐⭐
狀態持久化 ⭐⭐⭐⭐⭐ ⭐⭐ ⭐⭐⭐ ⭐⭐
人機協作 ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐
條件路由 ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐ ⭐
子圖巢狀 ⭐⭐⭐⭐⭐ ⭐⭐ ⭐ ⭐
學習曲線 ⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐⭐

總結：LangGraph讓你從「命令式Agent呼叫」進化為「宣告式圖編排」。StateGraph→條件路由→人機協作→檢查點持久化→子圖巢狀→串流輸出六位一體，是2026年Python多Agent編排的首選框架。核心原則：狀態即資料流、節點即處理單元、邊即控制流。

線上工具推薦

本站提供瀏覽器本地工具，免註冊即可試用 →

#LangGraph#AI Agent#多Agent编排#Python AI#2026#AI与大数据