第3章：Agent Loop —— 让模型转动起来的引擎

Python 阅读说明：本版与 TypeScript 版共享同一份事实与正文结构。下列 Python 代码只用于解释 TypeScript 源码的控制流，并非可安装的 Pi Python SDK；字段名和类型以链接的 v0.80.2 TypeScript 源码为准。

前一章我们看了 Pi 的分层架构。架构只是”骨架”——Agent 真正的生命力来自”循环”。这一章，我们从最基础的问题出发：为什么需要循环？循环怎么转？什么时候停？ 然后追踪一条用户消息的完整旅程，看清 Agent Loop 的每一次心跳。

校对口径：本章对应 Pi v0.80.2 的 packages/agent/src/agent-loop.ts。先区分三个信号：stopReason 负责异常硬停，工具调用与 terminate 决定工具链是否续转，steering / follow-up 队列还能在没有工具调用时开启下一 Turn。

一、引子：大模型的三种用法

在聊 Agent Loop 之前，我们先退一步，看看”使用大模型”这件事本身有几种模式。这对理解”为什么需要循环”至关重要。

模式 1：直接调用 —— “模型，回答我”

最原始、最直接的用法。你构建好提示词，调用一次 API，拿到结果，完事。

用户输入 → 构建提示词 → 调模型 → 模型输出 → 展示结果

代码大概长这样：

# ============================================================
# 【Python 改写】直接调用 LLM 的最简形态
# 原文 TS:
#   const response = await llm.chat({
#     messages: [
#       { role: "system", content: "你是一个翻译助手" },
#       { role: "user", content: "把这段代码翻译成 Python" },
#     ],
#   });
#   console.log(response.content);
# ============================================================

# 概念对照：TS 的 await → Python 的 await；messages 是同样的 role+content 字典列表

response = await llm.chat(messages=[
    {"role": "system", "content": "你是一个翻译助手"},
    {"role": "user", "content": "把这段代码翻译成 Python"},
])
print(response["content"])

核心工作在于”构建提示词”。提示词写得好，结果就好。一次调用，一次输出，没有来回。

适用场景：翻译、摘要、问答、代码补全——凡是”一问一答”能搞定的事。

模式 2：Workflow —— “模型，你先做第一步，我检查一下，再做第二步”

当任务变复杂，你发现一次性很难得到满意结果。于是你把大任务拆成多步，每步调一次模型，步骤之间由你的代码控制流转。

用户输入 → [步骤1: 调模型分析] → [你的代码: 提取关键信息]
         → [步骤2: 调模型生成草稿] → [你的代码: 检查质量]
         → [步骤3: 调模型润色] → 最终输出

每一步模型只负责自己的那份工作，决策权在你手上——你知道什么时候该进入下一步，模型只是流水线上的一环。

适用场景：文档生成流水线、代码审查自动化、RAG（检索增强生成）。

模式 3：Agent Loop —— “模型，你自己决定怎么做”

到了 Agent 模式，你把决策权交给了模型。

用户输入 → 调模型 → 模型说"我需要读文件" → 执行读文件 → 模型看结果
         → 模型说"还需要搜索代码" → 执行搜索 → 模型看结果
         → 模型说"我知道了，答案是..." → 输出 → 结束

关键区别：步骤之间的流转不再由你写死，而是由模型的输出内容来驱动。 你的代码只做两件事：

1. 把用户的输入和工具执行结果喂给模型
2. 如果模型输出的内容里包含了工具调用请求，就执行它；如果没有，就认为任务完成了

至于”该调什么工具”、“该调几次”——这些由模型的输出内容决定。“什么时候该停”——这是人类定义的规则：当模型的一次输出中不再包含工具调用时，我们就认为循环可以结束了。

用一个对比表格，三种模式的区别一目了然：

二、先搞清楚几个概念：Trace、Turn

在深入源码之前，有两个概念必须分清楚。它们经常被混用，但在 Pi 的代码里，每个都有精确的含义。

Trace（一次完整运行）

一个 Trace 是从用户按下回车、到 Agent 彻底停下来、发出 agent_end 事件的整个过程。一个 Trace 包含多个 Turn。

一个 Trace（一次 agent_start 到 agent_end）
│
├─ Turn 1
│  模型返回 toolUse(read) → 执行 read
│
├─ Turn 2
│  带着 read 结果再次调用模型
│  模型返回 toolUse(edit) → 执行 edit
│
└─ Turn 3
   带着 edit 结果再次调用模型
   模型返回 stop、无工具调用，且消息队列为空 → agent_end

Turn（一个轮次）

一个 Turn 的定义非常精确：一次模型调用 + 这次调用触发的所有工具执行。

每个 Turn 由一对 turn_start 和 turn_end 事件包裹。关键点：一个 Turn 只有一次模型调用。 模型返回了 toolUse → 执行那批工具 → 发送 turn_end → 这个 Turn 就结束了。把工具结果喂回去再调模型，那是下一个 Turn。

看代码就更清楚了。内层循环每一圈的结构（后面会详讲）：

while (hasMoreToolCalls || ...) {
    if (!firstTurn) emit(turn_start);    // ← 新 Turn 开始

    处理 pendingMessages
    streamAssistantResponse()             // ← 一次模型调用
    检查 stopReason
    executeToolCalls()                    // ← 执行这个 Turn 触发的一批工具
    emit(turn_end);                       // ← 这个 Turn 结束

    prepareNextTurn / shouldStopAfterTurn / 检查 steering
}

一圈内层循环 = 一个 Turn = 一次 turn_start → 一次模型调用 → 工具执行 → 一次 turn_end。

如果模型在一个 Turn 中一口气要求了 3 个工具（read + grep + find），那这 3 个工具都在同一个 Turn 里执行——因为它们都是同一次模型调用的产物。但执行完毕后把结果喂回去再调模型的那一刻，就已经进入下一个 Turn 了。

所以 Trace 和 Turn 的关系就是

Trace（一次完整运行）
│  agent_start
│
├── Turn 1
│   │  turn_start
│   ├── 调模型 → toolUse → 执行工具（read + grep）
│   │  turn_end
│   │
├── Turn 2
│   │  turn_start
│   ├── 调模型 → toolUse → 执行工具（edit）
│   │  turn_end
│   │
├── Turn 3
│   │  turn_start
│   ├── 调模型 → stop → 没有工具
│   │  turn_end
│   │
│   ├── steering / follow-up 均为空
│   │
│   agent_end

注意：首轮 Turn 的 turn_start 是在 runAgentLoop() 入口就发出的，然后 runLoop() 内用 firstTurn 标志跳过首圈的 turn_start，避免重复。

