Pi Durable 运行在 Cloudflare 上

Earendil Posts，Earendil 是一家公益公司，打造软件和开放协议，以增强人的主体性、弥合分歧、培育持久的快乐。关注

许多AI代理会响应用户请求或其他外部事件，完成工作后长时间闲置。让机器全天候运行会浪费计算，且成本迅速上升。

Pi Durable 是一款为长期工作特工设计的背带。阅读公告详细解释了其概念和存储模型。Cloudflare Durable Objects 提供了一个有状态的对象，在空闲时关闭，遇到请求、WebSocket 消息或警报时重新启动。

在这篇文章中，我们将构建一个聊天代理，它将每个会话存储在自己的持久对象中，管理每一次会话的多次对话，能够通过休眠存活，为安排的工作唤醒，并使用子代理进行跨对话搜索。

界面会带有90年代的氛围，但多玩家模式，因为现在是2026年：多个参与者连接在同一聊天会话中。

每个耐用对象一次 Pi 会话

Pi Durable 线束的作用域仅限于一个会话。该会话自然映射到一个由会话 ID 定位的 Durable Object 实例。其 SQLite 支持的存储会持久化会话数据，Durable Object 可以在空闲时休眠，闹钟可以唤醒它以处理待处理的工作。

React client
    │ PartySocket
    ▼
Worker
    │ route by session ID
    ▼
SQLite-backed Durable Object instance
    ├── Hibernatable WebSockets
    ├── embedded SQLite database
    │       └── conversations and tasks
    └── Pi Durable Harness ──▶ Model provider

工作者创建会话 ID，返回一个可共享的 URL，客户端通过 WebSockets 连接。所有打开相同会话 URL 的人都会到达同一个对象。不同的会话 ID 分别指向具有隔离状态的不同对象。

在持久对象存储中持久化Pi会话数据

Pi Durable 不假设会话存储在哪里。如果内置内存、JSONL 或 SQLite 支持不完全满足需求，你可以自带存储后端。SQLite 实现定义了一个运行时中立的异步数据库接口，包含一个用于Node.js的适配器，以及一个用于使用同步存储的 Cloudflare Durable 对象适配器SqlStorage通过 APIctx.storage.sql.

await openDurableObjectSqliteStorage(ctx.storage)

如果你需要自己运行SQLite适配器或其他存储后端，可以使用Pi Durable的一致性套件来检查它的行为是否正确。它能支持任何兼容Vitest或Jest的测试运行程序。

对于Durable Object，可以搭配Cloudflare Workers Vitest 集成:

registerStorageConformance(
  { describe, expect, it },
  "Cloudflare Durable Object SQLite",
  async (use) => {
    const stub = env.SESSIONS.get(env.SESSIONS.newUniqueId());
    await runInDurableObject(stub, async (_instance, state) => {
      await use(await openDurableObjectSqliteStorage(state.storage));
    });
  },
);

当耐用物体开始时打开圆周率

线束初始化时会用Session持久对象。Cloudflare 在对象启动时运行构造子：首次创建时，以及休眠或驱逐后请求、警报或 WebSocket 消息到达时运行构造函数。

所有事件都会被保留，直到初始化完成。

export class Session extends DurableObject {
  #harness!: Harness;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    void ctx.blockConcurrencyWhile(() => this.#initialize());
  }

  async #initialize(): Promise {
    const storage = await openDurableObjectSqliteStorage(this.ctx.storage);
    const chatOptions = createChatHarnessOptions((timestamp) =>
      this.#setAlarmNoLaterThan(timestamp),
    );
    const harness = await Harness.open(
      storage,
      {
        ...chatOptions,
        onReport: (error) => reportError("Harness background task failed",
 error),
      },
      BACKGROUND_CONTEXT,
    );

    // A storage error closes the harness. Reset the object so the next event reopens it.
    void harness.closed.then((end) => {
      if (end.reason === "failed") this.ctx.abort("Harness failed");
    });

    try {
      this.#harness = harness;
      await this.#restoreEventStreams();
      await this.#refreshAlarm();
    } catch (error) {
      await harness.close(BACKGROUND_CONTEXT);
      throw error;
    }
  }
}

这是该框架的最小集成。完整实现还会向客户端实时传输任务活动，并在创建子代理对话时告知他们。

