一次问答的背后

图 1-1 · 一次问答：问题 → Harness → 回答，背后是六步、五次持久提交

这是Pi《源码精读》的第一集，一次问答的背后。这个系列的目标是整体掌握，不抠细节。看完之后，你应该能说出它分成哪几块。一条消息怎么变成回答？你还能说出背后有哪几条关键设计？遇到具体问题，也知道该去翻哪个文件。所以这里不讲编程语法。

屏幕上的代码只放几句，作为路标。每一集都有四样东西。第一，跑一个包里自带的例子，看它打印出什么。第二，看几句关键代码，知道事情发生在哪里。第三，讲一条设计上的理由。第四，在全景地图上标出这次经过了哪些部分。整个系列一共九集，每集回答一个问题。

第一，问一个问题，背后存下了什么。第二，agent调用工具时，背后发生了什么。第三，agent忙的时候，你又说话会怎样。第四，崩溃了怎么办。第五，这些东西存在哪里。第六，模型每次看到的是什么。第七，对话太长了怎么办。第八，新功能怎么接进来。第九，多个agent怎么协作。

这一集，先回答第一个问题。我们跑包里自带的第 13 个示例，13-chat.ts，问法国的首都是哪里。为了不联网，例子里用的是一个假模型，预先排好了一句回答，内容是 Paris.。代码只有关键的几步，打开Harness，拿到一个对话，提交问题，然后等它回答。Harness就是总控，所有事情都从它开始。

源码对照 · packages/durable/test/examples/13-chat.ts:27

const storage = await openNodeJsonlStorage(directory, context);
const harness = await Harness.open(storage, { models, registry }, context);
const root = await harness.root(context, {
  agent: { model: { provider: "faux", modelId: "faux-1" } },
});
// …
const capital = await root.submit({ type: "input", content: "Capital of France?" }, context);
const answered = await capital.wait(context);

storage、models、registry 是前面准备好的；Harness.open() 把它们接起来。root() 拿到根对话，submit() 提交问题并返回一个提交项，wait() 等到这个提交项结算。正常结算时状态是 done、类型是 input，answered.answer 就是记录本里那条回答的 id；示例再按这个 id 把回答取出来打印。

图 1-2 · 代码里的五个变量：从 storage 到 answered

跑起来，回答是 Paris.；记录本里的顺序是 pi.user、pi.system、pi.assistant；数据落在当前目录的 pi-durable-chat-data。首次运行才会得到下面这五次持久提交；同一个目录再次运行会复用旧对话，条目会累加。要复现干净结果，换一个空目录运行。例子本身不打印每一次存档；为了看清背后，我在 Harness 上挂了一个提交监听，每提交一次打印一段。一次提交对应一个 seq；有些进度提交是 deferred，通常会并入下一次提交，不单独占 seq。结果一次问答，六个步骤，总共落盘五次。我们一条一条看。

第一次存档，发生在拿对话的时候。存储里出现了一个对话，你可以把它想成一本记录本。同一次提交里，还出现了五个文档。pi.live 记这条对话忙不忙、正在生成什么，失败时还会留下 runFailure，判断空闲要看 run 是否存在，不看它是不是空对象；pi.inbox 放还没轮到的输入；pi.usage 记 token 账；pi.provider 存 pi 发给模型提供方、用来标识这段对话的会话 ID；pi.agent 存这个对话的模型、工具这些设置。加上对话本身，这一次存档一共六样。

后面的问题和回答都会一条一条记进这本记录本。文档是挂在对话旁边的一小块状态。文档的内容可以改，但改动也要随着存档一起存下来。

源码对照 · packages/durable/src/harness/harness.ts:452

  protected override async conversationCreated(tx: Transaction, record: ConversationRecord): Promise<void> {
    await tx.doc(LiveDoc, record.id);
    await tx.doc(InboxDoc, record.id);
    await tx.doc(UsageDoc, record.id);
    await tx.doc(ProviderDoc, record.id);
    await createAgent(tx, record);
    await this.#options.conversationCreated?.(tx, record);
  }

五个文档的创建调用都在这段里，顺序也对得上：pi.live、pi.inbox、pi.usage、pi.provider、pi.agent。它们和对话在同一个 commit 里，所以要么一起出现，要么都不出现。

图 1-3 · 一次 commit：对话和五个文档一起出现，全有或全无

