Agent CLI 源码剖析 02:主循环与状态机

主循环是 Agent 的心脏

Agent CLI 最核心的代码不是 UI,不是命令列表,也不是某个工具,而是 QueryEngine。

QueryEngine 负责:

接收用户输入。
构建上下文。
调模型。
判断模型输出。
执行工具。
把工具结果回填。
重复直到结束。

最粗的伪代码:

代码块TS · 15 行收起展开

// 这是最粗糙的 Agent loop 伪代码。
// 它表达的是“模型 -> 工具 -> 模型”的循环，不是最终工程写法。
while (!done) {
  // 1. 把当前 messages 和工具列表发给模型。
  const response = await callModel(messages, tools);

  // 2. 如果模型直接回答文本，本轮结束。
  if (response.text) return response.text;

  // 3. 如果模型请求工具，执行工具并把结果回填。
  if (response.toolCall) {
    const result = await executeTool(response.toolCall);
    messages.push(result);
  }
}

但真正工程化时,不能只写成 while。它必须是状态机。

为什么是状态机

因为每一步都有明确状态和失败处理:

text tool deny fail

不用状态机的后果:

错误处理散落各处。
工具失败后不知道要不要重试。
max turns 写在多个地方。
trace 不知道记录哪一步。
用户中断可能留下半状态。

TurnContext

一次用户任务需要一个运行时上下文。

代码块TS · 18 行收起展开

// TurnContext = 一次用户任务的运行时状态。
// 状态机里的每个 state 都会读写这个 ctx。
export type TurnContext = {
  sessionId: string; // 串联 transcript、trace、audit。
  userInput: string; // 用户原始任务。
  messages: Message[]; // 当前模型上下文。

  turn: number; // 当前模型轮次。
  maxTurns: number; // 最大模型轮次。

  toolCallCount: number; // 当前已执行工具次数。
  maxToolCalls: number; // 最大工具调用次数。

  startedAt: number; // 本轮开始时间。
  timeoutMs: number; // 总超时时间。

  abortSignal?: AbortSignal; // 用户取消或系统取消信号。
};

字段含义:

字段	作用
`sessionId`	串联 transcript、trace、audit
`messages`	当前模型上下文
`turn`	模型轮次
`toolCallCount`	工具调用次数
`startedAt`	总耗时控制
`abortSignal`	用户取消

这里体现一个原则:

Agent 的状态不要隐含在局部变量堆里,要结构化保存。

EngineState

代码块TS · 11 行收起展开

// EngineState = QueryEngine 状态机的状态集合。
// 每个状态只携带自己需要的数据。
export type EngineState =
  | { name: "build_context" } // 构建初始 messages。
  | { name: "call_model" } // 调模型。
  | { name: "validate_tool"; call: ToolCall } // 校验工具名和参数。
  | { name: "ask_permission"; call: ToolCall } // 权限确认。
  | { name: "execute_tool"; call: ToolCall } // 执行工具。
  | { name: "append_tool_result"; call: ToolCall; result: ToolResult } // 回填工具结果。
  | { name: "recover"; error: AgentError } // 错误恢复。
  | { name: "done"; reason: StopReason }; // 结束。

这样每个状态携带自己需要的数据。

比如:

validate_tool 必须有 call。
append_tool_result 必须有 call 和 result。
recover 必须有 error。
done 必须有 reason。

这比一个大对象到处传更清楚。

StopReason

代码块TS · 11 行收起展开

// StopReason = 本轮停止原因。
// 用于最终提示、trace、eval，而不是只写 done。
export type StopReason =
  | "final_answer"      // 正常回答完成。
  | "max_turns"         // 模型轮次超限。
  | "max_tool_calls"    // 工具调用超限。
  | "timeout"           // 总耗时超时。
  | "user_abort"        // 用户取消。
  | "permission_denied" // 权限拒绝。
  | "tool_error"        // 工具错误。
  | "model_error";      // 模型错误。

不要只写 done。停止原因会影响用户提示和调试判断。

StopReason	用户该看到什么
`final_answer`	正常结果
`max_turns`	任务太复杂,建议拆分
`permission_denied`	因用户拒绝工具而停止
`timeout`	超时,可重试
`model_error`	模型请求失败

主循环骨架

代码块TS · 34 行收起展开

// query = 状态机主循环。
// 它只负责初始化 ctx、推进 state、产出事件。
export async function* query(userInput: string): AsyncGenerator<AgentEvent> {
  // 初始化本轮上下文。
  const ctx: TurnContext = {
    sessionId: crypto.randomUUID(),
    userInput,
    messages: [],
    turn: 0,
    maxTurns: 8,
    toolCallCount: 0,
    maxToolCalls: 12,
    startedAt: Date.now(),
    timeoutMs: 120_000
  };

  // 初始状态：构建上下文。
  let state: EngineState = { name: "build_context" };

  // 不断推进状态，直到 done。
  while (state.name !== "done") {
    // 把状态暴露成事件，方便 UI 和 trace。
    yield { type: "state", name: state.name };
    await appendTrace(ctx.sessionId, state);

    // 具体状态逻辑放在 step。
    state = await step(state, ctx, event => {
      // 复杂实现里可以把内部事件推到队列
    });
  }

  // done 状态里包含 reason。
  yield { type: "turn_end", reason: state.reason };
}

