自我改进 Agentself-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_version3说明当前 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建议至少覆盖以下事件事件类型产生时机关键 payloadtask_received收到任务时task_id,task_text,received_attask_completed策略执行结束时task_id,output,success,policy_versionevaluation_recorded评估器打分完成时task_id,score,threshold,evaluatortool_invokedAgent 调用外部工具时tool_name,input,output,latency_mspolicy_updated反思循环调整策略后added_rule,previous_version,change_reason这些事件之间不要求完全扁平。比如task_completed里可以引用task_id这样后续可以通过task_id把同一任务的事件串联起来。policy_updated里记录previous_version和change_reason方便定位策略变更的触发来源。要注意事件 payload 应该保存“当时的事实”而不是“现在的理解”。比如task_completed中保存policy_version1表示执行时使用的是第一版策略。之后策略升级到第二版历史事件里的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, pathEVENT_FILE): self.path path self.seq self._load_last_seq() def _load_last_seq(self): try: with open(self.path, r, encodingutf-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_idstr(uuid.uuid4()), event_typeevent_type, payloadpayload, timestampdatetime.now(timezone.utc).isoformat(), seqself.seq, ) with open(self.path, a, encodingutf-8) as f: f.write(json.dumps(asdict(event), ensure_asciiFalse) \n) return event def read_all(self): events [] try: with open(self.path, r, encodingutf-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: fadd 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 ftask-{i} store.append(task_received, {