第 6 讲Agent Loop 与事件流

async function streamAssistantResponse(context, config, signal, emit, streamFunction) {
  // ① AgentMessage 层：可选的上下文变换（裁剪、注入）——不写回 context.messages
  let messages = context.messages;
  if (config.transformContext) {
    messages = await config.transformContext(messages, signal);
  }

  // ② LLM 层：必需的格式转换（过滤 UI 消息、转换自定义类型）
  const llmMessages = await config.convertToLlm(messages);
  const llmContext = normalizeContext({ messages: llmMessages });

  // ③ 每次请求前动态解析 API key（应对会过期的 OAuth token）
  const resolvedApiKey =
    (config.getApiKey ? await config.getApiKey(config.model.provider) : undefined)
    || config.apiKey;

  const response = await streamFunction(config.model, llmContext, {
    ...config, apiKey: resolvedApiKey,
    identity: mergeClientIdentity(DEFAULT_AGENT_IDENTITY, config.identity),
    signal,
  });
  /* 之后 for await 消费 provider 事件流：start / *_delta / done / error …… */
}

/**
 * The loop passes a normalized transcript: the system prompt and tool
 * declarations are carried by the transcript's system messages, never by
 * `context.systemPrompt` or `context.tools`.
 *
 * Contract:
 * - Must not throw or return a rejected promise for request/model/runtime failures.
 * - Must return an AssistantMessageEventStream.
 * - Failures must be encoded in the returned stream via protocol events and a
 *   final AssistantMessage with stopReason "error" or "aborted" and errorMessage.
 */
export type StreamFn = (
  model: Model<Api>,
  context: TranscriptContext,
  options?: SimpleStreamOptions,
) => AssistantMessageEventStream | Promise<AssistantMessageEventStream>;

private async function handleRunFailure(error: unknown, aborted: boolean): Promise<void> {
  const failureMessage = {
    role: "assistant",
    content: [{ type: "text", text: "" }],
    stopReason: aborted ? "aborted" : "error",
    errorMessage: error instanceof Error ? error.message : String(error),
    /* usage / model 等字段省略 */
  };
  await this.processEvents({ type: "message_start", message: failureMessage });
  await this.processEvents({ type: "message_end",   message: failureMessage });
  await this.processEvents({ type: "turn_end", message: failureMessage, toolResults: [] });
  await this.processEvents({ type: "agent_end", messages: [failureMessage] });
}

private async function processEvents(event: AgentEvent): Promise<void> {
  switch (event.type) {
    case "message_end":
      this._state.streamingMessage = undefined;
      this._state.messages.push(event.message);   // 定稿并入列
      break;
    /* message_start/update、tool_execution_start/end、turn_end、agent_end 等省略 */
  }
  for (const listener of this.listeners) {
    await listener(event, signal);   // 逐个 await：每个事件都是一道 barrier
  }
}

if (context.messages.length === 0) {
  throw new Error("Cannot continue: no messages in context");
}
if (context.messages[context.messages.length - 1].role === "assistant") {
  throw new Error("Cannot continue from message role: assistant");
}
/**
 * **Important:** The last message in context must convert to a `user` or
 * `toolResult` message via `convertToLlm`. If it doesn't, the LLM provider
 * will reject the request.（自定义角色能否续跑由调用方负责——
 * 此处无法校验，因为 convertToLlm 每个 turn 只调用一次。）
 */

selected input events → prepareRequest → provider response
→ tool results → finishTurn → turn_end
→ existing continuation scheduling or agent_end

