nanobot 深入学习：会话管理与历史回放

1. 模块概述

Session 管理（nanobot/session/manager.py）负责会话的持久化、加载和历史回放。核心数据结构是 Session，由 SessionManager 管理其生命周期。

📍 源码：nanobot/nanobot/session/manager.py#L1-L876

1.1 会话文件格式

会话存储为 JSONL 文件（sessions/<safe_key>.jsonl），第一行是元数据记录：

{"_type":"metadata","key":"telegram:12345","created_at":"...","updated_at":"...","metadata":{...},"last_consolidated":0}
{"role":"user","content":"Hello","timestamp":"..."}
{"role":"assistant","content":"Hi! How can I help?","timestamp":"..."}

1.2 在系统中的位置

graph TB
    subgraph "SessionManager"
        GC[get_or_create]
        LOAD[_load]
        SAVE[save]
        REPAIR[_repair]
        FORK[fork_session_before_user_index]
    end

    subgraph "Session"
        GH[get_history]
        AM[add_message]
        CL[clear]
        RRS[retain_recent_legal_suffix]
        EFC[enforce_file_cap]
    end

    subgraph "消费者"
        AL[AgentLoop]
        AC[AutoCompact]
        WUI[WebUI API]
    end

    AL --> GC
    AL --> SAVE
    AL --> GH
    AC --> GC
    WUI --> FORK
    WUI --> LOAD

2. 命名体系与易混淆函数对比

2.1 命名规律拆解

前缀	操作对象	操作	含义
get_	or_	create	获取或创建会话
_get_	session_	path	获取会话文件路径
_get_	legacy_session_	path	获取旧版会话路径
retain_	recent_legal_	suffix	保留合法的最近消息后缀
enforce_	file_	cap	执行文件消息数上限
fork_	session_before_	user_index	在指定用户消息前分叉会话
_sanitize_	assistant_replay_	text	清理回放文本中的内部标记
_message_	preview_	text	生成消息预览文本
_metadata_	—	title	从元数据提取会话标题

2.2 易混淆函数对比表

对比维度	get_history	retain_recent_legal_suffix
目的	获取用于 LLM 输入的历史消息	获取用于持久化的最近消息
是否修改 session	❌ 只读	✅ 修改 messages 和 last_consolidated
返回值	消息列表	(dropped_messages, already_consolidated)
使用场景	AgentLoop._state_build()	AutoCompact + enforce_file_cap
一句话区分	get_history 为 LLM 准备输入	retain_recent_legal_suffix 为持久化截断会话

对比维度	save	flush_all
fsync	默认 False	强制 True
作用范围	单个 session	所有缓存的 session
使用场景	每次 turn 后	优雅关闭时
一句话区分	save 是常规持久化	flush_all 是关闭时的强制刷盘

3. API Signatures

@dataclass
class Session:
    key: str                                    # 会话键 (channel:chat_id)
    messages: list[dict[str, Any]]              # 消息列表
    created_at: datetime                        # 创建时间
    updated_at: datetime                        # 最后更新时间
    metadata: dict[str, Any]                    # 元数据
    last_consolidated: int = 0                  # 已巩固消息数

    def add_message(self, role, content, **kwargs):
        # 添加一条消息
        # 场景：AgentLoop._save_turn() 和 _persist_user_message_early()

    def get_history(
        self, max_messages=120, *, max_tokens=0,
        include_timestamps=False, extend_to_user=False,
    ) -> list[dict[str, Any]]:
        # 获取未巩固的消息用于 LLM 输入
        # 双重限制：消息数 + Token 数
        # 场景：AgentLoop._state_build()

    def clear(self):
        # 清空所有消息
        # 场景：/new 命令

    def retain_recent_legal_suffix(self, max_messages, *, extend_to_user=False):
        # 保留最近的合法消息后缀
        # 返回 (dropped, already_consolidated)
        # 场景：AutoCompact + enforce_file_cap

    def enforce_file_cap(self, on_archive=None, limit=2000):
        # 文件消息数上限保护
        # 场景：AgentLoop._state_save()


class SessionManager:
    def __init__(self, workspace: Path):
        # 初始化 sessions 目录和缓存
        # 场景：AgentLoop.__init__

    def get_or_create(self, key: str) -> Session:
        # 获取或创建会话（优先从缓存）
        # 场景：AgentLoop._process_message()

    def save(self, session: Session, *, fsync=False):
        # 原子写入会话文件
        # 场景：每个 turn 结束后

    def flush_all(self) -> int:
        # 强制刷盘所有缓存会话
        # 返回刷新的会话数
        # 场景：优雅关闭

    def delete_session(self, key: str) -> bool:
        # 删除会话文件和缓存
        # 场景：/new 命令

    def fork_session_before_user_index(
        self, source_key, target_key, before_user_index,
    ) -> Session | None:
        # 在指定用户消息前分叉会话
        # 场景：WebUI 的 "编辑并重试" 功能

    def list_sessions(self) -> list[dict]:
        # 列出所有会话（含预览）
        # 场景：WebUI 会话列表

4. 数据结构深度解析

4.1 Session

4.a 结构体存在的理由

Session 是对话上下文的持久化单元。它将消息列表、元数据和巩固游标封装在一起，支持增量保存和基于 Token 的历史截断。如果没有 Session，每次 LLM 调用都需要从零构建上下文，无法记住之前的对话。

4.b 字段三层分析表

