Day 06 / 共 20 天 · 第 2 周 核心引擎 · 加深版

QueryEngine 深入

这一页我们打开 QueryEngine.tsquery.ts真实源码,一段段贴出来、逐行讲"这在干嘛"。零基础也没关系——先补 3 个 JS 概念,再读代码就顺了。第 1 周(D03/D05)我们从宏观看清了循环这颗"心脏",第 2 周从今天起逐行读它的源码,Day 07 再进"手脚"(工具)。

📍 你在整门课的位置 · 第 2 周 核心引擎
D06 QueryEngine D07 工具接口 D08 内置工具 D09 权限 D10 上下文/Token
💡 用一个类比先兜住今天(本日世界观:餐厅"边炒边上菜"的传送带) 今天最难的三个 JS 概念,其实就是一条传送带async generator(yield= 后厨炒好一道就往传送带上放一道,不等整桌做完(这就是"流式");yield*= 三段传送带接力,后厨→传菜口→餐桌,菜原样一路传到你面前;class(QueryEngine 实例)= 你这桌的点菜档案袋,从坐下到结账一直跟着你(消息、花了多少钱都记里面)。记住"边炒边上、接力传菜、一桌一档案袋",今天的源码就有画面了。
L01

读代码前,先补 3 个 JS 概念

这一页会出现三个初学者常卡住的语法。先花 3 分钟建立直觉,后面读真实代码就不懵了。

① class(类)——一个"带状态的对象模板" class QueryEngine { ... } 定义了一种对象。new QueryEngine(...) 造出一个实例。类里的字段(如 this.mutableMessages)就是这个实例"自己记着的东西"——只要实例还活着,字段就一直保留。把它想成一个"会话档案袋":一次对话一个袋子,里面装着消息历史、花了多少钱等。
② async function*(异步生成器)——能"边算边一个个吐、还能中途等待" 普通函数 return 一次就结束。生成器函数(带 *)能用 yield 吐出很多个值,调用方用 for await (const x of ...) 一个个接。加 async 表示中间能 await(等待网络等异步操作)。这就是"流式"的根:模型每吐一段,yield 一段,界面就实时更新一段,不用等整段说完。
③ yield*(委托转发)——"把另一个生成器吐的东西,原样转发出去" yield* other() 意思是"other() 吐什么,我就替它往外吐什么"。Day 03 讲的三层结构就是层层 yield*:API 层吐的被循环层转发、循环层吐的被会话层转发——一段流式数据能一路穿三层直达界面,靠的就是它。
📝 举个例子:一个最小的 async generator 长啥样 async function* count(){ yield 1; yield 2; yield 3; }
消费:for await (const n of count()) print(n) → 依次打印 123吐一个、接一个,不是一次性返回数组)。
1/2/3 换成"模型吐的一段段文字/一次次工具调用",就是 submitMessage 在干的事——所以 UI 能边收边显。
记不住没关系。你只要有个印象:class = 会话档案袋async function* + yield = 边算边吐(流式)yield* = 转发。下面看到它们时回来对一眼即可。
L02

三层再定位(这页聚焦前两层)

复习 Day 03 的三层,今天钻进前两层的真实代码:

文件一句话职责
会话层QueryEngine.ts(1365 行)一次对话一个实例,管消息历史/成本/中断,产出对外消息
循环层query.ts(2057 行)queryLoopwhile(true)——真正的 agent 循环
API 层services/api/claude.ts(3574 行)真打 Anthropic API、流式累积(Day 03 讲过)
先记住调用方向:会话层 submitMessagefor await 消费 → 循环层 query → 循环层里调 callModel → API 层 queryModel。数据反着往上:API 吐 → 循环层转发/处理 → 会话层转成对外消息再吐给 UI。这一页从会话层入口开始,一路读到循环层的心脏。
yield* 接力:一段流式数据穿三层直达界面 API 层 后厨queryModel 炒菜 循环层 传菜口query yield* 会话层 餐桌normalize→UI 你👀 每 yield 一段就往右传一段——不必等整桌做完,所以你看到字"边流边冒"
图注:三层 generator 用 yield* 接力,一段段往外传;调用方向是右往左调,数据流是左往右吐。
L03

