Pi Agent 包架构

packages/agent/src/ 实现了 Pi 框架中的 Agent 运行时，用于构建工具增强型 LLM 对话代理（Tool-Augmented LLM Agent）。

整个包由五个模块组成：

架构总览

分层设计

无状态引擎 vs 有状态封装

Agent 包的核心设计思想是将引擎逻辑与状态管理分离：

类型系统（types.ts）

类型系统定义了五个核心类别：

1. 消息体系

AgentMessage 通过 TypeScript 的 declaration merging 实现扩展：

1
2
3
4
5
6
7
8

// 应用层扩展自定义消息类型
declare module "@mariozechner/agent" {
  interface CustomAgentMessages {
    artifact: ArtifactMessage;
    notification: NotificationMessage;
  }
}
// AgentMessage 自动包含 Message | ArtifactMessage | NotificationMessage

2. 事件体系（13 种）

3. 配置体系（AgentLoopConfig）

继承自 SimpleStreamOptions，增加：

• 消息转换：convertToLlm（AgentMessage[] → Message[]）、transformContext（上下文变换）
• 队列回调：getSteeringMessages、getFollowUpMessages
• 工具钩子：beforeToolCall（可阻拦执行）、afterToolCall（可覆盖结果）
• 停止判定：shouldStopAfterTurn
• 动态认证：getApiKey（用于短期 OAuth token）

4. 工具体系（AgentTool）

1
2
3
4
5
6

interface AgentTool<TParameters, TDetails> extends Tool<TParameters> {
  label: string;                          // UI 显示名称
  prepareArguments?: (args: unknown) => Static<TParameters>; // 参数兼容
  execute: (...) => Promise<AgentToolResult<TDetails>>;      // 执行
  executionMode?: "sequential" | "parallel";  // 单工具执行模式覆盖
}

5. 状态体系（AgentState）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

interface AgentState {
  systemPrompt: string;           // 系统提示词
  model: Model<any>;              // 当前模型
  thinkingLevel: ThinkingLevel;   // 推理级别
  tools: AgentTool<any>[];        // 工具列表（setter 自动 slice 拷贝）
  messages: AgentMessage[];       // 对话记录（setter 自动 slice 拷贝）
  readonly isStreaming: boolean;  // 是否正在运行
  readonly streamingMessage?: AgentMessage;  // 当前流式消息
  readonly pendingToolCalls: ReadonlySet<string>;  // 执行中的工具
  readonly errorMessage?: string; // 最近错误
}

无状态循环引擎（agent-loop.ts）

导出接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// 新 prompt 入口 → EventStream（流式封装）
function agentLoop(prompts, context, config, signal?, streamFn?):
  EventStream<AgentEvent, AgentMessage[]>

// 继续入口 → EventStream
function agentLoopContinue(context, config, signal?, streamFn?):
  EventStream<AgentEvent, AgentMessage[]>

// 底层异步版（被 agent.ts 调用）
function runAgentLoop(prompts, context, config, emit, signal?, streamFn?):
  Promise<AgentMessage[]>

function runAgentLoopContinue(context, config, emit, signal?, streamFn?):
  Promise<AgentMessage[]>

核心：双层循环（runLoop）

这是整个 Agent 包最核心的函数。

内层循环条件

内层持续的条件：有工具调用待处理 或 有 steering 消息待注入。

• hasMoreToolCalls：当前 assistant 消息有 tool call，且并非全部工具都要求终止
• pendingMessages：steering 队列中有消息

外层循环条件

内层结束后检查 follow-up 队列，有则设为 pending 继续外层。

LLM 调用（streamAssistantResponse）

函数逐项解读

以下逐一解读 agent-loop.ts 中各函数的设计细节。

agentLoop() / agentLoopContinue() — 公开入口（EventStream 封装）

1
2

function agentLoop(prompts, context, config, signal?, streamFn?): EventStream<AgentEvent, AgentMessage[]>
function agentLoopContinue(context, config, signal?, streamFn?): EventStream<AgentEvent, AgentMessage[]>

runAgentLoop() / runAgentLoopContinue() — 异步底层入口

两者最终都汇聚到 runLoop()。

runLoop() — 核心双层循环

已在前面用流程图详细展示，这里补充源码级要点：

关键设计点：

streamAssistantResponse() — LLM 调用与流式处理

处理流水线：

事件循环核心逻辑：

start   -> 推入 partialMessage 到 context.messages 末尾（占位），emit message_start
delta   -> 原地替换 context.messages 最后一条，emit message_update
done/error -> 取最终消息替换占位，emit message_end

设计要点：

• partialMessage 是引用，每次 delta 事件原地更新同一对象，避免频繁深拷贝
• context.messages 最后一条被用作"占位"：start 推入，update 替换，end 确定
• 若 start 事件从未到达（如空响应），则直接 push 最终消息

executeToolCalls() — 工具执行调度

调度逻辑：

