当前位置：首页 > news >正文

RAG 反馈回路设计：从用户点赞、点踩信号中自动优化检索策略

news 2026/7/12 17:23:54

RAG 反馈回路设计：从用户点赞、点踩信号中自动优化检索策略

一、深度引言与场景痛点

大家好，我是赵咕咕。

很多团队把 RAG 系统上线就完事了。数据向量化、索引建好、Prompt 模版嵌入——一套流水线跑通，看着搜索结果返回觉得不错，然后就再也没管过。

但现实是：检索质量会随时间退化。

新的文档加入，旧文档过时，用户问法在变化。你上线第一周精准的检索，到了第三周可能就偏了。更糟糕的是，你完全不知道偏了多少，因为没有反馈信号。

今天我们聊聊如何用最廉价的信号——用户点赞和点踩——来构建一个自动化的检索质量优化回路。

二、底层机制与原理深度剖析

2.1 反馈回路的核心思想

RAG 系统的反馈回路本质上是一个"信号采集 → 分析 → 策略调整 → 验证"的闭环：

flowchart LR A[用户提问] --> B[检索与生成] B --> C[展示结果] C --> D{用户反馈} D -->|👍 点赞| E[正反馈收集] D -->|👎 点踩| F[负反馈收集] D -->|无操作| G[忽略/默认反馈] E --> H[反馈分析引擎] F --> H H --> I{发现问题?} I -->|检索质量差| J[调整检索策略] I -->|文档过时| K[触发重新索引] I -->|Query 理解偏差| L[更新 Query 改写逻辑] J --> M[AB 验证] K --> M L --> M M --> A

这个回路的核心价值在于：它不用人工标注数据就能持续驱动优化。用户的每一次交互，都在为你提供"训练信号"。

2.2 反馈信号的分类与加权

并不是所有反馈都等值。点踩有两种可能：一种是"检索结果根本不对"，另一种是"答案对但不全"。我们需要区分处理：

显式反馈：用户直接点了赞/踩按钮。这是最强信号。

隐式反馈：用户复制了答案内容（正反馈）、立刻重新提问（可能在否定当前结果）、快速关闭页面（可能是满意，也可能是失望——需要结合上下文判断）。

信号衰减：一个反馈的价值会随时间递减。三天前的点赞对今天的检索策略调整，影响力应该小于昨天的点踩。

三、生产级代码实现

3.1 反馈采集与存储

import asyncio import json from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Any class FeedbackType(Enum): THUMBS_UP = "thumbs_up" THUMBS_DOWN = "thumbs_down" COPY = "copy" # 用户复制了答案 REWRITE = "rewrite" # 用户立刻重新提问 @dataclass class FeedbackRecord: query: str retrieved_docs: list[str] generated_answer: str feedback_type: FeedbackType timestamp: float = field(default_factory=lambda: datetime.now().timestamp()) user_id: str | None = None session_id: str | None = None metadata: dict[str, Any] = field(default_factory=dict) class FeedbackCollector: """异步反馈收集器，写入 Redis 或数据库。""" def __init__(self, redis_client, ttl_days: int = 30): self._redis = redis_client self._ttl = timedelta(days=ttl_days) async def record(self, feedback: FeedbackRecord) -> None: key = f"rag:feedback:{feedback.session_id}:{feedback.timestamp}" data = { "query": feedback.query, "feedback_type": feedback.feedback_type.value, "timestamp": feedback.timestamp, "user_id": feedback.user_id, "session_id": feedback.session_id, "retrieved_docs_count": len(feedback.retrieved_docs), } await self._redis.setex( key, int(self._ttl.total_seconds()), json.dumps(data, ensure_ascii=False) ) async def get_recent_negative( self, hours: int = 24 ) -> list[dict[str, Any]]: """获取最近 N 小时的负面反馈用于分析。""" pattern = "rag:feedback:*" cursor = 0 results = [] cutoff = (datetime.now() - timedelta(hours=hours)).timestamp() while True: cursor, keys = await self._redis.scan( cursor, match=pattern, count=100 ) for key in keys: raw = await self._redis.get(key) if raw: record = json.loads(raw) if record["timestamp"] > cutoff: results.append(record) if cursor == 0: break return results async def get_feedback_ratio( self, hours: int = 24 ) -> tuple[int, int, float]: """计算近 N 小时的点赞/点踩比例。""" recent = await self.get_recent_negative(hours) # 这里简化处理，实际应扫全量 ups = sum(1 for r in recent if r["feedback_type"] == "thumbs_up") downs = sum(1 for r in recent if r["feedback_type"] == "thumbs_down") total = ups + downs ratio = ups / total if total > 0 else 1.0 return ups, downs, ratio

3.2 根据反馈自动调整检索策略

from dataclasses import dataclass @dataclass class RetrievalConfig: top_k: int = 5 similarity_threshold: float = 0.7 enable_hyde: bool = True enable_rerank: bool = True query_rewrite_prompt: str = "请将用户问题改写为更精确的搜索查询: {query}" class AdaptiveRetrievalManager: """根据反馈信号动态调整检索策略。""" def __init__( self, collector: FeedbackCollector, config: RetrievalConfig | None = None, ): self.collector = collector self.config = config or RetrievalConfig() self._adjustment_log: list[dict] = [] async def evaluate_and_adjust(self) -> RetrievalConfig: """评估近期反馈质量，必要时调整策略。""" ups, downs, ratio = await self.collector.get_feedback_ratio(hours=1) new_config = RetrievalConfig( top_k=self.config.top_k, similarity_threshold=self.config.similarity_threshold, enable_hyde=self.config.enable_hyde, enable_rerank=self.config.enable_rerank, query_rewrite_prompt=self.config.query_rewrite_prompt, ) if ratio < 0.6 and downs > 10: # 差评率高 → 说明检索质量不行 # 策略 1：提高 top_k，让 LLM 看到更多文档 new_config.top_k = min(self.config.top_k + 5, 20) # 策略 2：降低阈值，扩大召回 new_config.similarity_threshold = max( self.config.similarity_threshold - 0.1, 0.4 ) self._adjustment_log.append({ "reason": f"差评率 {ratio:.2%}, 扩大召回", "old_top_k": self.config.top_k, "new_top_k": new_config.top_k, }) elif ratio > 0.85 and ups > 50: # 好评率高 → 可以尝试收紧召回、降低成本和延迟 new_config.top_k = max(self.config.top_k - 2, 3) new_config.similarity_threshold = min( self.config.similarity_threshold + 0.05, 0.9 ) self.config = new_config return new_config