// 内层循环：还有工具调用或 steering 时继续
while (hasMoreToolCalls || pendingMessages.length > 0) {
  let preparedMessages: AgentMessage[] = [];
  if (lastCompletedTurn) {
    const snapshot = await config.prepareNextTurn?.(lastCompletedTurn); // 可换 context/model/思维档
    if (snapshot) { /* 应用 snapshot 省略 */ }
    // prepareNextTurn 可能耗时很长（例如 compaction）：补 poll steering，
    // 但仅当前一次 poll 为空——否则 one-at-a-time 会在本轮送出两条
    if (pendingMessages.length === 0) {
      pendingMessages = (await config.getSteeringMessages?.()) || [];
    }
    await emit({ type: "turn_start" });
  }

  // 排队 / 预备消息先入列发事件，再准备请求
  for (const message of declareToolChanges(currentContext, [...preparedMessages, ...pendingMessages])) {
    await emit({ type: "message_start", message });
    await emit({ type: "message_end",   message });
    currentContext.messages.push(message); newMessages.push(message);
  }
  pendingMessages = [];

  const requestUpdate = await config.prepareRequest?.({ context: currentContext,
    model: config.model, thinkingLevel: config.reasoning ?? "off" }, signal);
  if (requestUpdate) { /* 替换 context/model/thinkingLevel 省略 */ }

  // ① 流式拿 assistant 消息（消息流水线在此执行，§6.2）
  const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);
  newMessages.push(message);

  // ② error / aborted 是硬出口：finishTurn 照跑，但决策被忽略
  if (message.stopReason === "error" || message.stopReason === "aborted") {
    lastCompletedTurn = { message, toolResults: [], context: currentContext, newMessages };
    await config.finishTurn?.(lastCompletedTurn, signal);
    await emit({ type: "turn_end", message, toolResults: [] });
    await emit({ type: "agent_end", messages: newMessages });
    return;
  }

  // ③ 执行工具批（parallel / sequential / length 截断批量失败）
  const toolCalls = message.content.filter((c) => c.type === "toolCall");
  const toolResults: ToolResultMessage[] = [];
  hasMoreToolCalls = false;
  if (toolCalls.length > 0) {
    const batch = message.stopReason === "length"
      ? await failToolCallsFromTruncatedMessage(toolCalls, emit)   // 参数可能被截断：不执行
      : await executeToolCalls(currentContext, message, config, signal, emit);
    toolResults.push(...batch.messages);
    hasMoreToolCalls = !batch.terminate;    // 全批 terminate 才提前停
    for (const r of toolResults) { currentContext.messages.push(r); newMessages.push(r); }
  }

  // ④ turn 收尾钩子 + turn_end + P3 poll steering
  lastCompletedTurn = { message, toolResults, context: currentContext, newMessages };
  const decision = await config.finishTurn?.(lastCompletedTurn, signal);
  await emit({ type: "turn_end", message, toolResults });
  if (decision?.action === "end") {
    await emit({ type: "agent_end", messages: newMessages });
    return;                                 // 队列不 poll、不消费
  }
  explicitContinuation = decision?.action === "continue";
  pendingMessages = (await config.getSteeringMessages?.()) || [];
  if (hasMoreToolCalls || pendingMessages.length > 0) explicitContinuation = false; // 自然调度优先
}

// 外层循环：本该停止时才查 followUp；否则若 finishTurn 要过 continue，补一次仅上下文请求
const followUpMessages = (await config.getFollowUpMessages?.()) || [];
if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
if (explicitContinuation) { explicitContinuation = false; continue; }
break;
/* 循环退出后 emit agent_end（略） */

import { Agent, type AgentEvent } from "@earendil-works/pi-agent-core";
import { obj, num, str } from "@earendil-works/pi-schema";

const calcParams = obj({ a: num(), b: num(), op: str({ description: "+ - * /" }) });
const calculate = {
  name: "calculate", label: "Calculate",
  description: "计算两个数的四则运算", parameters: calcParams,
  async execute(_id, p) {
    const r = p.op === "+" ? p.a + p.b : p.op === "*" ? p.a * p.b : await Promise.reject(new Error("unsupported"));
    return { content: [{ type: "text", text: String(r) }], details: { r } };
  },
};

const agent = new Agent({
  initialState: { systemPrompt: "你是计算助手，必须用工具计算。", model, tools: [calculate] },
  streamFn: models.streamSimple.bind(models),
  transformContext: async (msgs) => [/* 实验三：在此 prepend 一条备忘消息 */ ...msgs],
  convertToLlm: (msgs) => msgs.flatMap((m) => m.role === "notification" ? [] : [m]),
});

const t0 = Date.now();
const off = agent.subscribe((e: AgentEvent) =>
  console.log(String(Date.now() - t0).padStart(5), e.type));

setTimeout(() => agent.steer({ role: "user",
  content: [{ type: "text", text: "顺便用中文解释结果" }], timestamp: Date.now() }), 300);

await agent.prompt("计算 12*7");
off();