16 - LangGraph 基础
模块 16:LangGraph 基础
🎯 学习目标
本模块将帮助你理解 LangGraph 1.0 的核心概念,学会创建状态图来构建复杂的 AI 工作流。
📚 核心概念
什么是 LangGraph?
LangGraph 是一个用于构建状态化、多步骤 AI 应用的框架。它使用图(Graph) 的概念来组织工作流:
- 节点(Nodes):图中的处理单元,可以是 LLM 调用、工具执行或自定义函数
- 边(Edges):连接节点的路径,定义执行顺序
- 状态(State):在节点之间传递的数据结构
LangGraph vs LangChain
| 特性 | LangChain | LangGraph |
|---|---|---|
| 抽象级别 | 高级 | 低级 |
| 适用场景 | 快速构建标准 Agent | 复杂自定义工作流 |
| 控制粒度 | 通过中间件 | 完全控制每个节点 |
| 状态管理 | 自动 | 手动但灵活 |
核心组件
# 1. 定义状态 - 使用 TypedDict
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # 消息列表,自动累加
current_step: str # 当前步骤
# 2. 定义节点 - 接收和返回状态的函数
def my_node(state: State) -> dict:
# 处理逻辑
return {"current_step": "completed"}
# 3. 创建图
from langgraph.graph import StateGraph
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.add_edge(START, "my_node")
graph.add_edge("my_node", END)
# 4. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [], "current_step": "start"})🔑 关键 API
StateGraph
from langgraph.graph import StateGraph, START, END
# 创建图
graph = StateGraph(State)
# 添加节点
graph.add_node("node_name", node_function)
# 添加边
graph.add_edge("from_node", "to_node") # 普通边
graph.add_edge(START, "first_node") # 从入口开始
graph.add_edge("last_node", END) # 到出口结束
# 添加条件边
graph.add_conditional_edges(
"from_node",
condition_function, # 返回下一个节点名称
{"option1": "node1", "option2": "node2"}
)add_messages 注解
from langgraph.graph.message import add_messages
class State(TypedDict):
# add_messages 确保消息被追加而不是替换
messages: Annotated[list, add_messages]编译选项
from langgraph.checkpoint.memory import MemorySaver
# 添加内存检查点(用于持久化)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
# 使用 thread_id 进行会话管理
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(input_data, config=config)📝 本模块示例
main.py
包含三个递进的示例:
- 简单顺序图:展示基本的节点和边
- 条件分支图:根据条件选择不同路径
- 带内存的对话图:实现多轮对话
🧪 练习
- 修改
simple_workflow添加一个新的处理节点 - 在
conditional_workflow中添加第三个分支 - 扩展
conversation_workflow支持更多类型的用户意图
📖 延伸阅读
⚠️ 注意事项
- 状态更新是合并式的,只需返回要更新的字段
- 使用
add_messages注解时,消息会自动追加 - 条件函数必须返回有效的节点名称字符串
- 编译后的图是不可变的,需要修改时重新编译