自改进Agent的事件溯源：构建可回放、可归因的运行记忆系统

过去一年几乎所有做 AI Agent 的团队都在讨论同一个问题怎么让 Agent 自己变得更好。但真正开始落地的时候很多人会撞上一面墙——我们根本没有一份完整的、可回放、可归因的“经历”。Agent 的每一次模型调用、每一次工具返回、每一个错误决策散落在控制台日志、数据库字段和一堆 prompt 版本里彼此之间没有关联。任务失败后你只能靠记忆复盘甚至只能靠猜。自我改进的前提是拥有高质量的体验数据而体验数据的前提是先把运行过程完整地事件化。所以我的判断很明确Self-improving agents are event sourced自改进 Agent 的工程底座天然就是事件溯源。这篇文章会先解释为什么 Event Sourcing 不是“顺手选一个架构”而是自改进系统的必要条件然后给出事件模型设计、存储方案、回放逻辑和改进管线最后用一个最小示例把“记录—回放—反思—改进”的闭环跑通。读完你可以直接照着搭一套属于自己的 Agent 体验数据基础设施。1. 为什么“自我改进 Agent”听起来很美好落地却很难先看一个常见场景。你写了一个能调用工具的 Agent让它完成“帮用户查天气并安排行程”这种多步任务。前几次跑下来任务失败率很高。你希望 Agent 能从失败中学习于是你准备改进它。问题来了你如何知道它到底在哪一步失败如果你只靠打印日志你看到的是零散的tool_call、result、final_answer。日志之间没有层级关系你不知道某次错误的工具参数是模型自己拍脑袋生成的还是因为前一步的上下文被截断了。你也不知道用户反馈是“回答不对”还是“过程太慢”。如果你只把最终结果存到数据库那更麻烦。你存下来的是一个“状态”但状态不包含过程。Agent 的改进恰恰需要过程它先做了什么、看到了什么、修改了什么、结果如何。再往深一层说Agent 的非确定性让改进变得异常复杂。同样的任务今天跑和明天跑结果可能完全不同。你很难判断某个改进到底是“真的有效”还是“这次运气好”。这种情况下任何不基于完整历史数据的改进本质上都是在做实验但没有实验记录。从相关综述对“self-improving agents”的梳理来看这个领域正在从“self-evolution自我进化”走向“meta-evolution元进化”。自我进化是让 Agent 在具体任务上变得更好元进化是让 Agent 改进“自己改进自己的方式”。无论哪个阶段都存在一个共同前提你必须把“发生了什么”这件事变成可读取、可回放、可评估的数据。否则“进化”和“随机调整”没有区别。所以这篇文章真正要解决的问题是如何为自改进 Agent 建立一套可靠的“运行记忆系统”让每一次尝试都能被完整记录、事后复盘、并成为下一次改进的依据。2. 核心概念Self-improving Agent 与 Event Sourcing2.1 什么是 Self-improving AgentSelf-improving Agent 指能够根据自身运行产生的经验来改进策略的智能体。改进的对象可以是 prompt 指令、工具选择规则、上下文压缩策略、模型微调数据集甚至是改进流程本身。这里要区分两个层次Self-evolution自我进化Agent 在同一任务或同一任务分布上通过反复试错提升性能。例如一个写代码的 Agent 发现自己经常在“依赖安装”这一步骤失败于是它学会在代码执行前先检查包管理器。Meta-evolution元进化Agent 开始改进自己的改进机制。例如它发现“用最近 5 次失败的轨迹做反思”比“只反思最后一次失败”效果更好于是自动调整反思策略。无论哪个层次Agent 都需要“经验”。经验不是抽象的它就是一系列按时间顺序排列的输入、决策、动作、观察、结果、反馈。这套东西本质上就是一个事件流。2.2 什么是 Event SourcingEvent Sourcing 是一种软件架构模式。它的核心思想是不把当前状态当作唯一事实来源而是把引起状态变化的事件当作唯一事实来源。当前状态只是对这些事件做一次“投影Projection”得到的结果。举个例子。一个订单系统如果做成传统 CRUD数据库里会有一个orders表里面有个status字段从CREATED改成PAID再改成SHIPPED。但在 Event Sourcing 模式下你记录的是事件1OrderCreated订单创建事件2OrderPaid支付完成事件3OrderShipped已发货“当前状态”只是按顺序执行这三个事件后得到的结果。数据库里不会直接存status而是存一份不可变的、只能追加的事件表。Event Sourcing 最大的优势不是“存储方式新颖”而是它让系统拥有了时间维度和透明度。任何时刻的状态都可以通过回放事件重建任何问题都可以通过查看事件序列定位。它天然适合需要审计、复现、分析的领域。2.3 两者如何走到一起Agent 的运行时状态非常复杂。一个会话中有 LLM 消息列表、工具调用记录、临时变量、用户反馈、策略配置版本。如果你把这些都折叠成一个“状态”存进数据库那么事后你只能看到最终结果过程完全不可见。但如果把 Agent 的每一次内部决策、每一次工具调用、每一次结果返回都当作事件追加到日志中整个运行过程就变成了一条连续的事件流。自我改进就不再是“看着结果猜测原因”而是“回放事件流找到失败发生的准确位置再针对性地修改策略”。一句话总结Self-improving Agent 要改进的对象正是 Event Sourcing 所要记录的那条事件流。两者在数据本质上高度一致。3. 为什么 Self-improving Agents 天然适合 Event Sourcing3.1 改进需要可回放的历史任何“改进”都需要对照实验。你改了 prompt想要验证效果是否变好前提是能重现改进前的运行过程。如果历史数据只是一堆最终答案你无法重现当时的上下文和决策链条。事件日志因为是追加式、按顺序存储的天然支持精确回放。3.2 改进需要可归因的因果链一个多步 Agent 的失败原因往往靠后。比如 Agent 在第三步使用了错误工具真实原因可能是第二步的上下文没有包含必要信息。只有把事件链完整保存下来才能做因果归因。你会明确看到DecisionMade → ToolInvoked → ToolResultReceived中间到底哪一环断了。3.3 改进需要可差分的历史版本事件日志里可以附带agent_version、policy_version、model_config等字段。这样你在做策略更新时可以同时保留多个版本的实验结果。改进前的事件和改进后的事件放在同一个表里通过版本字段区分然后做对比分析。这比“新配置覆盖旧配置”的方式要可靠得多。3.4 改进需要可审计的安全边界Agent 在改进自己的过程中最危险的事情是“策略被悄悄改坏”。事件日志天然是一个审计日志谁在什么时候基于哪些历史事件更新了策略更新后的策略 diff 是什么都一目了然。这对生产环境尤其重要。下面用一个表格对比三种数据记录方式对比维度控制台日志状态型数据库CRUD事件日志Event Sourcing能否回放完整过程部分依赖开发者手动打点不能只有当前状态能追加事件即可回放能否定位因果链困难日志间缺少关联不能能通过 trace_id 串联能否做版本对比需自己额外记录困难旧状态已覆盖能事件本身带版本字段是否适合离线评测需要清洗无法还原过程天然适合生产环境安全审计弱弱强4. 事件模型设计最小可用事件结构要构建一个事件源驱动的 Agent第一步是定义事件模型。我建议从下面这套最小字段表入手字段类型说明sequenceLong自增序列用于排序event_idString全局唯一事件 ID幂等去重agent_idString哪个 Agent 产生的episode_idString同一次任务会话的 IDtrace_idString用于关联父子事件链parent_idString父事件 ID可选event_typeString事件类型event_versionInteger事件 schema 版本event_timeTimestamp事件发生时间payloadJSON事件主体内容在此基础上一个自改进 Agent 最核心的事件类型包括TaskReceived收到新任务ContextPrepared上下文组装完成DecisionMade模型输出一个决策ToolInvoked决定调用某个工具ToolResultReceived收到工具返回结果MessageDelivered向用户输出了一条消息HumanFeedbackReceived收到人工反馈EpisodeFinished一次任务结束PolicyUpdated策略发生变更来看一个DecisionMade事件的 JSON 示例{ event_id: evt_ab8312c4, agent_id: demo-agent-001, episode_id: ep_20250324_001, trace_id: trace_calc_0008, parent_id: null, event_type: DecisionMade, event_version: 1, event_time: 2025-03-24T10:15:32Z, payload: { step: 2, model_output: { action: call_tool, tool: calculator, tool_input: {expression: 123*456} }, context_summary: user asked to calculate 123*456 } }这里有一个容易被忽略的工程细节DecisionMade事件里除了模型输出最好还保存context_summary或者完整的消息列表快照。因为 LLM 调用是非确定性的回放时无法重新生成历史输出。事件日志不是“重新推理”而是“保存事实”。只有把原始输出保存下来回放才是准确的。事件类型不是越多越好。设计原则是凡是你未来可能需要分析、复盘、改进的关键节点都应该成为事件但普通日志和中间过程不一定要进事件表。事件表是结构化的事实日志是辅助排查的细节两者分开管理。5. 工程落地如何把 Agent 改造成事件源5.1 第一步在 Agent 运行时统一记录事件最直接的方式是在 Agent 的主循环里埋入一个事件记录器。下面这段 Python 代码展示了一个最小实现用 SQLite 作为存储足够本地开发和单机实验使用。文件路径event_store.pyimport json import sqlite3 import time import uuid class EventStore: def __init__(self, db_path: str agent_events.db): self.db_path db_path self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS agent_events ( seq INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL, agent_id TEXT NOT NULL, episode_id TEXT NOT NULL, trace_id TEXT NOT NULL, parent_id TEXT, event_type TEXT NOT NULL, event_version INTEGER NOT NULL DEFAULT 1, event_time REAL NOT NULL, payload TEXT NOT NULL ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_events_episode ON agent_events(episode_id, seq)) conn.execute(CREATE INDEX IF NOT EXISTS idx_events_trace ON agent_events(trace_id, seq)) def append(self, agent_id, episode_id, trace_id, event_type, payload, parent_idNone, event_version1): event { event_id: str(uuid.uuid4()), agent_id: agent_id, episode_id: episode_id, trace_id: trace_id, parent_id: parent_id, event_type: event_type, event_version: event_version, event_time: time.time(), payload: payload, } with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO agent_events (event_id, agent_id, episode_id, trace_id, parent_id, event_type, event_version, event_time, payload) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , (event[event_id], event[agent_id], event[episode_id], event[trace_id], event[parent_id], event[event_type], event[event_version], event[event_time], json.dumps(payload, ensure_asciiFalse)), ) return event注意这个 EventStore 是 append-only 的只有插入操作没有更新和删除。如果你未来团队协作或上生产建议在数据库层面把事件的写权限限制为“只能 INSERT”。接下来在 Agent 主循环中调用它。文件路径agent.pyimport json import uuid from event_store import EventStore class SelfImprovingAgent: def __init__(self, agent_id: str, episode_id: str, store: EventStore, llm_fn): self.agent_id agent_id self.episode_id episode_id self.trace_id uuid.uuid4().hex self.store store self.llm_fn llm_fn def run_task(self, task: str, max_steps: int 10): self.store.append( self.agent_id, self.episode_id, self.trace_id, TaskReceived, {task: task}, ) messages [ {role: system, content: self._load_policy()}, {role: user, content: task}, ] for step in range(max_steps): output self.llm_fn(messages) self.store.append( self.agent_id, self.episode_id, self.trace_id, DecisionMade, {step: step, messages: messages, model_output: output}, ) if output.get(action) finish: final_answer output.get(final_answer) self.store.append( self.agent_id, self.episode_id, self.trace_id, EpisodeFinished, {final_answer: final_answer}, ) return final_answer if output.get(action) call_tool: tool output[tool] tool_input output[tool_input] self.store.append( self.agent_id, self.episode_id, self.trace_id, ToolInvoked, {tool: tool, tool_input: tool_input}, parent_idself.trace_id, ) result self._run_tool(tool, tool_input) self.store.append( self.agent_id, self.episode_id, self.trace_id, ToolResultReceived, {tool: tool, result: result}, ) messages.append({role: assistant, content: json.dumps(output, ensure_asciiFalse)}) messages.append({role: tool, content: str(result)}) return None def _run_tool(self, tool: str, args: dict): if tool calculator: expr args.get(expression, ) return eval(expr) # 注意生产环境请勿直接 eval这里仅为演示 raise ValueError(funknown tool: {tool}) def _load_policy(self) - str: with open(policy/default.txt, r, encodingutf-8) as f: return f.read()在这个设计里llm_fn是外部注入的这样方便测试和替换真实模型。Agent 不关心事件存储的具体实现它只负责把关键决策点追加到事件流中。5.2 第二步选择事件存储本地原型阶段SQLite 足够。进入团队协作和生产环境后建议迁移到 PostgreSQL并使用 JSONB 保存 payload。下面是 PostgreSQL 的建表语句文件路径schema.sqlCREATE TABLE agent_events ( seq BIGSERIAL PRIMARY KEY, event_id TEXT NOT NULL UNIQUE, agent_id TEXT NOT NULL, episode_id TEXT NOT NULL, trace_id TEXT NOT NULL, parent_id TEXT, event_type TEXT NOT NULL, event_version INTEGER NOT NULL DEFAULT 1, event_time TIMESTAMPTZ NOT NULL DEFAULT now(), payload JSONB NOT NULL ); CREATE INDEX idx_agent_events_episode ON agent_events (episode_id, seq); CREATE INDEX idx_agent_events_trace ON agent_events (trace_id, seq);如果你的团队已经使用 Kafka、EventStoreDB 或 Axon也可以直接借用现成的事件基础设施。但我不建议一开始就上重武器。先用一个表把事件落下来数据量大了再换迁移成本并不高。5.3 第三步回放与投影事件记录的部分完成之后回放是核心能力。回放有两个用途一是重建某次任务的完整上下文二是生成训练和改进所需的结构化轨迹。下面是一段回放代码它从一个episode_id读取事件并重建消息列表和工具调用序列。文件路径replay.pyimport json import sqlite3 def load_episode(db_path: str, episode_id: str): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row rows conn.execute( SELECT seq, event_type, payload, event_time FROM agent_events WHERE episode_id ? ORDER BY seq ASC , (episode_id,), ).fetchall() return [dict(row) for row in rows] def reconstruct_trajectory(events): messages [] tool_calls [] for ev in events: payload json.loads(ev[payload]) if ev[event_type] DecisionMade: messages.append({ step: payload.get(step), model_output: payload.get(model_output), }) elif ev[event_type] ToolInvoked: tool_calls.append({ tool: payload[tool], tool_input: payload[tool_input], }) elif ev[event_type] ToolResultReceived: tool_calls[-1][result] payload.get(result) elif ev[event_type] EpisodeFinished: messages.append({final_answer: payload.get(final_answer)}) return messages, tool_calls回放得出的轨迹信息可以用于多种用途生成反思 prompt、构造微调训练集、计算工具成功率、对比不同版本策略的效果。5.4 第四步从事件流到改进管线有了完整事件流改进管线就可以开始工作。最基础的一条改进路径是“反思式改进”从事件流中筛选某次任务的全部事件。用脚本判断任务是否成功或者人工给这次任务打标签。把任务描述、事件轨迹、最终结果组装成反思 prompt交给一个反思模型。反思模型输出失败原因和改进建议。将建议写入策略文件或生成微调样本。这里要特别注意反思模型看到的事件轨迹不能只有结果必须包含过程和中间输入输出。因为改进 Agent 的关键是分析失败发生在哪一个决策点而不是简单地判断对错。6. 完整示例一个可运行的“记录-回放-反思-改进”循环下面我把上面的代码整合成一个最小示例。目录结构如下agent-improve-demo/ ├── agent.py ├── event_store.py ├── replay.py ├── policy/ │ └── default.txt ├── agent_config.yaml ├── main.py └── improve.py先看策略文件和配置文件。文件路径policy/default.txt你是一个擅长数学计算的助手。 规则 1. 如果任务是数值计算优先使用 calculator 工具。 2. 一次只调用一个工具。 3. 得到结果后直接给出最终答案。文件路径agent_config.yamlagent: id: demo-agent-001 policy_file: policy/default.txt max_steps: 8 llm_provider: openai_compatible event_store: path: agent_events.db接着写主程序把整个流程串起来。文件路径main.pyimport yaml from agent import SelfImprovingAgent from event_store import EventStore with open(agent_config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) store EventStore(config[event_store][path]) # 模拟 LLM 调用实际项目中请替换为真实模型客户端 def fake_llm(messages): last_user messages[-1][content] if 456 in str(last_user): return { action: call_tool, tool: calculator, tool_input: {expression: 123*456}, } return {action: finish, final_answer: 56088} agent_cfg config[agent] episode_id ep_demo_0001 agent SelfImprovingAgent( agent_idagent_cfg[id], episode_idepisode_id, storestore, llm_fnfake_llm, ) final_answer agent.run_task(请计算 123*456 的结果) print(final_answer:, final_answer)改进管线放在improve.py中。文件路径improve.pyimport json from replay import load_episode, reconstruct_trajectory def build_reflection_prompt(task, trajectory, final_answer): return f 你是一个 Agent 反思引擎。下面是某次任务的轨迹 任务{task} 轨迹 {json.dumps(trajectory, ensure_asciiFalse, indent2)} 最终答案{final_answer} 请分析 1. 这个 Agent 在哪些步骤做得好 2. 在哪些步骤可能出错 3. 如果要优化策略最值得修改的一条规则是什么 def reflect_on_episode(db_path, episode_id, task): events load_episode(db_path, episode_id) trajectory reconstruct_trajectory(events) final_answer trajectory[0][-1].get(final_answer) if trajectory[0] else None prompt build_reflection_prompt(task, trajectory, final_answer) # 这里调用真实 LLM 获取反思结果 reflection { success_steps: [调用了 calculator], risk_steps: [没有校验表达式安全性], policy_update: 禁止使用 eval 执行任意表达式改为白名单计算器。, } return reflection def update_policy(policy_path: str, reflection: dict): with open(policy_path, r, encodingutf-8) as f: content f.read() content \n改进记录 reflection[policy_update] \n with open(policy_path, w, encodingutf-8) as f: f.write(content) if __name__ __main__: reflection reflect_on_episode(agent_events.db, ep_demo_0001, 请计算 123*456 的结果) print(reflection:, reflection) update_policy(policy/default.txt, reflection)运行顺序如下# 1. 安装依赖 pip install pyyaml # 2. 运行一次任务记录事件 python main.py # 3. 回放本次任务并进行反思 python improve.py这个示例虽然用了fake_llm模拟模型但结构上已经覆盖了真实项目需要的全部环节。你只需要把fake_llm替换成真实模型调用把reflect_on_episode里的注释逻辑换成真实 LLM 调用即可。7. 运行结果与效果验证运行python main.py后预期输出类似于final_answer: 56088然后运行python improve.py预期输出reflection: {success_steps: [调用了 calculator], risk_steps: [没有校验表达式安全性], policy_update: 禁止使用 eval 执行任意表达式改为白名单计算器。}同时policy/default.txt文件末尾会追加一条改进记录。但这只是“跑通”不算“验证成功”。要验证事件溯源体系真的有效你需要检查三件事第一事件完整性。运行一次任务后查询事件表应该看到TaskReceived、DecisionMade、ToolInvoked、ToolResultReceived、EpisodeFinished这些事件且顺序正确。sqlite3 agent_events.db SELECT seq, event_type FROM agent_events ORDER BY seq ASC;预期输出类似1|TaskReceived 2|DecisionMade 3|ToolInvoked 4|ToolResultReceived 5|DecisionMade 6|EpisodeFinished第二回放一致性。用load_episode读取同一episode_id应该能重建出与运行时完全一致的消息序列和工具调用。如果回放结果和实际运行不一致说明事件记录缺失或顺序错乱这是第一个要修的 Bug。第三改进有效性。把改进前和改进后各跑 N 次任务对比成功率。如果改进后成功率没有提升说明反思方向不对或策略修改没有落到关键环节。此时不要盲目继续改要回到事件流里重新分析失败点。特别提醒由于 LLM 有随机性任何“改进是否有效”的判断都不能只看一两次运行。至少要跑多轮用成功率、工具调用错误率、平均步数等指标综合评估。8. 常见问题与排查思路问题现象可能原因排查方式解决方案事件记录缺失回放不完整某些分支没有调用 EventStore.append检查 Agent 所有 return 路径是否有提前退出在 Agent 的 finally 块中统一记录结束事件回放结果与运行时不一致事件中没有保存模型原始输出查看 DecisionMade 事件的 payload 是否包含完整输出保存完整 messages 快照和 model_output同一 episode 出现重复事件重试机制导致重复写入缺少幂等检查事件表是否设置 event_id 唯一约束在数据库层面对 event_id 加唯一索引事件表越来越大查询变慢没有按 episode_id 和 trace_id 建索引查看慢查询日志和执行计划为常用过滤字段建立联合索引并考虑按时间分区反思建议是空话没有操作性反思 prompt 缺少上下文细节检查反思模型是否能看到完整的中间事件在 prompt 中展示工具输入输出和每一步决策策略更新后 Agent 反而变差策略文件被直接覆盖没有版本隔离检查是否保留旧版本策略每次更新生成新版本文件事件表记录 policy_version数据中包含用户敏感信息事件 payload 直接保存了完整对话检查存储内容是否有非必要字段最小化保存对敏感字段脱敏或以内容哈希代替程序重启后 SQLite 表被锁多进程并发写入 SQLite查看锁等待错误日志单机原型避免多进程写生产环境换 PostgreSQL9. 最佳实践与工程建议第一把事件表当作单一事实来源。不要在 Agent 里再维护一套“运行记录”数据库也不要把事件表只当作辅助日志。事件流是改进系统的数据底座它的质量直接决定改进效果。第二保存原始信息不要只保存清洗后的结果。模型输出、工具返回原文、完整消息列表这些都要尽量保留。你可以在分析时再做清洗但原始信息一旦丢弃就再也找不回来。第三为每一个任务会话建立清晰的 ID 体系。episode_id表示一次任务trace_id表示一条调用链parent_id表示父子关系。当 Agent 支持子任务或并行分支时这三个字段缺一不可。第四给策略、模型配置和事件 schema 加上版本号。策略版本、模型版本、事件 schema 版本是改进实验的可控变量。只有把版本信息保存到事件中你才能回答“这次改进到底改了什么”这个问题。第五在事件记录层设置安全边界。不要把敏感对话原样入库至少要脱敏给数据库账号分配最小权限只允许 INSERT 和 SELECT配置数据保留策略超期事件归档或清理。改进系统的数据越完整越要重视权限控制。第六把“离线回放评测”纳入 Agent 发布流程。任何策略改动先用历史事件流做回放评测而不是直接在线上改完就跑。回放评测能发现很多在真实运行中才会暴露的问题成本却低得多。第七从简单方案起步。不要一开始就上 Kafka、事件溯源框架和分布式追踪系统。先用 SQLite 或 PostgreSQL 单表跑通整个闭环等事件量确实大了再逐步替换存储层。架构升级的收益要基于真实的量级而不是基于设想。10. 总结与下一步实践Self-improving Agent 这个方向不缺 idea缺的是可靠的经验基础设施。Event Sourcing 之所以和自改进 Agent 高度契合是因为它把“发生了什么”变成了一等公民。只有事件流完整回放、归因、对比、审计、反思才有依据没有事件流“自我改进”就只是一句口号。如果你准备动手我的建议是从一个最小闭环开始先给你的 Agent 加一个事件记录器把所有关键决策点落库再写一个回放脚本把每次任务的过程完整还原最后接一个反思模型让 Agent 从事件流中提炼策略改进建议。这个最小闭环跑通后再考虑更复杂的元进化让 Agent 自己决定该反思什么、该保留哪些经验、该在哪里调整改进策略。事件溯源是那种“前期看起来多写了很多代码后期却帮你省下无数排查时间”的架构。对于以不确定性为核心的 Agent 系统它不是一个可选项而是一个基础设施级别的选择。