自改进代理Self-Improving Agents和事件溯源Event-Sourcing看似属于两个不同的技术领域但当把代理的持续改进当成一个工程问题来看待时两者的关系非常紧密。自改进代理真正需要的不是“能改自己”这个能力而是一套可审计、可重放、可回滚的经验记录机制事件溯源恰好提供了一套成熟的状态演化模型。本文会从底层原理讲起逐步落地到一个最小可运行的 Python 示例并给出验证方法、常见排错路径和生产化建议。1. 为什么自改进代理会和事件溯源绑定在一起1.1 自改进代理的真正难点不在模型而在经验层很多人理解 self-improving agents 时第一反应是“用新的数据去微调或重新训练模型”。这个理解只有一半正确。在真实业务里代理的自改进往往不来自底层模型参数的变化而是来自上层环节的调整提示词模板、工具调用策略、问题拆解方式、检索知识库里的经验、打分规则、反思总结这些都可以成为持续改进的来源。换句话说模型参数在大多数生产场景里是相对固定的可变的通常是“策略层”和“经验层”。策略层的变化决定了一次迭代是否可观测、可回滚。一个需要自改进的代理至少要留下四类信息遇到什么情况context、observation、输入快照。当时做了什么所选的动作、调用的工具、生成的回复。结果如何reward、用户反馈、错误信息、任务是否成功。如果再来一次有什么要改提炼出的教训、修正后的规则。这四类信息不是孤立的它们构成了一条因果链。举例来说一个客服代理今天 10:00 遇到用户投诉物流异常采用了一套话术模板结果用户满意度很低于是晚间自动改进了话术模板。第二天要验证改进是否有效就必须把前一天 10:00 的那次事件和晚间提交策略更新的事件关联起来。如果系统里只保存“当前话术模板是 A”这样的最终状态中间的尝试、失败、教训、调整就全部丢失了改进闭环自然无法成立。所以自改进代理的工程难点是把“尝试、失败、教训、调整”整条因果链变成结构化、可重放的数据。这正是事件溯源擅长解决的问题。1.2 事件溯源的核心直觉状态只是事件流的投影事件溯源的基本思想是业务系统不把“最新状态”当作唯一的事实来源而是只保存每个状态变更事件当前状态由事件流推导出来。推导过程称为投影projection从零开始重放所有事件可以重建任意历史时刻的状态。用一个通俗类比说明传统记账方式账本上写“当前余额为 1000 元”。事件溯源方式账本上只写“收入 300 元”“支出 200 元”“收入 900 元”余额通过累加所有交易得到。传统方式简单直接但丢失了“这 1000 元如何形成”的证据。事件溯源虽然需要额外做一次读取和累加却完整保留了从零到当前状态的因果链。代理系统的当前状态至少包括当前策略版本号。积累下来的知识条目。反思总结出的一组 lessons。历史决策统计和累计 reward。最近一次快照指针。直接把这些状态落库也不难但问题在于代理的“自我改进”本质上是一种状态转换旧的策略版本加新的失败经验推导出新的策略版本。只有把“旧版本快照”和“导致版本变化的事件”同时保存下来才能回答三个问题当前代理为什么变成现在这样上一次更新是否真的改善了结果如果产生了回归能否干净地回退到上一个版本事件溯源并不是银弹它也有重放成本高、存储量大、事件结构兼容困难等缺点。但在需要长期演进、持续改进、严格审计的 agent 系统里这些代价是可以被收益覆盖的。1.3 两者如何对上改进等于重放一段历史经验把“自改进”与“事件溯源”映射起来可以得到一张清晰的对应表自改进代理的环节事件溯源中的概念对应事件举例记录当前环境与任务事实事件ObservationRecorded做出一次决策命令/动作事件ActionTaken拿到结果反馈结果事件RewardObserved、ErrorOccurred总结教训经验事件LessonExtracted修改策略聚合状态变更PolicyRevisionCommitted保存可运行版本快照SnapshotCreated这个映射说明了一个自改进代理并不是“一个持续运行的程序”而是“一系列事件的累计效果”。每次加载一个代理本质上是从某个事件起点重放出一条策略每次自我改进本质上是追加一条策略更新事件并刷新快照。接受这个映射后很多原本难处理的问题会变得清晰想回退到三天前的策略版本加载三天前的事件快照即可。想评估两种策略的好坏可以 fork 出两条事件分支分别测试。想确认某次反思是否真正有效可以只重放改进前后的事件比较各自决策输出。想排查一次幻觉或错误回复可以沿事件链回溯找到最早的错误输入。因此“Self-Improving Agents Are Event-Sourced”本质上是一个架构判断在代理需要持续从经验中进化时事件日志不是可选项而是保证可审计、可复现、可回退的基础设施。2. 设计一套用于代理改进的事件模型2.1 代理应该记录哪些事件设计事件模型时核心原则是记录“发生了什么”而不是“最终结果是什么”。每条事件必须携带足够的信息让另一个程序或者未来的同一个程序能够重建当时的情境。在常见的任务循环里代理事件可以分成五类生命周期事件AgentRegistered、AgentRetired。感知事件ObservationRecorded、ContextLoaded。决策事件ActionTaken、ToolInvoked、ResponseGenerated。反馈事件RewardObserved、UserFeedbackRecorded、ErrorOccurred。改进事件LessonExtracted、PolicyRevisionCommitted、SnapshotCreated。事件模型中应该有一个稳定的公共结构包含 event_id、aggregate_id、event_type、occurred_at、version、payload 这几个字段。aggregate_id 可以指向一个对话会话或一个代理实例让所有事件归属到同一个可查询主体上。下面是一个通用 payload 示例。实际项目里字段往往更多但这个结构可以作为起点。{ event_id: evt_01HKZX9Q2F0A1B2C3D4E5F6, aggregate_id: agent_support_001, event_type: ActionTaken, occurred_at: 2025-06-18T10:00:00Z, version: 1, payload: { task_id: task_1024, input_snapshot: { user_question: 我的订单显示派送中但超过48小时没有更新, intent: order_delivery_status }, chosen_action: use_tool:query_logistics, tool_params: { order_id: SO20250618001 }, context_refs: [ knowledge/policy_v20250601, memory/session_998 ] } }这段 JSON 里最容易被忽略的是context_refs。它把“决策”与“当时的背景”挂钩。如果没有 context回放或者排查时事件的还原价值会大打折扣。还需要注意不要把原始对话和完整模型输出全部塞进事件。实际项目里可以采用“小事件加大对象引用”策略事件只保存结构化摘要、hash、引用地址。完整的大文本比如几千字的 prompt 或模型输出放到对象存储或向量库中按 key 关联。离线分析时再按引用加载完整对象。2.2 事件负载的版本管理事件结构一定会随项目演进发生变化。最稳妥的做法是给每个事件类型增加 schema_version 字段并保留旧版本解析函数。{ event_type: ActionTaken, schema_version: 2, payload: { task_id: task_1024, input_snapshot: {}, chosen_action: use_tool:query_logistics, tool_params: {}, context_refs: [], latency_ms: 320 } }当同一事件类型出现多个版本时投影逻辑要通过 schema_version 分发到对应解析函数。否则旧事件一旦无法解析整个事件流的重放就会中断。2.3 事件模型与现有 agent 框架的关系目前很多 agent 框架已经有轨迹记录和记忆机制例如会话历史、trace、教训缓存、向量记忆。事件模型并不是要推翻这些机制而是要在它们之下补一层不可变因果日志。建议的分层如下层作用典型实现应用层与用户对话、调用工具、调用大模型自研循环、LangChain 等记忆和状态层当前会话上下文、当前策略缓存Memory、PolicyStore、Cache事件日志层不可变事实存储用于重放和改进PostgreSQL、Kafka、Event Store记忆层是可变的缓存事件日志层才是不可变事实。代理每次运行时应先通过事件日志重建记忆而不是直接信任进程内残留状态。这套分层也方便未来扩展离线分析、强化学习采样和审计查询。3. 从事件流构造一个自改进代理的最小示例3.1 目录结构和组件划分这里用 Python 实现一个最小可运行框架。它不接入具体大模型只把事件日志、策略投影、改进循环的核心逻辑展示出来方便替换成真实模型调用。self_improving_agent/ ├── event_store.py # 事件存储示例里用 SQLite ├── agent.py # 代理投影从事件重建状态 ├── reflection.py # 教训提取与策略修订 ├── snapshot.py # 快照读写 ├── main.py # 演示运行入口 └── schema.sql # 建表 SQLSQLite 适合学习和验证。生产环境建议使用 PostgreSQL 或者专门的 Event Store并设置分区、索引和权限隔离。先看建表 SQLCREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL UNIQUE, aggregate_id TEXT NOT NULL, event_type TEXT NOT NULL, occurred_at TEXT NOT NULL, version INTEGER NOT NULL, payload TEXT NOT NULL ); CREATE INDEX idx_events_aggregate_time ON events (aggregate_id, occurred_at);这个表结构抓住了事件溯源最核心的外型只追加不修改历史事件。新增事件使用 INSERT查询使用按 aggregate_id 和 occurred_at 排序的 SELECT修改数据只能通过追加修正事件不能 UPDATE 旧行。3.2 加载策略通过重放事件恢复状态代理的“当前状态”由事件投影而来。为了说明原理先演示从空状态开始完整重放。# agent.py class AgentProjection: def __init__(self, aggregate_id): self.aggregate_id aggregate_id self.policy_revision 0 self.lessons [] self.action_count 0 self.total_reward 0.0 def apply(self, event): event_type event[event_type] payload event[payload] if event_type PolicyRevisionCommitted: self.policy_revision payload.get(revision) self.lessons payload.get(lessons, self.lessons) elif event_type ActionTaken: self.action_count 1 elif event_type RewardObserved: self.total_reward float(payload.get(score, 0)) def load_agent(event_store, aggregate_id): events event_store.fetch_events(aggregate_id) projection AgentProjection(aggregate_id) for event in events: projection.apply(event) return projection这就是“可重放”的具体含义给同一批事件无论重放几次都能得到相同的 policy_revision、lessons 和累计 reward。只要事件不变投影结果就确定。3.3 执行任务并收集经验执行任务时代理需要把“观察、动作、结果”完整记录成事件。一个简化版如下import uuid def run_task(agent_projection, event_store, task_input, policy_fn): task_id uuid.uuid4().hex event_store.append({ aggregate_id: agent_projection.aggregate_id, event_type: ObservationRecorded, payload: { task_id: task_id, input_snapshot: task_input, lessons_at_start: list(agent_projection.lessons) } }) action policy_fn(task_input, agent_projection) event_store.append({ aggregate_id: agent_projection.aggregate_id, event_type: ActionTaken, payload: { task_id: task_id, action: action } }) result_score evaluate_result(action, task_input) event_store.append({ aggregate_id: agent_projection.aggregate_id, event_type: RewardObserved, payload: { task_id: task_id, score: result_score } }) return task_id上面代码里evaluate_result是一个占位函数。真实项目中它可以接入自动评测集、用户反馈、规则校验或人工标注。关键在于无论结果来自哪里都要以事件形式落到日志中因为一次“自改进”就是从这些经验事件中提炼出来的。3.4 提取反思并提交策略更新自改进的核心动作是扫描最近一段事件找出失败模式得到一条可复用的规则然后提交为新的策略版本。# reflection.py def extract_lessons(events_since_last_revision): failed [ e for e in events_since_last_revision if e[event_type] RewardObserved and float(e[payload].get(score, 0)) 0.5 ] if not failed: return [] # 实际项目中通常需要大模型总结共同失败原因 # 这里用占位逻辑演示合并失败样本中的 intent。 intents [] for e in failed: obs e[payload].get(input_snapshot, {}) if obs.get(intent): intents.append(obs[intent]) return [遇到意图 str(intents) 时先查询日志再回复] def commit_policy_update(event_store, aggregate_id, old_revision, new_revision, lessons): event_store.append({ aggregate_id: aggregate_id, event_type: PolicyRevisionCommitted, payload: { old_revision: old_revision, revision: new_revision, lessons: lessons, reason: batch reflection after task group run, created_by: reflection_worker } })策略更新本身也作为事件写入这一点很重要。这样任何一次“代理变好还是变坏”的变化都能精确对应到一个事件。排查回归问题时可以直接定位 old_revision 到 new_revision 的切换点。3.5 快照与分支控制重放成本和回滚风险当事件总量增大后每次都从第一条事件重放会越来越慢。事件溯源的标准解法是快照snapshot。# snapshot.py def save_snapshot(projection, event_store): last_event event_store.get_last_event(projection.aggregate_id) snapshot { aggregate_id: projection.aggregate_id, last_event_id: last_event[event_id], state: { policy_revision: projection.policy_revision, lessons: projection.lessons, action_count: projection.action_count, total_reward: projection.total_reward } } write_snapshot_to_disk_or_db(snapshot) return snapshot加载时先读快照再补放快照之后的事件def load_agent_with_snapshot(snapshot_store, event_store, aggregate_id): snapshot snapshot_store.load(aggregate_id) projection AgentProjection(aggregate_id) projection.apply_snapshot(snapshot[state]) for event in event_store.fetch_events_after(aggregate_id, snapshot[last_event_id]): projection.apply(event) return projection“分支”同样简单如果提交一次策略更新后发现回归不需要删除事件只需从失败的策略修订事件之前 fork 出一个新的 aggregate_id重放到该时间点再追加新的策略尝试。注意不要把“回滚”理解成删除事件。事件溯源里历史是事实删除历史会导致整个链路的因果失真。正确的回滚是从某时间点分支出一条新路径再在新的分支上继续改进。4. 如何验证这个设计是可用的4.1 必须验证的五个结论把代理改成事件溯源后不能只检查程序能启动还要验证事件模型是否带来了预期能力。至少验证以下五项重放一致性同样一批事件重放多次得到的聚合状态完全一致。无副作用重放重放事件时不会触发外部调用比如重复发邮件、重复调支付接口。快照等价性从“快照加增量事件”恢复的状态与从空库完整重放的状态一致。改进有效性基于同一批历史经验提交的新策略在新数据集上确实优于旧策略。回滚可用性某个策略更新导致明显下降时能够快速 fork 出前一版本并继续运行。在这五项里第 4 项最容易缺失。很多团队做到“代理能改自己的提示词”就停下却不统计改进前后在评测集上的分数变化导致所谓的 self-improving 变成自我折腾。4.2 回归验证脚本把验证固化下来建议写成自动化测试def assert_replay_consistency(event_store, aggregate_id): s1 load_agent(event_store, aggregate_id) s2 load_agent(event_store, aggregate_id) assert s1.policy_revision s2.policy_revision assert s1.lessons s2.lessons assert s1.total_reward s2.total_reward def assert_snapshot_equivalence(snapshot_store, event_store, aggregate_id): full load_agent(event_store, aggregate_id) with_snapshot load_agent_with_snapshot(snapshot_store, event_store, aggregate_id) assert full.__dict__ with_snapshot.__dict__运行方法python -m pytest tests/test_replay.py -v大型项目中建议把“无副作用重放”也作为一个测试规范。做法是在重放上下文里注入一个 mock 外部客户端任何外部调用都会让测试失败。这能有效拦截未来可能破坏重放纯度的代码改动。4.3 从指标到日志的追踪事件溯源的最大价值体现在审计链路上。一个改进循环完成后至少要能回答某次PolicyRevisionCommitted是由哪些RewardObserved和ErrorOccurred派生出来的。可以这样查询SELECT event_id, event_type, occurred_at, json_extract(payload, $.task_id) AS task_id FROM events WHERE aggregate_id agent_support_001 AND event_type IN (RewardObserved, PolicyRevisionCommitted) ORDER BY occurred_at;通过把 task_id 作为关联键可以串联出“观测到动作到奖励到教训到策略更新”的完整因果链。5. 常见问题与排错路径5.1 问题速查表问题现象常见原因检查方式处理建议重放后策略状态与线上不一致快照状态携带了事件流之外的隐式状态对比快照 last_event_id 与实际最后事件检查是否遗漏写入上下文引用把隐式变量也写入投影同一个流程重放多次结果不同事件重放时触发了外部副作用在重放上下文里打开外部调用 mock将外部调用与重放隔离事件里只存调用参数和结果事件表增长过快查询变慢没有为 aggregate_id 和 occurred_at 建索引或没有滚动快照查看慢查询与表大小增加索引、定期生成快照、大字段放外部存储策略更新后出现回归不确定是哪次触发每次策略更新没有可追踪的聚合 id 和版本差检查两次 PolicyRevisionCommitted 之间的事件用事件分支对比前后评测分数反思逻辑总结错误污染策略基于太少样本总结或缺乏足够上下文检查 LessonExtracted 的 payload 是否包含足够原始失败样本设置最小样本数和置信阈值对教训做审批导入旧事件时字段变化导致崩溃事件结构没有版本管理检查 payload 中是否包含 schema_version引入 schema 版本兼容新旧解析函数快照加载后状态丢失快照写入了不完整状态检查快照 state 字段是否包含所有投影字段为快照写一个“完整性校验”缺字段就告警5.2 典型排错链路回放结果不一致当出现“按事件日志重放出的策略与线上状态不一致”时按以下顺序排查输入是否正确确认重放时的 aggregate_id、事件范围、快照起始点正确。路径和命名是否正确确认事件日志版本、代码分支与线上一致。依赖版本是否匹配确认解析旧事件的兼容函数已部署。配置是否生效确认快照加载和完整重放解析的是同一份默认配置。是否混入外部副作用检查有没有把大模型调用、工具调用放到 apply 方法里而不是放到事件生成阶段。是否存在非事件状态检查内存变量里是否有没被事件持久化的配置例如当前时间、随机数、全局缓存。最容易踩的是第 6 个。把普通 Python 对象当成事件源模型对象内部却有一些没有写入事件的字段回放时这些字段从初始值开始自然与实时运行过的状态不一致。5.3 典型排错链路策略改进后没有效果这个问题通常不在事件日志本身而在改进信号的闭环上检查 RewardObserved 是否真实反映了任务结果。检查 LessonExtracted 是否真正来自失败样本。检查从教训到策略更新的转换是否有可解释、可复现的路径。检查新策略是否在下次运行时真的被加载。四个环节任何一个断裂都会表现为“改了很多次效果没变化”。事件日志的作用是让每一条链路都可追踪但不会自动保证信号质量。6. 工程落地建议与扩展方向6.1 从原型到生产需要补齐的部分学习环境里上面的最小框架可以很快跑通。进入生产环境还需要补齐以下内容事件存储高可用与分区PostgreSQL 按时间分区或使用事件总线配合消费者做投影。快照策略根据事件总量决定快照频率例如每 1000 条事件或每 24 小时生成一次。事件版本兼容为每个 event_type 引入 schema_version并提供迁移函数。幂等与去重事件写入基于唯一 event_id 做幂等防止重试造成重复事件。安全与权限事件可能包含敏感信息需要脱敏、加密、分级控制。可观测性把事件日志接入监控大盘每次策略更新打点标记版本号关联线上指标。人工审核位如果策略更新是全自动的建议加测试集 gate 或人工审批开关避免一次错误反思带偏整个策略。6.2 与提示缓存、向量库、模型微调的配合事件溯源不是孤立设计它可以与 agent 的其他基础设施形成配合提示缓存在事件尾部记录 prompt hash回放时优先返回缓存结果降低调用成本。向量库把经验事件的关键内容向量化作为后续任务的检索源让“记忆”从结构化查询升级成语义检索。微调流水线筛选 RewardObserved 高分或低分事件整理成训练集定期微调小模型或 reranker。强化学习反馈事件日志本质上相当于 experience replay buffer可以从中采样训练策略。关键思路是不要把各个系统都做成独立数据库而是以事件流为源派生多个投影提示缓存、向量库、评测报告、微调数据集。事件流是唯一事实来源所有投影都可以随时重建。6.3 适用边界和反模式事件溯源对自改进代理有价值但不是所有代理都值得上这套方案。判断依据可以这样梳理如果代理只是简单的工具调用没有策略变更也不需要反思事件溯源是过度设计。如果代理需要持续多轮改进、需要回滚、需要理解历史因果事件溯源是合理选择。如果项目里还没有专门的评测数据集和反馈链路先不要搞事件溯源优先解决“如何评估改进是否有用”。容易出现反模式的做法包括把事件表当操作日志只有记录没有重放和投影事件日志不会带来价值。事件里塞巨量数据把整段对话、大文件、完整模型输出无脑写入事件事件存储很快变成垃圾场。策略更新不写事件直接修改一个 policy.json却没让变更经过 PolicyRevisionCommitted 流程导致因果链断裂。删除“坏事件”认为删掉失败事件代理就能变好这违反事件溯源的事实保留原则也会让评测失真。6.4 下一步可以从哪里开始练手对刚接触这个方向的开发者建议从一个小而完整的项目开始做一个简单的分类、搜索或问答任务每轮记录事件。用规则或大模型提取教训提交一次策略更新。在固定评测集上对比 PolicyRevisionCommitted 前后的累计 reward。尝试 fork 出一个分支在分支上再改进一次验证回滚和分支能力。最后把事件存储换成 PostgreSQL加上快照和 schema 版本模拟接近生产的结构。完成这个练习后就能在真实项目中判断哪些改进需求适合事件溯源哪里需要快速决策哪里需要可追溯的回放能力。事件溯源不会让代理变得更加聪明但它能让“代理为什么变聪明了”变成一道可以打开和核查的推理题而这恰恰是自改进代理在生产环境落地时最稀缺的能力。