2.3 Token 预算与截断策略
协议层让我们能换后端，也能离线复现，但还有一条硬约束没处理：模型能吃的输入是有限的。对话一轮轮堆下去，历史会越来越长，迟早撞上上下文窗口；就算没撞上，每次把全部历史重发一遍，账单和延迟也在悄悄涨。这一节给 minagent 加上预算与裁剪。
本节把 minagent 推进到「会自己收口」：用近似函数估算 token，用「保留 system + 最近 N 轮 + 摘要」的规则裁剪超预算的消息，并把逻辑包成
BudgetedModel包装器。到本节结束，循环就算跑很久也不会把上下文撑爆。
2.3.1 为什么需要预算
预算不是一个优化项，而是三条硬约束的公共要求：
| 约束 | 触发条件 | 越界后果 |
|---|---|---|
| 上下文窗口 | 输入 token 超过模型上限 | 请求直接报错，无法降级 |
| 成本 | 每轮重发全部历史 | 长对话的输入费用指数上涨 |
| 延迟 | 输入越长首字越慢 | 用户等待变长，体验下降 |
关键在于每一次调用都要重发整个历史（无状态 API 的代价）。第 10 轮的请求包含了前 9 轮，第 100 轮包含前 99 轮——如果不裁剪，输入 token 是平方级增长的。所以预算要「每轮都做」，不是「快满了才做」。
三个约束里，上下文窗口是硬的（超了直接失败），成本和延迟是软的（超了只是变贵变慢）。工程上的顺序是：先用硬约束定一个安全上限，再在软约束下尽量多留有用的历史。这也解释了为什么预算是「每轮都做的小动作」，而不是「偶尔一次的大扫除」。
2.3.2 估算：字符近似 vs 真实 tokenizer
裁剪的前提是能算「现在多少 token」。有两条路：
| 方案 | 精度 | 依赖 | 适合 |
|---|---|---|---|
| 真实 tokenizer（tiktoken / SDK 的 count_tokens） | 高 | 需要额外包或网络 | 精确计费、临界判断 |
| 字符数近似 | 中 | 零依赖 | 日常裁剪、教学骨架 |
本机没有安装 tiktoken，而且它是一个第三方依赖——与全书「核心零依赖」的口径不符。所以 minagent 用字符近似：对英文等 ASCII 文本，大约 4 个字符 1 个 token；对中文，大约 1 个字符 1 个 token。这个精度对「决定要不要裁」完全够用，误差通常在一两成以内。
什么时候必须上真实 tokenizer？当你需要贴着上限做判断（比如刚好卡在 200K 窗口边缘），或者要做精确的成本核算时。真实适配器（2.2 节）可以把供应商返回的
usage.input_tokens回填进估算模型，做「近似 + 校准」的混合方案。
补充一点：Anthropic SDK 提供了 client.messages.count_tokens(...) 端点，可以在不真正生成的前提下精确数 token。它需要网络，所以不适合放进离线单元测试，但很适合在「上线前做一次精确预检」时调用。minagent 把它留作可选增强，核心路径仍走零依赖的近似函数。
2.3.3 实现近似估算
# minagent/budget.py
from __future__ import annotations
from typing import Callable
from minagent.types import Message, ModelResponse
def approx_tokens(text: str) -> int:
"""无 tiktoken 时的近似：ASCII 约 4 字符 1 token，CJK 约 1 字符 1 token。"""
ascii_n = sum(1 for c in text if ord(c) < 128)
cjk_n = len(text) - ascii_n
return cjk_n + (ascii_n + 3) // 4
这里把字符分成 ASCII 与非 ASCII 两类分别计费：(ascii_n + 3) // 4 是向上取整的除法，避免 len < 4 时算出 0。快速验证一下量级：
'hello world' -> 约 3 tokens
'你好，世界' -> 约 5 tokens
hello world 有 11 个 ASCII 字符，11 // 4 = 2……等等，输出是 3？因为代码里写的是 (11 + 3) // 4 = 3，向上取整生效了。中文那句 5 个字符全是非 ASCII，直接得 5。数量级是对的。
2.3.4 截断策略：保留什么、丢什么
估算有了，接下来定「丢什么」。随意丢弃会破坏对话的连贯性，所以我们用一条明确规则：保留 system 与最近若干轮，把更早的消息压缩成一条摘要。
| 消息类别 | 处理 | 理由 |
|---|---|---|
| system | 永远保留 | 指令、角色设定不能丢 |
| 最近 N 轮 | 永远保留 | 当前任务的直接上下文 |
| 更早的对话 | 丢弃并摘要 | 保留「发生过什么」的线索 |
| 工具结果 | 随所属轮次处理 | 大结果往往是最占 token 的 |
摘要不是把原文抄一遍，而是一句「已裁剪 K 条较早消息」的占位——第 6 章会用真正的摘要模型替换它。这里先把它做成一个可替换的钩子。
为什么「保留最近 N 轮」而不是「保留最近 N 条消息」？一轮通常包含 user + assistant 两条，按条数裁可能把一轮切成两半，模型看到「有回答没提问」会困惑。keep_recent 命名成「条」是为了实现简单，但实践里应设成偶数（比如 4），效果等价于保留最近 2 轮。
def render(m: Message) -> str:
parts = [m.content]
parts += [f"{tc.name}{tc.arguments}" for tc in m.tool_calls]
return "".join(parts)
def count_messages(messages: list[Message]) -> int:
return sum(approx_tokens(render(m)) for m in messages)
render 把一条消息「拍平」成参与计数的文本：正文加上工具调用的名字与参数。工具调用的参数常常很长，不把它们算进去会严重低估。
2.3.5 实现裁剪
def trim_to_budget(messages: list[Message], max_tokens: int,
keep_recent: int = 4) -> tuple[list[Message], list[Message]]:
"""保留 system + 最近 keep_recent 条；超预算再从最旧处继续丢。返回 (保留, 丢弃)。"""
system = [m for m in messages if m.role == "system"]
convo = [m for m in messages if m.role != "system"]
dropped: list[Message] = []
if len(convo) > keep_recent:
dropped = convo[:-keep_recent]
convo = convo[-keep_recent:]
while len(convo) > 1 and count_messages(system + convo) > max_tokens:
dropped.append(convo.pop(0))
return system + convo, dropped
逻辑分两层：先按条数保留最近 N 条（快，且符合直觉），再按 token 数从最旧处继续丢（兜底，应对单条消息特别长的情况）。while 里的 len(convo) > 1 保证至少留一条，避免把对话裁空。丢弃的消息按顺序收集，方便后续做摘要。
2.3.6 包装成 BudgetedModel
裁剪逻辑不该散落在循环里，而应包成一个装饰器式的模型——它本身实现 Model 协议，内部持有另一个模型：
def summarize(dropped: list[Message]) -> Message | None:
if not dropped:
return None
tools = [tc.name for m in dropped for tc in m.tool_calls]
tail = f"，涉及工具 {tools}" if tools else ""
return Message(role="user",
content=f"[已裁剪 {len(dropped)} 条较早消息{tail}]")
class BudgetedModel:
"""在调用内层模型前先裁剪消息；把裁剪量记进 usage 便于观测。"""
def __init__(self, inner, max_input_tokens: int = 200,
keep_recent: int = 4) -> None:
self.inner = inner
self.max_input_tokens = max_input_tokens
self.keep_recent = keep_recent
self.last_dropped: list[Message] = []
def complete(self, messages: list[Message], tools: list[dict]) -> ModelResponse:
kept, dropped = trim_to_budget(messages, self.max_input_tokens, self.keep_recent)
note = summarize(dropped)
if note is not None:
system = [m for m in kept if m.role == "system"]
rest = [m for m in kept if m.role != "system"]
kept = system + [note] + rest
self.last_dropped = dropped
resp = self.inner.complete(kept, tools)
resp.usage = {
**resp.usage,
"est_input_tokens": count_messages(kept),
"trimmed_messages": len(dropped),
}
return resp
设计要点有三：① 装饰器模式——BudgetedModel(EchoModel()) 或 BudgetedModel(AnthropicModel(...)) 都行，因为它只依赖内层的 complete；② 摘要插在 system 之后、最近对话之前，位置固定；③ 裁剪量写回 usage，这样一次调用花了多少、裁了多少，一眼可见。这种「包装器叠加」是全书的核心手法，第 14 章的重试、第 15 章的成本核算都会复用同一个形状。
2.3.7 真跑：裁剪前后对比
构造一段 17 条消息的长对话，预算设成 120 token：
# run23.py
import sys
sys.path.insert(0, "/tmp/ab_work")
from minagent.types import Message
from minagent.models import EchoModel
from minagent.budget import approx_tokens, count_messages, BudgetedModel
for s in ["hello world", "你好，世界"]:
print(f"{s!r:16} -> 约 {approx_tokens(s)} tokens")
convo = [Message(role="system", content="你是简洁的助手，只回答要点。")]
for i in range(1, 9):
convo.append(Message(role="user", content=f"第{i}个问题：请解释注意力机制的细节。"))
convo.append(Message(role="assistant", content=f"第{i}个回答：注意力通过 QK 点积加权。" * 3))
before_n, before_t = len(convo), count_messages(convo)
print(f"\n裁剪前：{before_n} 条消息，约 {before_t} tokens")
inner = EchoModel()
bm = BudgetedModel(inner, max_input_tokens=120, keep_recent=4)
resp = bm.complete(convo, tools=[])
passed, _ = inner.calls[-1]
print(f"裁剪后：{len(passed)} 条消息，约 {resp.usage['est_input_tokens']} tokens")
print("传给内层模型的消息：")
for m in passed:
head = m.content[:24].replace("\n", " ")
print(f" {m.role:9} {head}…")
print(f"丢弃 {resp.usage['trimmed_messages']} 条；usage={resp.usage}")
真跑输出：
'hello world' -> 约 3 tokens
'你好，世界' -> 约 5 tokens
裁剪前：17 条消息，约 550 tokens
裁剪后：4 条消息，约 91 tokens
传给内层模型的消息：
system 你是简洁的助手，只回答要点。…
user [已裁剪 14 条较早消息]…
user 第8个问题：请解释注意力机制的细节。…
assistant 第8个回答：注意力通过 QK 点积加权。第8个回…
丢弃 14 条；usage={'input_tokens': 0, 'output_tokens': 0, 'est_input_tokens': 91, 'trimmed_messages': 14}
结果符合预期：17 条、约 550 token 被裁成 4 条、约 91 token（system + 1 条摘要 + 最近一轮问答）。裁剪率约 83%，且 system 与最近上下文完好保留。usage 里多了 est_input_tokens 与 trimmed_messages 两个键，供后续观测使用。
顺带验证了一件事：BudgetedModel 是透明的——inner.calls[-1] 里拿到的正是裁剪后的消息列表，说明裁剪发生在「交给内层模型之前」，而不是事后补救。这正是装饰器模式该有的样子。
2.3.8 边界与取舍
裁剪永远是有损的，几条经验值得记住：
- 预算别贴着模型上限设。留出输出 token 与安全余量，通常按上限的 70%–80% 设输入预算。
- 摘要位置固定。插在 system 后、最近对话前，避免模型把摘要误当成最新指令。
- 别裁掉最后一条 user。
len(convo) > 1就是在保这条底线；否则模型会「忘记」当前问题。 - 近似不是免费的精确。做计费时要用真实
usage校准，近似只用于「决定裁不裁」。 - 能早裁就别等超。在每轮调用前做一次预算检查，比等撞上错误再补救便宜得多，也更容易测试。
还有一种情况不该裁：当历史本身就是要交付的内容（比如「总结我们前面聊的所有决定」）。这时正确的做法不是丢弃，而是换一个更大的上下文窗口或分层摘要——预算是手段，不是目的。
真实模型下会怎样
真实适配器返回的 usage.input_tokens 是精确值，可以用它来校准近似函数：跑一批请求，算出「近似 / 真实」的平均比值，再乘回去。真实 tokenizer 对中文、代码、JSON 的切分也和字符近似不同，越靠近上限越要小心。另外，真实场景里「工具结果特别长」是常见爆预算原因（比如一次读了整个文件），这时摘要策略要从「按条数」升级到「按内容类型」——第 6 章会专门处理。本节的 BudgetedModel 是一个能跑、能测、能替换的骨架，后面的增强都挂在它上面。
阅读导航：上一节：2.2 真实供应商适配与离线替身 · 下一节：3.1 朴素循环：跑到没有工具调用为止 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。