检索增强生成(RAG)系统评估实战：如何解决信息陈旧问题

1次阅读

没有评论

共计 2644 个字符，预计需要花费 7 分钟才能阅读完成。

检索增强生成（Retrieval-Augmented Generation, RAG）系统通过结合检索模块和生成模块，实现更准确的信息生成。其核心流程分为三步：

检索阶段：根据用户查询从知识库中检索相关文档
增强阶段：将检索结果与原始查询结合形成增强上下文
生成阶段：语言模型基于增强上下文生成最终响应

信息陈旧会导致系统检索到过时内容，进而影响生成结果的准确性。例如在金融、医疗等领域，使用过时信息可能导致严重后果。

我们曾遇到几个典型场景：

金融领域：使用旧的监管政策回答合规问题，导致错误建议
科技领域：基于过时的 API 文档生成代码示例，无法正常运行
医疗领域：引用已被推翻的研究结论，产生误导性建议

这些问题本质上是因为传统 RAG 系统将知识库视为静态数据源，缺乏对信息时效性的考量。

我们提出两阶段解决方案：

为每篇文档添加时间戳 metadata
设计时间衰减函数计算文档时效得分
将时效得分与语义相似度结合得到最终检索分数

常用衰减函数包括：

线性衰减：score = base_score * (1 - decay_rate * age)
指数衰减：score = base_score * exp(-decay_rate * age)
阶梯衰减：按时间区间分段设置权重

定期扫描知识库，识别过期文档
自动触发更新流程获取最新内容
版本控制确保更新原子性
灰度发布验证新内容效果

以下是带时间权重的检索模块核心代码：

import numpy as np
from datetime import datetime

class TimeAwareRetriever:
    def __init__(self, decay_rate=0.1, decay_type='exponential'):
        """
        初始化时间感知检索器
        :param decay_rate: 衰减系数，控制时间权重下降速度
        :param decay_type: 衰减类型，支持 linear/exponential/step
        """
        self.decay_rate = decay_rate
        self.decay_type = decay_type

    def calculate_time_decay(self, doc_date, current_date=None):
        """
        计算文档的时间衰减因子
        :param doc_date: 文档发布日期(datetime 对象)
        :param current_date: 当前日期，默认为系统当前时间
        :return: 时间衰减系数(0- 1 之间)
        """
        current_date = current_date or datetime.now()
        age_days = (current_date - doc_date).days

        if self.decay_type == 'linear':
            return max(0, 1 - self.decay_rate * age_days)
        elif self.decay_type == 'exponential':
            return np.exp(-self.decay_rate * age_days)
        elif self.decay_type == 'step':
            if age_days < 30: return 1.0
            elif age_days < 90: return 0.7
            else: return 0.3
        else:
            raise ValueError(f"Unknown decay type: {self.decay_type}")

    def retrieve(self, query, documents, top_k=3):
        """
        执行带时间权重的检索
        :param query: 查询文本
        :param documents: 文档列表，每个文档需包含 text 和 publish_date 字段
        :param top_k: 返回的文档数量
        :return: 排序后的文档列表
        """
        # 1. 计算语义相似度(简化示例，实际可用 sentence-transformers 等库)
        semantic_scores = [self._calc_semantic_score(query, doc['text']) 
                          for doc in documents]

        # 2. 计算时间衰减因子
        time_weights = [self.calculate_time_decay(doc['publish_date'])
                       for doc in documents]

        # 3. 综合得分 = 语义相似度 * 时间权重
        combined_scores = [semantic * time for semantic, time 
                          in zip(semantic_scores, time_weights)]

        # 4. 按综合得分排序并返回 top_k
        sorted_indices = np.argsort(combined_scores)[-top_k:][::-1]
        return [documents[i] for i in sorted_indices]

    def _calc_semantic_score(self, query, text):
        """简化的语义相似度计算(实际项目应使用预训练模型)"""
        # 这里应替换为真实的语义相似度计算
        return len(set(query.split()) & set(text.split())) / len(query.split())

我们在三个领域数据集上对比了优化前后的效果：