RAG 进阶之路5：Milvus 中实现“时间衰减”

在构建 RAG系统时，我们经常会遇到这样一个尴尬的场景：

用户问：“最新的 iPhone 摄像头参数怎么样？”
你的 RAG 系统兴致勃勃地从数据库里找出了 iPhone 11 的评测文章，因为那篇文章写得非常详尽，和问题的语义相似度（Semantic Similarity）极高。而关于 iPhone 12 的介绍可能比较简短，导致向量距离稍远，被排在了后面。

这就是向量检索的“时效性盲区”。

在 HNSW 或 IVF 这样的向量索引眼里，只有空间距离，没有时间概念。数据一旦写入，它在向量空间的位置就是永恒的。但在新闻、金融或日志分析等场景中，“新”往往比“准”更重要。

那么，在使用 Milvus 或 Faiss 时，我们如何引入时间衰减（Time Decay），让越新的数据得分越高？本文将为你揭示三种成熟的工程化解决方案。

为什么没有 `decay=0.5` 参数？

很多开发者第一次使用 Milvus 时，都在寻找类似 search(..., time_decay=True) 的参数，结果一无所获。

为什么原生不支持？
本质原因是索引的不可变性与查询效率的权衡。向量索引（如 HNSW 图）是基于几何距离构建的静态结构。如果引入动态的时间权重，意味着每次查询时，所有数据点在“综合打分空间”里的位置都会发生变化。这会导致索引失效，查询退化为全表扫描（Brute-force），这对于亿级数据是不可接受的。

因此，时间衰减必须在“索引之上”或“检索之后”通过工程手段实现。

策略一：扩大召回 + 外部重排

这是目前业界最通用、副作用最小的方案。

核心逻辑：
不要指望向量数据库一次性给你最终的 Top-K。让数据库多返回一些数据（Top-N），然后在你的应用层代码中，结合“向量分”和“时间分”重新排序。

算法公式

最终得分是内容相似度与时效性权重的乘积。

它的计算公式为：最终得分 = 原始向量相似度 × 时间衰减函数。

参数详解：

原始向量相似度： 衡量查询与文档在语义上的匹配程度，通常采用取值范围在 0 到 1 之间的余弦相似度（Cosine Similarity）。
时间差： 即“当前时间”减去“文档发布时间”，用于量化文档的新旧程度。
衰减函数： 用于对旧文档进行降权处理。常用的算法模型包括反比例函数或指数衰减函数 。

代码实战 (Python + Milvus)

import time
import math

def search_with_decay(collection, query_vector, top_k=10, decay_rate=0.001):
    # 1. 扩大召回：我们需要 Top-10，但向 Milvus 要 Top-100
    # 关键点：必须在 output_fields 中带回 timestamp
    results = collection.search(
        data=[query_vector],
        anns_field="embedding",
        param={"metric_type": "L2", "params": {"nprobe": 10}},
        limit=top_k * 10,  # 扩大 10 倍
        output_fields=["timestamp"] 
    )
    
    reranked_results = []
    now = time.time()
    
    for hit in results[0]:
        # 获取原始分数和时间
        original_score = hit.score  # 假设是内积或余弦，越大越好
        doc_time = hit.entity.get("timestamp")
        
        # 计算时间差 (单位转换成天或小时可能更直观)
        delta_seconds = max(0, now - doc_time)
        delta_days = delta_seconds / 86400
        
        # 2. 应用衰减公式 (这里使用简单的反比例函数)
        # decay_rate 越大，旧数据死得越快
        time_factor = 1.0 / (1.0 + decay_rate * delta_days)
        
        # 3. 计算混合分数
        new_score = original_score * time_factor
        
        reranked_results.append({
            "id": hit.id,
            "new_score": new_score,
            "original_score": original_score,
            "time_factor": time_factor
        })
    
    # 4. 重新排序并截取 Top-K
    reranked_results.sort(key=lambda x: x["new_score"], reverse=True)
    return reranked_results[:top_k]

优点： 灵活性极高，可以在代码中随意调整衰减曲线，不需要重建索引。
缺点： 增加了网络传输开销（fetch 100 条只用 10 条）。

策略二：使用迭代器防止内存爆炸

如果你需要极强的时间相关性，导致必须召回 10,000 条数据才能找到一条“既相关又新”的数据，策略一的 limit=10000 可能会撑爆内存或阻塞网络。

Milvus 2.3+ 引入了 Search Iterator (搜索迭代器)，这是处理大规模重排的神器。

核心逻辑：
像游标一样，分批次（Batch）从 Milvus 拉取数据。在客户端一边拉取，一边计算衰减分数。当发现“经过衰减后的最高理论分数”都已经低于“当前结果集的最低分”时，提前终止迭代。

这种方法类似于推荐系统中的 WAND 算法思想，能够极大地节省资源。

策略三：分区搜索与混合过滤

如果你不需要平滑的衰减曲线，而是更关注业务规则（例如：“优先看最近 3 个月的，如果没结果再看之前的”），那么使用 Milvus 的 Partition（分区） 功能效率最高。

实现方案：

数据入库时： 按月创建 Partition，例如 part_2023_10, part_2023_11。
检索时：
- Step 1: 指定 partition_names=["part_2023_11"] 进行搜索。
- Step 2: 检查结果的分数是否达标（如 score > 0.8）。
- Step 3: 如果不达标，再扩大范围搜索旧的分区。

或者直接使用标量过滤（Scalar Filtering）：

# 只看最近 30 天的数据
expr = f"timestamp > {time.time() - 30*86400}"
collection.search(..., expr=expr)

优点： 性能极致，完全利用数据库侧的能力。
缺点： 这是“硬截断”，不是“软衰减”。可能会漏掉一条 31 天前但极其精准的数据。

衰减函数的艺术：如何选择衰减因子

实现时间衰减，最难的不是代码，而是调参。

你需要问业务方一个问题：“一条昨天的新闻，其价值是一周前新闻的几倍？”

线性衰减 (Linear):
- 太硬了，容易减成负数，不推荐。
指数衰减 (Exponential):
- 特点： 最初衰减极快，随后变缓。适合新闻资讯（昨天的新闻是旧闻，上个月和上上个月的区别反倒不大了）。
高斯/反比例衰减:
- 特点： 比较平滑，适合技术文档或长尾知识库。

总结

在Milvus 中实现时间衰减，本质上是在“向量相似度”（语义相关性）和“时间戳”（新鲜度）之间寻找平衡。

对于大多数 RAG 应用，推荐使用 策略一（扩大召回 + 客户端重排）。它开发成本最低，且足以应对 90% 的场景。
对于海量数据流（如推特流、日志流），请研究 策略二（迭代器）。
永远不要试图把时间戳强行编码进向量里（Embedding Injection），除非你是算法专家且有无限的算力来重新训练模型。

解决了“时效性”问题后，我们可能会遇到另一个极端：为了追求语义匹配，我们将文档切得太碎，导致模型看懂了片段却看不懂全貌。下一篇，我们将探讨如何通过 Small-to-Big (父文档检索) 策略来解决这个问题。

为什么没有 decay=0.5 参数？