查询引擎 — while(true) 的力量

整个 Agent 系统的心脏就是一个 while 循环

📝 本章目标

读完本章，你将：

  1. 用三个 Prompt 让 AI 帮你构建一个可运行的查询引擎
  2. 理解自己刚刚构建的代码——while 循环为什么是 Agent 的心脏
  3. 了解流式处理、上下文压缩、模型降级背后的设计思想

想象你在指挥一个实习生完成一项复杂任务——比如”帮我把项目里所有的 TODO 注释整理成一份清单”。

你不会只说一句话就走开。你会看着他开始做，看到他读完第一个文件后问”接下来呢？“，你说”继续读下一个”。他可能读到一半发现有个子目录没权限，你帮他解决，然后他继续。最后他把清单交给你，你确认没问题，任务结束。

这个过程的核心是什么？一个循环——不断检查”完了没”，没完就继续。

查询引擎就是这个循环的程序化实现。别急着理解原理，我们先把它造出来——等手里有了能跑的代码，再回头看就清楚了。

动手：用三个 Prompt 构建查询引擎

你只需要两样东西：一台装了 Python 的电脑，和一个 Anthropic API Key。

如果你跟着第 1 章做了项目骨架，现在应该有一个 harness/ 目录。如果没有，去 GitHub 仓库 git checkout ch01-skeleton 获取起点。

打开 Claude Code，确认你在项目根目录（有 pyproject.toml 的那个目录），然后跟着走。

Prompt 1：让 AI 能对话

复制下面这段话，粘贴到 Claude Code 里：

帮我实现一个查询引擎，核心逻辑是这样的：
把用户说的话发给 Claude，Claude 可能回两种东西——
一种是正常的文字回答，一种是说"我要调用某个工具"。
如果是文字回答，说明事情做完了，把回答显示出来。
如果要调工具，就执行那个工具，把结果告诉 Claude，
然后继续问它下一步怎么办。
就这样一直循环，直到 Claude 觉得做完了为止。

工具执行的部分先占个位，后面再实现真的。
每轮帮我记一下用了多少 token，
再加一个 /cost 命令能随时看花了多少。

等 AI 跑完，它会帮你创建查询引擎的代码，并把命令行界面接上去。试一下：

$ export ANTHROPIC_API_KEY=sk-ant-你的密钥
$ pip install -e .
$ harness
Harness v0.1.0 — AI Agent Runtime
Type your message, or 'exit' to quit.

You > 你好
Assistant > 你好！有什么我可以帮助你的吗？
You > /cost
[tokens: ~150 | turns: 1 | tool calls: 0]

能收到回复就说明引擎跑起来了。但你会发现回答是等几秒后一次性蹦出来的——下一个 Prompt 解决这个问题。

Prompt 2：让回答逐字蹦出来

帮我改成像 ChatGPT 那样，一个字一个字蹦出来，
边生成边显示。其他功能不要动，就改这个显示方式。

跑完后再试：

$ harness
You > 给我讲个笑话
Assistant > 好的，给你讲一个程序员笑话...
（文字一个字一个字蹦出来，不再是干等几秒后一次性显示）

体验好多了。但还有一个问题：聊太久了对话历史会越来越长，最终超出 AI 的上下文窗口限制就崩了。

Prompt 3：聊再久也不怕

帮我解决对话越聊越长的问题，需要这几个能力：

1. 聊到一定长度后自动压缩——让 AI 自己总结之前聊了什么，
   用摘要替换掉旧对话，只保留最近几轮。
   这样上下文一下就短了，但关键信息还在。

2. 如果压缩了好几次还是太长，就别无限重试了，
   直接告诉用户"太长了，建议开个新会话"。

3. 如果某次工具返回的结果特别长，自动截断一下，
   别让一次返回就把上下文撑爆。

4. API 偶尔会调用失败，帮我加个自动重试，
   每次等久一点再试，别一失败就放弃。

再加一个 /compact 命令，让用户可以手动触发压缩。
$ harness
（跟 AI 聊很多轮...突然看到：）
[compact] ~82000 tokens, compressing...
[compact] → ~3200 tokens
（继续聊，AI 还记得之前聊了什么）

You > /compact
[compact] 12500 → 2800 tokens

You > /cost
[tokens: ~45000 | turns: 25 | tool calls: 0]

