观察一个正在工作的 Agent:View、Watch 与多人协作

系列第 7 篇,共 8 篇。

把 Agent 放进产品里,最难的往往不是让它会做事,而是让用户看见它在做什么。

传统做法是让前端直接消费模型流:token 一到就渲染,工具一跑就本地维护一份状态。一旦断线、刷新、换设备,前端就得靠自己拼出一个「可能差不多」的视图。多人同时观看时,每个人看到的还可能不一样。

Pi Durable 把这个问题倒过来:服务端先提交,客户端再观察。 所有 UI 需要的真相都在存储里,运行中的内容则通过 pi.live 按固定间隔提交。于是刷新、晚加入、第二个客户端,都只是「从当前视图开始」。

一、观察模型

flowchart LR
    W["工作:generation / tool / compaction"] -->|"commit"| S[("Storage")]
    S -->|"当前视图"| V["viewState()"]
    S -->|"操作流 ops"| K["watch()"]
    S -->|"事件流"| E["watchEvents()"]
    V --> U1["本地 UI"]
    K --> U2["远程客户端"]
    E --> U3["事件驱动宿主"]

三种接口面向不同消费者:

接口输出特点适用场景
viewState()完整视图的最新值简单、幂等本地渲染、首屏
watch()每次提交的精确 op增量小、可走 socket远程客户端、协作
watchEvents()编码 Agent 风格事件语义友好、字节更多已有事件总线

viewState() 的 value 包含:

  • entries:活跃 transcript;
  • docs["pi.live"]:运行中的 generation(流式 partial、retry、deferred)与工具调用(output、details);
  • docs["pi.inbox"]、docs["pi.usage"]、docs["pi.agent"]、docs["pi.provider"];
  • 对话元信息与所有权。
const view = await root.viewState(context);
view.subscribe((value) => render(value));
// 之后:view.dispose();

watch() 则同时给出当前值与之后的 ops:

const watch = await root.watch(context);
render(watch.value);
watch.start(async (value, ops) => {
    await send(ops); // 发给远程客户端,由它应用
});
// 之后:await watch.stop();

两者的区别可以用一句话概括:viewState() 是「状态」接口,watch() 是「操作」接口。 后者才适合走网络,因为 op 是幂等的状态变换,丢帧不影响正确性。

二、晚加入与断线重连

晚加入的客户端不需要回放历史。它先拿到当前完整视图,之后只接收增量:

sequenceDiagram
    autonumber
    participant C as 新客户端
    participant H as Harness
    participant S as Storage

    C->>H: viewState() / watch()
    H->>S: 读取当前已提交状态
    S-->>H: entries + docs + inbox + usage
    H-->>C: 完整视图(快照)
    loop 之后每次提交
        H-->>C: 增量 ops
    end
    Note over C: 断线重连 = 重新取一次快照

这条设计带来三个直接结果:

  1. 刷新页面不丢状态。 首屏就是当前真相,不需要「等待历史回放」。
  2. 重连不需要补课。 直接重新取快照,服务端不保存「你看到哪里了」。
  3. 多设备天然一致。 所有客户端观察的是同一份已提交状态。

2.1 背压:丢弃中间态,只保留最终态

慢速消费者不会无限积压。一个 watch 最多保留 100 个未投递的帧,超过之后,待投递的帧会被一个承载最新完整视图的帧替换。

flowchart LR
    A["提交 1"] --> Q["待投递队列"]
    B["提交 2"] --> Q
    C["... 提交 100"] --> Q
    D["提交 101"] -->|"队列满"| E["合并为一个最新快照帧"]
    Q --> F["慢消费者"]
    E --> F

这不是「丢数据」,而是「丢动画」。op 是幂等的状态变换,最终值仍然正确,只是中间过程被压缩了。

2.2 进度提交的粒度

partial 回答与工具输出默认最多每 100 ms 提交一次,所以崩溃最多丢掉这个窗口。这个值可以按存储位置调整:

settings: {
    progress: {
        partialIntervalMs: 100,
        outputIntervalMs: 100,
        // 远程存储可调大,例如 500
    },
}

更频繁的提交意味着更平滑的 UI,也意味着更多的存储写入。这是一个明确的权衡,而不是隐藏的实现细节。

三、多人协作

因为所有 UI 需要的状态都已提交,任意数量的客户端都能附着到同一批对话。任何客户端都能操控正在运行的对话,或排队一条后续消息。