在一个共享会话中运行多个对话

会话是持久的边界，但它不是一个庞大的文字记录。聊天参与者可以在同一会话内创建独立对话，或者从已提交的用户或助理条目中分叉对话。

为避免重复，例如在断开 WebSocket 连接后，线束会在同一提交中存储包含请求 ID 的回执：

const conversationId = await this.#harness.commit(async (tx) => {
  const doc = await tx.doc(ConversationCommandReceipts, requestId, null);
  if (doc.receipt !== undefined) return doc.receipt.conversationId;

  const conversation = await tx.createConversation({ ownership: { kind: "ownerless" } });
  doc.receipt = { command: "conversation.create", conversationId: conversation.id };
  return conversation.id;
}, BACKGROUND_CONTEXT);

创建分支类似，只不过它接收父节点直到所选条目的历史：

const fork = await tx.forkConversation(parentConversationId, entryId, {
  ownership: { kind: "ownerless" },
});

当聊天参与者发送消息时，会话会将其与显示名称关联，并作为后续提示提交。

const conversation = await this.#harness.conversation(conversationId, BACKGROUND_CONTEXT);
await conversation?.submit(
  {
    type: "input",
    content: attributedPrompt(username, content),
    requestId,
    whenBusy: "followUp",
  },
  BACKGROUND_CONTEXT,
);

如果多人在模型或工具运行时发送消息，消息会进入Pi的耐久收件箱，并按顺序回复。

流畅的持久对话状态

对于每个连接的客户端，会话会为所选对话打开 Pi 事件流。流从原子快照开始，然后发射已提交的转录修改、流式输出、工具、压缩、重置和收件箱变更的事件。

const stream = await watchEvents(
  this.#harness,
  conversationId,
  BACKGROUND_CONTEXT,
);

socket.serializeAttachment({ username, conversationId });
socket.send(JSON.stringify({
  type: "conversation.events",
  conversationId,
  events: [stream.snapshot],
}));

stream.start((events) => {
  socket.send(JSON.stringify({
    type: "conversation.events",
    conversationId,
    events,
  }));
});

React客户端会根据这些事件更新其选定的对话状态，并使用单独的会话快照填充侧边栏。

休眠后恢复工作

持久消息只是耐久性的一部分。应用程序还必须重建实时视图，恢复中断的Pi任务，并在未来的截止日期唤醒。

Cloudflare 在对象休眠时保持 WebSocket 开启。每个 socket 存储参与者的名字和所选对话，因此当对象唤醒时，会重新打开匹配的 Pi 事件流。

如果连接断开，客户端会用同一请求 ID 重新发送服务器未确认的内容，这样就不会重复运行。

恢复实时视图后，会话检查持久化的任务集。未来的截止日期会投影到持久对象警报上。对于可以立即运行的任务，waitForTask()启用 Pi 的调度器并监控结算，之后会话重新计算下一个闹钟。

将圆周率截止日期映射到耐用物体报警

该智能体被赋予schedule工具让它之后能行动，比如几分钟后查看天气。在幕后，这被转化为一个持久的圆周率任务，直到任务截止日期才会休眠。

Pi可以跟踪任意数量的截止日期，但耐用物体只有一个闹钟。如果新安排的任务截止日期先到，则会提前报警。每次目标唤醒时，都会根据Pi的任务重建闹钟。

仍在进行的工作会触发短暂的恢复闹钟，因此被淘汰的对象会回来完成：


override async alarm(): Promise {
  await this.#refreshAlarm();
}

#refreshAlarm(): Promise {
  return this.#serializeAlarmUpdate(async () => {
    const { tasks } = await this.#harness.inspect(BACKGROUND_CONTEXT);
    const now = Date.now();
    let nextAlarm: number | undefined;
    const tasksToResume: TaskId[] = [];

    for (const task of tasks) {
      if (task.state.kind === "blocked") continue;
      const scheduled = scheduledDeadline(task);
      const waitsForFuture =
        scheduled !== undefined && scheduled > now && !task.record.abortRequested;
      const deadline = waitsForFuture ? scheduled : now + RECOVERY_ALARM_DELAY_MS;
      nextAlarm = nextAlarm === undefined ? deadline : Math.min(nextAlarm, deadline);
      if (!waitsForFuture) tasksToResume.push(task.record.id);
    }

    await this.#replaceAlarm(nextAlarm);
    // waitForTask starts the scheduler, so persist its recovery alarm first.
    for (const taskId of tasksToResume) this.#refreshAlarmWhenTaskSettles(taskId);
  });
}