💡 三个 Prompt 做了什么

  • Prompt 1 建立了骨架——能对话了
  • Prompt 2 加上了体验——响应逐字蹦出来了
  • Prompt 3 解决了续航——聊再久也不会爆

现在你手里有了一个能用的查询引擎。完整代码在 GitHub 仓库，对应 tag ch02-query-engine。

接下来我们回过头，理解你刚刚构建的东西。

深入理解

理解查询循环：20 行核心代码

打开你刚生成的 harness/engine.py，找到 query_loop 函数。把所有辅助代码去掉，它的骨架就是下面这 20 行：

# harness/engine.py — 核心骨架
def query_loop(state):
    client = anthropic.Anthropic()

    while not state.abort:
        # 1. 调用 LLM
        response = client.messages.create(
            model=state.model,
            messages=state.messages,
            tools=state.tools,
        )

        # 2. 检查：AI 是想"说话"还是想"做事"？
        tool_blocks = [b for b in response.content
                       if b.type == "tool_use"]

        # 3. 只想说话 → 任务完成，退出
        if not tool_blocks:
            return extract_text(response)

        # 4. 想做事 → 执行工具，把结果告诉 AI，继续转
        for block in tool_blocks:
            result = execute_tool(block.name, block.input)
            state.messages.append(tool_result(block.id, result))

这就是全书最重要的代码片段。后续所有章节——工具系统、权限系统、Agent 编排——都是在这个骨架上增加能力。

💡 核心概念：查询循环（Query Loop）

查询循环的本质是一句话：

“不断问 AI 下一步要做什么，执行它说的操作，把结果告诉它，直到它说’我做完了’。”

退出条件：AI 的响应里没有 tool_use 块——说明它认为任务完成了。就像实习生放下笔说”好了”——你知道活干完了。

循环的四个阶段

你生成的 query_loop 比骨架版多了很多代码。那些代码在做什么？其实每一轮循环可以分成四个阶段：

图 2-1：查询循环的四个阶段

  1. Setup → 准备与压缩
  2. API Call → 模型调用
  3. Execution → 工具执行
  4. Decision → 继续判断
  • Setup——循环开始前，检查上下文是不是太长了，需不需要压缩。这就是你在 Prompt 3 里加的 estimate_tokens() + compact_context() 做的事
  • API Call——把消息发给 Claude。你的代码里用了 client.messages.stream()（流式版）或 client.messages.create()（同步版），外面包了一层 retry_with_backoff() 做失败重试
  • Execution——解析返回的 tool_use 块，执行工具。目前 execute_tool() 是占位函数，第 3 章会实现真的工具
  • Decision——有工具调用就继续转，没有就退出。除了这个”自然退出”，还有两种退出方式：

❗ 三种退出条件

  1. 自然退出——AI 不再调用工具，说明任务完成
  2. 用户中断——Ctrl+C，通过 state.abort 信号终止
  3. 资源耗尽——连续压缩失败触发熔断器（你代码里的 MAX_COMPACT_RETRIES = 3）

为什么用 QueryState 而不是普通参数

你可能注意到，所有状态都塞在一个 QueryState dataclass 里，而不是当作函数参数传来传去。为什么？

看看你的 QueryState：

# harness/engine.py — 你生成的 State
@dataclass
class QueryState:
    messages: list[dict] = field(default_factory=list)
    system: str = "You are a helpful assistant."
    model: str = DEFAULT_MODEL
    tools: list[dict] = field(default_factory=list)
    token_usage: int = 0
    tool_call_count: int = 0
    turns: int = 0
    compact_count: int = 0
    abort: bool = False

原因很实际：消息历史可能有几十 MB（包含大量工具结果），每轮复制一次内存开销太大。用一个可变对象原地修改，是工程上最高效的选择。

流式处理的工作原理

你在 Prompt 2 里加的流式输出，让回答”逐字蹦出来”。这背后发生了什么？

当开启流式模式时，API 不是等生成完再返回一个大 JSON，而是边生成边发送一系列事件：

图 2-2：流式事件序列

  1. message_start → 消息开始，返回 message_id
  2. content_block_start → 一个内容块开始（文本 或 tool_use）
  3. content_block_delta → 内容增量——就是那些逐个蹦出来的字
  4. content_block_stop → 一个内容块结束
  5. message_stop → 消息结束，返回 usage 统计

