LangChain v1 與 LangGraph:新一代 Agent 編排架構深度解析


0. 為什麼要再看一次 LangChain?

早期的 LangChain 給人的印象是「把 LLM + 工具 + 記憶串起來的便利框架」。但從 LangChain v1 開始,官方實際上做了兩個重要重構:

  1. 把「編排/工作流」能力沉到底層,變成獨立產品 LangGraph
  2. 把「Agent 框架」重寫在 LangChain v1 上,直接建立在 LangGraph 的狀態機/圖執行引擎之上

結果是:

  • LangGraph 成為一個可獨立使用的「有狀態、多節點、可持久化的 AI 工作流引擎」
  • LangChain v1 則成為「建在 LangGraph 之上的高階 Agent 框架」,提供中間件系統、結構化輸出、工具協作等

整體關係可以用一張圖概括:

flowchart TD
  classDef pad fill:transparent,stroke:transparent,color:transparent;

  subgraph Infra
    direction TB
    infra_pad[" "]:::pad
    LLM["LLM Providers<br/>OpenAI / Anthropic / ..."]
    Tools["Tools / APIs<br/>DB / HTTP / Search"]
    Memory["Stores / Checkpoints"]
  end

  subgraph LangGraph["LangGraph (Orchestration Layer)"]
    direction TB
    lg_pad[" "]:::pad
    SG["StateGraph / MessageGraph"]
    Pregel["Pregel Engine<br/>BSP, Channels, Checkpoints"]
  end

  subgraph LangChainV1["LangChain v1 <br/>(Agent Framework)<br/>"]
    direction TB
    lcv1_pad[" "]:::pad
    AgentFactory["create_agent()"]
    Middleware["AgentMiddleware Stack"]
    StructOut["Structured Output Strategies"]
  end

  LangChainCore["langchain-core<br/>BaseChatModel <br/>/ BaseTool <br/>/ Messages"]

  LLM --> LangChainCore
  Tools --> LangChainCore
  LangChainCore --> LangGraph
  Memory --> LangGraph
  LangGraph --> LangChainV1

接下來分別說明:

  • LangGraph:底層的 BSP 圖引擎 & 狀態系統
  • LangChain v1:建立在 LangGraph 上的 Agent 中間件架構
  • 如何在實戰中組合兩者

1. LangGraph:把「有狀態 AI 工作流」當成一張圖

1.1 LangGraph 的定位

LangGraph ≠ LangChain 的附屬套件,而是一個獨立的編排框架,特性包括:

  • 基於 Google PregelBulk Synchronous Parallel (BSP) 模型
  • 使用 StateGraph / MessageGraph 來定義有狀態工作流
  • 內建:
    • Channel-based 狀態管理
    • Checkpoint(持久化、時間旅行)
    • 人在迴圈(interrupt / resume)
    • 多代理(Network / Supervisor / Hierarchical)

你可以把 LangGraph 想像成「Kubernetes for AI workflows」:你定義節點與邊,LangGraph 負責調度、狀態、重試、中斷與恢復。

1.2 Pregel / BSP 執行模型

LangGraph 的核心是 Pregel-style 的 BSP 執行迴圈。每次執行由多個「超步」組成,每個超步包含四個階段:

flowchart TD
  Start([開始])
  Plan["1. Plan<br/>決定活躍節點"]
  Exec["2. Execute<br/>並行執行節點"]
  Update["3. Update<br/>套用通道更新"]
  Checkpoint["4. Checkpoint<br/>保存狀態"]
  Loop{還有活躍節點?}
  End([結束])

  Start --> Plan --> Exec --> Update --> Checkpoint --> Loop
  Loop -->|是| Plan
  Loop -->|否| End

流程細節:

Plan

  • 根據通道版本 [channel_versions] 與 [versions_seen] 判斷哪些節點需要執行
  • 不掃描完整狀態,只靠版本號比較 → O(1) 活躍節點判斷

Execute

使用 ThreadPoolExecutor 等並行執行活躍節點 (Async 用原生)
# 關鍵代碼
# _future.py:189-203 - run_coroutine_threadsafe
def run_coroutine_threadsafe(coro, loop, ...):
    if asyncio._get_running_loop() is loop:
        # 已在同一事件循環 → 直接創建 asyncio.Task
        return _ensure_future(coro, loop=loop, ...)
    else:
        # 從其他線程調用 → 用 call_soon_threadsafe 調度
        loop.call_soon_threadsafe(callback, ...)