类的字段:一次对话记着什么(真实代码)

打开 src/QueryEngine.ts,类定义前有一段官方注释,直接点明设计(QueryEngine.ts:183):

/**
 * One QueryEngine per conversation. Each submitMessage() call starts a new
 * turn within the same conversation. State (messages, file cache, usage, etc.)
 * persists across turns.
 */
export class QueryEngine {
  private config: QueryEngineConfig
  private mutableMessages: Message[]        // 跨 turn 持久的完整消息历史
  private abortController: AbortController   // 中断器:按 Esc 就 abort()
  private permissionDenials: SDKPermissionDenial[]  // 记录被拒的工具
  private totalUsage: NonNullableUsage      // 累计 token 用量(算钱用)
  private readFileState: FileStateCache     // 读过哪些文件(去重用,Day 08)
  private discoveredSkillNames = new Set<string>()  // 本轮发现的技能
  // ...
这段注释把整个类的哲学说透了:一次对话 = 一个 QueryEngine 实例你每发一句 = 调一次 submitMessage() = 开一个新 turn消息历史、文件缓存、用量这些状态跨 turn 一直留着
这些字段(this.xxx)是干嘛的? 它们就是 L01 说的"会话档案袋"里装的东西。举例:你聊了 5 轮,mutableMessages 里就攒着这 5 轮的所有消息;你花了 $0.1,totalUsage 里记着;你按 Esc,abortController.abort() 就取消正在进行的请求。因为它们是"实例字段",只要这次对话没结束,它们就一直在——这就是"状态跨 turn 保持"的物理实现。

构造时(:208)这些字段被初始化——注意 mutableMessages 用传入的初始消息(--continue 恢复会话时,Day 16,就是从这灌进来的):

constructor(config: QueryEngineConfig) {
  this.config = config
  this.mutableMessages = config.initialMessages ?? []   // 没有就空数组
  this.abortController = config.abortController ?? createAbortController()
  this.totalUsage = EMPTY_USAGE
  // ...
}
L04

submitMessage 的开头:一个 turn 从这里起

核心方法 async *submitMessage(...)QueryEngine.ts:217)。注意签名里的 *——它是 L01 说的异步生成器,所以 UI 能 for await 实时拿到它吐出的每条消息。开头先取配置、清理本轮状态:

