nanobot 源码解读 · 04 | 记忆与状态：Dream 长期记忆与会话压缩

self.memory_dir = ensure_dir(workspace / "memory")
self.memory_file   = self.memory_dir / "MEMORY.md"      # 项目/事实记忆
self.history_file  = self.memory_dir / "history.jsonl"  # 追加式历史（Dream 原料）
self.soul_file     = workspace / "SOUL.md"              # Agent 行为规则
self.user_file     = workspace / "USER.md"              # 用户画像
self._cursor_file  = self.memory_dir / ".cursor"        # 历史自增游标
self._dream_cursor_file = self.memory_dir / ".dream_cursor"  # Dream 消费到哪了

def append_history(self, entry, *, max_chars=None, session_key=None) -> int:
    limit = max_chars if max_chars is not None else _HISTORY_ENTRY_HARD_CAP
    ts = datetime.now().strftime("%Y-%m-%d %H:%M")
    raw = entry.rstrip()
    if len(raw) > limit:
        raw = truncate_text(raw, limit)        # 防御：单条上限 64K，防 LLM 把输入当摘要回灌
    content = strip_think(raw)                  # 剥离 <think> 泄漏 / <channel|> 标记
    with self._append_lock:                    # 线程锁：游标分配 + 追加必须原子
        cursor = self._next_cursor()
        record = {"cursor": cursor, "timestamp": ts, "content": content}
        if session_key:
            record["session_key"] = session_key
        with open(self.history_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
        self._cursor_file.write_text(str(cursor), encoding="utf-8")
    return cursor

def read_recent_history_for_prompt(self, since_cursor, *, session_key, unified_session=False):
    entries = self.read_unprocessed_history(since_cursor=since_cursor)
    if session_key is None:
        return entries
    if not unified_session:
        return [e for e in entries if e.get("session_key") == session_key]  # 只取本会话
    # unified_session：本会话 + 非内部会话（跨会话统一视图）
    return [e for e in entries
            if (s := e.get("session_key")) == session_key
            or not self._is_internal_history_session(s)]

def build_dream_prompt(self, *, max_entries: int = 20) -> tuple[str, int] | None:
    last_cursor = self.get_last_dream_cursor()           # 上次读到哪
    entries = self.read_unprocessed_history(since_cursor=last_cursor)
    if not entries:
        return None                                       # 没新东西就不跑
    batch = entries[:max_entries]                         # 每批最多 20 条
    history_text = "\n".join(
        f"[{e['timestamp']}] {truncate_text(e['content'], 1000)}" for e in batch)
    template = self._dream_template()                     # agent/dream.md
    files_section = self._render_current_memory_files()   # 当前三个记忆文件真实内容
    prompt = f"{template}\n\n{files_section}\n\n## Conversation History\n{history_text}"
    return (prompt, batch[-1]["cursor"])

editable_files = [self.memory_file, self.soul_file, self.user_file]
tools.register(EditFileTool(workspace=workspace, allowed_dir=skills_dir,
                            extra_write_allowed_files=editable_files, ...))
tools.register(ApplyPatchTool(workspace=workspace, allowed_dir=skills_dir,
                              extra_write_allowed_files=editable_files, ...))

resp = await loop.process_direct(prompt, session_key=key, ephemeral=True,
                                 tools=store.build_dream_tools(),
                                 on_progress=progress, runtime=dream_runtime)
diff_body = store.dream_content_diff()          # git 工作区真实 diff
completed = MemoryStore.dream_run_completed(resp, had_tool_errors=progress.had_tool_errors)
if completed:
    store.set_last_dream_cursor(last_cursor)    # 只有成功才推进游标
    ...
# finally:
if store.git.is_initialized():
    commit_msg = build_dream_commit_message("dream: manual run", diff_body)
    sha = store.git.auto_commit(commit_msg)     # 用真实 diff 写审计，而非模型自述
store.compact_history()

@dataclass
class Session:
    key: str
    messages: list[dict] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)
    last_consolidated: int = 0   # 已固化进文件（压缩掉）的消息条数
    provider_state = None

async def archive(self, messages, *, runtime, session_key=None, summary_messages=None):
    if not messages:
        return None
    messages_to_summarize = public_history_messages(summary_messages or messages)
    formatted = MemoryStore._format_messages(messages_to_summarize)
    formatted = self._truncate_to_token_budget(formatted, runtime=runtime)
    system_prompt = render_template("agent/consolidator_archive.md", strip=True)
    try:
        response = await runtime.provider.chat_with_retry(
            model=runtime.model,
            messages=[{"role": "system", "content": system_prompt},
                      {"role": "user", "content": formatted}],
            tools=None, ...)
    except Exception:
        self.store.raw_archive(messages, session_key=session_key)   # 失败降级：原样落盘
        return None
    summary = response.content or "[no summary]"
    self.store.append_history(summary, max_chars=_ARCHIVE_SUMMARY_MAX_CHARS,
                             session_key=session_key)
    return summary

budget = self._input_token_budget(runtime)          # 上下文窗口 - 生成预算 - 1024 安全余量
target = int(budget * self.consolidation_ratio)     # 默认 0.5
# ... 先处理 replay 窗口溢出 ...
for round_num in range(self._MAX_CONSOLIDATION_ROUNDS):   # 最多 5 轮
    if estimated <= target:
        break
    boundary = self.pick_consolidation_boundary(session, max(1, estimated - target))
    if boundary is None:
        break
    chunk = session.messages[session.last_consolidated:end_idx]
    summary = await self.archive(chunk, runtime=runtime, session_key=session.key)
    session.last_consolidated = end_idx              # 推进游标
    session.provider_state = None                     # 压缩后旧 provider 状态失效
    self.sessions.save(session)
    if not summary:
        break                                         # LLM 降级，停止硬压
    estimated, source = self.estimate_session_prompt_tokens(session, runtime=runtime)

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)

def check_expired(self, schedule_background, resolve_runtime, active_session_keys=()):
    for info in self.sessions.list_sessions():
        key = info.get("key", "")
        if not key or self._is_internal_session(key) or key in self._archiving:
            continue
        if key in active_session_keys:
            continue                                      # 正在聊的不压
        if self._is_expired(info.get("updated_at"), now) \
           and self._has_compactable_idle_tail(key):
            runtime = resolve_runtime(session)
            self._archiving.add(key)
            schedule_background(self._archive(key, runtime=runtime))   # 后台跑

meta = session.metadata.get("_last_summary")
if isinstance(meta, dict):
    text = meta.get("text")
    ...
    return session, self._format_summary(text, last_active)
# _format_summary: "Previous conversation summary (last active ...):\n{text}"

if session_summary:
    parts.append(f"[Archived Context Summary]\n\n{session_summary}")