🗺️ 全景：三层架构#

先记住这张图。整篇文章就是沿着它的箭头从上往下走一遍。

三层各管一件事：应用层管”做什么”（会话、工具、配置），运行时层管”循环怎么转”，模型层管”怎么跟 LLM 说话”。

两层之间只靠两个接口咬合，记住这两个词：

AgentTool 是应用层提供的，StreamFn 是模型层提供的，循环本身谁都不认识。这就是整个架构能拆成三个包的原因。

🚪 站 1：入口 main.ts 是个路由器#

packages/coding-agent/src/main.ts 自己不干活，它做三件事：解析参数、组装运行时、按模式分发。

1
if (appMode === "rpc") {
2
    await runRpcMode(runtime);
3
} else if (appMode === "interactive") {
4
    const interactiveMode = new InteractiveMode(runtime, {...});
5
    await interactiveMode.run();
6
} else {
7
    await runPrintMode(runtime, {...});
8
}

appMode 的判定藏着第一个小惊喜：

1
function resolveAppMode(parsed: Args, stdinIsTTY: boolean, stdoutIsTTY: boolean): AppMode {
2
    if (parsed.mode === "rpc") return "rpc";
3
    if (parsed.mode === "json") return "json";
4
    if (parsed.print || !stdinIsTTY || !stdoutIsTTY) return "print";  // 管道输入自动降级
5
    return "interactive";
6
}

stdin 不是终端就自动降级成 print 模式，所以 echo "hi" | pi 能一行跑完。同一个会话，三种驱动方式：TUI 事件驱动、一次性 prompt、JSON-RPC。读到这里先记住”会话”这个抽象，它是后面所有站点的共同底座。

🏭 站 2：总装车间，三个服务一个发动机#

main.ts 里有一段长长的依赖组装，产出的核心是三个服务：

1
const { settingsManager, modelRuntime, resourceLoader } = services;

• settingsManager 管配置（模型、工具开关、重试策略）
• modelRuntime 管模型目录、认证、发请求（站 7 的主角）
• resourceLoader 管加载扩展、技能、AGENTS.md

它们被装进 AgentSession，AgentSession 内部又创建一个 Agent，然后订阅它的事件：

1
// agent-session.ts:395
2
this._unsubscribeAgent = this.agent.subscribe(this._handleAgentEvent);

这行代码是整个项目事件架构的缩影：

数据单向流动：发动机产生事件，多个订阅者各取所需。这个设计让”重试、压缩、持久化”这些脏活全部挂在循环外面，循环本身保持纯净。后面你会看到这个”挂外面”的思路反复出现。

⌨️ 站 3：输入，prompt() 的预处理#

用户在 TUI 里按下回车，InteractiveMode 把文本交给 AgentSession.prompt()。注意它第一件事不是发消息：

1
async prompt(text: string, options?: PromptOptions): Promise<void> {
2
    // 1. 以 "/" 开头的先查扩展命令（/model、/compact 都是扩展命令）
3
    if (expandPromptTemplates && text.startsWith("/")) {
4
        const handled = await this._tryExecuteExtensionCommand(text);
5
        if (handled) return;  // 扩展命令自己管 LLM 交互
6
    }
7
    // 2. 展开 @file 模板、初始消息...
8
}

/model、/compact 这些你以为的”内置命令”，其实全是扩展命令。它们走的是另一条路，根本不进主循环。

然后进入 _runAgentPrompt，真正的转发：

1
private async _runAgentPrompt(messages: AgentMessage | AgentMessage[]): Promise<void> {
2
    this._isAgentRunActive = true;
3
    try {
4
        await this.agent.prompt(messages);
5
        while (await this._handlePostAgentRun()) {   // 重试 / 压缩循环
6
            await this.agent.continue();
7
        }
8
    } finally { ... }
9
}

_handlePostAgentRun 检查最后一条 assistant 消息：可重试错误就自动重试，上下文超限就自动压缩（把旧消息总结掉）。重试和压缩都是事后观察结果再干预，不是循环的一部分。这是事件驱动架构的直接收益：外科手术不用切开病人。

🔥 站 4：点火，Agent.prompt 的快照与钩子#

Agent.prompt() 干两件事：把输入规范成消息数组，然后带着上下文快照和循环配置点火：

1
private runPromptMessages(messages: AgentMessage[]) {
2
    await runAgentLoop(
3
        messages,
4
        this.createContextSnapshot(),   // { systemPrompt, messages, tools } 的拷贝
5
        this.createLoopConfig(),        // 所有钩子打包成配置
6
        (event) => this.processEvents(event),
7
        signal,
8
        this.streamFunction,            // ← 站 6 的主角
9
    );
10
}

两个细节值得停一下：

为什么拷贝？ 循环内部会原地修改 messages 数组（站 6 的 partial 替换），快照是为了隔离。