放大查看

配图说明：一个 Trace 外壳内嵌 3 个 Turn，每个 Turn 都是”模型调用 + 工具执行”的完整闭环。Turn 3 的响应恰好是 stop 且没有 ToolCall；随后 Turn 钩子未要求停止、steering / follow-up 也都为空，Loop 才退出。stopReason = stop 本身并不是独立的退出开关。

三、全景：一条消息的旅程，以及循环怎么转

你输入了”帮我读一下 src/main.ts”并按下回车。从这一刻起发生了什么？让我们站在高处看一遍全过程，同时把”循环怎么转起来的”和”循环什么时候停”一并讲清楚。不必纠结每个细节——后面会逐段拆解源码。

途中你会遇到四个反复出现的”实体”：你的输入变成消息（Message）；Loop 调用**模型（Model）来思考；模型要求的操作用工具（Tool）完成；每一步通过事件（Event）**通知外部。驱动这四者反复运转的机制就是本章主角——Agent Loop。

流程全景

你按下回车："帮我读一下 src/main.ts"
│
├─ ① 创建 UserMessage
│  role: user
│  content: "帮我读一下 src/main.ts"
│
└─ ② 进入 agentLoop，发送 agent_start
   │
   └─ runLoop()
      │
      ├─ ③ AgentMessage → pi-ai Message
      │
      ├─ Turn 1
      │  ├─ turn_start
      │  ├─ ④ streamSimple() 调用模型
      │  ├─ AssistantMessage
      │  │  stopReason: toolUse
      │  │  content: ToolCall(read)
      │  ├─ ⑤ 执行 read → ToolResultMessage
      │  └─ turn_end
      │
      ├─ 判断：存在未终止的 toolCall → 继续
      │
      ├─ Turn 2
      │  ├─ turn_start
      │  ├─ ⑥ 带着工具结果再次调用模型
      │  ├─ AssistantMessage
      │  │  stopReason: stop
      │  │  content: TextContent
      │  └─ turn_end
      │
      ├─ 判断：无 toolCall，steering 为空
      ├─ 内层循环退出，follow-up 为空
      └─ agent_end（共 2 个 Turn）

循环怎么转：三个信号，而不是一个字段

模型每次返回的 AssistantMessage 都有 stopReason，但 v0.80.2 的 Loop 并不只靠这个字段决定下一步。把三个信号分开，代码就不会被误读：

1. stopReason === "error" | "aborted"：立即结束整个 Trace；
2. content 中存在 toolCall，且这批工具没有全部 terminate: true：继续下一 Turn；
3. steering 或 follow-up 队列有消息：即使本轮没有 toolCall，也会继续。

但在此之前，必须澄清一个关键认知：Provider 会报告生成为什么结束，却不会提供一个可信的“现实任务已经完成”布尔值。 stop、length、toolUse 描述的是这次模型生成的终态；宿主仍要结合工具内容、终止标记和消息队列决定整个 Trace 是否继续。stopReason 的值来自两个不同的地方：

模型 API 真正返回的三种：

框架的流式层注入的两种（模型 API 本身不会返回这两种值）：

代码证据（packages/ai/src/）：当 streamSimple 内部的 API 调用抛出异常时，catch 块执行 output.stopReason = options?.signal?.aborted ? "aborted" : "error"。这不是模型说的，是框架替它”兜底”的。

正常工具链的主规则

排除硬停止和外部队列后，正常工具链的主规则是：模型输出里有没有工具调用。这背后是一条人类定义的工程约定：

如果模型一次输出中没有工具调用，就认为本轮不需要更多操作，循环可以停了。

这是一条宿主循环约定，不等于对现实任务完成度的证明。换个说法：“没有工具调用”只让当前工具链自然结束；如果 steering / follow-up 还有消息，Trace 仍会继续。

代码里最精炼的判断就是这个（伪代码示意，实际见 agent-loop.ts:202-216）：

# ============================================================
# 【Python 改写】简化逻辑（实际见 agent-loop.ts:202-216）
# 原文 TS:
#   const toolCalls = message.content.filter(c => c.type === "toolCall");
#   hasMoreToolCalls = false;
#   if (toolCalls.length > 0) {
#       const executedToolBatch = await executeToolCalls(...);
#       hasMoreToolCalls = !executedToolBatch.terminate;
#   }
# ============================================================

# 概念对照：TS 的 array.filter → Python 列表推导；
# TS 的 `arr.length > 0` → Python 的 `len(arr) > 0`；
# 只有整批工具都 terminate，has_more_tool_calls 才为 False

tool_calls = [c for c in message.content if c.type == "toolCall"]
has_more_tool_calls = False
if len(tool_calls) > 0:
    executed_tool_batch = await execute_tool_calls(...)
    has_more_tool_calls = not executed_tool_batch.terminate  # 整批工具都 terminate 才停止

注意：实际驱动循环的不是 stopReason === "toolUse"，而是 toolCalls 数组长度 > 0 && !terminate。这意味着：即使 stopReason === "length"（被截断），只要 content 里有 toolCall 块，循环仍会执行工具；反之，即使 stopReason === "toolUse"，如果所有工具结果都设置 terminate: true，循环也会停。

内层循环的条件是 while (hasMoreToolCalls || pendingMessages.length > 0)：

• 模型返回 toolCall 且工具未 terminate → hasMoreToolCalls = true → 继续转：执行工具，把结果喂回去再调模型
• 任意非硬停止响应若无 toolCall → hasMoreToolCalls = false → 工具链准备停（随后仍要看 Turn 钩子和消息队列）
• stopReason === "error" 或 "aborted" → 硬停止：立即退出整个循环，不检查 followUp

调模型
  ├─ stopReason = error / aborted
  │    └─ emit turn_end + agent_end → 硬停止
  │
  └─ 其他 stopReason
       ├─ content 有 toolCall → 执行整批工具
       │    └─ hasMoreToolCalls = !batch.terminate
       └─ content 无 toolCall → hasMoreToolCalls = false
                    │
                    ▼
       turn_end → prepareNextTurn → shouldStopAfterTurn
                    │
                    ▼
       拉取 steering → 检查 (hasMoreToolCalls || pendingMessages)
                    ├─ true  → 下一 Turn
                    └─ false → 拉取 follow-up → 续转或 agent_end

为什么不让代码更智能地判断“任务完成没”？ 在开放式任务里，宿主通常不知道模型要读几个文件、改几处代码。Pi 因而采用可观察的协议事实——工具调用、终止标记和消息队列——而不是另造一个“完成度”分类器。简单的代价是：没有 toolCall 只表示当前工具链自然结束，不证明现实中的任务一定完成。

