MD 状态:评估中 更新:2026/5/12

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 → Bgraph.add_edge("A", "B")
条件边根据状态动态路由graph.add_conditional_edges("A", router_fn, {"x": "B", "y": "C"})

Checkpointer(检查点)

每次节点执行后自动保存状态快照,支持:

  • 内存存储(MemorySaver,开发用)
  • 数据库持久化(SqliteSaverPostgresSaver,生产用)
  • 通过 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 团队构建,但两者是独立的框架,可以单独使用。

维度LangChainLangGraph
定位高层抽象,快速组装 LLM 应用低层编排,构建有状态 Agent
工作流线性 Chain(A→B→C)有向图(支持循环、分支)
状态管理无内置状态机制核心特性,自动管理
适合场景快速原型、简单 RAG、工具调用复杂 Agent、多步推理、生产部署
依赖关系独立可选集成 LangChain 组件

官方建议:入门或简单场景用 LangChain 的预置 Agent;需要持久化、循环、Human-in-the-Loop 时升级到 LangGraph。

🆚 竞品对比

维度LangGraphCrewAIAutoGen
抽象层级低(显式图结构)高(角色/任务)中(对话驱动)
学习曲线较陡平缓中等
Human-in-the-Loop★★★★★ 原生支持★★ 需自定义★★★ 代理模式
成本可预测性★★★★★★★★★★ 循环失控风险
可观测性★★★★★ LangSmith★★★★★★ Azure Monitor
生产可靠性★★★★★★★★★★★★
快速原型★★★★★★★★★★★★
企业生态LangChain 生态 200+ 集成独立,集成较少Microsoft 生态

⚠️ 缺点 & 注意事项

  • 学习曲线陡:图结构思维对习惯线性代码的开发者不直观,需要时间适应
  • 调试复杂:多节点状态流转时,不借助 LangSmith 很难追踪问题
  • 过度设计风险:简单的单次 LLM 调用用 LangGraph 是杀鸡用牛刀
  • 版本迭代快:API 变化频繁,升级时注意 breaking changes
  • 内存占用:长对话 + add_messages 模式下,消息列表会无限增长,需要手动截断
  • TypeScript 版本滞后:Python 版功能更完整,TS 版部分特性有延迟

[!warning] 常见坑 使用 add_messages reducer 时,消息会无限追加。长对话场景必须实现消息截断策略,否则 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(当前最新稳定版)

🔗 相关链接

📝 个人备注

(留白,供后续补充实际使用心得)

学习路径

AI 工具链路径

当前:LangGraph