更新日期: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 是当前默认答案;它的学习成本会在第一个需要"断点续跑"或"人工审批"的项目上一次性赚回来。
八、常见坑(提前避开)
- 旧教程陷阱:2025-10 之前的教程用
create_react_agent/langchain.chains——1.0 已改为create_agent/middleware,看文档先看版本 - 忘记 checkpointer:
interrupt()必须配 checkpointer(暂停后状态要能恢复),报错十有八九是这个 - Reducer 用错:默认行为是覆盖,列表类状态必须
Annotated[list, add_messages],否则并发节点互相覆盖 - thread_id 混用:多用户系统忘传 thread_id → 所有用户共享一个记忆(事故级 bug)
- 在节点里写死业务:图结构(流程)与节点内容(业务)分离,否则图改不动
- 本地开发端口:
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