你代码里的这一行就是”逐字蹦出来”的全部秘密：

print(event.delta.text, end="", flush=True)

end="" 不换行，flush=True 立即刷新到终端。没有任何黑魔法。

Claude Code 的进阶优化：流式工具执行

我们的实现是等流结束后才检查工具调用。Claude Code 做了一个更激进的优化——StreamingToolExecutor：

💡 进阶：StreamingToolExecutor

有些工具不需要等完整的 JSON 参数才能开始执行：

  • FileRead——一收到文件路径就能开始读
  • Bash——命令字符串到齐就能执行

StreamingToolExecutor 监听 JSON 片段，关键参数到齐就提前启动工具，与剩余的流式输出并行进行。这在长回复中能显著减少等待时间。

我们的简化版暂时不需要这个——等第 3 章有了真工具再考虑。

断流恢复：Tombstone 机制

网络不稳定时流可能中途断开。Claude Code 用 tombstone（墓碑）机制处理：把已收到的不完整内容标记为”死在了这里”，重试时模型看到 tombstone 就知道要从断点继续，而不是重复已做过的事。

模型降级与重试

你在 Prompt 3 里加了 retry_with_backoff()——API 调用失败时自动重试。这是最基本的容错。Claude Code 还有一个更巧妙的机制：模型降级。

图 2-3：模型降级流程

  1. 请求 Opus → 发送
  2. 529 错误（容量不足）→ 自动降级
  3. 降级到 Sonnet → 通知用户已降级

Anthropic API 的 529 错误码表示”系统过载”——你的请求没问题，只是服务端暂时扛不住。Claude Code 会自动切换到备选模型（比如从 Opus 降到 Sonnet），用户只看到一条通知。主模型恢复后自动切回。

⚠️ 降级的代价

Sonnet 比 Opus 快但能力弱，复杂推理任务质量可能下降。所以：

  • 降级只在瞬时过载时触发
  • 用户可以配置”宁可等也不降级”

上下文压缩：四种策略

你在 Prompt 3 里实现了最基本的压缩——超过阈值就让 AI 做摘要。Claude Code 的做法更精细，它有四种压缩策略，按侵入性从低到高排列：

图 2-4：四种压缩策略的层级关系

  1. 微压缩 — 去重、瘦身、零损耗 → 搞不定时升级
  2. 上下文折叠 — 渐进折叠、保留摘要 → 搞不定时升级
  3. 自动压缩 — ~87K 触发、LLM 摘要、恢复文件 → 搞不定时升级
  4. 响应式压缩 — 413 错误、紧急恢复、最后手段

策略一：微压缩（Micro-compact）

触发： 每次工具执行完自动运行。 做什么： 对工具结果做”瘦身”——去掉重复内容、截断大段输出。

这就是你代码里的 micro_compact() 函数。它的原则是零损耗——只处理明显冗余，不丢失语义。

策略二：上下文折叠（Context Collapse）

触发： 消息历史超过窗口的 60%。 做什么： 按”距今远近”渐进折叠——越久远的对话，保留的细节越少：

  1. 最近 5 轮——完整保留，一个字不动
  2. 5–20 轮前——保留用户消息和最终回复，折叠中间的工具调用
  3. 20 轮以前——只保留每轮一句话摘要

这借鉴了人类记忆：昨天的事记得清楚，上周的大概记得，上个月的只记结论。

策略三：自动压缩（Auto-compact）

触发： 约 87K token。 做什么： 用 LLM 自身对对话历史做摘要，然后用摘要替换原始历史。

这就是你的 compact_context() 做的事。但 Claude Code 多了一个关键步骤——恢复注入：

💡 压缩后恢复注入

压缩后 AI 会”忘记”之前读过的文件内容。Claude Code 会自动把最近操作过的 5 个文件重新读入上下文，确保 AI 不会”断片”。

摘要也不是自由文本，而是有固定结构：对话主题、关键决策、当前进展、活跃文件、未完成事项。结构化确保压缩后的信息是可操作的。

策略四：响应式压缩（Reactive-compact）

触发： API 返回 413 错误（请求体太大）。 做什么： 最后一道防线——激进删除所有非关键内容，只保留摘要 + 最近 3 轮。

熔断器

