LangGraph 完整實戰指南：用 StateGraph + Checkpointer 打造可 rollback 的 production Multi-Agent Pipeline（含 Claude Agent SDK 整合）

Klarna、Uber、J.P. Morgan、LinkedIn。這四家公司有什麼共通點？他們的 production-grade multi-agent system 都跑在 LangGraph 上。

這不是巧合。LangChain 官方資料點名 LangGraph 是「為長時運行、stateful agent 設計的 low-level orchestration framework」。同期 Pockit 與 Composio 等多份 2026 上半年的橫向比較顯示，LangGraph 在 GitHub stars 已經超車 CrewAI、Microsoft 把 AutoGen 轉維護模式之後，graph-based 架構成了 production multi-agent 的事實標準。

但你看完官方文件可能還是搞不懂三件事：State 到底要怎麼設計、Checkpointer 為什麼是 production 必要、為什麼大家說 LangGraph 比直接寫 if-else 工作流還清爽？

這篇是寫給工程師跟技術主管的深度實戰。讀完你會知道：LangGraph 的四個核心概念（State / Node / Edge / Checkpointer）怎麼設計、Conditional Edge 為什麼要寫成 code 不要塞進 prompt、Time Travel debugging 怎麼救你的 production、Human-in-the-Loop 用 interrupt() 怎麼接，以及——怎麼把 Claude Agent SDK 包成 LangGraph node，拿到兩邊的好處。

為什麼 production multi-agent pipeline 都選了 LangGraph

先說結論：LangGraph 解決的是「Agent 跑了一半當機怎麼辦」「Agent 跑了一半要人類接手怎麼辦」「Agent 跑出怪結果要倒帶 debug 怎麼辦」這三個 production 級工程問題。

Cursor、Copilot 是 IDE 外掛，CrewAI 是 role-based crew DSL，AutoGen 是會議室式對話 — 都不錯，但都沒原生處理上面那三個問題。LangGraph 透過 stateful graph + checkpointer 一次解掉。

看實際的市場接受度也能驗證。DataCamp 2026 年的 framework 比較指出，LangGraph 同一個任務只吃 2000 tokens，CrewAI 要 3500，AutoGen 高達 8000。差距來自 graph 結構讓「對話」變成「狀態轉換」，省掉 Agent 之間互相轉述背景的浪費。

如果你正在從零選型、目標是 production-grade pipeline，LangGraph 是最低風險的起點。如果你已經在 CrewAI 或 AutoGen 上跑得穩，本篇的概念依然適用 — checkpointing、time travel、human-in-the-loop 是 跨框架的工程模式，LangGraph 只是把它們做得最完整。

想看 4 個 Agent 框架的橫向 5 維度決策框架，可以搭配閱讀 AI Agent 框架選型完整指南；想看 Claude Agent SDK 端到端 pipeline 怎麼設計，可以看 Claude Agent SDK 端到端自動化開發完整實戰。

LangGraph 核心四件套：State、Node、Edge、Checkpointer

LangGraph 的全部複雜度可以歸結到四個概念。先把這四個搞清楚，後面所有 advanced topic 都是它們的延伸。

概念一：State（狀態）

State 是整個 graph 共享的記憶體。它是個 TypedDict 或 Pydantic BaseModel，定義所有 node 之間要傳遞的資料。每個 node 收到當前 state、做事、回傳 state 的更新。LangGraph 用 reducer 把更新 merge 進去。

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

class PipelineState(TypedDict):
    # 單值欄位：每次 node 回傳會覆蓋
    feature_name: str
    spec_path: str
    current_stage: str

    # List 欄位用 reducer 累加（Annotated 第二個參數）
    log_entries: Annotated[list[str], add]
    test_results: Annotated[list[dict], add]

重點是 Annotated[list[str], add] 那行。它告訴 LangGraph：當 node 回傳 `{"log_entries": ["new entry"]}`，請把它「加進」既有 list，而不是覆蓋。沒有 reducer 的欄位就是覆蓋行為。這個機制讓 state 既能傳大型累積資料、又能控制何時 reset。

概念二：Node（節點）

Node 是「做一件事」的函式。input 是 state，output 是 state 的部分更新。

def spec_agent(state: PipelineState) -> dict:
    # 讀 spec 檔、呼叫 LLM、產生 acceptance criteria
    spec_content = Path(state["spec_path"]).read_text()
    response = llm.invoke(f"Convert this PRD to BDD AC:\n{spec_content}")
    return {
        "current_stage": "spec_done",
        "log_entries": [f"spec_agent: produced {len(response.content)} chars"],
    }