async *submitMessage(
  prompt: string | ContentBlockParam[],
  options?: { uuid?: string; isMeta?: boolean },
): AsyncGenerator<SDKMessage, void, unknown> {
  const { cwd, commands, tools, mcpClients, maxTurns, maxBudgetUsd,
          taskBudget, canUseTool, ... } = this.config   // 取出这次要用的配置

  this.discoveredSkillNames.clear()   // 清空"本轮发现的技能"(防跨 turn 累积)
  this.permissionDenials = []         // 清空"被拒工具"记录
  setCwd(cwd)                         // 设当前工作目录
  // ...
AsyncGenerator<SDKMessage, void, unknown> 这个类型标注读法:它会一个个吐出 SDKMessage(对外消息),最终返回 void(没有返回值),不接收外部传入值。你不用记类型细节,看到 async * + yield 就知道"这是个会流式吐消息的方法"。
为什么每轮开头要 clear()? 有些状态是"本轮专属"的——比如"这一轮里发现了哪些技能"。如果不清空,聊 100 轮就累积 100 轮的脏数据(内存泄漏 + 逻辑错乱)。"跨 turn 保持的"(消息、用量)留着,"本轮专属的"(技能发现、被拒记录)每轮清零——这个区分很重要,是写有状态循环的常见讲究。
L05

核心:消费循环层的 for-await

做完准备,submitMessage 的心脏是这个 for awaitQueryEngine.ts:688把循环层 query() 吐出的每一条消息接过来处理:

for await (const message of query({
  messages,                    // 当前完整消息历史
  systemPrompt,                // 系统提示(告诉模型它是谁、有什么工具)
  canUseTool: wrappedCanUseTool,  // 权限回调(Day 09)
  toolUseContext,
  maxTurns,                    // 最大回合数上限
  taskBudget,                  // token 预算
})) {
  switch (message.type) {      // ← 见 L06:按消息类型分类处理
    case 'assistant': ...
    case 'stream_event': ...
    // ...
  }
}
for await 是什么? 普通 for...of 遍历一个数组。for await...of 遍历一个异步生成器——每当 query() yield 出一条消息,这个循环就转一圈处理它。因为 query() 是"边跑边吐"(模型说一段吐一段、调一次工具吐一次结果),这个 for await实时拿到每一步、实时更新界面。你在终端看到的"逐步冒出的工具行和回答",就是这个循环一圈圈转出来的。

注意传进去的 wrappedCanUseTool:253)——它包了一层权限回调,顺手把"被拒绝的工具"记进 permissionDenials,最后进结果。这就是权限系统(Day 09)接进循环的接口

L06

消息类型 switch:把"内部消息"翻译成"对外消息"

for await 的循环体是一个大 switch(message.type)。循环层吐出的消息有好几种类型,各自处理(QueryEngine.ts:772 起):

assistant

模型的回复。记下 stop_reason、push 进 mutableMessages(存进档案袋)、yield* normalizeMessage(msg) 把内部格式转成对外 SDK 消息再吐给 UI。

stream_event

原始流式片段。用它维护 usage/成本:message_start 重置本条、message_delta 累加、message_stop 把本条累进 totalUsage。仅当调用方要"打字机效果"时才把原始片段透传(Day 04)。

user

turnCount++——数这是第几轮。

attachment

处理结构化输出;碰到 max_turns_reached 就产出 error_max_turns 结果。

system

处理 compact_boundary(上下文压缩边界,Day 10)、api_error → 重试。

为什么要"内部消息 → 对外 SDK 消息"翻译(normalizeMessage)? 循环层内部的消息格式是"干活用的",带一堆中间状态;而给 UI/给外部调用者的,需要一个干净稳定的格式。normalizeMessage 做这层翻译。好处:内部实现可以随便改,只要翻译层不变,外部就感觉不到。这是软件设计里"内外格式分离"的通用做法——你写的库对外暴露稳定接口,内部想怎么重构都行。
switch 是什么? 就是"根据一个值,分几种情况走不同分支"。switch(message.type) = "看这条消息是什么类型,assistant 走这段、stream_event 走那段……"。等价于一串 if/else if,但更清爽。
L07

钻进循环层:queryLoop 的 State(真实代码)

现在从会话层进到循环层 src/query.ts。真正的循环是 queryLoop:393)。它开头定义一个可变的 State,并有一段极好的注释解释设计(query.ts:418):

