用大白话 + 类比 + 流程图,把 LangGraph 的核心组件、工作原理和实际用途一次性讲清楚。不需要任何前置知识。
整个流程共享一份数据(叫 State),每个步骤都能读取和修改它。就像共享文档,所有人看到的是同一份最新版本。
State不像传统流水线只能一条路走到黑,LangGraph 的图可以“回头看”、反复执行,直到条件满足才结束。这正是 Agent 反复“思考→行动→观察”的基础。
Cyclic关键节点可以让流程暂停,等人确认后再继续。比如“模型想执行一个删除操作,先弹出来让你点同意”。
Human-in-the-loop一个共享的数据结构,在整张图中流转。通常是一个字典,里面装着消息列表、计数器、用户输入等。每个节点都能读它、改它。定义时用 TypedDict 描述有哪些字段,还可以用 Annotated 指定“新值是覆盖旧的,还是追加到后面”。
一个普通的 Python 函数,接收当前 State,执行一些操作(调模型、调工具、跑代码),然后返回对 State 的更新。节点是“干活的”,它不需要知道整张图长什么样。
干活的函数决定“下一步走哪个节点”。有两种:普通边(固定走某条路)和条件边(根据 State 的内容动态决定)。边本身不干活,只负责指路。
指路的规则每执行完一步,就把整个 State 快照存一份。有了它,流程可以断点续跑、可以“时间旅行”回到之前的某一步重新走、可以让人慢慢审批而不怕状态丢失。生产环境常用数据库来存。
自动存档在节点内部调用 interrupt(),流程就会停在这里,把控制权交还给调用方。等人输入后,用 Command(resume=...) 继续跑。这是“人在回路”的实现基础。
把一张已经建好的图,当作另一张大图里的一个“节点”来用。适合多 Agent 系统——每个 Agent 有自己的内部流程,主图只管调度它们。
图套图LangGraph 的执行不是一步一步逐个节点来的,而是分“轮次”。同一轮里可以并行跑多个节点,等它们全部跑完、状态合并完,才算一轮结束。这借鉴了 Google Pregel 图计算系统的思想。
节点一开始都是“睡着”的。当一条边把新状态传给它时,它才醒来干活,干完把结果沿着边传出去。所有节点都不活跃、也没有消息在路上时,整个图就结束。
| 维度 | LangChain | LangGraph |
|---|---|---|
| 定位 | 高级框架,帮你快速搭出 Agent | 底层编排引擎,给你细粒度控制 |
| 编程模型 | 链式调用(线性流水线) | 图结构(节点 + 边 + 循环) |
| 状态管理 | 靠函数参数传递 | 共享 State 对象,可持久化 |
| 适合场景 | 简单问答、RAG 检索、快速原型 | 复杂 Agent、多步循环、需要人工审批、长期运行 |
| 关系 | LangChain 构建在 LangGraph 之上。从 2025 年起,LangChain 自己的 Agent 底层就是 LangGraph。 | |
流程跑到一半崩溃了,重启后能从最后一次成功的地方继续,不用从头再来。对长时间运行的任务至关重要。
在关键操作前暂停,把待审批的内容展示给人。人点了“同意”才继续,点了“拒绝”就走另一条路。这是 AI 安全落地的核心机制。
因为每一步都存了档,你可以“回到”之前的任意一个检查点,用不同的输入重新走一遍,方便调试和探索不同方案。
不需要等整个流程跑完才看到结果。模型每吐一个字、每个节点每完成一步,都可以实时推送到前端显示。
通过子图机制,让多个 Agent 各自拥有独立的内部流程,由一个“主管”负责调度和移交任务。
先跑通一张两个节点的图,理解 State 怎么在节点间流动。然后对比“覆盖”和“追加”两种状态更新方式,搞明白 add_messages 到底在干什么。
学会用条件边做分支和循环,理解 recursion_limit 怎么防止死循环。然后实现最小的 ReAct 模式:模型决策 → 调工具 → 回到模型。最后用 Checkpointer 让同一会话连续对话、回放历史。
用 interrupt 让人在关键点审批;用四种 stream_mode 看不同粒度的输出;把一张图当作另一张图的节点来用,理解父子状态映射。
用一个纯函数调度器在两个专职 Agent 之间移交任务;用标准库 sqlite3 做关键词检索,实现“证据不足就再检索”的回环;最后把前面所有能力串成一个完整的“检索 → 起草 → 人工审批 → 成稿”应用。
如果 graph.py 里直接 ChatOpenAI() 构造客户端,测试就没法离线跑了。正确做法是让模型作为参数传进来(依赖注入),测试时传一个假模型。
compile()定义完节点和边之后,必须调用 .compile() 才能得到可执行的图。compile 还会做结构检查(比如有没有孤儿节点),并绑定 Checkpointer。
如果用 Annotated[list, add_messages],新消息会追加;如果只写 list,新消息会直接覆盖旧列表。很多人以为在对话,其实每次都在重写历史。
PRD 里明确禁止 m05 去 import m03。每个模块应该是自包含的。如果两个模块需要共享逻辑,应该把共享部分放到 _shared/ 里。