再聊 nanobot 的记忆机制：从消息计数到 Token 预算
Posted on 六 21 3月 2026 in AI
再聊 nanobot 的记忆机制：从消息计数到 Token 预算
上次写过一篇 nanobot 记忆机制的文章，聊了双层记忆（MEMORY.md + HISTORY.md）的基本设计。那篇文章发出去之后，nanobot 的代码又迭代了不少——从 4000 行长到了 15000 行，记忆系统也有了几个重要的变化。
这篇算是续集，聊三件事：
- 整理策略从"消息计数"升级到了"Token 预算"
- 会话历史加了一道"合法性校验"
- 一个真实的 bug：subagent 结果导致连续 assistant 消息
一、从消息计数到 Token 预算
上次说的整理触发条件是"未整理消息超过 100 条就触发"。这个方案简单，但有个明显的问题：100 条消息的 token 数差异巨大。
一条"在吗？"只有几个 token，一条包含完整代码文件的工具调用结果可能有上万 token。用消息条数做阈值，要么整理太频繁（浪费 API 调用），要么整理太晚（prompt 撑爆上下文窗口）。
新版本改成了基于 Token 预算的策略。核心逻辑在 MemoryConsolidator.maybe_consolidate_by_tokens() 里：
async def maybe_consolidate_by_tokens(self, session: Session) -> None:
target = self.context_window_tokens // 2
estimated, source = self.estimate_session_prompt_tokens(session)
if estimated < self.context_window_tokens:
return # 还没撑爆，不用整理
# 循环整理，直到 prompt 大小降到窗口的一半以下
for round_num in range(self._MAX_CONSOLIDATION_ROUNDS):
if estimated <= target:
return
boundary = self.pick_consolidation_boundary(session, max(1, estimated - target))
if boundary is None:
return
end_idx = boundary[0]
chunk = session.messages[session.last_consolidated:end_idx]
if not await self.consolidate_messages(chunk):
return
session.last_consolidated = end_idx
self.sessions.save(session)
estimated, source = self.estimate_session_prompt_tokens(session)
几个关键设计：
目标是窗口的一半。不是刚好卡在上限，而是留出充足的余量。这样整理完之后，用户还能聊很多轮才需要再次整理。
循环整理。一次可能不够——如果积压了大量消息，可能需要多轮整理才能把 prompt 压到目标以下。最多 5 轮（_MAX_CONSOLIDATION_ROUNDS），防止无限循环。
边界选择只在 user 消息处切割。pick_consolidation_boundary() 遍历消息列表，只在 role == "user" 的位置设置切割点。为什么？因为一个完整的"对话轮次"是 user → assistant(tool_calls) → tool(results) → assistant(response)。如果在 assistant 和 tool 之间切割，就会产生"孤儿"消息——tool result 找不到对应的 tool_calls，或者 tool_calls 找不到对应的 result。
def pick_consolidation_boundary(self, session, tokens_to_remove):
start = session.last_consolidated
removed_tokens = 0
last_boundary = None
for idx in range(start, len(session.messages)):
message = session.messages[idx]
if idx > start and message.get("role") == "user":
last_boundary = (idx, removed_tokens)
if removed_tokens >= tokens_to_remove:
return last_boundary
removed_tokens += estimate_message_tokens(message)
return last_boundary
Token 估算用 tiktoken。estimate_message_tokens() 用 cl100k_base 编码器估算每条消息的 token 数。不精确，但够用——我们只需要一个大致的量级来决定是否需要整理。
二、会话历史的合法性校验
即使边界选择很小心，还是可能出现"不合法"的历史。比如：
- 进程崩溃导致 session 文件写了一半
- 旧版本的 bug 留下了不一致的数据
- 手动编辑了 session 文件
所以 get_history() 在返回历史之前，会做一道合法性校验：
def get_history(self, max_messages=500):
unconsolidated = self.messages[self.last_consolidated:]
sliced = unconsolidated[-max_messages:]
# 跳过开头的非 user 消息
for i, message in enumerate(sliced):
if message.get("role") == "user":
sliced = sliced[i:]
break
# 检查孤儿 tool result
start = self._find_legal_start(sliced)
if start:
sliced = sliced[start:]
# 合并连续 assistant 消息（防御性）
return self._fix_consecutive_assistants(out)
_find_legal_start() 的逻辑是：从头扫描消息列表，维护一个"已声明的 tool_call_id"集合。每遇到 assistant 消息的 tool_calls，就把 id 加入集合；每遇到 tool result，就检查它的 tool_call_id 是否在集合里。如果不在——说明对应的 assistant 消息被截掉了——就把起始位置推到这条 tool result 之后。
@staticmethod
def _find_legal_start(messages):
declared = set()
start = 0
for i, msg in enumerate(messages):
if msg.get("role") == "assistant":
for tc in msg.get("tool_calls") or []:
if isinstance(tc, dict) and tc.get("id"):
declared.add(str(tc["id"]))
elif msg.get("role") == "tool":
tid = msg.get("tool_call_id")
if tid and str(tid) not in declared:
start = i + 1 # 跳过这条孤儿 tool result
declared.clear()
return start
这个校验解决了一类常见的 400 错误：Anthropic API 要求每个 tool result 都能在前面的 assistant 消息中找到对应的 tool_call_id，否则直接拒绝请求。
不过它有一个盲区：只检查"孤儿 tool result"，不检查"缺失 tool result"。如果 assistant 声明了 tool_calls=[A, B]，但只有 tool(A) 的结果，tool(B) 丢了——_find_legal_start 不会发现这个问题。好在正常流程中不会出现这种情况，因为 _run_agent_loop 总是会执行所有 tool call 并添加所有结果。
三、一个真实的 bug：连续 assistant 消息
说个刚修的 bug，挺有代表性的。
背景
nanobot 支持 subagent——主 Agent 可以用 spawn 工具创建子 Agent 异步执行任务。子 Agent 完成后，结果通过消息总线发回主 Agent。
有人提了一个 PR（commit f72ceb7），把 subagent 结果的消息角色从 user 改成了 assistant：
# 改动前
current_role = "user"
# 改动后
current_role = "assistant" if msg.sender_id == "subagent" else "user"
意图是好的：subagent 的结果确实更像是 assistant 的输出，而不是用户的输入。但这个改动引入了一个微妙的问题。
问题
当 subagent 结果作为 assistant 角色注入后，LLM 可能会回复 tool_calls。这时消息序列变成：
[...history...]
assistant("subagent 完成了，结果是...") ← 注入的 subagent 结果
assistant(tool_calls=[Y]) ← LLM 的回复
tool(tool_call_id=Y, content="...") ← 工具执行结果
assistant("处理完毕") ← LLM 的最终回复
两条连续的 assistant 消息！这些消息被 _save_turn 保存到 session 后，下次用户发消息时，get_history() 会把它们原样返回。
Anthropic 的 API 不允许连续的同角色消息。通过 LLM Gateway 转发时，gateway 尝试合并这两条 assistant 消息，但合并过程中 tool_calls 的配对关系被破坏了。于是报错：
'tool_call_id' of 'toolu_011hc7X68Zw7n1eWkDoNdtZN' not found in 'tool_calls' of previous message
修复
方案 B：保留 assistant role 的语义，但修复保存逻辑。
第一层：不保存注入的 assistant 消息。subagent 结果是当前轮次的上下文，不是真正的对话历史。保存时跳过它：
save_skip = 1 + len(history)
if current_role == "assistant":
save_skip += 1 # 跳过注入的 assistant 消息
self._save_turn(session, all_msgs, save_skip)
第二层：防御性合并。在 get_history() 返回之前，扫描并合并连续的 assistant 消息。即使 session 里已经有了脏数据（比如修复前产生的），也能兜住：
@staticmethod
def _fix_consecutive_assistants(messages):
result = []
for msg in messages:
if (result
and msg.get("role") == "assistant"
and result[-1].get("role") == "assistant"
and not result[-1].get("tool_calls")):
# 前一条 assistant 没有 tool_calls，合并内容到当前消息
prev_content = result[-1].get("content") or ""
cur_content = msg.get("content") or ""
merged = f"{prev_content}\n\n{cur_content}".strip()
result[-1] = dict(msg)
if merged:
result[-1]["content"] = merged
else:
result.append(msg)
return result
合并规则：只合并"前一条没有 tool_calls"的情况。如果前一条有 tool_calls，说明它后面应该跟 tool results，不能随便合并。
教训
这个 bug 的教训是：改消息角色要非常小心。LLM API 对消息序列有严格的约束（尤其是 Anthropic），而这些约束不仅影响当前请求，还会通过 session 持久化影响后续所有请求。一个看似无害的 role 改动，可能在几轮对话之后才暴露问题。
四、记忆系统的全景图
把上面说的串起来，nanobot 的记忆系统现在长这样：
用户发消息
│
▼
┌─────────────────────────────────────────┐
│ ContextBuilder.build_messages() │
│ ┌─────────────────────────────────┐ │
│ │ System Prompt │ │
│ │ ├─ 身份 (nanobot 🐈) │ │
│ │ ├─ AGENTS.md / SOUL.md / ... │ │
│ │ ├─ MEMORY.md (长期记忆全文) │ │
│ │ └─ Skills 列表 │ │
│ ├─────────────────────────────────┤ │
│ │ History (未整理的消息) │ │
│ │ └─ _find_legal_start 校验 │ │
│ │ └─ _fix_consecutive_assistants│ │
│ ├─────────────────────────────────┤ │
│ │ 当前用户消息 │ │
│ └─────────────────────────────────┘ │
└─────────────────────────────────────────┘
│
▼
LLM 推理 ←→ 工具调用循环
│
▼
┌─────────────────────────────────────────┐
│ _save_turn() │
│ ├─ 截断过长的 tool result │
│ ├─ 剥离 runtime context 前缀 │
│ ├─ 跳过空 assistant 消息 │
│ └─ 追加到 session.messages │
└─────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ maybe_consolidate_by_tokens() │
│ ├─ 估算当前 prompt token 数 │
│ ├─ 超过上下文窗口？ │
│ │ ├─ 选择 user 消息边界 │
│ │ ├─ 取出待整理的消息块 │
│ │ ├─ 调用 LLM 做摘要 │
│ │ │ ├─ history_entry → HISTORY.md │
│ │ │ └─ memory_update → MEMORY.md │
│ │ └─ 推进 last_consolidated │
│ └─ 循环直到 prompt < 窗口/2 │
└─────────────────────────────────────────┘
整个流程的设计哲学没变：简单、透明、可恢复。两个 Markdown 文件，一个 JSONL session，一个整数指针。没有向量数据库，没有知识图谱，没有复杂的检索管线。
对于个人助手这个场景，这个方案的性价比很高。你的偏好、项目上下文、工作习惯——这些信息加起来也就几 KB 到几十 KB，完全塞得进一个 prompt。当信息量大到塞不进去的时候，HISTORY.md 的 grep 搜索就是你的后备方案。
不够优雅？也许。但它管用，而且你随时可以打开 MEMORY.md 看看 AI 到底记住了什么。这种透明度，是很多"高级"记忆方案做不到的。
项目地址：github.com/HKUDS/nanobot 上一篇：nanobot 的记忆机制：它为什么能记住你的习惯和喜好？