sequenceDiagram
    autonumber
    participant A as 客户端 A
    participant H as Harness
    participant B as 客户端 B

    A->>H: submit(input)
    H->>H: Run 开始
    B->>H: viewState()
    H-->>B: 当前视图(含运行中的工具)
    B->>H: submit({ whenBusy: "steer" })
    H->>H: 在当前工具轮后插入
    H-->>A: ops(新条目、工具输出)
    H-->>B: ops

第二个客户端可以在 Agent 工作时加入,并立刻看到:

  • 已经提交的 transcript;
  • 正在流式输出的回答;
  • 运行中的工具及其输出;
  • 排队中的消息;
  • agent 配置与用量。

它也可以立刻操控:whenBusy: "steer" 的消息会在当前工具调用之后加入正在运行的工作,而不是等整个 Run 结束。

四、事件流:另一种观察方式

watchEvents() 把提交转换成编码 Agent 熟悉的事件:

const stream = await watchEvents(harness, root.id, context);
initialize(stream.snapshot);
stream.start(async (events) => {
    for (const event of events) console.log(JSON.stringify(event));
});

语义:

  • 事件派生自提交,每次提交一批;
  • 消息与工具的更新携带 delta:文本与思考的追加、工具参数文本的追加、输出的裁剪与追加;
  • 消费者落后超过 100 批时,会直接收到一个全新的 snapshot 事件。

事件流适合已经有一套事件总线的宿主,例如需要把 Agent 状态映射到现有的前端框架或日志系统。代价是线上字节更多。

五、存储后端:观察能力的物理边界

观察接口的语义与存储后端无关,但性能和持久性取决于后端:

后端导入持久性适用场景
Memory包根的 MemoryStorage进程退出即失测试、演示
SQLitestorage/sqlite/node 的 openNodeSqliteStorage(file)进程崩溃可恢复;断电可能丢最新一笔单机长跑、桌面应用
JSONLstorage/jsonl/node 的 openNodeJsonlStorage(dir)追加写;fsync: true 每次提交刷盘调试、审计、小规模
Cloudflare DOstorage/sqlite/cloudflare提交即 Durable Object 的存储事务边缘部署、多客户端

几个工程约束:

  • 同一时刻一个存储只能被一个进程打开,没有跨进程锁;其他客户端通过 watch 附着。
  • JSONL 把全部历史放在内存里,内存随历史增长;长跑智能体请用 SQLite。
  • SQLite 存储支持 { prefix: "pi_" } 给表和索引加命名空间,便于与宿主表共享数据库;改前缀不会迁移已有数据。
  • 自定义存储可以用 @earendil-works/pi-durable/testing 的一致性套件自检。

六、存储故障时的观察行为

存储错误不是普通错误,它会终止整个 Harness:

flowchart TD
    E["Storage 错误"] --> A["当前调用 reject"]
    A --> B["所有 pending wait 以 SessionFailed reject"]
    B --> C["watches / 事件流以 session_failed 结束"]
    C --> D["运行中的任务收到 abort"]
    D --> E2["onReport 收到一次错误"]
    E2 --> F["Harness 自行关闭"]
    F --> G["宿主重开 Harness"]

宿主可以监听关闭事件并决定是否重启:

harness.closed.then((end) => {
    if (end.reason === "failed") reopen(end.error);
});

这里没有「自动重试」:存储不可信时继续运行,只会让更多客户端看到不一致的状态。恢复的责任在宿主,框架只保证存储本身是一致的。

七、把观察接进产品的建议

  1. 首屏用 viewState(),之后用 watch()。 不要用事件流从零重建状态。
  2. 把 pi.live 当作「运行中」的唯一来源。 不要在客户端另外维护一份工具输出。
  3. 断线重连时重新取快照。 不要试图补拉错过的 ops。
  4. 多人协作时只提交意图,不提交 UI 状态。 whenBusy: "steer" 比客户端自己判断「现在能不能打断」可靠。
  5. 远程存储调大 progress 间隔。 100 ms 是本地默认值,不是普适值。

小结

  • 服务端先提交,客户端再观察;观察接口只暴露已提交状态;
  • viewState() 给状态,watch() 给操作,watchEvents() 给事件;
  • 晚加入与重连从当前视图开始,不回放历史;
  • 慢消费者最多积压 100 帧,之后合并为最新快照;
  • 多人协作共享同一份已提交状态,steer 是安全的插入方式;
  • 存储错误会终止 Harness,恢复责任在宿主。

下一篇是落地手册:把前面七篇的机制收成一张选型表、一份配置速查和一份上线检查清单。