Pi Durable

# Pi Durable | Earendil 来源：https://earendil.com/posts/pi-durable/ 今天，Earendil 与 Pi 社区一起发布了 Pi 1.0（https://earendil.com/posts/pi-1-0/）。这体现了我们的信念：在经过无数小时的加固、维护和积极开发之后，Pi 现已是一个坚实的基础，可以在此之上构建。同时，Pi 也在持续演进。与 Pi 1.0 一同发布的，还有一个实验性的新包——Pi Durable。 Pi Durable 专为长期运行、可持久化且可塑性强的智能体而构建，可在任何地方运行。我们希望你能加入其中，帮助我们把它打造成最好的持久化运行时（durable harness）。 ## 为什么需要 Pi Durable？ 作为编码智能体的 Pi，其设计目标是在你的（远程）机器上运行，位于终端之中，由一个人驱动。如果进程挂掉了，你查看发生了什么，然后让它继续。这正是 Pi 1.0 专注并擅长的事情，这一点不会改变。 在 Earendil，我们希望把这项技术带给所有人，以最契合其需求的形式。为此，我们需要一个能在任何地方运行、可从不同入口访问、支持无限长对话、能在灾难性的内部与外部故障中存活、并允许多人同时引导同一个智能体的运行时。Pi Durable 正是这样的运行时。 它并不取代 Pi 编码智能体。它是一个用于构建任意智能体应用的框架，编码智能体也包括在内。它不仅与 Pi 编码智能体共享代码（比如 pi-ai），也共享其原则：极简主义与可塑性。它还让我们能够在这一领域探索各种设计，而不干扰 Pi 编码智能体。在 Pi Durable 上构建智能体应用的过程中，我们学到的经验教训，只要被证明有价值，就会回流到 Pi 编码智能体中。 ## 什么是运行时（harness）？ 每个人对运行时都有自己的定义。我们之前就此写过文章（https://earendil.com/posts/what-is-a-harness/），但让我们为 Pi Durable 重新介绍运行时的概念。 运行时（harness） = 存储 + 运行一个或多个与大语言模型并行对话所需的机制。它提供模型所调用的工具，以及工具运行的执行环境。 一次「对话」是指你与一个智能体之间的交互，被记录为一份对话记录（transcript）。智能体就是大语言模型本身，连同它的设置（比如思考等级）以及它能调用的工具。工具通过执行环境来完成工作，执行环境可以是你的笔记本电脑、一台远程虚拟机，或者一个内存沙箱。每个对话决定智能体获得哪些工具以及使用哪个执行环境。 运行时运行的一切，从调用模型到执行工具，都叫做一个任务（task）。和 Pi 中的一切一样，Pi Durable 的构建方式让智能体能够理解它。不含测试的全部源代码约 15,000 行，用 GPT 大约折合 150,000 个 token，用 Claude 大约 250,000 个。这是最坏情况。要在 Pi Durable 之上构建，你的智能体几乎很少需要全部代码；仅存储后端就有 3,000 行，它通常可以跳过。 现在让我们带你简单参观一下 Pi Durable，以说明我们构建了什么以及为什么这样构建。 ## 长时间、随处运行 我们希望智能体能长时间运行，并且能在任何地方运行——目前的「任何地方」指的是任何有 JavaScript 运行时的地方。在 Pi Durable 中，运行时通过打开一个存储后端来启动。Pi Durable 内置了内存、SQLite 和 JSONL 三种存储方式，还附带一套一致性测试套件，以及针对你自己的后端的基准测试。SQLite 和 JSONL 的存储代码不使用任何 Node API，因此通过一个小的适配器，它就能在 Bun 或 Cloudflare Durable Object 中运行。存储接口很小，在你现有的东西（比如键值存储或 Postgres）之上很容易实现。 同一时刻只有一个进程拥有某个存储，其他客户端附着到该进程上。在 SQLite 上，运行时只把工作集保留在内存中：活跃的对话记录、进行中的任务和待处理的提交。其他一切都在磁盘上，直到需要时才读取。活跃的对话记录天然被模型的上下文窗口所限，因为压缩机制会在旧消息溢出之前对它们进行摘要。因此，即使一个对话包含数万条消息，也能恰好放进内存。 需要文件或 shell 的工具从执行环境获得它们。Pi Durable 内置了一个 Node 执行环境，让工具能够访问你本地的文件。与存储一样，执行环境的接口也很小、很容易实现，因此你也可以为你的工具暴露远程执行环境。这就允许运行时运行在一台机器上，而它的工具运行在另一台机器上。你的 `env` 函数会为每次工具调用构建环境，从对话的工作目录出发，这样每个对话都可以运行在不同的位置。 ```js import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { createModels } from "@earendil-works/pi-ai/models"; import { openaiProvider } from "@earendil-works/pi-ai/providers/openai"; import { createRegistry, Harness } from "@earendil-works/pi-durable"; import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node"; import { openNodeSqliteStorage, } from "@earendil-works/pi-durable/storage/sqlite/node"; import { CodingTools } from "@earendil-works/pi-durable/tools"; const context = BACKGROUND_CONTEXT; // 每次调用都传入一个用于取消的上下文 const models = createModels(); models.setProvider(openaiProvider()); const registry = createRegistry(); registry.install(CodingTools); // 读、写、编辑、bash const env = ({ cwd }: { cwd?: string }) => new NodeExecutionEnv({ cwd: cwd ?? process.cwd() }); const harness = await Harness.open( await openNodeSqliteStorage("./agent.sqlite"), { models, registry, env }, context, ); // 根对话：首次使用时创建，之后每次重启后还是同一个 const root = await harness.root(context, { agent: { model: { provider: "openai", modelId: "gpt-6.1-sol" }, cwd: "/work/repo", }, }); ``` ## 在崩溃中存活 我们希望智能体即使进程挂掉也能存活下来——无论是笔记本睡眠、容器被重新部署，还是机器内存耗尽——并从中断处继续。在 Pi Durable 中，一次运行的每一步都是一个任务，它在继续之前会先存储一个检查点（checkpoint）。如果进程挂掉，一个新的进程会打开同一个存储，找到未完成的任务，并从各自的最后一个检查点继续。 被中断的模型请求会被重新发送；已有的部分答案会留在对话记录中，标记为已中止。被中断的工具调用，如果可以安全重跑就会重跑；否则会告知模型该调用被打断了。 Pi Durable 没有内置的子智能体，但只需几行代码就能构建，下面的 triage（分流）工具就演示了这一点。子智能体运行在自己的对话中，所以它也能从中断处继续；一个可以安全重跑的子智能体工具会重新找到它的子智能体，并等待它的回答。排队中的消息仍然保持排队。一个 `requestId` 让一次提交是「恰好一次」的，因此崩溃后重试的客户端会拿回原始提交结果，而不是提交两次。 ```js const job = { type: "input", content: "Fix the flaky login test", requestId: "job-42", } as const; await root.submit(job, context); // 进程在这里挂掉，卡在一次工具调用中间。 // 一个新的进程打开同一个存储。 const harness = await Harness.open( await openNodeSqliteStorage("./agent.sqlite"), { models, registry, env }, context, ); harness.resume(); // 继续被中断的运行 const root = await harness.root(context); // 同一个提交，已经得到回答 const settled = await (await root.submit(job, context)).wait(context); ``` ## 同时运行多个对话 我们希望一个运行时能同时运行多个对话，而不会相互阻塞。在 Pi Durable 中，一个运行时可以并发地运行你需要的任意数量的对话，且都享有相同的保证。 一个对话可以从零开始，也可以在任意时刻分叉出另一个对话，并且能看到父对话到那一刻为止的历史，而无需复制它。想象一个 Slack 频道，你的智能体回答来自任何人的 @ 提及。然后有人开启了一个帖子（thread）。频道可以是一个对话，帖子则是它在所回复那条消息处的分叉。两者同时运行，互不阻塞。 ```js const channel = await harness.root(context); const question = await channel.submit( { type: "input", content: "@agent why did the deploy fail?" }, context, ); const answered = await question.wait(context); // 有人在帖子中回复了智能体的答复。每个对话都会标明它的拥有者， // 这决定了「中止」会影响到谁（关于「任务」见下文）。 // 这个帖子没有拥有者。 const thread = await channel.fork( answered.answer!, { ownership: { kind: "ownerless" } }, context, ); // 两个对话同时工作。 const inThread = await thread.submit( { type: "input", content: "@agent can we roll it back?" }, context, ); const inChannel = await channel.submit( { type: "input", content: "@agent who is on call today?" }, context, ); await Promise.all([inThread.wait(context), inChannel.wait(context)]); ``` 每个对话还会存储它自己的智能体配置：模型、思考等级、所选用的扩展以及它们各自的哪些工具被激活、额外的指令，以及它在执行环境中的工作目录。主智能体旁边的审查者（reviewer）可以使用更便宜的模型、只读工具，以及它自己的代码检出。 ## 扩展 我们希望智能体能做的一切都是可插拔的，而每一个被插入的部件都要参与持久化。在 Pi Durable 中，一个扩展就是一组带名字的系统提示词片段、工具、钩子（hooks）和任务。应用程序把扩展安装到一个注册表中。每个对话选择它使用哪些扩展和工具，并且只存储它们的名字。 ### 系统提示词片段 系统提示词会在每次请求之前，由该对话所用扩展的各个片段重新构建而成，因此被修改的片段会在下一次请求时生效。Pi Durable 会在对话记录中记录哪里发生了变化，就记在变化发生的位置，因此重启或分叉后，看到的正是模型当时所看到的内容。在那些支持在对话中途修改系统提示词和工具的模型上，只会发送变更内容，因此提示词缓存仍然有效。 ```js 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)), ], }); ``` ### 工具 每次工具调用都作为一个独立的持久化任务运行，其意图（intent）会在执行之前存储。崩溃之后，只有工具自己声明可以重跑时才会重跑。否则，模型会被告知该调用被打断了，以及目前保存的输出，由模型决定下一步怎么做。每个对话还可以拥有自己的一套工具，比如前面提到的 Slack 帖子，它可能只能搜索而不能部署。 ```js import { Type } from "@earendil-works/pi-ai"; import { defineTool } from "@earendil-works/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) }], }; }, }); const deploy = defineTool({ name: "deploy", description: "Deploy a version to production", parameters: Type.Object({ version: Type.String() }), // 没有 replay：被崩溃打断的部署只会告知模型，绝不会重复执行。 execute: async (args) => ({ content: [{ type: "text", text: await ci.deploy(args.version) }], }), }); registry.install(defineExtension({ name: "ops", tools: [searchIssues, deploy] })); // 这个帖子可以搜索，但不能部署。 await thread.configure({ tools: { remove: [deploy] } }, context); ``` 工具在它的调用中会获得运行时 API：它可以提交条目和文档、启动任务和对话，以及与其他对话通信。这就让构建一个子智能体只需要几行代码。工具创建一个属于它的对话，给它一个更小的模型和自己的指令，然后等待它的回答。子智能体也是一个普通的对话，所以它能在崩溃中存活、自己计算成本，并且一个 UI 可以把它展示在那次调用之下。 ```js import type { AssistantMessage } from "@earendil-works/pi-ai"; import { AssistantEntry, configure } from "@earendil-works/pi-durable"; const triage = defineTool({ name: "triage", description: "Label an incoming issue as bug, feature, or question", parameters: Type.Object({ issue: Type.String() }), // 崩溃后的重跑会找到同一个子智能体和同一个提交 replay: "safe", execute: async (args, api, context) => { const child = await api.commit(async (tx) => { const existing = ( await tx.scanConversations({ ownerTaskId: api.taskId }, 1) ).items[0]; if (existing !== undefined) return existing.id; // 由这次调用拥有，所以中止调用就会中止子智能体。 const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId }, }); // 它一开始是这个对话智能体的副本。把它改成一个没有工具的小模型。 await configure(tx, created.id, { model: { provider: "openai", modelId: "gpt-6-luna" }, tools: [], instructions: "Answer with one word: bug, feature, or question.", }); return created.id; }, context); // 让 UI 能把子智能体展示在这次调用之下 await api.details({ conversationId: child }, context); const subagent = await api.conversation(child, context); const request = { type: "input", content: args.issue, requestId: `triage:${api.taskId}`, } as const; const settled = await ( await subagent!.submit(request, context) ).wait(context); // 回答是子智能体对话记录中的一个条目。读取它并取出文本。 const entry = await api.commit( (tx) => tx.entry(AssistantEntry, settled.answer!), context, ); const message = entry?.model?.[0] as AssistantMessage; const text = message.content .flatMap((content) => (content.type === "text" ? [content.text] : [])) .join(""); return { content: [{ type: "text", text }], }; }, }); ``` 扩展还可以修改其他扩展的工具。在更靠后的扩展中出现同名工具会替换掉先前的那个，例如一个在 Python 虚拟环境中运行的 bash。装饰（wrap）会作用于最终胜出的那个工具，无论选择装饰的扩展在哪里。 ```js import { wrapTool } from "@earendil-works/pi-durable"; import { createBashTool } from "@earendil-works/pi-durable/tools"; // 为每一次 bash 调用计时，无论对话最终用的是哪个 bash。 const Timing = defineExtension({ name: "timing", wraps: [ wrapTool(createBashTool(), (bash) => ({ ...bash, execute: async (args, api, context) => { const start = Date.now(); try { return await bash.execute(args, api, context); } finally { metrics.record("bash", Date.now() - start); } }, })), ], }); ``` ### 钩子 钩子（hooks）让扩展能够介入任务之中，包括那些内置任务，比如生成模型响应、调用工具或执行压缩。钩子可以在请求发往模型之前重写它，阻止或重写一次工具调用，替换一个结果，让一次运行继续下去，或者自己写一份摘要。钩子在崩溃后可能再次运行，所以一个需要做决策的钩子会把决策存进一个备忘录（memo）：一个小值，随任务一起存储，以第一次写入为准。 ```js import { hook, ToolTask } from "@earendil-works/pi-durable"; const Approval = defineExtension({ name: "approval", hooks: [ hook(ToolTask, { beforeTool: async (call, api, context) => { if (call.name !== "deploy") return undefined; // 重启之后，钩子找到的是存储好的答案，而不是重新询问。 let approved = await api.memo( "approval:deploy", context, ); approved ??= await api.m