iT邦幫忙

2026 iThome 鐵人賽

DAY 5
0
AI Engineering

地端 AI 建築學系列 第 5

05 從 LangChain 到 LangGraph (2) 先畫靶再射箭

  • 分享至 

  • xImage
  •  

LangGraph 到底在解決什麼問題

先講白話定位:LangGraph 是一個低階的編排框架(orchestration framework)與執行環境,專門用來蓋「長時間運行、有狀態」的 Agent。重點是「低階」— 它不會幫你包好一套現成的思考迴圈,而是把控制權完全交回你手上,讓你自己決定每一步的邏輯要怎麼走。


核心概念:混合「確定性」與「Agentic」的步驟

這是 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 的全部語彙 — 後面的範例會反覆用到。


用一個真實案例學「LangGraph 式思考」

原理講完了,光看定義還是很抽象。這裡用一個具體案例過一遍設計流程。情境是這樣:公司要蓋一個客服信箱 Agent,需求包括:

  • 讀取客戶來信
  • 依緊急程度與主題分類
  • 搜尋文件庫回答問題
  • 草擬回覆
  • 複雜案件轉真人處理
  • 需要時安排後續追蹤

步驟一:把流程拆成一個個 Node

先畫流程圖,每個獨立的步驟就是一個 Node,一個只做一件事的函式。以客服信箱為例,節點大概是:讀信 → 分類意圖 → (依分類走不同分支:查文件 / 建立 Bug 單 / 直接轉人工)→ 草擬回覆 → 人工審核(視情況)→ 寄出。

這裡有個細節值得注意:圖上的箭頭只是「可能的路徑」,實際走哪一條,是在節點內部的程式邏輯決定的,不是畫在圖上就自動發生。像「分類意圖」這個節點,自己會判斷接下來要跳到「查文件」還是「轉人工」。

步驟二:分辨每個節點的「類型」

節點大致分成四種:

類型 用途 範例
LLM 步驟 需要理解、生成、判斷的地方 分類意圖、草擬回覆
資料步驟 從外部系統取資料 搜尋文件庫、查客戶歷史
動作步驟 執行外部操作 寄出郵件、建立 Bug 單
使用者輸入步驟 需要真人介入 人工審核節點

分類型的好處是,你會很自然地想到:「這個節點需不需要重試策略」、「要不要加快取(cache)」、「靜態的 prompt 內容跟動態的 state 內容怎麼分開」— 這些設計決策,在動手開發前就該想清楚。

步驟三:設計 State — Agent 的共用筆記本

State 是所有節點共享的記憶體,你可以把它想成 Agent 一邊做事一邊寫的筆記本。判斷一筆資料該不該放進 State,官方給了一個很實用的判斷準則:

  • 這筆資料需要跨步驟保留嗎?要的話放進 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 寫成函式

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)

步驟五:把 Node 串成 Graph

最後一步反而最簡單,因為大部分的路由邏輯已經寫進節點裡的 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 的思考模式:

  1. 拆得夠細,才換得到韌性:每個節點的邊界就是一個 checkpoint,節點切得越細,失敗後要重跑的範圍就越小。

  2. State 是原始資料的倉庫,不是格式化好的訊息:格式化永遠留到真正要用的那一刻再做。

  3. Node 自己決定路由,Graph 只留必要的 Edge:控制流程因此變得清楚又可追溯 — 看目前卡在哪個節點,大致就能猜到下一步會發生什麼。

  4. Human in Loopsinterrupt() 讓「等人回覆」這件事,變成跟其他非同步步驟一樣自然。


上一篇
04 從 LangChain 到 LangGraph (1) 簡單起手式
下一篇
06 從 LangChain 到 LangGraph (3) 設計模式
系列文
地端 AI 建築學14
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言