【pi-mono 源码解析 2/4】展平的代价：LLM Agent 框架里被忽视的工具调用原子性问题

导读#

打开任意一个 LLM Agent 框架的上下文存储格式，你会发现一个几乎无处不在的设计选择：把所有消息展平成一个线性序列。

user 消息、assistant 消息、工具调用（toolUse）、工具结果（toolResult）——全部以同等身份躺在同一个 JSONL 文件或数组里，靠 role 字段区分。这个格式直觉上合理：LLM API 本身就接受这样的消息数组，展平存储减少了阻抗失配。

但有一个问题被这个展平格式默默掩盖了：工具调用和它的结果在语义上是原子的——一次 bash 调用和对应的执行结果，在逻辑上属于同一个操作单元。把它们拆成两条独立记录存入平面序列后，任何需要理解"这条记录是否是工具结果"的代码，都必须在运行时重新推断这个关系。

本文以 pi-mono 1 为解剖对象，用实测数据量化这个问题的规模，再从 Kafka 和 PostgreSQL 的历史中找到它的正名，最后提出一个不破坏现有格式的改进方案。

现象：数字不会说谎#

分析 pi-mono 项目中的两个真实会话文件，先看原始数字。

large-session.jsonl（914 条记录）：

before-compaction.jsonl（987 条记录）：

before-compaction.jsonl 更有意思：它的文件名和内部元数据（/Users/badlogic/workspaces/pi-mono、时间戳 2025-12-09、provider: anthropic、model: claude-opus-4-5）证明这是框架作者在自己机器上的真实会话，带有 branchedFrom 字段，说明分支特性也在生产中实际使用。框架作者在用自己的工具触碰自己工具的边界。

45% 的记录是工具结果。这个数字在每次上下文压缩时都要被重新发现。

根因：热路径上的重复验证#

定位到 compaction.ts:260：

function findValidCutPoints(entries: SessionTreeEntry[]): number[] {
  const cutPoints: number[] = [];
  for (let i = 0; i < entries.length; i++) {
    const entry = entries[i];
    if (entry.type !== "message") continue;
    // toolResult messages are not valid cut points
    if (entry.message.role === "user" && isToolResultMessage(entry.message)) {
      continue;
    }
    cutPoints.push(i);
  }
  return cutPoints;
}

这个函数在每次执行上下文压缩时被调用。它做的事情是：遍历所有消息记录，筛选出"不是 toolResult 的那些"作为合法切割点。

信息流是这样的：

写入时:    [toolUse 落盘] → [toolResult 落盘]   ← 原子语义在此建立
              ↓
压缩时:    [flatEntries.forEach] → [isToolResult? skip] → [cutPoints]
              ↓
          信息获取 → 丢弃 → 重建

每次压缩，都在重新发现同一个静态事实：toolResult 永远不是有效切点。

这不是算法复杂度问题——findValidCutPoints()（compaction.ts:260）本身是 O(n) 的。问题在于它违反了 Parse, Don't Validate 原则 2：在数据加载时就应该建立的结构，被推迟到热路径上反复重建。

根本原因在类型层：

// packages/agent/src/harness/types.ts
export type MessageEntry = {
  type: "message";
  message: AgentMessage;  // AgentMessage 的 role 可以是 "user" 或 "assistant"
  // ...
};

MessageEntry 是一个统一的包装类型。toolUse（role: "assistant"）和 toolResult（role: "user"）都被装入 MessageEntry，类型系统无法在编译期区分它们。因此运行时的每次 isToolResultMessage() 调用都是必要的——格式设计迫使验证必须发生在运行时。

外部参照：七年前的工业解法#

这个问题不是新问题。

Kafka KIP-98：生产者幂等性与原子批次#

Kafka 在 2017 年的 KIP-98 中引入了生产者幂等性和事务支持 3。核心设计是 producer_id 字段：同一个 producer_id 下的一批消息构成一个原子单元，Broker 侧可以按 producer_id 聚合，判断一批消息是否完整到达，而不是对每条消息逐一验证。

工具调用和 Kafka 批次的同构性：

PostgreSQL WAL：事务边界与 xid 组#

PostgreSQL 的 WAL 日志同样是展平的——每行 WAL 记录（heap insert、index insert、commit）都单独落盘。但 WAL 解析器（pg_waldump、逻辑复制解码器）在读取时会按 xid（事务 ID）重新组合，把一个事务的所有操作聚合为原子单元再交给消费者处理。

展平存储 + 读取时重组——这正是我们需要的模式。

命名这个模式：Aggregate Reconstitution#

