会话存储：Append-only 状态持久化

当 Agent 执行到一半崩溃了，你该怎么办？从头重来，还是从断点继续？答案藏在存储设计里。

前言

AI Agent 的会话状态是整个系统中最脆弱也最关键的部分。一次典型的 Agent 执行可能跨越数十轮工具调用、数千个 token 的上下文，以及数分钟乃至数小时的执行时间。在这漫长的生命周期中，任何环节的异常——网络中断、进程崩溃、用户主动暂停——都可能导致状态丢失。

传统的数据库方案采用"最新值覆盖"（Last-Write-Wins）策略来存储状态，这在 Agent 场景下存在根本性缺陷：你无法知道"什么时候的值"才是正确的，也无法回溯到某个中间状态重新开始。本文将深入探讨 Append-only 状态持久化 模式，这是 Claude Code 等主流 AI Agent 系统在实践中采用的核心设计哲学。

会话状态的挑战

会话状态的挑战

状态的多维性

Agent 会话状态远不止"对话历史"这么简单。一个完整的会话状态至少包含以下维度：

  • 对话历史：用户消息、助手回复、系统提示词
  • 工具调用记录：每个工具的输入、输出、执行状态
  • 中间决策：Agent 的推理链、分支选择、回退记录
  • 环境快照：文件系统状态、代码变更、外部 API 响应
  • 元数据：创建时间、最后修改时间、token 计数、模型参数

这些维度之间存在复杂的依赖关系。如果你只保存了对话历史而丢失了工具调用记录，那么后续的工具调用可能会产生重复副作用（比如重复发送邮件、重复创建文件）。

一致性困境

在分布式环境中，会话状态的一致性面临三大挑战：

  1. 并发写入：用户可能在 Agent 执行过程中发送新消息
  2. 部分失败：某个工具调用成功但后续步骤失败
  3. 因果顺序：消息的物理到达顺序可能与逻辑顺序不一致

传统的事务模型（ACID）在这种场景下过于重量级，而最终一致性模型又无法满足 Agent 对"可恢复性"的强需求。我们需要一种介于两者之间的方案。

Append-only 设计哲学

Append-only 设计哲学

核心思想：只追加，不修改

Append-only（仅追加）存储的核心原则极为简单：任何状态变更都以新记录的形式追加到日志末尾，永不修改已有记录。这看似低效，却解决了 Agent 状态管理中的几乎所有核心问题。

Append-only方式

event: A

event: B

event: C

event: D

传统方式

覆盖

覆盖

state = A

state = B

state = C 最终状态

Event Sourcing 模式

Append-only 存储本质上是 Event Sourcing（事件溯源）模式在 Agent 系统中的应用。Event Sourcing 最早由 Martin Fowler 在 2005 年提出，其核心思想是：不存储实体的当前状态，而是存储导致状态变化的所有事件。

在 Agent 上下文中，每个"事件"代表一次状态变更：

  • message_added：新增一条消息
  • tool_invoked：发起一次工具调用
  • tool_completed：工具调用完成
  • context_summarized：上下文被摘要压缩
  • checkpoint_created：创建恢复检查点

通过重放（replay）事件序列，你可以精确重建任意时间点的会话状态。

为什么不用 WAL？

Write-Ahead Log（预写日志）是数据库领域常见的持久化策略，但 WAL 和 Append-only 存储有几个关键区别：

特性WALAppend-only 日志
目的保证事务原子性保证状态可追溯性
生命周期事务完成后可清理永久保留或按策略归档
查询能力不支持直接查询支持按时间/类型查询
存储格式二进制、内部格式结构化、可读格式
适用场景数据库内部机制应用层状态管理

对于 Agent 系统，我们既需要 WAL 的原子性保证，也需要 Append-only 的可追溯性。实际上，很多 Agent 框架会同时使用两者：WAL 用于保证单次操作的原子性，Append-only 日志用于跨操作的状态追踪。

存储格式与序列化

存储格式与序列化

JSONL：简单但有效

Claude Code 的会话存储采用了 JSONL（JSON Lines）格式，每行一个 JSON 对象，代表一个事件。这种格式有几个显著优势：

  1. 流式写入：每个事件可以独立序列化和写入，无需等待整个文件完成
  2. 容错性：某一行损坏不影响其他行的读取
  3. 可读性：人类可以直接阅读和调试
  4. 跨语言：任何编程语言都能轻松解析

