QueryEngine 深入
这一页我们打开 QueryEngine.ts 和 query.ts 的真实源码,一段段贴出来、逐行讲"这在干嘛"。零基础也没关系——先补 3 个 JS 概念,再读代码就顺了。第 1 周(D03/D05)我们从宏观看清了循环这颗"心脏",第 2 周从今天起逐行读它的源码,Day 07 再进"手脚"(工具)。
yield)= 后厨炒好一道就往传送带上放一道,不等整桌做完(这就是"流式");yield*= 三段传送带接力,后厨→传菜口→餐桌,菜原样一路传到你面前;class(QueryEngine 实例)= 你这桌的点菜档案袋,从坐下到结账一直跟着你(消息、花了多少钱都记里面)。记住"边炒边上、接力传菜、一桌一档案袋",今天的源码就有画面了。读代码前,先补 3 个 JS 概念
这一页会出现三个初学者常卡住的语法。先花 3 分钟建立直觉,后面读真实代码就不懵了。
class QueryEngine { ... } 定义了一种对象。new QueryEngine(...) 造出一个实例。类里的字段(如 this.mutableMessages)就是这个实例"自己记着的东西"——只要实例还活着,字段就一直保留。把它想成一个"会话档案袋":一次对话一个袋子,里面装着消息历史、花了多少钱等。return 一次就结束。生成器函数(带 *)能用 yield 吐出很多个值,调用方用 for await (const x of ...) 一个个接。加 async 表示中间能 await(等待网络等异步操作)。这就是"流式"的根:模型每吐一段,yield 一段,界面就实时更新一段,不用等整段说完。yield* other() 意思是"other() 吐什么,我就替它往外吐什么"。Day 03 讲的三层结构就是层层 yield*:API 层吐的被循环层转发、循环层吐的被会话层转发——一段流式数据能一路穿三层直达界面,靠的就是它。async function* count(){ yield 1; yield 2; yield 3; }消费:
for await (const n of count()) print(n) → 依次打印 1、2、3(吐一个、接一个,不是一次性返回数组)。把
1/2/3 换成"模型吐的一段段文字/一次次工具调用",就是 submitMessage 在干的事——所以 UI 能边收边显。三层再定位(这页聚焦前两层)
复习 Day 03 的三层,今天钻进前两层的真实代码:
| 层 | 文件 | 一句话职责 |
|---|---|---|
| 会话层 | QueryEngine.ts(1365 行) | 一次对话一个实例,管消息历史/成本/中断,产出对外消息 |
| 循环层 | query.ts(2057 行) | queryLoop 的 while(true)——真正的 agent 循环 |
| API 层 | services/api/claude.ts(3574 行) | 真打 Anthropic API、流式累积(Day 03 讲过) |
submitMessage 里 for await 消费 → 循环层 query → 循环层里调 callModel → API 层 queryModel。数据反着往上:API 吐 → 循环层转发/处理 → 会话层转成对外消息再吐给 UI。这一页从会话层入口开始,一路读到循环层的心脏。类的字段:一次对话记着什么(真实代码)
打开 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>() // 本轮发现的技能
// ...
submitMessage() = 开一个新 turn;消息历史、文件缓存、用量这些状态跨 turn 一直留着。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
// ...
}
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 就知道"这是个会流式吐消息的方法"。核心:消费循环层的 for-await
做完准备,submitMessage 的心脏是这个 for await(QueryEngine.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...of 遍历一个数组。for await...of 遍历一个异步生成器——每当 query() yield 出一条消息,这个循环就转一圈处理它。因为 query() 是"边跑边吐"(模型说一段吐一段、调一次工具吐一次结果),这个 for await 就实时拿到每一步、实时更新界面。你在终端看到的"逐步冒出的工具行和回答",就是这个循环一圈圈转出来的。注意传进去的 wrappedCanUseTool(:253)——它包了一层权限回调,顺手把"被拒绝的工具"记进 permissionDenials,最后进结果。这就是权限系统(Day 09)接进循环的接口。
消息类型 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)。
userturnCount++——数这是第几轮。
attachment处理结构化输出;碰到 max_turns_reached 就产出 error_max_turns 结果。
system处理 compact_boundary(上下文压缩边界,Day 10)、api_error → 重试。
normalizeMessage 做这层翻译。好处:内部实现可以随便改,只要翻译层不变,外部就感觉不到。这是软件设计里"内外格式分离"的通用做法——你写的库对外暴露稳定接口,内部想怎么重构都行。switch(message.type) = "看这条消息是什么类型,assistant 走这段、stream_event 走那段……"。等价于一串 if/else if,但更清爽。钻进循环层: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 个字段。const { messages, turnCount } = state 意思是"从 state 这个对象里,把 messages 和 turnCount 两个字段取出来,各建一个同名变量"。之后写 messages 就等于 state.messages,少打字、更清爽。while(true) 是"无限循环"——靠里面的 return 跳出(Day 03 讲的各种停止原因)。👶 小白:改一个字段明明更省事,为啥非要每次把 9 个字段的新 state 整个重造一遍?
👨🏫 老师:因为这个循环有 7 个不同的"进下一轮"出口。要是每个出口手动改字段,某天你加了第 10 个字段、却漏在某个出口更新它——就留下上一轮的脏值,冒出诡异 bug 还极难查。改成"每个出口都造一个完整新 state 整体赋值",TypeScript 会强制你 9 个字段全给齐,漏一个直接编译报错。用"整体替换"把"忘记更新某字段"这类 bug 从根上消灭掉。
循环怎么判断"要不要继续"(真实代码)
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 就结束。type:'text',就是它说的话),有的是工具调用块(type:'tool_use',意思是"我要调 Read 工具,参数是这个文件路径")。系统看到 tool_use 块,就去执行对应工具。filter 是数组方法:"只留下满足条件的元素"。依赖注入 & 把结果拼回下一轮(真实代码)
① 依赖注入:为什么能不联网测试
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 接在后面拼成一个新数组"。这里把 历史 + 模型说的"我要调工具" + 工具跑出来的结果 三段拼一起,当作下一轮问模型的输入。tool_result 消息回填(Anthropic 协议要求),模型看到自己上轮说"我要 Read"、紧接着一条"这是文件内容…",就能接着推理。前面一轮结束若模型不再要工具,则走 return {reason:'completed'}(query.ts:1647),这个 turn 才真结束(Day 03/10 讲的各种停止/续命判断都在 while 体里)。今日小结 + 动手
🧠 今天你应该能回答
- 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) → 回答,就是循环转了一圈
Tool.ts 代码逐段讲:一个工具要实现什么、buildTool 的安全默认值、以及"发给模型的说明用 prompt() 不是 description()"这个反直觉点。