首页 › 参考文档

LangGraph(Agent 编排)完全指南:内容、能力与可用项目

参考

更新日期:2026-11-10 | 基于 LangGraph 1.0(2025-10 发布,当前稳定版) 官方文档:https://docs.langchain.com/oss/python/langgraph/overview | GitHub:https://github.com/langchain-ai/langgraph(约 45k star,MIT 许可,可商用)


一、一句话定位

LangGraph 是把 AI Agent 建模为"状态图(State Graph)"的低层编排框架——节点是步骤(调模型/调工具/调函数),边是控制流(顺序/条件/循环),全局状态在节点间流动并被持久化。它是 LangChain 官方对"链式调用(Chain)不适合 Agent"这一问题的回答:Agent 不是一条链,而是一个可以循环、可以中断、可以回退、可以人工接管的状态机。

是什么 不是什么
✅ 有状态、可循环、可持久化、可人工接管的 Agent 编排引擎 ❌ 不是模型(不训练、不推理,模型用 API 或本地)
✅ 从单 Agent 到多智能体的统一抽象 ❌ 不是低代码平台(写代码,可视化是辅助的 Studio)
✅ 生产级:断点续跑、容错、审计 ❌ 不是"提示词模板库"(那是 LangChain Hub 的事)

为什么值得你学(三个硬理由): 1. 市场定价最高:LangGraph 使用者中位年薪约 $209K,招聘 JD 最高频 Agent 技能词 2. 图结构/状态管理/checkpoint/human-in-the-loop 是可迁移到任何框架的工程思想(押注思想不押注 API) 3. 纯编排零算力:CPU 调 API 或本地 Ollama 即可完整学习——你的无 GPU 笔记本是合格学习机


二、核心思想:为什么是"图"而不是"链"

Chain(链):  输入 → 步骤A → 步骤B → 输出        (单向,无状态,一次过)
Graph(图):  输入 → [节点] ⇄ [节点] → [节点]      (可循环,共享状态,可中断)
                        ↑____条件边/人工审批____↓

真实 Agent 的本质动作就是循环:"思考 → 调工具 → 看结果 → 再思考"直到完成任务。链式框架写这种循环非常别扭;图把循环、条件分支、人工审批都变成一等公民。


三、核心概念详解(10 个,构成全部知识地图)

1. State(状态)—— 全局共享的记忆

用一个 TypedDict/Pydantic/dataclass 定义,所有节点读写同一份状态。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # add_messages = reducer:追加而非覆盖
    user_query: str
    approval: str | None

2. Node(节点)—— 干活的函数

接收状态、返回状态更新。可以包一次模型调用、一次工具执行、一段纯逻辑。