循环的退出路径（配图聚焦三类续转信号）

放大查看

配图说明：先处理 error/aborted 硬停止；其余响应从内容中提取 toolCall，并结合整批 terminate 记录工具链是否待续。无论有没有工具，普通路径都会经过 turn_end → prepareNextTurn → shouldStopAfterTurn：钩子要求停止时直接结束；否则只要工具链待续或 steering 非空，就进入下一 Turn。两者都为 false 时内层循环结束，再拉取 follow-up；follow-up 非空仍会进入下一 Turn，三者均空才正常发出 agent_end。

四、源码详解：基础 Loop 与 coding-agent 的叠加设计

§三 给了你概念全景：消息怎么流动、三个信号如何共同决定继续或停止。下面走进代码，回答“怎么做到的”。

在看 Pi 的源码之前，先搞清楚一件事：最简单的 Agent Loop 其实极其简短。

最简 Loop：工具型 Agent 的常见骨架

剥掉所有产品特性，一个能用的 Agent Loop 只需要这些：

# ============================================================
# 【Python 改写】最简 Agent Loop（伪代码）
# 原文 TS:
#   async function simpleLoop(messages, model, tools) {
#       while (true) {
#           const response = await callModel(model, messages, tools);
#           messages.push(response);
#           if (response.stopReason !== "toolUse") return messages;
#           for (const toolCall of response.toolCalls) {
#               const result = await executeTool(toolCall);
#               messages.push(result);
#           }
#       }
#   }
# ============================================================

# 概念对照：TS 的 while(true)+return → Python 同样写法；
# TS 的 for...of → Python 的 for...in；TS 的 !== → Python 的 !=
# TS 的数组 push → Python 的 list append

async def simple_loop(messages, model, tools):
    while True:
        # ① 调模型
        response = await call_model(model, messages, tools)
        messages.append(response)

        # ② 没有工具调用 → 结束
        if response.stop_reason != "toolUse":
            return messages

        # ③ 有工具调用 → 执行，把结果喂回去
        for tool_call in response.tool_calls:
            result = await execute_tool(tool_call)
            messages.append(result)

十几行代码。一个 while 循环，调模型、执行工具、再调模型，直到模型不再要求工具。这是工具调用型 Agent 的常见最小骨架；纯分类器、一次性生成器或确定性 Workflow 并不需要这条循环。

通用 Loop 暴露了什么，coding-agent 又接入了什么

低层 AgentLoopConfig 定义 steering、follow-up、prepareNextTurn 和 shouldStopAfterTurn。上层 Agent 封装了两类队列并暴露 prepareNextTurn，但 v0.80.2 的 AgentOptions 没有继续暴露 shouldStopAfterTurn。coding-agent 会接入 steering / follow-up 与 transformContext 扩展钩子；自动压缩则由 AgentSession 在一次 Agent 运行结束后检查，不通过 shouldStopAfterTurn 触发。

关键认知：agent-core 提供机制，但不同封装层暴露的接口并不完全相同，coding-agent 也只接入其中一部分。阅读时要区分 runAgentLoop()、Agent 和 AgentSession 三个边界。

但理解 coding-agent 怎么叠加这些设计很有价值——你自己的产品场景很可能也需要类似机制。接下来，我们以 coding-agent 的完整源码为例，逐段走这些设计。跟着那条”帮我读一下 src/main.ts”的消息，走完从入口到结束的整段旅程。

4.1 入口：runAgentLoop() 收到了什么

§三简要展示了流程全景，这里展开看代码细节——同一个过程，更深入的视角。

你按下回车后，调用链是：Agent.prompt() → runPromptMessages() → runAgentLoop()。停在入口：

# ============================================================
# 【Python 改写】agent-loop.ts:95-118 入口函数签名
# 原文 TS:
#   async function runAgentLoop(
#       prompts: AgentMessage[],     // 你的消息
#       context: AgentContext,       // 当前对话上下文（顶层浅拷贝）
#       config: AgentLoopConfig,     // 循环配置（模型、钩子、队列回调）
#       emit: AgentEventSink,        // 事件发射器
#       signal?: AbortSignal,        // 中止信号
#       streamFn?: StreamFn,         // 流式函数（可替换）
#   ): Promise<AgentMessage[]>
# ============================================================

# 概念对照：TS 的可选参数 `?` → Python 的默认 None；
# TS 的 AbortSignal → Python 的 asyncio.Event 或 cancellation；
# TS 的 `Promise<T>` → Python 的 `Coroutine[..., T]` / awaitable

from typing import Optional, Awaitable

async def run_agent_loop(
    prompts: list[AgentMessage],          # 你的消息
    context: AgentContext,                # 当前对话上下文（顶层浅拷贝）
    config: AgentLoopConfig,              # 循环配置（模型、钩子、队列回调）
    emit: AgentEventSink,                 # 事件发射器
    signal: Optional[AbortSignal] = None, # 中止信号
    stream_fn: Optional[StreamFn] = None, # 流式函数（可替换）
) -> list[AgentMessage]: ...

六个参数中最重要的三个：

prompts — 你的消息已经被包装成了标准格式：

# ============================================================
# 【Python 改写】prompts 数据结构
# 原文 TS:
#   [{ role: "user",
#      content: [{ type: "text", text: "帮我读一下 src/main.ts" }],
#      timestamp: 1748000000000 }]
# ============================================================

# 此时 prompts 的数据结构：
[{
    "role": "user",
    "content": [{"type": "text", "text": "帮我读一下 src/main.ts"}],
    "timestamp": 1748000000000,
}]

context — 对话上下文快照。这里的“快照”是顶层浅拷贝：createContextSnapshot() 分别对 messages 和 tools 调用 slice()，因此 Loop 向这两个数组追加元素或替换数组时，不会直接改写 Agent 持有的原数组；但数组里的消息对象和工具对象仍是共享引用，原地修改这些对象仍可能被两边观察到。Agent 自身的消息状态会另外通过事件归约更新，不能把这份浅拷贝理解成完全隔离的深拷贝。

# ============================================================
# 【Python 改写】context 数据结构——Agent 对话状态的完整快照
# 原文 TS:
#   {
#     systemPrompt: "You are a helpful coding assistant...",
#     messages: [ /* 之前的对话历史 */ ],
#     tools: [
#       { name: "read", description: "...", parameters: Type.Object({...}), execute: ... },
#       { name: "bash", description: "...", parameters: Type.Object({...}), execute: ... },
#     ]
#   }
# ============================================================

