LangGraph 入門：用圖結構構建你的第一個多智能體工作流

LangGraph 設計的一個核心是：多智能體工作流本質上是圖結構，而非線性鏈。早期 LLM 應用普遍采用"提示 → LLM → 響應"的線性模式，但這種架構難以應對真實智能體系統的復雜性。比如生產環境中的多智能體協作需要分支（基于數據選擇不同執行路徑）、循環（支持重試與迭代優化）、匯合（多個智能體向共享狀態寫入數據），以及條件路由（根據執行結果動態決定后續流程）。

LangGraph 如何表示工作流

LangGraph 里每個工作流都是一個 StateGraph——本質上是有向圖。節點就是智能體，或者說處理狀態的函數；邊是智能體之間的轉換；狀態則是在整個圖中流動的共享數據結構。

from langgraph.graph import StateGraph, END
from typing import TypedDict
# Define your state schema
class IncidentState(TypedDict):
incident_id: str
current_metrics: dict
proposed_solution: dict
issue_resolved: bool
retry_count: int
# Create the graph
workflow = StateGraph(IncidentState)
# Add agent nodes
workflow.add_node("diagnose", diagnose_agent)
workflow.add_node("plan_fix", planning_agent)
workflow.add_node("execute_fix", worker_agent)
workflow.add_node("verify", verification_agent)
# Define transitions
workflow.add_edge("diagnose", "plan_fix")
workflow.add_edge("plan_fix", "execute_fix")
workflow.add_edge("execute_fix", "verify")
# Conditional: retry or exit
workflow.add_conditional_edges(
"verify",
lambda state: "resolved" if state["issue_resolved"] else "retry",
{
"resolved": END,
"retry": "diagnose" # Loop back
}
)
workflow.set_entry_point("diagnose")

這樣做的好處非常明顯：圖本身就可以當作開發文檔文檔，一眼能看懂流程；加減節點不用動協調邏輯；狀態有類型約束；循環有內置的終止條件，不會跑成死循環。

節點、邊、狀態三者各司其職。節點封裝具體的邏輯操作，只管做事；邊定義節點間怎么交互、誰先誰后；狀態承載共享上下文，讓節點可以保持無狀態。這種職責分離讓系統好理解、好調試、好擴展，節點還能跨工作流復用。

運行時到底發生了什么

圖定義是聲明式的，但真正讓編排變得有意義的是運行時行為。

工作流啟動后，LangGraph 用狀態機來管理執行。首先從入口節點的初始狀態開始，然后調用智能體函數并傳入當前狀態。智能體返回的是增量更新而不是整個狀態的替換，LangGraph 拿到更新后原子性地合并到當前狀態，接著根據圖定義決定下一個節點，同時創建檢查點把當前狀態和執行位置持久化下來。這個過程一直重復，直到走到 END 節點或者達到最大迭代次數。

有一點很關鍵：智能體永遠不會直接改共享狀態。它們拿到的是只讀副本，算完之后返回更新，實際的狀態修改由 LangGraph 來做，可以保證了原子性和一致性。

邊遍歷機制

邊定義了哪些轉換是允許的，但具體什么時候轉換由運行時決定。

靜態邊沒什么花樣：

workflow.add_edge("diagnose", "plan_fix")

diagnose 節點跑完、檢查點創建好之后，LangGraph 立刻拿更新后的狀態去調 plan_fix。

條件邊就靈活多了：

workflow.add_conditional_edges(
"verify",
route_function,
{"retry": "diagnose", "resolved": END}
)

verify 完成后，LangGraph 調用 route_function(state) 來判斷下一步走哪條邊。函數返回 retry 就回到 diagnose，返回 resolved 就結束。

任何節點在執行前它的所有前置節點必須已經完成并創建了檢查點，這就避免了 Pub/Sub 系統里常見的那種"前面還沒跑完后面就開始了"的問題。

狀態管理的特殊之處

LangGraph 的狀態跟傳統系統不太一樣。

它不是存在 Redis 或數據庫里讓智能體直接訪問的共享內存。LangGraph 在內部維護狀態，給智能體的是受控訪問。對智能體來說狀態是不可變的——拿到的是快照，不能直接改，只能返回想要的變更。

多個智能體并行跑的時候（通過并行邊），LangGraph 收集所有更新，用 reducer 原子性地一起應用。讀-修改-寫的競態條件就這么解決了。

每個檢查點還會創建一個狀態版本。想看執行歷史中任意時刻的狀態？直接查檢查點就行，這就是所謂的時間旅行調試。

檢查點持久化

檢查點不只是日志，它們是恢復點。

每個檢查點記錄完整的狀態快照、當前在圖中的位置（剛執行完哪個節點）、還有元數據（時間戳、創建檢查點的節點、執行路徑）。

創建時機有三個：每個節點成功完成后、條件邊評估前、以及工作流暫停時（比如等人工審批）。

