Pi ↗ 是一个精简、可扩展的 Agent 执行框架。Agents SDK 为一等公民支持使用 Pi 执行框架构建 Agent。本页说明如何用 Agents SDK 的 PiHarness 在 Cloudflare Durable Object 中运行 Pi Durable ↗。
Pi 运行 Agent 循环：对话记录、后续与 steer 的 inbox、模型调用、工具、重试与崩溃恢复。PiHarness 在 Durable Object 的 SQLite 数据库中为 Pi 提供存储，并在 Pi 还有工作要完成时唤醒该对象。
要从完整应用程序开始，请部署 Pi harness 示例 ↗。它通过 WebSocket 将 session streaming 到浏览器 UI，并通过 @cloudflare/computer 为 Pi 提供持久文件系统和 JavaScript 执行功能。
PiHarness 是 Cloudflare Agents SDK 提供的新“Lifecycle capability”。Pi Durable 提供 Agent harness，而 Lifecycle 负责让 Agent 在 Durable Object 中持续运行。Lifecycle 是 Agents SDK 的核心概念，确保长时间运行的工作可以在 Durable Object 中执行，并能经受重启、崩溃和网络问题。Lifecycle capability 是 Durable Object 中可复用的部分，由 Lifecycle 启动，并获得 SQLite 存储和 durable job queue。PiHarness 以两种方式使用它们：
- 存储。 Pi 将对话记录、inbox 与任务保存在对象 SQLite 数据库的自有表中。表名以
pi_开头，因此不会与你的表冲突。 - 唤醒。 Pi 的 scheduler 在内存中运行，而被逐出的对象没有内存状态。
PiHarness根据 Pi 的wake()为对象计划一个 Lifecycle job。该 job 在 Pi 工作时保持 heartbeat，在 Pi 下一个休眠任务唤醒时到期，并在 Pi 无事可做时完成。如果对象在运行过程中被逐出，job 的 alarm 会重启对象，Pi 会重新打开其存储并继续运行。
PiHarness 不选择客户端如何到达会话。它提供每个会话的事件流，由你通过 WebSocket、HTTP 或 RPC 发送。
安装包含 Pi Durable、Pi AI 和 Chord 的 agents：
npm i agents @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chordyarn add agents @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chordpnpm add agents @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chordbun add agents @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chordPi 软件包是 agents 的可选 peer dependency，版本为 ~1.4.0。Pi Durable 是实验性功能，其 API 可能在次要版本中发生变化，因此每个 agents 版本仅支持一个 Pi 次要版本。请安装软件包管理器为 agents 解析出的版本。
Agents SDK 0.30.0 及更高版本支持 Pi 1.4。
agents/models/pi-ai 入口支持 AI Gateway 与 Workers AI 模型，因此可以立即使用 Cloudflare 模型，或使用现有 Pi AI provider。本页示例通过 AI 绑定与 Pi AI 模型 provider 使用 Workers AI。在 Wrangler 配置中添加绑定与 SQLite 支持的 Durable Object：
{
"name": "pi-agent",
"main": "src/index.ts",
"upload_source_maps": true,
// Set this to today's date
"compatibility_date": "2026-11-14",
"compatibility_flags": ["nodejs_compat"],
"ai": {
"binding": "AI",
},
"durable_objects": {
"bindings": [{ "name": "Assistant", "class_name": "Assistant" }],
},
"migrations": [{ "tag": "v1", "new_sqlite_classes": ["Assistant"] }],
}name = "pi-agent"
main = "src/index.ts"
upload_source_maps = true
# Set this to today's date
compatibility_date = "2026-11-14"
compatibility_flags = [ "nodejs_compat" ]
[ai]
binding = "AI"
[[durable_objects.bindings]]
name = "Assistant"
class_name = "Assistant"
[[migrations]]
tag = "v1"
new_sqlite_classes = [ "Assistant" ]PiHarness 是 Lifecycle capability，因此在普通 Durable Object 上运行。在 factory 中打开 Pi 的 Harness，在对象上安装 Lifecycle，并将 harness 添加到其中：
import { DurableObject } from "cloudflare:workers";
import { createModels } from "@earendil-works/pi-ai/models";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { PiHarness } from "agents/harness/pi";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/pi-ai";
export class Assistant extends DurableObject {
ai = createAI({ binding: this.env.AI });
registry = createRegistry();
harness = new PiHarness({
harness: ({ storage, context, onWake }) => {
const models = createModels();
models.setProvider(this.ai.provider);
return Harness.open(
storage,
{ models, registry: this.registry, onWake },
context,
);
},
defaults: { model: this.ai("@cf/zai-org/glm-5.3-flash") },
});
lifecycle = Lifecycle.install(this).use(this.harness);
async ask(prompt) {
const { text } = await this.harness.prompt(prompt);
return text;
}
}
export default {
async fetch(request, env) {
const agent = env.Assistant.getByName("demo");
return Response.json({ text: await agent.ask("What is 47 × 19?") });
},
};import { DurableObject } from "cloudflare:workers";
import { createModels } from "@earendil-works/pi-ai/models";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { PiHarness } from "agents/harness/pi";
import { Lifecycle } from "agents/lifecycle";
import { createAI } from "agents/models/pi-ai";
export class Assistant extends DurableObject<Env> {
ai = createAI({ binding: this.env.AI });
registry = createRegistry();
harness = new PiHarness({
harness: ({ storage, context, onWake }) => {
const models = createModels();
models.setProvider(this.ai.provider);
return Harness.open(
storage,
{ models, registry: this.registry, onWake },
context,
);
},
defaults: { model: this.ai("@cf/zai-org/glm-5.3-flash") },
});
lifecycle = Lifecycle.install(this).use(this.harness);
async ask(prompt: string) {
const { text } = await this.harness.prompt(prompt);
return text;
}
}
export default {
async fetch(request: Request, env: Env) {
const agent = env.Assistant.getByName("demo");
return Response.json({ text: await agent.ask("What is 47 × 19?") });
},
};harness factory 接收 storage（Pi 在对象 SQLite 数据库上的存储）、context（用于打开 Pi 的后台上下文）和 onWake（使对象可在 Pi 任务休眠时休眠）。Harness.open() 接收的其他所有内容都由你在 factory 中构建：models、tool 和 prompt section 的 registry、settings、env 和 onReport。
factory 在对象启动过程中、任何 session 使用之前运行；当 harness 因存储错误重新打开 Pi 时也会再次运行（请参阅恢复）。请确保它可安全运行多次。如果它抛出异常，启动它的操作会失败，下一个操作会重试。每个 PiHarness 操作都会等待启动完成，因此在启动前通过 RPC 到达的调用不会自行打开 Pi。
| 选项 | 描述 |
|---|---|
harness |
必需。接收 { storage, context, onWake } 并返回 Pi 的 Harness，通常来自 Harness.open()。将 onWake 传递给 Harness.open()。 |
defaults |
新会话的起始值：model（Pi AI Model，例如 ai("@cf/...")）与 thinkingLevel。之后用 session.setModel() 更改某个会话的模型。 |
没有模型时，会话的 prompt 会一直得不到回答，直到你设置一个。
整个执行框架的设置放在传给 Harness.open() 的 Pi settings 中。例如，settings: { retry: { enabled: true, maxRetries: 2, baseDelayMs: 500 } } 会重试失败的模型请求。compaction 与 toolExecution 设置 Pi 如何压缩长对话记录以及如何运行一轮工具。
工具与系统提示词 section 都通过 extension 提供给 Pi Harness。Pi extension 是带 name、tools 与 sections 的普通对象。将其安装到打开 Pi 所用的 registry 上：
import { Type } from "@earendil-works/pi-ai";
import type { ToolRegistration } from "@earendil-works/pi-durable";
const WordCount = Type.Object({ text: Type.String() });
const wordCount: ToolRegistration<typeof WordCount> = {
name: "word_count",
description: "Count the words in a text.",
parameters: WordCount,
replay: "safe",
async execute({ text }) {
const words = text.split(/\s+/).filter(Boolean).length;
return { output: [{ type: "text", text: String(words) }] };
},
};
this.registry.install({
name: "editor",
sections: [
{ key: "preamble", render: () => "You are an editor.", tag: false },
],
tools: [wordCount],
});创建与配置 extension（含 skills 与 replay 安全性）请参阅 Extensions。
harness.prompt() 向根会话提交并等待答案。它返回最终 assistant text、status，以及作为 messages 的会话对话记录。
对于长时间运行，分别提交与等待。submit() 在 Pi 持久存储输入后返回，此时模型尚未运行。wait() 在 Pi 完成时返回结果：
const receipt = await this.harness.submit("Summarize the latest report", {
operationId: "report-summary-42",
});
// Later, possibly from another request:
const result = await this.harness.wait(receipt.operationId);
// { operationId, session, status: "done" | "unanswered", text?, reason? }operationId 使提交幂等。再次提交相同 id 会返回同一操作，且 accepted: false。使用能在你自己的重试中存活的 id，例如入站事件 id。
会话运行期间做出的提交会排队，并在当前运行之后作为自己的运行回答。设置 whenBusy: "steer"，或调用 session.steer()，可在当前工具轮次之后加入正在运行的工作。
harness.pending() 列出 Pi 尚未完成的提交及其状态：queued、running 或 parked。当 Pi 的重启限制停止任务时，提交为 parked（请参阅恢复）。
Pi 将每个提交的输入记录为 pi.user 条目，如同由用户输入。你的代码通常也会自行提交输入，例如计划任务的 prompt、webhook payload 或审批结果。要记录该输入的来源，请传入包含自定义 kind 和要随其保留的任意 data 的 entry：
await this.harness.submit("The deploy was approved. Run the migration.", {
entry: { kind: "app.approval", data: { approved: true, by: "ben" } },
});await this.harness.submit("The deploy was approved. Run the migration.", {
entry: { kind: "app.approval", data: { approved: true, by: "ben" } },
});模型看不到差异：它仍会以用户消息的形式获得文本。只有 transcript 会变化。输入使用你的 kind 和 data 存储，而不是 pi.user；messages() 和 events() 会返回二者，因此 UI 可以用不同方式显示，例如显示“由 ben 批准”，而不是将其显示为用户所说的内容。Pi 保留以 pi. 开头的 kind，因此请使用自己的前缀，例如 app.。
kind 是标签而非权限。它记录代码对输入的声明，但不会进行检查。模型会像读取用户输入一样读取它，因此不要放入任何你不允许用户发送的内容。
当模型出错且 Pi 不重试，或错误持续时间超过 Pi 的重试期限时，运行会以 model_error 失败。当 session 没有可用模型时，它会以 no_model 失败。该运行自己的提交会以相应原因为 unanswered。settings.afterFailedRun 决定排在其后的提交会怎样处理：
| 值 | 结果 |
|---|---|
keep（默认） |
pending() 将它们列为 queued，wait() 会继续等待。下一个 submit() 会先按顺序运行它们，再运行新输入。下一个 reset() 也会运行它们。 |
continue |
下一个运行从它们开始。如果原因持续存在（例如缺少模型），每个输入都会在自己的运行中失败。 |
withdraw |
wait() 返回 { status: "unanswered", reason: "run_failed" }，且 pending() 不再列出它们。 |
在 factory 中设置它：
harness: ({ storage, context, onWake }) =>
Harness.open(
storage,
{
models,
registry: this.registry,
onWake,
settings: { afterFailedRun: "withdraw" },
},
context,
);harness: ({ storage, context, onWake }) =>
Harness.open(
storage,
{
models,
registry: this.registry,
onWake,
settings: { afterFailedRun: "withdraw" },
},
context,
);使用 keep 时，保留的提交不属于 Pi 的工作，因此对象可以休眠，直到下一个 submit() 或 reset()。如果调用方不应等待如此长时间，请向 wait() 传递 signal。要撤回保留的提交，请调用 abort(operationId)。
会话是一次 Pi 对话。根会话 id 为 "1"，从一开始就存在。harness 方法接受可选的 session 选项，harness.session(id) 返回单个会话的句柄：
const session = await this.harness.sessions.create();
await session.submit("Draft a release note");
const fork = await this.harness.sessions.fork(session.id);
const all = await this.harness.sessions.list();
// [{ id: "1", busy: false }, { id: "2", busy: true }, { id: "3", parent: "2", busy: false }]| 方法 | 描述 |
|---|---|
submit(input, options) |
使用 operationId、whenBusy 或 entry durable 地提交输入。返回 receipt。 |
prompt(input, options) |
提交并等待答案与更新后的对话记录。 |
steer(input) |
以 whenBusy: "steer" 提交。 |
wait(operationId, signal) |
等待操作。中止 signal 会停止等待，不会停止工作。 |
abort(operationId) |
撤回排队的操作，或中止它加入的运行。不带 id 时中止会话中的一切。 |
pending() |
Pi 尚未确定的操作：queued、running 或 parked。 |
retry() |
再次运行因达到 Pi 重启限制而 parked 的任务。没有 parked 任务时解析为 false。 |
reset(handoff) |
开始新上下文，可选地使用 handoff 说明。 |
setModel(model) |
将此 session 的模型更改为 Pi AI Model，例如 ai("@cf/zai-org/glm-5.3-flash")。 |
messages() |
自最近一次 reset 起的活跃对话记录，为 Pi 的 EntryRecord 条目。 |
events() |
此会话的 Pi 事件流。 |
busy() |
会话是否正在运行。 |
sessions.create() 使用执行框架默认值创建新的顶层会话。sessions.fork(id) 创建能看到另一会话直至其最新条目的历史的会话。sessions.list() 包含 Pi 的 subagent 工具创建的会话。
会话 id 是 Pi 的对话 id，因此你无法选择它们。要为每个用户或聊天提供自己的 Agent，请为每次对话使用一个 Durable Object，并在其中使用根会话。
session.events() 返回 Pi 的 AgentEventStream：会话的 snapshot，然后 Pi 提交的每次变更对应一批事件：
const stream = await this.harness.session().events();
send(stream.snapshot);
stream.start(async (events) => {
for (const event of events) send(event);
});
// When the client goes away:
await stream.stop();运行中途连上的客户端会得到包含部分答案的 snapshot，然后跟随实时事件。流存在于内存中。对象重启后，打开新流并发送新的 snapshot。
messages() 与 snapshot.entries 是 Pi 的对话记录条目，不是聊天 UI 格式。将它们映射为客户端渲染的内容。
对于 PiHarness 未覆盖的内容，await harness.pi() 返回已打开的 Pi Harness。
| 发生的情况 | 结果 |
|---|---|
| 对象被驱逐、崩溃，或超过内存或 CPU 限制 | 唤醒 job 的下一次闹钟重启对象，Pi 从其上次检查点继续。 |
| 运行中途发生部署 | 与崩溃相同。运行时给进行中工作 30 秒，然后闹钟重启对象。 |
| 模型正在流式传输 | Pi 保留已存储的部分答案，并再次进行模型调用。 |
带 replay: "safe" 的工具正在运行 |
Pi 再次运行该工具。 |
| 任何其他工具正在运行 | Pi 不会再次运行它。模型得到中断结果并决定下一步。 |
当两次运行无害时将工具标记为 replay: "safe"，例如读取或整文件写入。其他工具保持 unsafe，例如会因已替换文本而失败的编辑，或会扣费的调用。
Pi 有工作时，job 的 heartbeat 闹钟每 30 秒触发一次。崩溃的对象在下一次 heartbeat 时重启。
当 Pi 的任务处于休眠状态（例如重试 backoff 期间）时，job 会在第一个任务唤醒时到期。Pi 通过 factory 传给 Harness.open() 的 onWake callback 报告时间，并且不设置内存 timer，因此对象可以休眠到该时间。
持续使对象崩溃的任务会令其无限重启。如果任务连续 settings.restartLimit 次（默认 5 次）因崩溃或逐出而中断且没有进展，Pi 会将其 park。Parked 任务不属于工作，因此 job 会完成，对象可以休眠。其操作在 pending() 中显示为 parked，而 wait() 会继续等待。如果 factory 将 onReport 传给 Harness.open()，Pi 会在那里以 task_parked 报告。处理原因后，session.retry() 会再次运行任务；session.abort() 则放弃任务。
- 审批。 Pi Durable 尚无工具调用的审批或权限步骤。
- 事件重放。 事件流总是从 snapshot 开始。没有从某处恢复流的 cursor。
- Abort 与工具。
abort()会等到会话空闲。忽略 abort signal 的工具会让会话保持忙碌直到它返回。 - 长时间模型调用。 唤醒 job 在闹钟调用内等待，闹钟调用最多运行 15 分钟。执行框架每 10 分钟交接给新闹钟，但一次流式超过 15 分钟的模型请求可能被切断。
- 后台任务。 Pi 在后台运行的工作在每次 heartbeat 时检查，因此执行框架最多晚 30 秒才注意到它完成。
- 优雅驱逐。 运行中的会话会保持进行中工作，因此 Pi 运行时对象不会 drain。
- 会话删除。 Pi Durable 尚不能删除对话。