这次我们来看一个偏架构设计的主题Self-Improving Agents 与 Event-Sourced 模式如何结合。如果你近期关注 Agent 项目大概率会频繁看到self-improving agents、experience-driven agents、self-to-meta evolution这类概念。它们指向同一个方向智能体不应该只在一次对话里完成回答而是要在多轮任务之后把经验沉淀下来反过来优化自己下一次的规划、工具调用和提示词策略。这类系统的难点通常不在模型本身而在于“经验”如何被记录、重放、验证和发布。事件溯源Event Sourcing提供了非常契合的工程范式把所有行为当作不可变事件追加到日志状态通过事件重放推导。放在 Agent 自我改进场景里这等于给 Agent 装了一套完整的“成长记录仪”。这篇文章会从概念、架构、事件模型、Python 最小实现、批量经验回放、资源占用、排错清单和最佳实践几个角度展开最后给出一套可以落地参考的实现思路。1. 核心能力速览能力项说明项目类型架构设计 / 工程模式不是单一开源模型核心思想把 Agent 的感知、决策、执行、反馈全部记录为不可变事件用事件流驱动自我改进核心能力经验记录、行为审计、失败重放、策略版本化、技能沉淀、批量经验回放适用读者正在做 Agent 系统、RAG 应用、自动化工作流、AI 平台工程的开发者前置知识Python、事件驱动架构、LLM API 调用、SQLite/PostgreSQL 基础显存 / 硬件要求取决于底层 LLM 的部署方式如果只做架构验证仅普通开发机即可启动方式本地代码运行不依赖一键包是否支持 API可以推荐把事件流封装为内部 API 或消息队列接入是否支持批量任务支持事件存储天然适合批量重放与批量评估适用场景Agent 自我改进、多步骤决策日志、AI 工作流审计、策略回滚、Prompt 版本管理这里先说明一点这不是某个具体开源仓库的使用教程而是一套可复用的架构方案。本文会给出足够具体的代码设计和验证流程你可以直接把它迁移到自己的 Agent 项目里。2. 为什么 Agent 自我改进需要事件溯源2.1 常见实现方式有哪些问题现在很多 Agent 项目的“自我改进”是这样做的把用户反馈、任务结果写进数据库用一个新的 LLM 调用去总结经验把总结出来的经验塞回系统提示词下次任务直接使用更新后的提示词。这种方式跑通很快但会有几个隐患状态覆盖无法回滚。如果某次总结出来的经验是错误的新提示词会直接覆盖旧策略想回到上一个稳定版本很麻烦。缺少完整上下文。只保存“最终结论”中间为什么调用这个工具、什么参数、报了什么错全部丢失。难以审计。策略什么时候变的、哪条经验导致行为变化查不到。改进过程不可复现。同一个错误在新版本上修复了但没办法确切回答“是哪个事件触发的改进”。2.2 事件溯源解决的四个问题事件溯源把“状态改变”变成“事件追加”对 Agent 自我改进的价值非常直接可审计。每一次模型调用、工具返回、用户反馈、策略更新都是独立事件保留完整时间线。可重放。出问题时可以从事件流重新构建当时状态定位是输入问题、工具问题还是策略问题。可回滚。策略本身也是事件产生的投影结果回滚就是选择某个历史快照重新发布。可验证。新增一条经验后可以拿历史事件作为测试集重新跑评估而不是拍脑袋上线。一句话总结没有事件溯源的 Agent 改进是“黑盒成长”有了事件溯源就是“日志驱动的版本迭代”。3. 事件溯源的核心概念与事件模型3.1 领域事件、聚合状态与投影事件溯源里有三个核心概念概念在 Agent 场景中的含义Domain EventAgent 已经发生的不可变事实例如ActionExecutedAggregate一个 Agent 实例或一个任务会话的边界通过事件重放得到当前状态Projection / Read Model从事件流推导出的查询视图例如“当前提示词版本”“技能库列表”Agent 自我改进系统里典型的聚合根是Agent和TaskSession。每个聚合维护自己的事件流。投影层的职责是持续监听事件更新可查询的状态比如当前系统提示词版本技能注册表最近 N 轮失败记录策略评估指标。3.2 事件类型设计设计事件时不要保存“改进后的提示词”这种大而全的状态而要保存“发生了什么”。推荐从这些事件入手事件名称关键字段触发场景AgentInitializedagent_id, config_version创建 Agent 实例TaskReceivedtask_id, input, meta新任务进入PlanCreatedplan_id, steps, reasoningAgent 生成计划ToolCalledtool_name, arguments调用外部工具ToolResultReceivedtool_name, result_summary, error工具返回结果ModelInferencemodel, prompt_version, outputLLM 推理记录FeedbackReceivedsource, score, comment人工反馈 / 自动评估StrategyUpdatedstrategy_id, prompt_version, reason策略版本更新SkillCreatedskill_id, code_ref, description沉淀新技能事件字段统一带上event_id、agent_id、occurred_at、sequence后续排序、去重和审计都会用到。3.3 事件版本与兼容性事件会演进。比如ToolResultReceived一开始只有result_summary后来想加token_usage。直接改事件结构会让旧事件无法重放。建议每个事件带event_version事件类实现upcast方法旧版本事件重放时自动升级不要修改已发布的事件只追加新版本事件。dataclass class ToolResultReceived: event_version: int 1 def upcast(self): # 将 v1 事件升级为 v2 self.event_version 2 self.token_usage {prompt_tokens: 0, completion_tokens: 0} return self4. 自我改进 Agent 参考架构4.1 模块划分整个系统可以分成六个模块Agent Runtime负责任务执行调用 LLM 和工具Event Collector拦截 Agent 运行时产生的所有关键行为写入事件存储Event Store事件持久化层推荐 PostgreSQL 或 SQLite单机验证用 SQLite 足够Experience Miner从事件流中抽取高价值经验例如失败模式、工具超时、重复修正Strategy Builder根据经验生成新的系统提示词或技能定义Evaluator用历史事件和留出集评估新策略评估通过后发布。4.2 自我改进闭环任务输入 - Agent 感知 - 规划 - 工具调用 - 结果反馈 - 事件写入 Event Store - 周期性经验挖掘 - 策略更新事件 - 评估验证 - 发布新策略核心不是“每次任务都调模型总结经验”而是按事件窗口批量处理。例如每积累 200 条事件或者任务失败率超过阈值时触发一次改进循环。改进循环的关键步骤从事件存储读取最近一个时间窗口的事件用 LLM 或规则分析失败原因生成候选策略用历史事件重放评估评估通过后写入StrategyUpdated事件投影层更新当前策略版本。4.3 快照与投影事件流无限增长会有重放性能问题。常规做法是定期压缩快照Snapshot。快照不是替代事件而是缓存。例如每 1000 条事件生成一个聚合状态快照下次重放只从最近快照开始而不是从第一条事件开始。快照表设计CREATE TABLE agent_snapshot ( agent_id TEXT, snapshot_version INTEGER, state_json TEXT, created_at TIMESTAMP, PRIMARY KEY (agent_id, snapshot_version) );投影表需要实时更新CREATE TABLE agent_projection ( agent_id TEXT PRIMARY KEY, current_strategy_version INTEGER, current_system_prompt TEXT, skill_count INTEGER, updated_at TIMESTAMP );5. 最小实现Python SQLite 事件存储这一节给出可以直接跑起来的最小闭环。它不依赖框架方便你理解事件溯源在 Agent 场景下的工作方式。5.1 事件存储层使用 SQLite 作为事件存储包含四个核心方法append_event、read_events、snapshot、restore。import json import sqlite3 from datetime import datetime, timezone from typing import Any, Dict, List class SQLiteEventStore: def __init__(self, db_path: str): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS events ( event_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, sequence INTEGER NOT NULL, event_type TEXT NOT NULL, event_data TEXT NOT NULL, occurred_at TEXT NOT NULL, UNIQUE(agent_id, sequence) ) ) self.conn.commit() def append_event( self, agent_id: str, event_type: str, event_data: Dict[str, Any] ) - Dict[str, Any]: sequence self._next_sequence(agent_id) event { event_id: f{agent_id}-{sequence}, agent_id: agent_id, sequence: sequence, event_type: event_type, event_data: event_data, occurred_at: datetime.now(timezone.utc).isoformat() } self.conn.execute( INSERT INTO events (event_id, agent_id, sequence, event_type, event_data, occurred_at) VALUES (?, ?, ?, ?, ?, ?) , ( event[event_id], event[agent_id], event[sequence], event[event_type], json.dumps(event_data, ensure_asciiFalse), event[occurred_at] ) ) self.conn.commit() return event def _next_sequence(self, agent_id: str) - int: row self.conn.execute( SELECT MAX(sequence) FROM events WHERE agent_id ?, (agent_id,) ).fetchone() return 1 if row[0] is None else row[0] 1 def read_events(self, agent_id: str, after_sequence: int 0) - List[Dict[str, Any]]: rows self.conn.execute( SELECT event_id, agent_id, sequence, event_type, event_data, occurred_at FROM events WHERE agent_id ? AND sequence ? ORDER BY sequence ASC , (agent_id, after_sequence) ).fetchall() events [] for row in rows: events.append({ event_id: row[0], agent_id: row[1], sequence: row[2], event_type: row[3], event_data: json.loads(row[4]), occurred_at: row[5] }) return events这个实现满足单机场景。生产环境建议换成 PostgreSQL并增加事务写入和订阅分发。5.2 改进循环实现改进循环通过事件聚合触发。下面示例只提炼两个关键信号任务失败反馈和工具调用异常。from typing import List, Dict class AgentImprovementLoop: def __init__(self, event_store: SQLiteEventStore): self.event_store event_store def collect_history(self, agent_id: str) - List[Dict]: return self.event_store.read_events(agent_id) def extract_experiences(self, events: List[Dict]) - List[Dict]: experiences [] for event in events: if event[event_type] FeedbackReceived: data event[event_data] if data.get(score, 0) 0.6: experiences.append({ task_id: data.get(task_id), failure_reason: data.get(comment), event_id: event[event_id] }) if event[event_type] ToolResultReceived: data event[event_data] if data.get(error): experiences.append({ task_id: data.get(task_id), tool: data.get(tool_name), error: data.get(error), event_id: event[event_id] }) return experiences def build_candidate_strategy(self, experiences: List[Dict]) - str: if not experiences: return # 这里可以调用 LLM 生成候选提示词 # 简单实现只做去重和规则拼接 lines sorted({exp[failure_reason] for exp in experiences if exp.get(failure_reason)}) return \n.join(lines)实际项目中build_candidate_strategy应该交给 LLM 处理同时保留生成过程中的prompt_version和model信息方便后续审计。5.3 重放与验证重放目标是重建某个 Agent 在某时刻的策略状态。def replay_strategy(event_store: SQLiteEventStore, agent_id: str, target_sequence: int) - str: events event_store.read_events(agent_id) strategy_version 0 strategy_content for event in events: if event[sequence] target_sequence: break if event[event_type] StrategyUpdated: strategy_version event[event_data].get(strategy_version, strategy_version 1) strategy_content event[event_data].get(content, strategy_content) return fv{strategy_version}: {strategy_content}验证时可以拿历史任务输入重新执行一遍新策略比较成功率、工具调用次数和最终评分确认策略确实比旧版本更好。6. 接口 API 与批量经验回放6.1 事件流 API事件溯源很适合暴露为内部 API 服务方便多个 Agent 实例共用事件存储。推荐 API 设计接口方法说明/api/eventsPOST追加事件/api/agents/{agent_id}/eventsGET读取某 Agent 事件流/api/agents/{agent_id}/snapshotPOST生成快照/api/agents/{agent_id}/strategyGET读取当前策略投影/api/improvement/runPOST触发一次改进循环/api/improvement/evaluatePOST使用历史事件评估候选策略Python 侧可以使用 FastAPI 快速封装from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class EventPayload(BaseModel): agent_id: str event_type: str event_data: dict app.post(/api/events) def append_event(payload: EventPayload): event store.append_event( agent_idpayload.agent_id, event_typepayload.event_type, event_datapayload.event_data ) return event app.get(/api/agents/{agent_id}/events) def read_events(agent_id: str, after_sequence: int 0): events store.read_events(agent_id, after_sequence) if not events: raise HTTPException(status_code404, detailno events) return events6.2 批量回放设计批量经验回放是 Agent 自我改进的关键。它指的是用一个已经验证过的策略重新跑一批历史任务确认策略改动没有引入回归。批量回放的目录结构建议experiments/ 2025-06-01_v1_baseline/ config.json output/ task_001.jsonl task_002.jsonl 2025-06-02_v2_strategy_update/ config.json output/每次回放用到一个配置{ agent_id: agent-demo-001, strategy_version: 2, replay_task_ids: [task_001, task_002, task_003], evaluation: { success_threshold: 0.8, max_tool_calls: 10 } }批量回放流程从事件存储读取最近成功的 20 条任务记录依次喂给新策略 Agent每完成一条任务写入TaskCompleted事件全部结束后汇总成功率成功率高于阈值写入StrategyPublished事件低于阈值保留候选策略并通知人工介入。7. 资源占用与性能观察事件溯源本身不消耗 GPU。真正消耗资源的是底层 LLM 推理、事件存储写入和批量评估。可以从三个维度观察性能7.1 事件写入性能SQLite 单机每秒能追加几千条事件但对 Agent 系统来说事件数量远没有大到这个量级。需要关注的是每个事件是否有合理的sequence并发控制。生产环境换成 PostgreSQL 后要避免多个 Agent 实例同时向同一事件流写入时的唯一索引冲突。观察方式sqlite3 agent_events.db SELECT COUNT(*) FROM events;7.2 LLM 推理开销每次改进循环会触发多轮 LLM 调用例如经验提炼候选策略生成评估打分批量回放执行。这部分开销会直接反映在 API 延迟和账单上。如果使用本地模型则需要关注显存占用。稳妥做法是先小批量验证用 5 到 10 条任务作为评估集确定效果稳定后再扩大。7.3 事件流膨胀事件只追加不删除长期运行后存储会膨胀。快照机制能降低重放成本但无法阻止磁盘占用增长。建议设置事件保留策略原始事件保存 90 天90 天到 180 天的事件压缩为摘要事件超过 180 天只保留聚合快照和策略版本。8. 常见问题与排查方法问题现象可能原因排查方式解决方案事件重放结果和线上不一致事件顺序混乱或旧事件被修改检查sequence和事件表是否有UPDATE操作事件表只追加禁止更新用数据库触发器阻止修改策略越改越差评估集选择不当只用了出现过的问题样本检查评估集是否包含未见过的任务类型划分留出集新策略必须在留出集上评估提示词越来越长经验不断追加没有去重和压缩查看StrategyUpdated事件统计内容长度对经验做聚类和摘要超过阈值时触发压缩批量回放卡住某个工具调用超时查看事件流中ToolCalled和ToolResultReceived之间的时间间隔为工具调用设置超时和重试上限事件记录过多但改进不明显仅记录事件没有自动挖掘经验检查改进循环是否定期触发增加任务失败率阈值触发策略更新API 并发写入冲突多个 Agent 实例写入同一事件流查看数据库唯一索引报错使用分布式 ID 或按 agent_id 分片快照恢复失败快照 JSON 与事件版本不匹配检查state_json字段快照中带上 schema_version恢复时执行迁移Agent 自我改进出现循环依赖策略改进后又影响下一次策略更新查看事件时间线确认策略版本是否跳变增加策略发布冷却时间人工审核关键策略更新9. 最佳实践与安全边界9.1 工程化建议第一次先小参数测试。先用 10 条任务、单 Agent、一个事件存储文件跑通闭环再考虑分片和消息队列。保留一套最小可运行配置。把事件模型、改进循环、评估逻辑做成独立模块后续替换 LLM 或工具时不影响事件流。模型文件、输入素材、输出结果分目录管理。事件存储目录、任务输入目录、评估结果目录三者分开避免误删。批量任务要加日志和失败重试。每条任务的事件写入失败不能影响其他任务建议写入死信队列或单独事件表。接口服务要限制访问范围。事件流包含任务输入和 Agent 内部推理内容不要把事件 API 直接暴露到公网。策略发布前要做人工回退预案。即使评估通过新策略上线后仍要保留一键回滚到旧策略版本入口。9.2 涉及 LLM 内容的安全与合规边界事件流里可能包含用户输入、工具返回的文档内容、内部提示词。存储和回放时要做敏感信息脱敏。使用真实用户数据做自我改进前必须确认数据授权范围。不能拿未授权的对话内容训练或沉淀策略。涉及人脸、声音、版权素材、企业内部文档时要特别确认来源合法性和使用边界。Agent 自动生成的策略更新不能无限自动发布。会改变行为边界的策略变更需要人工审核。避免自我强化偏见。如果评估集全部来自 Agent 自身生成内容新策略可能只是在拟合过往错误。要定期混入人工标注数据和外部测试集。10. 总结与下一步这个设计思路最有价值的地方不是某个具体代码文件而是把 Agent 的“成长过程”变成了一串可以查询、重放、评估和回滚的事件。对正在搭建 AI Agent 平台的团队来说Event-Sourced 模式可以直接复用它让自我改进从“玄学调 Prompt”变成“可审计、可验证、可回滚的工程流程”。建议你先从最小闭环开始搭一个 SQLite 事件表让 Agent 每执行一次任务就追加几个关键事件再写一个简单的改进循环。不要一上来就做分布式事件总线。事件模型稳定之后再引入快照、消息队列、批量评估和策略发布机制。最容易踩的坑是事件记录了但改进循环没有真正触发或者新策略没有经过留出集评估。前者会让事件存储变成“废日志”后者会让系统越改越偏。先验证这两点再考虑扩大任务规模。方向可以继续扩展的方向包括从单 Agent 到多 Agent 协作的经验共享、从 Prompt 级策略升级到工具级/技能级自我进化、从人工反馈评估升级为自动评估加人工抽检。如果你正在设计 Agent 的经验沉淀系统建议收藏这篇作为起步参考。