Pi Durable：用于长期运行、多用户智能体应用程序的框架

TL;DR – Pi Durable 是什么及其重要性

Pi Durable 是随 Pi 1.0 发布的一个新的实验性框架，它使智能体能够在任何地方运行、在进程崩溃后存活、处理无限长的对话，并支持多用户和扩展。它提供了构建持久、可塑的智能体应用程序所需的存储、任务检查点和执行环境抽象。

核心问题：Pi 1.0 是一个单用户、终端绑定的智能体

Pi 1.0 旨在让单个用户在终端内驱动编码智能体。如果进程死亡，用户需要手动重启它并告诉智能体继续工作。这种模式适用于交互式编码，但不支持：

全天候智能体：无需人工干预运行数天或数周。
多用户协作：多个客户端同时观察或引导同一对话。
并发对话：例如 Slack 线程或并行子智能体。
稳健的恢复：从容器重启、虚拟机崩溃或内存溢出终止中恢复。

Pi Durable 在不改变原有 Pi 编码智能体的情况下解决了这些差距。

框架定义 – 存储 + 执行机制

Pi Durable 中的框架（harness）是以下各项的组合：

存储后端 – 持久化记录、任务和类型化的 JSON 文档。
执行环境 – 为智能体提供工具（文件访问、shell 等）。
任务引擎 – 将模型请求、工具调用和压缩作为持久任务运行。

该框架公开了一个简单的 API（Harness.open、harness.root、conversation.submit 等），智能体可以理解并操作这些 API。

在任何地方运行长期智能体

Pi Durable 开箱即用提供三个存储适配器：

Memory – 用于测试的纯内存存储。
SQLite – 基于磁盘的关系型存储。
JSONL – 行分隔的 JSON 日志。

所有适配器都公开相同的最小接口，使得实现自定义后端（例如 Postgres、DynamoDB 或 Cloudflare Durable Objects）变得非常简单。同一时间只有一个进程拥有存储实例；其他客户端作为读取者/写入者连接。

该框架仅在内存中保留工作集（活动记录、实时任务、待处理提交）。较旧的消息被压缩为摘要，因此即使是拥有数万条消息的对话也能舒适地适应模型的上下文窗口。

示例：打开 SQLite 框架

import { Harness, openNodeSqliteStorage } from "@earendil-works/pi-durable";
import { createModels, openaiProvider } from "@earendil-works/pi-ai";
import { createRegistry, CodingTools } from "@earendil-works/pi-durable/tools";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";

const models = createModels();
models.setProvider(openaiProvider());
const registry = createRegistry();
registry.install(CodingTools);

const env = ({ cwd }: { cwd?: string }) =>
  new NodeExecutionEnv({ cwd: cwd ?? process.cwd() });

const harness = await Harness.open(
  await openNodeSqliteStorage("./agent.sqlite"),
  { models, registry, env },
  BACKGROUND_CONTEXT,
);

const root = await harness.root(BACKGROUND_CONTEXT, {
  agent: { model: { provider: "openai", modelId: "gpt-6.1-sol" }, cwd: "/work/repo" },
});

崩溃恢复执行

每一步——模型请求、工具调用或压缩——都是一个任务，在继续之前会对其输入和输出进行检查点记录。如果进程死亡，新进程会重新打开相同的存储，发现未完成的任务，并从最后一个检查点恢复它们。

被中断的模型调用会被重新发送；部分答案在记录中被标记为已中止（aborted）。
工具调用声明一个 replay 策略（"safe" 或省略）。安全调用会自动重试；不安全调用会通知模型它们已被中断。
requestId 保证了提交的精确一次（exact-once）语义：崩溃后的重试会返回原始结果，而不是发出重复请求。

崩溃恢复示例

const job = { type: "input", content: "Fix the flaky login test", requestId: "job-42" } as const;
await root.submit(job, ctx); // 进程在工具调用期间死亡

// 新进程恢复
const harness2 = await Harness.open(await openNodeSqliteStorage("./agent.sqlite"), { models, registry, env }, ctx);
harness2.resume();
const root2 = await harness2.root(ctx);
const settled = await (await root2.submit(job, ctx)).wait(ctx); // 返回相同的答案