一个典型的会话日志文件可能看起来像这样：

{"type":"session_start","timestamp":"2025-01-15T10:30:00Z","model":"claude-sonnet-4-20250514","session_id":"sess_abc123"}
{"type":"message_added","role":"user","content":"帮我分析这段代码的性能问题","timestamp":"2025-01-15T10:30:05Z"}
{"type":"message_added","role":"assistant","content":"我来分析这段代码...","timestamp":"2025-01-15T10:30:08Z","usage":{"input_tokens":150,"output_tokens":80}}
{"type":"tool_invoked","tool_name":"read_file","input":{"path":"main.py"},"timestamp":"2025-01-15T10:30:10Z","invocation_id":"inv_001"}
{"type":"tool_completed","invocation_id":"inv_001","output":"def process_data(data):...","duration_ms":45,"timestamp":"2025-01-15T10:30:11Z"}
{"type":"checkpoint","state_hash":"a1b2c3d4","event_count":5,"timestamp":"2025-01-15T10:30:12Z"}

消息序列化策略

对话历史是会话状态中体积最大的部分。对于长对话，需要精心设计序列化策略：

import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from datetime import datetime
from enum import Enum

class EventType(Enum):
    SESSION_START = "session_start"
    MESSAGE_ADDED = "message_added"
    TOOL_INVOKED = "tool_invoked"
    TOOL_COMPLETED = "tool_completed"
    CHECKPOINT = "checkpoint"
    CONTEXT_SUMMARIZED = "context_summarized"

@dataclass
class SessionEvent:
    """会话事件基类 - 所有状态变更都通过此结构记录"""
    type: EventType
    timestamp: str
    event_id: str  # 全局唯一事件ID，用于去重和排序

    def to_jsonl(self) -> str:
        """序列化为 JSONL 格式（单行 JSON）"""
        data = asdict(self)
        data['type'] = self.type.value  # Enum 转字符串
        return json.dumps(data, ensure_ascii=False)

这段代码定义了事件的基础结构。注意 event_id 字段——它是保证事件幂等性的关键。即使同一个事件被意外写入两次，系统也能通过 event_id 去重。

索引与压缩

随着会话日志增长，直接顺序扫描的效率会下降。实践中通常会维护一个索引文件，记录关键事件的偏移量：

@dataclass
class SessionIndex:
    """会话索引 - 记录关键事件的位置，加速随机访问"""
    session_id: str
    checkpoints: List[dict]       # 检查点列表：{event_id, offset, timestamp}
    message_offsets: List[dict]   # 消息偏移量：{event_id, offset, role}
    last_event_id: str            # 最后一个事件的ID
    total_events: int             # 总事件数

    def find_nearest_checkpoint(self, target_event_id: str) -> Optional[dict]:
        """找到目标事件之前最近的检查点"""
        candidates = [
            cp for cp in self.checkpoints
            if cp['event_id'] <= target_event_id
        ]
        return max(candidates, key=lambda x: x['event_id']) if candidates else None

索引文件的存在使得"从断点恢复"变得高效——系统不需要扫描整个日志文件，只需要找到最近的检查点，然后从该位置开始重放后续事件。

断点恢复机制

检查点策略

检查点（Checkpoint）是断点恢复的基础。它本质上是一个状态快照，记录了到某个事件为止的完整会话状态。恢复时，系统只需要加载最近的检查点，然后重放后续事件即可。

检查点的创建时机有几种策略：

  1. 固定间隔：每 N 个事件创建一次检查点
  2. 时间间隔：每 M 分钟创建一次检查点
  3. 事件类型触发：在特定事件（如工具调用完成）后创建检查点
  4. 混合策略：结合以上多种条件
import hashlib
from typing import List, Dict, Any