{
    "system_prompt": "You are a helpful coding assistant...",
    "messages": [...],  # 之前的对话历史
    "tools": [
        {"name": "read", "description": "...", "parameters": {"type": "object"}, "execute": ...},
        {"name": "bash", "description": "...", "parameters": {"type": "object"}, "execute": ...},
    ],
}

config — Loop 的行为配置。这里有一组关键钩子（都是函数，不是数据）：

# ============================================================
# 【Python 改写】AgentLoopConfig 的关键字段
# 原文 TS:
#   {
#     model: Model,
#     convertToLlm: Function,
#     transformContext?: Function,
#     getSteeringMessages?: Function,
#     getFollowUpMessages?: Function,
#     shouldStopAfterTurn?: Function,
#     beforeToolCall?: Function,
#     afterToolCall?: Function,
#     toolExecution: "parallel",
#   }
# ============================================================

# 概念对照：TS 的可选字段 `?:` → Python 的 Optional[T]；TS 的函数类型 → Python 的 Callable
from typing import Callable, Optional

{
    "model":                      Model,                 # 用哪个 LLM
    "convert_to_llm":             Callable,              # list[AgentMessage] → list[Message] 转换
    "transform_context":          Optional[Callable],    # 调 LLM 前的上下文预处理（如压缩）
    "get_steering_messages":      Optional[Callable],    # 获取"紧急插队"消息
    "get_follow_up_messages":     Optional[Callable],    # 获取"追加任务"消息
    "should_stop_after_turn":     Optional[Callable],    # 每轮结束后是否该停
    "before_tool_call":           Optional[Callable],    # 工具执行前钩子
    "after_tool_call":            Optional[Callable],    # 工具执行后钩子
    "tool_execution":             "parallel",            # 工具执行模式
}

这些钩子都是函数而非数据——Loop 在运行时调用它们来”拉取”最新状态。这让 Loop 和外部消息来源彻底解耦。

入口函数只做了三步准备：

Step 1: 创建 newMessages 数组
        → 收集本轮 Trace 产生的所有新消息

Step 2: 把 prompts 追加到 context.messages
        → context.messages = [...context.messages, ...prompts]

Step 3: 发初始事件
        → emit("agent_start")    ← Trace 开始
        → emit("turn_start")     ← 首轮 Turn 开始
        → 对每条 prompt：
          emit("message_start") + emit("message_end")
        → 调用 runLoop()

数据变化：

入口前：
  context.messages = [user1, asst1, toolResult1]    ← 之前的对话
  newMessages = []

入口后：
  context.messages = [user1, asst1, toolResult1, user2]  ← 你的消息被追加
  newMessages = [user2]                                   ← 收集器开始记录

4.2 runLoop() 的骨架：先看内核，再看叠加

现在进入 runLoop()——整个系统最核心的代码。别被它的长度吓到，我们先看内核，再看叠加。

内核：内层循环

如果只保留最简 Loop 的逻辑，runLoop 长这样：

# ============================================================
# 【Python 改写】只保留内核的 run_loop（伪代码）
# 原文 TS:
#   while (hasMoreToolCalls) {
#       // 步骤 B：调 LLM
#       // 步骤 C：检查 stopReason → error/aborted 就退出
#       // 步骤 D：执行工具
#       // 步骤 E：emit turn_end
#   }
#   // 结束 → emit agent_end
# ============================================================

while has_more_tool_calls:
    # 步骤 B：调 LLM
    # 步骤 C：检查 stop_reason → error/aborted 就退出
    # 步骤 D：执行工具
    # 步骤 E：emit turn_end
    ...
# 结束 → emit agent_end

这就是最简工具 Loop——调模型、执行工具、turn_end，循环往复。内层循环的退出条件 hasMoreToolCalls 由 toolCalls 数组长度 > 0 && !terminate 驱动（§三讲过）。

可选机制：消息队列给内核加了两层外壳

通用 runLoop() 还支持两种消息队列；coding-agent 的交互层会向这些队列写入用户消息：

叠加 1：steering 消息注入（内层循环开头 + 每圈结尾各检查一次）。用户在 Agent 工作时输入了新指令——这些消息不能等当前任务跑完，得在下一圈开头紧急注入。所以内层循环条件多了一个 || pendingMessages.length > 0。

叠加 2：外层 followUp 循环（包在整个内层循环外面）。Agent 自然停了之后，系统可能还想追加任务（比如”顺便跑个测试”）。外层循环让这些追加任务在同一个 Trace 内继续跑，不需要重新启动一个新的 Loop。

把内核和两层叠加拼起来，才是完整的 runLoop 骨架：

# ============================================================
# 【Python 改写】runLoop 的完整骨架（内核 + 两层叠加）
# 原文 TS:
#   async function runLoop(currentContext, newMessages, config, signal, emit, streamFn) {
#       let firstTurn = true;
#       let pendingMessages = (await config.getSteeringMessages?.()) || [];
#       while (true) {
#           let hasMoreToolCalls = true;
#           while (hasMoreToolCalls || pendingMessages.length > 0) {
#               if (!firstTurn) emit({ type: "turn_start" });
#               firstTurn = false;
#               // 步骤 A~F
#           }
#           const followUpMessages = (await config.getFollowUpMessages?.()) || [];
#           if (followUpMessages.length > 0) {
#               pendingMessages = followUpMessages;
#               continue;
#           }
#           break;
#       }
#   }
# ============================================================

async def run_loop(current_context, new_messages, config, signal, emit, stream_fn):

    # 只在整个 Trace 的首个 Turn 跳过 turn_start；follow-up 不会重置它
    first_turn = True

    # ① 首次 steering 检查（在进入内层循环之前！）
    pending_messages = (await config.get_steering_messages()) if config.get_steering_messages else []

    # ========== 叠加2：外层循环（followUp 续命）==========
    while True:
        has_more_tool_calls = True

        # ========== 内核 + 叠加1：内层循环 ==========
        while has_more_tool_calls or len(pending_messages) > 0:
            #                                    ↑ 叠加1：steering 消息也驱动循环

            if not first_turn:
                await emit({"type": "turn_start"})
            first_turn = False

            # 步骤 A：注入 pending_messages（steering 消息）← 叠加1
            # 步骤 B：调 LLM → stream_assistant_response()  ← 内核
            # 步骤 C：检查 stop_reason                      ← 内核
            # 步骤 D：执行工具                              ← 内核
            # 步骤 E：emit turn_end                         ← 内核
            # 步骤 F：prepare_next_turn → should_stop_after_turn ← 叠加（钩子）
            #         → 再次检查 steering                          ← 叠加1
            ...

        # ========== 内层循环结束 ==========
        # 叠加2：检查 followUp 队列
        follow_up_messages = (await config.get_follow_up_messages()) if config.get_follow_up_messages else []
        if len(follow_up_messages) > 0:
            pending_messages = follow_up_messages
            continue   # 回到外层循环顶部，内层循环重开

        break      # 两个队列都空，真正退出