def think(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

3. Edge 与 Conditional Edge —— 控制流

  • 普通边:A 完了固定去 B
  • 条件边:根据状态动态路由(Agent 的"决策"就是条件边)
def should_continue(state: State):
    last = state["messages"][-1]
    return "tools" if last.tool_calls else END

graph.add_conditional_edges("agent", should_continue, ["tools", END])

4. Reducer —— 状态合并规则

Annotated[list, add_messages] 里的 add_messages 就是 reducer:定义"节点返回的更新如何并入全局状态"(覆盖?追加?合并?)。Reducer 是 LangGraph 并发安全与子图通信的基础。

5. Command 与 Send —— 动态控制

  • Command(goto="节点", update={...}):节点内部直接决定下一跳 + 更新状态
  • Send("节点", state):动态并行(Map-Reduce)——比如"把 10 个文档分发给 10 个并行节点各自总结"

6. Subgraph(子图)—— 复用与多智能体组合

一个图可以嵌进另一个图当一个节点——这是多智能体系统的组合方式:每个 Agent 是一个子图,主图负责调度。

7. Checkpointer(检查点)—— 持久化与断点续跑

每个超级步(superstep)后把状态存档(内存/SQLite/Postgres)。能力: - thread_id 会话隔离:同一图服务多个用户/对话 - 断点续跑:进程崩溃后从上次检查点恢复(durable execution) - 时间旅行:回到历史任一检查点,改状态重跑(调试神器)

from langgraph.checkpoint.memory import MemorySaver      # 学习用
from langgraph.checkpoint.sqlite import SqliteSaver     # 生产单机(你的笔记本就能跑)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "user-42"}}

8. interrupt() 与 Human-in-the-loop —— 人工审批

图可以在任意点暂停,等人类批准/修改/拒绝后用 Command(resume=...) 继续。这是企业级 Agent 的灵魂功能(高风险操作必须人签字)。

from langgraph.types import interrupt, Command

def risky_tool(state: State):
    decision = interrupt({"action": "转账", "amount": state["amount"]})  # 图在此暂停
    if decision["approved"]:
        ...  # 执行
    return {"approval": decision["status"]}

9. Streaming —— 四种流式粒度

stream_mode:values(每步全量状态)/ updates(每步增量)/ messages(token 级)/ custom(自定义进度)。前端打字机效果、进度条都靠它。

10. 多智能体模式(官方三模式)

模式 结构 适用
Supervisor(主管) 一个调度 Agent 把任务分给专家 Agent 任务可明确分工(最常用)
Swarm(蜂群) Agent 之间互相移交控制权(Handoff) 客服转接类场景
Network(网状) 任意 Agent 可通知任意 Agent 研究探索类,可靠性最难控

四、两套 API:高层 create_agent vs 低层 StateGraph

LangGraph 1.0 后的正确姿势:先用高层 API 起步,需要精细控制时下沉到低层。

高层:create_agent(替代旧 create_react_agent,10 行一个标准 Agent)

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_agent

model = init_chat_model("deepseek-chat", model_provider="deepseek")  # 或本地 ollama
agent = create_agent(
    model=model,
    tools=[search_tool, calculator_tool],
    system_prompt="你是一个严谨的研究助手,回答前先检索。",
)
agent.invoke({"messages": [{"role": "user", "content": "对比 LangGraph 与 AutoGen"}]})

自带运行时:持久化、短期记忆、流式、human-in-the-loop 钩子、1.0 新增的 Middleware API(在模型调用前后插逻辑,如敏感词过滤、工具审批)。

低层:StateGraph(当默认循环不满足时)

自定义节点/条件边/reducer/interrupt——上面第三节的所有能力在这一层。面试和生产里的复杂场景(审批流、多租户、并行 Map-Reduce)都在这层写。


五、配套生态(学习时一并认识)

组件 是什么 你需要吗
LangSmith 追踪/评估/监控平台(SaaS,有免费额度) 学习期强烈建议:可视化每次图执行,调试效率×10
LangGraph Studio 可视化图执行器(VS Code 插件/桌面端) 强烈建议:看着图跑,学习曲线最陡的部分变平
LangGraph Platform/Server 托管部署(持久化+队列+定时任务) 了解即可;你可以在笔记本上用 langgraph dev 本地起服务
MCP 适配 把 MCP 工具服务器接进 Agent 是:MCP 是协议层方向,两者组合是市场最高频组合
deepagents LangChain 官方"深度研究 Agent"模板(规划+子代理+文件系统) 进阶用:做投研/报告类项目直接复用

六、能用于什么项目(重点:12 个,按变现距离排序)

全部可在你的 i7-8550U/16GB 机器上开发(模型走 DeepSeek/Qwen API 或本地 Ollama 1.5B 测试)。★=与"长远价值项目"报告的 S/A 级对应。

第一梯队:直接能接单变现

1. 企业知识库问答 Agent(★对应 S1) - 做什么:上传企业文档 → RAG 检索 → 带引用溯源回答;敏感操作(删除/对外发送)走人工审批 - 用到的特性:条件边(检索足够则答/不足则追问)、Checkpointer(多租户 thread_id)、interrupt(管理员审核)、Subgraph(检索子图) - 难度 ★★★ | 变现:外包 MVP 2万-15万/单 - 为什么用 LangGraph:简单 RAG 链谁都会做,"检索不满意自动改写重查、多文档冲突自动仲裁、管理员审批"这套循环只有图能优雅表达——这就是溢价所在

2. 企微/微信智能客服 Agent(★对应 S3) - 做什么:接入企微 webhook,意图识别 → 知识库回答 → 转人工 Handoff(Swarm 模式) - 用到的特性:Swarm 移交、Checkpointer(跨天会话)、Streaming(打字机) - 难度 ★★★ | 变现:定制开发 1万-20万/单,2026 微信 AI 生态窗口期

3. 票据/文档处理流水线(★对应 S2 的 Agent 化升级) - 做什么:OCR 结果 → 字段校验 Agent → 异常单据人工审核 → 入账;批量单据用 Send 并行处理 - 用到的特性:Send(并行 Map-Reduce)、interrupt(异常审批)、Checkpointer(断点续跑——批量任务半夜崩了白天接着跑) - 难度 ★★ | 变现:与你的 S2 项目直接融合,客单价从"工具订阅"升级到"自动化流程交付"

第二梯队:作品集/接单进阶

4. Deep Research 投研/行业报告 Agent - 做什么:给定主题 → 规划提纲 → 并行检索 → 交叉验证 → 生成带引用报告(deepagents 模板) - 特性:Subgraph(规划/检索/写作三子图)、Send 并行、Reducer 合并多源结果 - 难度 ★★★★ | 变现:订阅制报告服务/卖给咨询与公关公司

5. NL2SQL 数据分析 Agent - 做什么:自然语言 → 生成 SQL → 执行 → 图表 → 异常自动重试修正 - 特性:条件边重试循环、interrupt(危险 SQL 人工放行——这个审批设计是企业敢用的唯一理由) - 难度 ★★★★ | 变现:企业内部工具外包

6. 邮件/公文自动化处理 Agent - 做什么:收件分类 → 起草回复 → 人工审批 → 发送;周报汇总 - 特性:Supervisor 多智能体(分类员/起草员/审核员)、HITL、持久化 - 难度 ★★★ | 变现:行政/外贸场景小工具订阅

7. 代码审查 Agent - 做什么:接 Git webhook → diff 分析 → 安全/风格/逻辑三专家并行审查 → 汇总评论 - 特性:Send 并行三专家、Reducer 合并意见、GitHub API 工具 - 难度 ★★★ | 变现:开源涨星 + 团队版订阅

第三梯队:学习与探索

8. 多智能体内容工厂:选题→写作→配图→审核流水线(Supervisor 模式,每个角色一个子图) 9. 个人知识库二脑:笔记自动归档/关联/每日摘要(Checkpointer 长期记忆 + 向量检索) 10. 游戏 NPC 编排:多 NPC 状态机 + 记忆(探索向,作品集亮点) 11. Agent 评测流水线:用 LangGraph 编排"批量跑测试集 → LLM-as-judge 评分 → 出报告"(对应评估方向 C17,稀缺技能) 12. 运维巡检 Agent:定时任务(Platform Cron)→ 检查服务 → 异常时 interrupt 通知人工 → 恢复后续跑

特性 × 项目映射速查

LangGraph 特性 没有它会怎样 哪些项目最依赖
Checkpointer 持久化 长任务崩了从头再来 1/3/5/12
interrupt 人工审批 企业不敢上生产 1/3/5/6/12
Send 并行 批量任务串行慢 10 倍 3/4/7/8
Subgraph 多智能体变面条代码 1/4/8
Swarm 移交 客服场景转接混乱 2
Streaming 前端体验差 全部
Middleware 敏感词/审批逻辑散落各处 1/2/6

七、与竞品对比(何时选 LangGraph)

框架 强项 弱项 选它当
LangGraph 状态/持久化/审批/流式最完整;生产级最成熟;生态最大 概念多、学习曲线最陡 要上生产、要审批流、要长任务
AutoGen 多智能体对话学术最强 生产持久化弱于 LangGraph 多 Agent 对话研究
CrewAI 角色分工上手最快 深度控制弱 快速 demo
OpenAI Agents SDK 极简、官方 绑 OpenAI 生态、持久化薄 快速验证
手写循环 零依赖 状态/恢复/审计全部自造轮子 只做一次性脚本

结论:写"能卖给企业的 Agent"→ LangGraph 是当前默认答案;它的学习成本会在第一个需要"断点续跑"或"人工审批"的项目上一次性赚回来。


八、常见坑(提前避开)

  1. 旧教程陷阱:2025-10 之前的教程用 create_react_agent/langchain.chains——1.0 已改为 create_agent/middleware,看文档先看版本
  2. 忘记 checkpointer:interrupt() 必须配 checkpointer(暂停后状态要能恢复),报错十有八九是这个
  3. Reducer 用错:默认行为是覆盖,列表类状态必须 Annotated[list, add_messages],否则并发节点互相覆盖
  4. thread_id 混用:多用户系统忘传 thread_id → 所有用户共享一个记忆(事故级 bug)
  5. 在节点里写死业务:图结构(流程)与节点内容(业务)分离,否则图改不动
  6. 本地开发端口:langgraph dev 默认 2024 端口,Windows 防火墙记得放行

九、最小可运行示例(在你的笔记本上直接跑)

安装(C 盘或 D 盘环境,约 200MB):

pip install -U langgraph langchain langchain-deepseek
set DEEPSEEK_API_KEY=你的key   # 或用 langchain-ollama 接本地模型,零 API 成本

示例 A:10 行工具 Agent(高层 API)

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_agent

@tool
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

agent = create_agent(
    model=init_chat_model("deepseek-chat", model_provider="deepseek"),
    tools=[add],
    system_prompt="你是计算助手,必须用工具计算。",
)
result = agent.invoke({"messages": [{"role": "user", "content": "3 加 5 等于几?"}]})
print(result["messages"][-1].content)

示例 B:带人工审批的状态图(低层 API,企业级核心模式)

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    messages: Annotated[list, add_messages]
    amount: int

def draft(state: State):          # 节点1:起草转账请求
    return {"amount": 5000}

def approve(state: State):        # 节点2:人工审批(图在此暂停)
    decision = interrupt({"question": "批准转账?", "金额": state["amount"]})
    if not decision["approved"]:
        return {"messages": [{"role": "assistant", "content": "已拒绝"}]}
    return {"messages": [{"role": "assistant", "content": "已执行"}]}

builder = StateGraph(State)
builder.add_node("draft", draft)
builder.add_node("approve", approve)
builder.add_edge(START, "draft")
builder.add_conditional_edges("draft", lambda s: "approve")  # 条件边(此处恒真,实战按金额路由)
builder.add_edge("approve", END)

graph = builder.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "demo-1"}}

