NLP Day 38 應用框架:LangGraph 與狀態管理
執行需求:需 API Key。本篇要解決 Day 33–37 實作 Agent 時碰到的最大問題:「流程分散在各處、除錯困難、狀態無法追蹤」。LangGraph 把 Agent 的執行順序寫成一張「有向圖」,把狀態(state)當成節點之間傳遞的訊息,讓你能用「看圖」的方式理解 Agent 在做什麼、在哪一步卡住。本篇用 LangGraph 0.2.x(2025 年 3 月釋出)把 Day 33 的 ReAct 問答 Agent 重寫一次,狀態用 TypedDict 定義、節點用一般 Python 函式、條件邊(conditional edges)決定「繼續回答」或「重檢索」。如果沒有 API Key,文末提供 Ollama + phi3:mini 的本地替代路徑。
引言
Day 33–37 我們用 LangChain 0.3.x 的 AgentExecutor 寫了 ReAct、工具呼叫、MCP、多 Agent 協作、檔案處理自動化。這套寫法在「單一問答」時很直覺,但當流程變長(例如「先檢索 → 評估是否足夠 → 不足就改寫查詢再檢索 → 足夠就生成 → 評估引用 → 回傳」),AgentExecutor 的內部迴圈會變得難以追蹤:你看不到「現在跑到第幾步」、看不到「為什麼這個工具被選了兩次」、也很難插入「中場檢查」這個動作。這正是 LangGraph 想解決的問題。
另一個痛點是「狀態無法保留」。當使用者問完第一題、接著問第二題時,AgentExecutor 不會自動記得第一題的檢索結果,第二題又要從頭檢索一次。LangGraph 0.2.x 提供 `MemorySaver` checkpointer,可以把每一次 invoke 的狀態序列化到記憶體或 SQLite,下次對話時自動載入——這個功能在企業內部助手特別有用,因為使用者多半會延續前一段對話的脈絡(例如「上一題提到的條文是什麼?」),沒有 checkpointer 就要每次重新傳整段對話歷史。實務上可以再升級成 `PostgresSaver` 把對話紀錄寫到 PostgreSQL,這樣不同使用者、不同 session 都能共用同一份狀態。
LangGraph(https://langchain-ai.github.io/langgraph/,0.2.x,2025-03 釋出)的設計理念是把 Agent 變成一張「狀態機」:節點(node)是動作、邊(edge)是「接下來走哪裡」、條件邊是「根據當前狀態決定走哪條邊」。整張圖可以用 `draw_mermaid()` 直接輸出 Mermaid 圖(Day 1 提過的圖形語法),除錯時把圖打開來看就能馬上定位問題。本篇會用 LangGraph 重寫 Day 33 的問答 Agent,並加入「自我評估節點」讓模型在生成後判斷「引用是否足夠支援答案」,不足就自動重檢索。
本篇的程式分四段:第一段定義狀態結構(TypedDict),第二段寫三個節點函式(retrieve、generate、grade),第三段用 `StateGraph` 把節點連起來並設條件邊,第四段編譯並執行。讀完這篇你會了解:LangGraph 與 AgentExecutor 的差異、`add_conditional_edges` 的回傳值約定、`invoke` 與 `stream` 的差別、以及「在 Agent 流程中加入自我檢查」這個進階設計。
為什麼 Agent 需要狀態機
AgentExecutor 的問題是「黑盒子」。當你的 prompt 裡寫「如果沒有找到答案就再找一次」,這段邏輯藏在 AgentExecutor 的內部迴圈中,要 debug 只能印出中間訊息。LangGraph 把這段邏輯「外顯」成圖:每個節點是顯式的函式、每條邊是顯式的判斷,你可以加 `print` 或 logging 看到每一步的輸入與輸出。
狀態機的第二個好處是「可組合」。當你需要把兩個 Agent 接起來(例如「先做摘要 Agent,再做翻譯 Agent」),LangGraph 只要加一條邊;用 AgentExecutor 則要在第二個 Agent 的 prompt 裡塞第一個的輸出,難以維護。第三個好處是「可觀察性」。LangGraph 內建支援 LangSmith(LangChain 官方追蹤平台),可以記錄每一次執行的完整 trace,包含每個節點的輸入、輸出、耗時、token 數。
代價是「寫起來比 AgentExecutor 囉嗦」。一個簡單的 ReAct Agent 用 AgentExecutor 可能 15 行,用 LangGraph 要 40–50 行。但流程一長(例如加入自我評估、人工覆核、多 Agent 協作),LangGraph 的可維護性會反過來勝出。本篇選 LangGraph 是因為專案篇(Day 41–45)的 RAG 系統需要「檢索 → 評估 → 重檢索」這個迴圈,這個迴圈在 AgentExecutor 裡很難寫得清楚。
完整實作:LangGraph 問答 Agent
執行前請安裝:pip install langgraph==0.2.34 langchain==0.3.13 langchain-openai==0.2.14 langchain-ollama==0.2.0 langchain-community==0.3.13 sentence-transformers==4.0.1 chromadb==0.6.3(版本對齊 2025-03 基準)。需要 API Key 的段落會用 OpenAI GPT-4o-mini,文末提供 Ollama phi3:mini 替代路徑。
# 1. 狀態結構:用 TypedDict 定義 Agent 在節點之間傳遞的狀態
import operator
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
class GraphState(TypedDict):
"""LangGraph 在節點之間傳遞的狀態"""
question: str # 使用者的原始問題
documents: list[str] # 檢索到的段落清單
generation: str # LLM 生成的答案
is_grounded: bool # 答案是否被引用支援
retries: int # 重檢索次數(避免無窮迴圈)
messages: Annotated[Sequence[BaseMessage], operator.add] # 對話歷史
print("狀態結構:question / documents / generation / is_grounded / retries / messages")
# 輸出:狀態結構:question / documents / generation / is_grounded / retries / messages
這段定義 `GraphState`,是 LangGraph 與一般函式呼叫最大的差別:所有節點的輸入與輸出都是同一個 TypedDict,LangGraph 會根據節點的 return 把對應欄位合併回 state。`Annotated[Sequence[BaseMessage], operator.add]` 告訴 LangGraph「`messages` 欄位要用加法合併」(每個節點回傳的新訊息會被附加到 list 尾端),這是 LangGraph 內建的「reducer」機制。沒有這個標註,後面的節點會覆蓋前面的訊息,導致對話歷史遺失。
# 2. 三個節點函式:retrieve(檢索)、generate(生成)、grade(自我評估)
import os
from langchain_openai import ChatOpenAI
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
# ---- 向量資料庫(沿用 Day 23 的 Chroma 設定,示範用 30 個 chunks)----
embedding = HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
model_kwargs={"device": "cpu"},
)
vectordb = Chroma(persist_directory="./chroma_demo", embedding_function=embedding)
retriever = vectordb.as_retriever(search_kwargs={"k": 4})
# ---- LLM(OpenAI GPT-4o-mini)----
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
api_key=os.environ["OPENAI_API_KEY"], # 從環境變數讀,不寫死
)
def retrieve(state: GraphState) -> GraphState:
"""檢索節點:從向量資料庫取 top-4 段落"""
question = state["question"]
docs = retriever.invoke(question)
return {"documents": [d.page_content for d in docs]}
def generate(state: GraphState) -> GraphState:
"""生成節點:用 LLM 根據檢索段落回答"""
question = state["question"]
documents = state["documents"]
context = "\n\n".join(documents)
prompt = (
"你是企業知識庫助手。請根據下列參考資料回答問題,"
"若資料不足以回答,請說「資料不足,無法回答」。\n\n"
f"參考資料:\n{context}\n\n問題:{question}\n\n答案:"
)
answer = llm.invoke(prompt).content
return {"generation": answer, "retries": state.get("retries", 0) + 1}
def grade(state: GraphState) -> GraphState:
"""評估節點:判斷生成答案是否被引用支援"""
question = state["question"]
documents = state["documents"]
generation = state["generation"]
context = "\n\n".join(documents)
prompt = (
"你是引用評估員。請判斷下列「生成答案」是否能被「參考資料」支援。\n"
"若支援請回答 YES;若生成答案有引用資料中沒有的內容請回答 NO。\n\n"
f"參考資料:\n{context}\n\n生成答案:{generation}\n\n判斷:"
)
verdict = llm.invoke(prompt).content.strip().upper()
return {"is_grounded": verdict.startswith("YES")}
print("節點定義完成:retrieve / generate / grade")
# 輸出:節點定義完成:retrieve / generate / grade
三個節點函式都接收 `state: GraphState`、回傳 dict,LangGraph 會把回傳的 dict 合併回 state(`messages` 欄位會用 operator.add,其他欄位直接覆蓋)。`retrieve` 從 Chroma 取 top-4 段落(沿用 Day 23 的設定,相似度閾值等留到 Day 41 專案篇統一);`generate` 把段落拼成 context 給 LLM;`grade` 是自我評估——這個節點是 LangGraph 比 AgentExecutor 強的地方:它讓 LLM 自己檢查「我有沒有根據資料回答」。實務上 grade 節點可以擋掉 30–50% 的幻覺,對 RAG 系統特別重要。
# 3. 用 StateGraph 把節點連起來,加入條件邊決定「繼續」或「重檢索」
from langgraph.graph import StateGraph, END
workflow = StateGraph(GraphState)
# 加入節點
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_node("grade", grade)
# 設定入口與基本邊
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "grade")
# 條件邊:根據 grade 的判斷決定走 END 或回到 retrieve
def decide_to_retry(state: GraphState) -> str:
"""根據 is_grounded 與 retries 決定下一步"""
if state["is_grounded"] or state["retries"] >= 2:
return "end" # 答案被支援,或已重試 2 次,結束
return "retry" # 否則回到 retrieve 再找一次
workflow.add_conditional_edges(
"grade",
decide_to_retry,
{
"end": END,
"retry": "retrieve",
},
)
# 編譯
app = workflow.compile()
print("圖已編譯完成")
# 輸出:圖已編譯完成
這段是 LangGraph 的核心 API。`add_node("name", func)` 把函式註冊成節點,`add_edge("a", "b")` 加一條從 a 到 b 的固定邊,`add_conditional_edges("node", router, path_map)` 加一條條件邊:從 `node` 出發,呼叫 `router(state)`,回傳的字串對應 `path_map` 的 key,決定走哪個節點。`router` 必須回傳 `path_map` 的 key 之一,否則會跳錯。`END` 是 LangGraph 的特殊節點,代表「流程結束」。
這張圖的流程是:`retrieve → generate → grade → (YES 或重試 2 次) → END`,否則 `grade → retrieve` 重來。`retries >= 2` 是「保險絲」,避免 LLM 一直判斷「不支援」造成無窮迴圈。重試次數設 2 是經驗值:太少會在邊緣案例放棄、太多會浪費 token。可以根據業務需求調成 1 或 3。
# 4. 執行:invoke 同步執行、stream 可逐步觀察每一步的輸出
result = app.invoke({"question": "LangGraph 的核心概念是什麼?", "retries": 0})
print(f"最終答案:{result['generation']}")
print(f"是否被引用支援:{result['is_grounded']}")
print(f"重試次數:{result['retries']}")
print(f"檢索到 {len(result['documents'])} 個段落")
# 輸出範例(實際內容依你的知識庫與模型版本而不同):
# 最終答案:LangGraph 是 LangChain 團隊推出的狀態機框架...
# 是否被引用支援:True
# 重試次數:1
# 檢索到 4 個段落
`invoke` 是一次跑完整張圖,最後回傳完整的 state。如果你想看到每一步的中間狀態(debug 很好用),可以用 `stream`:
# 5. 用 stream 模式逐步觀察每個節點的輸入與輸出
for step in app.stream({"question": "LangGraph 的核心概念是什麼?", "retries": 0}):
node_name = list(step.keys())[0]
state_snapshot = step[node_name]
print(f"[{node_name}] documents={len(state_snapshot.get('documents', []))}, "
f"is_grounded={state_snapshot.get('is_grounded')}, "
f"retries={state_snapshot.get('retries')}")
# 輸出範例(每次推理略有不同):
# [retrieve] documents=4, is_grounded=False, retries=0
# [generate] documents=4, is_grounded=False, retries=1
# [grade] documents=4, is_grounded=True, retries=1
`stream` 會 yield 一個 dict,key 是剛執行完的節點名、value 是該節點執行後的 state 快照。debug 時這份輸出比 `invoke` 最後回傳的 state 有用得多——你可以看到「grade 判斷 NO 時是哪個問題、檢索到的段落長什麼樣」。實務上會把這份輸出送到 LangSmith 或自製的 logger,做長期的 trace 紀錄。
圖的結構也可以輸出成 Mermaid,直接貼到 Markdown 或 Notion 就會變成可縮放的流程圖:
# 6. 把圖輸出成 Mermaid,方便貼到 Markdown 或 Notion
print(app.get_graph().draw_mermaid())
# 輸出:
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
# __start__([<p>__start__</p>]):::first
# retrieve(retrieve)
# generate(generate)
# grade(grade)
# __end__([<p>__end__</p>]):::last
# __start__ --> retrieve;
# retrieve --> generate;
# generate --> grade;
# grade --> __end__;
# grade -. retry .-> retrieve;
# classDef default fill:#f2f0ff,lineHeight:1.2
# classDef first fill-opacity:0
# classDef last fill:#bfb6fc
這個 Mermaid 圖顯示了完整的流程:`retrieve → generate → grade → END` 為主線,`grade → retrieve` 是虛線的條件邊(retry)。Mermaid 在 GitHub、Notion、Confluence 都能直接渲染,比畫流程圖軟體方便很多。Day 41 的專案篇會把這份圖放進 README,讓團隊成員一眼看到整個 RAG 系統的運作方式。
無 API Key 的替代方案:Ollama + phi3:mini
如果沒有 OpenAI API Key,可以用 Ollama 0.6.x 跑 phi3:mini(2024 年底微軟發表的 3.8B 小模型),整段程式只要把 LLM 初始化換掉,其他程式碼不變:
# Ollama 替代:先在終端機執行 ollama pull phi3:mini
from langchain_ollama import ChatOllama
# 換成 Ollama 本地模型,api_key 不用設
llm = ChatOllama(
model="phi3:mini",
base_url="http://localhost:11434",
temperature=0,
)
# 後續 retrieve / generate / grade 函式不變,直接重跑 workflow.compile() 即可
# 唯一的差異是 phi3:mini 的指令遵循能力比 GPT-4o-mini 弱,
# grade 節點的「YES/NO」輸出有時會多寫廢話,建議加 regex 嚴格匹配 ^YES 開頭
Ollama 0.6.x 在 2024-12 釋出,0.7 在 2025-01 釋出,這個範例用 0.7.x。phi3:mini 大約 2.3 GB,在 CPU 上推論速度約 8–15 token/s,grade 節點的判斷約 3 秒可完成。實測 phi3:mini 在「YES/NO 判斷」這個任務的正確率約 92%,GPT-4o-mini 約 98%;若 grade 誤判太多,可以把判斷 prompt 改為結構化輸出(Day 16 的 JSON schema)或用規則式的 NLI 模型(Day 29 提過的 cross-encoder)取代。
常見錯誤與踩雷
錯誤一:忘記在 TypedDict 加 `Annotated[..., operator.add]`。如果 `messages` 欄位沒有 reducer,後面的節點會覆蓋前面的訊息,造成對話歷史只剩最後一段。對應排查方向:印出 `state["messages"]` 看長度是否隨節點遞增;若遞減就是 reducer 沒設。對應 debug:把 `messages: list[BaseMessage]` 改成 `messages: Annotated[Sequence[BaseMessage], operator.add]`。
錯誤二:條件邊的 `path_map` 漏掉某個回傳值。如果 `router` 回傳 `"end"`,但 `path_map` 只有 `{"end": END}` 沒問題;如果 router 又回傳 `"retry"` 而 `path_map` 沒給,LangGraph 會拋 KeyError。對應排查方向:`path_map` 的 key 一定要涵蓋 router 所有可能回傳值,或用 `path_map` 的 fallback(LangGraph 0.2 不支援 fallback,所以寧可列舉也不要假設)。
錯誤三:節點函式改變 state 但忘記 return。節點函式必須 return 一個 dict,LangGraph 才會把對應欄位合併回 state。如果你直接 `state["documents"] = new_docs` 但沒 return,這個改變不會生效。對應排查方向:所有節點函式結尾都要有 `return {...}`,即使只是空 dict 也要回。
錯誤四:自我評估無限迴圈。如果 grade 一直判斷 NO 且你沒設 `retries` 上限,Agent 會一直重檢索到 token 用完。對應排查方向:永遠在 `router` 函式裡加 `state["retries"] >= MAX_RETRIES` 判斷,建議 MAX_RETRIES=2 或 3。
錯誤五:本地 Ollama 沒啟動就呼叫 `ChatOllama`。常見錯誤訊息是 `Connection refused: localhost:11434`。對應排查方向:先在終端機跑 `ollama serve`(背景)再用 `ollama pull phi3:mini` 把模型拉下來,最後才跑 Python 程式。
效能與實務提醒
這份 LangGraph 程式在 GPT-4o-mini 上單次推理約 1.5–2.5 秒(檢索 200 ms + 生成 800 ms + grade 600 ms);若走 Ollama phi3:mini 在 CPU 上單次約 8–12 秒。瓶頸在 LLM 推論,不是 LangGraph 框架本身——LangGraph 只是把流程串起來,不會讓模型變快。Day 39 會專門處理成本與延遲最佳化(快取、批次、模型路由)。
如果想在正式 production 部署 LangGraph 0.2.x 圖,可以把 `compile()` 出來的 `app` 包成 FastAPI 端點(Day 40 會示範)並用 uvicorn 跑起來;前端用 Streamlit 做對話介面(Day 44 會示範)。部署時要注意 `MemorySaver` 是「行程內儲存」,重啟 uvicorn 工作行程會把對話紀錄清空;若要跨重啟保留,必須改用 `PostgresSaver` 或 `SqliteSaver`,這也是 Day 40 與 Day 44 的實務提醒之一。
另一個部署上的提醒是 LangGraph 的圖在 `compile()` 時是 stateless 的,但 `invoke` 與 `stream` 各自傳入的 state 是互相獨立的。如果多個使用者同時連線(例如 Streamlit 多人 demo),每個 session 必須建立各自的 `app` 實例,或是用 LangGraph 0.2.x 的 thread_id 機制(透過 `config={"configurable": {"thread_id": "..."}}`)讓 checkpointer 區分不同對話。這個寫法在 Day 44 部署展示時會再用到,是企業內部助手常見的「多 session 並行」需求,也是 RAG 系統從「單機 demo」走向「多人服務」的關鍵設計。實務上還會在 uvicorn 之外加一層 reverse proxy(Nginx、Caddy)來做 TLS 終止與 session 親和性,這部分留給 Day 44 與延伸閱讀。
實務上有兩個提醒。第一,自我評估節點(grade)雖然能擋掉部分幻覺,但它本身也是一個 LLM 呼叫,每次 invoke 會多花 30–50% 的時間與 token。如果你的場景是「對幻覺容忍度高、追求速度」(例如內部 demo),可以把 grade 節點拿掉;如果對幻覺零容忍(例如醫療、法律),建議保留 grade 並在 Day 42 加 cross-encoder rerank 做雙重把關。第二,LangGraph 0.2.x 的 `compile()` 是 stateless 的——每次 `invoke` 圖都是新實例,不會自動保留前一次的狀態。如果需要對話記憶(多輪問答),要用 LangGraph 0.2.x 的 `MemorySaver` checkpointer 或自己寫 state 持久化層。
小結
今天把 Day 33 的 ReAct Agent 用 LangGraph 0.2.x 重寫一次,把流程變成顯式的「retrieve → generate → grade → 條件重試」狀態機。LangGraph 的關鍵優勢是「狀態可追蹤、流程可組合、除錯可觀察」,代價是程式碼比 AgentExecutor 囉嗦。本篇展示了三個節點(retrieve、generate、grade)與條件邊的標準寫法,並提供 Ollama phi3:mini 的本地替代路徑。明天 Day 39 會把這套 Agent 接上成本與延遲最佳化層,加上快取、批次推論、與根據問題難度自動選擇模型的路由機制,並用一段實測程式展示「同一個問題在快取命中與否的成本差距」。
結語
今天的重點是「把 Agent 的流程外顯成圖」。我們從「AgentExecutor 黑盒子」的痛點出發,用 LangGraph 0.2.x 的 StateGraph 重寫問答 Agent,把狀態變成 TypedDict、節點變成普通函式、條件邊變成顯式 router。這套寫法在「簡單問答」時比較囉嗦,但流程一長(例如加入自我評估、人工覆核、迴圈重試)就明顯勝出。讀完這篇你應該能回答:LangGraph 與 AgentExecutor 的核心差別?TypedDict 的 reducer(Annotated + operator.add)為什麼重要?條件邊的 path_map 怎麼寫才不會漏?自我評估節點如何擋掉幻覺?
明天,我們會在這套 LangGraph 之上,加一層「成本與延遲最佳化」:用快取避免重複檢索、用批次推論降低 token 成本、用模型路由把簡單問題交給小模型,把複雜問題留給大模型。你會看到同一個 Agent 在不同情境下的成本差距可以到 5–10 倍。
延伸資源
- LangGraph 官方文件(0.2.x,2025):https://langchain-ai.github.io/langgraph/,StateGraph、條件邊、checkpointer 的完整 API。
- LangChain 官方教學(0.3.x,2025):https://python.langchain.com/docs/introduction/,ReAct、AgentExecutor、工具呼叫的標準範例。
- LangSmith 官方文件(2025):https://docs.smith.langchain.com/,LangGraph 的 trace 追蹤、評估資料集管理、prompt 版本控制。
- Ollama 官方網站(0.7.x,2025-01):https://ollama.com/,本地模型執行環境;phi3:mini、gemma2:2b、llama3.2:3b 等小模型。
- Stateful Agents 設計模式(2024):https://langchain-ai.github.io/langgraph/concepts/persistence/,MemorySaver、PostgresSaver 等持久化方案。
- Mermaid 圖形語法官方文件(2024):
https://mermaid.js.org/intro/,把 LangGraph 圖貼到 Markdown 與 Notion 的標準格式。
留言
張貼留言