LangGraph-00-总览
0. 摘要
项目目标
LangGraph 是一个低级编排框架,用于构建、管理和部署长期运行的有状态Agent。作为一个基于图的计算框架,LangGraph 受到 Google 的 Pregel 和 Apache Beam 的启发,专门为处理复杂的多步骤工作流程和 Agent 执行而设计。
核心能力边界
- 持久化执行:构建能够在故障中持续运行并可以长期运行的 Agent,自动从中断点恢复
- 人机交互:在执行过程中的任何时刻无缝整合人类监督,检查和修改 Agent 状态
- 全面记忆:创建真正有状态的 Agent,具备短期工作记忆和跨会话的长期持久记忆
- 可视化调试:通过 LangSmith 深入了解复杂 Agent 行为,提供执行路径跟踪和详细运行时指标
- 生产就绪部署:具备处理有状态长期运行工作流程的可扩展基础设施
非目标
- 不抽象化提示词或架构设计
- 不提供开箱即用的预训练模型
- 不处理模型训练或微调
- 不强制特定的 Agent 架构模式
运行环境
- 语言:Python 3.9+ 和 TypeScript/JavaScript
- 运行时:支持同步和异步执行模式
- 依赖:基于 LangChain 生态系统,可独立使用
- 部署形态:支持单体应用、微服务、Sidecar 模式以及通过 LangGraph Platform 的容器化部署
1. 整体架构图
flowchart TB
subgraph "用户层"
USER[用户应用]
STUDIO[LangGraph Studio]
CLI[LangGraph CLI]
end
subgraph "API层"
PREBUILT[prebuilt<br/>高级API]
SDKPY[sdk-py<br/>Python SDK]
SDKJS[sdk-js<br/>JS/TS SDK]
end
subgraph "核心框架"
LANGGRAPH[langgraph<br/>核心框架]
PREGEL[Pregel引擎]
GRAPH[图构建器]
RUNTIME[运行时]
end
subgraph "持久化层"
CHECKPOINT[checkpoint<br/>检查点接口]
CPPOSTGRES[checkpoint-postgres]
CPSQLITE[checkpoint-sqlite]
STORE[存储系统]
CACHE[缓存系统]
end
subgraph "外部服务"
LLM[LLM模型]
TOOLS[外部工具]
DB[(数据库)]
LANGSMITH[LangSmith]
end
USER --> PREBUILT
USER --> SDKPY
USER --> SDKJS
STUDIO --> CLI
CLI --> LANGGRAPH
PREBUILT --> LANGGRAPH
SDKPY --> LANGGRAPH
SDKJS --> LANGGRAPH
LANGGRAPH --> PREGEL
LANGGRAPH --> GRAPH
LANGGRAPH --> RUNTIME
LANGGRAPH --> CHECKPOINT
CHECKPOINT --> CPPOSTGRES
CHECKPOINT --> CPSQLITE
CHECKPOINT --> STORE
CHECKPOINT --> CACHE
RUNTIME --> LLM
RUNTIME --> TOOLS
CPPOSTGRES --> DB
CPSQLITE --> DB
LANGGRAPH --> LANGSMITH
图解与要点
组件职责与耦合关系:
- 用户层:提供多种交互方式,支持可视化开发(Studio)、命令行操作(CLI)和程序化集成
- API层:prebuilt 提供高级抽象,SDK 提供编程接口,均为同步/异步模式
- 核心框架:langgraph 实现基于 Pregel 的图计算模型,支持事件驱动的节点执行
- 持久化层:checkpoint 系统提供状态持久化,支持多种存储后端,基于异步I/O模式
数据流与控制流:
- 控制流:用户请求 → API层路由 → 核心引擎执行 → 状态检查点保存
- 数据流:图状态在节点间传递,检查点数据跨进程/线程同步
- 跨协程路径:支持大规模并发执行,状态管理线程安全
高可用、扩展性与状态管理:
- 状态管理位置:检查点系统集中管理所有状态
- 扩展性:支持水平扩展,通过数据库实现状态共享
- 高可用:检查点机制保证故障恢复能力
2. 全局时序图
sequenceDiagram
autonumber
participant User as 用户
participant API as API层
participant Core as 核心引擎
participant Checkpoint as 检查点系统
participant LLM as 语言模型
participant Tools as 外部工具
participant Store as 存储系统
User->>API: 提交请求(state, config)
API->>Core: 创建图实例
Core->>Checkpoint: 加载检查点状态
Checkpoint->>Store: 查询历史状态
Store-->>Checkpoint: 返回状态数据
Checkpoint-->>Core: 恢复图状态
loop 图执行循环
Core->>Core: 计算可执行节点
Core->>LLM: 调用语言模型
LLM-->>Core: 返回响应
Core->>Tools: 执行工具调用
Tools-->>Core: 返回工具结果
Core->>Checkpoint: 保存中间状态
Checkpoint->>Store: 持久化状态
end
Core->>API: 返回最终结果
API-->>User: 响应用户请求
图解与要点
入口与鉴权:
- 请求通过 API 层统一入口,支持配置化鉴权机制
- 检查点加载时验证 thread_id 和 checkpoint_id 有效性
幂等性与重试:
- 每个超步骤(superstep)生成唯一检查点,支持精确重试
- 状态更新采用乐观并发控制,避免状态冲突
回退策略与超时:
- 节点执行失败时自动回退到上一个检查点
- 支持节点级别的超时配置,防止死锁
- 工具调用失败时提供优雅降级机制
资源上界:
- 执行步数限制(recursion_limit)防止无限循环
- 内存使用监控,大状态自动压缩存储
- 并发节点数量可配置限制
3. 模块边界与交互矩阵
模块列表与职责
| 模块 | 职责 | 对外API | 调用方式 |
|---|---|---|---|
| langgraph | 核心图计算引擎 | StateGraph, MessageGraph, compile() | 同步/异步 |
| checkpoint | 状态持久化接口定义 | BaseCheckpointSaver, CheckpointTuple | 同步/异步 |
| checkpoint-postgres | PostgreSQL存储实现 | PostgresSaver | 异步I/O |
| checkpoint-sqlite | SQLite存储实现 | SqliteSaver | 同步/异步 |
| prebuilt | 高级Agent构建工具 | create_react_agent, ToolNode | 同步/异步 |
| cli | 命令行工具 | 部署、开发、调试命令 | 进程调用 |
| sdk-py | Python平台SDK | LangGraphClient | HTTP/WebSocket |
| sdk-js | JS/TS平台SDK | LangGraphClient | HTTP/WebSocket |
交互关系矩阵
| 调用方 | 被调方 | 接口类型 | 错误语义 | 一致性要求 |
|---|---|---|---|---|
| prebuilt → langgraph | StateGraph.compile() | 同步调用 | 异常传播 | 强一致性 |
| langgraph → checkpoint | BaseCheckpointSaver | 异步调用 | 重试3次 | 最终一致性 |
| checkpoint-postgres → langgraph | 检查点回调 | 事件订阅 | 忽略失败 | 最终一致性 |
| sdk-py → langgraph | HTTP API | 异步请求 | 状态码返回 | 会话一致性 |
| cli → langgraph | 进程调用 | 同步执行 | 退出码 | 强一致性 |
交互说明
同步调用:
- prebuilt 到 langgraph:直接函数调用,错误立即传播
- cli 工具执行:进程同步,通过退出码判断成功失败
异步消息:
- 检查点保存:非阻塞式,失败时重试机制
- 工具执行结果:通过消息队列异步通知
共享存储:
- 检查点数据:多实例间共享,支持读写分离
- 缓存系统:Redis或内存缓存,提升性能
订阅/发布:
- 状态变更事件:支持外部系统订阅图执行状态
- 执行监控:LangSmith集成,实时追踪执行轨迹
4. 关键设计与权衡
数据一致性策略
- 强一致性:图构建和编译阶段,保证配置正确性
- 最终一致性:检查点保存和状态同步,优化性能
- 事务边界:单个超步骤内的所有状态更新为原子操作
- 并发控制:基于版本号的乐观锁机制,避免状态冲突
性能关键路径
- P95延迟指标:
- 简单Agent执行:< 200ms
- 复杂工具调用链:< 2s
- 检查点保存:< 50ms
- 内存峰值管理:
- 大状态自动分页存储
- LRU缓存策略限制内存使用
- I/O热点优化:
- 检查点写入批量化
- 数据库连接池复用
- 异步I/O避免阻塞
可观测性指标
- 执行指标:步数、耗时、成功率
- 资源指标:内存使用、数据库连接数、缓存命中率
- 业务指标:工具调用成功率、用户交互响应时间
关键配置项
- recursion_limit:最大执行步数,默认25,影响Agent复杂度上限
- checkpointer:检查点保存策略,影响持久化性能
- interrupt_before/after:人机交互控制点,影响用户体验
- stream_mode:输出流模式,影响实时性和资源使用
5. 典型使用示例与最佳实践
示例1:最小可运行ReAct Agent
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
# 定义工具
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
return f"{city}今天天气晴朗,气温25度"
# 创建Agent
model = ChatOpenAI(model="gpt-4")
agent = create_react_agent(
model=model,
tools=[get_weather],
prompt="你是一个有用的助手"
)
# 执行对话
result = agent.invoke({
"messages": [{"role": "user", "content": "北京天气怎么样?"}]
})
print(result["messages"][-1]["content"])
示例2:带持久化的复杂工作流
from langgraph.graph import StateGraph
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, List
class AgentState(TypedDict):
messages: List[dict]
current_task: str
completed_tasks: List[str]
def task_planner(state: AgentState):
# 任务规划逻辑(此处省略具体实现)
return {"current_task": "analyze_data"}
def task_executor(state: AgentState):
# 任务执行逻辑(此处省略具体实现)
completed = state["completed_tasks"] + [state["current_task"]]
return {"completed_tasks": completed}
# 构建复杂工作流
workflow = StateGraph(AgentState)
workflow.add_node("planner", task_planner)
workflow.add_node("executor", task_executor)
workflow.add_edge("planner", "executor")
workflow.set_entry_point("planner")
# 配置持久化
checkpointer = SqliteSaver.from_conn_string("agent_state.db")
app = workflow.compile(checkpointer=checkpointer)
# 执行工作流
config = {"configurable": {"thread_id": "workflow_001"}}
result = app.invoke({
"messages": [{"role": "user", "content": "开始执行任务"}],
"current_task": "",
"completed_tasks": []
}, config=config)
示例3:生产环境部署配置
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import create_react_agent
import os
# 生产环境配置
DB_URI = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost/langgraph")
checkpointer = PostgresSaver.from_conn_string(DB_URI)
# 配置监控和限制
agent = create_react_agent(
model="openai:gpt-4",
tools=production_tools,
checkpointer=checkpointer,
interrupt_before=["tools"], # 工具执行前需要确认
debug=False # 生产环境关闭调试
)
# 配置资源限制
config = {
"recursion_limit": 50, # 增加复杂任务的步数限制
"configurable": {
"thread_id": f"user_{user_id}",
"checkpoint_id": None # 从最新状态开始
}
}
最佳实践清单
开发阶段
- 状态设计:保持状态结构简单,避免深度嵌套对象
- 工具设计:每个工具功能单一,错误处理完善
- 调试配置:开启debug模式,集成LangSmith追踪
测试阶段
- 单元测试:模拟工具调用,测试各个节点逻辑
- 集成测试:使用内存检查点,测试完整流程
- 性能测试:监控执行时间和内存使用
生产阶段
- 资源配置:适当的递归限制和超时设置
- 监控告警:集成指标监控,设置关键阈值告警
- 错误处理:完善的异常捕获和用户友好的错误提示
- 扩展准备:使用数据库检查点,支持水平扩展
通过以上架构设计和最佳实践,LangGraph提供了构建生产级AI Agent系统的完整解决方案,平衡了灵活性、性能和可维护性的需求。