现在我们逐步骤展开。每个步骤会标注”内核”还是”叠加”，方便你区分。

4.3 【叠加1 · 步骤A】steering 消息注入

什么是 steering？ 这是 coding-agent 的一个交互功能。想象你让 Agent 帮你修一个 bug，Agent 正在读文件、分析代码。这时候你突然想到一个补充：“也检查一下测试文件”——你希望这条指令能插队，而不是等 Agent 把当前任务做完再说。

steering 就是这个”插队”机制。用户在 Agent 工作期间输入的新指令，会被放进 steering 队列。每圈内层循环开头，Loop 先检查这个队列，把紧急消息注入到当前对话中：

# ============================================================
# 【Python 改写】steering 消息注入
# 原文 TS:
#   if (pendingMessages.length > 0) {
#       for (const message of pendingMessages) {
#           await emit({ type: "message_start", message });
#           await emit({ type: "message_end", message });
#           currentContext.messages.push(message);
#           newMessages.push(message);
#       }
#       pendingMessages = [];
#   }
# ============================================================

# 概念对照：TS 的 `array.length > 0` → Python 的 `len(list) > 0`；
# TS 的 `for...of` → Python 的 `for...in`；TS 的 `array.push` → Python 的 `list.append`

if len(pending_messages) > 0:
    for message in pending_messages:
        await emit({"type": "message_start", "message": message})
        await emit({"type": "message_end",   "message": message})
        current_context.messages.append(message)
        new_messages.append(message)
    pending_messages = []   # 消费完毕，清空

这段代码就是把紧急消息逐条注入到上下文和消息收集器中。

pendingMessages 的第一个来源是 runLoop 一进来就执行的首次 steering 检查（agent-loop.ts:167）。它会接住第一次模型请求发出之前已经排队的消息，并让这些消息参与首个 Turn。第一次请求已经发出后才到达的输入，则要等当前 Turn 末尾的第二次 steering 检查（agent-loop.ts:253）才能注入。

4.4 【内核 · 步骤B】streamAssistantResponse() — 调 LLM

这是整个 Loop 最重的一步——把消息发给模型、拿回流式响应。分为四个阶段。

阶段 A：上下文预处理（可选）

# ============================================================
# 【Python 改写】上下文预处理
# 原文 TS:
#   let messages = context.messages;
#   if (config.transformContext) {
#       messages = await config.transformContext(messages, signal);
#   }
# ============================================================

messages = context.messages
if config.transform_context:
    messages = await config.transform_context(messages, signal)

如果配置了 transformContext，就在这里过滤、重排或注入消息。不配置就跳过。coding-agent v0.80.2 用它分发扩展的 context 事件；压缩由 AgentSession 另行生成 CompactionEntry 并重建消息，不是在这个函数里触发。

阶段 B：AgentMessage → Message 转换（两层消息的边界）

# ============================================================
# 【Python 改写】AgentMessage → Message 转换
# 原文 TS: const llmMessages = await config.convertToLlm(messages);
# ============================================================

llm_messages = await config.convert_to_llm(messages)

这一行站在 Agent 内核和 LLM 的边界上。要理解它为什么存在，得先知道”两层消息”的设计。

Agent 内部维护对话历史时，需要记录的不只是”用户说了什么、AI 回了什么”——它还需要记录自己的内部状态。比如 coding-agent 会记录：上下文被压缩过（CompactionSummaryMessage）、Bash 命令的执行详情（BashExecutionMessage）、会话树回退路径的摘要（BranchSummaryMessage）。这些是 Agent 自己用的”内部语言”，不能直接交给 Provider；进入 pi-ai 的标准消息仍只有 UserMessage、AssistantMessage、ToolResultMessage。

convertToLlm 就是站在这个边界上的翻译官。必须区分两层实现：如果只实例化裸 pi-agent-core 的 Agent、没有传入自定义转换器，core 的默认实现只是一个 .filter()，会只保留三种标准消息：

# ============================================================
# 【Python 改写】default_convert_to_llm
# 原文 TS:
#   function defaultConvertToLlm(messages: AgentMessage[]): Message[] {
#       return messages.filter(
#           (message) => message.role === "user"
#                     || message.role === "assistant"
#                     || message.role === "toolResult",
#       );
#   }
# ============================================================

# 概念对照：TS 的 array.filter(callback) → Python 的 [m for m in messages if pred(m)]
from typing import Callable

def default_convert_to_llm(messages: list[AgentMessage]) -> list[Message]:
    """只保留 LLM 认识的三种标准消息，过滤掉 Agent 内部状态消息。"""
    return [
        m for m in messages
        if m.role in ("user", "assistant", "toolResult")
    ]

裸 agent-core 的默认数据变换是：

转换前（AgentMessage[]）：
[
  { role: "user", content: "帮我读一下 src/main.ts", ... },    ← 保留
  { role: "assistant", content: [...], ... },                   ← 保留
  { role: "compactionSummary", summary: "之前的对话摘要..." },   ← 过滤掉
  { role: "toolResult", content: [...], ... },                  ← 保留
]

转换后（Message[]）：
[
  { role: "user", content: "帮我读一下 src/main.ts", ... },
  { role: "assistant", content: [...], ... },
  { role: "toolResult", content: [...], ... },
]

但完整的 coding-agent 不会使用这条默认过滤路径。core/sdk.ts 创建 Agent 时注入了自己的转换器：compactionSummary、branchSummary、未排除的 bashExecution 和 custom 都会转换为携带相应文本或内容的 UserMessage；原本的 user / assistant / toolResult 则原样保留。外层包装还会按 blockImages 设置过滤图片。因此，压缩摘要等产品消息会以标准消息的形式继续进入模型上下文，而不是被丢弃。

两层消息的完整设计，详见《第6章：消息系统》。

阶段 C：构建 Context 并调用模型

# ============================================================
# 【Python 改写】构建 llm_context 并调用模型
# 原文 TS:
#   const llmContext: Context = {
#       systemPrompt: context.systemPrompt,
#       messages: llmMessages,
#       tools: context.tools,
#   };
#   const streamFunction = streamFn || streamSimple;
#   const resolvedApiKey =
#       (config.getApiKey ? await config.getApiKey(config.model.provider) : undefined)
#       || config.apiKey;
#   const response = await streamFunction(config.model, llmContext, {
#       ...config, apiKey: resolvedApiKey, signal,
#   });
# ============================================================

