Agent 反馈闭环：用户纠错如何反向优化 Prompt 和工具链

一、用户反复纠正同一个错误，Agent 却永远学不会

Agent 上线两周后，客服团队反馈了 50 个类似问题。
"Agent 把退货地址搞错了"出现了 15 次。
"搜索结果的格式不是用户想要的"出现了 20 次。
每次用户纠正，Agent 当下改对了。
下次换个用户问类似问题，它又犯同样的错。

问题在于这些纠正信息是一次性的。
修改完当前对话就丢弃了，没有沉淀为系统记忆。
反馈闭环就是把这些纠正信息结构化存储。
下一次类似场景自动应用修正。

这是 Agent 产品中最常见的"记忆断层"问题。
传统的 Chatbot 每次对话是独立的——上一轮的纠正不会影响下一轮的行为。
Agent 比 Chatbot 多了工具调用和推理能力，但在学习能力上并没有本质提升。
它仍然是一个无状态的服务——每次 Prompt + 工具调用 → 输出，不会因为"上次用户纠正了退货地址"而在下次自动修正。

这种"记忆力缺失"背后是架构问题，不是模型问题。
模型有能力理解反馈并在当前对话中修正，但修正结果被丢弃在了对话历史中——下一轮全新的 System Prompt 里没有这些修正信息。
所以反馈闭环要解决的不是"如何让模型理解反馈"，而是"如何把一次修正确认为系统级的规则"。

二、三层反馈闭环架构

flowchart TB
    A[用户纠错反馈] --> B{反馈类型}
    B -->|短期| C[即时修正层]
    B -->|中期| D[Few-shot 示例库]
    B -->|长期| E[Prompt 模板优化]

    C --> F[当前对话中纠正输出]
    D --> G[存入示例库，相似场景自动引用]
    E --> H[定期批量更新 System Prompt]

    G --> I[相似度匹配选最优示例]
    I --> F

    H --> J[统计高频纠错 → 更新规则]
    J --> G

三层的时间尺度不同：即时修正是秒级（当前对话），Few-shot 注入是分钟级（相似场景触发），Prompt 优化是天/周级（定期统计）。
这三层不应该互相替代，而应该同时运行——即时修正解决当下问题，Few-shot 防止短期内同类错误，Prompt 优化从根本上减少某类错误的产生。

Few-shot 示例库的核心挑战不在存储，而在相似度匹配。
用简单的关键词匹配（如代码中的 find_similar 方法）只能覆盖显式关联——比如同样提到"地址"的场景。
但用户可能用不同的措辞表达同一个问题："地址不对"和"你发到哪里去了"在语义上指向同一个错误类型。
生产环境应该用 Embedding 做语义相似度匹配，而不是关键词匹配。

另外，Few-shot 示例不是越多越好。
注入 3 个高度相关的示例，效果远好于注入 10 个泛泛相关的示例。
示例库的去重和质量筛选是反馈闭环中最容易被忽略的工程工作。

三、反馈闭环的 Python 实现

下面的代码实现了一个三层反馈闭环：即时纠正记录、Few-shot 示例匹配和 Prompt 自动优化建议。

"""
feedback_loop.py - Agent 反馈闭环
"""

import hashlib
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from collections import defaultdict

logger = logging.getLogger(__name__)


@dataclass
class Correction:
    """单条纠正记录"""
    id: str
    original_output: str      # Agent 原始输出
    corrected_output: str     # 用户纠正后的输出
    context: str              # 对话上下文
    feedback_type: str        # 错误类型
    timestamp: float


