nanobot源码学习(七):Session 与记忆压缩

前 6 篇介绍了 Agent 跑起来的全部核心机制——消息总线、Tool、Provider、状态机、入口配置、多渠道架构、Tool 沙箱。

但有一个所有长期运行的 Agent 都会遇到的问题还没讲——

对话历史会无限增长。

这一篇讲 nanobot 怎么让 Agent 聊得久、又不超 token 预算、还不丢上下文。

对应教程的 s11 章节——Session 文件 + Manager 缓存 + Dream 两阶段压缩。

1. 一个问题：聊天记录会怎样增长

假设用户跟 Agent 聊了 50 轮。直觉的写法是每次都把全部 messages 喂给 LLM。

第 1 轮：2 条消息，没问题。

第 10 轮：20 条消息，响应开始慢。

第 50 轮：100 条消息。单次响应要 8 秒、烧 0.3 元。

聊到 500 轮？直接超过context window 报错。

但这还不是最坏的情况——还有一个隐藏问题。到第 100 轮时，LLM 的 context window 只剩 20-30% 给"早期历史"，前 50 轮的内容会被挤出去。用户问"我们第一天聊过的 X 是什么"，LLM 不是忘了，是根本没看到。

两个问题同时出现：

性能/成本：消息越长，调用越慢、越贵
体验：LLM 看不到早期信息，“失忆”

光"裁剪最旧的消息"治标不治本——直接裁会丢信息。光"全量保留"治不了 token 超限。

需要的是压缩：用更少的 token 表达同样的信息。

2. 解法：单阶段压缩

最直觉的压缩：让 LLM 把前 10 条消息总结成 1 段，替换进去。


  
    
    
    
  
  python
def maybe_compact(self):
    if len(self.messages) <= MAX_MESSAGES:
        return
    old = self.messages[:6]
    text = "\n".join(f"{m['role']}: {m['content']}" for m in old)
    summary = complete(f"Summarize in 2 short sentences:\n{text}")
    self.messages = [
        {"role": "user",      "content": f"[Earlier summary]: {summary}"},
        {"role": "assistant", "content": "OK, I'll remember that."},
        *self.messages[6:],
    ]

这是 s11 教学代码里的版本，简单、跑得通。但有一个隐藏问题——LLM 总结会"骗你"。

可能漏掉关键事实（“哦这个用户喜欢用中文回复”）、把决策张冠李戴（“他说要做 A，其实他说的是 B”）、甚至脑补一些对话里没说过的话。

这叫"压缩幻觉"——压缩后的内容看起来像那么回事，但其实有错。下游决策基于错的摘要，会越走越偏。

3. nanobot 的解法：Dream 两阶段压缩

nanobot 真实代码里的 nanobot/session/dream.py 不是单阶段总结，而是两阶段。

3.1 阶段 1：短期裁剪

保留最近 N 条（默认 20）原文不动——保证当下对话流畅。更早的消息"打包"成一个待压缩列表。


  
    
    
    
  
  python
# 伪代码
keep_recent = 20
old_messages = self.messages[:-keep_recent]   # 打包待压缩
recent_messages = self.messages[-keep_recent:]  # 保留最近 20 条

短期裁剪不需要 LLM 调用，零成本。

3.2 阶段 2：长期摘要（结构化）

让 LLM 把"待压缩"的消息摘要成 3 个结构化字段：


  
    
    
    
  
  python
@dataclass
class DreamSummary:
    key_facts: list[str]       # 已确认的事实
    decisions: list[str]       # 做出的决策
    open_questions: list[str]  # 未完成的 TODO

关键经验：摘要要结构化，不要纯文本总结。

纯文本总结里 LLM 可能把"事实"和"决策"混在一起写。结构化后，下次 LLM 看到摘要能精确引用"哪个 fact"“哪个 decision”——减少二次误解。

压缩后：


  
    
    
    
  
  python
self.messages = recent_messages
self.metadata.summary = DreamSummary(
    key_facts=["用户偏好用中文回复", "项目用 nanobot 框架"],
    decisions=["选了 OpenAI 兼容模式", "ToolRegistry 走 entry_point 自动发现"],
    open_questions=["MCP 协议要支持吗？"],
)

messages 还是最近 20 条原文，metadata.summary 是结构化摘要。新对话进来时，把 summary + 原文一起喂给 LLM——它能看到"过去的关键事实"，又不会超 token 预算。

4. 整个流程串起来

用户发来第 51 轮消息
  ↓
AgentLoop 启动 → RESTORE 状态
  ↓ SessionManager.get_or_create(session_key) 拿到/创建 Session
  ↓
COMPACT 状态
  ↓ messages 超 50 条 → 触发 Dream
  ↓ 阶段 1: 保留最近 20 条原文
  ↓ 阶段 2: LLM 把前 30 条摘要成结构化 summary
  ↓
