尧图网站建设 
会话存储Append-only 状态持久化当 Agent 执行到一半崩溃了你该怎么办从头重来还是从断点继续答案藏在存储设计里。前言AI Agent 的会话状态是整个系统中最脆弱也最关键的部分。一次典型的 Agent 执行可能跨越数十轮工具调用、数千个 token 的上下文以及数分钟乃至数小时的执行时间。在这漫长的生命周期中任何环节的异常——网络中断、进程崩溃、用户主动暂停——都可能导致状态丢失。传统的数据库方案采用最新值覆盖Last-Write-Wins策略来存储状态这在 Agent 场景下存在根本性缺陷你无法知道什么时候的值才是正确的也无法回溯到某个中间状态重新开始。本文将深入探讨Append-only 状态持久化模式这是 Claude Code 等主流 AI Agent 系统在实践中采用的核心设计哲学。会话状态的挑战状态的多维性Agent 会话状态远不止对话历史这么简单。一个完整的会话状态至少包含以下维度对话历史用户消息、助手回复、系统提示词工具调用记录每个工具的输入、输出、执行状态中间决策Agent 的推理链、分支选择、回退记录环境快照文件系统状态、代码变更、外部 API 响应元数据创建时间、最后修改时间、token 计数、模型参数这些维度之间存在复杂的依赖关系。如果你只保存了对话历史而丢失了工具调用记录那么后续的工具调用可能会产生重复副作用比如重复发送邮件、重复创建文件。一致性困境在分布式环境中会话状态的一致性面临三大挑战并发写入用户可能在 Agent 执行过程中发送新消息部分失败某个工具调用成功但后续步骤失败因果顺序消息的物理到达顺序可能与逻辑顺序不一致传统的事务模型ACID在这种场景下过于重量级而最终一致性模型又无法满足 Agent 对可恢复性的强需求。我们需要一种介于两者之间的方案。Append-only 设计哲学核心思想只追加不修改Append-only仅追加存储的核心原则极为简单任何状态变更都以新记录的形式追加到日志末尾永不修改已有记录。这看似低效却解决了 Agent 状态管理中的几乎所有核心问题。Append-only方式event: Aevent: Bevent: Cevent: D传统方式覆盖覆盖state Astate Bstate 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事件序列你可以精确重建任意时间点的会话状态。为什么不用 WALWrite-Ahead Log预写日志是数据库领域常见的持久化策略但 WAL 和 Append-only 存储有几个关键区别特性WALAppend-only 日志目的保证事务原子性保证状态可追溯性生命周期事务完成后可清理永久保留或按策略归档查询能力不支持直接查询支持按时间/类型查询存储格式二进制、内部格式结构化、可读格式适用场景数据库内部机制应用层状态管理对于 Agent 系统我们既需要 WAL 的原子性保证也需要 Append-only 的可追溯性。实际上很多 Agent 框架会同时使用两者WAL 用于保证单次操作的原子性Append-only 日志用于跨操作的状态追踪。存储格式与序列化JSONL简单但有效Claude Code 的会话存储采用了 JSONLJSON Lines格式每行一个 JSON 对象代表一个事件。这种格式有几个显著优势流式写入每个事件可以独立序列化和写入无需等待整个文件完成容错性某一行损坏不影响其他行的读取可读性人类可以直接阅读和调试跨语言任何编程语言都能轻松解析一个典型的会话日志文件可能看起来像这样{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}消息序列化策略对话历史是会话状态中体积最大的部分。对于长对话需要精心设计序列化策略importjsonfromdataclassesimportdataclass,asdictfromtypingimportList,OptionalfromdatetimeimportdatetimefromenumimportEnumclassEventType(Enum):SESSION_STARTsession_startMESSAGE_ADDEDmessage_addedTOOL_INVOKEDtool_invokedTOOL_COMPLETEDtool_completedCHECKPOINTcheckpointCONTEXT_SUMMARIZEDcontext_summarizeddataclassclassSessionEvent:会话事件基类 - 所有状态变更都通过此结构记录type:EventType timestamp:strevent_id:str# 全局唯一事件ID用于去重和排序defto_jsonl(self)-str:序列化为 JSONL 格式单行 JSONdataasdict(self)data[type]self.type.value# Enum 转字符串returnjson.dumps(data,ensure_asciiFalse)这段代码定义了事件的基础结构。注意event_id字段——它是保证事件幂等性的关键。即使同一个事件被意外写入两次系统也能通过event_id去重。索引与压缩随着会话日志增长直接顺序扫描的效率会下降。实践中通常会维护一个索引文件记录关键事件的偏移量dataclassclassSessionIndex:会话索引 - 记录关键事件的位置加速随机访问session_id:strcheckpoints:List[dict]# 检查点列表{event_id, offset, timestamp}message_offsets:List[dict]# 消息偏移量{event_id, offset, role}last_event_id:str# 最后一个事件的IDtotal_events:int# 总事件数deffind_nearest_checkpoint(self,target_event_id:str)-Optional[dict]:找到目标事件之前最近的检查点candidates[cpforcpinself.checkpointsifcp[event_id]target_event_id]returnmax(candidates,keylambdax:x[event_id])ifcandidateselseNone索引文件的存在使得从断点恢复变得高效——系统不需要扫描整个日志文件只需要找到最近的检查点然后从该位置开始重放后续事件。断点恢复机制检查点策略检查点Checkpoint是断点恢复的基础。它本质上是一个状态快照记录了到某个事件为止的完整会话状态。恢复时系统只需要加载最近的检查点然后重放后续事件即可。检查点的创建时机有几种策略固定间隔每 N 个事件创建一次检查点时间间隔每 M 分钟创建一次检查点事件类型触发在特定事件如工具调用完成后创建检查点混合策略结合以上多种条件importhashlibfromtypingimportList,Dict,AnyclassCheckpointManager:检查点管理器 - 负责创建和管理会话检查点def__init__(self,max_events_between_checkpoints:int50,max_seconds_between_checkpoints:int300):self.max_eventsmax_events_between_checkpoints self.max_secondsmax_seconds_between_checkpoints self.events_since_checkpoint0self.last_checkpoint_timeNonedefshould_create_checkpoint(self,event:SessionEvent)-bool:判断是否应该创建检查点 - 混合策略self.events_since_checkpoint1# 策略1工具调用完成后立即创建关键恢复点ifevent.typeEventType.TOOL_COMPLETED:returnTrue# 策略2固定事件间隔ifself.events_since_checkpointself.max_events:returnTrue# 策略3时间间隔如果上次检查点时间已知ifself.last_checkpoint_time:elapsed(datetime.fromisoformat(event.timestamp)-self.last_checkpoint_time).total_seconds()ifelapsedself.max_seconds:returnTruereturnFalsedefcreate_checkpoint(self,events:List[SessionEvent],current_state:Dict[str,Any])-SessionEvent:创建检查点事件 - 包含状态哈希用于校验state_jsonjson.dumps(current_state,sort_keysTrue,ensure_asciiFalse)state_hashhashlib.sha256(state_json.encode()).hexdigest()[:16]checkpointSessionEvent(typeEventType.CHECKPOINT,timestampdatetime.utcnow().isoformat()Z,event_idfcp_{state_hash},)# 附加检查点元数据checkpoint.state_hashstate_hash checkpoint.event_countlen(events)checkpoint.state_snapshotcurrent_state self.events_since_checkpoint0self.last_checkpoint_timedatetime.utcnow()returncheckpoint这段代码展示了混合检查点策略的实现。注意should_create_checkpoint方法中的三重判断工具调用完成后总是创建检查点因为工具调用可能有副作用需要精确恢复同时还有基于事件数量和时间间隔的兜底策略。恢复流程当系统需要恢复会话状态时执行以下流程定位检查点从索引文件中找到最近的检查点加载快照读取检查点中保存的状态快照重放事件从检查点之后开始依次重放每个事件校验状态对比最终状态的哈希值确保恢复正确classSessionRecovery:会话恢复引擎 - 从检查点和事件日志重建会话状态def__init__(self,storage:SessionStorage):self.storagestorageasyncdefrecover_session(self,session_id:str)-Dict[str,Any]:恢复会话状态 - 从最近检查点开始重放# 第一步加载索引找到最近的检查点indexawaitself.storage.load_index(session_id)ifnotindex:raiseValueError(fSession{session_id}not found)checkpointindex.find_nearest_checkpoint(index.last_event_id)ifcheckpoint:# 第二步从检查点加载状态快照statecheckpoint[state_snapshot]start_event_idcheckpoint[event_id]print(f[Recovery] Loaded checkpoint at event{start_event_id})else:# 无检查点从头开始stateself._create_empty_state(session_id)start_event_idNoneprint(f[Recovery] No checkpoint found, replaying from beginning)# 第三步重放检查点之后的所有事件eventsawaitself.storage.load_events_after(session_id,start_event_id)replayed_count0foreventinevents:stateself._apply_event(state,event)replayed_count1print(f[Recovery] Replayed{replayed_count}events, fsession state restored successfully)returnstatedef_apply_event(self,state:Dict[str,Any],event:SessionEvent)-Dict[str,Any]:将单个事件应用到状态上 - 状态机的核心转换函数ifevent.typeEventType.MESSAGE_ADDED:state[messages].append({role:event.role,content:event.content,timestamp:event.timestamp})elifevent.typeEventType.TOOL_INVOKED:state[pending_tools][event.invocation_id]{tool_name:event.tool_name,input:event.input,started_at:event.timestamp}elifevent.typeEventType.TOOL_COMPLETED:tool_recordstate[pending_tools].pop(event.invocation_id,{})state[completed_tools].append({**tool_record,output:event.output,duration_ms:event.duration_ms})elifevent.typeEventType.CONTEXT_SUMMARIZED:# 上下文摘要事件替换旧的对话历史为摘要版本state[messages]event.summarized_messages state[summary_applied_at]event.timestampreturnstate这段恢复引擎的代码揭示了 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 会话存储的最佳实践其核心优势在于可追溯性完整记录每个状态变更支持任意时间点回溯容错性系统崩溃后可以从检查点精确恢复不丢失任何信息确定性事件重放是确定性的恢复结果可验证简洁性基于文件系统的实现几乎零依赖在设计自己的 Agent 系统时建议从 JSONL 检查点的简单方案开始随着规模增长再逐步引入索引、压缩和分布式同步等高级特性。记住状态管理的第一原则是不丢失状态而不是优化查询性能。参考资料Claude Code 源码- Anthropic 的 CLI Agent 实现展示了生产级会话存储设计GitHub: anthropics/claude-codeMartin Fowler, “Event Sourcing” (2005)- Event Sourcing 模式的经典定义和设计原则Greg Young, “CQRS Documents” (2010)- 事件溯源与 CQRS 模式的深度结合Anthropic, “Claude Code: Best practices for agentic coding” (2025)- 官方文档中关于会话管理的工程实践Jay Kreps, “The Log: What every software engineer should know” (2013)- 关于日志作为数据基础设施的经典论文本系列覆盖AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理七大方向从入门到实战的全栈内容持续更新中。所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。 点赞 ⭐ 关注评论区扣「1」挨个发你领取方式 AG-16_会话存储：Append-only 状态持久化
会话存储Append-only 状态持久化当 Agent 执行到一半崩溃了你该怎么办从头重来还是从断点继续答案藏在存储设计里。前言AI Agent 的会话状态是整个系统中最脆弱也最关键的部分。一次典型的 Agent 执行可能跨越数十轮工具调用、数千个 token 的上下文以及数分钟乃至数小时的执行时间。在这漫长的生命周期中任何环节的异常——网络中断、进程崩溃、用户主动暂停——都可能导致状态丢失。传统的数据库方案采用最新值覆盖Last-Write-Wins策略来存储状态这在 Agent 场景下存在根本性缺陷你无法知道什么时候的值才是正确的也无法回溯到某个中间状态重新开始。本文将深入探讨Append-only 状态持久化模式这是 Claude Code 等主流 AI Agent 系统在实践中采用的核心设计哲学。会话状态的挑战状态的多维性Agent 会话状态远不止对话历史这么简单。一个完整的会话状态至少包含以下维度对话历史用户消息、助手回复、系统提示词工具调用记录每个工具的输入、输出、执行状态中间决策Agent 的推理链、分支选择、回退记录环境快照文件系统状态、代码变更、外部 API 响应元数据创建时间、最后修改时间、token 计数、模型参数这些维度之间存在复杂的依赖关系。如果你只保存了对话历史而丢失了工具调用记录那么后续的工具调用可能会产生重复副作用比如重复发送邮件、重复创建文件。一致性困境在分布式环境中会话状态的一致性面临三大挑战并发写入用户可能在 Agent 执行过程中发送新消息部分失败某个工具调用成功但后续步骤失败因果顺序消息的物理到达顺序可能与逻辑顺序不一致传统的事务模型ACID在这种场景下过于重量级而最终一致性模型又无法满足 Agent 对可恢复性的强需求。我们需要一种介于两者之间的方案。Append-only 设计哲学核心思想只追加不修改Append-only仅追加存储的核心原则极为简单任何状态变更都以新记录的形式追加到日志末尾永不修改已有记录。这看似低效却解决了 Agent 状态管理中的几乎所有核心问题。Append-only方式event: Aevent: Bevent: Cevent: D传统方式覆盖覆盖state Astate Bstate 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事件序列你可以精确重建任意时间点的会话状态。为什么不用 WALWrite-Ahead Log预写日志是数据库领域常见的持久化策略但 WAL 和 Append-only 存储有几个关键区别特性WALAppend-only 日志目的保证事务原子性保证状态可追溯性生命周期事务完成后可清理永久保留或按策略归档查询能力不支持直接查询支持按时间/类型查询存储格式二进制、内部格式结构化、可读格式适用场景数据库内部机制应用层状态管理对于 Agent 系统我们既需要 WAL 的原子性保证也需要 Append-only 的可追溯性。实际上很多 Agent 框架会同时使用两者WAL 用于保证单次操作的原子性Append-only 日志用于跨操作的状态追踪。存储格式与序列化JSONL简单但有效Claude Code 的会话存储采用了 JSONLJSON Lines格式每行一个 JSON 对象代表一个事件。这种格式有几个显著优势流式写入每个事件可以独立序列化和写入无需等待整个文件完成容错性某一行损坏不影响其他行的读取可读性人类可以直接阅读和调试跨语言任何编程语言都能轻松解析一个典型的会话日志文件可能看起来像这样{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}消息序列化策略对话历史是会话状态中体积最大的部分。对于长对话需要精心设计序列化策略importjsonfromdataclassesimportdataclass,asdictfromtypingimportList,OptionalfromdatetimeimportdatetimefromenumimportEnumclassEventType(Enum):SESSION_STARTsession_startMESSAGE_ADDEDmessage_addedTOOL_INVOKEDtool_invokedTOOL_COMPLETEDtool_completedCHECKPOINTcheckpointCONTEXT_SUMMARIZEDcontext_summarizeddataclassclassSessionEvent:会话事件基类 - 所有状态变更都通过此结构记录type:EventType timestamp:strevent_id:str# 全局唯一事件ID用于去重和排序defto_jsonl(self)-str:序列化为 JSONL 格式单行 JSONdataasdict(self)data[type]self.type.value# Enum 转字符串returnjson.dumps(data,ensure_asciiFalse)这段代码定义了事件的基础结构。注意event_id字段——它是保证事件幂等性的关键。即使同一个事件被意外写入两次系统也能通过event_id去重。索引与压缩随着会话日志增长直接顺序扫描的效率会下降。实践中通常会维护一个索引文件记录关键事件的偏移量dataclassclassSessionIndex:会话索引 - 记录关键事件的位置加速随机访问session_id:strcheckpoints:List[dict]# 检查点列表{event_id, offset, timestamp}message_offsets:List[dict]# 消息偏移量{event_id, offset, role}last_event_id:str# 最后一个事件的IDtotal_events:int# 总事件数deffind_nearest_checkpoint(self,target_event_id:str)-Optional[dict]:找到目标事件之前最近的检查点candidates[cpforcpinself.checkpointsifcp[event_id]target_event_id]returnmax(candidates,keylambdax:x[event_id])ifcandidateselseNone索引文件的存在使得从断点恢复变得高效——系统不需要扫描整个日志文件只需要找到最近的检查点然后从该位置开始重放后续事件。断点恢复机制检查点策略检查点Checkpoint是断点恢复的基础。它本质上是一个状态快照记录了到某个事件为止的完整会话状态。恢复时系统只需要加载最近的检查点然后重放后续事件即可。检查点的创建时机有几种策略固定间隔每 N 个事件创建一次检查点时间间隔每 M 分钟创建一次检查点事件类型触发在特定事件如工具调用完成后创建检查点混合策略结合以上多种条件importhashlibfromtypingimportList,Dict,AnyclassCheckpointManager:检查点管理器 - 负责创建和管理会话检查点def__init__(self,max_events_between_checkpoints:int50,max_seconds_between_checkpoints:int300):self.max_eventsmax_events_between_checkpoints self.max_secondsmax_seconds_between_checkpoints self.events_since_checkpoint0self.last_checkpoint_timeNonedefshould_create_checkpoint(self,event:SessionEvent)-bool:判断是否应该创建检查点 - 混合策略self.events_since_checkpoint1# 策略1工具调用完成后立即创建关键恢复点ifevent.typeEventType.TOOL_COMPLETED:returnTrue# 策略2固定事件间隔ifself.events_since_checkpointself.max_events:returnTrue# 策略3时间间隔如果上次检查点时间已知ifself.last_checkpoint_time:elapsed(datetime.fromisoformat(event.timestamp)-self.last_checkpoint_time).total_seconds()ifelapsedself.max_seconds:returnTruereturnFalsedefcreate_checkpoint(self,events:List[SessionEvent],current_state:Dict[str,Any])-SessionEvent:创建检查点事件 - 包含状态哈希用于校验state_jsonjson.dumps(current_state,sort_keysTrue,ensure_asciiFalse)state_hashhashlib.sha256(state_json.encode()).hexdigest()[:16]checkpointSessionEvent(typeEventType.CHECKPOINT,timestampdatetime.utcnow().isoformat()Z,event_idfcp_{state_hash},)# 附加检查点元数据checkpoint.state_hashstate_hash checkpoint.event_countlen(events)checkpoint.state_snapshotcurrent_state self.events_since_checkpoint0self.last_checkpoint_timedatetime.utcnow()returncheckpoint这段代码展示了混合检查点策略的实现。注意should_create_checkpoint方法中的三重判断工具调用完成后总是创建检查点因为工具调用可能有副作用需要精确恢复同时还有基于事件数量和时间间隔的兜底策略。恢复流程当系统需要恢复会话状态时执行以下流程定位检查点从索引文件中找到最近的检查点加载快照读取检查点中保存的状态快照重放事件从检查点之后开始依次重放每个事件校验状态对比最终状态的哈希值确保恢复正确classSessionRecovery:会话恢复引擎 - 从检查点和事件日志重建会话状态def__init__(self,storage:SessionStorage):self.storagestorageasyncdefrecover_session(self,session_id:str)-Dict[str,Any]:恢复会话状态 - 从最近检查点开始重放# 第一步加载索引找到最近的检查点indexawaitself.storage.load_index(session_id)ifnotindex:raiseValueError(fSession{session_id}not found)checkpointindex.find_nearest_checkpoint(index.last_event_id)ifcheckpoint:# 第二步从检查点加载状态快照statecheckpoint[state_snapshot]start_event_idcheckpoint[event_id]print(f[Recovery] Loaded checkpoint at event{start_event_id})else:# 无检查点从头开始stateself._create_empty_state(session_id)start_event_idNoneprint(f[Recovery] No checkpoint found, replaying from beginning)# 第三步重放检查点之后的所有事件eventsawaitself.storage.load_events_after(session_id,start_event_id)replayed_count0foreventinevents:stateself._apply_event(state,event)replayed_count1print(f[Recovery] Replayed{replayed_count}events, fsession state restored successfully)returnstatedef_apply_event(self,state:Dict[str,Any],event:SessionEvent)-Dict[str,Any]:将单个事件应用到状态上 - 状态机的核心转换函数ifevent.typeEventType.MESSAGE_ADDED:state[messages].append({role:event.role,content:event.content,timestamp:event.timestamp})elifevent.typeEventType.TOOL_INVOKED:state[pending_tools][event.invocation_id]{tool_name:event.tool_name,input:event.input,started_at:event.timestamp}elifevent.typeEventType.TOOL_COMPLETED:tool_recordstate[pending_tools].pop(event.invocation_id,{})state[completed_tools].append({**tool_record,output:event.output,duration_ms:event.duration_ms})elifevent.typeEventType.CONTEXT_SUMMARIZED:# 上下文摘要事件替换旧的对话历史为摘要版本state[messages]event.summarized_messages state[summary_applied_at]event.timestampreturnstate这段恢复引擎的代码揭示了 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 会话存储的最佳实践其核心优势在于可追溯性完整记录每个状态变更支持任意时间点回溯容错性系统崩溃后可以从检查点精确恢复不丢失任何信息确定性事件重放是确定性的恢复结果可验证简洁性基于文件系统的实现几乎零依赖在设计自己的 Agent 系统时建议从 JSONL 检查点的简单方案开始随着规模增长再逐步引入索引、压缩和分布式同步等高级特性。记住状态管理的第一原则是不丢失状态而不是优化查询性能。参考资料Claude Code 源码- Anthropic 的 CLI Agent 实现展示了生产级会话存储设计GitHub: anthropics/claude-codeMartin Fowler, “Event Sourcing” (2005)- Event Sourcing 模式的经典定义和设计原则Greg Young, “CQRS Documents” (2010)- 事件溯源与 CQRS 模式的深度结合Anthropic, “Claude Code: Best practices for agentic coding” (2025)- 官方文档中关于会话管理的工程实践Jay Kreps, “The Log: What every software engineer should know” (2013)- 关于日志作为数据基础设施的经典论文本系列覆盖AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理七大方向从入门到实战的全栈内容持续更新中。所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。 点赞 ⭐ 关注评论区扣「1」挨个发你领取方式