class FeedbackCollector:
    """反馈收集器"""

    def __init__(self, max_examples: int = 100):
        self.max_examples = max_examples
        self.corrections: List[Correction] = []
        self.error_stats: Dict[str, int] = defaultdict(int)

    def record(
        self,
        original: str,
        corrected: str,
        context: str,
        feedback_type: str = "general",
    ):
        """记录一次纠正"""
        correction = Correction(
            id=hashlib.md5(
                f"{original}{corrected}".encode()
            ).hexdigest()[:8],
            original_output=original,
            corrected_output=corrected,
            context=context,
            feedback_type=feedback_type,
            timestamp=datetime.now().timestamp(),
        )

        # 去重
        for c in self.corrections:
            if c.original_output == original and c.corrected_output == corrected:
                return  # 已有相同纠正

        self.corrections.append(correction)
        self.error_stats[feedback_type] += 1

        # 限制最大示例数
        if len(self.corrections) > self.max_examples:
            # 保留最新的
            self.corrections = self.corrections[-self.max_examples:]

    def find_similar(
        self, current_context: str, top_k: int = 3
    ) -> List[Correction]:
        """
        根据当前上下文找相似的纠正示例
        用于 Few-shot Prompt 构建
        """
        # 简化：关键词匹配
        scores = []
        context_lower = current_context.lower()

        for c in self.corrections:
            score = 0
            for word in context_lower.split():
                if word in c.context.lower():
                    score += 1
            if score > 0:
                scores.append((score, c))

        scores.sort(key=lambda x: x[0], reverse=True)
        return [c for _, c in scores[:top_k]]

    def build_few_shot_prompt(self, current_context: str) -> str:
        """构建 Few-shot 示例"""
        examples = self.find_similar(current_context)
        if not examples:
            return ""

        lines = ["## 历史纠正示例（请参考这些修正）:"]
        for i, ex in enumerate(examples):
            lines.append(f"\n示例 {i+1}:")
            lines.append(f"原始输出: {ex.original_output}")
            lines.append(f"纠正后: {ex.corrected_output}")

        return "\n".join(lines)

    def analyze_patterns(self) -> Dict:
        """分析高频错误模式，生成 Prompt 优化建议"""
        patterns = {}

        # 统计错误类型
        patterns["top_errors"] = sorted(
            self.error_stats.items(),
            key=lambda x: x[1],
            reverse=True,
        )[:5]

        # 检查是否有重复修正模式
        if self.corrections:
            corrections_by_type = defaultdict(list)
            for c in self.corrections:
                corrections_by_type[c.feedback_type].append(c)

            for ftype, corrections in corrections_by_type.items():
                if len(corrections) >= 3:
                    patterns[f"pattern_{ftype}"] = {
                        "count": len(corrections),
                        "suggestion": self._suggest_fix(ftype, corrections),
                    }

        return patterns

    def _suggest_fix(
        self, ftype: str, corrections: List[Correction]
    ) -> str:
        """针对错误类型建议 Prompt 修复"""
        suggestions = {
            "format_error": (
                "在 System Prompt 中添加：始终使用 Markdown 格式输出，"
                "列表使用 - 开头"
            ),
            "wrong_address": (
                "在工具调用前添加地址校验步骤，"
                "确认地址存在于地址库中"
            ),
            "missing_info": (
                "在回复前检查是否包含必要信息：名称、时间、地点"
            ),
        }
        return suggestions.get(
            ftype,
            f"累计 {len(corrections)} 次类似错误，建议人工分析",
        )


class PromptOptimizer:
    """Prompt 自动优化器"""

    def __init__(self, feedback_collector: FeedbackCollector):
        self.feedback = feedback_collector

    def optimize(
        self, current_prompt: str
    ) -> Tuple[str, str]:
        """
        根据反馈数据优化 System Prompt
        返回: (优化后的 Prompt, 优化说明)
        """
        patterns = self.feedback.analyze_patterns()
        suggestions = []
        optimized = current_prompt

        for key, info in patterns.items():
            if key.startswith("pattern_") and "suggestion" in info:
                suggestion = info["suggestion"]
                if suggestion not in optimized:
                    optimized += f"\n\n## 优化规则: {suggestion}"
                    suggestions.append(suggestion)

        return optimized, "\n".join(suggestions) if suggestions else "无需优化"


# ---- Agent 中集成反馈 ----