# 概念对照：TS 的对象字面量 → Python 的 dict 或 dataclass；
# TS 的 `a || b` 短路求值 → Python 的 `a if a else b`；
# TS 的 `...config` 展开 → Python 的 {**config, ...}

llm_context: Context = {
    "system_prompt": context.system_prompt,
    "messages":      llm_messages,
    "tools":         context.tools,
}

stream_function = stream_fn or stream_simple
resolved_api_key = (
    (await config.get_api_key(config.model.provider)) if config.get_api_key else None
) or config.api_key

response = await stream_function(config.model, llm_context, {
    **config, "api_key": resolved_api_key, "signal": signal,
})

构建 Context 是这一步的主线。 注意 llmContext 是一个全新对象，每圈内层循环都重建一次。它由三部分组成：

• systemPrompt —— 直接复用 Agent 上下文里的系统提示词，告诉模型”你是谁、该遵守什么规则”
• messages —— 就是上一步 convertToLlm 转换后的 llmMessages；自定义消息已被过滤或翻译成 pi-ai 的三种标准角色
• tools —— 工具列表（带 schema 定义），让模型知道”这次有哪些工具可以调”

这里的赋值只表示：本轮构造 wrapper 时，tools 取自当时的 currentContext.tools，systemPrompt 取当时的字符串值。若宿主没有替换 context，工具数组通常复用原引用，消息历史则随着 AssistantMessage / ToolResultMessage 追加而增长；但这不是跨 Turn 的不变量。

为什么每圈都重建 llmContext？transformContext 每轮都可能给出不同的 messages，prepareNextTurn 还可以替换后续 Turn 的整个 context——包括 systemPrompt、messages 和 tools——以及 model 或 thinking level。新 wrapper 表达的是“本轮实际发送的上下文”；不能据此宣称 tools 或 system prompt 在整个 Trace 中字节级稳定。

重建 JavaScript 对象不等于缓存一定失效。 Provider 最终看到的是序列化后的请求；是否命中、缓存多久以及如何计费，仍由各 Provider 的协议与服务端规则决定。v0.80.2 的 Anthropic 适配器会在稳定前缀附近设置 cache_control，实现见 anthropic-messages.ts：

第三个 breakpoint 会跟着最近的 user message 移动。可以把预期的前缀复用理解成：

Turn 1: 写入 [system + tools] → 写入 [messages §1]
Turn 2: 命中 [system + tools] → 命中 [messages §1] → 写入 [messages §2]
Turn 3: 命中 [system + tools] → 命中 [messages §1+§2] → 写入 [messages §3]

还有个容易误解的点：tools 不是被塞到 messages 末尾。Anthropic API 协议里 tools 是独立的顶层字段（位置在 messages 之前），这个协议设计本身就考虑了 cache——稳定的 tools 在前、变动的 messages 在后，prefix 越长越省。

OpenAI 兼容适配器和 OpenAI Responses 适配器还有各自的缓存参数路径。这里的工程结论只有一个：Agent Loop 只重建统一 Context；缓存语义必须留给 Provider 适配层解释。

阶段 D：流式处理响应 —— 原地替换的妙用

streamFunction 返回的是 AssistantMessageEventStream——一个异步迭代器。这步有一个很精巧的设计：

# ============================================================
# 【Python 改写】流式响应处理（原地替换）
# 原文 TS:
#   for await (const event of response) {
#       switch (event.type) {
#           case "start":
#               partialMessage = event.partial;
#               context.messages.push(partialMessage);
#               emit({ type: "message_start", ... });
#               break;
#           case "text_delta":
#           case "toolcall_delta":
#           case "thinking_delta":
#               partialMessage = event.partial;
#               context.messages[last] = partialMessage;   // ★ 原地替换
#               emit({ type: "message_update", ... });
#               break;
#           case "done":
#           case "error":
#               finalMessage = await response.result();
#               context.messages[last] = finalMessage;
#               emit({ type: "message_end", ... });
#               return finalMessage;
#       }
#   }
# ============================================================

# 概念对照：TS 的 for await...of → Python 的 async for...in；
# TS 的 switch/case → Python 的 if/elif（或 match/case，Python 3.10+）；
# TS 的 `array[last] = x` → Python 的 `array[-1] = x`

async def consume_response(response, context, emit):
    async for event in response:
        if event.type == "start":
            # 拿到一个“空壳”消息，直接 append 到 context
            partial_message = event.partial
            context.messages.append(partial_message)
            await emit({"type": "message_start", "message": partial_message})

        elif event.type in ("text_delta", "toolcall_delta", "thinking_delta"):
            partial_message = event.partial              # 更新后的部分消息
            context.messages[-1] = partial_message        # ★ 原地替换
            await emit({"type": "message_update", "message": partial_message})

        elif event.type in ("done", "error"):
            final_message = await response.result()
            context.messages[-1] = final_message          # ★ 用最终完整消息替换
            await emit({"type": "message_end", "message": final_message})
            return final_message

为什么要先 push 空壳再原地替换？注意”原地替换”的意思——不是往 context.messages 数组里 push 新条目，而是用新内容覆盖最后一条（context.messages[last] = partialMessage）。这样 context 的消息数量不变，但最后一条消息的内容在”长大”。

为了UI 能实时展示。如果等全部响应完成才 push 到 context，用户在 LLM 生成期间只能等待。通过“先放空壳、按 delta 替换”，UI 经由 message_update 拿到最新部分消息并持续刷新；一个 delta 可能包含一个或多个 token，不应把两者等同。

放大查看

配图说明：四个时间点的 context.messages[last] 演变——start 时空壳、text_delta 时文字在长、toolcall_delta 时工具调用出现、done 时最终完整消息替换。每一步同步展示 UI 显示状态。

数据在流式响应期间，context.messages[last] 的演变：

start
└─ { role: "assistant", content: [] }  ← push 空壳

text_delta
└─ { content: [{ type: "text", text: "好的..." }] }

toolcall_delta
└─ { content: [
       { type: "text", text: "好的..." },
       { type: "toolCall", name: "read", arguments: {...} }
   ] }

done
└─ { content: [...], stopReason: "toolUse", usage: {...} }
   最终完整消息替换

4.5 【内核 · 步骤C】检查 stopReason

§三详细讲了 stopReason 的来源和含义。这里看代码怎么处理它。拿到 AssistantMessage 后，立即做一次硬停止判断：