并发对话与分支

单个框架可以托管许多并行运行且互不阻塞的独立对话。对话可以在任何记录条目处分支（fork），继承父级直到分支点的历史记录，但此后会分道扬镳。

分支模拟了 Slack 频道和线程： 主频道是一个对话；每个线程都是一个并行运行的分支。

分支示例

const channel = await harness.root(ctx);
const q1 = await channel.submit({ type: "input", content: "@agent why did the deploy fail?" }, ctx);
const answered = await q1.wait(ctx);

const thread = await channel.fork(answered.answer!, { ownership: { kind: "ownerless" } }, ctx);
await Promise.all([
  thread.submit({ type: "input", content: "@agent can we roll it back?" }, ctx).then(t => t.wait(ctx)),
  channel.submit({ type: "input", content: "@agent who is on call today?" }, ctx).then(t => t.wait(ctx)),
]);

每个对话存储其自己的智能体配置（模型、工具、工作目录），从而实现廉价的审阅者或专门的子智能体。

可扩展架构 – 扩展、工具、钩子和任务

扩展

扩展捆绑了：

系统提示词部分（在每次模型请求前重新渲染）。
工具（模型可以调用的函数）。
钩子（拦截或修改任务执行）。
自定义任务（用户定义的持久工作流）。

对话按名称选择扩展；只有名称被持久化，因此重启后它们会自动加载最新的代码。

系统提示词部分

部分是读取对话执行环境并返回插入到系统提示词中的文本的函数。更改在记录中进行版本控制，因此重启后模型看到的完全一致。

import { defineExtension, section } from "@earendil-works/pi-durable";
const ProjectContext = defineExtension({
  name: "project-context",
  sections: [
    section("agents_md", (input) => agentsMd.latest(input.env)),
    section("skills", (input) => skills.latest(input.env)),
  ],
});

工具与重放语义

工具是持久任务。replay 标志告诉 Pi Durable 工具在崩溃后是否可以安全重运行。

const searchIssues = defineTool({
  name: "search_issues",
  description: "Search the issue tracker",
  parameters: Type.Object({ query: Type.String() }),
  replay: "safe",
  execute: async (args, api) => {
    api.output(`searching for ${args.query}\n`);
    return { content: [{ type: "text", text: await tracker.search(args.query) }] };
  },
});

像 deploy 这样的非安全工具会省略 replay；如果被中断，模型会收到通知，且操作不会重复。

钩子 – 拦截任务

钩子可以修改或阻止模型请求、工具调用或压缩。它们按对话所选扩展的顺序链式运行。

import { hook, ToolTask } from "@earendil-works/pi-durable";
const Approval = defineExtension({
  name: "approval",
  hooks: [
    hook(ToolTask, {
      beforeTool: async (call, api, ctx) => {
        if (call.name !== "deploy") return undefined;
        let approved = await api.memo<boolean>("approval:deploy", ctx);
        approved ??= await api.memo("approval:deploy", await askInSlack(call), ctx);
        return approved ? undefined : { block: "Nobody approved the deploy." };
      },
    }),
  ],
});

钩子在崩溃后依然有效，因为它们将决策存储在附加到任务的备忘录中。

自定义任务 – 持久工作流

任务是一等持久单元，具有明确的阶段、检查点和中止逻辑。它们可以拥有其他任务，形成一个所有权树，保证干净的中止传播。

const Payment = defineTask<{ card: string }, { phase: "charge" }, string>({
  name: "shop.payment",
  version: 1,
  initial: () => ({ phase: "charge" }),
  phases: {
    charge: async (task, rt, ctx) => {
      const charge = await bank.charge(task.input.card, `payment-${task.id}`);
      await rt.commit(() => ({
        status: "terminal",
        outcome: charge.ok ? { status: "completed", result: charge.receipt }
                           : { status: "failed", error: { message: charge.error } },
      }), ctx);
    },
  },
  abort: async (task, rt, ctx) => {
    await bank.refund(`payment-${task.id}`);
    await rt.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx);
  },
});

