Vibe Coding 时，遇到陌生、或者已经记不清的技术，可以用这个 skill 生成符合自己学习目标的学习材料。
它从问题出发，不预设应该使用某个框架。固定且短小的流程可能只需要普通服务代码；长时间业务流程可能更适合 Temporal 或云工作流；高吞吐投递通常要靠队列和 Outbox。材料的任务是把这些选择讲清楚。
它规定的是材料的讲法：概念先定义，接着说明输入到输出的过程，再解释关键组件、能力边界和替代方案。
它不替代项目经验，也不试图把一门技术讲成百科全书。
- 入门一项新技术，先建立能用于判断的基本认识；
- 重写一份“概念不少、但读完不会用”的技术材料；
- 审阅 AI 生成的技术方案或代码，找出它真正保证了什么、没有保证什么；
- 比较几个方案，避免因为某个框架热门就强行使用。
在 Codex 中说清楚技术、读者基础和学习边界即可。例如：
使用 $technology-learning-judgment，帮我完整但不过度地入门 LangGraph。
我会 Python，也做过基本的 LLM 调用。
我想知道何时该用 LangGraph，并能审阅 AI 生成的实现。请讲清 State、图执行、持久化、工具调用、人工审批和能力边界；不要写成 API 手册，也不要把示例说成生产就绪。
也可以拿已有材料来审阅：
使用 $technology-learning-judgment 审阅这份技术教材。指出概念链路断在哪里，哪些能力被写大了，哪些说法需要核验版本，并说明同类问题在业界还有什么常见做法。
下面只把 LangGraph 当作一份完整示例，不表示它是默认方案。这个案例之所以选用 LangGraph，是因为它恰好包含状态、工具调用、审批和恢复这些需要说明边界的机制。其他技术的材料应从自己的问题和替代方案开始。
展开：LangGraph 系统学习教材（完整文档）
验证日期：2026-08-05。本文以 LangGraph Python 官方 v1.x 文档为准。API 会演进；涉及安装、迁移或生产部署时，应再次核对文末一手资料。
学习 LangGraph，不是为了记住图 API，而是为了能判断：一个带 LLM 的流程何时需要显式状态、恢复、人工介入和工具边界；这些能力由什么机制提供；哪些问题仍必须交给数据库、队列、权限系统或工作流引擎。
读完后，你应能：
- 画出一次图调用中输入、State、节点更新和输出的流向；
- 为并行更新定义正确的 reducer，而不是假定“并发天然安全”；
- 区分 State、Checkpointer、Store 和业务数据库；
- 说明 interrupt 恢复为什么会重跑节点，并据此设计幂等副作用；
- 判断应使用普通服务代码、高层 Agent、LangGraph，还是其他工作流技术；
- 审阅 AI 生成的“可恢复”“安全”“可审计”实现是否真的兑现承诺。
前置知识：Python 函数、类型标注、HTTP/API 的基本概念，以及 LLM 消息与工具调用的基本概念。本文不试图教授每个集成商、部署平台或模型供应商的全部 API。
LangGraph v1 要求 Python 3.10 或更高版本。仅学习图 API 时可先安装：
pip install -U langgraph模型、持久化后端和观测工具按实际需要再安装相应集成包；不要为了运行第一个 StateGraph 一次性引入 PostgreSQL、向量库、消息队列和多个模型 SDK。先让最小图的执行机制清晰，再增加外部依赖。
先记住下面的最小概念地图；每个概念的详细机制会在后文展开。
| 概念 | 定义 | 一句话输入 → 处理 → 输出 |
|---|---|---|
| LangGraph | 编排有状态工作流和 Agent 的运行时。 | 初始 State 与运行配置 → 按图调度节点、合并更新 → 最终 State、流事件或中断。 |
| State | 图执行中共享的工作快照，由字段及其更新规则组成。 | 初始数据与节点局部更新 → reducer 合并 → 下一步可读取的新 State。 |
| Node | 图中的一个工作单元，可执行普通代码、模型调用或受控副作用。 | 当前 State → 计算/调用 → 局部 State 更新或控制命令。 |
| Edge | 定义一个节点完成后，哪些节点可以被激活。 | 当前节点及 State → 固定或条件路由 → 下一个节点、并行节点或结束。 |
| Reducer | 指定同一 State 字段如何接受新更新。 | 当前字段值与本次更新 → 合并规则 → 新字段值。 |
| Checkpointer | 保存同一执行线程的图状态快照，支持恢复与回放。 | 每个 super-step 的执行状态 → 持久化快照 → 同一 thread_id 后续恢复。 |
| Store | 保存跨线程复用的应用定义数据。 | namespace、键和值 → 存储或检索 → 供后续线程使用的数据。 |
| ToolNode | 执行模型提出的标准工具调用，并将工具结果写回消息流。 | 含 tool call 的模型消息 → 受限工具执行 → ToolMessage 供模型继续处理。 |
| interrupt | 在节点中暂停图并等待外部输入。 | 中断载荷 → 保存当前执行并返回等待状态 → Command(resume=...) 恢复。 |
先不追求写代码。假设用户要求 Agent 发布一份含高风险结论的报告；系统必须先查证据，再让人批准，最后才允许发布。这条链路会在后文不断复用。
用户请求
→ State 初始化
→ model 节点提出 read_policy 工具调用
→ ToolNode 执行受限读取，返回 ToolMessage
→ model 节点形成带 evidence_id 的报告草案
→ route 节点判断风险等级
→ approval 节点 interrupt 暂停，等待人工决定
→ publish 节点使用幂等键执行发布
→ 最终 State / 审计事件
| 阶段 | 输入 | 关键 State 变化 | 输出/下一步 |
|---|---|---|---|
| 初始化 | 用户消息、tenant_id、报告主题 |
写入 messages、tenant_id、risk_level=None |
激活 model |
| 选择工具 | 当前消息和可用工具 schema | 新增含 tool_calls 的 AIMessage |
路由到 tools |
| 执行工具 | AIMessage 的工具名与参数 | ToolNode 追加 ToolMessage |
回到 model |
| 形成草案 | 对话、工具结果、业务规则 | 写入 draft、evidence_ids、risk_level |
路由到审批或结束 |
| 人工审批 | 高风险草案和证据摘要 | 图暂停；检查点记录状态 | 返回中断载荷 |
| 恢复与发布 | 同一 thread_id、批准/拒绝结果 |
写入 approved；批准后写 publication_id |
返回最终报告状态 |
这里已经出现了 State、Node、Edge、ToolNode、Checkpointer、interrupt 和业务幂等，但它们的职责不同：模型只能提议工具和草案；ToolNode 只执行受限工具调用；服务端必须独立检查租户和对象权限；Checkpointer 只恢复图执行；业务数据库才记录报告和发布事实。
| 组件 | 输入 | 输出 | 在链路中的职责与边界 |
|---|---|---|---|
MessagesState |
初始消息或消息更新 | 带专用消息 reducer 的 messages State 字段 |
保存对话与工具结果；不是跨线程记忆库。 |
StateGraph(StateSchema) |
State schema | 图构建器 | 声明图有哪些状态字段及更新语义；本身不执行节点。 |
add_node(name, callable) |
节点名、接收 State 的函数 | 注册后的节点 | 连接“谁做工作”；函数返回局部 State 更新。 |
ToolNode(tools) |
BaseTool 列表、最后一条 AIMessage 的 tool call |
ToolMessage 更新 | 运行标准工具调用；不负责业务授权、沙箱或事实校验。 |
tools_condition |
当前消息 State | tools 或结束的路由结果 |
根据是否存在 tool call 做图内路由；不评估工具是否合理。 |
interrupt(payload) |
JSON 可序列化审批载荷 | 暂停图；恢复时返回 Command(resume=...) 的值 |
让外部系统介入；恢复会重跑所在节点。 |
compile(checkpointer=...) |
完整图和可选运行时组件 | 可调用图 | 检查结构并配置 checkpoint；不会自动创建业务表或审计策略。 |
假设初始调用是：
initial = {
"messages": [{"role": "user", "content": "发布本周高风险报告"}],
"tenant_id": "acme",
"risk_level": None,
"approved": None,
}第一轮，model 读取 messages 和工具 schema，返回一条 AIMessage。若它包含 read_policy 的 tool call，tools_condition 只检查这个结构事实并把控制流交给 ToolNode。第二轮，ToolNode 用参数调用受限工具，返回 ToolMessage；并不把工具返回文本当作可信结论。第三轮，模型读取 ToolMessage 后形成草案；一个确定性节点或服务端规则校验草案引用的 evidence_ids 和风险等级。只有校验通过的高风险草案才会抵达 approval。
审批发生时，interrupt() 把一个载荷交给 UI/任务系统；图并没有“在内存中等着”。同一 thread_id 后续带 Command(resume={"approved": True}) 恢复，审批节点会从头运行。因而真正的发布操作必须放在批准后的节点，并使用业务侧 idempotency key，防止恢复或重复投递造成重复发布。
后文学习每个概念时，都回到这条链路问三件事：它读什么输入？写什么输出？若失败、重试或恢复，状态会怎样？
LangGraph 是低层的状态化 Agent/工作流编排运行时。它允许确定性代码和模型驱动步骤出现在同一张图中，重点能力是持久化、流式、人工介入和细粒度流程控制。
它不是“让模型更会思考”的库，也不是业务数据库、事务协调器、队列或权限系统。
| 场景 | 更合适的选择 | LangGraph 的位置 |
|---|---|---|
| 单次摘要、分类、翻译、固定 RAG | 普通函数或 LangChain runnable | 通常不需要图 |
| 简单工具型聊天 Agent | LangChain create_agent 等高层工厂 |
只有要自定义状态/恢复/路由时再下沉 |
| 需要循环、动态工具、HITL、长时间恢复 | LangGraph | 强适配 |
| 订单、支付、跨服务补偿 | 领域数据库 + Temporal/Cadence/云工作流等 | 可在某个 Agent 步骤内部使用，不能替代事务/补偿 |
| 高吞吐异步投递 | 队列 + Worker + Outbox | 可作为单个任务中的推理编排，不能替代投递保证 |
| 批量数据管道 | Airflow、Dagster、Prefect 等 | 通常不应作为主调度器 |
一个简单标准：若流程是短小、固定、确定性的，先写普通代码；只有当“状态化分支/循环 + 恢复 + 人工或模型参与”确实带来价值时，再引入图。
LangGraph 的核心不是 LLM，而是下面三个对象：
- State：当前执行快照中的共享数据；按字段（channel）管理更新语义。
- Node：读取当前 State，完成计算或副作用，返回本节点的局部更新。
- Edge：决定下一批要运行的节点；可以是固定边或条件边。
运行时以 super-step 推进：同一批被激活、彼此独立的节点可以并行执行；本批更新根据每个字段的 reducer 合并；再激活下一批节点。没有活跃节点和在途消息时，图结束。
输入
↓
START → 节点 A ─→ 节点 B ─→ END
│返回局部更新 │读取更新后的 State
└──── State/reducer 合并 ────┘
以下代码只演示执行机制，不包含模型调用。
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class ArticleState(TypedDict):
topic: str
outline: str
article: str
def make_outline(state: ArticleState):
return {"outline": f"1. {state['topic']} 的定义\n2. 使用场景"}
def write_article(state: ArticleState):
return {"article": f"基于提纲：\n{state['outline']}"}
builder = StateGraph(ArticleState)
builder.add_node("make_outline", make_outline)
builder.add_node("write_article", write_article)
builder.add_edge(START, "make_outline")
builder.add_edge("make_outline", "write_article")
builder.add_edge("write_article", END)
graph = builder.compile()
result = graph.invoke({"topic": "LangGraph", "outline": "", "article": ""})一次运行的状态演化如下：
| 时刻 | 活跃节点 | 读取 | 返回的局部更新 | 合并后 State 的关键字段 |
|---|---|---|---|---|
| 输入 | START |
topic |
无 | topic="LangGraph" |
| super-step 1 | make_outline |
topic |
{"outline": ...} |
outline 被写入 |
| super-step 2 | write_article |
topic、outline |
{"article": ...} |
article 被写入 |
| 结束 | END |
无 | 无 | 返回输出 State |
compile() 不是可选装饰：它会检查图结构，并注入运行时能力，例如 checkpointer、缓存或中断配置。新代码优先用 START/END 表达入口与终点；set_entry_point() 和 set_finish_point() 仍可用，但只是等价的较旧写法。
TypedDict 常用于声明 State schema。它主要提供类型结构，不提供运行时校验；调用一个 TypedDict 类会得到普通字典，并非创建有行为的领域对象。需要默认值可考虑 dataclass；需要递归运行时校验可考虑 Pydantic，但通常有性能代价。
节点应返回局部更新，而非把读取到的完整 State 原地修改后整体返回。原因不是框架把 Python 对象变成了不可变对象，而是局部更新能让运行时明确：哪个字段被谁写入、何时合并、如何保存快照。
没有 reducer 的字段使用“后来的更新覆盖当前值”。有 reducer 时，运行时以 (left=当前值, right=本次更新) 调用它。
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
class SearchState(TypedDict):
query: str
documents: Annotated[list[str], add]
def search_a(_: SearchState):66
return {"documents": ["文档 A"]}
def search_b(_: SearchState):
return {"documents": ["文档 B"]}若 search_a 和 search_b 在同一 super-step 写入 documents，add 会将两份列表合并。它只定义图内同字段更新的合并方式，不等于数据库事务、乐观锁或跨服务一致性。
不要机械地给所有列表使用 add。例如最终风险等级、人工裁决和支付结果属于冲突型业务结论，应定义业务规则（最高风险优先、人工结果优先、拒绝冲突等），或进入人工审查。
消息有 ID，可能被更新，也存在工具调用和工具结果的对应关系。operator.add 只能拼接，不能正确表达“用同一 ID 更新既有消息”。因此聊天图通常用 MessagesState，其 messages 字段使用专用的 add_messages reducer。
from langgraph.graph import MessagesState
class SupportState(MessagesState):
customer_id: str
escalation_reason: str | None这不代表应无限累积对话；长对话需要裁剪、摘要或检索策略，也需要明确哪些原文不应保存。
条件边根据已有 State 选择路径。最佳实践是让路由函数尽量纯粹、确定：读取状态并返回目的地。把模型调用、数据库写入或网络请求藏进路由函数会使测试、重放和失败处理变得模糊；这不是框架禁止，而是工程上应避免的职责混杂。
from typing import Literal
from langgraph.graph import END
def route_review(state) -> Literal["approval", "report", "__end__"]:
if state["critical_findings"]:
return "approval"
return "report"
builder.add_conditional_edges("analyze", route_review)常见工具循环是“模型 → 工具 → 模型”。如果模型反复请求同一工具或错误工具，就会耗尽时间和费用。State 或 runtime 配置应有明确的轮次、总超时、工具调用次数、单次结果大小和重复请求检测。
def continue_or_stop(state):
if state["tool_calls_used"] >= 8:
return "report_limit_reached"
return "call_model"框架的递归上限是兜底，不是业务预算设计。真正的业务限额应可观测、可解释，并向用户报告降级或中止原因。
对彼此独立的数据（例如多个文件的只读扫描）可 fan-out；对依赖同一中间结论的步骤不应为了并发而并发。动态 fan-out 可用 Send 把每份输入映射为独立任务，再由 reducer 汇总。
并行设计先回答四件事：任务是否独立？结果如何排序？一项失败时整体是失败、降级还是重试？汇总字段的 reducer 是什么？这些规则不由 Send 自动决定。
节点可以是同步或异步函数。除了 State，它还可接收运行时对象，以访问本次调用的上下文、Store、流写入器和执行信息。
应区分三类数据：
| 数据 | 适合的位置 | 示例 |
|---|---|---|
| 会随图演化、需要 checkpoint 的工作数据 | State | 消息、计划、当前证据、循环次数 |
| 本次调用的稳定依赖或身份 | runtime context | user_id、仓库根目录、权限主体 |
| 跨线程复用的应用数据 | Store 或业务数据库 | 用户偏好、团队规则、知识条目 |
不要把密钥、数据库连接、请求对象或不可序列化的大对象塞入 State；不要把业务实体状态误当作图 checkpoint。
一个真正的工具型 Agent 至少有以下链路：
用户消息 → 模型节点（可能产生 tool_calls）
↓ 有调用
ToolNode（校验并执行受限工具）
↓ ToolMessage
模型节点（继续推理或结束）
ToolNode 负责执行标准工具调用并把结果写回消息流；tools_condition 根据最后消息是否含 tool call 路由。下面是骨架，模型、认证和工具实现必须按项目补充。
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
@tool
def read_invoice(invoice_id: str) -> str:
"""读取当前用户有权访问的发票摘要。"""
# 服务端必须在此处执行身份、租户和对象授权检查。
return f"invoice={invoice_id}"
tools = [read_invoice]
model_with_tools = model.bind_tools(tools) # model 由应用配置
def call_model(state: MessagesState):
return {"messages": [model_with_tools.invoke(state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", tools_condition)
builder.add_edge("tools", "model")
graph = builder.compile()这段图解决的是工具循环，不提供以下保证：
@tool不是授权系统；工具内部必须验证用户、租户、对象和操作权限。- 工具 schema 不等于安全沙箱；不要向模型暴露任意 shell、宽泛文件系统或生产数据库能力。
- 模型输出合法 JSON 不等于事实正确；发布、删除、付款等高影响操作仍要验证证据与业务约束。
- 工具超时、重试、幂等键、审计记录和资源配额必须由应用明确实现。
LangGraph v1 中，langgraph.prebuilt.create_react_agent 已弃用；简单 Agent 优先使用 LangChain 的 create_agent。需要显式 State、复杂路由、细粒度恢复或治理边界时，再使用 StateGraph 自行表达流程。
这三个概念最容易混淆：
| 组件 | 保存什么 | 作用域 | 典型用途 | 不能替代 |
|---|---|---|---|---|
| State | 当前图的工作数据 | 当前执行 | 节点间通信 | 持久化或业务事实 |
| Checkpointer | State 快照与执行历史 | 单个 thread_id |
恢复、HITL、回放、容错 | 订单/审批等领域数据库 |
| Store | 应用定义的键值/检索数据 | 跨线程 | 偏好、团队规则、长期知识 | 完整业务数据模型与权限体系 |
| 业务数据库 | 领域实体和业务约束 | 租户/业务域 | 订单、用户、授权、审计 | 图运行时控制 |
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver() # 仅开发/演示；进程重启后丢失
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "review-42"}}同一个 thread_id 是恢复同一执行的指针；新 ID 是新线程。生产环境使用持久化 checkpointer，例如 PostgreSQL；首次使用通常要执行相应后端的 setup()。同时设计保留期限、备份、删除和容量治理，因为 checkpoint 会持续累积。
Store 的 namespace 是隔离边界的一部分，不是命名习惯。多租户设计至少应将租户、主体和数据类别体现在 namespace 或服务端授权上下文中；不能仅靠模型提示词避免跨租户读取。
interrupt() 在节点中暂停执行，向调用方暴露 JSON 可序列化载荷；调用方用相同 thread_id 和 Command(resume=...) 恢复。恢复值成为 interrupt() 调用的返回值。
from langgraph.types import Command, interrupt
def approval_node(state):
decision = interrupt({
"action": "publish_report",
"summary": state["summary"],
})
return {"approved": decision["approved"]}
# 初次调用会暂停；批准后：
# graph.invoke(Command(resume={"approved": True}), config)关键语义：恢复时，包含 interrupt() 的整个节点从开头重新执行，而不是从那一行继续。因此中断前的外部副作用必须幂等，或移动到批准后的独立执行节点。
错误模式：
发送邮件 → interrupt(“是否批准？”) → 恢复
恢复会再次发送邮件。更安全的模式是：先创建带 idempotency key 的待执行动作，暂停审批；审批后由单独节点执行一次，并把执行结果写入业务审计。
批准、参数修改和自然语言反馈也不同：批准沿用原参数；修改应产生结构化、可验证的新参数；反馈应交回模型重新规划。三者都应记录操作者、时间、原始请求和最终决策。
流式输出是投影方式，不是业务逻辑。常见选择：
| 目标 | 适合的流 |
|---|---|
| 展示节点进度、排查局部更新 | stream_mode="updates" |
| 查看累计 State 演化 | stream_mode="values" |
| 聊天界面的增量消息/token | 消息或事件流模式 |
| 深度排错 | debug/事件流，但不要直接给终端用户 |
for update in graph.stream(input_state, config, stream_mode="updates"):
print(update)注意：输入/输出 schema 不等于流式隐私边界。内部/private channel 在某些 values 流中仍可能出现；向外部 UI 输出前应明确过滤允许字段。
生产可观测性至少记录：trace/run ID、节点名、模型与工具耗时、token/成本、重试、失败类别、审批等待时长、最终状态。日志必须脱敏，避免记录密钥、不必要的个人信息或原始敏感内容。
错误处理先分类：网络/限流可有限重试；参数和权限错误通常不可重试；模型格式错误可有限修复或降级；证据不足应转人工或返回“不足以判断”。所有副作用都应有超时、有限重试和幂等策略。
子图适合封装职责清楚的能力，例如“只读仓库检索”“静态扫描”“证据归并”。先定义其输入、输出、错误与预算，再把它作为父图节点使用。
多 Agent 不是让多个模型互相聊天。只有角色在权限、上下文、工具、生命周期或责任边界上真正不同，拆分才有收益。否则一个 Agent 加专用工具通常更便宜、更易调试和评估。
一个协调器至少要规定：最大委派次数、子任务超时、取消条件、结果 schema、失败传播规则和最终决策权。若这些规则没有写清，所谓“多 Agent”往往只是更昂贵的循环。
现在回到“高风险报告发布”示例。最终实战并不是再学习一批陌生组件，而是把已理解的链路写成可验证代码：
initial State
└─ MessagesState：用户请求与后续 ToolMessage
└─ model Node：读取消息，输出 AIMessage / tool call
└─ tools_condition：输出“去工具”或“结束”
└─ ToolNode：读取工具参数，输出 ToolMessage
└─ model Node：输出草案
└─ 规则/路由 Node：输出审批或报告路径
└─ interrupt：输出暂停载荷
└─ Command(resume)：输入人工决定
└─ publish Node：输出发布结果与审计引用
最终实战需要额外加入的不是“更多 LangGraph 魔法”，而是现实约束：
- State schema 中每个字段的 reducer 和输出边界；
- 工具内部的租户/对象授权、超时和参数校验；
- 持久化 checkpointer 与稳定
thread_id； - 批准后副作用的幂等键与业务数据库审计；
- 对工具失败、模型格式错误、拒绝审批和恢复重跑的测试。
如果你能把这段压缩链路还原成上一节的逐步状态表，再开始读或编写最终实战代码；否则代码只会变成一串不可解释的 API 调用。
先实现无模型的确定性基线，再将“形成候选发现”替换为受控的 LLM + ToolNode 循环。这样能区分“图的恢复/审批机制”与“模型推理”的责任，避免把两者混成黑盒。
下面的骨架展示 State、Node、条件边、Checkpointer 与 interrupt 的最小协作；它不连接真实模型、数据库或发布 API，因此只用于理解和改造，不宣称为生产就绪项目。
from typing import Literal
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
class ReviewState(TypedDict):
request: str # 输入：用户要发布的内容
evidence_ids: list[str] # 确定性证据的标识，不放原始敏感内容
risk_level: Literal["low", "high"]
approved: bool | None
publication_id: str | None
def collect_evidence(state: ReviewState):
# 输入：request；输出：可审计的证据 ID 更新
return {"evidence_ids": ["policy-42"]}
def assess_risk(state: ReviewState):
# 基线版本以确定性规则给出结论；Agent 版本可由模型提出候选，
# 但仍要在服务端验证证据和风险规则。
return {"risk_level": "high"}
def route_after_assessment(state: ReviewState) -> Literal["approval", "publish"]:
return "approval" if state["risk_level"] == "high" else "publish"
def approval(state: ReviewState):
# 输入：风险和证据；输出：暂停，或恢复后的 approved 更新。
decision = interrupt({
"action": "publish_report",
"evidence_ids": state["evidence_ids"],
"risk_level": state["risk_level"],
})
return {"approved": bool(decision["approved"])}
def publish(state: ReviewState):
if state["approved"] is False:
return {"publication_id": None}
# 真实实现：这里调用业务服务，并传入稳定 idempotency key；
# 不要在 approval 的 interrupt 之前执行发布。
return {"publication_id": "report-001"}
builder = StateGraph(ReviewState)
builder.add_node("collect_evidence", collect_evidence)
builder.add_node("assess_risk", assess_risk)
builder.add_node("approval", approval)
builder.add_node("publish", publish)
builder.add_edge(START, "collect_evidence")
builder.add_edge("collect_evidence", "assess_risk")
builder.add_conditional_edges("assess_risk", route_after_assessment)
builder.add_edge("approval", "publish")
builder.add_edge("publish", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "report-42"}}
initial = {
"request": "发布本周报告",
"evidence_ids": [],
"risk_level": "low",
"approved": None,
"publication_id": None,
}
paused = graph.invoke(initial, config)
# 读取 paused["__interrupt__"] 后，由外部 UI/服务让人工决策：
finished = graph.invoke(Command(resume={"approved": True}), config)| 符号 | 输入 | 输出 | 容易误解的点 |
|---|---|---|---|
collect_evidence |
完整 ReviewState，实际仅读取 request |
evidence_ids 局部更新 |
它应调用受限证据服务，而不是相信模型自造的证据。 |
assess_risk |
证据和请求 | risk_level 局部更新 |
基线可确定性；模型版只提供候选，不能绕过服务端规则。 |
route_after_assessment |
已更新的 State | 节点名 | 它只选择路径，不应偷偷发布、读库或调用模型。 |
approval |
风险和证据 | 中断，或恢复后 approved 更新 |
恢复会从函数开头重跑，因此此前不能有非幂等副作用。 |
publish |
已批准的 State | publication_id 更新 |
图状态更新不等于发布已可靠落库；真实发布需业务事务/幂等/审计。 |
将它升级为前述完整 Agent 时，只替换 assess_risk 周边：增加 MessagesState、model 节点、ToolNode 和 tools_condition 循环；保留证据校验、审批、发布和幂等边界。不要因为引入了模型就删除确定性控制点。
下面是应如何组织的生产链路：
PR diff/扫描结果
→ 确定性证据收集（白名单读取、静态扫描）
→ 模型形成“候选发现”并引用证据 ID
→ 服务端校验文件、行号、证据与严重等级
→ critical 发现进入 interrupt 审批
→ 批准后，以幂等发布动作写入评论与审计
这里的职责不能互换：扫描器和文件读取提供事实；模型归纳和提出候选；服务端授权与校验决定能否发布；checkpointer 恢复图执行；业务数据库保存审批和发布事实。把其中任意一层交给模型“自行保证”，都不是可靠设计。
让 AI 实现 LangGraph 功能时，不要只要求“加持久化”或“加人工审批”。改用可验收的问题：
- 图的 State 有哪些字段、各字段 reducer 是什么？并行写入时如何合并？
thread_id如何生成、归属哪个租户、谁有权恢复？- 哪个副作用会在恢复时重跑？它的幂等键在哪里验证？
- ToolNode 暴露了哪些窄工具？每个工具如何做服务端授权、审计和资源限制？
- 哪些数据进 checkpoint，哪些进 Store，哪些必须留在业务数据库？
- 模型输出的结构、事实和权限分别由什么验证？
- 达到循环/成本/超时上限时，用户得到什么状态和解释？
- 测试是否覆盖恢复、重复投递、权限拒绝、并行冲突和部分失败，而非只覆盖正常路径？
若回答不清楚，不应接受“已支持可恢复/安全/生产可用”的交付结论。
- 节点为什么返回局部更新？如果两个节点并行写同一字段，谁决定合并语义？
operator.add为什么不适合通用业务结论，也不适合需要更新既有消息的消息列表？- Checkpointer 与业务数据库各自保存什么？为什么前者不能保证支付只扣一次？
- interrupt 恢复后为什么可能重复副作用？你会如何拆节点或设计幂等？
- ToolNode 能做什么、不能做什么？为什么任意 shell 工具不是合理默认值？
- 什么条件下普通服务代码、高层
create_agent或 Temporal 比手写 StateGraph 更合适？ - 如何证明一份 AI 生成的实现真的满足“多租户隔离”和“可恢复”？
能把这些问题回答为具体机制、数据位置和失败路径，而不是口号，才算掌握了 LangGraph 的工程价值。
- 不把每个概念拆成一个练习文件；
- 不把框架的能力写成业务保证；
- 不因为是 AI/Agent 项目就默认推荐 LangGraph；
- 不用“复制即可运行”“生产可用”替代对版本、依赖和失败路径的核验。
如果你拿它学习或审阅过某项技术，欢迎开 Issue，附上技术主题、原始材料或你觉得不够清楚的地方。真实案例比空泛建议更有用。
如果这个技能确实帮你发现过一次 AI 方案里的漏洞，或让你更快做出技术判断，点个 Star 就够了。