# ============================================================
# 【Python 改写】agent-loop.ts:196-200 硬停止判断
# 原文 TS:
#   if (message.stopReason === "error" || message.stopReason === "aborted") {
#       await emit({ type: "turn_end", message, toolResults: [] });
#       await emit({ type: "agent_end", messages: newMessages });
#       return;
#   }
# ============================================================

if message.stop_reason in ("error", "aborted"):
    await emit({"type": "turn_end", "message": message, "tool_results": []})
    await emit({"type": "agent_end", "messages": new_messages})
    return   # ← 直接退出整个 run_loop，不检查 followUp

error 和 aborted 是”硬停止”——立即发 turn_end + agent_end，直接 return。连工具都不执行，连 followUp 都不检查。这是一种快速失败（fail fast）策略：既然模型调用本身就失败了（网络异常或用户中止），继续跑没有任何意义。

而 stop、toolUse、length 这三种，代码都会继续从 message.content 过滤 toolCall。Loop 并没有用这三个 stopReason 推导 toolCall 是否存在：例如被截断的 length 响应仍可能带有一个已经完整解析出的 toolCall，此时它照样会执行；只有过滤结果为空时，工具执行步骤才什么也不做。

4.6 【内核 · 步骤D】executeToolCalls() — 执行工具

模型返回的 AssistantMessage.content 里，可能有多个 type === "toolCall" 的块（模型一口气要求了多个操作）。先过滤出来：

# ============================================================
# 【Python 改写】过滤出工具调用
# 原文 TS: const toolCalls = message.content.filter((c) => c.type === "toolCall");
# ============================================================

tool_calls = [c for c in message.content if c.type == "toolCall"]

然后决定这批工具并行还是串行执行：

# ============================================================
# 【Python 改写】并行/串行选择
# 原文 TS:
#   if (config.toolExecution === "sequential" || hasSequentialToolCall) {
#       return executeToolCallsSequential(...);
#   }
#   return executeToolCallsParallel(...);
# ============================================================

if config.tool_execution == "sequential" or has_sequential_tool_call:
    return await execute_tool_calls_sequential(...)
return await execute_tool_calls_parallel(...)

“一票否决”策略：只要这批工具中有任何一个声明了 executionMode: "sequential"，整批都串行。调度器没有通用的资源冲突模型，这条规则用可能增加等待换取可预测性，但不自动保证工具安全。内置 edit 和 write 还通过 withFileMutationQueue 按同一文件路径串行化，这是工具内部的第二层协调。

两种执行模式的内部结构：

串行模式：
  ToolCall A: 准备 → 验证 → beforeHook → 执行 → afterHook → emit end
  ToolCall B: 准备 → 验证 → beforeHook → 执行 → afterHook → emit end
  （一个完全结束，才开始下一个）

并行模式（三阶段设计）：
  阶段1 - 准备（顺序）：  A 准备 → B 准备 → C 准备
      ↑ 实现按顺序 await 参数准备、验证和 beforeHook
  阶段2 - 执行与收尾（并行）：
      A、B、C 的 execute → afterHook → emit end 通过 Promise.all 并发
  阶段3 - 结果消息（有序）：
      ↑ Promise.all 返回值保持输入顺序，再按 ToolCall 顺序发送 result

注意：准备阶段按顺序等待，但 B 被拦截只会让 B 形成即时错误结果；只要 signal 没有终止，C 仍会继续准备。执行、后置钩子和 end 事件可以并发，最终 ToolResultMessage 再按调用顺序发出。

每个工具的结果被包装成 ToolResultMessage，追加到 context 和 newMessages：

工具执行后：
  context.messages = [..., user2, assistantMessage, {
    role: "toolResult", toolCallId: "toolu_01", toolName: "read",
    content: [{ text: "文件内容..." }], isError: false
  }]

terminate 机制：工具可以在返回结果中设置 terminate: true，表示”我觉得不该继续了”。但 Loop 不会因为某一个工具喊停就停——它用的是 every 而非 some：必须这批工具的全部结果都设置了 terminate 才真正退出。这是保守策略：只要有一个工具还在正常工作，Loop 就不中断。

工具执行的完整五步管道（prepareArguments → Schema 验证 → beforeToolCall → execute → afterToolCall）详见《第5章：工具系统》。

4.7 【内核+叠加 · 步骤E~F】turn_end + 钩子 + 再次检查 steering

Turn 的核心工作做完了（调模型 + 执行工具），接下来是低层 Loop 的收尾。四个步骤中，后三个都是可选配置；不应据此推断 coding-agent 一定配置了每个钩子：

# ============================================================
# 【Python 改写】turn_end + 钩子 + 再次检查 steering
# 原文 TS:
#   await emit({ type: "turn_end", message, toolResults });
#   const nextTurnSnapshot = await config.prepareNextTurn?.({...});
#   if (nextTurnSnapshot) {
#       currentContext = nextTurnSnapshot.context ?? currentContext;
#       config.model = nextTurnSnapshot.model ?? config.model;
#   }
#   if (await config.shouldStopAfterTurn?.({...})) {
#       await emit({ type: "agent_end", messages: newMessages });
#       return;
#   }
#   pendingMessages = (await config.getSteeringMessages?.()) || [];
# ============================================================

# ① emit turn_end —— 通知外部"这一轮结束了"（内核）
await emit({"type": "turn_end", "message": message, "tool_results": tool_results})

# ② prepare_next_turn —— 给外部一个机会"改装"下一轮（叠加）
# 返回值可包含 context / model / thinkingLevel 三者之一的覆盖
if config.prepare_next_turn:
    next_turn_snapshot = await config.prepare_next_turn({...})
    if next_turn_snapshot:
        current_context = next_turn_snapshot.context or current_context
        config.model    = next_turn_snapshot.model    or config.model
        # thinking_level 也在此处覆盖（详见 agent-loop.ts 中 prepareNextTurn 处理逻辑）

# ③ should_stop_after_turn —— 外部判断是否该停了（叠加）
if config.should_stop_after_turn and await config.should_stop_after_turn({...}):
    await emit({"type": "agent_end", "messages": new_messages})
    return

# ④ 再次检查 steering —— 有没有新的紧急消息？（叠加1）
pending_messages = (await config.get_steering_messages()) if config.get_steering_messages else []

prepareNextTurn —— 它在每个 turn_end 之后、下一轮开始前被调用，允许外部替换下一轮的 context、model 或 thinking level。下面只是接口可以支持的例子，不表示 v0.80.2 会自动按任务复杂度选模型：