BUILD 状态
  ↓ 拼 Prompt 时把 summary + 最近 20 条都喂给 LLM
  ↓
RUN 状态
  ↓ LLM 看到完整上下文 → 给出响应
  ↓
SAVE 状态
  ↓ SessionManager.update(session) 异步落盘
  ↓
RESPOND → DONE

s04 章节讲的状态机骨架，在 s11 章节有了具体的算法实现——TurnState.COMPACT 状态调的就是 Dream.compact(session)。

5. Session 怎么落盘和缓存

光有压缩算法还不够——Session 本身要存起来。重启 Agent、跨进程、跨设备，对话历史都要能恢复。

nanobot 的设计分三层：

第 1 层：Session 数据类


  
    
    
    
  
  python
@dataclass
class Session:
    key: str                          # session_key
    messages: list[dict]              # 当前消息列表
    metadata: SessionMetadata         # 包含 summary、created_at、last_active 等

第 2 层：SessionManager（缓存 + 锁 + 异步落盘）


  
    
    
    
  
  python
class SessionManager:
    def __init__(self):
        self._cache: dict[str, Session] = {}        # 热缓存
        self._locks: dict[str, asyncio.Lock] = {}   # per-key 锁
        self._store = FileSessionStore()             # 持久化后端

    async def get_or_create(self, key: str) -> Session:
        if key not in self._cache:
            async with self._lock_for(key):
                self._cache[key] = await self._store.load(key)
        return self._cache[key]

热缓存避免每次都从磁盘读
per-key 锁避免同一 session 并发更新冲突
异步落盘不阻塞 LLM 调用

第 3 层：FileSessionStore（落盘）


  
    
    
    
  
  python
class FileSessionStore:
    async def save(self, session: Session) -> None:
        path = self._session_path(session.key)  # ~/.nanobot/sessions/{hash}.json
        async with aiofiles.open(path, "w") as f:
            await f.write(json.dumps(asdict(session), default=str))

JSON 文件存 session 简单，但量大后 I/O 慢。真实 nanobot 还提供了 SQLite 后端——nanobot/session/backends/sqlite.py，通过配置切换，业务代码不动。

6. session_key

s08 章节讲过 session_key = hash(channel + chat_id + thread_id + user_id)。这个设计在 s11 章节兑现——

同一个飞书群的多轮对话，session_key 一样，Session 是同一个，记忆是连续的
不同群的 session_key 不同，记忆隔离
用户换群了，session_key 变了，LLM 不会"串台"


  
    
    
    
  
  bash
[飞书 项目 A 群]   → session_key=hash("feishu|chat_A|null|user_X")
                     ↓
                   Session A1（连续对话）
                     ↓
[飞书 项目 B 群]   → session_key=hash("feishu|chat_B|null|user_X")
                     ↓
                   Session B1（新对话，跟 A1 完全隔离）

Session 是对话的"身份证"，靠 session_key 索引。

7. 效果对比

没有压缩 vs 有压缩 的对比（粗略估算）：

轮次	没有压缩		有压缩
	消息数	单次 token	消息数	单次 token
10	20	~3k	20	~3k
50	100	~15k	20+summary	~5k
200	400	~60k	20+summary	~5k

加压缩后：

200 轮时单次 token 从 60k 降到 5k（省 90%）
早期信息（key facts / decisions）通过结构化摘要保留
用户体验"不失忆"

8. 常见的误区：只裁剪不压缩

如果 Agent 项目这么写：


  
    
    
    
  
  python
# 消息数超过 50 就裁掉前 30 条
if len(messages) > 50:
    messages = messages[-50:]

代码能跑。但有个隐藏问题——用户的关键信息被一刀切：

用户在第 5 轮说了"我姓张"——第 5 轮被裁掉了
用户在第 20 轮做了关键决策——第 20 轮被裁掉了
LLM 后面响应时完全看不到这些

正确做法：两阶段压缩——短期裁剪保流畅，长期结构化摘要保留关键信息。

从一开始就预留 metadata.summary 字段，别等到性能出问题才加压缩。前期投入小，后期不用重构。

9. 下一篇

下一篇进 s12 章节，讲周边能力——Cron（定时任务）、Command（斜杠命令）、Security（沙箱保护）。这些不是核心机制，但少了它们 Agent 上不了生产。

10. 参考资料

nanobot 源码：https://github.com/HKUDS/nanobot
nanobot-tutorial（14 章配套教程）：https://github.com/yaoweizhang/nanobot_tutorial
nanobot 官方 Roadmap：https://github.com/HKUDS/nanobot/discussions/431

跟读建议：跑 python s11_session_memory/code.py 体验"消息数超过 8 自动压缩"。再读 nanobot/session/dream.py 看完整两阶段压缩实现。