# pi 的 Agent 循环与回合执行器

> pi 的 runLoop 一轮怎么调模型、执行工具、停下来；agent-core 的 PiTurnRunner 与 agent-runtime 又怎样把它包成 MiniMax 运行时里的一个回合。

- 作者：David（道雾轩）
- 专栏：MiniMax Code 源码解读（https://daiw.org/manual/minimax-code.md）
- 最后更新：2026-09-19
- 原文：https://daiw.org/manual/minimax-code/pi-agent
- 转载与引用：请注明出处并附原文链接（https://daiw.org/about/copyright）

MiniMax Code 没有自己写 Agent 循环。每次“用户发一条消息、模型思考、调几个工具、再回答”，底层跑的都是 pi-agent-core 的 `runLoop`。MiniMax 在它外面包了两层：`packages/agent-core` 的 `PiTurnRunner` 把一次 pi 运行变成运行时里的一个回合，负责钩子、事件翻译与收尾；`packages/agent-runtime` 在每个回合开始前，把各扩展贡献的工具、提示词和钩子组装好交给它。

先统一术语。pi 把一次 `turn_start` 到 `turn_end` 定义为一个 assistant 响应加上它的工具调用与结果（`third_party/pi-mono/packages/agent/src/types.ts:420`）；MiniMax 管这个单位叫 step，而把一次完整的 Agent 运行叫 turn，`packages/agent-core/src/pi-turn-runner/hooks.ts:240` 的注释专门交代了这层对应。下文“回合”指 MiniMax 的回合，“一步”指 pi 的 turn。

## 怎么用

循环本身几乎没有直接的开关，用户能碰到的是这些：

| 场景 | 入口 | 在循环里的落点 |
| --- | --- | --- |
| 任务运行中按 `Enter` 发消息 | `README.md:138` | 作为 steering，在下一步请求模型之前注入 |
| `Alt+Enter` 排队后续消息 | `README.md:139` | 交给上层回合系统排队，不走 pi 的 follow-up 队列 |
| `Esc` 中断 | `README.md:144` | 转成 `agent.abort()`，尚未执行的工具调用补一条错误结果 |
| 单次模型请求超时 | `LLM_REQUEST_TIMEOUT_MS`，20 分钟 | `packages/agent-core/src/pi-turn-runner/defaults.ts:21` |
| 传输层重试 | 默认最多重试 5 次，退避基准 1 秒、单次上限 30 秒（带抖动），自首次失败起累计不超过 120 秒；只在模型开始可见输出之前重试 | `packages/agent-core/src/pi-turn-runner/llm-retry.ts:43`、`:458` |
| bash 子进程环境清洗 | 环境变量 `MAVIS_BASH_ENV_SANITIZE` 取 `off`、`scrub`、`strict` | `packages/agent-core/src/bash-subprocess-env.ts:167` |