// Mutable cross-iteration state. The loop body destructures this at the top
// of each iteration so reads stay bare-name (`messages`, `toolUseContext`).
// Continue sites write `state = { ... }` instead of 9 separate assignments.
let state: State = {
  messages: params.messages,
  toolUseContext: params.toolUseContext,
  turnCount: 1,
  maxOutputTokensRecoveryCount: 0,
  hasAttemptedReactiveCompact: false,
  stopHookActive: undefined,
  // ... 共约 9 个字段
}
// ...
while (true) {                          // ← 这就是 agent 循环的心脏
  let { toolUseContext } = state        // 每轮开头把 state 解构出来
  const { messages, turnCount, ... } = state
  // ... 一轮的全部逻辑
}
注释翻译:这是"跨轮次的可变状态"。每轮开头把它解构成裸变量(messages 而不是 state.messages),读起来干净。而每个"继续下一轮"的地方,用 state = {...} 整体替换,而不是分别改 9 个字段。
为什么"整体替换"而非"分别改 9 个字段"?(重要设计) 这个循环里有 7 个不同的"继续下一轮"的出口(对应 7 种继续原因)。如果每个出口都手动改 state 的 9 个字段,极易漏改一个、留下上一轮的脏值 → 诡异 bug。改成"每个出口都构造一个完整的新 state 整体赋值",就不可能漏——要么 9 个全给、要么 TypeScript 编译报错。这是用"不可变整体替换"消灭"部分更新遗漏"这类 bug 的经典手法,Day 18 的 Cursor 编辑也是同一思路。
解构(destructure)是什么? const { messages, turnCount } = state 意思是"从 state 这个对象里,把 messages 和 turnCount 两个字段取出来,各建一个同名变量"。之后写 messages 就等于 state.messages,少打字、更清爽。while(true) 是"无限循环"——靠里面的 return 跳出(Day 03 讲的各种停止原因)。

👶 小白:改一个字段明明更省事,为啥非要每次把 9 个字段的新 state 整个重造一遍?

👨‍🏫 老师:因为这个循环有 7 个不同的"进下一轮"出口。要是每个出口手动改字段,某天你加了第 10 个字段、却漏在某个出口更新它——就留下上一轮的脏值,冒出诡异 bug 还极难查。改成"每个出口都造一个完整新 state 整体赋值",TypeScript 会强制你 9 个字段全给齐,漏一个直接编译报错。用"整体替换"把"忘记更新某字段"这类 bug 从根上消灭掉。

L08

循环怎么判断"要不要继续"(真实代码)

Day 03 讲过判据。这是真实代码query.ts:1079)——每收到一条 assistant 消息,看它里面有没有 tool_use 块:

if (message.type === 'assistant') {
  const assistantMessage = message as AssistantMessage
  assistantMessages.push(assistantMessage)

  // 从这条回复的 content 里,筛出所有 type === 'tool_use' 的块
  const msgToolUseBlocks = (
    Array.isArray(assistantMessage.message?.content)
      ? assistantMessage.message.content : []
  ).filter(content => content.type === 'tool_use') as ToolUseBlock[]

  if (msgToolUseBlocks.length > 0) {   // 只要有工具调用块
    toolUseBlocks.push(...msgToolUseBlocks)
    needsFollowUp = true               // ← 唯一的"必须再来一轮"信号
  }
}
.filter(content => content.type === 'tool_use') = "从模型这段回复的所有内容块里,挑出所有'我要调工具'的块"。只要挑出至少一个,needsFollowUp = true——意思是"模型还想调工具,这一 turn 没完,得把工具跑了、结果喂回去再来一轮"。
为什么不看 stop_reason 字段而看"有没有 tool_use 块"? 因为流式响应里 stop_reason 可能不准或滞后(源码注释在 query.ts:751 附近明说不可靠)。而"这条回复里到底有没有工具调用块"是确定的事实用事实判断,不信可能出错的信号——这是很典型的健壮性设计。一句话记住:模型只要还在要工具,循环就继续;一旦只给文字不要工具了,这 turn 就结束。
tool_use 块是什么? 模型的一条回复是由若干"内容块"拼成的。有的是文字块(type:'text',就是它说的话),有的是工具调用块(type:'tool_use',意思是"我要调 Read 工具,参数是这个文件路径")。系统看到 tool_use 块,就去执行对应工具。filter 是数组方法:"只留下满足条件的元素"。
L09

依赖注入 & 把结果拼回下一轮(真实代码)

① 依赖注入:为什么能不联网测试

queryLoop 开头有一行(query.ts:416):

