LangGraph
LangChain 团队出品的低层级 Agent 编排框架,把 AI 工作流建模为有状态的有向图(StateGraph),原生支持循环、条件路由、持久化检查点和 Human-in-the-Loop,是 2026 年构建生产级 AI Agent 的主流选择。
🎯 为什么需要它
LangChain 的 Chain 是线性的:A → B → C → 结束。但真实 AI 任务往往需要:循环重试、条件分支、等待人工审批、跨会话记忆。LangGraph 用图结构解决了这个问题——每个节点是一个推理/行动步骤,边控制流转,状态在整个图中自动传递,不需要手动传参。
[!tip] 一句话心智模型 LangGraph = 状态机 + LLM 大脑 State(记忆)→ Node(推理/行动)→ Edge(控制流)→ Conditional Edge(决策)→ Loop(思考)
✅ 核心优势
- 持久化执行(Durable Execution):Agent 崩溃后可从检查点恢复,支持长时间运行的任务
- Human-in-the-Loop:原生支持在任意节点暂停图执行、等待人工输入后继续,★★★★★
- 完整状态管理:短期工作记忆(当前推理)+ 长期跨会话记忆,统一管理
- 流式输出:支持 token 级别和节点级别的实时流式响应
- 成本可预测:显式节点结构让每次 LLM 调用都是已知量,避免失控循环
- 可观测性:与 LangSmith 深度集成,完整追踪执行路径、状态转换、耗时和费用
- 不依赖 LangChain:可以单独使用,不强制绑定 LangChain 生态
🧩 核心概念
State(状态)
图的”记忆”,用 TypedDict 定义,在所有节点间共享和传递。节点返回的字典会合并(而非替换)到状态中。
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # add_messages 是 reducer,追加而非覆盖
topic: str
draft: str
iterations: int
Node(节点)
图中的一个处理单元,接收 State,返回要更新的字段。可以是 LLM 调用、工具调用、任意 Python 函数。
def writer_node(state: State) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response], "draft": response.content}
Edge(边)
连接节点的有向边,分两种:
| 类型 | 说明 | 用法 |
|---|---|---|
| 普通边 | 固定流转 A → B | graph.add_edge("A", "B") |
| 条件边 | 根据状态动态路由 | graph.add_conditional_edges("A", router_fn, {"x": "B", "y": "C"}) |
Checkpointer(检查点)
每次节点执行后自动保存状态快照,支持:
- 内存存储(
MemorySaver,开发用) - 数据库持久化(
SqliteSaver、PostgresSaver,生产用) - 通过
thread_id实现多会话隔离
🚀 快速上手
安装
pip install langgraph
# 如需 LangChain 集成
pip install langchain langchain-openai
最小示例:Hello World
from langgraph.graph import StateGraph, MessagesState, START, END
def mock_llm(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "hi!"}]})
条件路由示例
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
question: str
def route(state: State) -> Literal["code_node", "explain_node"]:
if "code" in state["question"].lower():
return "code_node"
return "explain_node"
def code_node(state: State) -> dict:
return {"messages": [llm.invoke(f"Write code for: {state['question']}")]}
def explain_node(state: State) -> dict:
return {"messages": [llm.invoke(state["question"])]}
graph = StateGraph(State)
graph.add_node("code_node", code_node)
graph.add_node("explain_node", explain_node)
graph.set_entry_point("router")
graph.add_conditional_edges("router", route, {
"code_node": "code_node",
"explain_node": "explain_node"
})
graph.add_edge("code_node", END)
graph.add_edge("explain_node", END)
app = graph.compile()
循环 + 检查点(持久化记忆)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
# 同一 thread_id = 同一会话,状态跨调用持久化
config = {"configurable": {"thread_id": "user-session-001"}}
result = app.invoke({"messages": [HumanMessage(content="你好")]}, config)
流式输出
# 流式获取每个节点的输出
for chunk in app.stream({"messages": [...]}, config):
for node_name, node_output in chunk.items():
print(f"[{node_name}]: {node_output}")
# 异步流式(token 级别)
async for event in app.astream_events({"messages": [...]}, version="v2"):
if event["event"] == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="", flush=True)
📦 典型使用场景
| 场景 | 说明 |
|---|---|
| ReAct Agent | 思考 → 调用工具 → 观察结果 → 循环,直到得出答案 |
| RAG with Retry | 检索 → 评估质量 → 质量差则重新检索或改写查询 |
| 多 Agent 协作 | Supervisor 节点分配任务给专业子 Agent,汇总结果 |
| 审批工作流 | 生成内容 → 暂停等待人工审核 → 通过后继续发布 |
| AI 面试机器人 | 多轮对话 + 状态追踪 + 评分循环 |
| 代码生成 + 测试 | 生成代码 → 运行测试 → 失败则修复 → 循环直到通过 |
| 长文档处理 | 分块处理 + 状态聚合,支持超长任务不丢失上下文 |
🔗 与 LangChain 的关系
[!info] 关系说明 LangGraph 由 LangChain 团队构建,但两者是独立的框架,可以单独使用。
| 维度 | LangChain | LangGraph |
|---|---|---|
| 定位 | 高层抽象,快速组装 LLM 应用 | 低层编排,构建有状态 Agent |
| 工作流 | 线性 Chain(A→B→C) | 有向图(支持循环、分支) |
| 状态管理 | 无内置状态机制 | 核心特性,自动管理 |
| 适合场景 | 快速原型、简单 RAG、工具调用 | 复杂 Agent、多步推理、生产部署 |
| 依赖关系 | 独立 | 可选集成 LangChain 组件 |
官方建议:入门或简单场景用 LangChain 的预置 Agent;需要持久化、循环、Human-in-the-Loop 时升级到 LangGraph。
🆚 竞品对比
| 维度 | LangGraph | CrewAI | AutoGen |
|---|---|---|---|
| 抽象层级 | 低(显式图结构) | 高(角色/任务) | 中(对话驱动) |
| 学习曲线 | 较陡 | 平缓 | 中等 |
| Human-in-the-Loop | ★★★★★ 原生支持 | ★★ 需自定义 | ★★★ 代理模式 |
| 成本可预测性 | ★★★★★ | ★★★ | ★★ 循环失控风险 |
| 可观测性 | ★★★★★ LangSmith | ★★★ | ★★★ Azure Monitor |
| 生产可靠性 | ★★★★★ | ★★★ | ★★★★ |
| 快速原型 | ★★★ | ★★★★★ | ★★★★ |
| 企业生态 | LangChain 生态 200+ 集成 | 独立,集成较少 | Microsoft 生态 |
⚠️ 缺点 & 注意事项
- 学习曲线陡:图结构思维对习惯线性代码的开发者不直观,需要时间适应
- 调试复杂:多节点状态流转时,不借助 LangSmith 很难追踪问题
- 过度设计风险:简单的单次 LLM 调用用 LangGraph 是杀鸡用牛刀
- 版本迭代快:API 变化频繁,升级时注意 breaking changes
- 内存占用:长对话 +
add_messages模式下,消息列表会无限增长,需要手动截断 - TypeScript 版本滞后:Python 版功能更完整,TS 版部分特性有延迟
[!warning] 常见坑 使用
add_messagesreducer 时,消息会无限追加。长对话场景必须实现消息截断策略,否则 token 消耗会爆炸。
👥 适合什么人 / 项目
适合:
- 需要构建生产级 AI Agent 的工程师
- 工作流包含循环、条件分支、重试逻辑的项目
- 需要Human-in-the-Loop(人工审批、人工干预)的业务场景
- 已在使用 LangChain 生态,需要升级到更复杂 Agent 架构
- 企业级应用,需要可观测性、持久化、成本控制
不适合:
- 简单的单次 LLM 调用或基础 RAG(用 LangChain 或直接调 API 更简单)
- 快速原型验证(CrewAI 上手更快)
- 团队完全没有 Python 异步编程经验
🌍 生态 & 社区
- 维护状态:极活跃,LangChain Inc 核心产品,持续高频发布
- LangSmith:官方配套的 Agent 调试/监控平台,与 LangGraph 深度集成
- LangGraph Platform:托管部署方案(付费),提供 API 端点、扩缩容、持久化存储
- LangChain Academy:官方免费课程,系统学习 LangGraph
- 社区:14k+ Stars,Discord 活跃,案例覆盖 Klarna、Replit、Elastic 等企业
💡 引入评估
| 维度 | 评分(/5) | 备注 |
|---|---|---|
| 上手难度 | ⭐⭐⭐ | 图结构思维需要适应期 |
| 文档完善度 | ⭐⭐⭐⭐⭐ | 官方文档 + Academy 课程质量高 |
| 社区活跃 | ⭐⭐⭐⭐⭐ | LangChain 生态背书 |
| 生产可靠性 | ⭐⭐⭐⭐⭐ | 企业级案例验证 |
| 功能完整性 | ⭐⭐⭐⭐⭐ | Human-in-Loop、持久化、流式全覆盖 |
| 综合 | ⭐⭐⭐⭐⭐ | 复杂 Agent 场景的首选框架 |
结论:强烈推荐用于生产级 AI Agent 开发。如果你的 Agent 需要循环推理、条件路由、人工干预或跨会话记忆,LangGraph 是 2026 年最成熟的选择。简单场景不必引入。
推荐引入版本:pip install -U langgraph(当前最新稳定版)
🔗 相关链接
- LangChain AI Agent 状态机 rag Human-in-the-Loop LLM编排
- 官方文档
- GitHub 仓库
- LangChain Academy(免费课程)
- LangSmith(调试平台)
- 案例研究
📝 个人备注
(留白,供后续补充实际使用心得)