LangChain v1 與 LangGraph:新一代 Agent 編排架構深度解析
0. 為什麼要再看一次 LangChain?
早期的 LangChain 給人的印象是「把 LLM + 工具 + 記憶串起來的便利框架」。但從 LangChain v1 開始,官方實際上做了兩個重要重構:
- 把「編排/工作流」能力沉到底層,變成獨立產品 LangGraph
- 把「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 Pregel 的 Bulk 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
...
)
一句話總結
「讀上一超步」= 所有節點讀同一份快照,寫入各自緩衝,超步結束後統一合併
這樣無論節點以什麼順序執行、在哪個線程執行,結果都是確定的。
- Update
- 收集節點產生的「通道寫入」,透過 [channel.update()] 聚合到通道中
- 通道可有多種聚合策略(LastValue / BinaryOperatorAggregate / Topic 等)
- 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
)
它實際做的事情:
- 根據你提供的 model / tools / system_prompt 建立一個 LangGraph StateGraph
- 注入一個預設的「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
- 把你設定的 middleware 一層一層包在:
- [before_agent / before_model / after_model / after_agent] 生命週期 hook
- [wrap_model_call] 模型呼叫包裝
- [wrap_tool_call] 工具呼叫包裝
- 最後 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 提供兩大類介面:
- 生命週期鉤子(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 等
- 攔截鉤子(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 無限用同一工具 |
LLMToolSelectorMiddleware | LLM 工具選擇 | 多工具選擇最佳一個/多個 |
LLMToolEmulator | 工具模擬 | 在沒有真工具時做 dry-run |
HumanInTheLoopMiddleware | 人工確認 | Human approval / 修改 |
ShellToolMiddleware | Shell 執行 | 代碼執行、DevOps 操作 |
FilesystemFileSearchMiddleware | 文件搜尋 | 本地知識庫檢索 |
PIIMiddleware | PII 檢測 | 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 驗證
- 如果 LLM 提供原生結構化 API,例如 OpenAI
-
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- 用
checkpointer與store管理會話與跨會話狀態
5.2 需要直接使用 LangGraph 的情境
適用條件:
- 多 Agent 複雜拓撲:
- Supervisor / Network / Hierarchical / Handoffs
- 多角色/多子系統互動
- 需要精細控制執行流程:
- 自訂節點拓撲+條件邊(conditional edges)
- 自訂 retry policy per node / per graph
- 自訂 streaming 行為(例如自定義 progress 事件)
- 重度的持久化與時間旅行需求:
- 要能隨時 inspect 中間狀態、回滾到某個 checkpoint
- 跨多個 workflow 的統一狀態管理
- 希望用 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 開發階段建議
- 先用 LangChain v1
create_agent()起步
- 快速把 LLM + Tools + Middleware + Checkpoint + Structured Output 拉起來
- 先解決「功能」與「可靠性」(重試、回退、限制、人審)
- 當流程變複雜,再引入 LangGraph
- 把既有 Agent 改寫成 LangGraph 的節點或子圖
- 逐步用 StateGraph 描述完整業務流程(例如多階段 pipeline、多角色協作)
- 中間件重用
- 把橫切 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」的加速器