更高级别的 Checkout 任务协调多个 Payment 任务，展示了快速失败（fail-fast）编排和中止时的自动退款。

无限上下文的自动压缩

当对话接近模型的上下文限制时，Pi Durable 会运行一个后台压缩任务来总结较旧的消息。摘要被插入到下一个轮次边界，将活动窗口保持在 reserveTokens（默认 16 384）内。如果提供商仍然拒绝请求，框架会再次压缩并重试。

await harness.root(ctx).compact("Keep the names of the failing tests", ctx);

reset() 操作可以启动一个新的上下文，同时保留较旧的消息以供后续搜索，从而实现交接模式。

通过类型化文档实现持久应用程序状态

应用程序状态（例如待办事项列表）存在于文档中——与记录原子存储的类型化 JSON 对象。文档支持：

作用域（conversation 或 global）。
历史模式（rewindable 用于版本化读取）。
分支行为（asOf 在分支点继承父级状态）。

const Todos = defineDoc<{ items: string[] }>({
  kind: "app.todos",
  version: 1,
  scope: "conversation",
  history: "rewindable",
  fork: "asOf",
  initial: () => ({ items: [] }),
});

工具可以在事务内读写文档，保证记录和状态永远不会分歧。

可塑性 – 热插拔扩展

注册表可以在运行时更新。以现有名称安装扩展会立即替换它。正在进行的工具调用以它们开始时的代码完成；后续调用使用新实现。因为对话只存储名称，重启会自动获取最新版本。

registry.install(await loadExtension("./ops.ts")); // 替换之前的 "ops" 扩展

多人协作

所有 UI 状态都源自已提交的存储，因此任意数量的客户端都可以连接到对话，接收当前视图，然后订阅增量更新。

const view = await thread.viewState(ctx);
view.subscribe(v => render(v));
await thread.submit({ type: "input", content: "Check the staging logs first", whenBusy: "steer" }, ctx);

客户端还可以观察低级提交流（thread.watch()）或更高级别的事件流（watchEvents()）。

社区反应（精选 HN 评论）

"很高兴看到 Pi 也构建了一个持久智能体框架。我自己在这个领域已经构建了很长一段时间……主要参与者正在这个领域构建产品：LangChain Deep Agents、Vercel Eve、OpenAI Agents API、Anthropic Managed Agents 等。" – lukebuehler

"整个源代码（不含测试）大约有 15,000 行，对于 GPT 来说大约是 150,000 个 token，对于 Claude 来说大约是 250,000 个。" – ireadmevs（强调了代码库大小和 token 数量的差异）。

"我最喜欢多用户那一点。应该能让我更容易构建我的远程控制工具……" – skeledrew

"太棒了。但我希望持久应用程序状态不仅仅局限于 json 文档。应该有一种集成的方式来实现发件箱模式……" – vmg12

"来自 1.0 线程的交叉发布。我正在为 Slack 构建一个框架……处理 JSONL 会话文件和 pod 中断增加了复杂性。我现在正在为此使用 DBOS。" – azuanrb

这些评论强调了社区对持久性、多用户支持和灵活存储后端的渴望。

入门

Pi Durable 是实验性的；API 可能会演变。要在本地尝试：

npm install @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chord
# 克隆 Pi 仓库并运行示例
npm install && npm run build
node packages/coding-agent/src/experimental/durable/main.ts
node packages/coding-agent/src/experimental/vacation/main.ts

探索 packages/durable/test/examples 中的 30 多个示例脚本以及度假规划 TUI，以获取具体的用例。

展望

Pi Durable 为构建全天候、协作式智能体提供了一个坚实、可扩展的基础。通过将框架与编码智能体分离，Earendil 可以在不破坏核心 Pi 体验的情况下迭代持久性、存储和多用户功能。在接下来的几周里，我们将看到更多生产级扩展（例如 Slack 机器人、GitHub 分流智能体）以及与外部状态存储的更深层集成。