主循环里不直接写所有逻辑,而是交给 step。

step 函数

代码块TS · 55 行收起展开

// step = 状态转移函数。
// 给定当前 state 和 ctx，返回下一个 state。
async function step(
  state: EngineState,
  ctx: TurnContext,
  emit: (event: AgentEvent) => void
): Promise<EngineState> {
  // 用户取消优先。
  if (ctx.abortSignal?.aborted) {
    return { name: "done", reason: "user_abort" };
  }

  // 总耗时超限。
  if (Date.now() - ctx.startedAt > ctx.timeoutMs) {
    return { name: "done", reason: "timeout" };
  }

  switch (state.name) {
    case "build_context":
      // 构建模型可见上下文。
      ctx.messages = await buildContext(ctx);
      return { name: "call_model" };

    case "call_model":
      // 调模型并判断返回文本还是工具调用。
      return callModelState(ctx, emit);

    case "validate_tool":
      // 工具存在性和参数校验。
      return validateToolState(state.call);

    case "ask_permission":
      // 权限判断。
      return askPermissionState(state.call);

    case "execute_tool":
      // 执行工具。
      return executeToolState(state.call, ctx, emit);

    case "append_tool_result":
      // 工具结果回填 messages，模型下一轮才能看到。
      ctx.messages.push({
        role: "tool",
        toolCallId: state.call.id,
        content: state.result.content
      });
      // 完成一次工具观察后，模型轮次 +1。
      ctx.turn++;
      return { name: "call_model" };

    case "recover":
      // 错误恢复。
      return recoverState(state.error, ctx);
  }
}

这个写法的关键价值:

每个状态可单独测试。
trace 可以记录每次状态转移。
退出条件集中。
错误恢复集中。

callModelState

代码块TS · 43 行收起展开

// callModelState = 调模型状态。
async function callModelState(
  ctx: TurnContext,
  emit: (event: AgentEvent) => void
): Promise<EngineState> {
  // 模型轮次超限就停止。
  if (ctx.turn >= ctx.maxTurns) {
    return { name: "done", reason: "max_turns" };
  }

  // 发起模型请求。
  const response = await callModel({
    messages: ctx.messages,
    tools: getToolSchemas(),
    abortSignal: ctx.abortSignal
  });

  // 记录 token usage。
  if (response.usage) {
    await recordUsage(ctx.sessionId, response.usage);
  }

  // 文本响应：输出并结束。
  if (response.type === "text") {
    emit({ type: "text_delta", text: response.text });
    await appendTranscript(ctx.sessionId, {
      role: "assistant",
      content: response.text
    });
    return { name: "done", reason: "final_answer" };
  }

  // 工具调用响应：进入 validate_tool，而不是直接执行。
  return {
    name: "validate_tool",
    call: {
      id: response.id,
      name: response.name,
      input: response.input,
      sessionId: ctx.sessionId
    }
  };
}

重要判断:

模型返回工具调用后,不立刻执行。先 validate,再 permission,再 execute。

这是安全边界。

validateToolState

代码块TS · 28 行收起展开

// validateToolState = 校验工具调用。
async function validateToolState(call: ToolCall): Promise<EngineState> {
  // 工具必须存在于注册表。
  const tool = findTool(call.name);

  if (!tool) {
    return {
      name: "recover",
      error: new AgentError("TOOL_NOT_FOUND", `Unknown tool: ${call.name}`, call)
    };
  }

  // 模型给的 input 必须符合工具 schema。
  const valid = validateJson(call.input, tool.inputSchema);

  if (!valid.ok) {
    return {
      name: "recover",
      error: new AgentError("TOOL_INPUT_INVALID", valid.message, {
        call,
        schema: tool.inputSchema
      })
    };
  }

  // 校验通过，进入权限判断。
  return { name: "ask_permission", call };
}

为什么工具名错、参数错可以进 recover?

因为模型有时能根据错误提示修正:

代码块TEXT · 1 行收起展开

Tool "read" not found. Available tools: read_file, grep, glob.

下一轮模型可能会改成正确工具。

askPermissionState

代码块TS · 32 行收起展开

// askPermissionState = 权限阶段。
async function askPermissionState(call: ToolCall): Promise<EngineState> {
  // 重新查工具，避免注册表异常或状态不一致。
  const tool = findTool(call.name);
  if (!tool) {
    return {
      name: "recover",
      error: new AgentError("TOOL_NOT_FOUND", `Unknown tool: ${call.name}`)
    };
  }

  // checkPermission 返回 allow/deny/ask。
  const decision = await checkPermission(tool, call.input);

  if (decision.type === "allow") {
    return { name: "execute_tool", call };
  }

  if (decision.type === "deny") {
    return { name: "done", reason: "permission_denied" };
  }

  // ask 表示需要用户明确批准。
  const approved = await askUser(decision.prompt);

  if (!approved) {
    return { name: "done", reason: "permission_denied" };
  }

  // 用户批准后执行工具。
  return { name: "execute_tool", call };
}

