Pi - Loop

Loop

• loop = outer_loop + inner_loop
• outer_loop允许在inner_loop执行中, 接收getSteeringMessages(),

Main loop logic

/**
 * `agentLoop()` 与 `agentLoopContinue()` 共用的主循环骨架。
 *
 * 可以把它理解成“两层循环”：
 * - 外层循环：agent 原本要停下时，看看 follow-up 队列里有没有“收尾接力”消息；
 * - 内层循环：持续处理 assistant 响应、tool call 批次，以及 turn 之间注入的 steering 消息。
 *
 * 核心不变量：
 * 1. 每次真正请求模型前，`currentContext.messages` 都是一份完整、可见、合法的 transcript；
 * 2. assistant turn 完成后，toolResult 会被补回 transcript，再决定是否下一轮继续；
 * 3. `prepareNextTurn()` 只影响“下一次 provider 请求”，绝不回改已经在飞的请求。
 */
// Outer loop: continues when queued follow-up messages arrive after agent would stop
while (true) {
    let hasMoreToolCalls = true;

    // Inner loop: process tool calls and steering messages
    while (hasMoreToolCalls || pendingMessages.length > 0) {

        // Process pending messages (inject before next assistant response)
        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 = [];
        }

        // ...

        // inner_loop每轮turn结束后检查是否还有pendingMessages
        // executeToolCalls()
        pendingMessages = (await config.getSteeringMessages?.()) || [];
    }

    // Agent would stop here. Check for follow-up messages.
    const followUpMessages = (await config.getFollowUpMessages?.()) || [];
    if (followUpMessages.length > 0) {
        // Set as pending so inner loop processes them
        pendingMessages = followUpMessages;
        continue;
    }

    // No more messages, exit
    break;
}

• 支持 runAgentLoop & runAgentLoopContinue

Agent

• 进入loop前先normalizePromptInput

/**
 * `Agent` 是 low-level loop 之上的有状态包装层。
 *
 * 它的职责不是替代 `runAgentLoop()`，而是把“运行时拥有者”需要承担的状态集中起来：
 * - 持有当前 transcript 与工具列表；
 * - 维护 activeRun / abortController / streamingMessage / pendingToolCalls；
 * - 对外暴露 `prompt()` / `continue()` / `steer()` / `followUp()` 等 API；
 * - 把 low-level loop 的事件归约回本地状态，并转发给订阅者。
 *
 * 可以把它理解成：
 * - `agent-loop.ts` 负责“如何跑一轮循环”；
 * - `agent.ts` 负责“谁拥有这次运行，以及如何把运行折叠回可观察状态”。
 */
export class Agent {}
// `normalizePromptInput()` 负责把多种输入形态统一折叠成标准 `AgentMessage[]`。
const messages = this.normalizePromptInput(input, images);
await this.runPromptMessages(messages);

最外层交互式loop

• TUI层循环接收用户输入

while (true) {
    const userInput = await this.getUserInput();
    try {
        // 里面最终会调用runLoop
        await this.session.prompt(userInput);
    } catch (error: unknown) {
        const errorMessage = error instanceof Error ? error.message : "Unknown error occurred";
        this.showError(errorMessage);
    }
}

Execute Tool Calls

• Sequential or Parallel

/**
 * 执行 assistant message 中携带的 tool calls。
 *
 * 这里先做一次“批次级调度决策”：
 * - 如果全局配置要求顺序执行，走 sequential；
 * - 如果任意单个 tool 声明自己必须顺序执行，也整体降级为 sequential；
 * - 否则才走 parallel。
 *
 * 之所以按“整批降级”处理，而不是部分并行、部分串行混搭，是为了保持批次语义简单，
 * 避免在一个 assistant turn 里出现难以解释的依赖竞态。
 */
async function executeToolCalls(
	currentContext: AgentContext,
	assistantMessage: AssistantMessage,
	config: AgentLoopConfig,
	signal: AbortSignal | undefined,
	emit: AgentEventSink,
): Promise<ExecutedToolCallBatch> {
	// 这里再次从 assistantMessage 中取 toolCall，保证调度逻辑只依赖最终定稿消息，
	// 不依赖 streaming 期间的 partial toolCall 片段。
	const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");
	const hasSequentialToolCall = toolCalls.some(
		(tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential",
	);
	if (config.toolExecution === "sequential" || hasSequentialToolCall) {
		return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit);
	}
	return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit);
}

executeToolCallsParallel

async function executeToolCallsParallel(
	currentContext: AgentContext,
	assistantMessage: AssistantMessage,
	toolCalls: AgentToolCall[],
	config: AgentLoopConfig,
	signal: AbortSignal | undefined,
	emit: AgentEventSink,
): Promise<ExecutedToolCallBatch> {
	// 并行模式下，prepare 阶段仍按 assistant 原顺序串行进行；
	// 真正并发的是 execute + finalize。
	// 这样做的好处是：
	// 1. 参数准备/校验/beforeHook 仍具确定性；
	// 2. 真正耗时的工具执行可以并发；
	// 3. 结果消息最终仍能按原 toolCall 顺序落回 transcript。
	const finalizedCalls: FinalizedToolCallEntry[] = [];

	for (const toolCall of toolCalls) {

		const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal);
        // immediate 直接执行了
		if (preparation.kind === "immediate") {
			const finalized = {
				toolCall,
				result: preparation.result,
				isError: preparation.isError,
			} satisfies FinalizedToolCallOutcome;
			finalizedCalls.push(finalized);
			if (signal?.aborted) {
				break;
			}
			continue;
		}
        // ts 语法糖并发
		finalizedCalls.push(async () => {
			const executed = await executePreparedToolCall(preparation, signal, emit);
			const finalized = await finalizeExecutedToolCall(
				currentContext,
				assistantMessage,
				preparation,
				executed,
				config,
				signal,
			);
			await emitToolExecutionEnd(finalized, emit);
			return finalized;
		});
		if (signal?.aborted) {
			break;
		}
	}

	const orderedFinalizedCalls = await Promise.all(
		finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
	);
	// 注意：`Promise.all` 返回结果顺序与输入数组顺序一致，
	// 因此即使并发执行，toolResult message 仍会按 assistant 原始 toolCall 顺序写回 transcript。
	const messages: ToolResultMessage[] = [];
	for (const finalized of orderedFinalizedCalls) {
		const toolResultMessage = createToolResultMessage(finalized);
		await emitToolResultMessage(toolResultMessage, emit);
		messages.push(toolResultMessage);
	}

	return {
		messages,
		terminate: shouldTerminateToolBatch(orderedFinalizedCalls),
	};
}