场景：按任务复杂度切换模型

Turn 1: 用户问了一个简单问题 → 模型用 Haiku（快、便宜）
        turn_end → prepareNextTurn 检查到问题很简单
        → 返回 { model: haiku } → 下一轮继续用 Haiku

场景：中途发现任务变复杂了

Turn 1: 用户让"重构这个模块" → Haiku 开始读文件
        turn_end → prepareNextTurn 发现要改的文件很多
        → 返回 { model: opus } → 下一轮自动切到 Opus（强、贵）

除了换模型，它还能换 context（比如注入新的上下文信息）和 thinkingLevel（思考强度）。返回 undefined 表示”什么都不换，按原配置继续”。

shouldStopAfterTurn —— 外部注入的优雅停止判断：当前响应和工具执行已经完成，再决定是否在下一次模型调用前退出。接口注释明确要求回调不要抛错；具体检查什么由宿主决定。

边界提醒：v0.80.2 的 runAgentLoop() 接受这个回调，但 Agent 包装类没有把它列入 AgentOptions；coding-agent 的自动压缩也不走这里。这里讲的是低层 Loop 能力。

④再次检查 steering —— 一圈内层循环结束，再看看 steering 队列有没有新积攒的紧急消息。如果有，下一圈继续跑。

4.8 回到循环顶部

工具执行完且收尾结束，代码回到内层 while 循环的条件判断。§三已经详细讲了 hasMoreToolCalls 由 toolCalls 数组长度 > 0 && !terminate 驱动——这里只看代码：

# ============================================================
# 【Python 改写】内层循环条件
# 原文 TS: while (hasMoreToolCalls || pendingMessages.length > 0)
# ============================================================

while has_more_tool_calls or len(pending_messages) > 0:
    ...

两个条件任一为 true 就继续。hasMoreToolCalls 由模型输出里有没有 toolCall 块（且未全部 terminate）决定（内核），pendingMessages 由 steering 队列决定（叠加1）。两个都为 false 时，内层循环退出。

4.9 【叠加2 · 步骤G】外层循环：followUp 的续命机制

内层循环退出了——没有工具要执行，也没有紧急消息。但 coding-agent 的设计是：此时检查一下 getFollowUpMessages()，看看系统有没有追加任务：

# ============================================================
# 【Python 改写】外层 followUp 检查
# 原文 TS:
#   const followUpMessages = (await config.getFollowUpMessages?.()) || [];
#   if (followUpMessages.length > 0) {
#       pendingMessages = followUpMessages;
#       continue;
#   }
#   break;
# ============================================================

follow_up_messages = (await config.get_follow_up_messages()) if config.get_follow_up_messages else []
if len(follow_up_messages) > 0:
    pending_messages = follow_up_messages   # 塞进 pending，触发新 Turn
    continue                                # 回到外层循环顶部
break                                       # 两个队列都空了，真正退出

如果有 followUp 消息，它们被塞进 pendingMessages，continue 回到外层循环顶部，内层循环检测到 pendingMessages 不为空，在同一个 Trace 内继续跑。这比重新启动一个新的 runAgentLoop() 好在哪？——连续性：同一个 newMessages 数组、同一个事件序列，不需要额外合并。

§四开头已经说过：如果你做的是简单 Agent，外层循环是多余的——内层循环退出后直接 break + emit agent_end 就行了。

4.10 steering vs followUp：一张表看清两种干预

本章出现了两种”外部消息注入”机制。虽然 §四开头已经分别介绍过，但放在一起对比更能看清它们的差异：

生活类比：steering 是你正在开会，有人敲门递了张纸条——“紧急，先看这个”。followUp 是开完会翻了翻信箱——“不急，但需要处理”。

放大查看

配图说明：左红右绿对照——steering 在每圈内层循环开头+结尾检查、紧急插队；followUp 在内层循环全部结束后检查、续命重启。底部列出各自时机/来源/影响/典型场景。

五、总结：Loop 的三条核心设计

回顾从按下回车到 Agent 完成的整段旅程：

1. ReAct 循环模式

Loop 的本质是 Reason（模型生成）→ Act（执行工具）→ Observe（把结果写回上下文）→ 再次生成。模型输出决定请求哪些工具，宿主则把“无 toolCall”“硬错误”“terminate”与队列状态解释成继续或停止；这是协议规则与模型输出共同作用的结果。

2. 三类信号共同驱动

error/aborted 负责硬停止；toolCall 与整批 terminate 决定工具链是否继续；steering / follow-up 队列决定是否追加 Turn。把这三类信号分开，比笼统说“由 stopReason 驱动”更准确。

3. 内核 + 叠加的架构思路

最小工具循环确实可以很短；低层 Loop 进一步定义 steering、follow-up、prepareNextTurn 和 shouldStopAfterTurn 等可选协议。Agent 与 coding-agent 选择性地封装、接入这些能力。内核提供机制，宿主决定实际暴露和使用哪些策略。

放大查看

配图说明：从里到外的四层洋葱——最内核是概念化的工具 Agent Loop；外层依次叠加 steering（紧急插队）、followUp（任务追加）和可选的 turn 钩子。图表达职责分层，不代表 coding-agent 把每个低层钩子都直接暴露为产品配置。

六、下一站

Loop 跑起来了——我们知道它怎么调用模型、怎么执行工具。但 Loop 调用的”模型”到底是什么？Pi 是怎么用同一套代码调用 OpenAI、Claude、Gemini 等 30+ 家不同供应商的？streamSimple() 内部做了什么？

下一章，我们拆开 Pi 的模型调用层：《第4章：模型调用 —— 一行代码驾驭多个模型》。

本章关键源码索引（v0.80.2 实际行号）：

• agent-loop.ts:95-118 — runAgentLoop() 入口
• agent-loop.ts:155-269 — runLoop() 双层循环完整实现
• agent-loop.ts:167 — steering 首次检查（进入内层循环之前）
• agent-loop.ts:175-179 — turn_start 的首轮跳过机制
• agent-loop.ts:196-200 — stopReason 硬停止判断
• agent-loop.ts:275-368 — streamAssistantResponse() 流式响应
• agent-loop.ts:313-357 — 流式处理中的原地替换
• agent-loop.ts:373-516 — executeToolCalls() 并行/串行调度
• agent-loop.ts:218-253 — turn_end + prepareNextTurn + shouldStopAfterTurn
• agent-loop.ts:253 — steering 每圈二次检查
• agent.ts:118-152 — PendingMessageQueue 和 drain 模式
• agent.ts:451-474 — runWithLifecycle() 生命周期管理