class CheckpointManager:
    """检查点管理器 - 负责创建和管理会话检查点"""

    def __init__(self, max_events_between_checkpoints: int = 50,
                 max_seconds_between_checkpoints: int = 300):
        self.max_events = max_events_between_checkpoints
        self.max_seconds = max_seconds_between_checkpoints
        self.events_since_checkpoint = 0
        self.last_checkpoint_time = None

    def should_create_checkpoint(self, event: SessionEvent) -> bool:
        """判断是否应该创建检查点 - 混合策略"""
        self.events_since_checkpoint += 1

        # 策略1：工具调用完成后立即创建（关键恢复点）
        if event.type == EventType.TOOL_COMPLETED:
            return True

        # 策略2：固定事件间隔
        if self.events_since_checkpoint >= self.max_events:
            return True

        # 策略3：时间间隔（如果上次检查点时间已知）
        if self.last_checkpoint_time:
            elapsed = (datetime.fromisoformat(event.timestamp) -
                      self.last_checkpoint_time).total_seconds()
            if elapsed >= self.max_seconds:
                return True

        return False

    def create_checkpoint(self, events: List[SessionEvent],
                         current_state: Dict[str, Any]) -> SessionEvent:
        """创建检查点事件 - 包含状态哈希用于校验"""
        state_json = json.dumps(current_state, sort_keys=True, ensure_ascii=False)
        state_hash = hashlib.sha256(state_json.encode()).hexdigest()[:16]

        checkpoint = SessionEvent(
            type=EventType.CHECKPOINT,
            timestamp=datetime.utcnow().isoformat() + 'Z',
            event_id=f"cp_{state_hash}",
        )
        # 附加检查点元数据
        checkpoint.state_hash = state_hash
        checkpoint.event_count = len(events)
        checkpoint.state_snapshot = current_state

        self.events_since_checkpoint = 0
        self.last_checkpoint_time = datetime.utcnow()
        return checkpoint

这段代码展示了混合检查点策略的实现。注意 should_create_checkpoint 方法中的三重判断：工具调用完成后总是创建检查点（因为工具调用可能有副作用，需要精确恢复），同时还有基于事件数量和时间间隔的兜底策略。

恢复流程

当系统需要恢复会话状态时，执行以下流程：

  1. 定位检查点：从索引文件中找到最近的检查点
  2. 加载快照：读取检查点中保存的状态快照
  3. 重放事件：从检查点之后开始，依次重放每个事件
  4. 校验状态：对比最终状态的哈希值，确保恢复正确
class SessionRecovery:
    """会话恢复引擎 - 从检查点和事件日志重建会话状态"""

    def __init__(self, storage: 'SessionStorage'):
        self.storage = storage

    async def recover_session(self, session_id: str) -> Dict[str, Any]:
        """恢复会话状态 - 从最近检查点开始重放"""
        # 第一步：加载索引，找到最近的检查点
        index = await self.storage.load_index(session_id)
        if not index:
            raise ValueError(f"Session {session_id} not found")

        checkpoint = index.find_nearest_checkpoint(index.last_event_id)
        if checkpoint:
            # 第二步：从检查点加载状态快照
            state = checkpoint['state_snapshot']
            start_event_id = checkpoint['event_id']
            print(f"[Recovery] Loaded checkpoint at event {start_event_id}")
        else:
            # 无检查点，从头开始
            state = self._create_empty_state(session_id)
            start_event_id = None
            print(f"[Recovery] No checkpoint found, replaying from beginning")

        # 第三步：重放检查点之后的所有事件
        events = await self.storage.load_events_after(session_id, start_event_id)
        replayed_count = 0

        for event in events:
            state = self._apply_event(state, event)
            replayed_count += 1

        print(f"[Recovery] Replayed {replayed_count} events, "
              f"session state restored successfully")
        return state

    def _apply_event(self, state: Dict[str, Any],
                     event: SessionEvent) -> Dict[str, Any]:
        """将单个事件应用到状态上 - 状态机的核心转换函数"""
        if event.type == EventType.MESSAGE_ADDED:
            state['messages'].append({
                'role': event.role,
                'content': event.content,
                'timestamp': event.timestamp
            })
        elif event.type == EventType.TOOL_INVOKED:
            state['pending_tools'][event.invocation_id] = {
                'tool_name': event.tool_name,
                'input': event.input,
                'started_at': event.timestamp
            }
        elif event.type == EventType.TOOL_COMPLETED:
            tool_record = state['pending_tools'].pop(event.invocation_id, {})
            state['completed_tools'].append({
                **tool_record,
                'output': event.output,
                'duration_ms': event.duration_ms
            })
        elif event.type == EventType.CONTEXT_SUMMARIZED:
            # 上下文摘要事件：替换旧的对话历史为摘要版本
            state['messages'] = event.summarized_messages
            state['summary_applied_at'] = event.timestamp

        return state

这段恢复引擎的代码揭示了 Append-only 模式的核心优势：恢复过程是确定性的。给定相同的事件序列，_apply_event 总是产生相同的状态。这意味着你可以：

  • 在不同机器上恢复同一个会话
  • 对恢复过程进行单元测试
  • 在恢复过程中检测状态不一致