const deps = params.deps ?? productionDeps()   // 没传 deps 就用生产实现
?? 是"空值合并"——左边有值就用左边,没有(null/undefined)才用右边。这里:如果外部传了 deps(假的模型调用),就用传的;没传就用 productionDeps()(真调 Anthropic API)。
这就是"依赖注入",也是测试能不联网的关键。 productionDeps() 里的 callModel 会真打 API。但测试时可以传一个假的 deps,其中 callModel 返回预设的固定响应——于是能测整个循环逻辑而不真调 API、不花钱、结果每次一样。这跟 gov-agents 教程的 FakeLLM 是完全一样的思想:把"外部依赖"抽成一个可替换的参数,测试时换成假的。好的可测试性是架构时就留好注入点,不是事后补的。

② 拼回下一轮:循环怎么"转起来"

一轮结束、准备进下一轮时,把三样东西拼成新的消息列表(query.ts:2043,真实代码):

const next: State = {
  messages: messagesForQuery.concat(assistantMessages, toolResults),
  //         历史消息          + 模型这轮的回复  + 工具执行结果
  toolUseContext: toolUseContextWithQueryTracking,
  turnCount: nextTurnCount,
  // ... 其余字段一次性给全(L07 讲的"整体替换")
  transition: { reason: 'next_turn' },
}
state = next   // 整体替换 → while 回到顶,下一轮 callModel 就带着工具结果再问模型
.concat(a, b) = "把数组 a、b 接在后面拼成一个新数组"。这里把 历史 + 模型说的"我要调工具" + 工具跑出来的结果 三段拼一起,当作下一轮问模型的输入。
这就是循环能"转"的本质:不断把"模型的动作"和"世界的反馈"拼进对话,让模型基于最新事实继续想下一步。工具结果以 user 角色tool_result 消息回填(Anthropic 协议要求),模型看到自己上轮说"我要 Read"、紧接着一条"这是文件内容…",就能接着推理。前面一轮结束若模型不再要工具,则走 return {reason:'completed'}query.ts:1647),这个 turn 才真结束(Day 03/10 讲的各种停止/续命判断都在 while 体里)。
L10

今日小结 + 动手

🧠 今天你应该能回答

  • class / async function* / yield* 各是什么?(会话档案袋 / 边算边吐 / 转发)
  • 为什么"一会话一 QueryEngine 实例"?跨 turn 保持了什么?(消息/用量/文件缓存)
  • submitMessage 为什么是 async generator?带来什么?(UI 能实时拿每条消息)
  • 消息 switch 为什么要"内部消息 → SDK 消息"翻译?
  • queryLoop 为什么"整体替换 State"而非分别改字段?(7 个出口防漏改)
  • 循环靠什么判断继续?(有没有 tool_use 块,不信 stop_reason)
  • 依赖注入 deps 怎么让循环不联网可测?

✋ 动手:对着真实代码读一遍

# 1. 读类定义 + 设计注释(本页 L03)
sed -n '183,215p' src/QueryEngine.ts

# 2. 读 submitMessage 开头(L04)
sed -n '217,250p' src/QueryEngine.ts

# 3. 读循环层 State 与 while(L07,注意那段注释)
sed -n '418,472p' src/query.ts

# 4. 读检测 tool_use(L08,核心判据)
sed -n '1079,1093p' src/query.ts

# 5. 读拼回下一轮(L09)
sed -n '2042,2056p' src/query.ts

# 6. 亲手验:dev 模式发一句"读 package.json 告诉我版本"
bun run dev   # 观察它 ● Read(package.json) → 回答,就是循环转了一圈
明天预告 · Day 07:循环的"大脑"钻透了。Day 07 进"手脚"——工具系统,同样会贴真实的 Tool.ts 代码逐段讲:一个工具要实现什么、buildTool 的安全默认值、以及"发给模型的说明用 prompt() 不是 description()"这个反直觉点。

← Day 05 完整旅程 Day 07 · 工具系统 Tool 接口 →