Node 不一定要呼叫 LLM。它可以是純 Python 程式碼（讀檔、跑測試、發 webhook），也可以是呼叫外部 service、甚至是另一個 LangGraph subgraph。「Node 不等於 Agent」是新手最常見的誤解 — node 是一個 unit of work，agent 只是其中一種。

概念三：Edge（邊）

Edge 決定 node 之間怎麼接。最簡單的是直線：A → B → C。但實戰幾乎都需要分支：「測試過了走 review、沒過走 retry」「review 同意走 deploy、駁回走 fix」。LangGraph 用 Conditional Edge 處理這個 — 一個 function 讀 state、回傳下一個 node 名字。後面有專節談。

概念四：Checkpointer（狀態持久化）

這是 LangGraph 最重要、最容易被低估的概念。LangChain 官方文件講得很直接：「checkpointer 是儲存後端，每個 node 執行後都會 save state — 不只是最後一次。」

意思是：你的 graph 跑到第 7 個 node 當機了，下次重啟可以從第 7 個繼續，前 6 個的狀態都還在。你想看 node 5 跑完當下的 state 長怎樣？讀 checkpoint。你想從 node 4 重跑一條不同的分支？讀 node 4 的 checkpoint，改 state，從那繼續。這就是 Time Travel debugging，下一節會深入。

State Schema 設計：把錯誤的選型代價講清楚

State 設計是 LangGraph 工程上最容易翻車的地方。看起來只是定義 TypedDict，但「該不該用 reducer」「Pydantic vs TypedDict」「巢狀結構放哪」這些決策會直接影響你的 pipeline 能不能 scale。

TypedDict vs Pydantic BaseModel

實戰經驗：先用 TypedDict 起步，跑通之後再對「從外部來的 input node」(例如 webhook 收到的 payload) 換成 Pydantic。一開始全用 Pydantic 會在每個 node transition 額外吃 5-15ms，多 node graph 累積起來不容忽視。

Reducer 是 state 設計的核心決策

預設行為（沒 Annotated）：node 回傳什麼，就覆蓋對應欄位的舊值。這對「累積型」資料是災難 — 第二個 node 把第一個 node 加的東西全清空。

Reducer 的三種常見模式：

from typing import Annotated
from operator import add

# 模式 1：累加 list (常見於 log / messages / tool_calls)
log_entries: Annotated[list[str], add]

# 模式 2：合併 dict (常見於 multi-key 結果集)
def merge_dict(a: dict, b: dict) -> dict:
    return {**a, **b}
results: Annotated[dict, merge_dict]

# 模式 3：自定義邏輯 (取最新、取最大、加總...)
def last_wins(a, b):
    return b if b is not None else a
status: Annotated[str, last_wins]

Conditional Edges：把分支邏輯寫進程式碼，不要塞進 prompt

新手用 LangGraph 最常做的反模式是「prompt 裡寫 if 邏輯」，例如「如果測試過了就回傳 GO_REVIEW，否則回傳 GO_RETRY」。然後在外層解析 LLM 回應字串、用 if-else 跳轉。

這是浪費 LangGraph 的核心價值。它就是為了讓你把分支邏輯從 prompt 拉出來、寫成 Python function 才存在的。

正確寫法：

from langgraph.graph import StateGraph, END

def route_after_test(state: PipelineState) -> str:
    failed = [r for r in state["test_results"] if r["status"] == "failed"]
    if failed and state.get("retry_count", 0) < 3:
        return "coder"          # 失敗且重試次數未滿 → 回去修
    elif failed:
        return "human_review"   # 重試已滿 → 升級給人類
    else:
        return "reviewer"        # 全綠 → 走 review