增量恢复与流式恢复

在实际生产环境中，完全重放所有事件可能很耗时。为此，可以采用两种优化策略：

增量恢复：维护多个检查点，恢复时只重放最近一个检查点之后的事件。这要求检查点创建策略更加积极。

流式恢复：在恢复过程中同时接收新事件。恢复引擎维护一个"追赶"状态——当历史事件重放完成时，新事件已经排好队等待处理。这对于用户在 Agent 恢复过程中继续发送消息的场景非常有用。

与传统数据库方案对比

关系型数据库方案

传统的关系型数据库（如 PostgreSQL）也可以用来存储会话状态，但存在以下问题：

对比维度Append-only 日志关系型数据库
写入性能追加写入，O(1) 复杂度需要索引维护，写放大严重
恢复能力天然支持任意时间点恢复需要额外实现 WAL 或备份机制
存储效率可压缩，事件粒度灵活行级存储，元数据开销大
查询灵活性按时间/类型查询高效复杂查询能力强
运维复杂度文件系统级别，极低需要 DBA 维护
适用规模单会话级别，GB 级别多会话级别，TB 级别

NoSQL 方案

MongoDB、DynamoDB 等 NoSQL 数据库在灵活性上更接近 Append-only 日志，但在"可追溯性"方面仍然不足。NoSQL 的文档模型天然适合存储会话记录，但要实现事件溯源，需要额外的变更为日志（Change Stream）或自定义事件表。

文件系统方案

Claude Code 等工具选择了最朴素的方案：直接使用文件系统。会话日志以 JSONL 文件存储在本地磁盘上，索引文件与之并列。这种方案的优势在于：

  • 零依赖：不需要安装和维护数据库
  • 可移植：用户可以直接复制会话文件
  • 可调试：用 cat 或 jq 就能查看会话内容
  • 隐私友好：数据完全在本地，不经过第三方服务

当然，文件系统方案的劣势也很明显：不支持跨设备同步、缺乏并发控制、大文件性能问题。但对于大多数 Agent 使用场景（单用户、单设备），这些劣势是可以接受的。

总结

Append-only 状态持久化是 AI Agent 会话存储的最佳实践，其核心优势在于：

  1. 可追溯性：完整记录每个状态变更，支持任意时间点回溯
  2. 容错性：系统崩溃后可以从检查点精确恢复，不丢失任何信息
  3. 确定性：事件重放是确定性的，恢复结果可验证
  4. 简洁性：基于文件系统的实现几乎零依赖

在设计自己的 Agent 系统时，建议从 JSONL + 检查点的简单方案开始，随着规模增长再逐步引入索引、压缩和分布式同步等高级特性。记住：状态管理的第一原则是不丢失状态，而不是优化查询性能。

参考资料

  1. Claude Code 源码 - Anthropic 的 CLI Agent 实现，展示了生产级会话存储设计（GitHub: anthropics/claude-code）
  2. Martin Fowler, “Event Sourcing” (2005) - Event Sourcing 模式的经典定义和设计原则
  3. Greg Young, “CQRS Documents” (2010) - 事件溯源与 CQRS 模式的深度结合
  4. Anthropic, “Claude Code: Best practices for agentic coding” (2025) - 官方文档中关于会话管理的工程实践
  5. Jay Kreps, “The Log: What every software engineer should know” (2013) - 关于日志作为数据基础设施的经典论文

本系列覆盖 AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理 七大方向，从入门到实战的全栈内容持续更新中。

所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。

👍 点赞 + ⭐ 关注，评论区扣「1」，挨个发你领取方式 👇

# 安全 # 人工智能
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

avatar AI Agent技术社区
cover

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

avatar AI Agent技术社区

写给 TypeScript 工程师的 OpenClaw.NET 上手指南：用你熟悉的 TS 思维，跑起一个生产级 AI Agent

读代码层：TS 和 C# 本来就是「远房亲戚」——asyncawait??、接口、泛型、严格空检查，你的心智模型几乎无缝平移。真正要适应的只有两件事：类型是具名而非结构化的（多一句显式声明），以及 NativeAOT 下的「禁反射」约束。写扩展层：你甚至可以不怎么碰 C#——Node.js 插件桥意味着你的 TypeScript 插件就是一等公民，而且能在生产 aot 车道跑。内核给你 .NET

avatar AI Agent技术社区