自我改进 Agent（self-improving agents）听起来像是一个纯粹的算法问题：多给模型一些反馈，多跑几轮训练，再让评估器筛选出更好的策略。但真正做过 Agent 工程后会发现，自我改进首先是一个数据完整性问题。一个 Agent 要在一次次执行中变好，前提是它知道自己上一次做了什么、为什么那么做、结果如何、下一次哪些决策需要修正。这些信息本质上就是一个不可变的事件序列。换句话说，一个能自我改进的 Agent，内部组织方式天然是事件溯源（event sourced）的。
这篇文章会从这个判断出发，先把“自我改进”拆成可观察的状态变化，再说明事件溯源如何为这种变化提供可靠的数据基础。之后会给出一个可运行的最小 Python 示例，演示一个 Agent 如何通过追加事件日志完成策略更新。最后会讨论快照、投影、常见故障和生产落地的工程要点。读完你会理解：事件溯源不是后端系统才需要关心的持久化模式，它也是 Agent 记忆、反思、评估和策略迭代的统一抽象。
1. 自我改进的 Agent 如果不记录历史，就谈不上“改进”
1.1 自我改进 Agent 到底在改什么
很多讨论把“自我改进 Agent”等同于“让大模型自己优化 prompt”，但这只是其中一种表现。站在工程角度看，Agent 可以改进的对象至少包括以下几类：
- 模型参数：通过微调或强化学习更新底层权重。
- 提示词策略：调整 system prompt、few-shot 示例、推理步骤模板。
- 工具选择逻辑：面对同一目标，尝试不同工具或调整调用顺序。
- 评估标准：从只看最终答案，到同时检查中间步骤、耗时、成本。
- 元策略：决定“什么时候该反思”“什么时候该重试”“什么时候该换方案”。
无论改进的是哪一层，它都对应一个状态迁移：从策略 A 变成策略 B。比如“之前遇到加法任务只会原样返回，现在遇到加法任务会调用计算器”，这种变化不是凭空发生的，它必须基于某次或某几次具体执行结果。如果 Agent 没有保存这些结果，下一次“改进”就只能靠随机重试或模型参数里的隐性记忆，无法回答一个关键问题：现在的策略为什么比之前的策略好。
1.2 复盘和回放是改进的前提
一次 Agent 任务的生命周期通常包含感知、决策、执行、评估四个阶段。感知阶段拿到任务描述，决策阶段选择策略或工具，执行阶段产生输出，评估阶段判断输出是否符合预期。这四个阶段任何一个节点都可以成为改进的素材。
但只有素材还不够。真正的改进需要“复盘”，而复盘需要“回放”。一个任务失败了，你想知道失败是因为任务理解错误、工具调用错误，还是评估标准不合理。此时你需要回到那个时间点，查看当时的输入、当时的策略版本、当时的工具返回、当时的评估分数。没有事件记录，你只能看到最终失败结果，无法判断失败发生在哪个环节。
更麻烦的是，一次策略调整后效果变好，可能是真实改进，也可能是测试任务太简单造成的假象。要区分这两种情况，必须拿同一批历史任务去重放新旧策略，并对比评估结果。事件日志在这里的作用，就像版本控制里的提交历史：它让你可以随时切回任意版本，验证某次变更是否真的带来了改进。
1.3 从“状态存储”思维切到“事件日志”思维
传统系统的常见做法是只保存当前状态。比如策略表里有一行记录current_prompt_version=3，说明当前 Agent 使用第三版提示词。但你不知道第三版是怎么来的，是修改了哪个失败样本得到的，也不知道第一版到第二版之间发生了什么。
如果 Agent 要自我改进，这种“只存结果”的模型是不够的。改进过程中最有价值的不是“当前策略”，而是“策略为什么变成当前这样”。因此需要把视角切换成事件日志：不直接保存最终状态，而是保存每一条导致状态变化的事件。当前状态可以从事件流中重新计算出来。
这件事听起来像数据库里的 binlog，也像 Git 的提交历史。事件溯源正是把这种思想变成一种明确的架构模式：所有状态变化都以不可变事件追加到日志里，状态本身只是一个可以由事件流推导出来的投影。
2. 为什么事件溯源是自我改进 Agent 的天然底座
2.1 事件溯源的核心不是日志，而是“状态可推导”
事件溯源（Event Sourcing）常被误解为“把操作过程记下来”。实际上，普通的操作日志只是辅助排错的记录，系统状态仍然直接存储在业务表里。事件溯源的真正特征是：事件的序列是状态的唯一事实来源。
也就是说，当前状态不是从某个current_state字段读出来的，而是通过按照顺序重放事件得到的。比如一个 Agent 的策略状态是“支持 reverse、upper、echo 三种规则”，它不是靠一个字段保存的，而是因为事件流里有一条policy_updated: add rule upper和一条policy_updated: add rule echo。删除这两条事件，状态就不存在；追加新事件，状态就发生变化。
这个特性对自我改进 Agent 非常关键。因为 Agent 的策略更新本质上就是“追加一条策略变更事件”，而不是“覆盖一个策略变量”。当你怀疑当前策略有问题时，可以回放事件流，定位到是哪次变更引入了问题，甚至可以在副本上回滚到最后一条可靠事件，重新执行后续任务。
2.2 自我改进过程如何映射到事件溯源
把 Agent 的一次完整改进循环映射到事件溯源，会产生这样一条事件流：
- Agent 收到任务，追加
task_received事件。 - Agent 选择策略并执行，追加
task_completed事件。 - 评估器对输出打分，追加
evaluation_recorded事件。 - 反思循环读取事件流，发现某类任务失败率过高。
- 反思循环生成新策略，追加
policy_updated事件。
这里最关键的一点是：策略更新也被建模成事件。很多人在做 Agent 时会把“策略”单独存成一个配置对象，失败了就改配置。但如果策略更新不进入事件流，那么事件流只能回答“Agent 做过什么”，不能回答“Agent 如何变成现在这样”。自我改进的过程要想可审计、可回滚、可复现，策略更新就必须与任务执行一样，成为事件流的一部分。
在这个模型里，“经验”不再是一个模糊的概念，而是一串有序的、不可变的事件。每个事件都携带任务上下文、执行结果、评估分数和策略变更信息。后续无论是做数据筛选、模型微调，还是进行 A/B 对比，都可以直接从事件流中提取。
2.3 命令、事件、投影、快照的关系
事件溯源涉及四个经常被混淆的概念：命令、事件、投影、快照。
- 命令表示“希望发生什么”，是一个意图，可能被拒绝。例如
ImprovePolicy命令。 - 事件表示“已经发生了什么”，是一个事实，不可变。例如
PolicyUpdated事件。 - 投影表示“从事件流中生成的可查询状态”，可以理解为事件的读取模型。
- 快照表示“某一时刻的投影备份”，用来避免每次都从第一条事件开始重放。
放到 Agent 场景中，对应关系如下：
| 事件溯源概念 | 在 Agent 中的对应物 | 作用 |
|---|---|---|
| 命令 | 接收任务、请求工具调用、触发反思 | 表达意图，可能不被执行 |
| 事件 | 任务执行、评估打分、策略更新 | 记录已经发生的事实 |
| 投影 | 当前策略、失败统计、改进报告 | 从事件流中生成可查询状态 |
| 快照 | 定期保存的策略状态副本 | 加速重放，避免全量计算 |
这个模型的好处是，命令和事件分离。命令层可以做校验、限流、权限控制；事件层只负责忠实记录事实。Agent 的“自我改进”就发生在命令层与事件层的交互之间：反思循环消费事件投影，生成新的改进命令，改进成功后追加新的事件。
3. 设计一个事件溯源式自我改进 Agent：事件模型先行
3.1 需要记录的几类关键事件
设计事件溯源系统时，第一件事不是写代码，而是定义事件类型。事件类型决定了系统可以回答哪些问题。对于一个自我改进 Agent，建议至少覆盖以下事件：
| 事件类型 | 产生时机 | 关键 payload |
|---|---|---|
task_received | 收到任务时 | task_id,task_text,received_at |
task_completed | 策略执行结束时 | task_id,output,success,policy_version |
evaluation_recorded | 评估器打分完成时 | task_id,score,threshold,evaluator |
tool_invoked | Agent 调用外部工具时 | tool_name,input,output,latency_ms |
policy_updated | 反思循环调整策略后 | added_rule,previous_version,change_reason |
这些事件之间不要求完全扁平。比如task_completed里可以引用task_id，这样后续可以通过task_id把同一任务的事件串联起来。policy_updated里记录previous_version和change_reason，方便定位策略变更的触发来源。
要注意，事件 payload 应该保存“当时的事实”，而不是“现在的理解”。比如task_completed中保存policy_version=1，表示执行时使用的是第一版策略。之后策略升级到第二版，历史事件里的policy_version不应被修改。一旦修改历史事件，事件溯源就退化成普通数据库更新。
3.2 事件日志的数据结构
最小实现中，事件日志可以采用 JSONL 文件，每行一个事件。每个事件需要包含以下公共字段：
event_id：全局唯一事件 ID。event_type：事件类型。payload：事件负载，保存与该事件相关的业务数据。timestamp：事件发生时间。seq：单调递增序号，用于保证重放顺序。
JSONL 格式便于学习和查看，也便于用命令行工具直接检索：
{ "event_id": "7e2a6c18-1d0f-4b2e-b0a9-5c3d2e4f8a10", "event_type": "policy_updated", "payload": { "added_rule": "upper", "change_reason": "detect high failure rate on upper tasks" }, "timestamp": "2025-06-01T12:30:00.123Z", "seq": 7 }生产环境通常不会使用 JSONL 文件，而会使用数据库表或消息队列。例如关系型数据库中可以设计一张agent_events表：
CREATE TABLE agent_events ( seq BIGSERIAL PRIMARY KEY, event_id UUID NOT NULL UNIQUE, aggregate_id VARCHAR(64) NOT NULL, event_type VARCHAR(64) NOT NULL, payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_agent_events_aggregate ON agent_events (aggregate_id, seq);这里的aggregate_id非常重要。它可以对应一个任务、一个会话或一个 Agent 实例。通过aggregate_id可以只重放某个任务的事件，而不必加载全局事件流。
3.3 改进循环如何运转
一个事件溯源式自我改进 Agent 的改进循环可以这样描述：
- 执行阶段：任务进入后追加
task_received，策略执行后追加task_completed。 - 评估阶段：评估器对输出打分，追加
evaluation_recorded。 - 阈值判断：如果分数低于阈值，进入反思阶段。
- 反思阶段：读取最近一段时间的事件投影，分析失败任务的共性。
- 策略更新阶段：生成新的策略规则，追加
policy_updated。 - 验证阶段：用新策略重放失败任务，追加
verification_recorded事件。
这个循环的关键不是“反思多聪明”，而是每一步都留下事件。反思本身可以很简单，比如统计哪个前缀任务失败最多，就为它注册一个新规则。真正让 Agent 稳定改进的，是事件流保证了每一步都有据可查。
4. 最小可运行示例：用 Python 实现事件驱动改进循环
4.1 环境准备与项目结构
这个示例只使用 Python 标准库，不需要安装任何第三方依赖。建议使用 Python 3.10 及以上版本，因为示例中使用了dataclass和类型标注。创建一个学习目录，例如agent_event_sourcing_demo，目录结构如下：
agent_event_sourcing_demo/ └── self_improving_agent.py这个示例的核心逻辑都放在一个文件里，便于阅读和运行。实际项目可以拆分成事件存储、策略、评估、反思等多个模块，但学习阶段先看单文件更容易理解事件流如何运作。
4.2 事件模型与事件存储
先定义事件结构和基于 JSONL 的事件存储。事件存储只需要两个能力：追加事件、读取全部事件。追加时自动生成序号seq，读取时按文件行顺序返回事件。
import json import uuid from dataclasses import asdict, dataclass from datetime import datetime, timezone EVENT_FILE = "events.jsonl" @dataclass class Event: event_id: str event_type: str payload: dict timestamp: str seq: int class JsonlEventStore: def __init__(self, path=EVENT_FILE): self.path = path self.seq = self._load_last_seq() def _load_last_seq(self): try: with open(self.path, "r", encoding="utf-8") as f: last_line = None for line in f: line = line.strip() if line: last_line = line if last_line is None: return 0 return int(json.loads(last_line)["seq"]) except FileNotFoundError: return 0 def append(self, event_type: str, payload: dict) -> Event: self.seq += 1 event = Event( event_id=str(uuid.uuid4()), event_type=event_type, payload=payload, timestamp=datetime.now(timezone.utc).isoformat(), seq=self.seq, ) with open(self.path, "a", encoding="utf-8") as f: f.write(json.dumps(asdict(event), ensure_ascii=False) + "\n") return event def read_all(self): events = [] try: with open(self.path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: data = json.loads(line) events.append(Event(**data)) except FileNotFoundError: pass return events这里的关键点是：读取顺序等于写入顺序，而写入顺序由seq记录。生产环境中如果使用关系型数据库，可以使用自增主键保证顺序；如果使用消息队列，则需要确保同一个aggregate_id的事件进入同一个分区。
4.3 Agent 策略与执行逻辑
接下来定义 AgentPolicy。这个策略非常简单，内部维护一个规则字典，每种任务前缀对应一个处理函数。初始策略只支持reverse规则，也就是说 Agent 刚启动时只会做字符串反转，遇到其他任务会失败。
class AgentPolicy: def __init__(self): self.version = 1 self.rules = { "reverse": lambda text: text[::-1], } def add_rule(self, rule_name: str) -> dict: if rule_name == "upper": self.rules["upper"] = lambda text: text.upper() elif rule_name == "echo": self.rules["echo"] = lambda text: text else: self.rules[rule_name] = lambda text: text self.version += 1 return { "added_rule": rule_name, "change_reason": f"add rule {rule_name}", } def execute(self, task_text: str): prefix, _, content = task_text.partition(":") rule = self.rules.get(prefix) if rule is None: return task_text, False return rule(content), True执行结果由execute返回两个值：输出内容和是否成功。如果规则不存在，会返回原始任务文本，并标记为失败。这个设计对应真实 Agent 中“某个策略分支不可用”的情况。
评估函数根据任务前缀计算期望输出，然后与实际输出比较：
def evaluate(task_text: str, output: str) -> float: prefix, _, content = task_text.partition(":") if prefix == "reverse": expected = content[::-1] elif prefix == "upper": expected = content.upper() elif prefix == "echo": expected = content else: expected = None if expected is None: return 0.0 return 1.0 if output == expected else 0.0评估结果不写入任务状态，而是作为独立事件追加到事件日志，这样可以保留多次评估视角。
4.4 反思循环与策略更新
反思循环读取事件流，寻找失败的task_completed事件，提取失败任务的任务前缀。如果当前策略还没有支持该前缀的规则，就调用add_rule添加规则，并把policy_updated事件追加到日志。
def reflect_and_update(store: JsonlEventStore, policy: AgentPolicy): events = store.read_all() failed_prefixes = set() for event in events: if event.event_type == "task_completed": payload = event.payload if payload.get("success") is False: task_text = payload["task_text"] prefix, _, _ = task_text.partition(":") if prefix not in policy.rules: failed_prefixes.add(prefix) changes = [] for prefix in sorted(failed_prefixes): change = policy.add_rule(prefix) store.append("policy_updated", change) changes.append(change) return changes这里反射逻辑非常朴素：只要发现某类前缀任务失败，并且策略里没有对应规则，就直接添加规则。真实项目中的反思会更复杂，比如先判断失败样本是否足够多，再判断新规则是否会影响已有任务。但这个最小示例足以说明事件流如何驱动策略变化。
4.5 从事件流重建策略状态
事件溯源的关键能力是从事件流重建状态。下面的函数读取全部事件，只处理policy_updated事件，重新执行规则添加操作：
def rebuild_policy_from_events(events): policy = AgentPolicy() for event in events: if event.event_type == "policy_updated": rule_name = event.payload["added_rule"] policy.add_rule(rule_name) return policy注意，重建时不读取事件 payload 里的version字段，而是按照事件顺序重新调用add_rule。这样即使事件中的描述信息有误，状态仍然由事件序列决定。这个函数就是“状态可推导”的直接体现。
4.6 主流程与运行验证
主流程负责执行一组任务，并在每个任务后追加事件。任务列表故意设计成让 Agent 在运行过程中遇到两类失败：upper和echo。第一次遇到upper:hello时失败并触发反思，之后遇到upper:world时已经成功，说明策略更新产生了实际效果。
def main(): store = JsonlEventStore() policy = AgentPolicy() tasks = [ "reverse:abc", "upper:hello", "upper:world", "reverse:world", "echo:keep", ] for i, task_text in enumerate(tasks): task_id = f"task-{i}" store.append("task_received", {
