在昨天的文章中,我們建立了 LangGraph 的核心認知:將 Agent 建模為具備「全域狀態(Global State)」的有向圖(Directed Graph),讓 LLM 的推論與工具調用在可控、可觀察的狀態機中運作。
今天我們進入實戰篇:如何使用 Python 與 langgraph 套件,從零搭建一個包含「State 定義 ➔ LLM 決策節點 ➔ 工具執行節點 ➔ 條件路由(Conditional Edge)➔ MemorySaver 持久化」的完整生產級 Agent!
我們要構建的工作流拓撲結構如下:
┌───────────────┐
│ __start__ │
└───────┬───────┘
│
▼
┌───────────────┐
│ agent_node │ ◄─────────────┐
└───────┬───────┘ │
│ (條件路由: should_continue)
[需要使用工具?] │
├── 是 ──> ┌───────────────┐ │ (循環回傳結果)
│ │ tools_node │ ──┘
│ └───────────────┘
└── 否 ──> ┌───────────────┐
│ __end__ │
└───────────────┘
在 LangGraph 中,第一步是定義整個圖共享的資料結構。我們使用 TypedDict 並搭配 Annotated 與 add_messages reducer,確保新產生的對話訊息會**追加(Append)**進狀態,而不是覆蓋舊資料。
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# 定義圖的全域狀態結構
class State(TypedDict):
# 使用 add_messages 機制:新訊息會自動追加至 list
messages: Annotated[list, add_messages]
我們定義一個具體的業務查詢工具,並建立兩個核心節點:
agent_node:負責呼叫 LLM 進行推理與決策。tools_node:負責執行 LLM 發起的工具調用(直接使用 LangGraph 內建的 ToolNode)。from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
# 1. 定義自訂業務工具
@tool
def check_order_status(order_id: str) -> str:
"""根據訂單編號查詢物流狀態。"""
mock_db = {
"ORD-101": "已出貨,預計明日送達",
"ORD-202": "備貨中,預計 3 個工作天內寄出"
}
return mock_db.get(order_id, f"找不到訂單編號【{order_id}】的資料。")
tools = [check_order_status]
# 2. 初始化 LLM 並綁定工具
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
llm_with_tools = llm.bind_tools(tools)
# 3. 定義 Agent 節點函數
def agent_node(state: State):
"""執行 LLM 思考,回傳決策訊息"""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# 4. 建立工具節點
tool_node = ToolNode(tools)
條件邊負責檢查 LLM 的最新輸出。如果模型回傳了 tool_calls,表示需要執行工具,流程導向 tools;如果沒有,表示任務已完成,直接前往 END。
def should_continue(state: State) -> str:
"""判斷下一步是前往工具節點還是結束"""
last_message = state["messages"][-1]
# 若模型輸出了工具呼叫請求,前往 "tools" 節點
if hasattr(last_message, "tool_calls") and len(last_message.tool_calls) > 0:
return "tools"
# 否則任務完成,前往結束
return END
我們使用 StateGraph 將節點與邊組裝在一起,並掛載 MemorySaver(記憶檢查點機制),實現多輪對話持久化與狀態隔離。
from langgraph.checkpoint.memory import MemorySaver
# 1. 初始化狀態圖
workflow = StateGraph(State)
# 2. 註冊節點
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
# 3. 建立連線 (Edges)
workflow.add_edge(START, "agent") # 起點直達 agent 節點
# 條件邊:agent 執行完後,依據 should_continue 決定分流
workflow.add_conditional_edges(
"agent",
should_continue,
{
"tools": "tools",
END: END
}
)
# 普通邊:tools 節點執行完畢後,必定循環回到 agent 進行下一輪思考
workflow.add_edge("tools", "agent")
# 4. 掛載記憶檢查點並完成編譯
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
透過傳入 thread_id,LangGraph 會在背景自動管理該對話 Session 的完整狀態與記憶。
from langchain_core.messages import HumanMessage
# 設定特定會話的 Thread ID
config = {"configurable": {"thread_id": "session_tw_001"}}
# --- 第 1 輪對話:觸發工具調用循環 ---
print("=== 第 1 輪對話 ===")
input_msg = [HumanMessage(content="你好,幫我查一下訂單 ORD-101 的狀態。")]
# 使用 stream 逐步檢視節點執行狀態
for event in app.stream({"messages": input_msg}, config=config, stream_mode="values"):
latest_msg = event["messages"][-1]
print(f"[{latest_msg.__class__.__name__}] {latest_msg.content or latest_msg.tool_calls}")
# --- 第 2 輪對話:延續上一輪的記憶狀態 ---
print("\n=== 第 2 輪對話 ===")
input_msg_2 = [HumanMessage(content="那它大概什麼時候會送到?")]
for event in app.stream({"messages": input_msg_2}, config=config, stream_mode="values"):
latest_msg = event["messages"][-1]
print(f"[{latest_msg.__class__.__name__}] {latest_msg.content}")
START ➔ agent:LLM 讀取提問,輸出 tool_calls=[check_order_status(order_id='ORD-101')]。agent ➔ tools:條件邊判定有工具呼叫,路由至 tools 節點執行查詢,產出 ToolMessage(content='已出貨,預計明日送達')。tools ➔ agent:普通邊循環回到 agent,LLM 根據工具回傳輸出自然語言總結。agent ➔ END:條件邊判定已無工具呼叫,安全抵達 END。checkpointer,模型能立即根據 thread_id 讀取上一輪「預計明日送達」的記憶並給出準確回答。| 核心元件 | 實作手段 | 解決痛點 |
|---|---|---|
| 狀態集中管理 | State(TypedDict) + add_messages |
消除模組間分散傳參的混亂,確保資料單一來源 |
| 動態分支控制 | add_conditional_edges |
取代寫死的多層 if-else,依據狀態自主分流 |
| 可控循環回溯 | add_edge("tools", "agent") |
讓 Agent 具備自我驗證、多次調用與修復能力 |
| 持久化會話 | MemorySaver + thread_id |
天然支援多用戶狀態隔離、斷點續傳與上下文延續 |
透過 StateGraph、Node、Conditional Edge 與 Checkpointer,我們成功用乾淨、可擴展的圖結構實作了一個標準的 ReAct Agent 狀態機。
然而,在真實的大型專案中,工作流往往需要由多個具備專屬技能的 Agent 協同合作(例如:Router Agent 分流 ➔ Research Agent 蒐集 ➔ Writer Agent 產出)。這時我們該如何建立多層次、多節點的協作系統?
明天 【Day 12】多 Agent 協同架構:Supervisor 與 Hierarchical 工作流,我們將探討現代 Multi-Agent 系統的核心設計模式與溝通協定!