DDD（领域驱动设计）语境中，"聚合重组"（Aggregate Reconstitution）描述的是：把持久化层的展平记录在内存中还原为完整的领域对象。

• PostgreSQL WAL 解析器：按 xid 聚合 → 返回完整事务
• Kafka 消费者：按 producer_id 聚合 → 返回完整批次
• Agent 框架（待实现）：按 toolCallId 聚合 → 返回完整工具调用对

数据库和消息队列在 2017-2018 年就解决了这个问题。Agent 框架在 2025 年仍在热路径上重复验证。

横向对比#

当前主流 Agent 框架的处理方式：

四个框架，四个运行时验证，全部在热路径上重复同一个问题。差异只在语法层面（Python isinstance vs TypeScript isToolResultMessage），模式完全一致。

改进方案#

ToolCallGroupEntry：在加载时建立结构#

核心思路：JSONL 格式不变，只改变读取后的内存表示。

引入一个新的内存类型 ToolCallGroupEntry，在 JSONL 加载时（parse-time）按 toolCallId 聚合：

// 只存在于内存，不写入 JSONL
type ToolCallGroupEntry =
  | { kind: "complete";         toolCallId: string; use: ToolUseEntry; result: ToolResultEntry }
  | { kind: "missing_results";  toolCallId: string; use: ToolUseEntry }
  | { kind: "orphaned_result";  toolCallId: string; result: ToolResultEntry }
  | { kind: "unknown";          entries: MessageEntry[] };

4 个 variant 对应 4 种现实情况：

• complete：toolUse 和 toolResult 都存在，通过 toolCallId 精确匹配
• missing_results：toolUse 存在但结果尚未到达（工具仍在执行中）
• orphaned_result：结果存在但找不到对应的 toolUse（会话被截断）
• unknown：无法判断归属关系的记录（容错兜底）

聚合依据来自 agent-loop.ts:736 的显式 ID：

// packages/agent/src/agent-loop.ts:736
const toolResultMessage = createToolResultMessage({
  toolCallId: finalized.toolCall.id,  // ← 显式 ID，非位置匹配
  // ...
});

这意味着聚合是 ID-join，不是位置匹配——对乱序到达的并行工具结果同样安全。

Tolerant Reader：容错优先#

orphaned_result 和 unknown variant 确保解析器不会在格式异常时崩溃。这是 Tolerant Reader 模式 4 的体现：只处理你理解的字段，对未知字段宽容，而不是防御性地抛出异常。

改造后的 findValidCutPoints#

function findValidCutPoints(entries: SessionEntry[]): number[] {
  const cutPoints: number[] = [];
  for (let i = 0; i < entries.length; i++) {
    const entry = entries[i];
    // ToolCallGroupEntry 在加载时已标记，O(1) tag check
    if (entry.type === "toolCallGroup") continue;
    if (entry.type === "message") cutPoints.push(i);
  }
  return cutPoints;
}

isToolResultMessage() 调用消失了。类型系统在编译期保证了 ToolCallGroupEntry 不是有效切点，运行时验证变成了 O(1) 的 discriminant check。

影响范围#

由于 ToolCallGroupEntry 只存在于内存：

DAG 分支场景（branchedFrom 字段）同样安全：prepareCompaction() 在 compaction.ts:568 已经把 DAG 路径线性化，ToolCallGroupEntry 的聚合在线性化序列上执行，不受原始分支结构影响。

开放问题#

这个改进解决了热路径上的重复验证，但它引出了几个更深的问题：

Q1：并行工具执行的聚合时机

pi-mono 支持并行工具调用（ToolExecutionMode: "parallel"）。在 Promise.all 尚未 settle 时，部分工具已完成，部分还在执行——此时 ToolCallGroupEntry 的状态是 missing_results。聚合器应该在什么时间点触发从 missing_results 到 complete 的转换？是每次工具结果到达时增量更新，还是批次结束后统一重组？

Q2：会话分支与聚合一致性

branchedFrom 字段标记了一条会话从另一条分叉的点。如果分叉点恰好落在一个工具调用的 toolUse 和 toolResult 之间——toolUse 在主干，toolResult 在分支——ToolCallGroupEntry 的聚合应该产生 orphaned_result 还是跨分支查找？

Q3：类型系统的语义边界

引入 ToolCallGroupEntry 后，SessionEntry 的 union type 变宽了。在 exhaustive match 场景（TypeScript 的 switch/case + default: never 检查），消费方需要显式处理新 variant。这是一个"零感知升级"还是实际上需要所有消费方更新？Tolerant Reader 原则和严格类型安全之间，边界在哪里？