1. 从 assistantMessage.content 中过滤出所有 type === "toolCall"
2. 检查是否有工具标记了 executionMode: "sequential"
3. config.toolExecution === "sequential" 或存在串行工具 -> executeToolCallsSequential
4. 否则 -> executeToolCallsParallel

executeToolCallsSequential() — 串行执行

对每个 tool call 逐个：
  emit tool_execution_start
  -> prepareToolCall()
    -> 若 immediate：直接记 error result
    -> 若 prepared：executePreparedToolCall() -> finalizeExecutedToolCall()
  -> emit tool_execution_end
  -> createToolResultMessage + emit message_start/end
-> 全部完毕后计算 shouldTerminateToolBatch

• 每一步都严格串行：prepare -> execute -> finalize -> 下一个

executeToolCallsParallel() — 并行执行

两阶段设计：

prepareToolCall() — 工具调用前置管道

toolCall -> 1. 查找工具定义（context.tools.find()）
           -> 不存在 -> return { kind: 'immediate', error }
           -> 找到 -> 2. prepareArguments() 参数兼容转换
                  -> 3. validateToolArguments() Schema 校验
                  -> 4. beforeToolCall 钩子
                    -> block -> return { kind: 'immediate', error }
                    -> 允许 -> return { kind: 'prepared', tool, args }

executePreparedToolCall() — 工具执行

• 调用工具自带的 execute()，传入 signal 支持取消
• 第四个参数是 onProgress 回调，工具可在执行中多次调用来发射 tool_execution_update
• 执行抛错 -> catch 后生成 error tool result（永不崩溃）

finalizeExecutedToolCall() — 工具后处理

• afterToolCall 钩子可以覆盖工具执行的任何输出字段（content / details / terminate / isError）
• 钩子本身抛错 -> catch 后覆盖为 error tool result（防止钩子崩溃波及主流程）

辅助函数一览

AM --> TC --> CL --> M --> LLM --> Stream --> Events --> Partial

### 工具执行（executeToolCalls）

