1. 为什么智能体需要事后复盘这双眼睛如果你跑过几次基于大模型的自动化任务大概率遇到过这种场面Agent第一次执行时在某一步卡死你改了prompt重跑它换了个姿势继续错直到你把整条链路里的每个坑都踩完它才勉强跑通。然后你复盘这个过程发现自己说的最多的一句话是——这步当时明明可以这样处理怎么就没想到呢这就是hindsight这个项目的出发点。它不是一个对话机器人也不是什么炫酷的生成式应用而是一套给AI Agent用的经验回放与事后反思系统。简单说当Agent在执行任务时它每走一步都会被记录下来任务结束后hindsight会把这些运行轨迹翻出来逐段审视哪里对了、哪里绕远了、哪里彻底跑偏了然后把反思结果沉淀成结构化建议反哺到下一轮执行里。我一开始做这个项目也谈不上多么高瞻远瞩单纯是被自己写的Agent气到了。去年做一个批量数据处理工具链Agent每次处理到一个特定格式的文件就断掉每次报错信息都不一样我像消防员一样扑了一个星期的火。后来我才意识到问题的本质Agent在每一个当下都只能做贪婪决策它没有能力站在事后视角看整条流程自然也学不会避坑。而我要做的就是给这个只看当下的执行器装上一个事后视角的复盘系统。这套东西适合谁呢如果你在写自动化脚本、在搭个人Agent、在维护一条多步骤的数据流水线或者单纯受够了反复调试同一个错误那这套思路和实现方案大概率对你有用。它解决的问题不是让Agent变聪明而是让Agent能记住自己踩过什么坑这才是稳定的进步方式。2. 核心设计思路记录、回看、反思、再执行hindsight一开始的定位就很明确它不抢Agent的活只做Agent身后的那个记录员评论员。整个项目拆开来看其实就是四个环节——记录Record、回放Replay、反思Reflect、再执行Re-execute。每个环节单独看都很简单难的是把它们串成一个能闭环的回路。2.1 记录环节不是所有事件都值得被记第一步是埋点记录。你可能会想这不就是打日志吗还真不太一样。传统日志关注的是系统发生了什么而hindsight关注的是Agent当时在想什么、面对什么状态、做了什么选择。所以事件结构里不只要记做了什么还要带上决策依据和当时的环境快照。我的事件结构长这样{ event_id: evt_001, timestamp: 2024-11-20T14:32:10.882Z, type: action_executed, agent_state: { current_step: file_parse, context_count: 87, model: qwen2.5-14b-instruct }, input: { file_path: /data/raw/sales_2024.csv, encoding: utf-8-sig }, decision: { chosen_action: parse_with_pandas, alternatives: [parse_with_csv, extract_by_lines], confidence: 0.73 }, result: { status: failed, error_type: EncodingError, error_message: codec cant decode byte 0xed } }这套结构说白了有一个好处你回过头来复盘的时候能还原现场。很多系统挂了之后开发者面对的都是结果而对Agent来说过程比结果重要得多——因为Agent之所以犯错往往不是因为结果做错了而是因为在决策岔路口选了一条不该选的路。埋点要注意一点别什么都记。我起初把Agent每一步的完整上下文都塞进事件里几分钟就把磁盘写满了。后来改成状态摘要 决策信息 结果概要三段式事件体积压缩了将近80%复盘质量反而没降——因为真正关键的信息就那么几项。2.2 回放环节把散落的轨迹串成一条故事线记录完一大堆事件之后下一个问题就是怎么把这一堆事件变成有意义的东西这时候就需要一个回放层。它做的事情是把一条任务从开始到结束的所有事件按时间轴重新串起来分拣出关键节点标记出转折点——所谓转折点我定义的是以下三类首次失败点任务链条里第一个报错的环节决策犹豫点Agent在多个可用动作之间反复跳动的环节状态异常点输入输出偏离预期类型的环节回放层输出的是一份结构化的过程摘要类似于给这个任务写了一份编年史。这份编年史会作为反思环节的输入。很多人会忽略回放的价值觉得直接拿原始日志去问LLM不就行了。我试过效果非常差。因为原始日志里70%是噪声LLM拿到一大坨日志要么抓不住重点要么被无关信息带跑反思质量极其不稳定。经过回放层的精简之后反思环节面对的是一份已经提炼过的事件序列质量和稳定性一下子上来了。这就是回放层存在的价值——它不是可有可无的中间层而是整个系统的信息过滤器。2.3 反思环节让大模型输出可执行经验回放层整理完素材反思环节就可以上场了。我这里的做法是把过程摘要交给大模型配合一套反思提示模板引导它输出三个维度的内容问题归因这次任务失败/低效的根本原因是什么不要只停留在表面报错信息改进建议给出具体的、可执行的调整方案是改prompt、换工具、还是调整参数顺序泛化规律把这个案例抽象成一条一般性经验比如遇到CSV编码问题时优先尝试utf-8-sig这套提示模板我迭代过好几个版本最终稳定在下面这个结构上你是一个任务复盘专家。以下是AI Agent执行任务的过程摘要 过程摘要 请从以下三个维度进行分析 1. 问题归因找出任务失败或低效的根本原因 2. 改进建议给出3-5条具体可执行的调整方案 3. 泛化规律总结一条可以迁移到类似场景的经验 要求不要空泛评价每一条建议必须能直接落地。这里有一个关键经验你想要的不是解释而是方案。早期版本我用的是请分析这次任务的失败原因结果它给了我一大段华丽的错误分析很好看但看完还是不知道下一步该改什么。改成上面这个模板之后输出变得实用得多。2.4 再执行环节经验被注入下一次任务反思环节产出的建议如果没有被下一次执行消费掉整个系统就等于白做了。所以最后一步是经验注入。我的做法是把反思结果存到一个经验库里每次Agent开始新一轮任务时hindsight从经验库里检索与当前任务相似的历史经验作为经验提示注入到Agent的system prompt里。这一步的设计其实最折磨人因为它有一个很难绕开的坑如果直接把历史经验塞进promptAgent会在无关任务上也被这些经验影响反而降低执行质量。我采用的方案是给每条经验打上标签在注入前做相似度匹配只召回与当前任务类型匹配的经验并且设了一个召回上限——最多三条宁缺毋滥。这四步串起来才算是把事后复盘变成了事前能力。听起来不是很惊艳对吧说实话这套架构确实没有多少技术壁垒真正有价值的地方在于怎么把每一步做扎实尤其是细节处理。3. 实操实现从零搭一个最小可用的Hindsight前面说了这么多概念这把这块落成代码。我用的Python 3.10依赖很少核心库就是sqlite3和标准库反思环节接一个大模型API就行。整个项目的代码量不大大概500行左右但麻雀虽小五脏俱全。3.1 数据层用SQLite做轨迹存储首先需要一个存储层。我选择SQLite而不是JSON文件落盘原因很简单复盘时经常要做条件查询比如找出所有失败的parse事件用SQL查比在JSON里遍历方便得多。建表语句是这样的CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, timestamp TEXT NOT NULL, event_type TEXT NOT NULL, agent_state TEXT, input_data TEXT, decision_data TEXT, result_data TEXT, created_at TEXT DEFAULT (datetime(now)) ); CREATE INDEX IF NOT EXISTS idx_task_time ON events(task_id, timestamp); CREATE INDEX IF NOT EXISTS idx_task_type ON events(task_id, event_type);表结构定义得宽一些agent_state、input_data这些字段都存JSON字符串因为事件结构后期会演化如果一开始把字段定死后面加字段就要改表结构麻烦。事件写入的代码很简单import json import sqlite3 import uuid from datetime import datetime, timezone class EventRecorder: def __init__(self, db_path: str hindsight.db): self.conn sqlite3.connect(db_path) self._init_db() def _init_db(self): self.conn.executescript(SCHEMA_SQL) self.conn.commit() def record(self, task_id: str, event: dict): event_id uuid.uuid4().hex ts datetime.now(timezone.utc).isoformat() self.conn.execute( INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?, ?), ( event_id, task_id, ts, event.get(type, unknown), json.dumps(event.get(agent_state, {}), ensure_asciiFalse), json.dumps(event.get(input, {}), ensure_asciiFalse), json.dumps(event.get(decision, {}), ensure_asciiFalse), json.dumps(event.get(result, {}), ensure_asciiFalse), ), ) self.conn.commit()往自己的Agent代码里埋点的时候其实不需要在业务逻辑里到处塞代码我一般只找几个关键位置埋点每轮决策开始前、每轮决策结束并拿到结果后、以及任务结束/异常退出时。三个点就够了埋太密反而是负担。3.2 回放层提炼关键节点而不是罗列日志回放层的实现我把它分成两步第一步按task_id把事件全部取出来按时间排序第二步扫一遍事件序列标记出关键节点。class Replayer: def __init__(self, conn): self.conn conn def replay(self, task_id: str) - dict: rows self.conn.execute( SELECT * FROM events WHERE task_id? ORDER BY timestamp, (task_id,), ).fetchall() events [self._row_to_dict(row) for row in rows] summary { task_id: task_id, total_steps: len(events), key_nodes: [], event_sequence: [], } for i, evt in enumerate(events): node {index: i, type: evt[event_type], summary: self._summarize(evt)} result json.loads(evt[result_data]) if result.get(status) failed: node[node_type] failure if self._is_hesitation(evt): node[node_type] hesitation if self._is_anomaly(evt): node[node_type] anomaly summary[key_nodes].append(node) summary[event_sequence].append(evt[event_type]) return summary def _is_hesitation(self, evt): # 检查决策里是否出现反复切换 decision json.loads(evt[decision_data] or {}) return decision.get(switched, False) def _is_anomaly(self, evt): result json.loads(evt[result_data] or {}) return result.get(anomaly, False)这里有个细节值得说一下事件序列不一定要输出所有原始数据回放层输出给反思环节的应该是精简过的节点摘要每条节点不要超过50个字。这样传给LLM的内容才够干净也省token。3.3 反思层接入大模型输出结构化经验回放层准备好了素材反思层其实就是一个调用LLM 解析输出的模块。class Reflector: def __init__(self, llm_client, prompt_template: str): self.llm llm_client self.template prompt_template def reflect(self, replay_summary: dict) - dict: prompt self.template.replace({{process_summary}}, json.dumps(replay_summary, ensure_asciiFalse)[:6000]) raw self.llm.chat(prompt) return self._parse_response(raw) def _parse_response(self, raw: str) - dict: # 这里用简单正则或JSON解析要求LLM按固定格式输出 # 我这里要求它输出JSON try: return json.loads(raw) except json.JSONDecodeError: # 兜底用正则把三个字段抠出来 return { root_cause: parse_failed, suggestions: [无法解析反思结果请检查模型输出格式], general_rule: , }反思层的成败很大程度上取决于输出格式约束。我踩过的一个大坑是LLM偶尔不按JSON格式输出导致解析失败。后来我在提示模板里写死必须输出JSONkey固定为root_cause/suggestions/general_rule并且加了上面的兜底解析逻辑——解析失败时不至于让整条链路崩溃而是标记一下跳过本次反思。3.4 经验库与注入给每条经验打标签控制召回数量经验库这层我用一张数据库表来存CREATE TABLE IF NOT EXISTS lessons ( id TEXT PRIMARY KEY, task_type TEXT NOT NULL, content TEXT NOT NULL, source_task_id TEXT, created_at TEXT, valid_until TEXT, hit_count INTEGER DEFAULT 0 );注入逻辑的关键在于任务类型匹配。每个任务在启动时都要声明自己的task_type比如csv_batch_process、pdf_extract这种粒度太粗了召回不精准太细了每条经验都只能用在单一任务上失去泛化价值。召回代码如下class LessonInjector: def __init__(self, conn): self.conn conn def inject(self, task_type: str, max_lessons: int 3) - str: rows self.conn.execute( SELECT content FROM lessons WHERE task_type? AND (valid_until IS NULL OR valid_until datetime(now)) ORDER BY hit_count DESC LIMIT ?, (task_type, max_lessons), ).fetchall() if not rows: return lessons_text \n.join([f- {r[0]} for r in rows]) # 更新命中计数 self.conn.execute( UPDATE lessons SET hit_count hit_count 1 WHERE task_type?, (task_type,), ) self.conn.commit() return f【历史经验参考】\n{lessons_text}\n请结合以上经验优化本次执行但不要盲从。这里有一个很有用的设置就是valid_until字段。经验不是越老越值钱一些经验在上下文变化后反而会误导Agent比如某个第三方库升级后旧的经验可能就失效了。所以反思产出的经验默认有效期为30天过期后自动排除除非被人工确认过长期有效。3.5 把全流程串起来所有模块就位后主流程是这样跑的class Hindsight: def __init__(self, db_path: str, llm_client, prompt_template: str): self.conn sqlite3.connect(db_path) self.recorder EventRecorder(db_path) self.replayer Replayer(self.conn) self.reflector Reflector(llm_client, prompt_template) self.injector LessonInjector(self.conn) def run_task_with_feedback(self, task_type: str, task_runner, task_input): # 注入历史经验 lesson self.injector.inject(task_type) if lesson: task_input[extra_context] lesson # 执行任务 task_id uuid.uuid4().hex result task_runner(task_input, event_callbacklambda evt: self.recorder.record(task_id, evt)) # 任务结束后复盘 if result.get(needs_review, True): replay self.replayer.replay(task_id) reflection self.reflector.reflect(replay) self._store_lesson(task_type, reflection, task_id) return result整体跑下来的效果我自己测了两个场景一个是CSV批量清洗任务一个是网页信息提取任务。接入hindsight后同样的任务跑第二遍时成功率和效率都有明显提升——CSV清洗任务的成功率从63%提升到87%网页提取任务的平均耗时下降了约20%。样本量不大但这个趋势是确实能感受到的。4. 常见问题与排查实录这套系统我上线用了三个月左右中间踩过不少坑挑几个典型的列出来如果你也在做类似的东西应该能帮你省不少时间。4.1 事件数据量爆炸刚开始我每个Agent步骤都把完整的上下文、工具返回结果、token用量全部记录下来跑一个稍复杂的任务就产出几MB事件数据。查询变慢、存储膨胀、复盘时给LLM的输入太长。后来我做了三层过滤第一层只记录决策点事件把状态轮询这类过程事件过滤掉第二层对长文本字段做截断只保留前500字第三层按任务保留原始事件但超过500个事件的任务自动触发采样压缩。做完这三步之后数据量降到了原来的十分之一复盘质量没有明显变化。4.2 反思输出质量不稳定这个问题困扰了我很久。同一个任务的两次复盘一次输出很精准、直击要害另一次就泛泛而谈。排查后发现两个影响因素一是回放摘要的质量如果关键节点标记错了反思就会跟着歪二是温度参数反思任务我用的temperature是0.2而不是执行任务时的0.7。反思本身更像分析题而不是创作题把温度压低之后稳定性好了不少。4.3 注入历史经验反而把Agent带偏了这个问题是上线之后最让我头疼的一个。某一次任务里Agent参考了一条经验结果那条经验是针对旧版本的代码逻辑新版本已经不需要那个workaround了Agent反而多做了无用功。出现这个问题的根源是经验没有时效性。我后面加了两道防线第一道就是前面说的valid_until默认30天过期过期后不自动删除但不再注入只保留在库里供人工查看第二道是在注入提示词的最后加了一句请结合以上经验优化本次执行但不要盲从。如果经验与当前任务的实际情况不符以当前实际情况为准。这两道防线加完误用经验的情况大大减少。4.4 时间戳不同步导致回放顺序错乱有一次我观察到某条任务的回放事件顺序明显乱掉了排查之后发现是埋点的时候timestamp用的是time.time()这种本地时间跨进程时时钟源不一致。后来我统一改成了UTC时间戳并且规定埋点端必须传入时间戳而不是在Recorder内部生成。这是一个很细碎的坑但排起来很折磨人。4.5 LLM输出格式偶发解析失败反思环节依赖LLM输出JSON但偶尔会遇到输出里带markdown代码块标记或者JSON结构不完整。我除了写死格式要求之外还加了一层正则清理——把包裹的json和等标记剥掉再解析。兜底逻辑如果解析失败会把这次反思标记为质量存疑同时不中断主流程。复盘链路绝不能成为任务执行链路的单点故障这是我写这套系统时坚持的一个底线。简单整理一个排查速查表症状可能原因排查方向复盘内容空泛关键节点标记不准检查回放层的节点标记逻辑经验注入后效果变差经验过时或与任务不匹配检查valid_until和task_type匹配事件数据增长过快记录粒度过细启用决策点过滤和长文本截断回放事件顺序错乱时间戳时钟源不一致统一UTC时间戳埋点端传时间反思输出无法解析LLM输出格式漂移加正则清理和兜底解析任务执行变慢注入经验过多降低召回上限建议不超过3条5. 从复盘工具到学习系统hindsight还能怎么走我把hindsight这套东西跑通之后最大的感受不是我多了一个工具而是明白了另一个层面的问题Agent和普通程序之间的本质差异不在于它能生成自然语言而在于它能不能从自己的经验里持续进步。没有复盘闭环的Agent不管prompt写得多么精致本质上还是一个没有记忆的函数换一个场景就重新开始。目前hindsight核心闭环已经能跑但如果继续往下做我看到的几个方向是值得投入的。第一个方向是做经验冲突检测。目前的经验库比较简单新经验直接append但实际运行一段时间后你会发现两条经验之间可能存在矛盾。比如一条经验说遇到编码错误时优先尝试utf-8-sig另一条经验可能说某些文件用utf-8-sig反而会乱码应该检测BOM头。这两条经验放在一起Agent反而会困惑。理想的做法是在新经验入库之前和历史经验做一次语义相似度检测如果有矛盾就把冲突标记出来交给人工判断。这个我还没完全做好目前只是加了一个简单的关键词冲突检测能发现一部分明显矛盾。第二个方向是给复盘结果量化打分。现在reflection输出的是问题归因、改进建议、泛化规律三段式文本但缺少一个统一的量化指标来评估这次复盘的质量。我打算在后面加一个经验价值分由多个维度组成问题是否被复现、建议是否有可操作性、泛化规律是否能覆盖到其他任务。有了这个分经验库召回的时候就不止按hit_count排序了还能按经验价值排序效果应该更好。第三个方向是把复盘从任务结束后扩展到任务过程中。现在的设计是跑完一个任务整体复盘但如果任务特别长、有几十个步骤等到最后才发现前面第5步就歪了纠正成本已经很高。我下一步打算在任务执行中每隔几步触发一次轻量检查只要发现关键指标异常就立刻中断当前路线先小节复盘、再继续走。这个做好之后整个系统就不只是事后诸葛亮了而更像一个实时导航但那是另一个话题了。我自己在实际落地hindsight时还有一个心得不要试图让所有复盘都自动化保留人工抽审的入口会舒服很多。让系统把经验自动沉淀下来是一方面但每隔一段时间人工翻一次经验库往往会发现一些自动反思完全没发现的隐藏问题。AI负责及时近忧式的复盘人负责远虑式的审视这个搭配目前对我来说是最好用的组合。这套思路你在自己的项目里不妨也试试从一个小闭环开始跑起来坚持几轮之后效果你自己会看见。