由于警报至少执行一次，处理器必须是幂等型的。#refreshAlarm()它会重新计算存储任务的所有内容，而稳定的调度请求ID会阻止指令重复提交。

搜索与持久分级代理的对话记录

在一次会话中有多个对话时，参与者可能会询问对话上下文之外的问题。与其将所有文字记录加载到父对话中，不如search_history工具将检索委托给子代理。

在 Pi Durable 中，子代理是创建该工具任务的普通对话。

它会搜索来自顶级对话的用户和助手消息，包括发起搜索的对话中更早的消息。它无法搜索自身对话或由之前搜索创建的消息，防止生成的搜索结果成为未来搜索的证据。

Parent conversation
    │ search_history(question)
    ▼
Durable tool task
    │ owns
    ▼
Child conversation
    │ search_conversations
    ▼
Final answer returned to parent

该工具被标记为可重放安全，因此如果耐用对象在调用中途被淘汰，Pi 会在下一次开始时重新运行，而不是报告为中断：

const childId = await api.commit(async (tx) => {
  // Reuse the original child if this tool task is replayed.
  const existing = (
    await tx.scanConversations(
      { ownerTaskId: api.taskId },
      1,
    )
  ).items[0];

  if (existing !== undefined) return existing.id;

  const child = await tx.createConversation({
    ownership: {
      kind: "task",
      taskId: api.taskId,
    },
  });

  await configure(tx, child.id, {
    extensions: [conversationSearchExtension],
    tools: [searchConversationsTool],
  });

  return child.id;
}, context);

// Let the interface link from the parent tool call to its subagent.
await api.details({ conversationId: childId }, context);

const child = await api.conversation(childId, context);
if (child === undefined) {
  throw new Error(`Search conversation ${childId} is missing`);
}

const submission = await child.submit(
  {
    type: "input",
    content:
      `Answer this question using search_conversations: ${question}\n\n` +
      "Try a few relevant terms. Reply briefly, cite conversation IDs, " +
      "and say so if nothing relevant was found.",
    requestId: `search-history:${api.taskId}`,
  },
  context,
);

const settled = await submission.wait(context);
if (settled.status !== "done" || settled.type !== "input") {
  throw new Error(`History search did not answer: ${settled.status}`);
}

const answer = await api.commit(
  (tx) => tx.entry(AssistantEntry, settled.answer),
  context,
);

return {
  output: [{
    type: "text",
    text: messageText(answer?.model?.[0]),
  }],
  details: { conversationId: childId },
};

所有权赋予孩子持久的身份。如果工具被打断并重播，它会找到由原始任务创建的对话并继续。

该search_conversations工具是唯一提供给子代理的工具，在本示例实现中采用了刻意简单的子字符串搜索。原始匹配和中间工具调用在子代理对话中保持隔离。

父对话只接收最终答案。

本地运行

你需要Node.js 22.19或更新版本，pnpm 11，以及OpenAI API密钥。克隆样本库，安装其依赖，并复制示例环境文件：

git clone https://github.com/earendil-works/pi-durable-cloudflare.git
cd pi-durable-cloudflare
pnpm install
cp .dev.vars.example .dev.vars
OPENAI_API_KEY=your-key
pnpm dev

创建一个会话，在第二个浏览器中打开其URL，然后尝试以下操作：

  1. 在第一次对话中，可以说：“我们决定用PostgreSQL来做事件商店。”
  2. 再开一个对话。
  3. 问：“我们决定了活动商店什么？”
  4. 从侧边栏打开搜索子代理，检查其搜索结果。
  5. 让客服两分钟后查看天气，然后关闭浏览器，截止时间后再回来。

请注意，演示版故意没有认证。

用Pi耐用度构建

这个例子刻意较小，但它展示了 Pi Durable 的关键功能：持久会话、可插拔存储、长时间运行和计划工作以及子代理。

开始在 Pi Durable 上构建你的代理。我们很想听听你的想法以及你是如何使用它的。

添加评论
点赞收藏
点踩分享查看原文
评论
?
参与讨论