r1 = graph.invoke({"messages": [], "amount": 0}, cfg)         # 第一次运行:暂停在审批
print("暂停,等待人工决策")
r2 = graph.invoke(Command(resume={"approved": True}), cfg)    # 人类批准后:继续执行
print(r2["messages"][-1].content)                              # → 已执行

本地可视化:pip install "langgraph-cli[inmem]" → langgraph dev → 浏览器打开 Studio,看着图一步步跑。


十、四周学习路线(配合你的机器)

周 目标 产出
第1周 高层 API:create_agent + 工具 + Ollama 本地模型跑通 示例 A 变体 3 个(计算器/搜索/天气)
第2周 低层 API:State/条件边/Reducer/Send 手写一个"并行总结 10 篇文章"的图
第3周 Checkpointer + interrupt + Studio 调试 示例 B 完整版 + 一个"危险操作审批"Agent
第4周 Subgraph + Supervisor 多智能体 + MCP 接入 多智能体客服(知识库 Agent + 转人工)→ 这就是 S1 项目的骨架

模型供给:学习期用本地 Ollama(qwen3:1.7b,10-18 tok/s,零成本);需要质量时切 DeepSeek API(约 1 元/百万 token,一个月学习成本 <10 元)。


十一、资源清单

  • 官方文档(Python):https://docs.langchain.com/oss/python/langgraph/overview
  • v1.0 变更说明:https://docs.langchain.com/oss/python/releases/langgraph-v1
  • 1.0发布公告:https://www.langchain.com/blog/langchain-langgraph-1dot0
  • GitHub:https://github.com/langchain-ai/langgraph | 深度研究模板:https://github.com/langchain-ai/deep-agents
  • 官方教程里的经典范例(建议顺序):01-create-agent → persistence → human-in-the-loop → map-reduce → multi-agent supervisor