LangGraph 是一個基于狀態的工作流框架,它通過 節點(Nodes) 和 邊(Edges) 的組合,構建出復雜的工作流邏輯。這種設計特別適合處理需要動態決策、循環、多步驟交互的場景(比如對話系統、智能代理等)。下面我們通過一個簡單的例子,逐步講解 LangGraph 中的圖(Graph)是如何定義的。
1. 什么是 StateGraph?
StateGraph
是 LangGraph 的核心類,用于定義和管理基于狀態的圖結構。它的核心思想是:
- 節點:表示執行的操作(如調用模型、處理數據)。
- 邊:定義節點之間的執行順序(支持無條件跳轉、條件跳轉、循環)。
- 狀態(State):貫穿整個圖的數據結構,存儲和傳遞上下文信息。
舉個例子:
想象你正在開發一個聊天機器人,用戶輸入一條消息后,機器人需要依次完成以下步驟:
- 解析用戶意圖(節點A)。
- 查詢數據庫(節點B)。
- 生成回復(節點C)。
通過 StateGraph
,你可以將這三個步驟定義為節點,并通過邊連接它們,同時用狀態對象傳遞上下文(比如用戶的消息、查詢結果等)。
2. 如何定義一個圖?
定義一個圖的核心步驟包括:創建狀態、添加節點、設置邊,并最終編譯圖。以下是具體操作:
(1) 定義狀態
狀態是圖中傳遞的數據結構,通常是一個字典(TypedDict
或自定義類)。例如:
from typing_extensions import TypedDict
class State(TypedDict):messages: list # 存儲對話歷史query: str # 用戶當前問題
(2) 創建圖實例
使用 StateGraph
類初始化一個圖:
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
(3) 添加節點
節點是執行具體任務的函數。例如,添加一個解析用戶意圖的節點:
def parse_intent(state: State) -> dict:# 假設解析出用戶意圖是"查詢天氣"return {"query": "查詢天氣"}
graph_builder.add_node("parse_intent", parse_intent)
(4) 添加邊
邊定義了節點的執行順序。例如,將解析意圖節點連接到查詢數據庫節點:
graph_builder.add_edge("parse_intent", "query_database")
(5) 設置入口和出口
指定圖的起點和終點:
graph_builder.set_entry_point("parse_intent") # 起點
graph_builder.set_finish_point("generate_response") # 終點
(6) 編譯圖
將圖編譯為可執行對象:
compiled_graph = graph_builder.compile()
3. 核心方法詳解
LangGraph 提供了多個方法來構建圖,以下是關鍵方法的簡明說明:
方法名 | 作用 |
---|---|
add_node() | 添加一個節點到圖中。可以指定節點名稱和執行函數。 |
add_edge() | 添加一條有向邊,定義兩個節點的執行順序。如果多個起點,所有起點完成后才會執行終點。 |
add_conditional_edges() | 添加條件邊,根據狀態動態選擇下一個節點。例如:根據用戶意圖跳轉到不同的處理節點。 |
compile() | 將圖編譯為可執行對象,支持調用、流式處理等功能。 |
條件邊的使用示例
假設需要根據用戶意圖選擇不同的處理路徑:
def decide_next_node(state: State) -> str:if state["query"] == "查詢天氣":return "query_weather"else:return "handle_other"graph_builder.add_conditional_edges("parse_intent", decide_next_node)
4. 實際應用示例
場景:用戶詢問訂單狀態,機器人需要驗證身份后查詢數據庫。
代碼實現:
from langgraph.graph import StateGraph, START, ENDclass State(TypedDict):user_id: strorder_status: strdef verify_user(state: State) -> dict:# 模擬驗證用戶身份return {"user_id": "12345"}def query_order(state: State) -> dict:# 模擬查詢訂單狀態return {"order_status": "已發貨"}builder = StateGraph(State)
builder.add_node("verify_user", verify_user)
builder.add_node("query_order", query_order)
builder.add_edge(START, "verify_user")
builder.add_edge("verify_user", "query_order")
builder.add_edge("query_order", END)compiled = builder.compile()
result = compiled.invoke({"user_id": "", "order_status": ""})
print(result) # 輸出: {'user_id': '12345', 'order_status': '已發貨'}
5. 狀態管理的關鍵技巧
- 狀態更新:每個節點接收當前狀態,并返回更新后的狀態。例如,節點A修改
state["x"]
,節點B可以直接讀取新值。 - 狀態持久化:通過
checkpointer
參數保存狀態,支持斷點續傳(例如用戶中斷對話后恢復上下文)。 - 條件邏輯:利用條件邊(
add_conditional_edges
)動態決定下一步操作,比如根據用戶輸入選擇不同分支。
6. 常見問題與注意事項
- 節點命名沖突:確保每個節點名稱唯一。
- 狀態字段設計:狀態字段應包含所有必要的上下文,避免遺漏關鍵信息。
- 調試技巧:啟用
debug=True
編譯圖,觀察節點執行過程和狀態變化。
總結
LangGraph 的 StateGraph
通過節點和邊的靈活組合,讓開發者能夠輕松構建復雜的工作流。無論是簡單的線性流程,還是帶條件分支、循環的復雜邏輯,StateGraph
都能提供清晰的結構和強大的狀態管理能力。通過合理設計狀態和節點,你可以快速實現對話系統、智能代理等應用。