```mermaid
flowchart TD
    TC[tool calls 列表] --> Mode{"配置 + 工具标记<br/>sequential 还是 parallel?"}

    Mode -->|sequential| Seq["executeToolCallsSequential()<br/>逐个执行"]
    Mode -->|parallel| Par["executeToolCallsParallel()<br/>prepare 串行 → execute 并发"]

    subgraph "每个工具的完整生命周期"
        Prep["prepareToolCall()"]
        Exec["executePreparedToolCall()"]
        Final["finalizeExecutedToolCall()"]
    end

    Prep --> Exec --> Final

    subgraph "prepareToolCall 细节"
        Find["1. 查找工具定义"]
        PrepArgs["2. prepareArguments()<br/>参数兼容"]
        Validate["3. validateToolArguments()<br/>Schema 校验"]
        Before["4. beforeToolCall 钩子<br/>可 block 执行"]
    end

    Find --> PrepArgs --> Validate --> Before

串行 vs 并行策略

终止语义（terminate）

• 仅当批次内所有工具的 result.terminate === true 时，循环才会终止
• 单一工具返回 terminate: true 不会立即终止
• 这是设计选择：避免一个工具意外终止整个 agent

有状态封装层（agent.ts）

Agent 类结构

消息队列设计

drain() 模式：

事件归约（processEvents）

类似 Redux reducer，接收 agent-loop 的 emit 事件，更新 _state：

函数逐项解读

以下逐一解读 agent.ts 中各成员的设计细节。

subscribe() — 事件订阅系统

内部调用方式（在 processEvents 中）：

1
2
3

for (const listener of this.listeners) {
  await listener(event, signal);
}

prompt() — 发起对话

1
2
3

// 三种重载
async prompt(message: AgentMessage | AgentMessage[]): Promise<void>
async prompt(input: string, images?: ImageContent[]): Promise<void>

流程：

prompt(input, images?)
  -> normalizePromptInput() 统一成 AgentMessage[]
  -> 检查 activeRun？存在 -> throw Error
  -> runPromptMessages(messages)
    -> runWithLifecycle()
      -> createContextSnapshot()
      -> createLoopConfig()
      -> runAgentLoop(..., emit -> processEvents)

normalizePromptInput() 规则：

continue() — 继续对话

逻辑分支：

continue()
  -> activeRun 存在？ -> throw Error
  -> 最后一条消息角色？
    -> 无消息 -> throw Error
    -> assistant -> steeringQueue 有消息？
      -> 有 -> drain() -> runPromptMessages(steering, skipInitialSteeringPoll=true)
      -> 无 -> followUpQueue 有消息？
        -> 有 -> drain() -> runPromptMessages(followUps)
        -> 无 -> throw Error "Cannot continue from assistant"
    -> user/toolResult -> runContinuation() -> runAgentLoopContinue()

steer() / followUp() — 消息队列

PendingMessageQueue 的 drain() 模式通过 QueueMode 控制：

abort() / waitForIdle() / reset() — 生命周期控制

runWithLifecycle() — 并发控制核心

runWithLifecycle(executor)
  -> activeRun 存在？ -> throw Error
  -> 创建 AbortController + Promise
  -> activeRun = { promise, resolve, abortController }
  -> isStreaming = true
  -> try { executor(signal) }
    -> 成功 -> finally { finishRun() }
    -> 失败 -> handleRunFailure(error, aborted) -> finally { finishRun() }

handleRunFailure() — 失败处理

将错误编码为一条 stopReason: "error" / "aborted" 的 AgentMessage 并推入对话历史：

1
2
3
4
5
6
7

const failureMessage = {
  role: "assistant", content: [{ type: "text", text: "" }],
  stopReason: aborted ? "aborted" : "error",
  errorMessage: error instanceof Error ? error.message : String(error),
};
this._state.messages.push(failureMessage);
await this.processEvents({ type: "agent_end", messages: [failureMessage] });

所有失败都被编码为正常的 AgentMessage 推入对话历史，应用层可通过 subscribe() 监听到 agent_end 事件。

finishRun() — 运行终结

1
2
3
4
5
6
7

private finishRun(): void {
  this._state.isStreaming = false;
  this._state.streamingMessage = undefined;
  this._state.pendingToolCalls = new Set<string>();
  this.activeRun?.resolve();
  this.activeRun = undefined;
}

清理 isStreaming、streamingMessage、pendingToolCalls，兑现 waitForIdle() 的 Promise，清除 activeRun 以允许下一次调用。

processEvents() — 事件归约（状态机）

类似 Redux reducer，接收 AgentEvent 并更新 _state：

然后按订阅顺序依次 await 所有 listener。

不可变性保证： pendingToolCalls 每次更新都新建 Set，外部通过 AgentState.pendingToolCalls 只读到 ReadonlySet，防止篡改。

createContextSnapshot() / createLoopConfig() — 适配层

然后按订阅顺序依次 await 所有 listener。

并发控制（ActiveRun）

完整调用链

HTTP 代理（proxy.ts）

streamProxy 是一个替代 streamSimple 的流式函数，用于需要经过代理服务器的场景。

关键设计：

• 服务器端去掉 partial 字段以减少带宽
• 客户端根据事件序列重建完整的 AssistantMessage
• 支持 text / thinking / toolcall 所有三种内容类型
• 异常时生成 stopReason: "error" / "aborted" 的消息，不抛异常

容错设计

Agent 包的设计哲学是永不崩溃——所有可能的失败都被捕获并编码为正常事件流：

所有错误最终都以 AgentEvent 的形式通过 emit 输出，应用层可以通过 subscribe() 统一处理。

使用示例

基础用法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import { Agent } from "@mariozechner/agent";

const agent = new Agent({
  initialState: {
    systemPrompt: "你是一个有用的助手。",
    model: myModel,
  },
});

// 订阅事件（更新 UI）
agent.subscribe((event) => {
  switch (event.type) {
    case "message_update":
      updateUI(event.message);
      break;
    case "turn_end":
      updateToolPanel(event.toolResults);
      break;
  }
});

// 发起对话
await agent.prompt("Hello!");

// 注入中间消息（不打断当前 LLM 调用）
agent.steer({ role: "user", content: [{ type: "text", text: "等一下..." }] });

自定义消息类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 扩展自定义消息
declare module "@mariozechner/agent" {
  interface CustomAgentMessages {
    artifact: { role: "artifact"; content: ArtifactContent[]; timestamp: number };
  }
}

// 自定义转换函数
const agent = new Agent({
  convertToLlm: (messages) =>
    messages.flatMap((m) => {
      if (m.role === "artifact") return []; // 过滤掉 artifact 消息
      return [m];
    }),
});

使用代理

1
2
3
4
5
6
7
8

const agent = new Agent({
  streamFn: (model, context, options) =>
    streamProxy(model, context, {
      ...options,
      authToken: await getAuthToken(),
      proxyUrl: "https://genai.example.com",
    }),
});

设计决策

为什么分离引擎层和管理层？

1. 可测试性：agent-loop.ts 是纯函数，无需 mock 状态即可单元测试
2. 复用性：无状态引擎可以被不同的管理层复用
3. 关注点分离：引擎只管循环逻辑，管理只管状态和并发

为什么用双层循环（steering + follow-up）？

• steering：用户想在 agent 工作过程中"插话"，需要立即处理
• follow-up：用户想等 agent 完成后再补充，不打断当前工作流
• 双层结构让这两种场景各得其所

为什么工具终止需要全部一致？

• 防止一个工具意外终止整个 agent
• 只有所有工具都明确要求终止时才生效
• 调用方可以通过 afterToolCall 钩子调整 terminate 标志

为什么用 Set + 替换而非直接 mutable？

• AgentState.pendingToolCalls 对外是 ReadonlySet，保证状态不可变性
• 内部每次更新都新建 Set，防止外部持有引用后篡改