📝 研究进展 #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 策略：
优先级（从高到低）：
- 当前任务状态 - 正在做什么
- 错误与解决方案 - 遇到的问题和解决方法
- 代码演变 - 最终工作版本（删除中间尝试）
- 系统上下文 - 项目结构、依赖、环境
- 设计决策 - 架构选择和理由
- 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() │ （如果超出限制）
└─────────────┘ └─────────────┘
│
▼
...循环...
七、关键发现 💡
- 检查点是 D-Mail 的基础 - 没有检查点就无法实现时间旅行
- 压缩保留最近 2 条消息 - 确保当前轮次的上下文不被破坏
- 压缩时不允许工具调用 - 避免在压缩过程中产生副作用
- 文件轮转策略 - revert/clear 时自动备份旧文件，防止数据丢失
- Token 计数异步更新 - 每次 LLM 返回后更新，用于触发压缩
八、性能考量
| 操作 | 时间复杂度 | 说明 |
|---|
append_message | O(1) | 追加写入文件 |
restore | O(n) | 读取整个历史文件 |
revert_to | O(k) | k 为检查点前的消息数 |
compact | O(m) + LLM | m 为压缩的消息数 |
九、下一步研究计划
| 阶段 | 目标 |
|---|
| #5 | Wire 协议 - UI 与 Soul 的通信机制 |
| #6 | Agent Spec - Agent 配置系统 |
| #7 | Approval 系统 - 用户审批流程 |
研究时间：2026-02-23
当前进度：Context & Compaction ✓