# _executor.py:122-140 - AsyncBackgroundExecutor
class AsyncBackgroundExecutor:
    def __init__(self, config):
        self.loop = asyncio.get_running_loop()  # ← 用 asyncio loop
        self.semaphore = asyncio.Semaphore(max_concurrency)  # ← asyncio 信號量
節點執行中只能讀「上一超步」的通道狀態 → 避免 race condition

Explain: 這是 Pregel BSP 模型的核心設計。讓我用具體例子解釋:

假設場景

三個節點同時執行,都要讀寫同一個 channel “counter”

初始值: counter = 10

def node_A(state): val = state[“counter”] # 讀 return {“counter”: val + 1} # 寫 +1

def node_B(state): val = state[“counter”] # 讀 return {“counter”: val + 2} # 寫 +2

def node_C(state): val = state[“counter”] # 讀 return {“counter”: val + 3} # 寫 +3


❌ 如果是「即時讀寫」(傳統共享內存)

時間軸 → T1 T2 T3 T4 Node A: 讀=10 寫=11 Node B: 讀=11 寫=13 Node C: 讀=13 寫=16

結果: counter = 16 (順序執行的結果)

問題:

  • 執行順序不同 → 結果不同(不確定性)
  • 需要加鎖 → 效能差
  • Race condition:如果 A 和 B 同時讀到 10,都寫回去,一個會被覆蓋

✅ BSP 模型:「讀上一超步」

超步 N 開始時: counter = 10 (快照)


┌─────────────────────────────────────────────┐
│ 並行執行階段 (所有節點讀的都是同一個快照) │
│ │
│ Node A: 讀=10 → 本地緩衝寫 +1 │
│ Node B: 讀=10 → 本地緩衝寫 +2 │
│ Node C: 讀=10 → 本地緩衝寫 +3 │
│ │
│ (三個節點讀到的都是 10,互不干擾) │
└─────────────────────────────────────────────┘

┌─────────────────────────────────────────────┐
│ Barrier (等待所有節點完成) │
└─────────────────────────────────────────────┘

┌─────────────────────────────────────────────┐
│ apply_writes() 批量合併 │
│ │
│ 緩衝區: [+1, +2, +3] │
│ 合併策略 (reducer): 10 + 1 + 2 + 3 = 16 │
└─────────────────────────────────────────────┘

超步 N+1 開始時: counter = 16


關鍵差異

傳統共享內存BSP 模型
讀到「當前」值(可能被其他線程改過)讀到「超步開始時的快照」
寫立即生效寫入本地緩衝,超步結束後批量合併
需要鎖/原子操作無需鎖,天然並行安全
結果依賴執行順序結果確定性(與執行順序無關)

源碼對應

# \_loop.py - tick() 準備任務時

def tick(self): # 基於當前 checkpoint 的 channels 狀態建立任務 # 這裡的 channels 是「快照」,不會在執行中被修改
  self.tasks = prepare_next_tasks(
  self.checkpoint,
  self.channels, # ← 快照
  ...
)

# \_loop.py - after_tick() 超步結束後

def after_tick(self): # 所有任務完成後,才統一 apply 寫入
  self.updated_channels = apply_writes(
  self.checkpoint,
  self.channels,
  self.tasks.values(), # ← 收集所有任務的 writes
  ...
)

一句話總結

「讀上一超步」= 所有節點讀同一份快照,寫入各自緩衝,超步結束後統一合併

這樣無論節點以什麼順序執行、在哪個線程執行,結果都是確定的。

  1. Update
  • 收集節點產生的「通道寫入」,透過 [channel.update()] 聚合到通道中
  • 通道可有多種聚合策略(LastValue / BinaryOperatorAggregate / Topic 等)
  1. Checkpoint
  • 可選同步或異步,保存:
    • channel_values
    • channel_versions
    • versions_seen
  • 對應到 LangGraph 的 Checkpointer 介面(Memory / SQLite / Postgres / 自訂)

這個 BSP 模型帶來:

  • 確定性:同樣的輸入、同樣的圖,結果可重現
  • 易恢復:任意超步都可作為最新 checkpoint 恢復
  • 無競態條件:節點執行階段不會互相寫同一通道(寫入在 Update phase 聚合)