钩子即配置。 createLoopConfig 把 Agent 类的公共字段（beforeToolCall、afterToolCall、getSteeringMessages……）打包成 AgentLoopConfig。没有继承，没有接口实现类，全是回调函数。想扩展循环？往配置里塞一个函数就行。

💓 站 5：心脏，双层循环#

packages/agent/src/agent-loop.ts 的 runLoop 是整个项目最重要的函数。它是个双层循环：

1
while (true) {                          // 外层: 处理 follow-up 消息
2
    let hasMoreToolCalls = true;
3
    while (hasMoreToolCalls || pendingMessages.length > 0) {   // 内层: 一轮对话
4
        // 1. 注入排队中的 steering 消息
5
        // 2. 请求 LLM → assistant 消息
6
        // 3. 有 toolCall? 执行 → ToolResultMessage 进上下文 → hasMoreToolCalls = true
7
        // 4. 没 toolCall → 内层退出
8
    }
9
    // 外层: 查 follow-up 队列, 有就继续, 没有就 agent_end
10
}

为什么是双层？ 因为用户可以在 agent 干活时插话（steering，比如”等等，先别改那个文件”），也可以排队等它干完再问（follow-up）。内层循环处理”一轮对话”（一次 LLM 调用加若干工具执行），外层循环处理”这轮完了还有没有新指令”。

顺带一提：内层循环里有个小细节，stopReason === "length"（输出被 token 上限截断）时，会把这一批工具调用全部判失败而不是执行。因为截断的工具参数可能是残缺的 JSON，执行了反而危险。

🧱 站 6：LLM 边界，两次转换和 partial 替换#

streamAssistantResponse 是 agent 层和模型层的唯一接触点，也是全项目最优雅的流式处理：

1
// 转换 1: AgentMessage[] → Message[] (过滤 UI-only 消息, 转换自定义消息)
2
const llmMessages = await config.convertToLlm(messages);
3

4
// 转换 2: 调 StreamFn — 返回事件流而不是 Promise
5
const response = await streamFunction(config.model, llmContext, {...});
6

7
let partialMessage: AssistantMessage | null = null;
8
for await (const event of response) {
9
    switch (event.type) {
10
        case "start":
11
            partialMessage = event.partial;
12
            context.messages.push(partialMessage);   // 占位对象入数组
13
            break;
14
        case "text_delta": case "toolcall_delta": ...
15
            partialMessage = event.partial;
16
            context.messages[context.messages.length - 1] = partialMessage;  // 原地替换!
17
            emit({ type: "message_update", ... });
18
            break;
19
        case "done": case "error":
20
            return await response.result();
21
    }
22
}

partialMessage 是一个不断被整体替换的占位对象。每次 text_delta 到来，事件携带的是累积的完整 partial，循环直接拿新对象换掉数组最后一项：

好处是连锁的：订阅者每次收到 message_update 都是完整快照，不用自己拼增量；context.messages 的最后一项永远是最新状态，任何时刻读取都是安全的。增量拼装的复杂度被收敛在 provider 内部（processResponsesStream 负责解析 SSE 并产出累积 partial），上游全部享受完整快照的简单。

对比常见做法（UI 自己拼 delta 字符串），这个设计让每个环节都少了一个状态。

💰 站 7：钱的边界，ModelRuntime.prepareRequest#

默认的 StreamFn 是谁装的？sdk.ts 里一行：

1
setDefaultStreamFn(streamSimple);   // 来自 @earendil-works/pi-ai/compat

这行的意义在于：agent 包完全不依赖 ai 包，依赖方向由宿主注入。ModelRuntime.streamSimple 再套一层：

1
// model-runtime.ts:636
2
streamSimple(model, context, options) {
3
    return lazyStream(model, async () => {
4
        const prepared = await this.prepareRequest(model, options);   // ← 关键
5
        return prepared.provider.streamSimple(prepared.model, context, prepared.options);
6
    });
7
}

prepareRequest 是认证边界：解析该 provider 的凭证（API key、OAuth token、headers、baseUrl 覆盖），注入请求选项。所有 provider 的请求都从这里过，所以”凭证过期”只在这一层处理。它是整条链路上唯一碰钱和身份的地方。

🌐 站 8：真正的 HTTP，provider 适配器#

streamSimple 只是薄封装，真正干活的是 stream：

1
// openai-responses.ts:101
2
const client = createClient(model, context, apiKey, ...);
3
const { data: openaiStream, response } = await retryProviderRequest(
4
    () => client.responses.create(params, requestOptions).withResponse(),
5
    {...}
6
);
7
stream.push({ type: "start", partial: output });
8
await processResponsesStream(openaiStream, output, stream, model, {...});
9
stream.push({ type: "done", reason: output.stopReason, message: output });

两个学习要点：