第二次存档，发生在提交问题的时候。这一次同时存下了四样东西。第一样是一条用户条目。你问的那句话，就这样记进了记录本。第二样是一个提交项。提交问题后拿回来的就是它，像一张回执。它的状态是 placed，也就是已放进对话，之后我们就靠它等这个问题被回答。

第三样是一个任务，叫生成任务。状态是 pending，等待执行。真正去问模型的就是它。第四样是 pi.live 里的一笔：run 指向这个生成任务，这条对话现在正忙，由它负责。

图 1-4 · 六步、五次持久提交：每一步写了什么，以及记录本、五个文档、提交项与任务

源码对照 · packages/durable/src/harness/submissions.ts:312

  const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now);
  const input = { ...fields, type: "input", status: "placed", entry: entry.id } as const;
  const { id } = await tx.createSubmission(input);
  await startRun(tx, conversationId, live, { inputs: [id], first: entry.id });
  return { id, status: "placed" };

用户条目、提交项、生成任务和 pi.live 的 run，就是这四样。它们也在同一个 commit 里：问题被接住的同时，后台已经有一份任务在等它。

第三次存档，是后台的调度器接走了这个任务，把它的状态从 pending 改成 running。这时 checkpoint（任务自己的进度）还是 prepare，它还没开始整理提示词。

第四次存档，prepare 这一步跑完。它把系统提示词准备好，记成一条 pi.system 条目；同时把任务的 checkpoint 从 prepare 推到 request。这一次写了两样：系统条目和任务进度。请记住这个进度，它也在存储里，后面会用到。

第五步，任务开始执行 request 这一步。它先在 pi.live 里记下 generation: { attempt: 1 }，界面看到这笔，就知道它正在生成。这一笔是进度展示，默认最多一秒后落盘，通常并入最后一次回答，不单独占一个 seq。然后才去请求模型。

第五次存档（最后一次），模型的回答回来了。这一次同时存下了五样东西。回答条目记进了记录本。任务进入 terminal，结果是 completed。提交项变成 done，answer 指向回答条目 #11。pi.live 里的 run 和 generation 被清掉，对话回到空闲。pi.usage 记下这次用了多少 token。直到这一刻，我们的等待才结束。提交项告诉我们，问题已经回答了，答案是记录本里的哪一条。

源码对照 · packages/durable/src/harness/generation.ts:1148

async function appendAssistant(
  tx: Tx,
  conversationId: ConversationId,
  message: AssistantMessage,
): Promise<TypedEntry<never>> {
  await recordUsage(tx, conversationId, "models", `${message.provider}/${message.model}`, message.usage);
  return tx.appendEntry(AssistantEntry, conversationId, { model: [message] });
}

回答条目和用量结算在同一个函数里完成；这个函数就在第五次存档的 commit 里被调用。同一个 commit 里的 endRun 再把提交项结算成 done，并清掉 pi.live 里的运行状态。

图 1-5 · 状态不是一步到位：placed / pending / running / request / terminal

把这次运行的五次持久提交按 seq 排开，是这样。一次存档对应一个 seq，里面的“写入”可能有好几笔：

存储实况 · 五次提交（SQLite 后端，observeCommits 记录）

seq	这次提交写下的东西	写入数
1	对话 #1；文档 pi.live #2、pi.inbox #3、pi.usage #4、pi.provider #5、pi.agent #6	6
2	条目 pi.user #7；提交项 #8 placed；任务 #9 pi.generation pending；pi.live 的 run 指向任务 #9（turns: 0）	4
3	任务 #9 改为 running，checkpoint 仍为 prepare	1
4	条目 pi.system #10；任务 #9 的 checkpoint 改为 request	2
5	条目 pi.assistant #11；提交项 #8 done；任务 #9 terminal；pi.live 清空；pi.usage 记下 faux/faux-1（input 0 / output 2 / totalTokens 16）	5

这些 ID 是本次运行的真实值；编号属于不同的记录类型：对话、文档、条目、提交项、任务。第 1、3 次没有新条目，所以只看记录本会漏掉它们；第 1 次写对话和五个文档，第 3 次只改任务状态。第 5 次才写回答条目、提交项结算、任务终态、pi.live 和 pi.usage。

条目表可以直接用 SQL 复核：

存储实况 · 条目落在哪次提交

SELECT id, json_extract(record, '$.kind') AS kind, commit_seq FROM entries ORDER BY id;

id	kind	commit_seq
7	pi.user	2
10	pi.system	4
11	pi.assistant	5