1.3 StateGraph:用 Channel 來表達「AI Agent 的狀態」

LangGraph 不直接操作「變數」,而是操作「通道(Channel)」。可以想像成:

  • 每個 State 欄位背後,是一個對應的 Channel
  • Channel 負責定義「如何合併多個節點更新」

簡化示例:

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

class State(TypedDict):
    messages: Annotated[list, add]  # 使用 add 作為 reducer
    count: int                      # 覆蓋策略(LastValue)

builder = StateGraph(State)
builder.add_node("chatbot", chatbot_func)
builder.add_node("tools", tools_func)

builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", should_continue, {"continue": "tools", "end": END})
builder.add_edge("tools", "chatbot")

graph = builder.compile(checkpointer=checkpointer)

實際對應到通道層大致如下:

flowchart LR
  subgraph State
    M["messages<br/>BinaryOperatorAggregate(add)"]
    C["count<br/>LastValue"]
  end

  ChatbotNode -->|append message| M
  ToolsNode -->|append message| M
  ChatbotNode -->|set count| C

Channel 的設計提供了:

  • 明確的狀態聚合策略
  • 分離「節點邏輯」與「狀態合併」

常見通道類型包括:

  • LastValue:永遠保留最新值(預設)
  • BinaryOperatorAggregate:用指定 Binary Operator 聚合多個更新
  • Topic:pub/sub 風格
  • EphemeralValue:不持久化的臨時值
  • NamedBarrierValue:用於同步屏障

1.4 Checkpoint 與版本追蹤

LangGraph 的 Checkpoint 結構大約是:

class Checkpoint(TypedDict):
    v: int                              # Checkpoint 版本
    id: str                             # 單調遞增 ID
    ts: str                             # ISO 時間戳
    channel_values: dict[str, Any]      # 通道值
    channel_versions: dict[str, int]    # 每個通道的版本
    versions_seen: dict[str, dict[str, int]]  # 節點已見版本

其中:

  • [channel_versions]:每個通道有版本號,每次更新 +1
  • [versions_seen]:每個節點記錄「當前已讀到的各通道版本」

節點是否需要被啟動的判斷條件為:

[ \exists\ channel\ c,\ channel_versions[c] > versions_seen[node][c] ]

這使得「活躍節點偵測」只需要看版本號,不需要對比整個狀態 → 重度依賴 Channel & Checkpoint 設計。


2. LangChain v1:建立在 LangGraph 之上的 Agent 中間件框架

2.1 LangChain v1 的核心定位

在 v1 之前,很多人把 LangChain 當作:

  • 調 LLM 的 SDK
  • 封裝工具、記憶、Chain

在 v1 中,LangChain 的核心 Agent 模塊已經變成:

  • 建立在 LangGraph 的 StateGraph 之上的 高階 Agent 工廠
  • 提供一套 中間件(Middleware)系統,用來攔截:
    • Agent 生命週期(before/after_agent, before/after_model)
    • 模型呼叫(wrap_model_call)
    • 工具呼叫(wrap_tool_call)
  • 內建 10+ AgentMiddleware,例如:
    • 模型重試 / 回退 / 調用限制
    • 工具重試 / 調用限制
    • PII 檢測
    • 人在迴圈
    • Shell 執行環境
    • 文件搜尋、摘要、TodoList、Context Editing 等

整體堆疊可以用下圖表示:

flowchart TD
  subgraph LangGraph["LangGraph (StateGraph Engine)"]
    SG["StateGraph<br/>AgentState"]
    Pregel["Pregel Execution<br/>BSP + Channels"]
  end

  subgraph LangChainV1["LangChain v1 Agents"]
    CA["create_agent()"]
    MW["AgentMiddleware<br/>Model/Tool hooks"]
    Struct["Structured Output<br/>ResponseFormat Strategies"]
  end

  UserCode["User Code<br/>LLM / Tools / Middleware Config"]

  UserCode --> CA
  CA --> SG
  SG --> Pregel
  MW --> CA
  Struct --> CA

2.2 create_agent():用一行 API 建出一個 StateGraph Agent

v1 的核心 API 是:

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-sonnet-4-5-20250929",
    tools=[get_weather],
    system_prompt="You are a helpful assistant",
    middleware=[...],
    response_format=...,     # Pydantic / TypedDict / Dataclass
    checkpointer=...,        # MemorySaver / Sqlite / Postgres / ...
    store=...,               # 跨會話 store
)