返回值永远是事件流，错误也走事件流。 catch 里是 stream.push({type:"error"}) 而不是 throw。这与 StreamFn 的契约一致（types.ts 里写明：“必须不抛异常，失败编码在事件流里”）。调用方不需要 try/catch 包一层，for await 就能处理所有情况。

每个 provider 文件导出相同形状的 stream + streamSimple。 anthropic-messages.ts、google-generative-ai.ts、bedrock 系列……全是同一个接口的不同适配器。所谓”统一多 provider API”，全部秘密就这一句：一个接口，N 个适配器。

🛠️ 站 9：工具执行，三段式流水线#

模型说”我要调用 bash”，循环进入 executeToolCalls。每个工具调用走三段式：

1
// 第一段: prepare — 校验 + 前置钩子
2
const tool = currentContext.tools?.find((t) => t.name === toolCall.name);
3
const validatedArgs = validateToolArguments(tool, preparedToolCall);  // TypeBox schema
4
const beforeResult = await config.beforeToolCall?.(...);
5
if (beforeResult?.block) return createErrorToolResult(...);           // 拦截 → 错误结果
6

7
// 第二段: execute — 真正的工具函数
8
const result = await prepared.tool.execute(toolCall.id, prepared.args, signal, onUpdate);
9

10
// 第三段: finalize — 后置钩子 + 归一化
11
const afterResult = await config.afterToolCall?.(...);

最值得学的是错误处理哲学：工具不存在、参数校验失败、被钩子拦截，全都变成一条错误 ToolResultMessage 喂回给模型，让模型自己改正。错误是对话的一部分，循环不中断。beforeToolCall / afterToolCall 正是扩展系统（tool_call / tool_result 钩子）的挂载点。

📡 站 10：回声，一个事件四个去处#

Agent.processEvents 把每个事件先更新内部状态（streamingMessage、messages.push……），再广播给所有订阅者。订阅者一号是 AgentSession._handleAgentEvent：

1
// agent-session.ts:610
2
private _handleAgentEvent = async (event: AgentEvent): Promise<void> => {
3
    // 1. 用户消息若来自队列, 先从队列移除 (UI 看到最新队列状态)
4
    // 2. 转发给扩展 (tool_call / tool_result 等钩子)
5
    await this._emitExtensionEvent(event);
6
    // 3. 转发给 UI 订阅者
7
    this._emit(event.type === "agent_end" ? {...} : event);
8
    // 4. 持久化到磁盘 (session jsonl)
9
    if (event.type === "message_end") {
10
        if (event.message.role === "user" || event.message.role === "assistant" || ...) {
11
            this.sessionManager.appendMessage(event.message);
12
        }
13
    }
14
};

一个事件，四个去处。UI 不需要自己记历史，sessionManager 落盘的 jsonl 就是完整会话，这也是 --continue / --resume 能恢复会话的原因。

🐣 站 11：工具的出生，从定义到可执行#

最后补上工具是怎么来的。_buildRuntime 里：

1
const baseToolDefinitions = createAllToolDefinitions(this._cwd, {...});
2
// → read / bash / edit / write 等内置工具的定义

_refreshToolRegistry 里，内置工具和扩展工具统一包装（每个工具外面包一层扩展钩子），最后：

1
this.setActiveToolsByName([...new Set(nextActiveToolNames)]);
2
// → this.agent.state.tools = tools  (setter 拷贝数组)

工具的”出生证明”是 ToolDefinition（声明 TypeBox schema 和 prompt 片段），可执行的 AgentTool（带 execute）是运行时形态。schema 一份定义，两个用途：既校验参数，又生成发给模型的 tools JSON。

🧠 记忆锚点#

🧪 动手验证#

光读不练容易飘，这四个实验每个十分钟：

1. 看会话文件：跑 ./pi-test.sh 问一个问题，然后打开 ~/.pi/sessions/ 下的 jsonl。你会看到 user / assistant / toolResult 消息和 toolCall 的完整往返，这就是站 10 落盘的东西。
2. 数 LLM 调用次数：跑 pi -p "读一下 package.json"，观察它先调 read 工具、再调一次模型。两次 LLM 调用 = 一个回合，这就是内层循环。
3. 打断点看 partial：在 agent-loop.ts 的 streamAssistantResponse 里临时打印每次 text_delta 的累积长度，观察站 6 的替换机制。
4. 改一个数字感受重试：把 openai-responses.ts 里 maxRetries: 0 改成 3，感受重试层的位置。

结尾#

这条主线走完，Pi 的骨架就清楚了：一个双层循环，两个接口，一种事件流。剩下的内容，模型目录、OAuth、差分渲染、远程协议，都是挂在这根主干上的器官，想读哪块读哪块，不会再迷路。

这也是整个系列的地基。下一篇沿着模型层往下钻，看模型目录怎么生成、认证怎么解析、provider 适配器怎么接；再往后是工具、会话层。每一篇都会像本文一样，从一条具体的路径出发，带着图走完全程。