如果你正在做一个“会自己改自己”的 AI Agent 项目大概率已经遇到过这几个让人头疼的问题Agent 上一轮表现很好下一轮忽然变笨了模型自己改了 Prompt但没人记得改之前是什么跑了一晚上实验第二天完全复现不了昨天的结果。这个问题的本质不是模型能力不够而是工程架构上缺少了一个东西可追溯、可回放、可回滚的事实记录。换句话说Self-improving agents are event sourced——自我改进的 Agent本质上就应该是一个事件溯源系统。这篇文章会把这个判断拆开讲清楚为什么自改进 Agent 不能只靠“保存快照”来管理事件溯源到底怎么落到 Agent 工程里以及你该怎么用一套可运行的事件存储与回放机制把 Agent 的每一次自我修改都变成可控的、可审计的、可复现的工程行为。1. 这篇文章真正要解决的问题先对齐一个核心痛点。假设你正在做一个支持“自我改进”的 Agent。它不是简单的 RAG 问答机器人而是具备以下能力的一种程序根据历史对话和用户反馈自动调整系统 Prompt根据任务执行结果新增或修改工具调用策略根据评测分数修改自身的决策规则在多次运行中不断迭代自己的行为策略。这个时候你面临的已经不是“怎么让模型答对题”而是一个典型的有状态系统问题Agent 的状态在变Agent 的代码逻辑在变Agent 的 Prompt 配置在变而且这些变化是 Agent 自己发起的。如果你用传统方式管理这个系统——数据库里存一份当前配置代码里写死一套行为规则模型运行时直接读当前配置——你会很快撞上三个问题。第一个问题是不可回溯。Agent 自己改坏了 Prompt你想恢复到“昨天下午那个表现最好的版本”但你根本没有昨天的版本记录。你只知道现在坏了不知道坏在哪一步。第二个问题是不可复现。Agent 的改进效果和上下文强相关。同一套 Prompt在某个会话里表现好换一个会话可能就崩。如果每一次修改没有伴随完整的行为事件记录你根本无法判断“这次改进到底是因为 Prompt 变了还是因为测试数据变了”。第三个问题是不可审计。对生产环境来说Agent 的行为需要可追踪它为什么改成这样是基于哪条用户反馈哪次工具调用触发了这个决策如果没有事件记录这些问题永远没有答案。传统软件开发里我们早就解决过类似的问题——通过 Git 做代码版本管理通过数据库日志做数据恢复通过事件溯源做金融交易记账。当 Agent 开始自我修改时它实际上变成了一个“修改自身源码和配置的程序”所以它必须复用这些成熟的工程思想。而事件溯源Event Sourcing恰恰是最贴合这个场景的架构模式。这篇文章适合三类读者正在做 Agent 自改进/自进化项目的工程师需要给 Agent 增加审计和回滚能力的架构师以及想理解“LLM Agent 工程化落地”到底难在哪里的技术爱好者。2. 事件溯源的核心概念从 Git 和银行流水说起要理解“自改进 Agent 为什么需要事件溯源”先得把事件溯源本身讲明白。很多人第一次接触事件溯源时会被“CQRS”“聚合根”“事件总线”这些词吓退其实它的核心思想非常简单——不存当前状态只存导致状态改变的事实。想象你在用 Git 做代码管理。Git 里并没有一个叫“最新代码”的神奇文件。Git 存储的是一串提交记录commit每个提交包含上一次提交的哈希、这次改了什么、谁改的、为什么改。你要拿到“最新代码”只需要从初始提交开始把所有提交按顺序应用一遍。这就是事件溯源。再想一个更贴近生活的例子银行账户。传统数据库存的是“当前余额”比如 1000 元。假设用户转账 300 元出去数据库会直接把余额改成 700 元。问题在于如果发生纠纷银行怎么知道这 300 元到底怎么转出去的事件溯源的做法不同银行存储的是“开户时存入 1000 元”“转出 300 元”“转入 500 元”这些事件。当前余额不是直接存储的而是通过重放事件计算出来的。这样每一分钱的来源都可追溯任何时候都可以回到历史中的某一个时间点。事件溯源系统里有几个关键角色事件Event)一个已经发生的事实不可修改、不可删除。例如PromptUpdated、ToolCalled、FeedbackReceived。事件存储Event Store)只支持追加写入的数据库。你可以往里面加新事件但不能修改或删除已有事件。聚合Aggregate)从事件流中重建出来的当前状态。比如 Agent 当前的完整配置和行为规则。投影Projection)把事件流转换成适合查询的视图。比如一份“当前 Prompt 版本”的普通数据库表。重放Replay)从最早的某个位置开始把所有事件按顺序重新应用一遍从而还原某个历史时刻的 Agent 状态。讲到这你可能会说这不就是“操作日志”吗过去做系统不也经常记录操作日志这里有一个重要的技术区别普通的操作日志通常只服务于监控和排查它记录的是“发生了什么”但系统当前状态还是直接存储在业务表里日志和真实状态可能不一致。事件溯源不一样——事件不仅仅是日志它就是系统状态本身。系统没有一份独立存储的“当前状态”所有状态都是从事件中推导出来的。事件是唯一的事实来源Single Source of Truth。这个区别对自改进 Agent 至关重要。因为 Agent 的“状态”不是简单的几个字段而是包含 Prompt 文本、工具配置、评分记录、历史反馈在内的大量上下文。如果你把这些状态直接存在一张表里那么每次自我修改都会覆盖旧值一旦修改逻辑有 bug历史信息就永久丢失。而事件溯源把这些修改过程全部保留任何时刻都能重建状态。3. 自改进 Agent 的本质一个修改自身程序的程序现在我们把目光转回自改进 Agent。先说一个容易混淆的点。很多人谈论 Agent 时会说“它会学习”“它会进化”但事实是当前主流的 LLM Agent 并不像人一样有持续更新的记忆机制。所谓“自改进”在工程上通常是指以下几种具体行为的组合。Prompt 自优化Agent 根据历史任务表现自动修改系统 Prompt比如增加约束条件、补充示例、调整输出格式要求。这是成本最低、最普遍的自改进形式。工具自编排Agent 根据任务需求自动调整工具调用的顺序、参数甚至动态生成新的工具配置。某些框架里Agent 可以把一套多步骤工具链抽象成新工具。知识自积累Agent 从已完成的任务中提取经验写入外部记忆库。后续任务开始时Agent 会检索这些经验作为额外上下文。策略自更新Agent 根据评测结果调整决策策略。这个最接近“元学习”的概念——不仅解决任务还解决“如何解决任务”的方法。无论哪种形式自改进 Agent 在技术上都有一个共性它会在运行过程中修改自己的行为配置并且这个修改会影响后续所有行为。这就产生了一个非常特殊的工程问题普通程序有明确的“程序”和“数据”边界而自改进 Agent 的行为配置Prompt、策略、工具定义介于程序和数据之间。它不是硬编码在代码里但也不是纯粹的查询数据——它直接决定 Agent 怎么思考、怎么行动。如果你把行为配置当成普通数据来管理最常见的结果就是数据库里存着“当前最佳 Prompt”一旦 Agent 写入了一个新版本旧版就没了。如果你把行为配置当成普通代码来管理结果也差不多Git 仓库里会有很多历史提交但 Agent 运行时的行为配置未必和 Git 提交一一对应。自改进 Agent 真正需要的是一个能把“行为配置的变化过程”完整记录下来、并且能够在任意时刻恢复任意版本历史的机制。这就是事件溯源架构的用武之地。4. Agent 改进过程的事件模型六大核心事件前面说的是概念这一节开始落地。要让自改进 Agent 接入事件溯源第一步是定义事件模型。也就是想清楚Agent 在自改进过程中到底会发出哪些不可变的事件根据 Agent 自改进的典型流程我整理出了六类核心事件。它们覆盖了从任务执行、反馈收集、改进决策到配置变更的完整链路。第一类任务执行事件。记录 Agent 每次任务执行的输入、输出、耗时、调用链。这是最底层的事实。没有任务执行的完整记录后面所有改进分析都无从谈起。第二类反馈接收事件。记录外部反馈包括用户显式评价、自动评测分数、人工纠偏内容。这类事件是自我改进的“信号源”。第三类改进决策事件。记录 Agent 或外部评测系统做出的改进决策。它回答的是“为什么改”这个问题基于哪条反馈、哪个评分、哪个失败案例触发了修改。第四类配置变更事件。这是最核心的事件类型。它记录 Agent 对自身 Prompt、工具定义、策略参数的具体修改。包括变更之前的内容、变更之后的内容、变更类型。第五类评测验证事件。记录变更前后的评测结果。这一步非常关键改进不是只要做了变更就算完成必须要有评测验证来证明这个变更是有效的。第六类整体演进事件。记录 Agent 的版本演进轨迹。比如某次变更通过评测决定正式进入生产某次变更效果回退决定回滚到上一版本。下面是这六类事件的关键字段设计可以直接用于你的事件表设计参考。事件类型事件名称示例核心字段解决什么问题任务执行TaskExecutedtaskId, taskInput, taskOutput, traceId记录每次任务的完整上下文反馈接收FeedbackReceivedtaskId, feedbackType, score, comment记录改进的信号来源改进决策ImprovementDecidedtaskId, reason, strategyType记录“为什么改”配置变更ConfigChangedconfigType, oldValue, newValue, changeReason记录“改了什么”评测验证EvaluationCompletedconfigVersion, scoreBefore, scoreAfter, verdict记录“改得怎么样”演进事件EvolutionCommittedfromVersion, toVersion, rollbackFlag记录版本演进和回滚这套事件模型的设计原则有三个事件不可变只能追加事件包含足够上下文不能只记“发生了什么”而不记“为什么发生”事件粒度要能支撑重放和审计不能太粗也不能太细。从实际经验来看很多团队在设计事件模型时容易犯两个错误。第一个错误是只记录配置变更事件不记录任务执行和反馈事件。这样做会导致一个问题事后追查时你只知道 Prompt 被改了但不知道是基于哪条用户反馈改的。第二个错误是把事件设计得过于细粒度比如把模型 API 的每次 token 调用都记为事件。这会让事件存储变得非常庞大而且大部分事件对改进决策没有价值。5. 环境准备与基础依赖进入实操环节之前先明确环境。本文用 Python 作为示例语言核心思路不依赖特定版本。只要你本机有 Python 3.9 及以上环境就可以跑通下面的代码。需要准备的工具和依赖如下Python 3.9建议使用虚拟环境管理项目依赖。一个 JSON 序列化库Python 内置json即可。时间处理使用 Python 内置datetime。后续如果要接入消息队列或流处理可以考虑 Kafka、Redis Stream 或 NATS JetStream。为了防止环境差异本文示例不依赖第三方框架全部用 Python 标准库实现。这样你拿到代码就能跑不需要先部署 Kafka 或数据库。等理解了核心逻辑之后再迁移到真正的生产级事件存储如 PostgreSQL、EventStoreDB、Kafka也不迟。创建项目目录结构mkdir -p self-improving-agent-event-sourcing cd self-improving-agent-event-sourcing python3 -m venv venv source venv/bin/activate虚拟环境激活后创建一个空的requirements.txt文件即可。本文代码不依赖外部包。touch requirements.txt接下来创建三个 Python 文件event_store.py事件存储的核心实现。agent_events.pyAgent 改进场景的事件定义。demo_workflow.py完整的演示流程。6. 事件存储核心实现一个最小可运行的版本先写最底层的事件存储。这里实现一个基于追加日志Append-Only Log的最小事件存储。它支持三个操作追加事件、按聚合 ID 读取事件流、通过重放事件流重建状态。# 文件路径event_store.py import json from datetime import datetime, timezone from typing import Any, Dict, List, Optional class Event: 事件基类。事件一经创建不允许修改。 def __init__( self, event_id: str, aggregate_id: str, event_type: str, data: Dict[str, Any], occurred_at: Optional[str] None, ) - None: self.event_id event_id self.aggregate_id aggregate_id self.event_type event_type self.data data self.occurred_at occurred_at or datetime.now(timezone.utc).isoformat() def to_dict(self) - Dict[str, Any]: return { event_id: self.event_id, aggregate_id: self.aggregate_id, event_type: self.event_type, data: self.data, occurred_at: self.occurred_at, } class EventStore: 最小事件存储基于追加日志支持重放。 def __init__(self, storage_path: str event_log.jsonl) - None: self.storage_path storage_path # 每个 aggregate_id 对应一个事件索引 self._index: Dict[str, List[Dict[str, Any]]] {} def append(self, event: Event) - None: 追加事件。只许追加禁止修改已有事件。 event_dict event.to_dict() # 每个聚合追加一条事件记录 with open(self.storage_path, a, encodingutf-8) as f: f.write(json.dumps(event_dict, ensure_asciiFalse) \n) # 更新内存索引 if event.aggregate_id not in self._index: self._index[event.aggregate_id] [] self._index[event.aggregate_id].append(event_dict) def load_events(self, aggregate_id: str) - List[Dict[str, Any]]: 读取某个聚合的全部事件。 if aggregate_id in self._index: return self._index[aggregate_id] events: List[Dict[str, Any]] [] with open(self.storage_path, r, encodingutf-8) as f: for line in f: if not line.strip(): continue event_dict json.loads(line) if event_dict[aggregate_id] aggregate_id: events.append(event_dict) self._index[aggregate_id] events return events def replay_all(self, aggregate_id: str) - List[Dict[str, Any]]: 重放事件流返回按时间排序的事件列表。 events self.load_events(aggregate_id) events.sort(keylambda x: x[occurred_at]) return events这段代码做了三件事。第一定义了Event类事件包含event_id、aggregate_id、event_type、data、occurred_at五个字段。第二实现了一个追加写入的 JSONL 文件存储。第三提供了按聚合 ID 加载事件并重放的能力。这里有一些设计细节值得解释。aggregate_id是聚合根的唯一标识。什么是聚合根简单说它代表一个需要保证完整性的业务实体。在 Agent 场景里一个 Agent 实例、一个 Agent 配置版本、一个 Agent 的改进迭代周期都可以作为聚合根。选择哪种粒度取决于你的业务需求。本文示例把单个 Agent 会话作为聚合根这样每个会话内有完整的任务执行、反馈、改进记录。occurred_at使用 UTC 时间而不是本地时间。这避免了不同服务器时区不一致导致的事件排序问题。JSONL 格式之所以够用是因为它简单、轻量、方便调试。你可以在生产环境换成 PostgreSQL 表或 Kafka Topic但这套事件模型和重放逻辑本身不需要改。7. Agent 改进事件定义与状态重建有了基础事件存储接下来定义 Agent 自改进场景里的具体事件和状态重建逻辑。这段代码放在agent_events.py里。它定义三类事件任务执行、反馈接收、配置变更。然后定义一个AgentStateBuilder把事件流重放成 Agent 当前的状态。# 文件路径agent_events.py from typing import Any, Dict, List, Optional from event_store import Event, EventStore class AgentEventFactory: Agent 改进事件的工厂类负责创建各种事件。 staticmethod def task_executed(aggregate_id: str, task_id: str, task_input: str, task_output: str) - Event: return Event( event_idfevt-{task_id}-task, aggregate_idaggregate_id, event_typeTaskExecuted, data{ task_id: task_id, task_input: task_input, task_output: task_output, }, ) staticmethod def feedback_received( aggregate_id: str, task_id: str, feedback_type: str, score: float, comment: str ) - Event: return Event( event_idfevt-{task_id}-feedback, aggregate_idaggregate_id, event_typeFeedbackReceived, data{ task_id: task_id, feedback_type: feedback_type, score: score, comment: comment, }, ) staticmethod def config_changed( aggregate_id: str, config_type: str, old_value: str, new_value: str, change_reason: str, task_id: str, ) - Event: return Event( event_idfevt-{task_id}-config, aggregate_idaggregate_id, event_typeConfigChanged, data{ config_type: config_type, old_value: old_value, new_value: new_value, change_reason: change_reason, task_id: task_id, }, ) class AgentState: 从事件流重建出来的 Agent 状态。 def __init__(self) - None: self.system_prompt: str self.tool_config: Dict[str, Any] {} self.task_history: List[Dict[str, Any]] [] self.feedback_history: List[Dict[str, Any]] [] self.config_history: List[Dict[str, Any]] [] def apply(self, event_dict: Dict[str, Any]) - None: 把单个事件应用到当前状态。 event_type event_dict[event_type] data event_dict[data] if event_type TaskExecuted: self.task_history.append(data) elif event_type FeedbackReceived: self.feedback_history.append(data) elif event_type ConfigChanged: self.config_history.append(data) if data[config_type] system_prompt: self.system_prompt data[new_value] elif data[config_type] tool_config: self.tool_config data[new_value] classmethod def rebuild_from_events(cls, events: List[Dict[str, Any]]) - AgentState: 通过重放事件重建 Agent 状态。 state cls() for event_dict in events: state.apply(event_dict) return state def rebuild_agent_state(event_store: EventStore, aggregate_id: str) - AgentState: 完整的重建入口从事件存储中读取事件重建状态。 events event_store.replay_all(aggregate_id) return AgentState.rebuild_from_events(events)这个文件最重要的部分是AgentState.apply方法。它定义了一个规则每种事件类型如何修改当前状态。这个规则必须和业务语义严格一致。比如ConfigChanged事件里如果config_type是system_prompt那new_value就会覆盖当前的system_prompt如果是tool_config就会覆盖工具配置。这段代码里隐藏着一个值得思考的架构原则Agent 的当前状态永远不直接存储而是通过事件重放计算出来。这意味着如果你发现“当前 Prompt 有问题”你可以回到任何历史节点重建那个节点上的状态然后从这个节点继续演进而不需要担心覆盖任何数据。这种模式在传统事件溯源里被称为“状态重建”在 Agent 场景里有更实际的用途——它天然支持 A/B 测试和分支演进。你可以从同一个历史节点分叉出两个 Agent 版本一个继续按原策略改进一个尝试新策略然后对比评测结果。8. 完整演示Agent 自改进流程的事件溯源实战现在把前面的代码串起来写一个完整演示。场景设计如下有一个 Agent初始系统 Prompt 是“请用简洁的语言回答问题”。它在两个任务上表现不佳收到了用户反馈“回答太啰嗦”。于是 Agent 决定修改自己的系统 Prompt改成“请用不超过三句话回答问题”。修改完成后再执行一个验证任务。完整演示代码在demo_workflow.py里。# 文件路径demo_workflow.py from event_store import EventStore from agent_events import AgentEventFactory, rebuild_agent_state def main() - None: # 1. 初始化事件存储指定日志文件 store EventStore(demo_agent_events.jsonl) # 2. 模拟一个 Agent 会话aggregate_id 为 agent-session-001 aggregate_id agent-session-001 # 3. 记录事件任务执行 1 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-001, task_input解释什么是事件溯源, task_output事件溯源是一种架构模式它不直接存储系统当前状态而是存储所有导致状态改变的事件通过重放事件来推导系统当前状态。这个模式有很多优点比如可以追溯历史状态可以方便地回滚还可以通过重放来复现故障现场。, ) ) # 4. 记录事件任务执行 2 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-002, task_input写出 Python 的 Hello World, task_output所谓 Hello World 程序是编程领域中最基础的一种程序它的功能就是向屏幕输出一句话Hello, World! 在 Python 语言中实现这个功能的方法非常简单只需要调用内置的 print 函数并传入对应的字符串即可。, ) ) # 5. 记录事件用户反馈 store.append( AgentEventFactory.feedback_received( aggregate_idaggregate_id, task_idtask-002, feedback_typeuser_comment, score0.4, comment回答太啰嗦了希望简洁一点。, ) ) # 6. 重建当前状态查看修改前的 Prompt state_before rebuild_agent_state(store, aggregate_id) print( 修改前 Agent 状态 ) print(fSystem Prompt: {state_before.system_prompt}) print(f执行任务数: {len(state_before.task_history)}) print(f接收反馈数: {len(state_before.feedback_history)}) # 7. 记录事件配置变更Agent 自我改进 store.append( AgentEventFactory.config_changed( aggregate_idaggregate_id, config_typesystem_prompt, old_valuestate_before.system_prompt, new_value请用简洁的语言回答问题不超过三句话。, change_reason用户反馈回答过于啰嗦需要压缩输出长度。, task_idtask-002, ) ) # 8. 记录事件任务执行 3使用新 Prompt 后的验证任务 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-003, task_input写出 Python 的 Hello World, task_outputprint(Hello, World!), ) ) # 9. 记录事件二次反馈 store.append( AgentEventFactory.feedback_received( aggregate_idaggregate_id, task_idtask-003, feedback_typeuser_comment, score0.9, comment回复简洁明了很好。, ) ) # 10. 再次重建状态查看修改后的结果 state_after rebuild_agent_state(store, aggregate_id) print(\n 修改后 Agent 状态 ) print(fSystem Prompt: {state_after.system_prompt}) print(f配置变更次数: {len(state_after.config_history)}) print(f执行任务数: {len(state_after.task_history)}) # 11. 从事件流中回溯历史重建“第一次改进决策前”的状态 print(\n 回溯历史状态 ) all_events store.replay_all(aggregate_id) # 找到 ConfigChanged 事件的位置 config_change_index None for i, event in enumerate(all_events): if event[event_type] ConfigChanged: config_change_index i break if config_change_index is not None: events_before_change all_events[:config_change_index] from agent_events import AgentState past_state AgentState.rebuild_from_events(events_before_change) print(f修改前的 Prompt: {past_state.system_prompt}) print(f历史任务数量: {len(past_state.task_history)}) if __name__ __main__: main()运行方式python demo_workflow.py预期输出类似 修改前 Agent 状态 System Prompt: 执行任务数: 2 接收反馈数: 1 修改后 Agent 状态 System Prompt: 请用简洁的语言回答问题不超过三句话。 配置变更次数: 1 执行任务数: 3 回溯历史状态 修改前的 Prompt: 历史任务数量: 2输出里有一行值得注意修改前的System Prompt是空的。这个结果是正常的因为模拟场景里初始 Prompt 并没有作为事件写入而 AgentState 只从事件流重建状态。实际工程中Agent 的初始配置应该以“初始化事件”的形式写入事件存储。比如定义一个AgentInitialized事件system_prompt的初始值在那一刻被记录下来。这样重放时才能完整还原状态。9. 运行结果与效果验证运行上面这段演示代码你能验证到三件事。第一件事事件追加是否生效。打开生成的demo_agent_events.jsonl文件可以看到每一行都是一条事件记录。注意看事件是追加在文件末尾的历史记录没有被修改。这个特性保证了事件流不会因为后续写入而被污染。第二件事状态重建是否正确。执行两次rebuild_agent_state第一次拿到的是修改前的状态第二次拿到的是修改后的状态。中间只隔了一次ConfigChanged事件状态却完全不同。这说明事件重放确实起到了“时间旅行”的作用。第三件事历史回溯是否可行。演示代码最后部分手动找到ConfigChanged事件的位置用它之前的事件重建了历史状态。这个操作等价于 Git 的checkout commit它证明了任意历史节点的状态都可以被重建。如果运行失败优先检查以下几个方面Python 版本是否在 3.9 及以上。是否在项目根目录运行event_store.py和agent_events.py是否正常导入。是否重复运行过 demo 脚本导致demo_agent_events.jsonl里积累了多套agent-session-001的事件。如果是删除该文件再重新运行。实际上第二个问题值得展开说一下。如果你连续运行两次python demo_workflow.py第二次运行时事件存储里已经存在agent-session-001的事件脚本会再次追加两套一模一样的事件状态重建结果却不受影响——因为事件是追加写入的重放时只是多了一倍的记录。但你如果统计任务数量会发现翻倍了。这说明在生产环境里你要保证同一个 Agent 会话不会被重复写入通常通过唯一聚合 ID 或幂等机制来解决。10. 自改进 Agent 事件溯源的常见问题与排查思路事件溯源在自改进 Agent 项目里用起来之后会遇到一些比“代码报错”更隐蔽的问题。下面整理了几个高频问题。问题现象可能原因排查方式解决方案重放后状态与预期不符事件顺序错乱依赖了错误的时间字段检查occurred_at是否统一使用 UTC确保事件写入顺序与业务语义一致使用单调递增的序列号代替时间戳排序或者在事件存储层维护全局序重建状态时 Prompt 为空Agent 初始配置没有作为事件写入查看事件流里是否有AgentInitialized事件把初始 Prompt、初始工具配置都建模为初始化事件确保任何状态都来源于事件事件日志文件无限增长没有做事件归档或快照观察日志文件大小和查询延迟定期生成状态快照重放时从最近快照开始而不是从第一条事件开始同一个 Agent 会话数据重复客户端重复提交了事件检查写入路径是否有重试机制在事件存储层做幂等控制按event_id去重历史状态可以重建但无法直接查询“当前配置”事件溯源不适合按照当前状态做高频查询Prisma 或 SQL 直接查询是查不到当前状态的增加投影Projection模块把事件流实时同步到关系表或 NoSQL供查询端使用回滚到一个旧版本后新事件丢失回滚实现有误直接删除了旧事件确认回滚实现是否采用了“追加补偿事件”的方式回滚不要删除历史事件而是追加一个ConfigChanged或EvolutionCommitted类型的补偿事件这个表格里有一个设计原则值得反复强调回滚不等于删除事件。在事件溯源系统里没有任何操作能删除或修改已追加的事件。如果你想从当前 Prompt 回滚到上一个版本正确做法是追加一条ConfigChanged事件把new_value设为上一个版本的旧值。这样做的结果是事件流里完整保留了“V1 - V2 - 回滚到 V1”的全部轨迹而不是只留下一个孤零零的 V1。这条轨迹对你分析“为什么 V2 失败”非常宝贵。另一个常见误区是把事件溯源和“记录日志”混为一谈。普通日志只记录系统运行过程中的关键输出不承担状态恢复的职责事件溯源则把事件当作状态的唯一来源。如果团队里有人试图用日志框架来实现事件溯源会很快发现系统重启后无法重建状态因为日志框架很少保证有序性和完整性。11. 自改进 Agent 事件溯源最佳实践与生产建议前面几节已经把原理和代码讲清楚了这一节补充一些工程落地方案。11.1 从最小事件模型开始不要贪多新手最容易犯的错误是试图把所有细节都建模成事件结果事件类型膨胀到几十种维护成本急剧上升。建议从一版最小事件模型开始初始化事件、任务执行事件、反馈接收事件、配置变更事件、评测验证事件。运行一段时间后再根据实际分析需求增加事件类型。事件是追加式的不存在“改表结构”的迁移压力所以晚一点增加事件类型完全来得及。11.2 用状态快照优化重放性能事件溯源有个天然的成本问题事件越来越多重放会越来越慢。假设一个 Agent 每天执行 10 万次任务每个任务产生 3 条事件一个月就是 900 万条事件。每次重建状态都要从第一条事件开始算性能无法接受。解决方案是定期生成状态快照Snapshot。比如每 1000 条事件把当时的 AgentState 序列化保存下来。重放时先加载最近的快照然后只重放快照之后的事件。这个思路和 Redis 持久化的 AOF RDB 组合非常相似。11.3 投影层负责查询事件流负责事实事件存储本身不适合承担“查询当前 Prompt 是什么”“查询某个分数段的任务记录”这类高频查询。推荐的架构是事件流作为唯一事实来源投影模块监听新事件实时更新一个“当前状态表”或“统计报表表”。比如你可以维护一张agent_current_config表每当出现新的ConfigChanged事件就同步更新这张表。这样查询端可以直接走 SQL而审计和回溯场景再走事件重放。11.4 给改进决策增加人工审批节点自改进 Agent 在开发环境里可以全自动迭代但进入生产环境后建议给“配置变更”这个环节增加人工审批或复核机制。不是所有自动改进都应该直接上线。具体做法是在事件链路里增加一个ImprovementReviewRequired事件。当 Agent 产生改进建议后它不会直接发ConfigChanged事件而是先发一个“待审核”事件。人工或外部系统审核通过后再追加ConfigChanged事件。这个流程保证了配置变更可以被控制也方便追踪每一次正式变更的审批人。11.5 事件设计要有业务语义不要过度抽象有些工程师会把所有事件都塞进一个通用结构比如data字段里装一个 JSON 字符串这样前期写代码很爽后期分析时却很痛苦。更推荐的做法是相同类型的事件结构保持稳定。比如ConfigChanged事件的data里就要固定包含config_type、old_value、new_value、change_reason、task_id。这样后续做数据分析和可视化展现时不需要每次都对data内容做二次解析。11.6 明确事件溯源和安全边界涉及 Agent 自我修改的代码有一个红线必须守住事件存储本身不能存放凭据和密钥。很多 Agent 的改进需要调用 LLM APIAPI Key 应该存放在独立的密钥管理系统里事件里只记录“调用了模型使用了多少 token”不要记录 Authorization 头或 Token 明文。另外对生产环境的 Agent 配置变更务必遵守最小权限原则。Agent 只应该能修改自己负责范围内的 Prompt、工具配置和策略参数不应该有权限去修改事件存储系统本身的代码。把“Agent 的改进权限”和“平台管理员的运维权限”隔离开能避免很多安全风险。11.7 用事件溯源辅助回归评测自改进 Agent 有一个比其他系统更麻烦的问题模型输出有随机性。同样的 Prompt 在不同温度下效果不同同样的修改在不同上下文里可能表现不一致。事件溯源提供的完整历史记录恰好可以支撑更严谨的回归评测。你可以在历史事件流中选出一组具有代表性的任务组成回归测试集然后对 Agent 的多个历史版本分别做评测。这比只依赖最近几条反馈来判断“改得好不好”更可靠。12. 总结与后续学习方向这篇文章的核心判断可以总结成一句话当 Agent 开始修改自身行为配置时它的状态管理就不能再用“保存当前值”的思路而应该采用事件溯源把每一次任务执行、反馈接收、配置变更都变成不可变事件。文章里不仅解释了为什么还给出了一个完整的最小实现。你用它跑通的流程包括定义六类核心事件、实现追加式事件存储、通过事件重放重建 Agent 状态、从事件流回溯任意历史节点、追加补偿事件实现回滚。如果你接下来要继续深入有四个方向值得探索。第一个方向是生产级事件存储选型。PostgreSQL 的 append-only 表、EventStoreDB、Kafka 都可以作为事件存储的底座它们各有优劣适合不同的吞吐和一致性要求。第二个方向是快照与归档策略。事件量大了之后怎么做快照、怎么做压缩、怎么保留全部历史又兼顾查询性能是一套需要在真实数据量下验证的工程方法。第三个方向是投影和查询优化。把事件流同步到 Elasticsearch、ClickHouse 或关系型数据库构建适合分析 Agent 行为的查询视图。第四个方向是结合评测系统做自动回归。事件溯源给你提供了完整的 Agent 版本历史能不能在此基础上做自动化的版本对比和回归验证直接决定了你的 Agent 自改进系统能否从 demo 走向生产。对正在搭建自改进 Agent 的团队我的建议是别急着接复杂框架先用本文的代码把事件模型定义清楚把状态重建流程跑通然后观察真实 Agent 运行数据里哪些事件对你最有价值。事件溯源的核心价值不在于技术本身而在于它逼着你先想清楚一件事你的 Agent 到底在改变什么以及你如何证明这些改变是有效的。
事件溯源：让自改进AI Agent的每一次修改都可追溯、可回放、可回滚
如果你正在做一个“会自己改自己”的 AI Agent 项目大概率已经遇到过这几个让人头疼的问题Agent 上一轮表现很好下一轮忽然变笨了模型自己改了 Prompt但没人记得改之前是什么跑了一晚上实验第二天完全复现不了昨天的结果。这个问题的本质不是模型能力不够而是工程架构上缺少了一个东西可追溯、可回放、可回滚的事实记录。换句话说Self-improving agents are event sourced——自我改进的 Agent本质上就应该是一个事件溯源系统。这篇文章会把这个判断拆开讲清楚为什么自改进 Agent 不能只靠“保存快照”来管理事件溯源到底怎么落到 Agent 工程里以及你该怎么用一套可运行的事件存储与回放机制把 Agent 的每一次自我修改都变成可控的、可审计的、可复现的工程行为。1. 这篇文章真正要解决的问题先对齐一个核心痛点。假设你正在做一个支持“自我改进”的 Agent。它不是简单的 RAG 问答机器人而是具备以下能力的一种程序根据历史对话和用户反馈自动调整系统 Prompt根据任务执行结果新增或修改工具调用策略根据评测分数修改自身的决策规则在多次运行中不断迭代自己的行为策略。这个时候你面临的已经不是“怎么让模型答对题”而是一个典型的有状态系统问题Agent 的状态在变Agent 的代码逻辑在变Agent 的 Prompt 配置在变而且这些变化是 Agent 自己发起的。如果你用传统方式管理这个系统——数据库里存一份当前配置代码里写死一套行为规则模型运行时直接读当前配置——你会很快撞上三个问题。第一个问题是不可回溯。Agent 自己改坏了 Prompt你想恢复到“昨天下午那个表现最好的版本”但你根本没有昨天的版本记录。你只知道现在坏了不知道坏在哪一步。第二个问题是不可复现。Agent 的改进效果和上下文强相关。同一套 Prompt在某个会话里表现好换一个会话可能就崩。如果每一次修改没有伴随完整的行为事件记录你根本无法判断“这次改进到底是因为 Prompt 变了还是因为测试数据变了”。第三个问题是不可审计。对生产环境来说Agent 的行为需要可追踪它为什么改成这样是基于哪条用户反馈哪次工具调用触发了这个决策如果没有事件记录这些问题永远没有答案。传统软件开发里我们早就解决过类似的问题——通过 Git 做代码版本管理通过数据库日志做数据恢复通过事件溯源做金融交易记账。当 Agent 开始自我修改时它实际上变成了一个“修改自身源码和配置的程序”所以它必须复用这些成熟的工程思想。而事件溯源Event Sourcing恰恰是最贴合这个场景的架构模式。这篇文章适合三类读者正在做 Agent 自改进/自进化项目的工程师需要给 Agent 增加审计和回滚能力的架构师以及想理解“LLM Agent 工程化落地”到底难在哪里的技术爱好者。2. 事件溯源的核心概念从 Git 和银行流水说起要理解“自改进 Agent 为什么需要事件溯源”先得把事件溯源本身讲明白。很多人第一次接触事件溯源时会被“CQRS”“聚合根”“事件总线”这些词吓退其实它的核心思想非常简单——不存当前状态只存导致状态改变的事实。想象你在用 Git 做代码管理。Git 里并没有一个叫“最新代码”的神奇文件。Git 存储的是一串提交记录commit每个提交包含上一次提交的哈希、这次改了什么、谁改的、为什么改。你要拿到“最新代码”只需要从初始提交开始把所有提交按顺序应用一遍。这就是事件溯源。再想一个更贴近生活的例子银行账户。传统数据库存的是“当前余额”比如 1000 元。假设用户转账 300 元出去数据库会直接把余额改成 700 元。问题在于如果发生纠纷银行怎么知道这 300 元到底怎么转出去的事件溯源的做法不同银行存储的是“开户时存入 1000 元”“转出 300 元”“转入 500 元”这些事件。当前余额不是直接存储的而是通过重放事件计算出来的。这样每一分钱的来源都可追溯任何时候都可以回到历史中的某一个时间点。事件溯源系统里有几个关键角色事件Event)一个已经发生的事实不可修改、不可删除。例如PromptUpdated、ToolCalled、FeedbackReceived。事件存储Event Store)只支持追加写入的数据库。你可以往里面加新事件但不能修改或删除已有事件。聚合Aggregate)从事件流中重建出来的当前状态。比如 Agent 当前的完整配置和行为规则。投影Projection)把事件流转换成适合查询的视图。比如一份“当前 Prompt 版本”的普通数据库表。重放Replay)从最早的某个位置开始把所有事件按顺序重新应用一遍从而还原某个历史时刻的 Agent 状态。讲到这你可能会说这不就是“操作日志”吗过去做系统不也经常记录操作日志这里有一个重要的技术区别普通的操作日志通常只服务于监控和排查它记录的是“发生了什么”但系统当前状态还是直接存储在业务表里日志和真实状态可能不一致。事件溯源不一样——事件不仅仅是日志它就是系统状态本身。系统没有一份独立存储的“当前状态”所有状态都是从事件中推导出来的。事件是唯一的事实来源Single Source of Truth。这个区别对自改进 Agent 至关重要。因为 Agent 的“状态”不是简单的几个字段而是包含 Prompt 文本、工具配置、评分记录、历史反馈在内的大量上下文。如果你把这些状态直接存在一张表里那么每次自我修改都会覆盖旧值一旦修改逻辑有 bug历史信息就永久丢失。而事件溯源把这些修改过程全部保留任何时刻都能重建状态。3. 自改进 Agent 的本质一个修改自身程序的程序现在我们把目光转回自改进 Agent。先说一个容易混淆的点。很多人谈论 Agent 时会说“它会学习”“它会进化”但事实是当前主流的 LLM Agent 并不像人一样有持续更新的记忆机制。所谓“自改进”在工程上通常是指以下几种具体行为的组合。Prompt 自优化Agent 根据历史任务表现自动修改系统 Prompt比如增加约束条件、补充示例、调整输出格式要求。这是成本最低、最普遍的自改进形式。工具自编排Agent 根据任务需求自动调整工具调用的顺序、参数甚至动态生成新的工具配置。某些框架里Agent 可以把一套多步骤工具链抽象成新工具。知识自积累Agent 从已完成的任务中提取经验写入外部记忆库。后续任务开始时Agent 会检索这些经验作为额外上下文。策略自更新Agent 根据评测结果调整决策策略。这个最接近“元学习”的概念——不仅解决任务还解决“如何解决任务”的方法。无论哪种形式自改进 Agent 在技术上都有一个共性它会在运行过程中修改自己的行为配置并且这个修改会影响后续所有行为。这就产生了一个非常特殊的工程问题普通程序有明确的“程序”和“数据”边界而自改进 Agent 的行为配置Prompt、策略、工具定义介于程序和数据之间。它不是硬编码在代码里但也不是纯粹的查询数据——它直接决定 Agent 怎么思考、怎么行动。如果你把行为配置当成普通数据来管理最常见的结果就是数据库里存着“当前最佳 Prompt”一旦 Agent 写入了一个新版本旧版就没了。如果你把行为配置当成普通代码来管理结果也差不多Git 仓库里会有很多历史提交但 Agent 运行时的行为配置未必和 Git 提交一一对应。自改进 Agent 真正需要的是一个能把“行为配置的变化过程”完整记录下来、并且能够在任意时刻恢复任意版本历史的机制。这就是事件溯源架构的用武之地。4. Agent 改进过程的事件模型六大核心事件前面说的是概念这一节开始落地。要让自改进 Agent 接入事件溯源第一步是定义事件模型。也就是想清楚Agent 在自改进过程中到底会发出哪些不可变的事件根据 Agent 自改进的典型流程我整理出了六类核心事件。它们覆盖了从任务执行、反馈收集、改进决策到配置变更的完整链路。第一类任务执行事件。记录 Agent 每次任务执行的输入、输出、耗时、调用链。这是最底层的事实。没有任务执行的完整记录后面所有改进分析都无从谈起。第二类反馈接收事件。记录外部反馈包括用户显式评价、自动评测分数、人工纠偏内容。这类事件是自我改进的“信号源”。第三类改进决策事件。记录 Agent 或外部评测系统做出的改进决策。它回答的是“为什么改”这个问题基于哪条反馈、哪个评分、哪个失败案例触发了修改。第四类配置变更事件。这是最核心的事件类型。它记录 Agent 对自身 Prompt、工具定义、策略参数的具体修改。包括变更之前的内容、变更之后的内容、变更类型。第五类评测验证事件。记录变更前后的评测结果。这一步非常关键改进不是只要做了变更就算完成必须要有评测验证来证明这个变更是有效的。第六类整体演进事件。记录 Agent 的版本演进轨迹。比如某次变更通过评测决定正式进入生产某次变更效果回退决定回滚到上一版本。下面是这六类事件的关键字段设计可以直接用于你的事件表设计参考。事件类型事件名称示例核心字段解决什么问题任务执行TaskExecutedtaskId, taskInput, taskOutput, traceId记录每次任务的完整上下文反馈接收FeedbackReceivedtaskId, feedbackType, score, comment记录改进的信号来源改进决策ImprovementDecidedtaskId, reason, strategyType记录“为什么改”配置变更ConfigChangedconfigType, oldValue, newValue, changeReason记录“改了什么”评测验证EvaluationCompletedconfigVersion, scoreBefore, scoreAfter, verdict记录“改得怎么样”演进事件EvolutionCommittedfromVersion, toVersion, rollbackFlag记录版本演进和回滚这套事件模型的设计原则有三个事件不可变只能追加事件包含足够上下文不能只记“发生了什么”而不记“为什么发生”事件粒度要能支撑重放和审计不能太粗也不能太细。从实际经验来看很多团队在设计事件模型时容易犯两个错误。第一个错误是只记录配置变更事件不记录任务执行和反馈事件。这样做会导致一个问题事后追查时你只知道 Prompt 被改了但不知道是基于哪条用户反馈改的。第二个错误是把事件设计得过于细粒度比如把模型 API 的每次 token 调用都记为事件。这会让事件存储变得非常庞大而且大部分事件对改进决策没有价值。5. 环境准备与基础依赖进入实操环节之前先明确环境。本文用 Python 作为示例语言核心思路不依赖特定版本。只要你本机有 Python 3.9 及以上环境就可以跑通下面的代码。需要准备的工具和依赖如下Python 3.9建议使用虚拟环境管理项目依赖。一个 JSON 序列化库Python 内置json即可。时间处理使用 Python 内置datetime。后续如果要接入消息队列或流处理可以考虑 Kafka、Redis Stream 或 NATS JetStream。为了防止环境差异本文示例不依赖第三方框架全部用 Python 标准库实现。这样你拿到代码就能跑不需要先部署 Kafka 或数据库。等理解了核心逻辑之后再迁移到真正的生产级事件存储如 PostgreSQL、EventStoreDB、Kafka也不迟。创建项目目录结构mkdir -p self-improving-agent-event-sourcing cd self-improving-agent-event-sourcing python3 -m venv venv source venv/bin/activate虚拟环境激活后创建一个空的requirements.txt文件即可。本文代码不依赖外部包。touch requirements.txt接下来创建三个 Python 文件event_store.py事件存储的核心实现。agent_events.pyAgent 改进场景的事件定义。demo_workflow.py完整的演示流程。6. 事件存储核心实现一个最小可运行的版本先写最底层的事件存储。这里实现一个基于追加日志Append-Only Log的最小事件存储。它支持三个操作追加事件、按聚合 ID 读取事件流、通过重放事件流重建状态。# 文件路径event_store.py import json from datetime import datetime, timezone from typing import Any, Dict, List, Optional class Event: 事件基类。事件一经创建不允许修改。 def __init__( self, event_id: str, aggregate_id: str, event_type: str, data: Dict[str, Any], occurred_at: Optional[str] None, ) - None: self.event_id event_id self.aggregate_id aggregate_id self.event_type event_type self.data data self.occurred_at occurred_at or datetime.now(timezone.utc).isoformat() def to_dict(self) - Dict[str, Any]: return { event_id: self.event_id, aggregate_id: self.aggregate_id, event_type: self.event_type, data: self.data, occurred_at: self.occurred_at, } class EventStore: 最小事件存储基于追加日志支持重放。 def __init__(self, storage_path: str event_log.jsonl) - None: self.storage_path storage_path # 每个 aggregate_id 对应一个事件索引 self._index: Dict[str, List[Dict[str, Any]]] {} def append(self, event: Event) - None: 追加事件。只许追加禁止修改已有事件。 event_dict event.to_dict() # 每个聚合追加一条事件记录 with open(self.storage_path, a, encodingutf-8) as f: f.write(json.dumps(event_dict, ensure_asciiFalse) \n) # 更新内存索引 if event.aggregate_id not in self._index: self._index[event.aggregate_id] [] self._index[event.aggregate_id].append(event_dict) def load_events(self, aggregate_id: str) - List[Dict[str, Any]]: 读取某个聚合的全部事件。 if aggregate_id in self._index: return self._index[aggregate_id] events: List[Dict[str, Any]] [] with open(self.storage_path, r, encodingutf-8) as f: for line in f: if not line.strip(): continue event_dict json.loads(line) if event_dict[aggregate_id] aggregate_id: events.append(event_dict) self._index[aggregate_id] events return events def replay_all(self, aggregate_id: str) - List[Dict[str, Any]]: 重放事件流返回按时间排序的事件列表。 events self.load_events(aggregate_id) events.sort(keylambda x: x[occurred_at]) return events这段代码做了三件事。第一定义了Event类事件包含event_id、aggregate_id、event_type、data、occurred_at五个字段。第二实现了一个追加写入的 JSONL 文件存储。第三提供了按聚合 ID 加载事件并重放的能力。这里有一些设计细节值得解释。aggregate_id是聚合根的唯一标识。什么是聚合根简单说它代表一个需要保证完整性的业务实体。在 Agent 场景里一个 Agent 实例、一个 Agent 配置版本、一个 Agent 的改进迭代周期都可以作为聚合根。选择哪种粒度取决于你的业务需求。本文示例把单个 Agent 会话作为聚合根这样每个会话内有完整的任务执行、反馈、改进记录。occurred_at使用 UTC 时间而不是本地时间。这避免了不同服务器时区不一致导致的事件排序问题。JSONL 格式之所以够用是因为它简单、轻量、方便调试。你可以在生产环境换成 PostgreSQL 表或 Kafka Topic但这套事件模型和重放逻辑本身不需要改。7. Agent 改进事件定义与状态重建有了基础事件存储接下来定义 Agent 自改进场景里的具体事件和状态重建逻辑。这段代码放在agent_events.py里。它定义三类事件任务执行、反馈接收、配置变更。然后定义一个AgentStateBuilder把事件流重放成 Agent 当前的状态。# 文件路径agent_events.py from typing import Any, Dict, List, Optional from event_store import Event, EventStore class AgentEventFactory: Agent 改进事件的工厂类负责创建各种事件。 staticmethod def task_executed(aggregate_id: str, task_id: str, task_input: str, task_output: str) - Event: return Event( event_idfevt-{task_id}-task, aggregate_idaggregate_id, event_typeTaskExecuted, data{ task_id: task_id, task_input: task_input, task_output: task_output, }, ) staticmethod def feedback_received( aggregate_id: str, task_id: str, feedback_type: str, score: float, comment: str ) - Event: return Event( event_idfevt-{task_id}-feedback, aggregate_idaggregate_id, event_typeFeedbackReceived, data{ task_id: task_id, feedback_type: feedback_type, score: score, comment: comment, }, ) staticmethod def config_changed( aggregate_id: str, config_type: str, old_value: str, new_value: str, change_reason: str, task_id: str, ) - Event: return Event( event_idfevt-{task_id}-config, aggregate_idaggregate_id, event_typeConfigChanged, data{ config_type: config_type, old_value: old_value, new_value: new_value, change_reason: change_reason, task_id: task_id, }, ) class AgentState: 从事件流重建出来的 Agent 状态。 def __init__(self) - None: self.system_prompt: str self.tool_config: Dict[str, Any] {} self.task_history: List[Dict[str, Any]] [] self.feedback_history: List[Dict[str, Any]] [] self.config_history: List[Dict[str, Any]] [] def apply(self, event_dict: Dict[str, Any]) - None: 把单个事件应用到当前状态。 event_type event_dict[event_type] data event_dict[data] if event_type TaskExecuted: self.task_history.append(data) elif event_type FeedbackReceived: self.feedback_history.append(data) elif event_type ConfigChanged: self.config_history.append(data) if data[config_type] system_prompt: self.system_prompt data[new_value] elif data[config_type] tool_config: self.tool_config data[new_value] classmethod def rebuild_from_events(cls, events: List[Dict[str, Any]]) - AgentState: 通过重放事件重建 Agent 状态。 state cls() for event_dict in events: state.apply(event_dict) return state def rebuild_agent_state(event_store: EventStore, aggregate_id: str) - AgentState: 完整的重建入口从事件存储中读取事件重建状态。 events event_store.replay_all(aggregate_id) return AgentState.rebuild_from_events(events)这个文件最重要的部分是AgentState.apply方法。它定义了一个规则每种事件类型如何修改当前状态。这个规则必须和业务语义严格一致。比如ConfigChanged事件里如果config_type是system_prompt那new_value就会覆盖当前的system_prompt如果是tool_config就会覆盖工具配置。这段代码里隐藏着一个值得思考的架构原则Agent 的当前状态永远不直接存储而是通过事件重放计算出来。这意味着如果你发现“当前 Prompt 有问题”你可以回到任何历史节点重建那个节点上的状态然后从这个节点继续演进而不需要担心覆盖任何数据。这种模式在传统事件溯源里被称为“状态重建”在 Agent 场景里有更实际的用途——它天然支持 A/B 测试和分支演进。你可以从同一个历史节点分叉出两个 Agent 版本一个继续按原策略改进一个尝试新策略然后对比评测结果。8. 完整演示Agent 自改进流程的事件溯源实战现在把前面的代码串起来写一个完整演示。场景设计如下有一个 Agent初始系统 Prompt 是“请用简洁的语言回答问题”。它在两个任务上表现不佳收到了用户反馈“回答太啰嗦”。于是 Agent 决定修改自己的系统 Prompt改成“请用不超过三句话回答问题”。修改完成后再执行一个验证任务。完整演示代码在demo_workflow.py里。# 文件路径demo_workflow.py from event_store import EventStore from agent_events import AgentEventFactory, rebuild_agent_state def main() - None: # 1. 初始化事件存储指定日志文件 store EventStore(demo_agent_events.jsonl) # 2. 模拟一个 Agent 会话aggregate_id 为 agent-session-001 aggregate_id agent-session-001 # 3. 记录事件任务执行 1 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-001, task_input解释什么是事件溯源, task_output事件溯源是一种架构模式它不直接存储系统当前状态而是存储所有导致状态改变的事件通过重放事件来推导系统当前状态。这个模式有很多优点比如可以追溯历史状态可以方便地回滚还可以通过重放来复现故障现场。, ) ) # 4. 记录事件任务执行 2 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-002, task_input写出 Python 的 Hello World, task_output所谓 Hello World 程序是编程领域中最基础的一种程序它的功能就是向屏幕输出一句话Hello, World! 在 Python 语言中实现这个功能的方法非常简单只需要调用内置的 print 函数并传入对应的字符串即可。, ) ) # 5. 记录事件用户反馈 store.append( AgentEventFactory.feedback_received( aggregate_idaggregate_id, task_idtask-002, feedback_typeuser_comment, score0.4, comment回答太啰嗦了希望简洁一点。, ) ) # 6. 重建当前状态查看修改前的 Prompt state_before rebuild_agent_state(store, aggregate_id) print( 修改前 Agent 状态 ) print(fSystem Prompt: {state_before.system_prompt}) print(f执行任务数: {len(state_before.task_history)}) print(f接收反馈数: {len(state_before.feedback_history)}) # 7. 记录事件配置变更Agent 自我改进 store.append( AgentEventFactory.config_changed( aggregate_idaggregate_id, config_typesystem_prompt, old_valuestate_before.system_prompt, new_value请用简洁的语言回答问题不超过三句话。, change_reason用户反馈回答过于啰嗦需要压缩输出长度。, task_idtask-002, ) ) # 8. 记录事件任务执行 3使用新 Prompt 后的验证任务 store.append( AgentEventFactory.task_executed( aggregate_idaggregate_id, task_idtask-003, task_input写出 Python 的 Hello World, task_outputprint(Hello, World!), ) ) # 9. 记录事件二次反馈 store.append( AgentEventFactory.feedback_received( aggregate_idaggregate_id, task_idtask-003, feedback_typeuser_comment, score0.9, comment回复简洁明了很好。, ) ) # 10. 再次重建状态查看修改后的结果 state_after rebuild_agent_state(store, aggregate_id) print(\n 修改后 Agent 状态 ) print(fSystem Prompt: {state_after.system_prompt}) print(f配置变更次数: {len(state_after.config_history)}) print(f执行任务数: {len(state_after.task_history)}) # 11. 从事件流中回溯历史重建“第一次改进决策前”的状态 print(\n 回溯历史状态 ) all_events store.replay_all(aggregate_id) # 找到 ConfigChanged 事件的位置 config_change_index None for i, event in enumerate(all_events): if event[event_type] ConfigChanged: config_change_index i break if config_change_index is not None: events_before_change all_events[:config_change_index] from agent_events import AgentState past_state AgentState.rebuild_from_events(events_before_change) print(f修改前的 Prompt: {past_state.system_prompt}) print(f历史任务数量: {len(past_state.task_history)}) if __name__ __main__: main()运行方式python demo_workflow.py预期输出类似 修改前 Agent 状态 System Prompt: 执行任务数: 2 接收反馈数: 1 修改后 Agent 状态 System Prompt: 请用简洁的语言回答问题不超过三句话。 配置变更次数: 1 执行任务数: 3 回溯历史状态 修改前的 Prompt: 历史任务数量: 2输出里有一行值得注意修改前的System Prompt是空的。这个结果是正常的因为模拟场景里初始 Prompt 并没有作为事件写入而 AgentState 只从事件流重建状态。实际工程中Agent 的初始配置应该以“初始化事件”的形式写入事件存储。比如定义一个AgentInitialized事件system_prompt的初始值在那一刻被记录下来。这样重放时才能完整还原状态。9. 运行结果与效果验证运行上面这段演示代码你能验证到三件事。第一件事事件追加是否生效。打开生成的demo_agent_events.jsonl文件可以看到每一行都是一条事件记录。注意看事件是追加在文件末尾的历史记录没有被修改。这个特性保证了事件流不会因为后续写入而被污染。第二件事状态重建是否正确。执行两次rebuild_agent_state第一次拿到的是修改前的状态第二次拿到的是修改后的状态。中间只隔了一次ConfigChanged事件状态却完全不同。这说明事件重放确实起到了“时间旅行”的作用。第三件事历史回溯是否可行。演示代码最后部分手动找到ConfigChanged事件的位置用它之前的事件重建了历史状态。这个操作等价于 Git 的checkout commit它证明了任意历史节点的状态都可以被重建。如果运行失败优先检查以下几个方面Python 版本是否在 3.9 及以上。是否在项目根目录运行event_store.py和agent_events.py是否正常导入。是否重复运行过 demo 脚本导致demo_agent_events.jsonl里积累了多套agent-session-001的事件。如果是删除该文件再重新运行。实际上第二个问题值得展开说一下。如果你连续运行两次python demo_workflow.py第二次运行时事件存储里已经存在agent-session-001的事件脚本会再次追加两套一模一样的事件状态重建结果却不受影响——因为事件是追加写入的重放时只是多了一倍的记录。但你如果统计任务数量会发现翻倍了。这说明在生产环境里你要保证同一个 Agent 会话不会被重复写入通常通过唯一聚合 ID 或幂等机制来解决。10. 自改进 Agent 事件溯源的常见问题与排查思路事件溯源在自改进 Agent 项目里用起来之后会遇到一些比“代码报错”更隐蔽的问题。下面整理了几个高频问题。问题现象可能原因排查方式解决方案重放后状态与预期不符事件顺序错乱依赖了错误的时间字段检查occurred_at是否统一使用 UTC确保事件写入顺序与业务语义一致使用单调递增的序列号代替时间戳排序或者在事件存储层维护全局序重建状态时 Prompt 为空Agent 初始配置没有作为事件写入查看事件流里是否有AgentInitialized事件把初始 Prompt、初始工具配置都建模为初始化事件确保任何状态都来源于事件事件日志文件无限增长没有做事件归档或快照观察日志文件大小和查询延迟定期生成状态快照重放时从最近快照开始而不是从第一条事件开始同一个 Agent 会话数据重复客户端重复提交了事件检查写入路径是否有重试机制在事件存储层做幂等控制按event_id去重历史状态可以重建但无法直接查询“当前配置”事件溯源不适合按照当前状态做高频查询Prisma 或 SQL 直接查询是查不到当前状态的增加投影Projection模块把事件流实时同步到关系表或 NoSQL供查询端使用回滚到一个旧版本后新事件丢失回滚实现有误直接删除了旧事件确认回滚实现是否采用了“追加补偿事件”的方式回滚不要删除历史事件而是追加一个ConfigChanged或EvolutionCommitted类型的补偿事件这个表格里有一个设计原则值得反复强调回滚不等于删除事件。在事件溯源系统里没有任何操作能删除或修改已追加的事件。如果你想从当前 Prompt 回滚到上一个版本正确做法是追加一条ConfigChanged事件把new_value设为上一个版本的旧值。这样做的结果是事件流里完整保留了“V1 - V2 - 回滚到 V1”的全部轨迹而不是只留下一个孤零零的 V1。这条轨迹对你分析“为什么 V2 失败”非常宝贵。另一个常见误区是把事件溯源和“记录日志”混为一谈。普通日志只记录系统运行过程中的关键输出不承担状态恢复的职责事件溯源则把事件当作状态的唯一来源。如果团队里有人试图用日志框架来实现事件溯源会很快发现系统重启后无法重建状态因为日志框架很少保证有序性和完整性。11. 自改进 Agent 事件溯源最佳实践与生产建议前面几节已经把原理和代码讲清楚了这一节补充一些工程落地方案。11.1 从最小事件模型开始不要贪多新手最容易犯的错误是试图把所有细节都建模成事件结果事件类型膨胀到几十种维护成本急剧上升。建议从一版最小事件模型开始初始化事件、任务执行事件、反馈接收事件、配置变更事件、评测验证事件。运行一段时间后再根据实际分析需求增加事件类型。事件是追加式的不存在“改表结构”的迁移压力所以晚一点增加事件类型完全来得及。11.2 用状态快照优化重放性能事件溯源有个天然的成本问题事件越来越多重放会越来越慢。假设一个 Agent 每天执行 10 万次任务每个任务产生 3 条事件一个月就是 900 万条事件。每次重建状态都要从第一条事件开始算性能无法接受。解决方案是定期生成状态快照Snapshot。比如每 1000 条事件把当时的 AgentState 序列化保存下来。重放时先加载最近的快照然后只重放快照之后的事件。这个思路和 Redis 持久化的 AOF RDB 组合非常相似。11.3 投影层负责查询事件流负责事实事件存储本身不适合承担“查询当前 Prompt 是什么”“查询某个分数段的任务记录”这类高频查询。推荐的架构是事件流作为唯一事实来源投影模块监听新事件实时更新一个“当前状态表”或“统计报表表”。比如你可以维护一张agent_current_config表每当出现新的ConfigChanged事件就同步更新这张表。这样查询端可以直接走 SQL而审计和回溯场景再走事件重放。11.4 给改进决策增加人工审批节点自改进 Agent 在开发环境里可以全自动迭代但进入生产环境后建议给“配置变更”这个环节增加人工审批或复核机制。不是所有自动改进都应该直接上线。具体做法是在事件链路里增加一个ImprovementReviewRequired事件。当 Agent 产生改进建议后它不会直接发ConfigChanged事件而是先发一个“待审核”事件。人工或外部系统审核通过后再追加ConfigChanged事件。这个流程保证了配置变更可以被控制也方便追踪每一次正式变更的审批人。11.5 事件设计要有业务语义不要过度抽象有些工程师会把所有事件都塞进一个通用结构比如data字段里装一个 JSON 字符串这样前期写代码很爽后期分析时却很痛苦。更推荐的做法是相同类型的事件结构保持稳定。比如ConfigChanged事件的data里就要固定包含config_type、old_value、new_value、change_reason、task_id。这样后续做数据分析和可视化展现时不需要每次都对data内容做二次解析。11.6 明确事件溯源和安全边界涉及 Agent 自我修改的代码有一个红线必须守住事件存储本身不能存放凭据和密钥。很多 Agent 的改进需要调用 LLM APIAPI Key 应该存放在独立的密钥管理系统里事件里只记录“调用了模型使用了多少 token”不要记录 Authorization 头或 Token 明文。另外对生产环境的 Agent 配置变更务必遵守最小权限原则。Agent 只应该能修改自己负责范围内的 Prompt、工具配置和策略参数不应该有权限去修改事件存储系统本身的代码。把“Agent 的改进权限”和“平台管理员的运维权限”隔离开能避免很多安全风险。11.7 用事件溯源辅助回归评测自改进 Agent 有一个比其他系统更麻烦的问题模型输出有随机性。同样的 Prompt 在不同温度下效果不同同样的修改在不同上下文里可能表现不一致。事件溯源提供的完整历史记录恰好可以支撑更严谨的回归评测。你可以在历史事件流中选出一组具有代表性的任务组成回归测试集然后对 Agent 的多个历史版本分别做评测。这比只依赖最近几条反馈来判断“改得好不好”更可靠。12. 总结与后续学习方向这篇文章的核心判断可以总结成一句话当 Agent 开始修改自身行为配置时它的状态管理就不能再用“保存当前值”的思路而应该采用事件溯源把每一次任务执行、反馈接收、配置变更都变成不可变事件。文章里不仅解释了为什么还给出了一个完整的最小实现。你用它跑通的流程包括定义六类核心事件、实现追加式事件存储、通过事件重放重建 Agent 状态、从事件流回溯任意历史节点、追加补偿事件实现回滚。如果你接下来要继续深入有四个方向值得探索。第一个方向是生产级事件存储选型。PostgreSQL 的 append-only 表、EventStoreDB、Kafka 都可以作为事件存储的底座它们各有优劣适合不同的吞吐和一致性要求。第二个方向是快照与归档策略。事件量大了之后怎么做快照、怎么做压缩、怎么保留全部历史又兼顾查询性能是一套需要在真实数据量下验证的工程方法。第三个方向是投影和查询优化。把事件流同步到 Elasticsearch、ClickHouse 或关系型数据库构建适合分析 Agent 行为的查询视图。第四个方向是结合评测系统做自动回归。事件溯源给你提供了完整的 Agent 版本历史能不能在此基础上做自动化的版本对比和回归验证直接决定了你的 Agent 自改进系统能否从 demo 走向生产。对正在搭建自改进 Agent 的团队我的建议是别急着接复杂框架先用本文的代码把事件模型定义清楚把状态重建流程跑通然后观察真实 Agent 运行数据里哪些事件对你最有价值。事件溯源的核心价值不在于技术本身而在于它逼着你先想清楚一件事你的 Agent 到底在改变什么以及你如何证明这些改变是有效的。