它實際做的事情:

  1. 根據你提供的 model / tools / system_prompt 建立一個 LangGraph StateGraph
  2. 注入一個預設的「ReAct 類型代理迴圈」:
flowchart TD
 START --> BeforeAgent
 subgraph AgentLoop["Agent Loop (may repeat)"]
   BeforeModel --> ModelNode --> AfterModel
   AfterModel -->|no tool calls| END
   AfterModel -->|tool calls| ToolsNode --> BeforeModel
 end
 BeforeAgent --> BeforeModel
 AfterModel --> AfterAgent --> END
  1. 把你設定的 middleware 一層一層包在:
  • [before_agent / before_model / after_model / after_agent] 生命週期 hook
  • [wrap_model_call] 模型呼叫包裝
  • [wrap_tool_call] 工具呼叫包裝
  1. 最後 compile 成 LangGraph 的 CompiledStateGraph,提供:
  • [invoke] / [ainvoke] 同步/非同步執行
  • [stream] / [astream] 各種 stream mode(values / updates / messages / debug)
  • [batch] 批次處理

換句話說,create_agent() 的回傳值,其實仍然是一個 LangGraph Graph,只是已經幫你把「Agent 迴圈/工具節點/中間件」都定義好了。

2.3 AgentState / ModelRequest / ModelResponse:Agent 的狀態與呼叫模型的資料模型

LangChain v1 把 Agent 的狀態,也定義為一個 TypedDict(實際上會轉成 LangGraph 的 State):

AgentState (TypedDict)
  ├── messages: list[AnyMessage]         # 有 add_messages reducer
  ├── jump_to: Optional[str]             # tools / model / end
  └── structured_response: Any           # 結構化輸出結果

呼叫模型與工具則使用強型別的資料類:

  • ModelRequest:模型呼叫前的上下文

    ModelRequest
      ├── model: BaseChatModel
      ├── messages: AnyMessage[]
      ├── system_message: SystemMessage | None
      ├── tools: (BaseTool | dict)[]
      ├── response_format: ResponseFormat | None
      ├── tool_choice: Any | None
      ├── state: AgentState
      ├── runtime: Runtime
      └── model_settings: dict
  • ModelResponse:模型呼叫後的結果

    ModelResponse
      ├── result: list[BaseMessage]        # 通常 1 個 AIMessage + 可選 ToolMessage
      └── structured_response: Any

這些類型正是中間件攔截點(wrap_model_call / wrap_tool_call)的操作對象。

2.4 AgentMiddleware:把 Agent 當成一條「中間件管線」

AgentMiddleware 提供兩大類介面:

  1. 生命週期鉤子(Lifecycle Hooks)
class AgentMiddleware(Generic[StateT, ContextT]):
    def before_agent(self, state, runtime) -> dict | None: ...
    def before_model(self, state, runtime) -> dict | None: ...
    def after_model(self, state, runtime) -> dict | None: ...
    def after_agent(self, state, runtime) -> dict | None: ...
  • 回傳的 dict 會 merge 回 AgentState
  • 可以用來修改狀態、增加 messages、設定 jump_to 等
  1. 攔截鉤子(Interception Hooks)
class AgentMiddleware(...):
    def wrap_model_call(self, request, handler) -> ModelResponse: ...
    def wrap_tool_call(self, request, execute) -> ToolMessage | Command: ...
  • [handler] 是「下一層」的模型呼叫
  • 你可以:
    • 在呼叫前後記錄 log
    • 修改 request(e.g. 改模型、改溫度、改工具列表)
    • 捕捉錯誤並重試/回退

這些 hook 可以用 class 方式實作,也可以用 decorator:

from langchain.agents.middleware import (
    AgentMiddleware, wrap_model_call, before_model
)

class LoggingMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        print(f"Calling model with {len(request.messages)} messages")
        response = handler(request)
        print(f"Model returned: {response.result[0].content[:50]}...")
        return response


@wrap_model_call()
def my_model_wrapper(request, handler):
    print(f"[Wrapper] Before: {request.model}")
    result = handler(request)
    print(f"[Wrapper] After")
    return result

中間件的組合順序 則非常關鍵:

graph LR
  Base[Base Model Call]
  MW3[Middleware 3]
  MW2[Middleware 2]
  MW1[Middleware 1]

  Base --> MW3 --> MW2 --> MW1 --> User

  %% 執行順序:MW1 -> MW2 -> MW3 -> Base -> MW3 -> MW2 -> MW1

實際上,LangChain v1 透過:

  • [_chain_model_call_handlers]
  • [_chain_tool_call_wrappers]
  • 以及對應的 async 版本

來包成類似:

[ outer(inner(innermost(base_handler))) ]

的形式。([\text{handlers[0]}] 為最外層)

2.5 內建中間件:把常見場景封裝成 Middleware

LangChain v1 預設提供 14+ 中間件,包含典型的生產場景需求:

中間件功能典型用途
ModelRetryMiddleware模型呼叫重試LLM API 不穩、偶發錯誤
ModelFallbackMiddleware模型回退主要模型掛了換備援模型
ModelCallLimitMiddleware模型呼叫限制防止 prompt 自我反覆迴圈
ToolRetryMiddleware工具呼叫重試後端 API 不穩
ToolCallLimitMiddleware工具呼叫限制防止 Agent 無限用同一工具
LLMToolSelectorMiddlewareLLM 工具選擇多工具選擇最佳一個/多個
LLMToolEmulator工具模擬在沒有真工具時做 dry-run
HumanInTheLoopMiddleware人工確認Human approval / 修改
ShellToolMiddlewareShell 執行代碼執行、DevOps 操作
FilesystemFileSearchMiddleware文件搜尋本地知識庫檢索
PIIMiddlewarePII 檢測Email/信用卡等敏感資訊處理
SummarizationMiddleware對話摘要長對話壓縮、長期記憶
TodoListMiddleware任務追蹤任務列表、自動追蹤進度
ContextEditingMiddleware上下文編輯對 prompt / context 做自動改寫

與其自己寫「重試、回退、限制、Human-in-the-loop」,不如直接把這些中間件掛上去。


3. 結構化輸出(Structured Output):從 Pydantic 到 Provider Strategy

LangChain v1 的一大設計亮點 是針對「結構化輸出」的抽象:ResponseFormat

3.1 ResponseFormat 的策略模式

當你在 create_agent() 中指定:

from pydantic import BaseModel

class WeatherResponse(BaseModel):
    location: str
    temperature: int
    condition: str

agent = create_agent(
    model="openai:gpt-4o",
    tools=[get_weather],
    response_format=WeatherResponse
)

框架會自動決定用哪種策略讓 LLM 產生結構化輸出:

  • ProviderStrategy(優先):

    • 如果 LLM 提供原生結構化 API,例如 OpenAI response_format = {"type": "json_object"} 或 JSON mode
    • 則自動產生對應的 provider-specific 設定
    • 支援 strict 驗證
  • ToolStrategy(退而求其次):

    • 如果模型不支援 ProviderStrategy
    • 則使用「虛擬 tool」方式:
  • 綁定一個 schema 對應的 tool

  • 要求 LLM 呼叫這個 tool,並把參數 parse 成結構化輸出

  • AutoStrategy(包裝 raw schema):

    • 若你直接傳入 Pydantic / Dataclass / TypedDict
    • 會被包成一個 AutoStrategy,內部再由上述兩者擇一使用

自動決策邏輯大致是:

if raw_schema provided:
    if model_supports_provider_strategy():
        use ProviderStrategy
    else:
        use ToolStrategy

結果是:

  • 你在 user code 只需要關心「我要得到一個強型別 [WeatherResponse]」
  • 框架會根據 model provider 能力選擇:用原生 JSON mode 還是 tool-call workaround

深度分析