這樣如果節點執行到一半崩了，可以從最后一個檢查點重試就行；長時間運行的工作流可以暫停再恢復，進度不會丟；調試的時候能從任意檢查點開始重放。

一個完整的運行時示例

假設用戶發起請求："修復服務延遲問題"。

T0: Workflow starts
- Initial state: {incident_id: "INC-123", retry_count: 0}
- Entry point: "diagnose"
T1: "diagnose" node executes
- Receives: {incident_id: "INC-123", retry_count: 0}
- Agent calls Data Agent, fetches metrics
- Returns: {current_metrics: {cpu: 95, latency: 500ms}}
- LangGraph merges: state now has metrics
- Checkpoint created
T2: Static edge triggers: "diagnose" → "plan_fix"
- "plan_fix" node executes
- Receives merged state (incident_id + retry_count + current_metrics)
- Agent calls Knowledge Agent for runbook
- Returns: {proposed_solution: "restart_service"}
- LangGraph merges
- Checkpoint created
T3: Static edge triggers: "plan_fix" → "execute_fix"
- "execute_fix" node executes
- Calls Worker Agent
- Returns: {action_status: "completed"}
- Checkpoint created
T4: Static edge triggers: "execute_fix" → "verify"
- "verify" node executes
- Calls Data Agent again
- Returns: {current_metrics: {cpu: 90, latency: 480ms}, issue_resolved: false}
- Checkpoint created
T5: Conditional edge evaluation
- LangGraph calls route function with current state
- route_function checks: state["issue_resolved"] == false and retry_count < 3
- Returns: "retry"
- LangGraph increments retry_count
- Routes back to "diagnose" (cycle)
T6: "diagnose" executes again (retry #1)
- Process repeats with updated state...

狀態在節點間累積——指標、方案、操作結果都在里面。每個節點都能看到之前所有節點產出的完整信息。重試邏輯是圖結構強制的，不是寫在智能體代碼里。出了故障檢查點可以讓程序隨時恢復運行。

用 LangGraph 的話，智能體只管返回自己的更新。協調、狀態合并、路由、持久化，運行時全包了。

關鍵架構模式

傳統多智能體系統喜歡累積對話歷史：

# Common pattern - append-only log
messages = [
{"role": "user", "content": "Service X is slow"},
{"role": "data", "content": "CPU at 95%"},
{"role": "knowledge", "content": "Try restarting"},
{"role": "action", "content": "Restarted service"},
...
]

這東西會無限增長，智能體每次都得在歷史里翻來翻去找有用的數據。

LangGraph 換了個思路，狀態就是當前世界的快照：

class State(TypedDict):
# Current values, not history
incident_id: str
current_cpu: float
recommended_action: str
action_status: str
retry_count: int

智能體讀當前值、更新當前值。歷史通過檢查點單獨維護，調試用得著，但工作狀態保持精簡。訪問狀態 O(1)，不用解析歷史；數據所有權清晰，一眼看出哪個字段歸誰管；推理也簡單，當前狀態是啥就是啥。

Reducer 解決并行協調

多個智能體要往同一個狀態字段寫數據怎么辦？LangGraph 提供 reducer——專門合并并發更新的函數。

傳統 A2A 模型里，智能體得自己搞協調：搶鎖、讀-修改-寫、重試、沖突檢測。這套東西各團隊實現得五花八門，一旦出現部分故障就容易出問題。Reducer 把沖突解決挪到編排層，智能體級別的協調邏輯直接省掉。

比如說下面的例子，三個監控智能體并行檢查不同的服務副本：

from typing import Annotated
from operator import add
class State(TypedDict):
# Reducer: combine all health check results
health_checks: Annotated[list, add]

三個 Data Agent 各自返回健康檢查結果，reducer（這里就是列表的 add 操作）自動把三份結果合成一個列表。沒有智能體需要知道其他智能體的存在，不用搶鎖，不用協調更新。

沒有 reducer 的話，需要手動加鎖防覆蓋、寫協調邏輯合并結果、還得擔心更新丟失。有了 reducer，編排層自動處理。

檢查點用于調試和恢復

每次節點執行都會創建檢查點，狀態和執行位置的快照會持久化到 Postgres、Redis 或文件系統。

生產環境出故障了？可以檢查檢查點的內容，看看每個智能體觀察到了什么、做了什么決定。這相當于給智能體工作流裝了黑匣子，決策鏈條一清二楚。

服務器中途崩了也可以從最后一個檢查點恢復，不用從頭來。對那些要調用昂貴 API 或者收集大量數據的長時間任務來說，這太重要了。

而且工作流可以暫停幾小時甚至幾天，狀態通過檢查點保持現有狀態，從暫停的地方精確恢復，上下文完整保留。

修改工作流的靈活性

LangGraph的另外一個賣點是工作流改起來容易。

假設初始工作流是 Diagnose → Fix → Verify，現在要加個需求："修復之前先查一下 Jira 有沒有已知問題"。

代碼改動就這么點：

# Add the new agent
workflow.add_node("check_jira", jira_agent)
# Rewire the flow
workflow.add_edge("diagnose", "check_jira") # New path
workflow.add_conditional_edges(
"check_jira",
lambda state: "known_issue" if state["jira_ticket"] else "unknown",
{
"known_issue": "apply_known_fix", # New path
"unknown": "plan_fix" # Original path
}
)

單個智能體的實現不用動，狀態協調邏輯不用動，檢查點處理不用動，錯誤恢復不用動。

如果換成換成 Pub/Sub 呢？事件路由邏輯要改，完成跟蹤要改（現在是 4 個智能體不是 3 個了），狀態模式協調要改，所有集成點都得重新測。

再看重試邏輯的修改。原來是最多重試 3 次：

# Before
workflow.add_conditional_edges(
"verify",
lambda state: "retry" if state["retry_count"] < 3 else "end",
{"retry": "diagnose", "end": END}
)

新需求："只有臨時性錯誤（網絡問題）才重試，永久性錯誤（配置問題）不重試"。改條件函數就行：

# After - just change the condition function
def should_retry(state):
if state["issue_resolved"]:
return "success"
if state["error_type"] == "config":
return "escalate" # Don't retry config errors
if state["retry_count"] >= 3:
return "max_retries"
return "retry"
workflow.add_conditional_edges(
"verify",
should_retry,
{
"success": END,
"retry": "diagnose",
"escalate": "human_review",
"max_retries": "alert_team"
}
)

業務邏輯在工作流結構里一目了然，改起來也順手。

LangGraph 支持的典型模式

生成的方案不夠好，可以直接加個循環：

workflow.add_node("generate_solution", llm_agent)
workflow.add_node("validate_solution", validation_agent)
workflow.add_node("refine_solution", refinement_agent)
workflow.add_conditional_edges(
"validate_solution",
lambda state: "valid" if state["solution_quality"] > 0.8 else "refine",
{
"valid": "execute_fix",
"refine": "refine_solution"
}
)
workflow.add_edge("refine_solution", "generate_solution") # Loop back

方案不斷迭代，直到質量達標。

并行信息收集時需要同時從多個來源拉數據：

from langgraph.graph import START
# Parallel nodes
workflow.add_node("fetch_metrics", data_agent)
workflow.add_node("fetch_logs", elasticsearch_agent)
workflow.add_node("fetch_config", knowledge_agent)
# All start in parallel
workflow.add_edge(START, "fetch_metrics")
workflow.add_edge(START, "fetch_logs")
workflow.add_edge(START, "fetch_config")
# All must complete before analysis
workflow.add_node("analyze", analysis_agent)
workflow.add_edge("fetch_metrics", "analyze")
workflow.add_edge("fetch_logs", "analyze")
workflow.add_edge("fetch_config", "analyze")

LangGraph 保證 analyze 節點在三個數據源都拿完之后才開始跑。

高風險操作需要人來進行確認：

workflow.add_node("propose_fix", planning_agent)
workflow.add_node("await_approval", approval_gate)
workflow.add_node("execute_fix", action_agent)
workflow.add_edge("propose_fix", "await_approval")
# Workflow pauses at await_approval
# State is persisted
# When human approves, workflow resumes
workflow.add_conditional_edges(
"await_approval",
lambda state: "approved" if state["human_approved"] else "rejected",
{
"approved": "execute_fix",
"rejected": "propose_alternative"
}
)

這個確認過程可以等幾小時甚至幾天，不消耗任何的資源。

什么場景適合 LangGraph

復雜工作流（5 個以上智能體、有條件邏輯、有循環）、業務邏輯經常變、需要事后調試分析、有人工審批或質量門控、長時間任務需要崩潰恢復——這些場景 LangGraph 很合適。

簡單的線性流程（A → B → C，沒分支）、智能體完全獨立不需要協調、對延遲極度敏感（編排開銷要控制在 10ms 以內）、或者團隊有深厚的分布式系統功底想自己搞狀態機——這些場景替代方案也挺好。

總結

編排框架在復雜系統中的價值已經被反復驗證：Kubernetes 之于容器、Airflow 之于數據管道、Temporal 之于通用工作流。LangGraph 將同樣的理念帶入多智能體 AI 領域，提供了 LLM 感知的編排能力。

其核心價值在于：圖結構讓工作流易于修改和擴展，檢查點機制保障了可調試性和故障恢復，reducer 和原子狀態更新解決了并行協調難題。開發者可以專注于智能體邏輯本身，而非協調管道的實現細節。

對于正在構建多智能體系統的團隊，LangGraph 提供了一條從實驗原型到生產系統的可行路徑。

https://avoid.overfit.cn/post/207f7dd3b4b2488983645d365c9e0b89

作者：ravikiran veldanda