持久化执行

跑了半天崩了

你的 Agent 跑到一半，LLM 超时了 / 网络断了 / 服务器重启了 —— 然后呢？

开始之前

什么是持久化

一句话：在关键节点保存进度，暂停后能从断点精准恢复

持久化执行（Durable Execution）不是什么新概念——数据库事务、消息队列都有类似的"至少一次"语义。在 LangGraph 中，它的含义很简单：

三个前提

两个启用条件 + 一个编码规范

// MemorySaver 用于开发
const graph = workflow.compile({
  checkpointer: new MemorySaver()
});

// 同一任务用相同 thread_id
const config = {
  configurable: {
    thread_id: "task-001"
  }
};

import { task } from "@langchain/langgraph";

const result = await task(
  "my-api-call",
  async () => fetch(url)
);

// 完整示例：满足三个前提的持久化执行图
import { StateGraph, MemorySaver, START, END, task } from "@langchain/langgraph";
import { Annotation } from "@langchain/langgraph";

const State = Annotation.Root({
  url: Annotation<string>,
  result: Annotation<string>,
});

const fetchNode = async (state) => {
  // 用 task() 包裹网络请求——重放时自动跳过
  const html = await task("fetch-url", async () => {
    const resp = await fetch(state.url);
    return await resp.text();
  });
  return { result: html.slice(0, 100) };
};

const graph = new StateGraph(State)
  .addNode("fetch", fetchNode)
  .addEdge(START, "fetch")
  .addEdge("fetch", END)
  .compile({ checkpointer: new MemorySaver() });  // 前提 1

const config = { configurable: { thread_id: "crawl-1" } };  // 前提 2

📄 完整代码：06_task_wrapping.ts

回放机制

最重要的概念：恢复时从头重放，不是从断点继续

编码原则

理解了回放机制后，写代码时要遵循三条原则

// 错误：直接调用
const data = await fetch(url);

// 正确：用 task() 包裹
const data = await task("fetch-data",
  async () => await fetch(url)
);

// 错误：每次重放结果不同
const id = crypto.randomUUID();

// 正确：重放时使用缓存的结果
const id = await task("gen-id",
  () => crypto.randomUUID()
);

// 使用幂等写入：upsert 而非 insert
await task("save-result",
  async () => db.collection.updateOne(
    { _id: taskId },  // 唯一键
    { $set: { result } },
    { upsert: true }
  )
);

📄 完整代码：06_coding_principles.ts

持久化模式

三种模式控制检查点的保存时机和频率

// 三种模式的使用方式——只在调用时指定
const config = { configurable: { thread_id: "task-1" } };

// exit 模式：只在图结束时保存
await graph.stream({ input: "test" }, { ...config, durability: "exit" });

// async 模式：异步保存（默认）
await graph.stream({ input: "test" }, { ...config, durability: "async" });

// sync 模式：每步同步保存——最安全
await graph.stream({ input: "test" }, { ...config, durability: "sync" });

📄 完整代码：06_durability_modes.ts

task() 详解

用 task() 包裹副作用，让重放时自动跳过已执行的操作

// task() 的签名
import { task } from "@langchain/langgraph";

// 第一个参数：唯一名称（同一个节点内不能重复）
// 第二个参数：实际要执行的异步函数
const result = await task("fetch-url", async () => {
  const response = await fetch(state.url);
  return await response.text();
});

在节点中使用 task()

import { task } from "@langchain/langgraph";

const researchNode = async (state) => {
  // 任务 1：获取网页内容（副作用 → 用 task）
  const html = await task("fetch-page", async () => {
    const resp = await fetch(state.url);
    return resp.text();
  });

  // 任务 2：调用 LLM 分析（副作用 + 花钱 → 用 task）
  const analysis = await task("analyze", async () => {
    const llm = new ChatOpenAI({ model: "gpt-4o-mini" });
    const result = await llm.invoke(html.slice(0, 3000));
    return result.content;
  });

  // 纯计算不需要 task()——没有副作用，重放也无害
  const summary = analysis.slice(0, 200);

  return { result: summary };
};

const graph = new StateGraph(State)
  .addNode("research", researchNode)
  .addEdge(START, "research")
  .addEdge("research", END)
  .compile({ checkpointer: new MemorySaver() });

📄 完整代码：06_task_wrapping.ts

恢复场景

两种核心恢复方式：暂停/恢复 和 故障恢复

场景一：暂停与恢复（HITL）

import { interrupt, Command } from "@langchain/langgraph";

const reviewNode = async (state) => {
  // 暂停，等待人工审核
  const humanDecision = interrupt("请审核这个结果：" + state.draft);

  // humanDecision 就是恢复时传入的值
  if (humanDecision === "approve") {
    return { status: "approved" };
  }
  return { status: "rejected", feedback: humanDecision };
};

// 恢复执行——传入人工决策
await graph.invoke(
  null,  // 不需要新输入
  {
    ...config,
    command: new Command({ resume: "approve" }),
  }
);

场景二：故障恢复

// 第一次调用——假设在节点 C 崩溃了
try {
  await graph.invoke({ input: "start" }, config);
} catch (e) {
  console.log("崩溃了，准备恢复...");
}

// 恢复：用同一个 thread_id 重新调用，自动从断点恢复
const result = await graph.invoke(
  null,  // 不需要新输入，检查点已有状态
  config  // 同一个 thread_id
);
console.log(result);  // 从崩溃点继续执行

📄 完整代码：06_fault_recovery.ts

排错指南

持久化执行中的常见问题与修复方案

动手练习

构建一个带 task() 和故障恢复的可靠 Agent

// 参考答案：练习 1 & 2
import { StateGraph, MemorySaver, START, END, task, interrupt, Command } from "@langchain/langgraph";
import { Annotation } from "@langchain/langgraph";

const State = Annotation.Root({
  url: Annotation<string>,
  content: Annotation<string>,
  analysis: Annotation<string>,
  report: Annotation<string>,
});

const fetchNode = async (state) => {
  const html = await task("fetch-page", async () => {
    const resp = await fetch(state.url);
    return resp.text();
  });
  return { content: html.slice(0, 500) };
};

const analyzeNode = async (state) => {
  // 等待人工确认分析内容
  const approved = interrupt(`内容预览：${state.content.slice(0, 100)}...确认分析？`);

  const result = await task("llm-analyze", async () => {
    // 模拟 LLM 调用
    return `分析结果：${state.content.length} 字符`;
  });
  return { analysis: result };
};

const reportNode = (state) => {
  return { report: `报告：${state.analysis}` };
};

const graph = new StateGraph(State)
  .addNode("fetch", fetchNode)
  .addNode("analyze", analyzeNode)
  .addNode("report", reportNode)
  .addEdge(START, "fetch")
  .addEdge("fetch", "analyze")
  .addEdge("analyze", "report")
  .addEdge("report", END)
  .compile({ checkpointer: new MemorySaver() });

const config = { configurable: { thread_id: "exercise-1" } };

// 第一次调用——会在 analyze 节点暂停
await graph.invoke({ url: "https://example.com" }, config);

// 查看暂停状态
const snap = await graph.getState(config);
console.log("下一步:", snap.next);  // ["analyze"]

// 恢复——传入确认
await graph.invoke(null, { ...config, command: new Command({ resume: "yes" }) });

本章回顾

持久化执行让长时间运行的智能体变得可靠

延伸阅读

继续深入持久化执行，或探索 LangGraph 的其他核心能力