字段	设计动机	反事实	替代方案
last_consolidated	标记已归档到 history.jsonl 的消息数量	每次都需要扫描全部消息判断哪些已归档	可以用消息上的 archived 标志，但游标模式在消息被删除后仍正确
metadata	存储 session 级别的运行时状态（checkpoint、goal_state、title）	需要额外的键值存储来跟踪这些状态	可以在消息中嵌入元数据，但污染 LLM 上下文
key	channel:chat_id 格式的唯一标识	无法区分不同渠道的相同 chat_id	可以用 UUID，但 channel:chat_id 具有业务含义

4.c 生命周期状态图

stateDiagram-v2
    [*] --> Created: get_or_create()
    Created --> Active: add_message()
    Active --> Active: add_message() / save()
    Active --> Consolidated: consolidate_by_tokens()
    Consolidated --> Active: 新消息到达
    Active --> Truncated: retain_recent_legal_suffix()
    Truncated --> Active: 新消息到达
    Active --> Cleared: clear() / delete_session()
    Cleared --> [*]

5. 函数逐行精讲

5.1 Session.get_history() — 历史回放

def get_history(self, max_messages=120, *, max_tokens=0, ...):
    # → 双重限制：先按消息数切片，再按 Token 数裁剪
    unconsolidated = self.messages[self.last_consolidated:]
    # → 只取未巩固的消息（已归档的不再发送给 LLM）

    start_idx = recent_message_start_index(unconsolidated, max_messages, extend_to_user=extend_to_user)
    sliced = unconsolidated[start_idx:]
    # → 从尾部保留最多 max_messages 条

    # 确保从 user 消息开始（Provider 要求）
    for i, message in enumerate(sliced):
        if message.get("role") == "user":
            start = i
            if i > 0 and sliced[i - 1].get("_channel_delivery"):
                start = i - 1  # → 包含主动推送的 assistant 消息
            sliced = sliced[start:]
            break

    # 过滤命令消息和清理回放文本
    for message in sliced:
        if message.get("_command"):
            continue  # → 命令不进入 LLM 上下文
        content = _sanitize_assistant_replay_text(content)
        # → 移除 [Message Time: ...] 等内部标记

    # Token 预算裁剪（如果指定了 max_tokens）
    if max_tokens > 0 and out:
        kept = []
        used = 0
        for message in reversed(out):
            tokens = estimate_message_tokens(message)
            if kept and used + tokens > max_tokens:
                break
            kept.append(message)
            used += tokens
        kept.reverse()

5.2 SessionManager.save() — 原子持久化

def save(self, session, *, fsync=False):
    path = self._get_session_path(session.key)
    tmp_path = path.with_suffix(".jsonl.tmp")
    # → 先写入临时文件，再原子替换

    with open(tmp_path, "w", encoding="utf-8") as f:
        # 第一行：元数据记录
        metadata_line = {"_type": "metadata", ...}
        f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n")
        # 后续行：消息记录
        for msg in session.messages:
            f.write(json.dumps(msg, ensure_ascii=False) + "\n")
        if fsync:
            f.flush()
            os.fsync(f.fileno())
            # → 强制刷盘（仅关闭时使用）

    os.replace(tmp_path, path)
    # → 原子替换（Unix: rename 是原子的；Windows: 如果目标存在会替换）

6. 设计决策分析

6.1 为什么使用 JSONL 而不是 SQLite？

决策：会话存储使用 JSONL 文本文件。

权衡：

• ✅ 人类可读，便于调试和手动修复
• ✅ 追加写入性能好
• ✅ 无外部依赖
• ❌ 不支持索引查询（如”所有包含 tool_calls 的消息”）
• ❌ 大文件需要全量扫描

6.2 为什么 last_consolidated 使用消息数而不是消息 ID？

决策：使用整数索引跟踪已巩固的消息。

权衡：

• ✅ 简单直接，O(1) 切片
• ✅ 不依赖消息中的 ID 字段
• ❌ 如果消息被外部修改导致顺序变化，游标失效

6.3 为什么 session 分叉只保留到指定 user_index 之前？

决策：fork_session_before_user_index 在用户编辑某条消息后创建新会话分支。

权衡：

• ✅ 用户可以在任意历史点”重新开始”，类似 Git branch
• ✅ 原始会话不受影响，保留完整历史
• ❌ 分叉后的会话不共享后续历史（这是设计意图）

7. 学习检查点

📝 本章小结

1. Session 是对话上下文的持久化单元，包含消息列表、元数据和巩固游标
2. JSONL 格式：第一行元数据，后续行消息，人类可读
3. 双重历史限制：先按消息数（max_messages），再按 Token 数（max_tokens）
4. 原子写入：先写临时文件，再 os.replace 替换
5. 会话分叉支持 WebUI 的”编辑并重试”功能

🤔 思考题

1. 为什么 get_history 中要先按消息数切片再按 Token 数裁剪，而不是反过来？

   参考答案

   消息数切片是粗粒度的快速过滤，Token 数裁剪是细粒度的精确控制。如果反过来（先按 Token 裁剪再按消息数切片），可能保留了不足 max_messages 条消息——因为 Token 裁剪可能已经切掉了很多短消息。实际上，消息数限制防止历史过长导致 LLM 注意力分散，Token 限制防止超出上下文窗口。两者服务于不同的约束。

2. enforce_file_cap 的默认上限 2000 条消息有什么考量？

   参考答案

   2000 条消息（nanobot/nanobot/session/manager.py#L28）约对应 10-20 万 Token（取决于消息长度），远超任何模型的上下文窗口。这个上限不是为了 Token 预算，而是防止 session 文件无限制增长导致磁盘 I/O 和加载性能下降。在达到此上限前，Consolidator 的 Token 预算巩固早已触发，因此 enforce_file_cap 主要作为最后的安全网。