iT邦幫忙

2026 iThome 鐵人賽

DAY 3
0
Vibe Coding

agent工作流系列 第 11

【Day 11】怎麼應用 LangGraph:從零構建具備循環、條件路由與持久化的狀態機

  • 分享至 

  • xImage
  •  

在昨天的文章中,我們建立了 LangGraph 的核心認知:將 Agent 建模為具備「全域狀態(Global State)」的有向圖(Directed Graph),讓 LLM 的推論與工具調用在可控、可觀察的狀態機中運作。

今天我們進入實戰篇:如何使用 Python 與 langgraph 套件,從零搭建一個包含「State 定義 ➔ LLM 決策節點 ➔ 工具執行節點 ➔ 條件路由(Conditional Edge)➔ MemorySaver 持久化」的完整生產級 Agent!


一、實戰目標:打造支援自我循環的 ReAct 狀態圖

我們要構建的工作流拓撲結構如下:

       ┌───────────────┐
       │   __start__   │
       └───────┬───────┘
               │
               ▼
       ┌───────────────┐
       │  agent_node   │ ◄─────────────┐
       └───────┬───────┘               │
               │ (條件路由: should_continue)
        [需要使用工具?]                │
        ├── 是 ──> ┌───────────────┐   │ (循環回傳結果)
        │          │  tools_node   │ ──┘
        │          └───────────────┘
        └── 否 ──> ┌───────────────┐
                   │    __end__    │
                   └───────────────┘

二、步驟 1:安裝依賴與定義全域狀態(State)

在 LangGraph 中,第一步是定義整個圖共享的資料結構。我們使用 TypedDict 並搭配 Annotatedadd_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]

三、步驟 2:定義工具與節點函數(Nodes)

我們定義一個具體的業務查詢工具,並建立兩個核心節點:

  1. agent_node:負責呼叫 LLM 進行推理與決策。
  2. 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)

四、步驟 3:定義條件路由(Conditional Edge)

條件邊負責檢查 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

五、步驟 4:組裝狀態圖並編譯(Compile with Memory)

我們使用 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)

六、步驟 5:執行與多輪對話驗證

透過傳入 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}")

執行軌跡解析:

  1. START ➔ agent:LLM 讀取提問,輸出 tool_calls=[check_order_status(order_id='ORD-101')]
  2. agent ➔ tools:條件邊判定有工具呼叫,路由至 tools 節點執行查詢,產出 ToolMessage(content='已出貨,預計明日送達')
  3. tools ➔ agent:普通邊循環回到 agent,LLM 根據工具回傳輸出自然語言總結。
  4. agent ➔ END:條件邊判定已無工具呼叫,安全抵達 END
  5. 第 2 輪對話:由於掛載了 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 系統的核心設計模式與溝通協定!


上一篇
【Day 10】甚麼是 LangGraph:從線性鏈路邁向循環狀態圖的 Agent 架構
下一篇
【Day 12】多 Agent 協同架構:Supervisor 與 Hierarchical 工作流
系列文
agent工作流15
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言