图 1-6 · 五次持久提交 ≠ 三条条目：记录本之外还有对话、文档、任务和 pi.live

tasks 表里任务 #9 是 terminal、结果 completed；submissions 表里提交项 #8 是 done，answer=11。

这五次持久提交分别发生在哪里？屏幕上是四个路标。拿对话、建好五个文档，在 harness.ts 里，就是上面那段 conversationCreated。提交问题这一步，在 submissions.ts 里，条目、提交项、任务和 pi.live 的 run 在同一次提交里写下。调度器接手任务，在 scheduler.ts 里，把 pending 改成 running。整理提示词、请求模型、存下回答，都在 generation.ts 里。

图 1-7 · 四个路标：harness.ts、submissions.ts、scheduler.ts、generation.ts

现在不用记住这些文件名，后面几集会再讲到。五次持久提交看下来，规律是：会改变最终结果的状态，先落盘再对外生效；流式正文和运行中的工具输出是例外，它们先展示，最多约一秒后才随下一次提交落盘。wait() 也不是靠任务在内存里打个招呼，它等的是提交里出现一个已经结算的提交项。程序外面能看到的关键结果，比如提交项的状态和回答，都已经存好了。这就是 Durable 最核心的一条设计：关键结果先存档，再展示。

图 1-8 · 关键结果提交过，外面才看得见；流式进度先展示、随后落盘；wait() 等的是已结算的提交项

它有什么用？先说一个前提，存储要能扛住进程死掉，就得放在进程外面，比如磁盘上的 SQLite 文件，或者一个 JSONL 目录。这个例子用的是当前目录下的 pi-durable-chat-data，就是磁盘上的 JSONL；为了下面好查，我也用 SQLite 跑了同一场景。有了这个前提，最终回答结算时，回答条目已经落盘；流式过程中的文字可能还没落盘，崩溃会丢最多约一秒。进程被杀，已经落盘的部分还在。再回到开场的问题，要是进程刚做完第四次存档就死了，会怎么样呢？

图 1-9 · 如果进程在第 4 次存档之后死掉：前 4 次已落盘，最后一次回答提交还没发生

我们再打开存储，里面留着最要紧的三样东西，分别是你问的那句话、状态是 placed 的提交项，以及一个 checkpoint 停在 request 的生成任务。任务记录在崩溃前还标着 running；重开时，调度器把它改回 pending，但 checkpoint 不动。request 阶段发现上次请求可能已经在途，会按重试策略把这次尝试记为失败：有重试额度就退避后重来，没有就以 runFailure 的 reason: "interrupted" 结束，不会直接重发同一个请求。这是怎么做到的？删除文件这种做到一半的工具，要不要再执行一次？

图 1-10 · 重开存储：问题、提交项和 checkpoint=request 的任务还在；是否重发由重试策略决定

这些留到第四集再讲。最后，我们看看全景地图。从代码看，Durable 可以分成三层。最上面是 Harness，负责把 agent 跑起来；中间是 Session，专门负责存档；最底下是 Storage，真正把东西写下去。

对话，条目，文档，任务，提交项，都由 Session 存取。一次 commit 里的东西，要么一起落下去，要么都不落。Harness 管怎么让 Agent 跑起来：接收、提交、调度任务、生成回答、调用工具，还有接入各种扩展，都在这一层。代码里 Harness 直接继承 Session，所以它每走一步，最后都落到同一个 commit 上。

最底下是存储，可以放在内存里，也可以是 SQLite 文件、JSONL 目录，或者别的地方。刚才那次问答，从上层的提交开始，经过调度器交给生成任务。

图 1-11 · 三层：Harness 管运行，Session 管存档，Storage 落盘

每走一步，都通过 Session，存进最底下的存储。这一次问答，背后是六步、五次持久提交。除了总控 Harness，我们还认识了五样东西。对话是记录本，条目是记录本里的一条。提交项是提交后的回执。任务是在后台干活的那份工作。文档是挂在对话旁边的小块状态。

图 1-12 · 五样东西：对话、条目、提交项、任务、文档；关键结果先存档，再展示

贯穿这一切的设计，是让关键结果先存档，再展示，流式进度先展示、随后落盘。下一集，我们沿着这条路再走一遍。看看Agent调用工具时，背后发生了什么。你让Agent改一个文件，他回了一句，改好了。看起来只是一问一答，其实在你看不到的地方。他先读了文件，又改了文件，还跑了两条命令，最后才回答你。

这几步背后，Durable都做了什么？