執行流程圖

  Agent Loop:
  ┌──────────────────────────────────────────────────────────┐
  │  Step 1: Model 輸出                                       │
  │                                                          │
  │  ┌─────────────────┐     ┌─────────────────────────────┐ │
  │  │ AIMessage       │     │ 有 tool_calls?              │ │
  │  │ (模型回應)       │ ──► │                             │ │
  │  └─────────────────┘     └─────────────────────────────┘ │
  │                                 │                        │
  │              ┌──────────────────┼──────────────────┐     │
  │              ▼                  ▼                  ▼     │
  │     ┌────────────────┐  ┌────────────────┐ ┌───────────┐ │
  │     │ 普通 tool call │  │ 結構化輸出工具  │ │ 無 tool   │ │
  │     │ (search, read) │  │ (WeatherResp)  │ │ call      │ │
  │     └───────┬────────┘  └───────┬────────┘ └─────┬─────┘ │
  │             │                   │                │       │
  │             ▼                   ▼                ▼       │
  │     ┌────────────────┐  ┌────────────────┐ ┌───────────┐ │
  │     │ 執行工具       │  │ Parse 結構化   │ │ Provider  │ │
  │     │ 繼續迴圈       │  │ 輸出, 結束迴圈 │ │ Strategy  │ │
  │     │ (中間階段)     │  │ (最終輸出)     │ │ Parse     │ │
  │     └────────────────┘  └────────────────┘ └───────────┘ │
  │                                                          │
  └──────────────────────────────────────────────────────────┘

實際例子

agent = create_agent(
    model="openai:gpt-4o",
    tools=[search, read_file],
    response_format=WeatherResponse  # Pydantic model
)

# 執行過程:
# Turn 1: Model → "I'll search for weather" + tool_call(search)
#         → 執行 search 工具,繼續迴圈 ✅
#         → structured_response = None (未 parse)

# Turn 2: Model → "Found data, let me read file" + tool_call(read_file)
#         → 執行 read_file 工具,繼續迴圈 ✅
#         → structured_response = None (未 parse)

# Turn 3: Model → tool_call(WeatherResponse, args={...})
#         → 匹配到結構化輸出工具 ❗
#         → Parse args 為 WeatherResponse
#         → structured_response = WeatherResponse(...)
#         → 結束迴圈,返回結果 ✅

一句話總結

response_format 只在模型「決定輸出最終答案」時生效:

  • 中間的 reasoning + tool calls → 不受影響,正常執行
  • 最終輸出(無 tool call 或 call 結構化工具)→ Parse 為指定格式,結束迴圈

4. LangChain v1 與 LangGraph 的結構對應關係

可以用一張「分層架構圖」來統整:

flowchart TD
  subgraph UserApp["User Application"]
    UC1["Custom Tools<br/>DB / HTTP / Search"]
    UC2[Custom Middleware]
    UC3[Business Logic]
  end

  subgraph LangChainV1["LangChain v1 (Agents)"]
    CF["create_agent()"]
    MWStack[Middleware Stack]
    SO["Structured Output<br/>(ResponseFormat)"]
    ToolsBinding[Tool Binding]
  end

  subgraph LangGraph["LangGraph (Stateful Orchestration)"]
    SG[StateGraph / MessageGraph]
    Channels["Channels<br/>(messages, state, ...)"]
    Pregel["Pregel Engine<br/>BSP, Checkpoints"]
    Checkpointer["Checkpointer<br/>Memory / SQLite / Postgres / Custom"]
  end

  subgraph LangChainCore["langchain-core"]
    Models[BaseChatModel / Chat Models]
    BaseTools[BaseTool]
    MessagesCore[BaseMessage / System / Human / AI]
  end

  UC1 --> BaseTools
  UC2 --> MWStack
  UC3 --> CF

  Models --> CF
  BaseTools --> CF
  MessagesCore --> CF

  CF --> SG
  MWStack --> CF
  SO --> CF
  ToolsBinding --> CF

  SG --> Channels --> Pregel
  Pregel --> Checkpointer

核心結論

  • LangGraph 是 狀態機 + 工作流執行引擎
  • LangChain v1 的 Agent 創建其實就是:「幫你定義了一個特定拓撲的 StateGraph + 一組中間件堆疊」
  • 你可以:
    • 完全只用 LangGraph(自己畫圖/定義所有節點)
    • 或先用 LangChain v1 的 create_agent() 生出一個 Agent Graph,再把這個 Agent 當成 子圖 放進你自己的 LangGraph workflow

5. 實戰視角:什麼時候用 LangGraph?什麼時候用 LangChain v1?

5.1 只用 LangChain v1 就夠的情境

適用條件:

  • 單 Agent,或少量 Agent
  • 流程是典型的 ReAct:LLM + 工具 + 記憶
  • 需求集中在:
    • 模型重試/回退、調用限制
    • 工具重試/限制
    • PII 處理
    • 人在迴圈
    • 結構化輸出
  • 不需要複雜的自訂 graph 拓撲

