LangGraph 快速入門
- LangGraph 簡介 ===============
隨着大模型技術的不斷演進,**AI Agent **成爲了大模型領域的焦點。開發者們紛紛投身其中,期望構建出能夠自主決策、靈活交互的智能體。在這片競爭激烈的技術藍海中,LangGraph 作爲從 LangChain 孵化而來的獨立平臺,迅速崛起爲 Multi-Agent 開發框架中的佼佼者。
LangGraph 基於有向無環圖(DAG)這一核心數據結構,通過節點(Node)和邊(Edge)的組合,使得開發者能夠快速地搭建起一個 Multi-Agent 系統。
對於追求高效、可靠開發的工程師們來說,LangGraph 的核心優勢在於:
-
敏捷高效:憑藉簡潔統一的開發架構,將複雜 AI Agent 開發流程標準化,降低開發門檻;開發者可快速複用節點與工作流模板,減少重複性開發工作,大幅縮短項目週期,實現從構思到落地的高效轉化 。
-
可靠性和可控性:通過審覈檢查與人工介入機制,確保智能體行動精準合規,同時能在長期工作流中保留上下文,讓智能體行動始終保持正確方向。
-
高可擴展性:開發者可擺脫固有框架束縛,使用完全描述性的底層原語自由定製智能體,輕鬆設計滿足各類需求的可擴展多智能體系統。
-
原生流式支持:支持逐個 token 流式傳輸和中間步驟流式輸出,實時呈現智能體推理過程,增強交互透明度。
從企業級複雜業務流程自動化,到創新型智能交互應用開發,LangGraph 都能成爲開發者手中的得力工具。接下來,就讓我們開啓 LangGraph 的快速入門之旅,探索其強大功能與魅力。
- LangGraph 的核心概念 ==================
LangGrapg 有三個最核心的概念,分別是:State、Node 和 Graph ,對這幾個概念的具體理解如下:
-
**State:狀態,其中記錄了智能體運行過程中的所有信息,涵蓋輸入數據、中間計算結果和最終輸出,爲智能體的持續運行提供上下文。**狀態內的參數可以在多個節點間保存和流轉,是節點間通信的核心機制。
-
Node:節點,它是構成 LangGraph 的基本單元。每個 Node 可實現特定功能,例如數據處理、邏輯判斷或調用外部 API。
-
Graph:Graph 就是最終的圖結構,由衆多 Node 通過特定邏輯連接而成,定義了智能體的整體工作流程,各 Node 在其中協同運作,推動 AI Agent 完成複雜任務。
- 把大象裝進冰箱 ==========
接下來,我們通過一個例子,展示下如何使用 LangGraph 快速構建一個簡單的多步驟工作流。
相信大家一定看過趙本山和宋丹丹的經典小品《鐘點工》,裏面有這樣一段臺詞:
要把大象裝冰箱,總共分三步:第一步,把冰箱門打開;第二步,把大象裝進去;第三步,把冰箱門帶上。
首先,讓我們看看完整的實現代碼:
from IPython.display import Image, display
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict
# 1. 定義狀態
class ElephantInFridgeState(TypedDict):
"""
定義狀態——大象在冰箱中
狀態內的參數可以在多個節點間保存和流轉
"""
fridge_open: bool # 冰箱門是否打開
elephant_inside: bool # 大象是否在冰箱內
# 2. 定義每個步驟的處理節點
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第一步: 把冰箱門打開"""
print("正在打開冰箱門...")
# 執行當前節點業務邏輯
print("冰箱門已打開!")
# 更新狀態
state["fridge_open"] = True
# 當前節點處理完成,更新狀態
return state
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第二步: 把大象放進去"""
# 獲取當前狀態
if not state["fridge_open"]:
raise ValueError("冰箱門未打開,無法放入大象!")
# 執行當前節點業務邏輯
print("正在把大象放入冰箱...")
print("大象已放入冰箱!")
# 更新狀態
state["elephant_inside"] = True
# 當前節點處理完成,更新狀態
return state
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""第三步: 把冰箱門帶上"""
# 獲取當前狀態
if not state["elephant_inside"]:
print("警告:冰箱內沒有大象,是否確認關閉?")
# 執行當前節點業務邏輯
print("正在關閉冰箱門...")
print("冰箱門已關閉!")
# 更新狀態
state["fridge_open"] = False
# 當前節點處理完成,更新狀態
return state
# 3. 構建Graph工作流圖
def build_graph() -> CompiledStateGraph:
"""構建"把大象裝冰箱"的工作流程圖"""
# 創建圖對象StateGraph
graph = StateGraph(ElephantInFridgeState)
# 添加節點Node
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 添加起始邊
graph.add_edge(START, "open_fridge")
# 添加邊Edge,將節點連接起來,形成工作流
graph.add_edge("open_fridge", "put_elephant")
graph.add_edge("put_elephant", "close_fridge")
# 添加結束邊
graph.add_edge("close_fridge", END)
# 返回編譯好的圖Graph
return graph.compile()
# 4. 運行Graph工作流
def run_workflow():
"""運行"把大象裝冰箱"的工作流"""
# 構建圖
graph = build_graph()
# 定義初始狀態
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
# 運行圖
result = graph.invoke(initial_state)
# 輸出最終狀態
print("\n工作流執行完畢,最終狀態:")
print(f"冰箱門狀態: {'打開' if result["fridge_open"] else '關閉'}")
print(f"大象是否在冰箱內: {'是' if result["elephant_inside"] else '否'}")
# 可視化工作流圖
display(Image(graph.get_graph().draw_mermaid_png()))
return result
# 執行入口
if __name__ == "__main__":
final_state = run_workflow()
接下來,我們詳細分析下這段代碼的功能。
定義狀態類
我們首先創建了 ElephantInFridgeState
類,它繼承自 LangGraph 的 TypedDict
基類,是一種 key-value 的結構,用於管理整個工作流的狀態。這個類包含兩個 bool 類型屬性:
-
fridge_open:表示冰箱門是否打開
-
elephant_inside:表示大象是否在冰箱內
這些狀態變量將在各個節點之間傳遞,通過節點間的通信與協作,確保工作流的連貫性。
定義工作流節點
我們爲三個步驟分別創建了函數作爲節點:
-
open_fridge:負責將冰箱門狀態設置爲打開。
-
put_elephant:檢查冰箱門是否打開,並將大象放入冰箱。
-
close_fridge:關閉冰箱門,並在冰箱內沒有大象時發出警告。
我們可以看到,每個節點都遵循這樣的處理範式:
-
從狀態中獲取所需數據。
-
執行當前節點的業務邏輯。
-
更新狀態。
-
返回處理後的狀態。
這種設計確保了狀態在整個工作流中的一致性和可追蹤性。
構建工作流圖
build_graph
函數負責將各個節點連接成完整的工作流:
-
創建
StateGraph
實例 -
添加三個工作步驟作爲節點
-
通過 add_edge 方法添加邊,定義節點之間的執行順序
-
添加起始邊和結束邊。這裏 START 和 END 是 LangGraph 內置的常量,分表代表圖的起始和結束節點。
這種圖結構的設計讓我們可以輕鬆定義複雜的工作流,包括條件分支、循環、回退、異步等高級特性。
運行工作流
run_workflow
函數演示瞭如何編譯和執行工作流:
-
構建圖
-
初始化狀態
-
編譯圖並執行
-
輸出最終狀態
通過這種方式,我們可以清晰地看到每個步驟的執行結果,以及整個工作流的最終狀態。
這裏我還通過可視化的方式,打印了整個圖的結構:
運行結果
最後,我們執行入口函數,就可以看到如下的效果:
這個簡單的例子展示了 LangGraph 的核心功能:通過狀態管理、節點定義和圖構建,我們可以創建複雜的工作流,每個步驟都可以獨立開發和測試,同時保持整體的一致性和可控性。
- 小結 =====
通過對 LangGraph 的優勢解析、核心概念拆解及 “把大象裝冰箱” 的實戰演練,我們清晰看到:LangGraph 以簡潔高效的架構,將複雜 AI Agent 開發流程化繁爲簡。它通過 State、Node 和 Graph 的有機結合,賦予開發者精準掌控智能體行爲的能力,無論是基礎任務的有序推進,還是複雜場景的多智能體協作,都能從容應對。在 AI Agent 蓬勃發展的當下,LangGraph 不僅是開發者提升開發效率的利器,更爲創新應用的落地提供了堅實支撐,有望推動人工智能邁向更智能、更靈活的新階段。
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/yvGYDAWjPb96eto08CcJ8w