YAOTU INSIGHTS

AI智能体生产反馈闭环:从失败样本到Prompt持续优化

AI智能体生产反馈闭环:从失败样本到Prompt持续优化
上线一个 AI 智能体真正让人崩溃的往往不是模型效果差而是它在生产环境里的表现和你在测试集上看到的结果完全不像同一个系统。你精心设计了 prompt构造了二十条测试用例结果全过。一上生产用户换个问法、给个上下文、传个格式不规范的文件它就开始胡说。更麻烦的是这种错误反馈散落在用户投诉、客服工单、后台日志和数据库里没有任何机制把它们收集起来更不用说变成下一次迭代的训练素材。过去我们解决这个问题的办法也很原始定期导日志人工翻聊天记录把典型错误整理成文档再手动改 prompt。这套流程慢、漏、依赖个人责任心而且改完之后无法证明问题真的被修复了。Reflexio 这个项目想解决的正是这个环节的自动化让 AI 智能体具备一个从生产反馈中收集问题、评估错误、触发改进的学习回路。这篇文章我会把它的设计思路拆开来讲同时给出一个可以落到自己项目里的最小闭环实现。1. 为什么 AI 智能体需要“生产反馈闭环”先厘清一个概念AI 智能体AI Agent和普通的接口调用有一个本质差异。普通接口的输入输出结构是确定的请求参数少了、类型错了系统能立刻报错。但 Agent 是目标驱动的它接收的是一个开放式任务自己决定调用哪些工具、按什么顺序执行、中间结果是否可信。这意味着 Agent 的失败模式不是单一的错误码而是多种多样的“结果质量不达标”。我把常见的失败模式归纳为三类任务完成度缺失用户要求生成一份包含三个维度的市场分析报告Agent 只做了两个维度没有提醒用户遗漏。工具使用不当在天气查询和航班查询之间选错工具或者传入工具的参数格式有误导致返回结果本身是对的但 Agent 理解错了。上下文误判用户在一个多轮对话里说“那个文件帮我处理一下”Agent 没有结合上文确认“那个”指哪个文件直接对不存在的文件发起了操作。这些错误有一个共同点只能在真实使用中暴露很难在设计阶段穷举。传统的做法是建设离线的评测集但评测集有两个天然局限覆盖度有限。无论测试工程师多么努力构造用例的速度永远追不上用户提问的多样性。静态数据会过期。用户的使用习惯、业务政策甚至语言风格都在变化半年前构造的测试集今天可能已经不具备代表性。所以 Agent 系统必须有一个生产反馈的采集通道把真实世界里出现的“预期偏差”捕获下来形成一个可持续生长的数据库。这个数据库既可以用来回归测试也可以用来触发改进。Reflexio 的核心价值就是把这条链路从概念变成了可运行的项目骨架。2. Reflexio 的架构思路与核心概念Reflexio 不是一个全新的框架它更像是在现有 Agent 架构之上加了一个“反馈处理中枢”。一个典型的 Agent 生产系统通常包含这几个部分用户输入、编排层、大模型、工具层、知识库、输出层。Reflexio 不替代这些部分而是在它们旁边增加一个旁路系统用于采集数据、评估结果和生成改进建议。它解决的核心问题可以拆成四个环节采集Capture把一次 Agent 的完整执行过程记录下来包括输入、输出、调用了哪些工具、每步耗时、用户后续反馈。评估Evaluate判断这次执行结果到底好不好用规则、模型或用户反馈进行自动化打标。改进Improve把评估失败的样本聚合成问题模式生成可供开发者修改的测试用例或提示词优化建议。沉淀Memorize将新发现的问题模式加入回归集避免同一个问题在后续版本中再次出现。这四个环节里最核心的设计决定是把“评估”和“改进”解耦。很多团队在设计这类系统时容易急于让模型自动修改 prompt 或自动换个模型版本。这个方向听起来很美好风险也极大如果评估本身不可靠模型自动改进的每一次动作都可能引入更隐蔽的回归。Reflexio 的做法更稳先把评估做扎实改进环节保留人工确认的闸门让系统提出建议由人做最终裁决。用传统软件工程来类比这个过程非常像“测试驱动开发”。在写新功能之前先写下失败的测试用例。在 Agent 开发中这些“测试用例”就是生产环境采集到的失败对话样本。开发者要做的是先让 AI 智能体在这些真实失败样本上不再犯错然后才思考新功能的开发。Reflexio 实际上是在给 AI 智能体开发流程引入一套质量保障体系。3. 从反馈到学习生产数据如何变成改进信号理解了整体结构接下来要解决一个关键问题用户的原始反馈往往不是结构化的怎么把它变成可学习、可执行的信号举例来说用户在对话框里留下一句话“你刚才给的那个文件里没有第三章内容而且格式跟我要求的不一样。”这句话包含了几个信息层次操作对象是“刚才给的文件”。问题描述是“缺少第三章”。质量要求是“格式不符”。但如果用户只是默默关掉页面或者点了一个“无帮助”按钮那采集到的信号就更微弱了。Reflexio 的采集层正是为了处理这种多层级的反馈而设计的。它的输入通道分为三类第一类是显式反馈即用户主动表达的负面评价。这类反馈价值最高应当被完整保留并及时处理。第二类是隐式反馈例如用户复制了 Agent 的答案去搜索、直接要求转人工客服、或者对某句回复点了踩。这类反馈需要靠规则或者模型推断意图。第三类是系统自检反馈即由另一个模型评测模型或规则引擎来评估 Agent 的回复质量例如检查是否包含幻觉信息、是否完成了用户的所有指令。三类反馈的结构化程度不同处理优先级也不一样。显式反馈应进入人工审核队列隐式反馈应当进行聚合分析系统自检反馈则可以自动化进入回归集。在设计数据结构时反馈样本不需要保存所有原始对话但必须保留足够重建上下文的关键信息。{ sample_id: sample_20250610_001, agent_task: 生成长达50页的港股行业比较报告, user_input: 帮我生成一份港股和A股医疗器械行业的比较报告包含近三年的财务指标和估值分析, agent_output: 报告已给出包含11页行业概述和5页财务数据……, tool_calls: [ {tool: search_stock, input: {query: 港股医疗器械行业龙头}, result_summary: 获取到12家公司}, {tool: calc_financial_metrics, input: {code: 02160.HK}, result_summary: 完成ROE和PE计算} ], user_feedback: 报告内容太短要求50页但明显没有展开财务指标维度也不全, feedback_type: explicit, quality_score: 0.2, error_category: task_completeness, occurred_at: 2025-06-10T13:25:11Z }在命名上这条记录就是一次“失败样本”如果你维护了一个评测集一般会叫它fail_case.csv。它代表了一次需要进入改进流程的质量事件。4. 落地最小闭环环境搭建与工具选择如果要把这套理念真正用起来不必等 Reflexio 的完整版本发布。只要自己有 Python 基础就能基于它的核心思路搭建一个最小闭环系统。推荐的技术栈如下Python 3.10 及以上版本较多 Agent 框架已经默认支持新版 type hint。LangChain 或 LlamaIndex作为 Agent 编排层。本文的示例不依赖任何特定框架逻辑是通用的。Pydantic用于执行记录的 schema 校验和持久化。FastAPI用于搭建一个简单的反馈接收 HTTP 服务。SQLite 或 PostgreSQL。需要在本地快速演示用 SQLite 足够生产环境建议使用 PostgreSQL 并开启全文检索能力。OpenAI SDK 或兼容的模型接口用于评测环节的自动打分。版本说明模型接口版本变化较快本文不会写死某个具体版本号。实际使用时以官方文档发布版本为准。环境准备阶段用 venv 创建一个隔离的 Python 虚拟环境是稳妥做法python -m venv .venv source .venv/bin/activate随后创建requirements.txtpydantic2.5 fastapi0.110 uvicorn0.27 openai1.12 python-dotenv1.0安装依赖pip install -r requirements.txt5. Feedback Collector 的代码实现与配置先实现最核心的反馈采集模块作用是把生产环境的各种反馈统一收敛到本地数据库。假设项目的工作目录如下reflexio-demo/ ├── app/ │ ├── main.py │ ├── collect.py │ ├── store.py │ └── schema.py ├── requirements.txt └── .envschema.py定义反馈入库的数据模型# 文件路径app/schema.py from datetime import datetime from typing import Optional from uuid import uuid4 from pydantic import BaseModel, Field class FeedbackIn(BaseModel): agent_task: str user_input: str output: str tool_calls: list Field(default_factorylist) user_feedback: str feedback_type: str explicit # explicit | implicit | selfcheck quality_score: Optional[float] None class FeedbackRecord(FeedbackIn): sample_id: str Field(default_factorylambda: fsample_{uuid4().hex[:12]}) created_at: datetime Field(default_factorydatetime.utcnow)store.py负责持久化写入# 文件路径app/store.py import json import sqlite3 from pathlib import Path from .schema import FeedbackRecord DB_PATH Path(reflexio.db) def init_db() - None: with sqlite3.connect(DB_PATH) as conn: conn.execute( CREATE TABLE IF NOT EXISTS feedback_samples ( sample_id TEXT PRIMARY KEY, agent_task TEXT, user_input TEXT, agent_output TEXT, tool_calls TEXT, user_feedback TEXT, feedback_type TEXT, quality_score REAL, created_at TEXT ) ) def insert_record(record: FeedbackRecord) - None: with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT INTO feedback_samples (sample_id, agent_task, user_input, agent_output, tool_calls, user_feedback, feedback_type, quality_score, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( record.sample_id, record.agent_task, record.user_input, record.output, json.dumps(record.tool_calls, ensure_asciiFalse), record.user_feedback, record.feedback_type, record.quality_score, record.created_at.isoformat(), ), ) conn.commit()collect.py提供一个简化的 Agent 执行采集装饰器。在实际项目中你可能想在 Agent 的invoke方法里埋点。这里用一个简单版本演示# 文件路径app/collect.py import functools import time from typing import Any from .schema import FeedbackIn, FeedbackRecord from .store import insert_record def track_agent_call(func): 装饰器自动记录Agent一次调用的输入输出与耗时。 functools.wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) latency_ms int((time.time() - start) * 1000) record FeedbackRecord( agent_taskgetattr(func, __name__, unknown_task), user_inputstr(kwargs.get(task) or args[0] if args else ), outputstr(result), tool_calls[ {tool: unknown, latency_ms: latency_ms} ], feedback_typeselfcheck, quality_scoreNone, ) insert_record(record) return result return wrappermain.py负责把 FastAPI 服务和 Agent 调用串联起来# 文件路径app/main.py from fastapi import FastAPI, HTTPException from .schema import FeedbackIn, FeedbackRecord from .store import init_db, insert_record app FastAPI(titleReflexio Demo) app.on_event(startup) def startup() - None: init_db() app.post(/feedback) def record_feedback(payload: FeedbackIn) - dict: if not payload.user_input: raise HTTPException(status_code422, detailuser_input is required) record FeedbackRecord(**payload.model_dump()) insert_record(record) return {status: ok, sample_id: record.sample_id}启动服务需要先初始化环境变量。在.env文件中配置模型服务密钥。如果你使用的是兼容 OpenAI 格式的国内大模型服务则填入相应的base_url和api_key。OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.example.com/v1启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload这个阶段你已经有一个可以接收反馈并入库的服务。你可以用 curl 模拟一条反馈curl -X POST http://localhost:8000/feedback \ -H Content-Type: application/json \ -d { agent_task: generate_report, user_input: 帮我写一份2025年港股医疗器械行业分析报告, output: 报告已生成共8页。, user_feedback: 报告页数太少缺少头部企业深度对比, feedback_type: explicit }如果返回 JSON 中包含sample_id说明写入成功。6. Evaluate如何设计一套可用的自动评估规则采集只是第一步。如果每一条反馈都要人来评估这个系统就只是一个工单系统谈不上自我改进。要让系统能自动给失败样本分类需要加一个评估模块。评估模块的职责是判断一条样本是否属于质量事件以及属于哪类质量问题。评估有两种策略分别适用不同场景6.1 规则评估规则评估适合错误模式固定的场景。这些规则可以快速捕获明显的问题消耗成本低执行速度极快。# 文件路径app/evaluate_rules.py def rule_based_evaluate(feedback: FeedbackIn) - dict: issues [] quality_score 1.0 # 规则1用户要求生成指定页数/长度实际输出明显不足 if 页 in feedback.user_input: # 简单示例如果要求包含“50页”但输出明显不足 if 50页 in feedback.user_input and 50 not in feedback.output: issues.append(completeness_drop) quality_score - 0.5 # 规则2输出为空或报错 if not feedback.output or error in feedback.output.lower(): issues.append(empty_output) quality_score - 0.7 # 规则3用户的反馈中包含模型承认错误的标志 if 对不起 in feedback.user_feedback or 抱歉 in feedback.user_feedback: issues.append(error_admission) return { quality_score: max(0.0, quality_score), issues: issues, needs_review: quality_score 0.6 }规则评估的优点是成本极低能覆盖最频发的错误。缺点是无法覆盖语义层面的质量问题。用户只回复“你写的第二段不对”这种反馈规则就捕捉不到具体错在哪里。6.2 模型评估模型评估适合需要理解语义的场景。拿一条 Agent 的完整输出让评测模型判断是否存在指定类型的质量问题。# 文件路径app/evaluate_model.py import os from openai import OpenAI SYSTEM_PROMPT 你是一个AI智能体质量评估器。你将收到一次Agent执行的记录包括用户输入、Agent输出与用户反馈。 你的任务 1. 判断用户反馈中是否隐含质量投诉。 2. 如果存在质量投诉分类到以下类别 - task_completeness: 任务完成度不足 - tool_usage_error: 工具误用 - hallucination: 幻觉/事实错误 - format_error: 格式不符合要求 - other: 其他 3. 给出0到1之间的质量评分。 只输出JSON不要输出额外文字。 JSON格式 {score: 0.8, error_category: task_completeness, reason: 简要说明} def model_evaluate(feedback: FeedbackIn) - dict: client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) prompt_content f 用户输入 {feedback.user_input} Agent输出 {feedback.output} 调用工具 {feedback.tool_calls} 用户反馈 {feedback.user_feedback} response client.chat.completions.create( modelgpt-4o-mini, temperature0, response_format{type: json_object}, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt_content} ], ) content response.choices[0].message.content return json.loads(content)注意这个模块依赖大模型接口调用生产环境中需要做好超时、重试和限流控制避免评测服务不可用时拖垮主流程。你可以通过一条脚本把规则评估和模型评估串联起来先跑低价规则规则覆盖不了的高风险样本再调用模型。# 文件路径app/evaluate_pipeline.py from .schema import FeedbackIn from .evaluate_rules import rule_based_evaluate from .evaluate_model import model_evaluate def evaluate_feedback(feedback: FeedbackIn) - dict: # 先走规则评估 rule_result rule_based_evaluate(feedback) if rule_result[needs_review] is False and rule_result[quality_score] 0.9: return rule_result # 存在风险时升级到模型评估进行语义分类 try: model_result model_evaluate(feedback) return model_result except Exception as exc: # 模型调用失败时保守处理标记为需要人工确认 return { score: 0.0, error_category: needs_human_review, reason: fmodel evaluate failed: {exc} }这套设计背后的一个重要原则是宁可让样本进入人工审核队列也不要让它直接成为自动优化的输入。自动优化基于错误样本一旦样本本身脏了整个 prompt 优化方向就会出现偏差。7. Improve如何不断优化 Prompt 与评测数据集当评估模块标记了一批质量事件后就可以进入“改进”环节。改进的第一步不是直接修改 prompt而是分析失败样本找出共性问题。在实际项目中可以使用一个 AI 分析助手来生成诊断建议# 文件路径app/improve.py import json import os from openai import OpenAI def analyze_failures(samples: list[dict]) - str: 聚合分析失败样本并输出改进建议。 client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) sample_text json.dumps(samples, ensure_asciiFalse, indent2) response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: 你是一个AI智能体质量改进顾问。根据给出的失败样本列表分析它们的共同模式 并给出改进AI智能体prompt或流程的具体建议。 请按照以下结构输出 1. 共性问题描述 2. 对应失败样本序号 3. prompt改进建议 4. 是否需要新增测试用例 , }, { role: user, content: f失败样本列表{sample_text} } ], temperature0.3 ) return response.choices[0].message.content这一步输出的建议本质上是在帮开发者回答三个问题当前的系统提示词里缺少什么约束边界哪些场景需要新增 few-shot 示例来引导模型是否应该调整 Agent 的工具选择逻辑或者编排流程值得强调的是这个环节的输出并不建议直接进入生产 prompt。更稳妥的做法是先根据分析结果编写新的测试用例。将新测试用例加入回归集。在测试环境中验证新 prompt 的通过率。对比原始 prompt 在同样测试集上的表现。如果通过率提升且没有引入新回归再发布到生产。回顾 Reflexio 的设计哲学它并不想完全取代开发者在优化过程中的地位而是让开发者的每一次修改都更有依据让 AI 智能体的缺陷更快暴露让系统的知识积累不依赖个人记忆。8. 针对不同场景的落地建议Reflexio 这种“生产反馈闭环”的实现层级可以根据团队规模和业务形态灵活调整。现实中并没有一套万能方案需要结合自己的系统场景来判断。如果你是个人开发者或者小团队处于 Agent 的快速验证阶段建议使用最轻量的方式维护一个存放失败样本的 JSONL 文件每次处理反馈时在文件末尾追加优化 prompt 时使用脚本批量统计错误类型。在这一阶段不要花时间搭平台重点是用反馈数据帮助自己做判断。如果你的 Agent 已经有一定的用户量进入灰度或小范围商用阶段这时候需要把反馈采集做厚。至少要做到以下几点在 Agent 调用的主链路增加日志埋点。把用户的每一次显式负面反馈结构化入库。每周对一个时间窗口内的失败样本做一次聚类分析。对自动评测产生的低分样本做人工抽检校准评测标准。如果你的 Agent 服务于企业客户有明确的 SLA 要求那么你需要的是一套更完整的质量运营体系。除了上述功能外还必须把每个失败样本追溯到具体的 Agent 版本、Prompt 版本和模型版本因为任何一方变化都可能影响最终结果。没有版本关联的失败数据很难帮助你快速定位问题。9. 常见问题与排查思路在搭建和使用这套系统时有几个问题出现频率比较高可以参考下表排查问题现象可能原因排查方式解决方案反馈写入失败数据库文件冲突或并发写入问题查看 uvicorn 日志和 SQLite 报错信息生产环境切换为 PostgreSQL避免单文件并发瓶颈模型评估接口超时OpenAI SDK 默认超时时间过短确认模型服务响应耗时在OpenAI客户端设置timeout30并增加重试机制评估结果不准确评测 prompt 的描述过于模糊检查评测 prompt 中分类定义是否清晰引入 few-shot 示例让评测模型按示例输出格式同一个错误反复出现失败样本未回到 Agent 的上下文知识库检查改进流程中是否更新了 prompt 对应的 few-shot 部分将典型失败案例转化为 prompt 或 RAG 中的参考样本正反馈误报为失败规则评估阈值过敏感查看规则的判定条件和截断阈值为规则增加白名单或提高触发阈值除了上述列表有一个容易被忽视的运维问题如果 Agent 调用量很大把每一次执行都完整记录到数据库数据量会快速增长。建议设计保留策略例如只保留最近 90 天的原始样本超过 90 天的样本只保留聚合统计。用户隐私也要提前考虑比如对话中可能包含个人信息或商业敏感信息入库前应做脱敏处理。10. AI 智能体测试的数据集该怎么设计很多开发者在了解 Reflexio 的流程后会追问一个更基础的问题AI 智能体测试的数据集到底该怎么设计生产反馈只是数据集的来源之一真正的数据集需要同时覆盖不同维度。一个合格的 AI 智能体测试数据集应当至少包含以下五类数据第一类是用户需求达成类。用于验证 Agent 是否在正确的边界条件下完成了用户的目标。每个用例应包含原始需求、隐含需求、期望输出结构、验收条件。第二类是工具调用正确性类。用一组需要调用工具的输入来测试 Agent 是否会正确选择工具、构造参数和解析结果。这类用例容易设计也最容易在模型版本升级后发生回归。第三类是上下文理解类。模拟真实的多轮对话场景测试 Agent 能否准确识别指代、省略和默认值。第四类是鲁棒性边界类。包含模糊表达、错别字、超长文本、空输入和诱导性输入用于测试 Agent 的容错能力。第五类是安全与合规类。测试 Agent 是否会在不具备足够信息时给出主观判断是否会泄露提示词中的系统指令是否会绕过权限限制执行工具。生产反馈闭环最重要的价值正是让上述数据集的第五类乃至其余类别的数据能随着系统运行不断地补充和更新。传统的测试集是跨版本长期不变的金标准生产驱动型测试集则是一个能随用户习惯同步演变的活数据集。11. 总结与上手路径建议Reflexio 真正解决的并不是某个单一技术难点而是当前 AI 智能体工程化中的一个系统性问题如何让 Agent 在真实交互中持续变好而不是停留在开发者的实验环境里自嗨。对我们开发者的启示也很明确未来做 AI 智能体应用不会写 prompt 的价值正在降低会用数据驱动的方式维护 prompt 和评测闭环的价值在持续升高。注意热搜词提示“AI智能体开发人才需求大涨 244%”这也印证了市场需要的不是会调用接口的人而是懂得用工程方法让 Agent 稳定产生结果的工程师。如果你想在自己项目里动手尝试我会建议按以下路线执行从本周的开始记录每次 Agent 失败整理成一个 JSONL 文件。为前 50 条错误手动分类寻找错误模式的分布规律。为最常见的前 3 类错误编写自动评测规则。把这 3 类错误转化成新测试用例补进回归集。每周迭代一次 prompt并用回归集验证改进前后效果。完成这些步骤后你会发现自己对 Agent 系统的掌控力明显提升。你不会再凭感觉改 prompt而是知道每一条修改想解决哪个真实问题以及它在历史失败样本上是否真的有效。这种工作方式的转变比任何框架和工具都更有长期价值。这个基于 Reflexio 思路的最小闭环本身也很值得收藏备用因为它在不同 Agent 项目里都能复用。