- 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 不仅是开发者提升开发效率的利器,更为创新应用的落地提供了坚实支撑,有望推动人工智能迈向更智能、更灵活的新阶段。