第一方代码里 `agent.followUp` 只在 after-LLM 钩子的 retry 决策里用到一次（`packages/agent-core/src/pi-turn-runner/events.ts:128`），用户排队的消息走的是[回合系统](https://daiw.org/manual/minimax-code/turn-system)。

## pi 的循环：一步怎么走

核心在 `third_party/pi-mono/packages/agent/src/agent-loop.ts:155` 的 `runLoop`：外层循环处理 follow-up，内层循环只要“还有工具要继续或有待注入的消息”就一直跑。每一步的主体是这段（`agent-loop.ts:193`）：

```ts
const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
newMessages.push(message);

if (message.stopReason === "error" || message.stopReason === "aborted") {
	await emit({ type: "turn_end", message, toolResults: [] });
	await emit({ type: "agent_end", messages: newMessages });
	return;
}

// Check for tool calls
const toolCalls = message.content.filter((c) => c.type === "toolCall");

const toolResults: ToolResultMessage[] = [];
hasMoreToolCalls = false;
if (toolCalls.length > 0) {
	const executedToolBatch = await executeToolCalls(currentContext, message, config, signal, emit);
	toolResults.push(...executedToolBatch.messages);
	hasMoreToolCalls = !executedToolBatch.terminate;

	for (const result of toolResults) {
		currentContext.messages.push(result);
		newMessages.push(result);
	}
}
```

**调模型**。`streamAssistantResponse`（`agent-loop.ts:279`）先用 `transformContext` 在 `AgentMessage` 层面改写上下文，再用 `convertToLlm` 转成 pi-ai 能懂的消息，拼上系统提示词与工具表，交给 `streamFn`（缺省为 `streamSimple`）。API Key 每次调用都经 `getApiKey` 重新取一遍，方便短期 token 在长时间的工具执行后刷新。流的第一个事件就把未完成的 assistant 消息推进上下文，之后每个增量事件都替换它，并向外发 `message_update`。

**执行工具**。`executeToolCalls`（`agent-loop.ts:377`）只要本批里有一个工具声明了 `executionMode: "sequential"`，整批就顺序执行；否则先逐个做准备（找工具、校验参数、跑 `beforeToolCall`），再用 `Promise.all` 并发执行，结束事件按完成先后发出，而写回上下文的 toolResult 消息仍按模型给出的顺序（`agent-loop.ts:574`）。工具不存在、参数校验失败、被钩子拦下、执行抛错，都会变成一条 `isError` 的工具结果，保证每个工具调用都有配对结果。

**停下来**。循环的出口有这几个：

| 条件 | 结果 | 位置 |
| --- | --- | --- |
| 模型响应 `stopReason` 为 `error` 或 `aborted` | 立即结束 | `agent-loop.ts:196` |
| `shouldStopAfterTurn` 返回真 | 这一步收尾后结束，不再轮询 steering | `agent-loop.ts:241` |
| 轮询 steering 之后 `shouldStopAfterSteering` 返回真（MiniMax 补丁） | 结束，不发下一次请求 | `agent-loop.ts:254` |
| 本批结果都带 `terminate`，或有钩子返回 `terminateAgent` | 不再为这批工具结果请求模型，没有待注入消息就结束 | `agent-loop.ts:663` |
| 没有工具要继续、没有 steering、也没有 follow-up | 自然结束 | `agent-loop.ts:261` |

顺序执行时，一旦某个钩子返回 `terminateAgent`，同批剩下的调用直接补一条“被钩子停止”的错误结果，不再执行（`agent-loop.ts:452`）。

`third_party/pi-mono/packages/agent/src/agent.ts:174` 的 `Agent` 类是这个无状态循环的有状态外壳：持有 transcript、工具表与模型，提供 `prompt()`、`continue()`、`steer()`、`steerBatch()`、`followUp()` 和 `abort()`；两个待注入队列默认每次只取一条（`agent/src/agent.ts:226`），工具默认并行（`agent/src/agent.ts:232`）。订阅者按注册顺序被 `await`，`agent_end` 的监听者全部返回后运行才算结束（`agent/src/agent.ts:575`）。运行中抛出的异常由 `handleRunFailure`（`agent/src/agent.ts:498`）合成一条失败的 assistant 消息，照常发完 `message_end`、`turn_end`、`agent_end`。

```mermaid
flowchart TD
  A["prompt：放入用户消息"] --> B["transformContext（before-LLM 钩子）"]
  B --> C["convertToLlm 后经 streamFn 请求模型"]
  C --> D{"stopReason 为 error 或 aborted？"}
  D -- 是 --> Z["agent_end"]
  D -- 否 --> E{"有工具调用？"}
  E -- 有 --> F["beforeToolCall、execute、afterToolCall"]
  E -- 无 --> G["turn_end"]
  F --> G
  G --> H{"shouldStopAfterTurn？"}
  H -- 是 --> Z
  H -- 否 --> I["轮询 steering，再问 shouldStopAfterSteering"]
  I --> J{"工具要继续或有新消息？"}
  J -- 是 --> B
  J -- 否 --> K{"有 follow-up？"}
  K -- 有 --> B
  K -- 无 --> Z
```

## pi 自己怎么用这个循环

pi 的命令行产品在 `Agent` 外面又包了一层 `AgentSession`（`third_party/pi-mono/packages/coding-agent/src/core/agent-session.ts:256`）：订阅事件写 JSONL 会话、派发扩展事件；`_runAgentPrompt`（`agent-session.ts:936`）在 `agent.prompt()` 返回后循环调用 `_handlePostAgentRun`，出错可重试就重试，需要压缩就压缩，还有排队消息就 `agent.continue()`。

MiniMax 完全没有用这层外壳：会话存 SQLite、重试放进流函数、压缩改成 before-LLM 钩子、续跑由宿主判定，这些都由下面的 `PiTurnRunner` 与它的调用方承担。

## PiTurnRunner：一个回合的八个步骤

`packages/agent-core/src/pi-turn-runner/pi-turn-runner.ts:83` 的 `PiTurnRunner` 只持有进程级默认值，每次 `runTurn()` 都新建 Agent、EventBridge、事件队列与历史游标。`runTurn`（`pi-turn-runner.ts:114`）的注释把流程标成八步：

1. `newTurn` 组装本回合状态：合成流函数、工具表、EventBridge、事件编号器。
2. 发出第一条 `session.status`（running）。
3. `newAgent` 创建 pi 的 `Agent`，注入系统提示词、模型、工具与历史，并建立历史游标。
4. 安装工具钩子与 LLM 钩子。
5. 订阅 Agent 事件，所有事件进同一个串行队列。
6. 接好取消：调用方的 `AbortSignal` 触发时转成 `agent.abort()`。
7. `runAgent`：`prompt` 用户消息，或在已有历史上 `continue`，然后等 Agent 空闲。
8. 排空事件队列、补发遗漏的历史、写最终的 `session.status`。

流函数的包装顺序在 `packages/agent-core/src/pi-turn-runner/llm.ts:13`：

```ts
export function composeStreamFn(resolved: LLMModelConfig): StreamFn {
  // Order matters: each wrapper only fills the option it owns, while
  // explicit per-call options from upstream wrappers still win. The host
  // ceiling is the exception and therefore the innermost link: it has to see
  // the fully accumulated options in order to shrink them.
  const callerHeaders =
    resolved.headers && Object.keys(resolved.headers).length > 0 ? resolved.headers : undefined;
  const hostClamped = withHostMaxOutputTokens(resolved.streamFn, resolved.hostMaxOutputTokens);
  const cacheWrapped = withCacheRetention(hostClamped, resolved.cacheRetention);
  const maxTokensWrapped =
    typeof resolved.maxTokens === 'number' && resolved.maxTokens > 0
      ? withMaxTokens(cacheWrapped, resolved.maxTokens)
      : cacheWrapped;
  const headerWrapped = callerHeaders
    ? withHeaders(maxTokensWrapped, callerHeaders)
    : maxTokensWrapped;
  const fetchWrapped = resolved.fetch ? withFetch(headerWrapped, resolved.fetch) : headerWrapped;
  return wrapStreamFnWithTimeout(fetchWrapped, LLM_REQUEST_TIMEOUT_MS);
}
```

`newTurn`（`packages/agent-core/src/pi-turn-runner/turn.ts:127`）再在外面套指标、传输重试（主请求与压缩等辅助请求各包一份）、开发用的上下文捕获，以及 before-LLM 钩子要求“直接回答”时用的合成响应。交给 `Agent` 的配置在 `packages/agent-core/src/pi-turn-runner/agent.ts:78`：

```ts
    streamFn: turn.streamFn,
    getApiKey: (provider) => (provider === turn.llm.model.provider ? turn.llm.apiKey : undefined),
    toolExecution: 'parallel',
    sessionId: turn.input.sessionId,
    ...(turn.input.shouldStopAfterSteering
      ? { shouldStopAfterSteering: turn.input.shouldStopAfterSteering }
      : {}),
    ...(turn.input.shouldStopAfterTurn
      ? { shouldStopAfterTurn: turn.input.shouldStopAfterTurn }
      : {}),
    ...(turn.llm.payloadTransform ? { onPayload: turn.llm.payloadTransform } : {}),
    ...(turn.llm.responseObserver ? { onResponse: turn.llm.responseObserver } : {}),
```

`convertToLlm` 接的是 `projectAgentMessagesForModel`（同文件 `:42`）：先滤掉只给宿主看的消息，再调 pi-coding-agent 的 `convertToLlm` 转换，之后合并同批发送的用户消息、去掉宿主字段、删掉孤立的工具结果，最后过滤过小的图片。`onPayload` 是模型系统生成的请求补丁生效的地方，见[模型系统](https://daiw.org/manual/minimax-code/model-system)。

## 钩子怎样挂进 pi

`PiTurnHooks`（`hooks.ts:252`）有八类钩子，各自挂在 pi 的不同接缝上：

| 钩子 | 挂在哪里 | 能做什么 |
| --- | --- | --- |
| `beforeLlmCallHook` | 改写 `agent.transformContext`（`llm.ts:37`） | 继续、跳过、直接给出回答、只替换本次请求的消息、持久替换历史（压缩）、追加一条隐藏消息、中止 |
| `onLlmCallPreparedHook` | 包在流函数外（`turn.ts:273`） | 只观察最终请求，失败不影响调用 |
| `afterLlmCallHook` | 在 `message_end` 事件里执行（`events.ts:87`） | 替换文本、丢弃响应并重试、判为失败；后两者会先剥掉响应里的工具调用 |
| `beforeToolCallHook` | `agent.beforeToolCall`（`packages/agent-core/src/pi-turn-runner/tools.ts:78`） | 拦下工具，标注是否由权限拦截 |
| `onToolExecutionStartHook` | `agent.onToolExecutionStart`（`tools.ts:97`） | 准入通过、真正执行前的时间点 |
| `afterToolCallHook` | `agent.afterToolCall`（`tools.ts:104`） | 逐个合并对结果的改写 |
| `onHistoryChangedHook` | `message_end` 与 `tool_execution_end` 后刷新历史游标（`events.ts:320`） | 把新增消息交给宿主持久化 |
| `onStepEndHook` | `turn_end` 事件（`events.ts:323`） | 每步结束后的处理，可再排入消息 |

before-LLM 钩子抛异常时，运行时选择“关门”：直接中止这次模型调用（`llm.ts:212`），不带着没处理好的上下文去请求。

## steering 与收尾

每步的 `turn_end` 之后，事件订阅者会向宿主轮询外部消息（`events.ts:195`）：

```ts
      for (;;) {
        // The poll reports whether more steps already follow. At an exit
        // boundary the host may withhold messages so the close below hands
        // them back to the session instead of extending a finished answer.
        const boundary = hasToolCalls || agent.hasQueuedMessages() ? 'mid-turn' : 'exit';
        const steering = [...((await turn.input.getSteeringMessages?.({ boundary })) ?? [])];
        enqueueSteering(agent, steering);
        if (steering.length > 0 || hasToolCalls || agent.hasQueuedMessages()) return;
```

`boundary` 区分“后面还有步”和“这一步已是结尾”。在结尾处，宿主不会把用户的 steering 塞进来延长一个已经答完的回合，而是留着让关闭流程把它交还会话、当成新的请求（`packages/local-runtime-v2/src/service/turn-system/agent-host/execution/user-input-control.ts:49`）。轮询为空时调用 `tryBeginClose` 原子地关闭“继续受理”；关闭被拒说明有新工作抢先到达，就再轮询一次。同批立即发送的用户消息用 `steerBatch` 一次注入（`events.ts:373`），对应台账里 2026-09-08 那条补丁。

Agent 空闲后还有一道 `convergeAtIdle`（`events.ts:222`）：如果这时又冒出 steering，就 `agent.continue()` 再跑，直到关闭成功。最终结局只有三种，`completed`、`aborted`、`failed`（`packages/agent-core/src/event-bridge/types.ts:90`），失败信息会先经错误分类器映射成稳定的状态码。

## EventBridge：从 AgentEvent 到 RuntimeEvent

pi 的事件不直接给界面用。`packages/agent-core/src/event-bridge/bridge.ts:282` 的 `processEvent` 把 `message_*` 与 `tool_execution_*` 翻译成 `stream.resp` 分块（文本、思考、工具调用），把 `agent_*`、`turn_*` 和 `tool_execution_update` 降为 `debug.trace`。两个细节：assistant 消息的 ID 由宿主分配，桥在 `message_start` 时先请求 ID，拿到之前的增量不会发出；带工具调用的完整消息要等所有 `tool_execution_end` 到齐才发（`bridge.ts:174` 的注释解释了原因：否则持久化下来的工具状态永远停在“已开始”）。

`packages/agent-core/src/protocol/` 负责词汇：`packages/agent-core/src/protocol/runtime-event.ts:77` 把 `@mavis/protocol` 里由 thrift 生成的 `IRuntimeEvent` 以旧名字 `RuntimeEvent` 重新导出；`packages/agent-core/src/protocol/agent-event.ts:5` 的注释说明，这里镜像一份 pi 的 `AgentEvent` 类型，是为了让宿主不必直接依赖 pi 也能消费同一套事件形状。

## 工具契约：RuntimeTool 变成 AgentTool

宿主提供的工具是 `RuntimeTool`：LLM 看得见的定义 `def` 与执行实现 `impl` 分开（`packages/agent-core/src/tools/types.ts:142`），可用 `defineRuntimeTool` 或装饰器 `bindTool` 声明。`newTools`（`tools.ts:26`）把它们包成 pi 的 `AgentTool`：重名直接报错；工具没声明 `executionMode` 时按 `sequential` 处理（`tools.ts:140`）；执行时把本回合上下文、`toolCallId` 与 assistant 消息 ID 注入 `impl.execute`。

若宿主没绑定 read、write、edit、bash，`newTools` 会退回 pi 的 `createReadTool` 等内置工具（`tools.ts:44`）。v2 默认关掉了这条退路（`packages/local-runtime-v2/src/service/turn-system/agent-host/runner/local-agent-turn-runner.ts:285`），产品里的这几个工具来自 `agent-tools` 对 pi 工具工厂的再包装，详见[文件与 Shell 工具](https://daiw.org/manual/minimax-code/file-and-shell)。

由于 pi 遇到一个顺序工具就让整批顺序执行，只有同一批里全是声明了 `parallel` 的工具（如 `read`、`grep`、`glob`，`packages/agent-core/src/tools/builtin-defs.ts:24`）才会真正并发。

## 两个小模块

- **`prompt-read.ts`**：宿主无关的提示词模板读取契约。`PromptSnapshotSource` 负责捕获快照、读取某个模板键，`PromptReadScope` 让扩展在一次模型调用里读同一份快照；`BoundedInternalTurnPromptReadRegistry` 在内部回合受理之前暂存已渲染的提示词，容量默认 128（`packages/agent-core/src/prompt-read.ts:59`）。
- **`bash-subprocess-env.ts`**：bash 子进程环境的两层清洗（`bash-subprocess-env.ts:6`）。A 层总是剥掉运行时边界相关的变量，防止子进程冒充父会话；B 层按模式处理凭据：交互默认 `off`，检测到 `CI` 或 `GITHUB_ACTIONS` 时自动 `scrub`，按黑名单删 `ANTHROPIC_API_KEY`、`OPENAI_API_KEY`、`MCODE_PROVIDER_API_KEY` 等；`strict` 则按名字模式删。`createBashEnvSpawnHook` 把它做成 pi `createBashTool` 的 `spawnHook`。

## agent-runtime：每个回合组装扩展

`packages/agent-runtime` 只有约 1400 行，做一件事：让各功能模块以“扩展”的形式挂到回合上。入口 `createAgentRuntime`（`packages/agent-runtime/src/api.ts:16`）先用 `resolveExtensions`（`packages/agent-runtime/src/resolve.ts:21`）按 profile 的 overlay 对 `base` 做禁用、启用、替换，任何对不上的 ID 都在构造时直接抛错；再由 `Registry` 依次调用每个扩展的 `init(pi)`。

`init` 拿到的 `ExtensionAPI` 只在该扩展自己的初始化窗口内有效，窗口外调用 `registerTool`、`on` 之类都会抛错（`packages/agent-runtime/src/registry.ts:307`）。扩展能贡献五类东西：工具（静态或每回合工厂）、系统提示词前缀、用户提示词前缀、提醒（`<system-reminder>` 块，按优先级排序）和钩子。`assembleTurn`（`registry.ts:345`）先同步快照各扩展的启用状态，再逐个收集，钩子按表映射到 agent-core 的字段（`registry.ts:126`）：

```ts
const HOOK_MAPPINGS = [
  { pi: 'onLlmCallPreparedHook', event: 'on_llm_call_prepared', adapter: 'context' },
  { pi: 'beforeLlmCallHook', event: 'before_llm_call', adapter: 'context' },
  { pi: 'afterLlmCallHook', event: 'after_llm_call', adapter: 'context' },
  { pi: 'beforeToolCallHook', event: 'before_tool_call', adapter: 'tool-context' },
  { pi: 'afterToolCallHook', event: 'after_tool_call', adapter: 'tool-context' },
  { pi: 'onHistoryChangedHook', event: 'on_history_changed', adapter: 'context' },
  { pi: 'onStepEndHook', event: 'on_step_end', adapter: 'context' },
] as const satisfies readonly {
```

表里没有 `onToolExecutionStartHook`，这个时间点只能由宿主直接接。`turn_start`、`turn_end` 两类处理器不进 `PiTurnHooks`，以原始数组交给宿主在回合前后调用。`packages/agent-runtime/src/factories.ts:71` 起提供 `createToolExtension`、`createHookExtension`、`createPromptExtension`、`createReminderExtension` 和组合型的 `createExtension`，省掉手写 `init` 的样板。

v2 在 `packages/local-runtime-v2/src/service/turn-system/agent-host/native-production-dependencies.ts:131` 用 `createAgentRuntime({ base: ... })` 建出普通回合用的运行时，挂上会话报告、失控保护、工具输出预算等扩展（`packages/local-runtime-v2/src/service/turn-system/production-composition.ts:515`）；`assembleTurn` 的结果再由宿主拼成 `RunTurnInput`，交给包着 `PiTurnRunner` 的 `LocalAgentTurnRunner`（`local-agent-turn-runner.ts:86`）。回合怎样被受理、排队、调度，见[回合系统](https://daiw.org/manual/minimax-code/turn-system)。

下一篇：[模型协议层 pi-ai](https://daiw.org/manual/minimax-code/pi-ai)——循环里的 `streamSimple` 背后，九种 API 形态怎样统一成同一套流式事件。