範例(你已在附件中提供):

  • create_agent() + ModelRetryMiddleware + ToolCallLimitMiddleware + HumanInTheLoopMiddleware
  • checkpointerstore 管理會話與跨會話狀態

5.2 需要直接使用 LangGraph 的情境

適用條件:

  1. 多 Agent 複雜拓撲
  • Supervisor / Network / Hierarchical / Handoffs
  • 多角色/多子系統互動
  1. 需要精細控制執行流程
  • 自訂節點拓撲+條件邊(conditional edges)
  • 自訂 retry policy per node / per graph
  • 自訂 streaming 行為(例如自定義 progress 事件)
  1. 重度的持久化與時間旅行需求
  • 要能隨時 inspect 中間狀態、回滾到某個 checkpoint
  • 跨多個 workflow 的統一狀態管理
  1. 希望用 Functional API 寫 workflow
  • 使用 @task / @entrypoint,像寫普通 Python 函數一樣,但擁有 checkpoint + interrupt。

5.3 混合使用:用 LangChain v1 的 Agent 當 LangGraph 子圖

一個常見 pattern 是:

  • create_agent() 生成一個有工具 + 中間件的「專家 Agent」
  • 在 LangGraph 的 StateGraph 中,將這個 Agent 當成一個節點/子圖

偽碼示例(概念):

expert_agent = create_agent(
    model="anthropic:claude-sonnet-4-5-20250929",
    tools=[...],
    middleware=[ModelRetryMiddleware(max_retries=3)]
)

builder = StateGraph(GlobalState)
builder.add_node("router", router_node)
builder.add_node("expert_workflow", expert_agent)  # Agent 當 node/subgraph
builder.add_node("another_workflow", another_subgraph)
...
graph = builder.compile(checkpointer=...)

這樣:

  • LangGraph 管整體 workflow(路由、協作、多 Agent)
  • LangChain v1 管單一 Agent 的「模型/工具/中間件」細節

6. 實作建議與架構選型指南

6.1 開發階段建議

  1. 先用 LangChain v1 create_agent() 起步
  • 快速把 LLM + Tools + Middleware + Checkpoint + Structured Output 拉起來
  • 先解決「功能」與「可靠性」(重試、回退、限制、人審)
  1. 當流程變複雜,再引入 LangGraph
  • 把既有 Agent 改寫成 LangGraph 的節點或子圖
  • 逐步用 StateGraph 描述完整業務流程(例如多階段 pipeline、多角色協作)
  1. 中間件重用
  • 把橫切 concern(logging、metrics、安全審查)實作成 AgentMiddleware
  • 在不同 Agent 上重複使用

6.2 典型架構演進路徑

flowchart LR
  A["階段 1<br/>單 Agent + 工具"] --> B["階段 2<br/>多工具 + 中間件"]
  B --> C["階段 3<br/>多 Agent + LangGraph"]
  C --> D["階段 4<br/>自訂 Channels / Checkpointer"]
  • 階段 1:簡單聊天+一兩個工具
  • 階段 2:加入中間件、結構化輸出、持久化
  • 階段 3:多 Agent + 複雜 workflow,開始直接寫 StateGraph
  • 階段 4:自訂 Channel / Checkpointer,針對特殊需求優化(例如高頻交易、超大狀態)

7. 結語:LangChain v1 + LangGraph 的真正價值

總結一下兩者的關係與價值:

  • LangGraph

    • 是「有狀態 AI workflow 的引擎」:圖、通道、BSP、checkpoint、human-in-the-loop、多 Agent
    • 適合當你的「AI 作業系統」
  • LangChain v1

    • 是「建立在 LangGraph 上的高階 Agent 框架」
    • 把常見的 Agent 模式(ReAct、工具調用、結構化輸出、中間件)封裝為一行 API create_agent()
    • 讓你在不用自己畫圖的情況下,也能直接享受 LangGraph 的能力(持久化、流式、人在迴圈)

對工程團隊來說:

  • 若你現在還停留在「直接調 LLM API」,那麼:
    • LangChain v1 是你走向「可觀測、可管控 Agent」的第一步
  • 若你已經在做「多 Agent、複雜 workflow」,那麼:
    • LangGraph 是你需要的底層引擎,而 LangChain v1 則是「快速組裝單一 Agent」的加速器