权限状态不要藏在具体工具里。否则每个工具各写一套,很快失控。

executeToolState

代码块TS · 52 行收起展开

// executeToolState = 工具执行阶段。
async function executeToolState(
  call: ToolCall,
  ctx: TurnContext,
  emit: (event: AgentEvent) => void
): Promise<EngineState> {
  // 工具调用次数也要限制。
  if (ctx.toolCallCount >= ctx.maxToolCalls) {
    return { name: "done", reason: "max_tool_calls" };
  }

  // 计数 + 发开始事件。
  ctx.toolCallCount++;
  emit({ type: "tool_start", call });

  try {
    // 真正执行工具。
    const result = await executeToolCall(call);

    // 发工具结果事件。
    emit({ type: "tool_result", call, result });

    // 成功审计。
    await appendToolAudit({
      sessionId: ctx.sessionId,
      call,
      result,
      ok: true
    });

    // 成功后进入回填状态。
    return {
      name: "append_tool_result",
      call,
      result
    };
  } catch (error) {
    // 工具失败统一归一化。
    const agentError = normalizeToolError(error, call);

    // 失败也要审计。
    await appendToolAudit({
      sessionId: ctx.sessionId,
      call,
      error: agentError,
      ok: false
    });

    // 交给 recover 判断能否恢复。
    return { name: "recover", error: agentError };
  }
}

注意这里不是工具失败就直接 throw 到顶层。先进入 recover,让恢复策略决定。

recoverState

代码块TS · 32 行收起展开

// recoverState = 错误恢复策略。
function recoverState(error: AgentError, ctx: TurnContext): EngineState {
  // 快到 maxTurns 时不再恢复，直接停止。
  if (ctx.turn >= ctx.maxTurns - 1) {
    return { name: "done", reason: "max_turns" };
  }

  // 参数错误：把错误回填给模型，让模型按 schema 重试。
  if (error.code === "TOOL_INPUT_INVALID") {
    ctx.messages.push({
      role: "tool",
      toolCallId: "validation-error",
      content: `Tool input invalid: ${error.message}. Retry with valid JSON.`
    });
    ctx.turn++;
    return { name: "call_model" };
  }

  // 工具不存在：告诉模型可用工具列表。
  if (error.code === "TOOL_NOT_FOUND") {
    ctx.messages.push({
      role: "tool",
      toolCallId: "tool-not-found",
      content: `Tool not found. Available tools: ${getToolSchemas().map(t => t.name).join(", ")}`
    });
    ctx.turn++;
    return { name: "call_model" };
  }

  // 其他错误默认不可恢复。
  return { name: "done", reason: "tool_error" };
}

可恢复错误和不可恢复错误要分开:

错误	能否恢复	说明
工具名错	能	给工具列表
参数格式错	能	给 schema 错误
文件不存在	可能	让模型先 glob
权限拒绝	不能	用户明确拒绝
危险命令	不能	安全边界
鉴权失败	不能	配置问题

状态机的测试方式

不要只测最终答案。要测状态序列。

代码块TS · 15 行收起展开

// 状态机测试示例：验证一次工具调用链路存在。
it("runs tool then returns final answer", async () => {
  // 收集 query 产出的所有事件。
  const events = await collectEvents(query("read README"));

  // 应该出现工具开始和工具完成。
  expect(events.map(e => e.type)).toContain("tool_start");
  expect(events.map(e => e.type)).toContain("tool_result");

  // 最后应该正常结束。
  expect(last(events)).toEqual({
    type: "turn_end",
    reason: "final_answer"
  });
});

状态机可测,Agent 才可维护。

源码验收门槛

主循环的验收重点不是“能回答一次问题”,而是每个状态转移都有边界、证据和退出条件。

验收项	必须看到的源码证据	不合格表现
状态枚举清晰	`EngineState` 或等价 union 覆盖 model/tool/permission/recover/final	用多个 boolean 拼状态
转移集中	`step(state, ctx)` 或等价调度点决定下一状态	状态跳转散在多个 callback
停止原因细分	`final_answer/max_turns/user_abort/permission_denied/error`	只有 success/fail
工具回填闭环	tool result 写回 messages/transcript 后再 call model	工具执行完直接给用户,不让模型整合
失败恢复有界	可恢复错误重试或改策略,不可恢复错误立即停止	所有异常都继续循环
每轮可观测	turn、state、tool call、usage、error 都进入 trace	debug 只能看终端残留输出

最小回归样例:构造一个模型先返回非法工具参数、再修正参数、最后输出答案的场景。合格状态机应出现 call_model -> validate_tool -> recover -> call_model -> execute_tool -> call_model -> final,并且 turn 数、错误原因、修复后的参数都可在 trace 中检查。

读源码抓手

读 QueryEngine 时,问这些问题:

一次 turn 的状态存在哪里?
最大轮次在哪里限制?
工具调用次数是否单独限制?
工具失败是否可恢复?
用户拒绝权限后怎么停止?
模型 usage 是否记录?
状态变化是否进入 trace?
tool result 是否回填 messages?

如果这些问题答不出来,就还没读懂主循环。