1. 从“记忆型 AI”到“开工型 Agent”的认知转变1.1 为什么“能记住”不等于“能干活”过去大半年我陆陆续续试过不少所谓的个人 Agent 项目。一开始我也被“记忆”这个卖点吸引——能记住我的偏好、能回忆上次聊到哪、能在我提到某个项目时自动关联历史上下文。听起来很美好但真正用起来问题很快就暴露了。最典型的场景是这样的我让 Agent 帮我整理一份会议纪要它确实记得我上次说过“纪要要分行动项和决策项”也记得我偏好用表格呈现。但当我真正把一段两小时的录音转写文本丢给它时它给出的东西依然是一堆散乱的摘要行动项没有负责人决策项没有时间戳表格格式也对不齐。换句话说它“记得”我的偏好但它“不会”执行我的流程。这就是“记忆型 AI”和“开工型 Agent”之间最本质的差别。记忆型 AI 的核心能力是检索和关联它像一个博闻强识但缺乏动手能力的顾问而开工型 Agent 的核心能力是执行和交付它需要把模糊的意图翻译成确定的步骤把确定的步骤拆解成可验证的动作再把动作的结果组装成可交付的产物。我后来复盘发现问题的根源在于大多数个人 Agent 项目把“记忆”当成了终点而不是起点。记忆本身不产生价值记忆只有被编排进一个可重复执行的工作流里才真正有意义。这就引出了我们这次改造的核心思路——把 Agent 从“知道什么”推向“能做什么”。1.2 我们到底想让 Agent 干什么在动手之前我先花了一个下午把需求写清楚。这个过程很痛苦因为“让 Agent 帮我干活”这句话太模糊了模糊的需求必然导致模糊的实现。我最后把目标收敛成三个具体的场景第一个场景是会议到行动项的转化。输入是一段会议转写文本输出是一份结构化的行动项清单每个行动项包含负责人、截止时间、依赖项和验收标准。这个场景的关键在于Agent 不能只是“总结”它必须“提取”并“补全”——很多会议里负责人是隐含的截止时间是模糊的Agent 需要根据上下文推断并标注置信度。第二个场景是技术调研到决策备忘录的转化。输入是一个技术问题输出是一份包含候选方案、对比维度、推荐结论和风险提示的备忘录。这个场景的关键在于Agent 不能只是“罗列”它必须“权衡”——它需要知道我们团队的约束条件比如部署成本、维护复杂度、学习曲线然后在这些约束下给出推荐。第三个场景是零散笔记到项目计划的转化。输入是一堆碎片化的想法和待办输出是一份有优先级、有里程碑、有依赖关系的项目计划。这个场景的关键在于Agent 不能只是“分类”它必须“排序”——它需要理解任务之间的依赖关系识别关键路径并给出合理的排期建议。这三个场景有一个共同点它们都不是“问答”而是“加工”。Agent 需要接收一种形态的输入经过一系列确定的处理步骤输出另一种形态的产物。这就决定了我们的技术选型不能围绕“对话”来设计而必须围绕“流水线”来设计。1.3 为什么选择“事件模式”作为核心抽象在架构设计阶段我们面临一个关键选择用什么作为 Agent 的核心抽象是“对话轮次”还是“任务”还是“事件”对话轮次的抽象最自然但问题也最明显——它假设用户和 Agent 之间的交互是线性的、同步的。但真实的干活场景不是这样的。我可能在开会时丢给 Agent 一段录音然后去忙别的事半小时后回来检查结果。这中间 Agent 可能需要调用外部工具、可能需要等待某个依赖、可能需要向我确认某个模糊点。这些都不是“对话轮次”能很好表达的。任务的抽象更接近干活的本质但它太粗粒度了。一个任务从创建到完成中间会经历很多状态变化如果只用一个“任务”对象来承载状态管理会变得非常复杂。最后我们选择了事件模式。具体来说Agent 的整个生命周期被建模成一系列事件的追加。用户丢进来一段录音这是一个input.received事件Agent 开始处理这是一个processing.started事件Agent 调用了一个外部工具这是一个tool.invoked事件Agent 需要用户确认这是一个confirmation.requested事件用户确认了这是一个confirmation.resolved事件Agent 输出了最终产物这是一个output.produced事件。这个设计的妙处在于它把 Agent 的“记忆”和 Agent 的“执行”统一到了同一个抽象下。记忆不再是独立于执行之外的一个模块而是执行过程中自然产生的事件流。Agent 要回忆什么只需要回放相关的事件Agent 要执行什么只需要追加新的事件。这种统一性大大简化了系统的复杂度。更重要的是事件模式天然支持追加写入。这意味着 Agent 的整个执行历史是不可变的、可审计的、可回放的。如果某次执行出了问题我可以精确地回放到出问题的那一步检查当时的事件上下文而不是面对一个黑盒。2. 核心细节解析事件模式、访谈协议与追加写入2.1 事件模式的设计要点与常见陷阱事件模式听起来简单但真正落地时有很多细节需要斟酌。我踩过的第一个坑是事件粒度过细。一开始我把每个微小的状态变化都定义成一个事件结果事件流变得极其冗长回放一次要处理上千个事件性能很差。后来我把事件粒度调整到“有语义意义的业务动作”这一层比如“工具调用完成”是一个事件但“工具调用的参数序列化”就不是一个独立事件而是前一个事件的属性。第二个坑是事件类型的命名。我见过很多项目用event_type_1、event_type_2这种命名过两周自己都忘了哪个是哪个。我的建议是采用领域.动作的命名规范比如meeting.transcribed、action_item.extracted、action_item.confirmed。这样一看就知道事件属于哪个领域、发生了什么动作。第三个坑是事件的版本兼容。Agent 的 schema 一定会随着需求变化而演进如果事件结构变了旧的事件怎么处理我们的做法是在每个事件里加一个schema_version字段回放时根据版本号做兼容性转换。这个成本不高但能避免很多麻烦。下面是我们实际使用的事件 schema 的核心字段{ event_id: evt_20250115_001, event_type: action_item.extracted, schema_version: 1.2, timestamp: 2025-01-15T10:23:45Z, session_id: sess_meeting_20250115, payload: { item_id: ai_001, description: 完成竞品分析报告初稿, owner: 张三, owner_confidence: 0.85, due_date: 2025-01-20, due_confidence: 0.6, dependencies: [ai_000], acceptance_criteria: 包含至少3个竞品的功能对比表 }, source_event_ids: [evt_20250115_000], produced_by: extractor_v2 }这个 schema 里有几个设计决策值得说明。owner_confidence和due_confidence是我强烈建议加的字段。会议里很多信息是模糊的与其让 Agent 假装确定不如让它诚实地标注置信度。置信度低于阈值的Agent 会主动发起确认请求而不是默默猜测。source_event_ids是另一个关键字段。它记录了当前事件是从哪些事件派生出来的这样整个事件流就形成了一张有向无环图而不是一条简单的线。当需要追溯某个行动项的来源时可以沿着source_event_ids一路回溯到原始的会议转写文本。produced_by字段记录了是哪个处理器产生了这个事件。当 Agent 的处理器升级后我可以对比新旧处理器的输出差异评估升级效果。2.2 访谈协议让 Agent 学会“问对问题”事件模式解决了“怎么记录”的问题但没解决“怎么获取信息”的问题。在会议到行动项的转化场景里很多关键信息在原始文本里是缺失的。比如会议里有人说“这个事我来跟进”但没说截止时间有人说“等那边确认了再说”但没说“那边”是谁。传统的做法是让 Agent 直接输出一个不完整的行动项然后人工补全。但这违背了我们“开工型 Agent”的初衷——Agent 应该主动补全而不是被动等待。我们借鉴了结构化访谈的思路设计了一套访谈协议。核心思想是Agent 在提取信息时如果发现关键字段缺失或置信度低于阈值就生成一个confirmation.requested事件把问题结构化地抛给用户。用户回答后生成confirmation.resolved事件Agent 继续处理。访谈协议的关键在于问题的生成策略。我们定义了三种问题类型补全型问题某个必填字段缺失直接问。比如“行动项‘完成竞品分析报告’的截止时间是什么”澄清型问题某个字段有多个候选值让用户选择。比如“会议里提到了‘张三’和‘张总’行动项的负责人是哪一个”确认型问题某个字段的置信度低于阈值让用户确认。比如“我推断截止时间是1月20日置信度60%是否正确”这三种问题的优先级不同。补全型问题必须问否则行动项不完整澄清型问题尽量问但如果有默认值可以先用默认值确认型问题可以批量问减少对用户的打扰。下面是我们实际使用的问题 schema{ question_id: q_001, question_type: completion, target_event_id: evt_20250115_001, target_field: due_date, question_text: 行动项‘完成竞品分析报告初稿’的截止时间是什么, candidate_answers: [], default_answer: null, priority: high, batch_id: batch_001 }batch_id是为了支持批量提问。如果一次处理产生了多个确认型问题Agent 会把它们打包成一个批次一次性呈现给用户而不是逐个弹窗。这个细节看起来小但对用户体验影响很大。2.3 追加写入为什么不可变日志是 Agent 的基石追加写入是我们整个架构里最底层的一个决策也是我花了最多时间说服团队的一个决策。很多人的第一反应是为什么不能直接修改事件如果发现某个行动项的负责人写错了直接改掉不就行了我的回答是修改会丢失信息而信息是 Agent 最宝贵的资产。假设 Agent 提取了一个行动项负责人是“张三”后来用户纠正说是“李四”。如果直接修改那么“Agent 曾经认为是张三”这个信息就丢失了。这个信息有什么用它可以用来评估 Agent 的提取准确率可以用来分析哪些类型的表述容易导致误判可以用来训练更好的提取器。追加写入的做法是不修改原事件而是追加一个action_item.corrected事件记录修正的内容和原因。回放时Agent 会先看到原始事件再看到修正事件最终状态是修正后的状态。但整个修正历史被完整保留。这个设计还有一个额外的好处支持时间旅行。我可以回放到任何一个时间点查看当时 Agent 的状态。这在调试时极其有用。比如我想知道为什么 Agent 在某个时间点做出了一个奇怪的决策我可以回放到那个时间点之前逐步执行观察每一步的事件上下文。当然追加写入也有代价。事件流会越来越长存储成本会上升。我们的应对策略是分层存储最近的事件放在内存里支持快速回放较旧的事件放在磁盘上按需加载超过一定时间的事件归档到冷存储只在审计时访问。还有一个代价是回放性能。如果事件流有十万条从头回放一次要很久。我们的做法是定期生成快照。快照记录了某个时间点的完整状态回放时只需要从最近的快照开始而不是从头开始。快照的生成频率可以根据事件流的增长速度动态调整。3. 实操过程从零搭建一个可开工的 Agent3.1 环境准备与依赖选型我们的技术栈选择遵循一个原则能用成熟库就不自己造轮子但核心抽象必须自己掌控。运行时我们选了 Python 3.11原因是生态成熟而且 3.11 的性能比 3.10 有明显提升。事件存储我们用了 SQLite 作为起步方案原因是零配置、单文件、支持事务。很多人一上来就上 PostgreSQL 或 MongoDB但对于个人 Agent 这个场景SQLite 完全够用而且部署成本低得多。等事件流真的增长到 SQLite 扛不住了再迁移也不迟。事件 schema 的校验我们用了 Pydantic。Pydantic 的好处是它把 schema 定义和校验逻辑合二为一而且错误信息很清晰。下面是我们的事件基类from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List, Dict, Any class BaseEvent(BaseModel): event_id: str event_type: str schema_version: str 1.0 timestamp: datetime Field(default_factorydatetime.utcnow) session_id: str payload: Dict[str, Any] source_event_ids: List[str] [] produced_by: str class ActionItemExtractedEvent(BaseEvent): event_type: str action_item.extracted class ActionItemPayload(BaseModel): item_id: str description: str owner: Optional[str] None owner_confidence: float 0.0 due_date: Optional[str] None due_confidence: float 0.0 dependencies: List[str] [] acceptance_criteria: Optional[str] None payload: ActionItemPayload这个基类定义了所有事件的公共字段子类只需要定义自己的 payload 结构。Pydantic 会自动处理校验和序列化。工具调用我们用了httpx而不是requests原因是httpx原生支持异步而 Agent 的执行流程里有大量 IO 等待异步能显著提升吞吐。当然如果你不熟悉异步编程用requests也完全可以只是性能会差一些。3.2 事件存储层的实现细节事件存储层是整个系统的地基它的设计直接决定了上层能做什么。我们的存储层提供了四个核心接口append(event)追加一个事件返回事件 IDget_events(session_id, sinceNone, untilNone)获取某个会话的事件流get_snapshot(session_id, at_time)获取某个时间点的快照replay(session_id, from_event_id)从某个事件开始回放append的实现很简单就是往 SQLite 里插一条记录。但有一个细节需要注意事件 ID 的生成必须是单调递增的。我们用了evt_{session_id}_{sequence}的格式其中sequence是会话内自增的。这样既能保证唯一性又能保证顺序性。get_events的实现需要注意分页。如果事件流很长一次性加载所有事件会爆内存。我们用了游标分页每次最多加载 1000 条事件。get_snapshot的实现是我们花时间最多的地方。快照的本质是把事件流“折叠”成一个状态对象。折叠的逻辑取决于事件类型。比如action_item.extracted事件会把行动项加入状态action_item.corrected事件会修改状态中的行动项action_item.deleted事件会从状态中移除行动项。def fold_events(events: List[BaseEvent]) - Dict[str, Any]: state {action_items: {}, confirmations: {}} for event in events: if event.event_type action_item.extracted: state[action_items][event.payload.item_id] event.payload.dict() elif event.event_type action_item.corrected: item_id event.payload.item_id if item_id in state[action_items]: state[action_items][item_id].update(event.payload.changes) elif event.event_type action_item.deleted: state[action_items].pop(event.payload.item_id, None) elif event.event_type confirmation.requested: state[confirmations][event.payload.question_id] event.payload.dict() elif event.event_type confirmation.resolved: state[confirmations].pop(event.payload.question_id, None) return state这个折叠函数是纯函数没有副作用所以可以安全地缓存和复用。我们会在每次追加事件后异步更新快照这样回放时只需要从最近的快照开始而不是从头开始。3.3 访谈协议的实现与问题生成策略访谈协议的实现分为三个部分问题生成、问题呈现、答案处理。问题生成是在提取器输出行动项后触发的。提取器会为每个行动项计算字段置信度如果某个必填字段的置信度低于阈值我们设的是 0.7就生成一个补全型问题如果某个字段有多个候选值就生成一个澄清型问题如果某个字段的置信度在 0.7 到 0.9 之间就生成一个确认型问题。def generate_questions(action_item: dict, threshold: float 0.7) - List[dict]: questions [] if not action_item.get(owner): questions.append({ question_type: completion, target_field: owner, question_text: f行动项‘{action_item[description]}’的负责人是谁, priority: high }) elif action_item.get(owner_confidence, 0) threshold: questions.append({ question_type: confirmation, target_field: owner, question_text: f我推断负责人是{action_item[owner]}是否正确, default_answer: action_item[owner], priority: medium }) if not action_item.get(due_date): questions.append({ question_type: completion, target_field: due_date, question_text: f行动项‘{action_item[description]}’的截止时间是什么, priority: high }) elif action_item.get(due_confidence, 0) threshold: questions.append({ question_type: confirmation, target_field: due_date, question_text: f我推断截止时间是{action_item[due_date]}是否正确, default_answer: action_item[due_date], priority: medium }) return questions问题呈现我们做了一个简单的 CLI 界面把同一批次的问题一次性列出用户可以用编号回答也可以直接回车接受默认值。这个界面很粗糙但足够用。关键是它把“打扰用户”的次数降到了最低。答案处理是把用户的回答转换成confirmation.resolved事件然后触发提取器重新处理受影响的行动项。这里有一个细节重新处理时提取器会跳过已经有确认答案的字段避免重复提问。3.4 从会议转写到行动项清单的完整流程让我用一个真实的例子把整个流程串起来。假设我有一段会议转写文本内容是关于一个产品迭代计划的讨论。第一步是输入接收。我把转写文本丢给 Agent生成一个input.received事件payload 里包含文本内容和元数据会议时间、参与人等。第二步是预处理。Agent 对文本做分句、去噪、说话人分离。这一步的输出是一个text.preprocessed事件payload 里包含结构化的对话片段。第三步是行动项提取。Agent 用提取器从对话片段里识别行动项。提取器是一个基于规则和模型混合的组件——规则负责识别“我来做”“你负责”“下周完成”这类模式模型负责处理更复杂的表述。提取器的输出是一批action_item.extracted事件。第四步是问题生成。Agent 检查每个行动项的字段完整性生成confirmation.requested事件。这些事件被批量呈现给我。第五步是答案处理。我回答了问题Agent 生成confirmation.resolved事件并更新行动项。第六步是产物组装。Agent 把所有行动项组装成一份 Markdown 格式的清单生成output.produced事件。整个流程走下来从丢入文本到拿到清单大概需要 30 秒到 2 分钟取决于文本长度和问题数量。这个速度不算快但考虑到它省掉了我手动整理的一两个小时完全值得。4. 常见问题与排查技巧实录4.1 事件流异常增长的排查与治理上线两周后我发现事件流的增长速度远超预期。一个小时的会议产生了将近 500 个事件。按这个速度一个月下来事件表就要爆了。我排查后发现问题出在工具调用的日志事件上。每次 Agent 调用外部工具都会生成一个tool.invoked事件和一个tool.completed事件payload 里包含了完整的请求和响应。这些 payload 很大而且大部分是冗余的。治理方案是分级记录。对于成功的工具调用只记录摘要信息工具名、耗时、结果状态不记录完整的请求响应对于失败的工具调用才记录完整信息方便排查。这个改动把事件数量降到了原来的三分之一。另一个治理方案是事件合并。对于高频的、语义相近的事件比如连续多个action_item.extracted事件可以合并成一个action_items.extracted事件payload 里包含一个数组。这个改动进一步把事件数量降到了原来的五分之一。下面是我们的事件分级策略事件类型记录级别保留时长说明input.received完整永久原始输入必须保留text.preprocessed摘要30天只保留统计信息action_item.extracted完整永久核心业务事件tool.invoked (成功)摘要7天只保留工具名和耗时tool.invoked (失败)完整90天保留完整请求响应confirmation.requested完整永久核心业务事件output.produced完整永久最终产物这个策略的核心思想是业务事件永久保留技术事件定期清理。业务事件是 Agent 的记忆技术事件只是执行过程的痕迹。4.2 提取器误判的典型模式与修正方法提取器误判是另一个高频问题。我整理了最常见的几种误判模式第一种是指代消解错误。会议里有人说“这个事我来跟进”提取器需要判断“这个事”指的是哪个事。如果前文有多个候选提取器很容易选错。我们的修正方法是引入指代消解模块专门处理代词和指示词的消解并把消解结果作为行动项的一个字段记录下来方便追溯。第二种是时间表达歧义。“下周完成”到底是下周一还是下周五我们的修正方法是引入时间归一化模块把相对时间表达转换成绝对日期并标注置信度。置信度低的会触发确认问题。第三种是负责人误判。会议里有人说“张三你来负责这个”提取器可能把负责人识别成说话人而不是张三。我们的修正方法是引入角色识别模块区分“指派者”和“被指派者”并优先把被指派者作为负责人。第四种是行动项边界模糊。一段话里可能包含多个行动项也可能一个都不包含。我们的修正方法是引入边界检测模块用序列标注的方式识别行动项的起止位置。这些修正模块不是一次性的而是持续迭代的。我们建了一个误判案例库每次发现新的误判模式就把它加入案例库并针对性地调整提取器。这个案例库现在有 200 多条记录是我们最宝贵的资产之一。4.3 确认问题的用户体验优化确认问题是一把双刃剑。问得太少行动项不完整问得太多用户嫌烦。我们在这个平衡上踩了不少坑。第一个坑是问题太多。一开始我们对所有置信度低于 0.9 的字段都生成确认问题结果一次处理产生了 20 多个问题用户直接放弃了。后来我们把阈值降到 0.7并且只对必填字段生成问题问题数量降到了 5 个以内。第二个坑是问题太碎。每个问题单独呈现用户要反复切换上下文。后来我们引入了批量提问把同一批次的问题一次性列出用户可以用编号回答也可以直接回车接受默认值。第三个坑是问题太模糊。比如“截止时间是什么”这个问题用户可能不知道该怎么回答。后来我们引入了候选答案Agent 根据上下文生成几个候选用户只需要选择而不是从零输入。第四个坑是问题太频繁。每次处理都问用户很快就烦了。后来我们引入了学习机制Agent 会记住用户的历史回答对于相似的问题如果用户之前已经回答过就不再重复提问。下面是我们的问题优先级策略问题类型触发条件优先级是否批量补全型必填字段缺失高是澄清型字段有多个候选中是确认型置信度低于阈值低是学习型历史已回答过最低否这个策略的核心思想是能不问就不问能批量问就不单独问能选就不让用户输入。4.4 追加写入的性能优化实践追加写入的性能问题在事件流增长到一定规模后才会显现。我们的事件表在达到 10 万条记录时回放一次要 3 秒以上这在交互场景里是不可接受的。优化方案有三个层次第一个层次是索引优化。我们给session_id和timestamp建了联合索引这样按会话和时间范围查询事件时不需要全表扫描。这个改动把查询时间从 3 秒降到了 200 毫秒。第二个层次是快照优化。我们每 1000 个事件生成一个快照回放时从最近的快照开始而不是从头开始。这个改动把回放时间从 200 毫秒降到了 20 毫秒。第三个层次是缓存优化。我们把最近的事件流缓存在内存里回放时优先从缓存读取。这个改动把回放时间进一步降到了 5 毫秒以内。下面是我们的事件表结构CREATE TABLE events ( event_id TEXT PRIMARY KEY, event_type TEXT NOT NULL, schema_version TEXT NOT NULL, timestamp DATETIME NOT NULL, session_id TEXT NOT NULL, payload TEXT NOT NULL, source_event_ids TEXT, produced_by TEXT NOT NULL ); CREATE INDEX idx_session_timestamp ON events(session_id, timestamp); CREATE INDEX idx_event_type ON events(event_type); CREATE TABLE snapshots ( snapshot_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, at_event_id TEXT NOT NULL, state TEXT NOT NULL, created_at DATETIME NOT NULL ); CREATE INDEX idx_snapshot_session ON snapshots(session_id, at_event_id);这个表结构很简单但足够支撑我们的需求。关键是索引的设计——session_id和timestamp的联合索引覆盖了最常见的查询模式。4.5 常见问题速查表问题现象可能原因排查方法解决方案事件流增长过快技术事件记录过细统计各类型事件的数量占比分级记录技术事件只记摘要回放速度慢事件流太长无快照测量回放耗时与事件数量的关系定期生成快照加索引提取器误判多指代、时间、角色未处理抽样检查误判案例引入专门的消解和归一化模块确认问题太多阈值过低问题太碎统计每次处理的问题数量提高阈值批量提问引入学习机制行动项不完整必填字段缺失未补全检查行动项的字段完整率补全型问题必须问不能跳过用户回答后未更新答案处理逻辑有 bug检查 confirmation.resolved 事件确保答案事件触发重新处理快照与事件流不一致快照生成有竞态对比快照状态与回放状态快照生成加锁或改为异步生成事件 schema 不兼容版本演进未处理检查旧事件的 schema_version回放时做版本兼容转换这张表是我在实际运维中逐步积累的每一条都对应着一次真实的故障或优化。我建议你也建一张类似的表把每次踩坑的经验记录下来这比任何文档都有价值。5. 一些个人体会与后续扩展方向这套系统跑了一个多月我最深的体会是Agent 的价值不在于它有多聪明而在于它有多可靠。一个能稳定完成 80% 工作的 Agent比一个偶尔能完成 100% 工作但经常出错的 Agent 有用得多。可靠性来自于确定的流程、清晰的抽象和持续的迭代而不是来自于更大的模型或更复杂的提示词。事件模式、访谈协议和追加写入这三个设计决策本质上都是在为可靠性服务。事件模式让执行过程可追溯访谈协议让信息缺口可补全追加写入让历史状态可回放。这三个能力加起来才让 Agent 从“玩具”变成了“工具”。后续我打算在这几个方向继续扩展一是多会话协同让多个 Agent 可以共享事件流协同完成一个复杂任务二是事件流的可视化把事件流渲染成时间线方便人工审查和调试三是提取器的持续学习把用户的修正反馈自动转化为训练数据让提取器越用越准。如果你也在做类似的事情我的建议是先从一个小场景开始把事件模式跑通把访谈协议用起来把追加写入坚持住。不要一上来就追求大而全先把一个场景做可靠再扩展到其他场景。可靠性是积累出来的不是设计出来的。 从记忆型AI到开工型Agent：事件模式与追加写入实战
1. 从“记忆型 AI”到“开工型 Agent”的认知转变1.1 为什么“能记住”不等于“能干活”过去大半年我陆陆续续试过不少所谓的个人 Agent 项目。一开始我也被“记忆”这个卖点吸引——能记住我的偏好、能回忆上次聊到哪、能在我提到某个项目时自动关联历史上下文。听起来很美好但真正用起来问题很快就暴露了。最典型的场景是这样的我让 Agent 帮我整理一份会议纪要它确实记得我上次说过“纪要要分行动项和决策项”也记得我偏好用表格呈现。但当我真正把一段两小时的录音转写文本丢给它时它给出的东西依然是一堆散乱的摘要行动项没有负责人决策项没有时间戳表格格式也对不齐。换句话说它“记得”我的偏好但它“不会”执行我的流程。这就是“记忆型 AI”和“开工型 Agent”之间最本质的差别。记忆型 AI 的核心能力是检索和关联它像一个博闻强识但缺乏动手能力的顾问而开工型 Agent 的核心能力是执行和交付它需要把模糊的意图翻译成确定的步骤把确定的步骤拆解成可验证的动作再把动作的结果组装成可交付的产物。我后来复盘发现问题的根源在于大多数个人 Agent 项目把“记忆”当成了终点而不是起点。记忆本身不产生价值记忆只有被编排进一个可重复执行的工作流里才真正有意义。这就引出了我们这次改造的核心思路——把 Agent 从“知道什么”推向“能做什么”。1.2 我们到底想让 Agent 干什么在动手之前我先花了一个下午把需求写清楚。这个过程很痛苦因为“让 Agent 帮我干活”这句话太模糊了模糊的需求必然导致模糊的实现。我最后把目标收敛成三个具体的场景第一个场景是会议到行动项的转化。输入是一段会议转写文本输出是一份结构化的行动项清单每个行动项包含负责人、截止时间、依赖项和验收标准。这个场景的关键在于Agent 不能只是“总结”它必须“提取”并“补全”——很多会议里负责人是隐含的截止时间是模糊的Agent 需要根据上下文推断并标注置信度。第二个场景是技术调研到决策备忘录的转化。输入是一个技术问题输出是一份包含候选方案、对比维度、推荐结论和风险提示的备忘录。这个场景的关键在于Agent 不能只是“罗列”它必须“权衡”——它需要知道我们团队的约束条件比如部署成本、维护复杂度、学习曲线然后在这些约束下给出推荐。第三个场景是零散笔记到项目计划的转化。输入是一堆碎片化的想法和待办输出是一份有优先级、有里程碑、有依赖关系的项目计划。这个场景的关键在于Agent 不能只是“分类”它必须“排序”——它需要理解任务之间的依赖关系识别关键路径并给出合理的排期建议。这三个场景有一个共同点它们都不是“问答”而是“加工”。Agent 需要接收一种形态的输入经过一系列确定的处理步骤输出另一种形态的产物。这就决定了我们的技术选型不能围绕“对话”来设计而必须围绕“流水线”来设计。1.3 为什么选择“事件模式”作为核心抽象在架构设计阶段我们面临一个关键选择用什么作为 Agent 的核心抽象是“对话轮次”还是“任务”还是“事件”对话轮次的抽象最自然但问题也最明显——它假设用户和 Agent 之间的交互是线性的、同步的。但真实的干活场景不是这样的。我可能在开会时丢给 Agent 一段录音然后去忙别的事半小时后回来检查结果。这中间 Agent 可能需要调用外部工具、可能需要等待某个依赖、可能需要向我确认某个模糊点。这些都不是“对话轮次”能很好表达的。任务的抽象更接近干活的本质但它太粗粒度了。一个任务从创建到完成中间会经历很多状态变化如果只用一个“任务”对象来承载状态管理会变得非常复杂。最后我们选择了事件模式。具体来说Agent 的整个生命周期被建模成一系列事件的追加。用户丢进来一段录音这是一个input.received事件Agent 开始处理这是一个processing.started事件Agent 调用了一个外部工具这是一个tool.invoked事件Agent 需要用户确认这是一个confirmation.requested事件用户确认了这是一个confirmation.resolved事件Agent 输出了最终产物这是一个output.produced事件。这个设计的妙处在于它把 Agent 的“记忆”和 Agent 的“执行”统一到了同一个抽象下。记忆不再是独立于执行之外的一个模块而是执行过程中自然产生的事件流。Agent 要回忆什么只需要回放相关的事件Agent 要执行什么只需要追加新的事件。这种统一性大大简化了系统的复杂度。更重要的是事件模式天然支持追加写入。这意味着 Agent 的整个执行历史是不可变的、可审计的、可回放的。如果某次执行出了问题我可以精确地回放到出问题的那一步检查当时的事件上下文而不是面对一个黑盒。2. 核心细节解析事件模式、访谈协议与追加写入2.1 事件模式的设计要点与常见陷阱事件模式听起来简单但真正落地时有很多细节需要斟酌。我踩过的第一个坑是事件粒度过细。一开始我把每个微小的状态变化都定义成一个事件结果事件流变得极其冗长回放一次要处理上千个事件性能很差。后来我把事件粒度调整到“有语义意义的业务动作”这一层比如“工具调用完成”是一个事件但“工具调用的参数序列化”就不是一个独立事件而是前一个事件的属性。第二个坑是事件类型的命名。我见过很多项目用event_type_1、event_type_2这种命名过两周自己都忘了哪个是哪个。我的建议是采用领域.动作的命名规范比如meeting.transcribed、action_item.extracted、action_item.confirmed。这样一看就知道事件属于哪个领域、发生了什么动作。第三个坑是事件的版本兼容。Agent 的 schema 一定会随着需求变化而演进如果事件结构变了旧的事件怎么处理我们的做法是在每个事件里加一个schema_version字段回放时根据版本号做兼容性转换。这个成本不高但能避免很多麻烦。下面是我们实际使用的事件 schema 的核心字段{ event_id: evt_20250115_001, event_type: action_item.extracted, schema_version: 1.2, timestamp: 2025-01-15T10:23:45Z, session_id: sess_meeting_20250115, payload: { item_id: ai_001, description: 完成竞品分析报告初稿, owner: 张三, owner_confidence: 0.85, due_date: 2025-01-20, due_confidence: 0.6, dependencies: [ai_000], acceptance_criteria: 包含至少3个竞品的功能对比表 }, source_event_ids: [evt_20250115_000], produced_by: extractor_v2 }这个 schema 里有几个设计决策值得说明。owner_confidence和due_confidence是我强烈建议加的字段。会议里很多信息是模糊的与其让 Agent 假装确定不如让它诚实地标注置信度。置信度低于阈值的Agent 会主动发起确认请求而不是默默猜测。source_event_ids是另一个关键字段。它记录了当前事件是从哪些事件派生出来的这样整个事件流就形成了一张有向无环图而不是一条简单的线。当需要追溯某个行动项的来源时可以沿着source_event_ids一路回溯到原始的会议转写文本。produced_by字段记录了是哪个处理器产生了这个事件。当 Agent 的处理器升级后我可以对比新旧处理器的输出差异评估升级效果。2.2 访谈协议让 Agent 学会“问对问题”事件模式解决了“怎么记录”的问题但没解决“怎么获取信息”的问题。在会议到行动项的转化场景里很多关键信息在原始文本里是缺失的。比如会议里有人说“这个事我来跟进”但没说截止时间有人说“等那边确认了再说”但没说“那边”是谁。传统的做法是让 Agent 直接输出一个不完整的行动项然后人工补全。但这违背了我们“开工型 Agent”的初衷——Agent 应该主动补全而不是被动等待。我们借鉴了结构化访谈的思路设计了一套访谈协议。核心思想是Agent 在提取信息时如果发现关键字段缺失或置信度低于阈值就生成一个confirmation.requested事件把问题结构化地抛给用户。用户回答后生成confirmation.resolved事件Agent 继续处理。访谈协议的关键在于问题的生成策略。我们定义了三种问题类型补全型问题某个必填字段缺失直接问。比如“行动项‘完成竞品分析报告’的截止时间是什么”澄清型问题某个字段有多个候选值让用户选择。比如“会议里提到了‘张三’和‘张总’行动项的负责人是哪一个”确认型问题某个字段的置信度低于阈值让用户确认。比如“我推断截止时间是1月20日置信度60%是否正确”这三种问题的优先级不同。补全型问题必须问否则行动项不完整澄清型问题尽量问但如果有默认值可以先用默认值确认型问题可以批量问减少对用户的打扰。下面是我们实际使用的问题 schema{ question_id: q_001, question_type: completion, target_event_id: evt_20250115_001, target_field: due_date, question_text: 行动项‘完成竞品分析报告初稿’的截止时间是什么, candidate_answers: [], default_answer: null, priority: high, batch_id: batch_001 }batch_id是为了支持批量提问。如果一次处理产生了多个确认型问题Agent 会把它们打包成一个批次一次性呈现给用户而不是逐个弹窗。这个细节看起来小但对用户体验影响很大。2.3 追加写入为什么不可变日志是 Agent 的基石追加写入是我们整个架构里最底层的一个决策也是我花了最多时间说服团队的一个决策。很多人的第一反应是为什么不能直接修改事件如果发现某个行动项的负责人写错了直接改掉不就行了我的回答是修改会丢失信息而信息是 Agent 最宝贵的资产。假设 Agent 提取了一个行动项负责人是“张三”后来用户纠正说是“李四”。如果直接修改那么“Agent 曾经认为是张三”这个信息就丢失了。这个信息有什么用它可以用来评估 Agent 的提取准确率可以用来分析哪些类型的表述容易导致误判可以用来训练更好的提取器。追加写入的做法是不修改原事件而是追加一个action_item.corrected事件记录修正的内容和原因。回放时Agent 会先看到原始事件再看到修正事件最终状态是修正后的状态。但整个修正历史被完整保留。这个设计还有一个额外的好处支持时间旅行。我可以回放到任何一个时间点查看当时 Agent 的状态。这在调试时极其有用。比如我想知道为什么 Agent 在某个时间点做出了一个奇怪的决策我可以回放到那个时间点之前逐步执行观察每一步的事件上下文。当然追加写入也有代价。事件流会越来越长存储成本会上升。我们的应对策略是分层存储最近的事件放在内存里支持快速回放较旧的事件放在磁盘上按需加载超过一定时间的事件归档到冷存储只在审计时访问。还有一个代价是回放性能。如果事件流有十万条从头回放一次要很久。我们的做法是定期生成快照。快照记录了某个时间点的完整状态回放时只需要从最近的快照开始而不是从头开始。快照的生成频率可以根据事件流的增长速度动态调整。3. 实操过程从零搭建一个可开工的 Agent3.1 环境准备与依赖选型我们的技术栈选择遵循一个原则能用成熟库就不自己造轮子但核心抽象必须自己掌控。运行时我们选了 Python 3.11原因是生态成熟而且 3.11 的性能比 3.10 有明显提升。事件存储我们用了 SQLite 作为起步方案原因是零配置、单文件、支持事务。很多人一上来就上 PostgreSQL 或 MongoDB但对于个人 Agent 这个场景SQLite 完全够用而且部署成本低得多。等事件流真的增长到 SQLite 扛不住了再迁移也不迟。事件 schema 的校验我们用了 Pydantic。Pydantic 的好处是它把 schema 定义和校验逻辑合二为一而且错误信息很清晰。下面是我们的事件基类from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List, Dict, Any class BaseEvent(BaseModel): event_id: str event_type: str schema_version: str 1.0 timestamp: datetime Field(default_factorydatetime.utcnow) session_id: str payload: Dict[str, Any] source_event_ids: List[str] [] produced_by: str class ActionItemExtractedEvent(BaseEvent): event_type: str action_item.extracted class ActionItemPayload(BaseModel): item_id: str description: str owner: Optional[str] None owner_confidence: float 0.0 due_date: Optional[str] None due_confidence: float 0.0 dependencies: List[str] [] acceptance_criteria: Optional[str] None payload: ActionItemPayload这个基类定义了所有事件的公共字段子类只需要定义自己的 payload 结构。Pydantic 会自动处理校验和序列化。工具调用我们用了httpx而不是requests原因是httpx原生支持异步而 Agent 的执行流程里有大量 IO 等待异步能显著提升吞吐。当然如果你不熟悉异步编程用requests也完全可以只是性能会差一些。3.2 事件存储层的实现细节事件存储层是整个系统的地基它的设计直接决定了上层能做什么。我们的存储层提供了四个核心接口append(event)追加一个事件返回事件 IDget_events(session_id, sinceNone, untilNone)获取某个会话的事件流get_snapshot(session_id, at_time)获取某个时间点的快照replay(session_id, from_event_id)从某个事件开始回放append的实现很简单就是往 SQLite 里插一条记录。但有一个细节需要注意事件 ID 的生成必须是单调递增的。我们用了evt_{session_id}_{sequence}的格式其中sequence是会话内自增的。这样既能保证唯一性又能保证顺序性。get_events的实现需要注意分页。如果事件流很长一次性加载所有事件会爆内存。我们用了游标分页每次最多加载 1000 条事件。get_snapshot的实现是我们花时间最多的地方。快照的本质是把事件流“折叠”成一个状态对象。折叠的逻辑取决于事件类型。比如action_item.extracted事件会把行动项加入状态action_item.corrected事件会修改状态中的行动项action_item.deleted事件会从状态中移除行动项。def fold_events(events: List[BaseEvent]) - Dict[str, Any]: state {action_items: {}, confirmations: {}} for event in events: if event.event_type action_item.extracted: state[action_items][event.payload.item_id] event.payload.dict() elif event.event_type action_item.corrected: item_id event.payload.item_id if item_id in state[action_items]: state[action_items][item_id].update(event.payload.changes) elif event.event_type action_item.deleted: state[action_items].pop(event.payload.item_id, None) elif event.event_type confirmation.requested: state[confirmations][event.payload.question_id] event.payload.dict() elif event.event_type confirmation.resolved: state[confirmations].pop(event.payload.question_id, None) return state这个折叠函数是纯函数没有副作用所以可以安全地缓存和复用。我们会在每次追加事件后异步更新快照这样回放时只需要从最近的快照开始而不是从头开始。3.3 访谈协议的实现与问题生成策略访谈协议的实现分为三个部分问题生成、问题呈现、答案处理。问题生成是在提取器输出行动项后触发的。提取器会为每个行动项计算字段置信度如果某个必填字段的置信度低于阈值我们设的是 0.7就生成一个补全型问题如果某个字段有多个候选值就生成一个澄清型问题如果某个字段的置信度在 0.7 到 0.9 之间就生成一个确认型问题。def generate_questions(action_item: dict, threshold: float 0.7) - List[dict]: questions [] if not action_item.get(owner): questions.append({ question_type: completion, target_field: owner, question_text: f行动项‘{action_item[description]}’的负责人是谁, priority: high }) elif action_item.get(owner_confidence, 0) threshold: questions.append({ question_type: confirmation, target_field: owner, question_text: f我推断负责人是{action_item[owner]}是否正确, default_answer: action_item[owner], priority: medium }) if not action_item.get(due_date): questions.append({ question_type: completion, target_field: due_date, question_text: f行动项‘{action_item[description]}’的截止时间是什么, priority: high }) elif action_item.get(due_confidence, 0) threshold: questions.append({ question_type: confirmation, target_field: due_date, question_text: f我推断截止时间是{action_item[due_date]}是否正确, default_answer: action_item[due_date], priority: medium }) return questions问题呈现我们做了一个简单的 CLI 界面把同一批次的问题一次性列出用户可以用编号回答也可以直接回车接受默认值。这个界面很粗糙但足够用。关键是它把“打扰用户”的次数降到了最低。答案处理是把用户的回答转换成confirmation.resolved事件然后触发提取器重新处理受影响的行动项。这里有一个细节重新处理时提取器会跳过已经有确认答案的字段避免重复提问。3.4 从会议转写到行动项清单的完整流程让我用一个真实的例子把整个流程串起来。假设我有一段会议转写文本内容是关于一个产品迭代计划的讨论。第一步是输入接收。我把转写文本丢给 Agent生成一个input.received事件payload 里包含文本内容和元数据会议时间、参与人等。第二步是预处理。Agent 对文本做分句、去噪、说话人分离。这一步的输出是一个text.preprocessed事件payload 里包含结构化的对话片段。第三步是行动项提取。Agent 用提取器从对话片段里识别行动项。提取器是一个基于规则和模型混合的组件——规则负责识别“我来做”“你负责”“下周完成”这类模式模型负责处理更复杂的表述。提取器的输出是一批action_item.extracted事件。第四步是问题生成。Agent 检查每个行动项的字段完整性生成confirmation.requested事件。这些事件被批量呈现给我。第五步是答案处理。我回答了问题Agent 生成confirmation.resolved事件并更新行动项。第六步是产物组装。Agent 把所有行动项组装成一份 Markdown 格式的清单生成output.produced事件。整个流程走下来从丢入文本到拿到清单大概需要 30 秒到 2 分钟取决于文本长度和问题数量。这个速度不算快但考虑到它省掉了我手动整理的一两个小时完全值得。4. 常见问题与排查技巧实录4.1 事件流异常增长的排查与治理上线两周后我发现事件流的增长速度远超预期。一个小时的会议产生了将近 500 个事件。按这个速度一个月下来事件表就要爆了。我排查后发现问题出在工具调用的日志事件上。每次 Agent 调用外部工具都会生成一个tool.invoked事件和一个tool.completed事件payload 里包含了完整的请求和响应。这些 payload 很大而且大部分是冗余的。治理方案是分级记录。对于成功的工具调用只记录摘要信息工具名、耗时、结果状态不记录完整的请求响应对于失败的工具调用才记录完整信息方便排查。这个改动把事件数量降到了原来的三分之一。另一个治理方案是事件合并。对于高频的、语义相近的事件比如连续多个action_item.extracted事件可以合并成一个action_items.extracted事件payload 里包含一个数组。这个改动进一步把事件数量降到了原来的五分之一。下面是我们的事件分级策略事件类型记录级别保留时长说明input.received完整永久原始输入必须保留text.preprocessed摘要30天只保留统计信息action_item.extracted完整永久核心业务事件tool.invoked (成功)摘要7天只保留工具名和耗时tool.invoked (失败)完整90天保留完整请求响应confirmation.requested完整永久核心业务事件output.produced完整永久最终产物这个策略的核心思想是业务事件永久保留技术事件定期清理。业务事件是 Agent 的记忆技术事件只是执行过程的痕迹。4.2 提取器误判的典型模式与修正方法提取器误判是另一个高频问题。我整理了最常见的几种误判模式第一种是指代消解错误。会议里有人说“这个事我来跟进”提取器需要判断“这个事”指的是哪个事。如果前文有多个候选提取器很容易选错。我们的修正方法是引入指代消解模块专门处理代词和指示词的消解并把消解结果作为行动项的一个字段记录下来方便追溯。第二种是时间表达歧义。“下周完成”到底是下周一还是下周五我们的修正方法是引入时间归一化模块把相对时间表达转换成绝对日期并标注置信度。置信度低的会触发确认问题。第三种是负责人误判。会议里有人说“张三你来负责这个”提取器可能把负责人识别成说话人而不是张三。我们的修正方法是引入角色识别模块区分“指派者”和“被指派者”并优先把被指派者作为负责人。第四种是行动项边界模糊。一段话里可能包含多个行动项也可能一个都不包含。我们的修正方法是引入边界检测模块用序列标注的方式识别行动项的起止位置。这些修正模块不是一次性的而是持续迭代的。我们建了一个误判案例库每次发现新的误判模式就把它加入案例库并针对性地调整提取器。这个案例库现在有 200 多条记录是我们最宝贵的资产之一。4.3 确认问题的用户体验优化确认问题是一把双刃剑。问得太少行动项不完整问得太多用户嫌烦。我们在这个平衡上踩了不少坑。第一个坑是问题太多。一开始我们对所有置信度低于 0.9 的字段都生成确认问题结果一次处理产生了 20 多个问题用户直接放弃了。后来我们把阈值降到 0.7并且只对必填字段生成问题问题数量降到了 5 个以内。第二个坑是问题太碎。每个问题单独呈现用户要反复切换上下文。后来我们引入了批量提问把同一批次的问题一次性列出用户可以用编号回答也可以直接回车接受默认值。第三个坑是问题太模糊。比如“截止时间是什么”这个问题用户可能不知道该怎么回答。后来我们引入了候选答案Agent 根据上下文生成几个候选用户只需要选择而不是从零输入。第四个坑是问题太频繁。每次处理都问用户很快就烦了。后来我们引入了学习机制Agent 会记住用户的历史回答对于相似的问题如果用户之前已经回答过就不再重复提问。下面是我们的问题优先级策略问题类型触发条件优先级是否批量补全型必填字段缺失高是澄清型字段有多个候选中是确认型置信度低于阈值低是学习型历史已回答过最低否这个策略的核心思想是能不问就不问能批量问就不单独问能选就不让用户输入。4.4 追加写入的性能优化实践追加写入的性能问题在事件流增长到一定规模后才会显现。我们的事件表在达到 10 万条记录时回放一次要 3 秒以上这在交互场景里是不可接受的。优化方案有三个层次第一个层次是索引优化。我们给session_id和timestamp建了联合索引这样按会话和时间范围查询事件时不需要全表扫描。这个改动把查询时间从 3 秒降到了 200 毫秒。第二个层次是快照优化。我们每 1000 个事件生成一个快照回放时从最近的快照开始而不是从头开始。这个改动把回放时间从 200 毫秒降到了 20 毫秒。第三个层次是缓存优化。我们把最近的事件流缓存在内存里回放时优先从缓存读取。这个改动把回放时间进一步降到了 5 毫秒以内。下面是我们的事件表结构CREATE TABLE events ( event_id TEXT PRIMARY KEY, event_type TEXT NOT NULL, schema_version TEXT NOT NULL, timestamp DATETIME NOT NULL, session_id TEXT NOT NULL, payload TEXT NOT NULL, source_event_ids TEXT, produced_by TEXT NOT NULL ); CREATE INDEX idx_session_timestamp ON events(session_id, timestamp); CREATE INDEX idx_event_type ON events(event_type); CREATE TABLE snapshots ( snapshot_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, at_event_id TEXT NOT NULL, state TEXT NOT NULL, created_at DATETIME NOT NULL ); CREATE INDEX idx_snapshot_session ON snapshots(session_id, at_event_id);这个表结构很简单但足够支撑我们的需求。关键是索引的设计——session_id和timestamp的联合索引覆盖了最常见的查询模式。4.5 常见问题速查表问题现象可能原因排查方法解决方案事件流增长过快技术事件记录过细统计各类型事件的数量占比分级记录技术事件只记摘要回放速度慢事件流太长无快照测量回放耗时与事件数量的关系定期生成快照加索引提取器误判多指代、时间、角色未处理抽样检查误判案例引入专门的消解和归一化模块确认问题太多阈值过低问题太碎统计每次处理的问题数量提高阈值批量提问引入学习机制行动项不完整必填字段缺失未补全检查行动项的字段完整率补全型问题必须问不能跳过用户回答后未更新答案处理逻辑有 bug检查 confirmation.resolved 事件确保答案事件触发重新处理快照与事件流不一致快照生成有竞态对比快照状态与回放状态快照生成加锁或改为异步生成事件 schema 不兼容版本演进未处理检查旧事件的 schema_version回放时做版本兼容转换这张表是我在实际运维中逐步积累的每一条都对应着一次真实的故障或优化。我建议你也建一张类似的表把每次踩坑的经验记录下来这比任何文档都有价值。5. 一些个人体会与后续扩展方向这套系统跑了一个多月我最深的体会是Agent 的价值不在于它有多聪明而在于它有多可靠。一个能稳定完成 80% 工作的 Agent比一个偶尔能完成 100% 工作但经常出错的 Agent 有用得多。可靠性来自于确定的流程、清晰的抽象和持续的迭代而不是来自于更大的模型或更复杂的提示词。事件模式、访谈协议和追加写入这三个设计决策本质上都是在为可靠性服务。事件模式让执行过程可追溯访谈协议让信息缺口可补全追加写入让历史状态可回放。这三个能力加起来才让 Agent 从“玩具”变成了“工具”。后续我打算在这几个方向继续扩展一是多会话协同让多个 Agent 可以共享事件流协同完成一个复杂任务二是事件流的可视化把事件流渲染成时间线方便人工审查和调试三是提取器的持续学习把用户的修正反馈自动转化为训练数据让提取器越用越准。如果你也在做类似的事情我的建议是先从一个小场景开始把事件模式跑通把访谈协议用起来把追加写入坚持住。不要一上来就追求大而全先把一个场景做可靠再扩展到其他场景。可靠性是积累出来的不是设计出来的。