AI 大模型

前端转型 Agent 开发 05 之 Agent Hooks 与 Checkpointer（让 Agent 从全自动转变人为可掌控）

掘金 2026-09-22 约 12 分钟读完 0 阅读
文章来源声明： 原文作者：Setsuna_F_Seiei； 来源站点：掘金； 原文链接：https://juejin.cn/post/7687876412298805248； 本文基于上述来源整理/加工，觅优补充点评，仅供技术学习交流。版权归原作者所有。
觅优短评

文章以检查站和存档点为喻，讲清 Hook 与 Checkpointer 如何让 Agent 可控可回溯。适合用 LangGraph 做审批、长任务、人机协作的前端开发者参考。

本文目录
![](https://p6-xtjj-sign.byteimg.com/tos-cn-i-73owjymdk6/721d2c63e9e840549ead973af821d14b~tplv-73owjymdk6-jj-mark-v1:0:0:0:0:5o6Y6YeR5oqA5pyv56S-5Yy6IEAgU2V0c3VuYV9GX1NlaWVp:q75.awebp?rk3s=f64ab15b&x-expires=1790610230&x-signature=DhNy697r%2BshzccsrdVSUanrBpZM%3D)

一、基础概念

1.1、Agent Hook

这是什么？

Agent Hook 是在 Agent 执行循环的关键节点上插入的回调函数——让外部代码能在"模型即将推理前""工具即将执行前""这一轮结束时"等时机介入，做拦截、修改、记录或中止。

把 Agent Loop 想象成一条流水线，Hook 就是流水线上的一道道"检查站"——每个检查站都可以放行、修改、拦截或中止。

如果没有 Hook，Agent 就是一个"黑盒"：你给它任务，它自己跑到结束，中间发生了什么、调了什么工具、花了多少钱，你既看不见也管不了。Hook 就是打开这个黑盒的开关。

主要解决的问题

问题没有 Hook有 Hook
**可观测性**不知道 Agent 中间做了什么`postModelHook` / `postToolCallHook` 记录每一步
**成本控制**模型可能无限循环烧钱`preModelHook` 里检查 token 用量，超预算就停
**安全审批**危险工具（删文件/发请求）直接执行`preToolCallHook` 拦截危险工具，等用户批准
**人为介入**模型卡住只能干等在任意节点 `interrupt` 暂停，等用户给信息再恢复
**上下文管理**历史无限增长导致 token 爆炸`preModelHook` 里检测并触发摘要压缩

1.2、Checkpointer

这是什么？

Checkpointer 是 Agent 执行状态的存档/读档机制——像游戏里的"存档点"。每执行完一个节点（node），Checkpointer 就把当前完整的 State（消息历史、变量、工具结果）持久化到存储（内存/SQLite/Postgres）。任何时候你都能"读档"回到任意一个存档点。

Hook 是"实时介入"，Checkpointer 是"事后回溯"。两者经常配合：Checkpointer 存档 → 用 Hook 在某个存档点暂停 → 用户介入 → 从该存档点恢复执行。

主要解决的问题

问题没有 Checkpointer有 Checkpointer
**进程崩溃**Agent 跑到一半挂了，全部重来从最近的 checkpoint 恢复，继续执行
**长任务断点续跑**跑 2 小时的任务必须一次性跑完随时暂停、随时恢复
**时间旅行调试**不知道哪一步出错了回退到出错前的 checkpoint，重放观察
**分支实验**想试"如果这步换种做法会怎样"从某 checkpoint fork 出新分支跑
**人机协作**暂停等用户审批后恢复interrupt 暂停时自动存档，恢复时自动读档

二、Hooks - 掌握执行操作/方向的可能性

2.1、在 Agent Loop 当中增加运行钩子 Hook

LangGraph 的 createAgent 支持 preModelHook / postModelHook 两类钩子，分别在"模型推理前"和"模型推理后"触发。返回值可以更新 State 或注入消息：

<span>import</span> { createAgent } <span>from</span> <span>'@langchain/langgraph'</span>;
<span>import</span> { <span>ChatAnthropic</span> } <span>from</span> <span>'@langchain/anthropic'</span>;
<span>import</span> { <span>Annotation</span> } <span>from</span> <span>'@langchain/langgraph'</span>;

<span>// ① 定义 State（带一个 turnCount 字段，用于在 Hook 里计数）</span>
<span>const</span> <span>StateAnnotation</span> = <span>Annotation</span>.<span>Root</span>({
  ...<span>Annotation</span>.<span>MessagesState</span>.<span>spec</span>, <span>// 内置 messages 数组</span>
  <span>turnCount</span>: <span>Annotation</span><<span>number</span>>({ <span>default</span>: <span>() =></span> <span>0</span>, <span>reducer</span>: <span>(<span>a, b</span>) =></span> a + b }),
});

<span>const</span> agent = <span>createAgent</span>({
  <span>llm</span>: <span>new</span> <span>ChatAnthropic</span>({ <span>model</span>: <span>'claude-sonnet-4-20250514'</span> }),
  <span>tools</span>: [<span>/* ... */</span>],
  <span>stateSchema</span>: <span>StateAnnotation</span>,

  <span>// ② preModelHook：模型推理前触发</span>
  <span>//    返回的对象会被 merge 进 State（这里累加轮次 + 注入提醒）</span>
  <span>preModelHook</span>: <span>(<span>state</span>) =></span> {
    <span>console</span>.<span>log</span>(<span>`[Hook] 即将进行第 <span>${state.turnCount + <span>1</span>}</span> 轮推理`</span>);

    <span>// 成本控制：超过 50 轮就强制注入"请尽快收尾"的提醒</span>
    <span>if</span> (state.<span>turnCount</span> >= <span>50</span>) {
      <span>return</span> {
        <span>turnCount</span>: <span>1</span>,
        <span>messages</span>: [{
          <span>role</span>: <span>'system'</span>,
          <span>content</span>: <span>'⚠️ 已达 50 轮，请在下一轮给出最终答案。'</span>,
        }],
      };
    }
    <span>return</span> { <span>turnCount</span>: <span>1</span> };
  },

  <span>// ③ postModelHook：模型推理后、工具执行前触发</span>
  <span>//    可以看到模型本轮的决定（有没有 tool_calls）</span>
  <span>postModelHook</span>: <span>(<span>state</span>) =></span> {
    <span>const</span> lastMsg = state.<span>messages</span>.<span>at</span>(-<span>1</span>);
    <span>if</span> (lastMsg?.<span>tool_calls</span>?.<span>length</span>) {
      <span>console</span>.<span>log</span>(<span>`[Hook] 模型决定调用: <span>${lastMsg.tool_calls.map(t => t.name).join(<span>', '</span>)}</span>`</span>);
    } <span>else</span> {
      <span>console</span>.<span>log</span>(<span>'[Hook] 模型本轮无工具调用，即将输出最终答案'</span>);
    }
    <span>// 返回 undefined 表示不修改 State</span>
  },
});

Hook 的执行时机图：

用户输入
  ↓
[<span>preModelHook</span>] ← 你可以在这里修改 messages、注入提醒、检查预算
  ↓
模型推理（stream）
  ↓
[<span>postModelHook</span>] ← 你可以在这里记录模型决定、拦截危险 tool_call
  ↓
工具执行
  ↓
回到 preModelHook（下一轮循环）

2.2、审批处理 Hook

判断需要审批的情况

并非所有工具调用都需要审批——只读工具（查询、搜索）可以直接放行，而有副作用的工具（删文件、发请求、改配置、花钱）应该拦截等用户批准。实现方式是在 postModelHook 里检查 tool_calls，遇到危险工具就触发 interrupt：

<span>import</span> { createAgent, interrupt, <span>Command</span> } <span>from</span> <span>'@langchain/langgraph'</span>;

<span>// 定义哪些工具需要审批</span>
<span>const</span> <span>DANGEROUS_TOOLS</span> = <span>new</span> <span>Set</span>([<span>'delete_file'</span>, <span>'execute_command'</span>, <span>'send_email'</span>]);

<span>const</span> agent = <span>createAgent</span>({
  <span>llm</span>: <span>new</span> <span>ChatAnthropic</span>({ <span>model</span>: <span>'claude-sonnet-4-20250514'</span> }),
  <span>tools</span>: [<span>/* ... */</span>],

  <span>postModelHook</span>: <span>(<span>state</span>) =></span> {
    <span>const</span> lastMsg = state.<span>messages</span>.<span>at</span>(-<span>1</span>);
    <span>const</span> calls = lastMsg?.<span>tool_calls</span> ?? [];

    <span>// 找出本轮调用中需要审批的危险工具</span>
    <span>const</span> dangerous = calls.<span>filter</span>(<span><span>c</span> =></span> <span>DANGEROUS_TOOLS</span>.<span>has</span>(c.<span>name</span>));

    <span>if</span> (dangerous.<span>length</span> > <span>0</span>) {
      <span>// interrupt：暂停执行，把审批请求抛给外部</span>
      <span>// 外部（UI）会展示"模型想调用 delete_file，是否批准？"</span>
      <span>const</span> decision = <span>interrupt</span>({
        <span>type</span>: <span>'approval_request'</span>,
        <span>toolCalls</span>: dangerous.<span>map</span>(<span><span>c</span> =></span> ({
          <span>name</span>: c.<span>name</span>,
          <span>args</span>: c.<span>args</span>,
        })),
        <span>message</span>: <span>`模型请求执行 <span>${dangerous.length}</span> 个危险操作，请审批`</span>,
      });

      <span>// decision 是用户在外部恢复时传入的决定</span>
      <span>if</span> (decision === <span>'allow_once'</span>) {
        <span>console</span>.<span>log</span>(<span>'[审批] 用户：本次允许'</span>);
        <span>// 不做任何修改，正常继续执行</span>
      } <span>else</span> <span>if</span> (decision === <span>'allow_session'</span>) {
        <span>console</span>.<span>log</span>(<span>'[审批] 用户：本会话允许'</span>);
        <span>// 可以在这里把工具加入白名单（简化示例省略状态持久化）</span>
      } <span>else</span> {
        <span>console</span>.<span>log</span>(<span>'[审批] 用户：拒绝'</span>);
        <span>// 拒绝：把 tool_calls 从消息里移除，注入"用户拒绝"的 ToolMessage</span>
        <span>return</span> {
          <span>messages</span>: [{
            <span>role</span>: <span>'tool'</span>,
            <span>content</span>: <span>'用户拒绝了此操作。'</span>,
            <span>tool_call_id</span>: dangerous[<span>0</span>].<span>id</span>,
          }],
        };
      }
    }
  },
});

审批操作行为的记录

单次允许

用户选"单次允许"——只放行这一回，下次再调同一个工具还是要审批。适合偶尔用的高危操作（如删除生产数据）。

对话允许

用户选"本会话允许"——把工具加入当前会话的白名单，后续不再弹审批。适合用户信任后想连续操作的场景（如批量重命名文件）。

拒绝

用户选"拒绝"——把这个 tool_call 拦截掉，注入一条 ToolMessage 告诉模型"用户拒绝了这个操作"，模型会据此调整后续行为（比如换一种方案或询问用户原因）。注意：拒绝不是中止 Agent，Agent 会继续运行，只是这一步被否决了。

2.3、询问确认 Hook

不确定的处理方向

有时模型不是要执行危险操作，而是自己拿不准该走哪个方向——比如用户的需求模糊，或有多个可行方案。这时可以用 interrupt 主动暂停，向用户提问：

<span>const</span> <span>agent</span> = <span>createAgent</span>({
  <span>llm</span>: model,
  <span>tools</span>: [<span>/* ... */</span>],

  // 用一个专门的<span>"问用户"</span>工具，模型遇到歧义时调用它
  // 这里通过 postModelHook 检测到 ask_user 工具被调用，触发 interrupt
  <span>postModelHook</span>: (state) => {
    <span>const</span> <span>lastMsg</span> = state.messages.<span>at</span>(-<span>1</span>);
    <span>const</span> <span>askCall</span> = lastMsg?.tool_calls?.<span>find</span>(c => c.name === <span>'ask_user'</span>);

    <span>if</span> (askCall) {
      <span>// interrupt 暂停，把问题抛给用户</span>
      <span>const</span> <span>answer</span> = <span>interrupt</span>({
        <span>type</span>: <span>'question'</span>,
        <span>question</span>: askCall.args.question,
        <span>options</span>: askCall.args.options, // 可选：预设选项
      });

      <span>// 用户回答后恢复，把答案作为 tool 结果注入</span>
      <span>return</span> {
        messages: [{
          role: <span>'tool'</span>,
          tool_call_id: askCall.id,
          content: answer,
        }],
      };
    }
  },
});

单选/多选/自定义处理

ask_user 工具的 schema 可以声明不同的回答模式，UI 据此渲染不同交互：

import { tool } <span>from</span> <span>'@langchain/core/tools'</span>;
import { z } <span>from</span> <span>'zod'</span>;

<span>const</span> <span>askUser</span> = <span>tool</span>(
  // 这个工具的真正<span>"执行"</span>由 postModelHook 的 interrupt 接管，这里只是占位
  <span>async</span> () => <span>'等待用户回答'</span>,
  {
    <span>name</span>: <span>'ask_user'</span>,
    <span>description</span>: <span>'当你不确定用户意图时，向用户提问。不要自己瞎猜。'</span>,
    <span>schema</span>: z.<span>object</span>({
      <span>question</span>: z.<span>string</span>().<span>describe</span>(<span>'要问用户的问题'</span>),
      mode: z.<span>enum</span>([<span>'single'</span>, <span>'multiple'</span>, <span>'text'</span>]).<span>describe</span>(
        <span>'single=单选, multiple=多选, text=自由文本'</span>,
      ),
      options: z.<span>array</span>(z.<span>string</span>()).<span>optional</span>().<span>describe</span>(
        <span>'当 mode 为 single/multiple 时的候选项'</span>,
      ),
    }),
  },
);
modeUI 渲染例子
`single`单选按钮组"用 React 还是 Vue？" → \[React\] \[Vue\]
`multiple`多选复选框"要导出哪些格式？" ☑PDF ☑HTML ☐DOCX
`text`文本输入框"你的目标用户是谁？" → \[\_\_\_\_\]

实际应用中，桌面端（Electron）和终端端（TUI）通常各自把 interrupt 事件渲染成对应的 UI 组件——前者弹出审批对话框，后者用终端全屏 overlay 展示选项。

三、Checkpointer - 掌控对话存档与回溯之力

承继着前面的 Agent Hook，我们能够基于这个点进一步的实现一个类似存档的能力。

3.1、快照存档点

LangGraph 的 Checkpointer 在每个 node 执行完毕后自动存档——你不需要手动调用 save。只要在 compile 时传入一个 checkpointer 实例，存档就自动开启。

<span>import</span> { createAgent } <span>from</span> <span>'@langchain/langgraph'</span>;
<span>import</span> { <span>MemorySaver</span> } <span>from</span> <span>'@langchain/langgraph'</span>;
<span>import</span> { <span>SqliteSaver</span> } <span>from</span> <span>'@langchain/langgraph-checkpoint-sqlite'</span>;
<span>import</span> * <span>as</span> sqlite <span>from</span> <span>'node:sqlite'</span>;

<span>// ① 选择存档后端</span>
<span>//    MemorySaver：存内存，进程退出就丢（适合开发调试）</span>
<span>const</span> memoryCheckpointer = <span>new</span> <span>MemorySaver</span>();

<span>//    SqliteSaver：存 SQLite，进程重启后可恢复（适合生产）</span>
<span>const</span> db = <span>new</span> sqlite.<span>DatabaseSync</span>(<span>'checkpoints.sqlite'</span>);
<span>const</span> sqliteCheckpointer = <span>SqliteSaver</span>.<span>fromConn</span>(db);

<span>// ② 把 checkpointer 传给 Agent</span>
<span>const</span> agent = <span>createAgent</span>({
  <span>llm</span>: <span>new</span> <span>ChatAnthropic</span>({ <span>model</span>: <span>'claude-sonnet-4-20250514'</span> }),
  <span>tools</span>: [<span>/* ... */</span>],
  <span>checkpointer</span>: sqliteCheckpointer, <span>// ← 关键：传入后自动存档</span>
});

<span>// ③ 每次调用要带一个 thread_id（会话标识）</span>
<span>//    同一个 thread_id 的多次调用共享同一份存档历史</span>
<span>const</span> config = { <span>configurable</span>: { <span>thread_id</span>: <span>'session-001'</span> } };

<span>// 第一次调用：执行到一半时被 interrupt 暂停（比如等审批）</span>
<span>const</span> result1 = <span>await</span> agent.<span>invoke</span>(
  { <span>messages</span>: [{ <span>role</span>: <span>'user'</span>, <span>content</span>: <span>'帮我删除 /tmp/old.log'</span> }] },
  config,
);
<span>// 此时 Agent 暂停在审批 interrupt 处，状态已存档</span>

<span>// 用户批准后，第二次调用：从存档点恢复，继续执行</span>
<span>const</span> result2 = <span>await</span> agent.<span>invoke</span>(
  <span>new</span> <span>Command</span>({ <span>resume</span>: <span>'allow_once'</span> }), <span>// 传入审批决定</span>
  config, <span>// 同一个 thread_id → 读档</span>
);
<span>// Agent 从暂停处继续，执行删除操作，输出最终结果</span>

存档的内容：每次存档保存完整的 State 快照——包括 messages 数组、所有自定义变量、当前执行到哪个 node、interrupt 状态。这意味着你可以在任意时刻"读档"回到任何一个历史节点。

3.2、读档回溯重来

存档是为了能在需要时回退到历史状态。LangGraph 提供两套 API：getStateHistory 列出某 thread 的所有 checkpoint，updateState 把 State 回退/修改到指定版本。

import { Agent } from <span>'./my-agent'</span>; <span>//</span> 上面创建的 agent 实例

const config = { configurable: { thread_id: <span>'session-001'</span> } };

<span>//</span> ① 列出该 thread 的所有 checkpoint（按时间倒序）
const history = await agent.getStateHistory(config);

<span>for</span> (const <span>state</span> of history) {
  console.log({
    checkpointId: state.config.configurable.checkpoint_id,
    step: state.values.step ?? <span>0</span>,           <span>//</span> 第几步
    messages: state.values.messages.length, <span>//</span> 这一步的消息条数
    <span>next</span>: state.next,                       <span>//</span> 下一步要执行的节点
    createdAt: state.metadata?.createdAt,
  });
}

// 输出示例：
// [
<span>//</span>   { checkpointId: <span>'1f2c...'</span>, step: <span>5</span>, messages: <span>12</span>, <span>next</span>: [<span>'tools'</span>], createdAt: <span>'...'</span> },
<span>//</span>   { checkpointId: <span>'a8d1...'</span>, step: <span>4</span>, messages: <span>10</span>, <span>next</span>: [<span>'agent'</span>], createdAt: <span>'...'</span> },
<span>//</span>   { checkpointId: <span>'5e3b...'</span>, step: <span>3</span>, messages: <span>8</span>,  <span>next</span>: [<span>'tools'</span>], createdAt: <span>'...'</span> },
<span>//</span>   ...
// ]

回退到指定 checkpoint：用 updateState 选一个历史版本作为新起点，然后把修改后的 State 作为下一次 invoke 的输入。LangGraph 会从那个 checkpoint 重放后续节点，而不是简单"跳过去"——这保证 State 一致性。

<span>// ② 找到出错前的那一步（比如第 4 步）</span>
<span>const</span> targetCheckpoint = history.<span>find</span>(s => s.values.step === <span>4</span>);

<span>// ③ 从这个 checkpoint 恢复，并修改 State（可选）</span>
await agent.<span>updateState</span>(
  targetCheckpoint.config, <span>// 要恢复到的 checkpoint 配置</span>
  {
    <span>// 覆盖 State 字段（这里是纠正用户的错误输入）</span>
    messages: [
      ...targetCheckpoint.values.messages.<span>slice</span>(<span>0</span>, <span>-1</span>), <span>// 去掉最后那条坏消息</span>
      { role: <span>'user'</span>, content: <span>'正确的问题描述'</span> },
    ],
  },
);

<span>// ④ 再次 invoke——LangGraph 会从更新后的 State 继续执行</span>
<span>const</span> result = await agent.<span>invoke</span>(
  null, <span>// 不传新输入，从 checkpoint 恢复</span>
  targetCheckpoint.config,
);

关键点：updateState 是修改 + 重放，不是"删除历史"。原 checkpoint 还在 store 里，你随时可以回到任何历史版本。生产环境的 Checkpoint store 通常保留所有历史（按 thread_id 索引），UI 上可以可视化时间线让用户选择回退点。

3.3、分叉实验：从一个 checkpoint 跑出多个分支

存档的另一个高级用法是分支实验（fork） ——从同一个 checkpoint 出发，复制多份分别跑不同方案，最后对比结果。这在 prompt 调优、A/B 测试、方案对比时非常有用。

<span>// 场景：模型在第 4 步生成了"建议 A"和"建议 B"两个分支，</span>
<span>//       想分别跑完看哪个效果好</span>

<span>// ① 找到第 4 步的 checkpoint</span>
<span>const</span> checkpoint4 = (<span>await</span> agent.<span>getStateHistory</span>(config))
  .<span>find</span>(<span><span>s</span> =></span> s.<span>values</span>.<span>step</span> === <span>4</span>);

<span>// ② 分叉 1：跑"建议 A"路线</span>
<span>await</span> agent.<span>updateState</span>(checkpoint4.<span>config</span>, {
  <span>// 注入"建议 A"作为模型下一轮的输入</span>
  <span>messages</span>: [...checkpoint4.<span>values</span>.<span>messages</span>, {
    <span>role</span>: <span>'user'</span>,
    <span>content</span>: <span>'走方案 A：用 TypeScript 重写'</span>,
  }],
});
<span>const</span> branchA = <span>await</span> agent.<span>invoke</span>(<span>null</span>, checkpoint4.<span>config</span>);

<span>// ③ 分叉 2：跑"建议 B"路线（注意用不同的 thread_id，否则会覆盖）</span>
<span>const</span> branchBConfig = {
  <span>configurable</span>: {
    <span>thread_id</span>: <span>'session-001-branch-B'</span>, <span>// 不同的 thread_id</span>
    <span>checkpoint_id</span>: checkpoint4.<span>config</span>.<span>configurable</span>.<span>checkpoint_id</span>,
  },
};
<span>await</span> agent.<span>updateState</span>(branchBConfig, {
  <span>messages</span>: [...checkpoint4.<span>values</span>.<span>messages</span>, {
    <span>role</span>: <span>'user'</span>,
    <span>content</span>: <span>'走方案 B：用 Python 重写'</span>,
  }],
});
<span>const</span> branchB = <span>await</span> agent.<span>invoke</span>(<span>null</span>, branchBConfig);

<span>// ④ 对比两个分支的最终结果</span>
<span>console</span>.<span>log</span>(<span>'方案 A:'</span>, branchA.<span>messages</span>.<span>at</span>(-<span>1</span>).<span>content</span>);
<span>console</span>.<span>log</span>(<span>'方案 B:'</span>, branchB.<span>messages</span>.<span>at</span>(-<span>1</span>).<span>content</span>);

为什么分叉要用不同 thread_id：thread_id 是 checkpoint 的"命名空间"，同名会覆盖。生产场景下 fork 通常配合新的 thread_id（如 xxx-branch-1、xxx-branch-2），让两条线可以独立读档、独立存档，方便后续对比和回溯。

3.4、读档回溯的副作用与幂等性

在读档 Checkpoint 恢复时，节点可能被重新执行（replay）。如果节点有副作用（发请求、写文件、扣款），重放会重复执行！

解决副作用的一个形式：

幂等性设计——给每个工具调用附带一个唯一的 id（比如用 tool_call_id），执行时先查这个 id 是否已经执行过（用 Checkpoint 里的记录或外部状态库）。执行过就直接返回上次的结果，不真正再执行一遍。

<span>import</span> { tool } <span>from</span> <span>'@langchain/core/tools'</span>;
<span>import</span> { z } <span>from</span> <span>'zod'</span>;

<span>// 已执行操作的记录（生产环境用 SQLite/Redis 持久化）</span>
<span>const</span> executedOps = <span>new</span> <span>Map</span><<span>string</span>, <span>any</span>>();

<span>const</span> sendEmail = <span>tool</span>(
  <span>async</span> ({ to, subject, body }, config) => {
    <span>const</span> callId = config.<span>toolCallId</span>; <span>// 每次调用的唯一 id</span>

    <span>// 幂等检查：这个 callId 执行过吗？</span>
    <span>if</span> (executedOps.<span>has</span>(callId)) {
      <span>console</span>.<span>log</span>(<span>`[幂等] <span>${callId}</span> 已执行过，跳过`</span>);
      <span>return</span> executedOps.<span>get</span>(callId); <span>// 直接返回上次结果</span>
    }

    <span>// 真正执行</span>
    <span>const</span> result = <span>await</span> mailgun.<span>send</span>({ to, subject, body });

    <span>// 记录，防止重放时重复执行</span>
    executedOps.<span>set</span>(callId, result);
    <span>return</span> result;
  },
  {
    <span>name</span>: <span>'send_email'</span>,
    <span>description</span>: <span>'发送邮件（幂等：同一 callId 不会重复发送）'</span>,
    <span>schema</span>: z.<span>object</span>({ <span>to</span>: z.<span>string</span>(), <span>subject</span>: z.<span>string</span>(), <span>body</span>: z.<span>string</span>() }),
  },
);

Checkpoint 的另一面：它存的只是"Agent 的内部状态"，不包括"外部世界已经发生的改变"。所以 Checkpoint 回溯 + 副作用工具 = 潜在的重复执行风险。生产实践中对所有有副作用的工具（文件写入、命令执行、扣款、发邮件）都做幂等性处理，是规避此类风险的标准手段。

四、人机协作完整流程

Hook + Checkpointer + interrupt 三者配合，构成 LangGraph 的人机协作（Human-in-the-Loop） 核心机制。完整流程的时序图：

┌──────────┐       ┌──────────┐       ┌──────────┐       ┌──────────┐
│   User   │       │  Agent   │       │  Store   │       │   UI     │
│ (用户)   │       │ (LangGr.)│       │ (Checkpt)│       │ (前端)   │
└────┬─────┘       └────┬─────┘       └────┬─────┘       └────┬─────┘
     │ invoke           │                   │                   │
     │─────────────────▶│                   │                   │
     │                  │ preModelHook      │                   │
     │                  │──┐                │                   │
     │                  │  │记录上下文       │                   │
     │                  │◀─┘                │                   │
     │                  │ 模型推理           │                   │
     │                  │ 决策: 调工具A      │                   │
     │                  │ postModelHook      │                   │
     │                  │──┐                │                   │
     │                  │  │检测到危险工具   │                   │
     │                  │◀─┘                │                   │
     │                  │ <span>interrupt</span>()       │                   │
     │                  │──┐                │                   │
     │                  │  │存档当前 State   │                   │
     │                  │  │────────────────▶│ save checkpoint  │
     │                  │◀─┘                │                   │
     │                  │ 抛出 interrupt    │                   │
     │                  │─────────────────────────────────────▶  │
     │                  │                   │  推送审批请求      │
     │                  │                   │                   │
     │                  │                   │    用户决策        │
     │                  │                   │  ◀───────────────  │
     │                  │                   │                   │
     │                  │                   │  用户决定+参数     │
     │                  │  ◀───────────────────────────────────── │
     │                  │ <span>invoke</span>(Command)   │                   │
     │                  │──┐                │                   │
     │                  │  │读 checkpoint    │                   │
     │                  │  │◀───────────────│ load checkpoint  │
     │                  │◀─┘                │                   │
     │                  │ 从中断处继续       │                   │
     │                  │ 工具执行           │                   │
     │                  │ 模型继续推理       │                   │
     │                  │ 最终回答           │                   │
     │ ◀────────────────│                   │                   │
     │  显示流式回答     │                   │                   │

关键时序点：

  1. interrupt 触发：模型决定调危险工具 → postModelHook 调用 interrupt() → LangGraph 立即存档当前完整 State 到 Store → 抛出 interrupt 异常给上层
  2. UI 接收：上层（Electron / TUI）捕获 interrupt 事件 → 解析出审批请求内容（哪个工具、参数是什么） → 渲染审批对话框
  3. 用户决策：用户点"允许/拒绝/修改参数" → UI 把决策包装成 Command({ resume: ... }) → 再次调用 agent.invoke()
  4. resume 恢复：LangGraph 用相同 thread_id 找到之前的 checkpoint → 加载 State → 从中断点继续执行 → 工具按用户决策执行 → 模型继续推理 → 最终回答
  5. 存档保留：所有 checkpoint 都保留在 Store 里，UI 可以让用户"回到任何一步"重新决策

为什么这个流程可靠：

  • 状态不丢：中断时已存档，恢复时从存档加载，不会因为进程崩溃或断电丢失中间状态
  • 决策可追溯：每次 interrupt 的请求内容、用户决策、时间戳都记录在 checkpoint metadata 里
  • 可重放：用户可以"撤销"自己的决策，回到上一步重新选——LangGraph 读 checkpoint 重放后续节点
  • 跨进程：存档在 SQLite 等持久层后，A 进程的 interrupt 可以由 B 进程响应（多窗口/远程协作场景）

参考资料：

Agent 开发系列文章

前端转型 Agent 开发 01 之 Agent API 调用（和 Agent 的基础对话）：juejin.cn/post/767744…

前端转型 Agent 开发 02 之 Provider 与 Structured Output（规范化模型输入输出）：juejin.cn/post/767745…

前端转型 Agent 开发 03 之 Agent Tools（给 Agent 装上手脚）：juejin.cn/post/768007…

前端转型 Agent 开发 04 之 MCP 与 Skill（赋予 Agent 更广工作能力）：juejin.cn/post/768563…

前端转型 Agent 开发 05 之 Agent Hooks 与 Checkpointer（让 Agent 从全自动转变人为可掌控）：juejin.cn/spost/76878…

相关推荐

不只看跑分，Step 5 Preview 两个真实编程任务实测 AI 大模型
AI 现状：哪怕只回答一个“是”或“否”，大模型背后的“算力”也一点没省 AI 大模型
不会写代码，我画了张 A4 草图扔给豆包，二十分钟后拿到能点的网页 AI 大模型
别再跟AI说“请”了，它根本不 care——但有个东西它超在意 AI 大模型
你的 harness 技巧有保质期：176 组对照实验显示，上下文管理的收益从 35.7 分跌到 2.7 分 AI 大模型
一天一个开源项目（第224篇）：ARTEMIS —— 谷歌开源的移动端 AI 自动化框架，让 AI 助手像人一样操作手机 AI 大模型
一个周末，6个项目，我第一次感觉 AI 编程真的进入了新阶段 AI 大模型
LLM 自动化测试系列（01）：为什么测试是 LLM 落地的新战场 AI 大模型