class AgentWithFeedback:
    """带反馈闭环的 Agent"""

    def __init__(self, system_prompt: str):
        self.system_prompt = system_prompt
        self.feedback = FeedbackCollector()
        self.optimizer = PromptOptimizer(self.feedback)

    def respond(self, user_input: str, context: str = "") -> str:
        """生成回复（注入 Few-shot 示例）"""
        # 构建增强 Prompt
        enhanced_prompt = self.system_prompt

        # 注入历史纠正示例
        few_shot = self.feedback.build_few_shot_prompt(
            context + " " + user_input
        )
        if few_shot:
            enhanced_prompt += "\n\n" + few_shot

        # 调用 LLM（实际项目）
        response = call_llm(enhanced_prompt, user_input)
        return response

    def collect_feedback(
        self,
        original_output: str,
        corrected_output: str,
        context: str,
    ):
        """收集用户纠正反馈"""
        # 自动识别错误类型
        ftype = self._classify_error(original_output, corrected_output)
        self.feedback.record(
            original_output, corrected_output, context, ftype
        )

        # 检查是否需要更新 Prompt
        stats = self.feedback.error_stats
        if max(stats.values(), default=0) >= 5:
            logger.warning("累计纠正已达阈值，建议执行 Prompt 优化")

    def optimize_periodically(self):
        """定期优化（如每周触发）"""
        new_prompt, changes = self.optimizer.optimize(self.system_prompt)
        if changes != "无需优化":
            logger.info(f"Prompt 优化: {changes}")
            self.system_prompt = new_prompt

    def _classify_error(
        self, original: str, corrected: str
    ) -> str:
        """自动分类错误类型"""
        if len(original) != len(corrected):
            return "format_error"
        if "地址" in original or "地址" in corrected:
            return "wrong_address"
        if len(corrected) > len(original) * 1.5:
            return "missing_info"
        return "general"

FeedbackCollector 的去重逻辑不容忽视。
如果同一个纠正被重复记录（不同用户的相似反馈），示例库会膨胀，Few-shot Prompt 也会越来越长。
MD5 去重可以保证相同内容的纠正只存一份，但对于"相似但不完全相同"的纠正（比如两次"地址错了"的具体地址不同），需要额外的语义去重策略。

四、反馈闭环的成本考量

Few-shot 示例会增加 Prompt Token 消耗。
每个示例约 100-300 Token。
建议限制最多引用 3 个示例。
Token 增加约 10-20%，但准确率提升 5-15%。

自动错误分类可能不准。
_classify_error 的简单规则覆盖不了复杂场景。
建议先人工标注一段时间，积累数据后再做自动分类。

还有一个容易被忽略的成本：Prompt 自动优化的风险。
每次自动修改 System Prompt 都有可能引入新的问题——比如优化了一条规则但和已有的规则冲突。
建议 Prompt 优化生成的是"建议"而非自动应用——由人工 Review 后手动执行，而不是让系统自动替换 System Prompt。
在你有了足够的 A/B 测试数据证明自动优化的效果后，再考虑自动化。

另外，反馈数据本身的价值随时间衰减。
三个月前的纠正可能对应一个已经被修复的问题——但如果问题修复了，对应的 Few-shot 示例还留在库里，反而会误导模型。
建议反馈库设置 TTL（如 30 天），过期示例自动清理，确保示例库始终反映的是"最近的有效反馈"。

五、总结

Agent 反馈闭环让用户纠错沉淀为系统能力。
三层机制：即时修正、Few-shot 示例库、Prompt 定期优化。
相似度匹配保证示例的相关性。
Token 成本增加约 10-20%，但可显著提升准确率。
周期性 Prompt 优化防止同类错误反复出现。

落地优先级：先做反馈收集（记录每次纠正），再做 Few-shot 注入（相似场景复用），最后做自动 Prompt 优化（人工审核）。
不要一上来就追求全自动——反馈闭环的难点不在技术实现，而在如何保证反馈数据的质量和时效性。