AG Day 11 LangGraph 起步:StateGraph、節點與邊
執行需求:CPU 可跑。本篇文章聚焦於 LangGraph 1.x 的核心抽象與圖編譯機制,所有範例均在本地 CPU 環境下執行確定性狀態躍遷,不需要外部 API 金鑰即可完成完整的狀態圖建置、編譯、單步追蹤與視覺化驗證。
引言
在上一篇 AG Day 10(原文連結)中,我們深入剖析了手刻代理迴圈在狀態爆炸、分支控制、持久化恢復與人機互動等面向的本質瓶頸,並親手實作了一個微型圖排程器,驗證了「純函式節點+集中狀態機」相較於傳統 while 迴圈的壓倒性架構優勢。今天,我們正式邁入本系列的核心編排技術——LangGraph 1.x。
LangGraph 是 LangChain 生態系在經歷數年工程實踐後,專門為「週期性、長生命週期、狀態驅動」的 AI Agent 所打造的現代分散式圖編排框架。不同於傳統以單向無環圖(DAG)為假設的管線引擎,LangGraph 將「迴圈(Cycles)」與「可控狀態轉移」提升為系統架構的第一公民。它拋棄了複雜晦澀的黑盒子抽象,以最純粹的 Python 型別標註(TypedDict)為基底,提供了一套嚴謹而優雅的狀態機建置語法。
在本文中,我們將為貫穿專案 research-agent 引進 LangGraph 1.x 核心庫,並在 src/research_agent/graph.py 中親手搭建第一個專業的狀態圖。讀者將徹底掌握 StateGraph 的架構藍圖、START 與 END 特殊節點、純函式節點的撰寫準則、確定型邊的連接方式,以及如何透過編譯後的圖引擎進行單步串流除錯與 Mermaid 圖表視覺化輸出。
LangGraph 核心抽象:圖狀態機的三位一體
要駕馭 LangGraph,首先必須在腦海中建立清晰的抽象模型。LangGraph 的架構可以精煉為「三位一體」:狀態(State)、節點(Nodes)與邊(Edges):
- 狀態(State):圖的「單一真實來源(Single Source of Truth)」。在 LangGraph 中,狀態通常繼承自標準庫的
typing.TypedDict。它定義了圖在運作時所有節點可以讀取與寫入的欄位鍵值及其型別規格。狀態在整個圖的生命週期中集中維護,節點之間不進行直接的參數私下傳遞,所有的資訊流動皆透明地體現於狀態的躍遷之中。 - 節點(Nodes):圖的「計算單元」。每個節點本質上都是一個獨立的 Python 函式。它的輸入簽章接收當前完整的系統狀態(
state: State),內部執行特定的計算(如呼叫模型、讀寫資料庫或執行數學運算),並回傳一個包含局部更新的字典。LangGraph 會依據狀態架構自動將該回傳字典與既有狀態進行合併。 - 邊(Edges):圖的「控制流導軌」。邊定義了節點與節點之間的流轉順序。LangGraph 1.x 提供了兩大類邊:固定指向邊(Normal Edges,如從節點 A 無條件轉移至節點 B)與條件邊(Conditional Edges,依據動態評估函式決定分支)。此外,LangGraph 內建了兩個常數標記:
START(代表整個圖的進入點)與END(代表圖執行的終止匯流點)。
這三者透過 StateGraph 容器類別進行組合宣告,最後透過 .compile() 方法編譯成一個可呼叫的執行個體(CompiledGraph)。編譯過程會進行靜態拓撲檢查,確保沒有孤立節點或無法抵達的死路,為動態語言 Python 帶來了編譯期的架構安全保障。
生命週期剖析:從宣告、編譯到呼叫
理解 LangGraph 執行個體的生命週期,對於撰寫高效能的生產環境程式碼至關重要。其生命週期包含三個關鍵階段:
- 圖構建階段(Graph Definition):宣告
builder = StateGraph(MyState),並反覆呼叫builder.add_node()與builder.add_edge()。在此階段,所有的函式只是以參照形式被註冊進拓撲對照表中,沒有任何節點會被實際呼叫。 - 圖編譯階段(Compilation):呼叫
app = builder.compile()。編譯器會驗證進入點是否已正確連接至START、檢查所有邊所指向的節點名稱是否存在、並建構底層的 Pregel 排程執行引擎。編譯是一個具備確定性的運算,在實務架構中,圖應該在模組啟動時編譯一次並全域複用,切忌在每次 HTTP 請求中重複編譯。 - 圖執行階段(Invocation & Streaming):透過
app.invoke(inputs)或app.stream(inputs)啟動圖。排程器將初始輸入灌入狀態架構中,從START邊出發,依序驅動對應的節點函式,直到狀態流轉至END或達到步數限制為止。
完整實作:為 research-agent 打造三階段研究管線圖
現在,我們進入具體工程實作。我們將在 research-agent 專案的 src/research_agent/graph.py 中,建立一個結構清晰的三階段研究助理狀態圖。該圖依序包含:擬定檢索規劃(Planner)、收集文獻佐證(Evidence Gatherer)以及整合研究報告(Synthesizer)。
第一步,首先確認專案依賴。我們在 pyproject.toml 中加入 langgraph>=0.2.0 與 langchain-core,並透過 uv 工具鏈進行環境同步:
# 安裝 LangGraph 核心套件
uv add "langgraph>=0.2.0" "langchain-core>=0.3.0"
第二步,在 src/research_agent/graph.py 中定義狀態結構。我們利用標準 TypedDict 宣告包含主題、規劃串列、佐證紀錄與報告草稿的核心欄位:
"""src/research_agent/graph.py:定義研究助理核心狀態架構。"""
from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, START, END
class ResearchState(TypedDict):
"""研究助理在圖中流轉的核心狀態架構。"""
topic: str
plan: List[str]
evidence_items: List[str]
report_draft: Optional[str]
step_count: int
第三步,實作三個獨立的純函式節點。每個節點接收完整的 ResearchState,並嚴格遵循無副作用原則,只回傳屬於自己的增量字典:
"""定義圖狀態機的三個純函式節點。"""
def plan_research_node(state: ResearchState) -> dict:
"""階段一:拆解研究主題,擬定檢索規劃細項。"""
topic = state.get("topic", "")
print(f"[節點: plan_research] 正在為主題 '{topic}' 規劃檢索方向...")
generated_plan = [
f"{topic} 的底層運作原理",
f"{topic} 的架構限制與常見踩雷",
f"{topic} 的生產環境最佳化策略",
]
# 回傳增量字典,僅包含本次變更的欄位
return {
"plan": generated_plan,
"step_count": state.get("step_count", 0) + 1,
}
def gather_evidence_node(state: ResearchState) -> dict:
"""階段二:依據規劃收集客觀文獻證據與技術論點。"""
plan = state.get("plan", [])
print(f"[節點: gather_evidence] 正在依據 {len(plan)} 項規劃收集文獻佐證...")
gathered = []
for item in plan:
gathered.append(f"【佐證紀錄】已確認 '{item}' 之實測數據與架構規範。")
return {
"evidence_items": gathered,
"step_count": state.get("step_count", 0) + 1,
}
def synthesize_report_node(state: ResearchState) -> dict:
"""階段三:彙整所有佐證資料,產出整合研究報告草稿。"""
topic = state.get("topic", "")
evidence = state.get("evidence_items", [])
print(f"[節點: synthesize_report] 正在彙整 {len(evidence)} 筆佐證產出最終摘要...")
draft_lines = [
f"# 研究專題報告:{topic}",
f"總計提煉論點佐證:{len(evidence)} 項。",
"## 核心技術摘要",
]
draft_lines.extend([f"- {e}" for e in evidence])
draft_lines.append("## 結論\n本研究架構在離線與生產環境中皆展現高度確定性。")
return {
"report_draft": "\n".join(draft_lines),
"step_count": state.get("step_count", 0) + 1,
}
第四步,建立圖組裝函式 build_research_graph()。我們使用 StateGraph 註冊節點,並利用 START 與 END 常數精確定義拓撲流向:
"""組裝 StateGraph 拓撲結構並完成圖編譯。"""
def build_research_graph():
"""組裝節點與固定邊,並編譯為可執行的應用實體。"""
# 1. 以狀態架構初始化 StateGraph
workflow = StateGraph(ResearchState)
# 2. 註冊三個核心運算節點
workflow.add_node("planner", plan_research_node)
workflow.add_node("gatherer", gather_evidence_node)
workflow.add_node("synthesizer", synthesize_report_node)
# 3. 使用 LangGraph 1.x 標準常數建立固定指向邊
# 從起點 START 進入 planner
workflow.add_edge(START, "planner")
# planner 執行完畢後固定流向 gatherer
workflow.add_edge("planner", "gatherer")
# gatherer 執行完畢後固定流向 synthesizer
workflow.add_edge("gatherer", "synthesizer")
# synthesizer 執行完畢後導向終點 END
workflow.add_edge("synthesizer", END)
# 4. 編譯狀態圖拓撲
app = workflow.compile()
return app
第五步,實作執行入口並示範 LangGraph 強大的「串流追蹤(Streaming)」與「Mermaid 圖表導出」功能。這能讓開發者清晰檢視每一步狀態字典的增量躍遷:
"""執行狀態圖並展示串流步階日誌與拓撲圖。"""
from src.research_agent.graph import build_research_graph
app = build_research_graph()
# 導出 Mermaid 語法字串,可直接貼至 Markdown 預覽
mermaid_code = app.get_graph().draw_mermaid()
print("=== 狀態圖 Mermaid 拓撲定義 ===")
print(mermaid_code[:120], "...(省略後續語法)")
print("=" * 32)
# 初始輸入狀態
initial_inputs = {
"topic": "LangGraph 狀態機與圖論編排",
"plan": [],
"evidence_items": [],
"report_draft": None,
"step_count": 0,
}
print("\n=== 開始執行圖串流躍遷 ===")
# 使用 .stream() 逐步觀察每個節點的輸出增量
for step_output in app.stream(initial_inputs):
for node_name, node_delta in step_output.items():
print(f"\n--> 完成節點 [{node_name}]:")
for k, v in node_delta.items():
if isinstance(v, list):
print(f" {k}: 清單共 {len(v)} 筆資料")
elif isinstance(v, str) and len(v) > 60:
print(f" {k}: {v[:50]}...(長文字省略)")
else:
print(f" {k}: {v}")
# 輸出(範例輸出):
# === 狀態圖 Mermaid 拓撲定義 ===
# %%{init: {'flowchart': {'curve': 'linear'}}}%% ...(省略後續語法)
# ================================
#
# === 開始執行圖串流躍遷 ===
# [節點: plan_research] 正在為主題 'LangGraph 狀態機與圖論編排' 規劃檢索方向...
# --> 完成節點 [planner]:
# plan: 清單共 3 筆資料
# step_count: 1
# [節點: gather_evidence] 正在依據 3 項規劃收集文獻佐證...
# --> 完成節點 [gatherer]:
# evidence_items: 清單共 3 筆資料
# step_count: 2
# [節點: synthesize_report] 正在彙整 3 筆佐證產出最終摘要...
# --> 完成節點 [synthesizer]:
# report_draft: # 研究專題報告:LangGraph 狀態機與圖論編排
總計提煉論點佐證:3 項。...(長文字省略)
# step_count: 3
第六步,展示一次性呼叫(.invoke())並檢驗最終狀態物件,確認完整資料皆被乾淨收斂:
"""示範 invoke 一次性取得終端狀態。"""
final_state = app.invoke({
"topic": "LangGraph 檢查點機制",
"plan": [],
"evidence_items": [],
"report_draft": None,
"step_count": 0,
})
print("\n=== 最終成果報告 ===")
print(final_state["report_draft"])
print(f"總執行步數累計: {final_state['step_count']}")
# 輸出(範例輸出):
# === 最終成果報告 ===
# # 研究專題報告:LangGraph 檢查點機制
# 總計提煉論點佐證:3 項。
# ## 核心技術摘要
# - 【佐證紀錄】已確認 'LangGraph 檢查點機制的底層運作原理' 之實測數據與架構規範。
# - 【佐證紀錄】已確認 'LangGraph 檢查點機制的架構限制與常見踩雷' 之實測數據與架構規範。
# - 【佐證紀錄】已確認 'LangGraph 檢查點機制的生產環境最佳化策略' 之實測數據與架構規範。
# ## 結論
# 本研究架構在離線與生產環境中皆展現高度確定性。
# 總執行步數累計: 3
第七步,我們為 src/research_agent/graph.py 撰寫嚴謹的自動化單元測試 tests/test_graph_basics.py,透過 pytest 驗證圖拓撲結構、節點覆蓋度與終端狀態欄位完整性:
"""tests/test_graph_basics.py:驗證 LangGraph 基礎狀態圖功能。"""
import pytest
from src.research_agent.graph import build_research_graph
def test_compiled_graph_nodes_presence():
"""測試編譯後的圖結構包含所有宣告的節點。"""
app = build_research_graph()
graph_repr = app.get_graph()
# 檢查節點清單
node_names = set(graph_repr.nodes.keys())
assert "planner" in node_names
assert "gatherer" in node_names
assert "synthesizer" in node_names
def test_research_graph_end_to_end_execution():
"""測試端到端執行能夠正確走完三階段並產出報告。"""
app = build_research_graph()
result = app.invoke({
"topic": "自動化單元測試專題",
"plan": [],
"evidence_items": [],
"report_draft": None,
"step_count": 0,
})
assert len(result["plan"]) == 3
assert len(result["evidence_items"]) == 3
assert result["report_draft"] is not None
assert "研究專題報告:自動化單元測試專題" in result["report_draft"]
assert result["step_count"] == 3
讀者可以在終端機透過以下指令執行單元測試以確認所有測試案例皆順利通過:
# 執行 LangGraph 基礎拓撲單元測試
uv run pytest tests/test_graph_basics.py -v
常見錯誤與踩雷
初學者在接觸 LangGraph 1.x 時,極常在以下四個語法與觀念細節上踩雷:
- 混淆 0.x 與 1.x 的進入點語法:在 LangGraph 早期版本中,設定起點採用的是
workflow.set_entry_point("node_name")。到了 LangGraph 1.x,官方統一了設計哲學,引進了虛擬起點常數START,標準寫法全面改為workflow.add_edge(START, "node_name")。雖然舊語法目前大多仍具備向下相容性,但若在新專案中混用會造成拓撲視覺化圖表語意不一致。 - 節點函式誤回傳完整狀態或就地修改物件:LangGraph 的節點運算子設計原則是「增量回傳(Delta Update)」。例如節點只負責更新
plan,就應該僅回傳{"plan": new_plan}。如果初學者在節點函式中寫下state["plan"] = new_plan; return state,不僅破壞了函式的純粹性,更會在大規模平行分支或持久化回溯時引發難以追蹤的競態條件(Race Conditions)。 - 未呼叫
.compile()便直接執行:StateGraph只是圖的結構建構器(Builder),本身並不具備.invoke()或.stream()方法。許多初學者忘記執行app = workflow.compile(),直接嘗試呼叫workflow.invoke(),導致直譯器直接拋出AttributeError: 'StateGraph' object has no attribute 'invoke'。 - 節點回傳的鍵值未在 TypedDict 狀態中宣告:若節點函式回傳了一個不在
ResearchState宣告範圍內的欄位(例如拼寫錯誤{"paln": [...]}),在預設情況下雖然 Python 字典允許合併,但靜態型別檢查器(如 Mypy / Pyright)會直接發出嚴厲警告,且後續依賴該鍵值的下游節點將無法透過型別提示取得程式碼自動補全。
效能與實務提醒
在生產架構中引進 LangGraph,我們應該掌握以下幾項重要的效能心法:
第一點是編譯執行個體的單例化複用(Singleton Reuse)。workflow.compile() 內部會進行完整的拓撲排序、邊界驗證與執行管線初始化。這段計算雖然僅需數毫秒,但若被錯誤地放置在 API 請求處理函式內部(例如在每個 FastAPI 端點中重複 new 一個 StateGraph 並 compile),在高並行請求下將造成不必要的 CPU 負載。最佳實務是將編譯好的 app 物件宣告在模組層級或依賴注入容器中,作為長駐的單例執行個體重複使用。
第二點是善用 .stream() 進行漸進式除錯與使用者體驗最佳化。在傳統 .invoke() 中,系統必須等待所有節點全部執行完畢才一次性回傳結果,這會讓終端使用者面臨長達數秒的白畫面等待。透過 app.stream(),前端介面可以在 planner 完成時立即印出檢索方向、在 gatherer 完成時即時更新進度條,大幅改善人機互動的即時感與系統透明度。
第三點是無狀態與有狀態的邊界分離。今天我們建置的圖尚未掛載持久化檢查點(Checkpointer),它是一個純記憶體、無狀態(Stateless)的運算圖。無狀態圖的優勢在於輕巧、水平擴充極為容易,非常適合用於單次確定型的分析管線。但在需要中斷恢復或人機核准的場景中,我們將需要導入持久化狀態,這將在未來的篇章中逐步解鎖。
小結
今天我們正式邁入了 LangGraph 1.x 的世界。我們釐清了 StateGraph、START、END、純函式節點與固定邊的架構藍圖,擺脫了手刻迴圈的混亂控制流。我們為 research-agent 專案建立了清晰的三階段研究管線,展示了增量狀態更新、串流步階監控以及 Mermaid 拓撲圖輸出,並透過嚴格的單元測試鞏固了程式品質。
透過 LangGraph 的抽象,原本盤根錯節的邏輯被理順為一條條清晰可辨的導軌。每一個節點只專注於自己負責的微小任務,所有的協同全部交由底層狀態機排程器進行標準化調度。
結語
在今天的範例中,我們的狀態更新採用的是標準的字典覆蓋模式:節點回傳什麼鍵值,就直接覆蓋原有的狀態。然而,在真實的對話式代理或反覆迭代的研究情境中,很多資料不是覆蓋、而是「累積」——例如對話歷史訊息清單(Messages)、搜尋到的眾多文獻碎片(Documents)。如果每一次更新都直接覆蓋舊資料,我們將無法保留寶貴的對話記憶與多輪研究軌跡。
明天,我們會進入「AG Day 12 狀態設計:TypedDict、reducer 與訊息累積」,深入拆解 LangGraph 的狀態聚合中樞,掌握 Annotated 與 add_messages 縮減器(Reducer)的核心心法,學習如何優雅地管理多輪對話訊息與增量集合累積。
延伸資源
- LangGraph 官方快速上手指南:
https://langchain-ai.github.io/langgraph/tutorials/introduction/。掌握 StateGraph 最新的建置慣用法。 - LangGraph 1.x 核心 API 官方文件:
https://langchain-ai.github.io/langgraph/reference/graphs/。深入查閱 StateGraph、START 與 END 規格說明。 - Python 官方 typing.TypedDict 規格指引(PEP 589):
https://peps.python.org/pep-0589/。理解結構化型別提示的語法與原理。 - Mermaid.js 流程圖官方規範:
https://mermaid.js.org/syntax/flowchart.html。學習如何將狀態圖拓撲渲染為高互動性圖表。
留言
張貼留言