你代码里的 MAX_COMPACT_RETRIES = 3 就是熔断器——连续压缩 3 次还超限，说明真的搞不定了，报错让用户开新会话。有限度地重试，不无限制浪费资源。

策略触发条件核心动作侵入性
微压缩每次工具执行后去重、截断零
上下文折叠>60% 窗口渐进折叠旧对话低
自动压缩~87K tokenLLM 摘要 + 恢复注入中
响应式压缩413 错误激进删除高

表 2-1：四种压缩策略对比

Token 计数与系统提示

精确值 vs 估算值

你代码里的 estimate_tokens() 用的是”4 字符 ≈ 1 token”的粗算法。为什么不用精确值？因为压缩决策必须在发请求之前做——你得先知道消息有多长，才能决定要不要压缩。精确值要等 API 返回 usage 才有，那时候已经晚了。

方式精确计数估算计数
来源API 返回的 usage本地：4 字符 ≈ 1 token
时机响应返回后请求发送前
用途成本追踪（/cost 显示的）决定是否压缩
准确度100%±10%

表 2-2：Token 计数的两种方式

系统提示的组装顺序

你的代码里 system 是一个简单字符串。Claude Code 的系统提示则是动态组装的，关键原则是稳定的内容在前，动态的在后：

图 2-5：系统提示的分层结构

  1. 基础指令 — 角色定义、行为规范 → 几乎不变 → 可缓存
  2. 工具 Schema — 工具定义、JSON Schema → 会话内不变 → 可缓存
  3. 缓存边界 — DYNAMIC_BOUNDARY → 以下每轮变化
  4. 用户上下文 — 项目指令、Git 状态
  5. 记忆 + 日期 — 跨会话记忆、环境信息

这样做的好处是 Prompt Cache 优化：边界线之前的内容不变，API 会缓存它们，后续请求只为动态部分付费。在 50 轮工具调用的会话中，这能省 60–80% 的输入 token 成本。

Prompt Cache 的工作原理

Prompt Cache 不是你主动开启的功能，而是 API 的自动行为——只要连续两次请求的前缀相同，第二次就能复用缓存。关键是你要保证前缀尽可能长且稳定。

Claude Code 的系统提示因此严格遵循”不变在前、变化在后”：基础指令（~3000 token）→ 工具 Schema（~8000 token）→ 缓存边界标记 → 动态上下文。前面约 11K token 在整个会话期间一字不变，每次 API 调用都能命中缓存。

缓存命中时输入价格降低 90%。一个 50 轮的会话，如果每轮发送 11K 缓存前缀 + 5K 动态内容，总共 550K 缓存 token 只按 55K 计费——这就是系统提示分层设计的经济价值。

动态区域的容量预算

缓存边界以下的动态区域并非无限。Claude Code 为不同来源设定了软性预算：

  • 项目指令（CLAUDE.md）：~2000 token
  • 记忆注入：~1500 token
  • 环境信息（Git 状态、当前目录等）：~500 token

总共约 4000 token 的动态区域。超过预算时按优先级截断——环境信息最先砍，项目指令最后砍。这和记忆系统（第 6 章）的容量控制逻辑一脉相承：越重要的信息，被裁剪的优先级越低。

延伸思考

在进入下一章之前，回头看看你的 harness/engine.py，思考几个问题：

  1. query_loop 和 query_loop_stream 有大量重复代码，你会怎么重构？
  2. 现在 execute_tool() 是占位的——如果你要实现一个”读文件”工具，它需要什么输入、返回什么输出？
  3. 压缩摘要的质量直接影响后续对话——如果摘要漏掉了关键信息，会发生什么？怎么防止？

章节小结

  • 三个 Prompt 构建了一个可运行的查询引擎：基础对话 → 流式输出 → 上下文压缩
  • 查询引擎的本质是一个 while True 循环：调用 LLM → 检查 tool_use → 执行工具 → 继续
  • 循环分四个阶段：Setup → API Call → Tool Execution → Continue Decision
  • 流式处理的秘密：print(text, end="", flush=True) 逐字蹦出来
  • Claude Code 有四种压缩策略按侵入性递进：微压缩 → 折叠 → 自动 → 响应式
  • 系统提示按”稳定在前”组装，最大化 Prompt Cache 命中率
  • 下一章给 Agent 装上手和脚——真正的文件读写和命令执行
English EN 简体中文 ZH