📝 研究进展 #4：Context 管理与 Compaction 机制

一、Context 系统概述

Context 负责管理 Agent 的对话历史，包括：

• 存储用户和助手的消息
• 持久化到文件（支持会话恢复）
• 检查点机制（支持 D-Mail 时间旅行）
• Token 计数追踪

二、Context 数据结构设计

class Context:
    _history: list[Message]        # 消息历史
    _token_count: int              # 当前 Token 数
    _next_checkpoint_id: int       # 下一个检查点 ID
    _file_backend: Path            # 持久化文件路径

文件存储格式（JSON Lines）：

{"role": "user", "content": [...]}
{"role": "assistant", "content": [...], "tool_calls": [...]}
{"role": "_usage", "token_count": 1234}
{"role": "_checkpoint", "id": 0}
{"role": "user", "content": [...]}

三、检查点机制（Checkpoint）

检查点是实现 D-Mail（时间旅行） 功能的核心：

async def checkpoint(self, add_user_message: bool):
    checkpoint_id = self._next_checkpoint_id
    self._next_checkpoint_id += 1
    
    # 写入检查点标记
    await f.write(json.dumps({"role": "_checkpoint", "id": checkpoint_id}) + "\n")
    
    # 可选：在上下文中添加系统消息标记
    if add_user_message:
        await self.append_message(
            Message(role="user", content=[system(f"CHECKPOINT {checkpoint_id}")])
        )

回滚机制：

async def revert_to(self, checkpoint_id: int):
    # 1. 轮转当前文件（备份）
    rotated_file = await next_available_rotation(self._file_backend)
    await aiofiles.os.replace(self._file_backend, rotated_file)
    
    # 2. 重新读取直到指定检查点
    async for line in old_file:
        data = json.loads(line)
        if data["role"] == "_checkpoint" and data["id"] == checkpoint_id:
            break
        await new_file.write(line)

四、D-Mail 机制（时间旅行）

D-Mail 是项目中最有趣的设计之一，灵感来自《命运石之门》：

# denwa_renji.py 中
def send_dmail(self, checkpoint_id: int, message: str):
    """从未来发送消息回到过去的检查点"""
    self._pending_dmail = DMail(checkpoint_id, message)

# kimisoul.py 中
if dmail := self._denwa_renji.fetch_pending_dmail():
    # 抛出异常，让主循环捕获并回滚
    raise BackToTheFuture(
        dmail.checkpoint_id,
        [Message(role="user", content=[system(f"D-Mail: {dmail.message}")])]
    )

应用场景：

• 子 Agent 完成工作后通知父 Agent
• 跨会话的状态同步
• 错误恢复和重试

五、Compaction 机制

当对话历史接近 Token 限制时，需要压缩上下文：

触发条件：

# kimisoul.py
reserved = self._loop_control.reserved_context_size
if self._context.token_count + reserved >= self._runtime.llm.max_context_size:
    logger.info("Context too long, compacting...")
    await self.compact_context()

SimpleCompaction 算法：

class SimpleCompaction:
    def __init__(self, max_preserved_messages: int = 2):
        # 保留最近 N 条消息不压缩
        self.max_preserved_messages = max_preserved_messages
    
    async def compact(self, messages: Sequence[Message], llm: LLM):
        # 1. 准备阶段：分离需要压缩和保留的消息
        compact_message, to_preserve = self.prepare(messages)
        
        # 2. 调用 LLM 进行压缩
        result = await kosong.step(
            chat_provider=llm.chat_provider,
            system_prompt="You are a helpful assistant that compacts conversation context.",
            toolset=EmptyToolset(),  # 压缩时不允许调用工具
            history=[compact_message],
        )
        
        # 3. 构建压缩后的消息列表
        compacted_messages = [
            Message(role="user", content=[
                system("Previous context has been compacted..."),
                ...result.message.content...
            ]),
            *to_preserve  # 保留的最近消息
        ]
        return compacted_messages

压缩 Prompt 策略：

优先级（从高到低）：

1. 当前任务状态 - 正在做什么
2. 错误与解决方案 - 遇到的问题和解决方法
3. 代码演变 - 最终工作版本（删除中间尝试）
4. 系统上下文 - 项目结构、依赖、环境
5. 设计决策 - 架构选择和理由
6. TODO 项 - 未完成的任务

<current_focus>[当前工作重点]</current_focus>
<environment>[关键配置]</environment>
<completed_tasks>[已完成任务]</completed_tasks>
<active_issues>[活跃问题]</active_issues>
<code_state>[代码状态摘要]</code_state>
<important_context>[重要上下文]</important_context>

六、Context 生命周期

会话开始
    │
    ▼
┌─────────────┐
│   restore   │ ◀── 从文件恢复历史（如果有）
└─────────────┘
    │
    ▼
┌─────────────┐
├─ checkpoint ─┤ ◀── 创建检查点 0
└─────────────┘
    │
    ▼
┌─────────────┐
│  user msg   │
└─────────────┘
    │
    ▼
┌─────────────┐
├─ checkpoint ─┤ ◀── 每轮开始前创建检查点
└─────────────┘
    │
    ▼
┌─────────────┐
│  LLM step   │
└─────────────┘
    │
    ▼
┌─────────────┐     ┌─────────────┐
│  compact?   │ ──▶ │  compact()  │ （如果超出限制）
└─────────────┘     └─────────────┘
    │
    ▼
  ...循环...

七、关键发现 💡

1. 检查点是 D-Mail 的基础 - 没有检查点就无法实现时间旅行

1. 压缩保留最近 2 条消息 - 确保当前轮次的上下文不被破坏

1. 压缩时不允许工具调用 - 避免在压缩过程中产生副作用

1. 文件轮转策略 - revert/clear 时自动备份旧文件，防止数据丢失

1. Token 计数异步更新 - 每次 LLM 返回后更新，用于触发压缩

八、性能考量

九、下一步研究计划

研究时间：2026-02-23
当前进度：Context & Compaction ✓