graph = StateGraph(PipelineState)
graph.add_node("spec", spec_agent)
graph.add_node("coder", coder_agent)
graph.add_node("tester", test_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_node("human_review", human_review_node)

graph.add_edge("spec", "coder")
graph.add_edge("coder", "tester")
graph.add_conditional_edges(
    "tester",
    route_after_test,
    {"coder": "coder", "reviewer": "reviewer", "human_review": "human_review"},
)
graph.add_edge("reviewer", END)
graph.add_edge("human_review", END)

看到 `route_after_test` 了嗎？它是純 Python 函式 — 可以單元測試、可以 step debug、可以 log、可以加 metric。把 retry policy、escalation rule、business logic 全部寫在這裡，LLM 完全不碰。

這個設計哲學叫 「結構在 graph、智慧在 node、不要把結構塞進 prompt」。Graph 是確定性的（你寫 if，它就走 if），node 內部是非確定性的（LLM 怎麼回答你不知道）。把這兩層分開，整個 pipeline 才能被 audit、可 reproduce、可 debug。

Checkpointer 與 Time Travel：可 rollback pipeline 的工程意義

Checkpointer 不只是「斷線重連」這麼簡單。DEV.to 上有篇文章把這個價值講得很到位：「debug 非確定性的 LLM agent 最痛的是『同一個 input 兩次跑出不同結果』。Checkpoint 讓你能 freeze 中間任何一個 state，用 deterministic 方式重跑後面」。

Time Travel 的三個工程場景

• 場景一：production 出包後 reproduce bug。客戶回報「我昨天跑 export 出來的 JSON 少了三筆」。從 checkpointer 撈出那 thread 的歷史，找到 export node 跑完當下的 state，看到他傳的 filter 是 `created_at > 7 days ago`，少的三筆是 6 天 23 小時前的 — 時區 bug。整個 trace 不到 10 分鐘。
• 場景二：分支實驗（What If）。「如果 Reviewer Agent 用 Sonnet 而不是 Opus，結果會差很多嗎？」從昨天某個 review failed 的 checkpoint 開始，改一行 model name，從那個 state 繼續跑。完全不需要重跑前面 1 小時的 work。
• 場景三：Human-in-the-Loop 的天然支援。Pipeline 跑到一半 interrupt 給人類確認，人類在週末才回。週一 resume — 因為 state 在 Postgres 持久化著，三天前的 graph 完整繼續走，不需要從頭跑。

PostgresSaver 設定範例：（production 標準配置）

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import StateGraph

DB_URI = "postgresql://user:pass@host:5432/agent_state"

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 第一次跑要 setup 建表
    await checkpointer.setup()

    graph = builder.compile(checkpointer=checkpointer)

    # 每次 invoke 都要帶 thread_id（identifier for the conversation）
    config = {"configurable": {"thread_id": "pipeline-user-export-2026-05-17"}}

    result = await graph.ainvoke(
        {"feature_name": "user-export", "spec_path": "docs/prd.md"},
        config=config,
    )

    # 撈歷史 checkpoint
    async for snapshot in graph.aget_state_history(config):
        print(f"Step {snapshot.metadata['step']}: {snapshot.next}")

Human-in-the-Loop：interrupt 與 resume 的工程實作

Production 級 multi-agent pipeline 一定會有「需要人類確認」的點。Agent 要刪資料庫前、要 deploy 到 production 前、要花超過 $50 的 API 呼叫前。LangGraph 用 interrupt() 函式原生處理這個。

LangChain 官方部落格的解釋很乾脆：「呼叫 interrupt() 時，graph 暫停執行、把 thread 標記為 interrupted、把你傳給 interrupt 的東西放進 persistence layer。你之後用 graph.invoke(Command(resume="value"), thread) 把人類的決定傳回去。」

最小可用範例：

from langgraph.types import interrupt, Command

def deploy_approval_node(state: PipelineState) -> dict:
    # 跳出來等人類審核
    decision = interrupt({
        "action": "deploy",
        "pr_url": state["pr_url"],
        "test_pass_rate": state["test_pass_rate"],
        "estimated_cost": state["estimated_cost"],
        "question": "確定要 deploy 到 production 嗎？(approve/reject/defer)",
    })

    # 程式從這裡繼續，decision 就是 resume value
    return {
        "deploy_decision": decision,
        "log_entries": [f"human decided: {decision}"],
    }

# 第一次跑：interrupt 觸發
result = graph.invoke(initial_state, config={"configurable": {"thread_id": "deploy-2026"}})
# result 裡有 __interrupt__ 欄位帶出 interrupt 的內容

# 把它推給人類介面（Slack / Email / Admin 後台），等他回覆

# 拿到人類決定後 resume
graph.invoke(
    Command(resume="approve"),
    config={"configurable": {"thread_id": "deploy-2026"}},
)

interrupt() 的神奇之處在於：人類審核可以等三天，你的 worker process 可以重啟、可以擴展、可以 deploy 新版本，state 都在 Postgres 不會掉。這跟你自己用 webhook + queue 自幹一套的差距是工程量級的。

把 Claude Agent SDK 包成 LangGraph node：拿到兩邊的好處

前面提過 Claude Agent SDK 是 Anthropic 出的 agent harness — Loop / Tool / Context / Verification 都內建。而 LangGraph 是上層的 orchestration framework。兩個其實互補：

• LangGraph 處理「整條 pipeline 怎麼串、stage 之間的 routing、checkpoint、human approve」這些外層工程議題
• Claude Agent SDK 處理「在某個 stage 裡 Claude 要怎麼讀檔、用工具、跑 subagent」這些內層執行細節

Mager 部落格把這個 pattern 講得很直接：「LangGraph for workflow skeleton, Claude Agent SDK for heavy lifting inside nodes」。

整合範例：把一個 Coder Agent 用 Claude Agent SDK 寫成、再包成 LangGraph node

import asyncio
from typing import TypedDict, Annotated
from operator import add
from pathlib import Path

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from claude_agent_sdk import query, ClaudeAgentOptions

class PipelineState(TypedDict):
    feature_name: str
    spec_path: str
    log_entries: Annotated[list[str], add]
    code_result: str
    test_passed: bool

async def coder_node(state: PipelineState) -> dict:
    """用 Claude Agent SDK 跑一個完整的 coder agent。"""
    options = ClaudeAgentOptions(
        system_prompt=Path("prompts/coder.md").read_text(),
        allowed_tools=["Read", "Write", "Edit", "Bash", "Grep", "Glob"],
        permission_mode="acceptEdits",
        cwd="/workspace/my-project",
    )

    final_text = ""
    async for msg in query(
        prompt=f"Read {state['spec_path']}, implement {state['feature_name']}, "
               f"run pytest, commit when green.",
        options=options,
    ):
        if msg.type == "result":
            final_text = msg.text

    return {
        "code_result": final_text,
        "log_entries": [f"coder_node: finished, {len(final_text)} chars output"],
    }

async def tester_node(state: PipelineState) -> dict:
    # 純 Python，不需要 LLM — 直接跑 pytest subprocess
    import subprocess
    result = subprocess.run(["pytest", "--tb=short"], capture_output=True, text=True)
    return {
        "test_passed": result.returncode == 0,
        "log_entries": [f"tester_node: {'passed' if result.returncode == 0 else 'failed'}"],
    }

def route_after_test(state: PipelineState) -> str:
    return "reviewer" if state["test_passed"] else "coder"

async def main():
    builder = StateGraph(PipelineState)
    builder.add_node("coder", coder_node)
    builder.add_node("tester", tester_node)
    builder.add_node("reviewer", reviewer_node)

    builder.set_entry_point("coder")
    builder.add_edge("coder", "tester")
    builder.add_conditional_edges("tester", route_after_test,
                                   {"coder": "coder", "reviewer": "reviewer"})
    builder.add_edge("reviewer", END)

    async with AsyncPostgresSaver.from_conn_string(DB_URI) as cp:
        await cp.setup()
        graph = builder.compile(checkpointer=cp)
        config = {"configurable": {"thread_id": "user-export-2026-05-17"}}
        result = await graph.ainvoke({
            "feature_name": "user-export",
            "spec_path": "docs/prd.md",
            "log_entries": [],
            "code_result": "",
            "test_passed": False,
        }, config=config)
        print(result)

asyncio.run(main())

看出整合的優雅：LangGraph 把流程結構講清楚（誰跑完誰、失敗回哪），Claude Agent SDK 在 node 內部把實作做掉。每個 node 可以是 SDK call，也可以是純 Python，graph 一視同仁。

Production 部署：observability、cost、failure modes

把 LangGraph graph 跑起來不難，跑得穩才難。Production 部署有三個工程責任你必須認領：observability、cost control、failure mode handling。

Observability：LangSmith 是預設選項

DigitalApplied 2026 年 4 月的整理指出，2026 上半年 agent observability 領域已收斂到六個 production-grade 平台：LangSmith、Langfuse、Arize Phoenix、Helicone、Datadog LLM Observability、Honeycomb LLM Observability。LangGraph stack 上 LangSmith 是預設選項，整合只要 `LANGSMITH_TRACING=true` 一個環境變數，每個 node、每個 tool call、每個 state diff 都自動 trace。

Cost：state diff 才是省 token 關鍵

前面提過 LangGraph 同樣任務只吃 LangChain 約 1/2 token、AutoGen 約 1/4。差在哪？差在 state diff 機制 — node 之間只傳「變化了什麼」，不是把整個對話歷史塞回 prompt。Cost 直接砍掉重複轉述的部分。

Failure Mode：三個一定會遇到的

• Recursion limit hit。LangGraph 預設 25 次 node transition 上限。複雜 graph 一定要主動調整，並且在 routing function 加防呆 — 用「retry_count 超過 3 就強制走 human_review」這種顯式上限。
• Checkpoint table 膨脹。每個 node 都寫 checkpoint，跑大型 pipeline 一週可以堆出幾 GB。要設 retention policy：完成的 thread 保留 30 天、進行中的不限。寫成排程刪除任務。
• LLM provider rate limit。一個 graph 同時 spawn 多個 LLM 呼叫的 node，撞到 Anthropic / OpenAI rate limit。LangGraph 沒有原生 rate limit，需要在 node 層接 tenacity 或自己的 token bucket。

業界踩過的五個坑

下面這些是業界把 LangGraph 推上 production 的真實踩坑紀錄。寫出來給你繞開。

坑一：State 設計成「巨型 dict」，後期重構地獄

MVP 階段為了快，所有資料都塞 state 同一層。三個月後 state 有 40 個欄位，沒人說得清哪些 node 會改哪些。正解：用巢狀 TypedDict 分區，把欄位歸類成 `pipeline_meta`、`coder_output`、`test_results`、`review_state` 等子物件，每個 node 的責任邊界一目瞭然。

坑二：把 LLM 呼叫塞到 conditional edge 函式

想偷懶把「該走哪條分支」用 LLM 判斷。Conditional edge function 變成「呼叫 LLM 問下一步」。災難 — edge function 應該是純 Python、deterministic、可單元測試。LLM 判斷該回到 node 內部、回傳 state 欄位，edge 讀那個欄位決定走向。

坑三：用 MemorySaver 上 production，重啟就忘

教學範例幾乎全用 MemorySaver，新人 copy 過來就 deploy。Process 一重啟所有 thread 通通沒了。產品經理回報「客戶按 approve 之後系統忘記他按過」— debug 半天才發現是 checkpointer 問題。Production 永遠用 PostgresSaver。寫成 lint rule 擋住 MemorySaver 進 main branch。

坑四：沒設 recursion_limit，graph 跑無限迴圈燒錢

Conditional edge 寫錯一個條件，retry 永遠不會停。預設 25 次跑完才丟 GraphRecursionError，但 25 次 Claude Opus 呼叫已經 $5+。正解：在 routing function 內顯式檢查 retry_count，超過上限強制走 human_review 或 END，不要依賴 recursion_limit 救你。

坑五：把 secret 直接放進 state

新手把 API key、DB credential 寫進 state 方便 node 取用。Checkpoint 表會把整個 state JSON 化存到 Postgres，連 secret 一起持久化。如果 LangSmith tracing 開著，secret 還會送去 LangChain 雲端。正解：secret 永遠從 env 讀、永遠不進 state。Node 函式直接讀 env，state 只放「需要跨 node 共享的業務資料」。

什麼時候用 LangGraph、什麼時候別用

技術選型最後一定要回到「適不適合」這個問題。LangGraph 強，但不是所有場景都需要它。

如果你還在「自建 vs 採購 vs 接案」的選型階段，建議搭配閱讀 AI Agent 系統採購完整框架：老闆視角的 Workflow vs Agent 判斷；如果你的痛點是 Agent 在 production 翻車，可以看 LLM Evals 完整指南：技術主管的 AI 評估流程。

常見問題

從一個小 graph 開始，半年內把 production pipeline 搬上來

讀到這裡你應該理解了：LangGraph 不只是另一個 framework，是把 multi-agent 從「實驗室 demo」推到「production 系統」的工程基礎建設。State、Node、Edge、Checkpointer 四個概念，加上 Conditional Edge 與 Interrupt 兩個機制，就能涵蓋 production 級工作流 90% 的需求。

建議的學習路徑：第一週用 MemorySaver 跑通一個三 node 的 toy graph（讀檔 → call LLM → 寫檔）；第二週加 Conditional Edge 跟 retry；第三週換 PostgresSaver、加 Time Travel；第四週加 interrupt 跟 human approve；第五週把 Claude Agent SDK 包進 node 做真實工作。一個月後你就有能力把任何重複性開發工作流變成可 audit、可 rollback、可 human approve 的 production pipeline。

完整 Agent 工程知識鏈推薦這樣讀：先看 Claude Code /loop 完整教學 了解最簡單的定時 agent，再看 Multi-Agent Debate 架構設計與實作指南 理解多 agent 協作概念，本篇接著看 Claude Agent SDK 端到端自動化開發完整實戰，三篇連起來剛好走完從原理到實戰的完整路徑。

延伸閱讀：如果你正在規劃 Multi-Agent 系統的記憶層或知識檢索層，建議搭配看 圖資料庫 Neo4j 完整介紹：何時該升級成 Graph DB？——LangGraph 管「Agent 推理流程的圖」，Neo4j 管「Agent 知識記憶的圖」，兩者搭配能撐起完整的長期記憶與結構推理能力。