先講白話定位:LangGraph 是一個低階的編排框架(orchestration framework)與執行環境,專門用來蓋「長時間運行、有狀態」的 Agent。重點是「低階」— 它不會幫你包好一套現成的思考迴圈,而是把控制權完全交回你手上,讓你自己決定每一步的邏輯要怎麼走。
這是 LangGraph 官方文件反覆強調的核心重點:在同一張圖裡,自由混搭寫死的邏輯與由 LLM 驅動的判斷。需要可靠、可預期、可稽核的地方,就用寫死的程式碼;需要彈性、需要模型自己判斷的地方,才交給 LLM。這種細粒度的控制,是 LangGraph 跟一般「Agent 框架」最大的差異。
除此之外,LangGraph 提供的基礎建設還包括:
持久化(Persistence):Agent 可以在失敗後恢復,從中斷的地方繼續執行,而不用整個重跑。
人機協作(Human-in-the-loop):可以在任何一個節點暫停,讓人檢查甚至修改 Agent 當下的狀態。
完整的記憶機制:同時支援短期的工作記憶(這次對話中的推理過程)與跨會話的長期記憶。
可觀測性:能追蹤執行路徑、捕捉狀態變化,方便除錯複雜的 Agent 行為。
正式環境部署:針對長時間、有狀態的工作流設計的可擴展部署架構。
順帶一提,LangGraph 的設計靈感來自 Google 的 Pregel 論文與 Apache Beam,對外介面則參考了圖論函式庫 NetworkX — 所以你會看到「node」、「edge」、「graph」這些詞彙,其實就是把 Agent 的流程用圖(graph)的方式建模。
安裝很單純:
pip install -U langgraph
官方給的 hello world 大概長這樣:
from langgraph.graph import StateGraph, MessagesState, START, END
def mock_llm(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
三個關鍵動作:定義一個 State(狀態的資料結構)、加入 Node(做事的函式)、用 Edge 把節點串起來,最後 compile() 變成可以執行的 Graph。這幾個字,幾乎就是 LangGraph 的全部語彙 — 後面的範例會反覆用到。
原理講完了,光看定義還是很抽象。這裡用一個具體案例過一遍設計流程。情境是這樣:公司要蓋一個客服信箱 Agent,需求包括:
先畫流程圖,每個獨立的步驟就是一個 Node,一個只做一件事的函式。以客服信箱為例,節點大概是:讀信 → 分類意圖 → (依分類走不同分支:查文件 / 建立 Bug 單 / 直接轉人工)→ 草擬回覆 → 人工審核(視情況)→ 寄出。
這裡有個細節值得注意:圖上的箭頭只是「可能的路徑」,實際走哪一條,是在節點內部的程式邏輯決定的,不是畫在圖上就自動發生。像「分類意圖」這個節點,自己會判斷接下來要跳到「查文件」還是「轉人工」。
節點大致分成四種:
| 類型 | 用途 | 範例 |
|---|---|---|
| LLM 步驟 | 需要理解、生成、判斷的地方 | 分類意圖、草擬回覆 |
| 資料步驟 | 從外部系統取資料 | 搜尋文件庫、查客戶歷史 |
| 動作步驟 | 執行外部操作 | 寄出郵件、建立 Bug 單 |
| 使用者輸入步驟 | 需要真人介入 | 人工審核節點 |
分類型的好處是,你會很自然地想到:「這個節點需不需要重試策略」、「要不要加快取(cache)」、「靜態的 prompt 內容跟動態的 state 內容怎麼分開」— 這些設計決策,在動手開發前就該想清楚。
State 是所有節點共享的記憶體,你可以把它想成 Agent 一邊做事一邊寫的筆記本。判斷一筆資料該不該放進 State,官方給了一個很實用的判斷準則:
還有一條原則我覺得特別重要:State 只存原始資料,不要存格式化好的文字。也就是說,不要把已經套好模板的 prompt 存進 State,而是存乾淨的原始資料,等到某個節點真的要組 prompt 的時候,再即時格式化。這樣做的好處是不同節點可以用同一份資料組出不同的 prompt,換模板也不用動 State 的結構,debug 時也能清楚看到每個節點收到的到底是什麼。
例如,一個客服信箱的 State 大概長這樣:
from typing import TypedDict, Literal
class EmailClassification(TypedDict):
intent: Literal["question", "bug", "billing", "feature", "complex"]
urgency: Literal["low", "medium", "high", "critical"]
topic: str
class EmailAgentState(TypedDict):
email_content: str
sender_email: str
classification: EmailClassification | None
search_results: list[str] | None
draft_response: str | None
可以注意到:分類結果就是一個單純的 dict,沒有任何 prompt 樣板混在裡面。
Node 的本質很單純:接收 State,做點事,回傳更新。當節點需要自己決定下一步要去哪裡時,LangGraph 提供 Command 物件,同時帶著「要更新的資料」跟「要跳去哪個節點」:
from langgraph.types import Command
def classify_intent(state: EmailAgentState) -> Command:
classification = structured_llm.invoke(build_prompt(state))
if classification["urgency"] == "critical":
goto = "human_review"
elif classification["intent"] == "question":
goto = "search_documentation"
else:
goto = "draft_response"
return Command(update={"classification": classification}, goto=goto)
最後一步反而最簡單,因為大部分的路由邏輯已經寫進節點裡的 Command 了,圖本身只需要幾條「一定會走」的 Edge:
from langgraph.checkpoint.memory import MemorySaver
workflow = StateGraph(EmailAgentState)
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("search_documentation", search_documentation)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", END)
app = workflow.compile(checkpointer=MemorySaver())
這裡要搭配 checkpointer 才能讓 interrupt() 生效,因為人工審核節點會讓整張圖暫停,狀態要先存下來,才有辦法「幾天後」再回來接著跑,而且靠 thread_id 把同一個對話的狀態綁在一起。
把五個步驟走完一輪,可以歸納出幾條 LangGraph 的思考模式:
拆得夠細,才換得到韌性:每個節點的邊界就是一個 checkpoint,節點切得越細,失敗後要重跑的範圍就越小。
State 是原始資料的倉庫,不是格式化好的訊息:格式化永遠留到真正要用的那一刻再做。
Node 自己決定路由,Graph 只留必要的 Edge:控制流程因此變得清楚又可追溯 — 看目前卡在哪個節點,大致就能猜到下一步會發生什麼。
Human in Loops:interrupt() 讓「等人回覆」這件事,變成跟其他非同步步驟一樣自然。