RAG 检索链路中相似度阈值不稳定导致召回波动的工程化调参体系

大家好，今天我们来深入探讨一下在检索增强生成 (RAG) 系统中，如何解决由于相似度阈值不稳定而导致的召回波动问题，并建立一套可行的工程化调参体系。

RAG 系统通过从外部知识库检索相关信息，并将其融入到生成过程中，从而增强模型的知识储备和生成质量。检索环节的质量直接影响着后续生成效果，而相似度阈值作为检索环节的关键参数，其稳定性和调优至关重要。

一、问题定义：相似度阈值不稳定与召回波动

在 RAG 系统中，我们通常使用向量相似度来衡量查询 (query) 与知识库文档 (document) 之间的相关性。一个预先设定的相似度阈值决定了哪些文档会被召回，并传递给生成模型。

然而，实际应用中，由于以下几个原因，相似度阈值的设置往往面临挑战：

数据分布不均： 知识库中的文档质量参差不齐，向量表示的质量也因此各异。某些主题的文档向量可能整体相似度偏高，而另一些主题则偏低。
查询意图多样性： 用户的查询意图千差万别，有些查询表达明确，容易找到相关文档；而有些查询较为模糊，导致相似度分数普遍偏低。
向量模型偏差： 不同的向量模型（例如 Sentence Transformers, OpenAI Embeddings）在向量空间中的分布特性不同，对相似度分数的绝对值和相对差异产生影响。
动态变化： 知识库的内容会随着时间推移而更新，导致文档向量的分布发生变化，原先设定的阈值可能不再适用。

这些因素会导致在固定的相似度阈值下，召回结果出现波动。例如，对于某些查询，可能召回过多不相关的文档 (高召回率，低准确率)；而对于另一些查询，则可能漏掉重要的相关文档 (低召回率，高准确率)。

二、影响召回波动的原因分析

为了更好地解决这个问题，我们需要深入分析导致召回波动的原因。

原因	表现	影响
知识库文档质量差异	部分文档内容冗余、噪声多，向量表示质量差；部分文档信息丰富、表达清晰，向量表示质量高。	低质量文档即使与查询相关，相似度也可能较低，导致漏召；高质量文档即使与查询无关，相似度也可能较高，导致误召。
查询语句表达模糊	查询语句过于宽泛、缺少关键信息，导致向量表示难以准确捕捉用户意图。	相似度分数普遍偏低，难以区分相关和不相关文档，容易导致低召回率。
向量模型特性差异	不同的向量模型在向量空间中的分布特性不同，相似度分数的绝对值和相对差异也不同。例如，有些模型倾向于给出较高的相似度分数，而另一些模型则倾向于给出较低的分数。	使用不同的向量模型，即使是相同的查询和文档，相似度分数也会有所不同，导致需要调整阈值。
向量检索方法选择	不同的向量检索方法 (例如暴力搜索、HNSW、IVF) 在效率和精度之间有所权衡。一些方法可能牺牲精度来提高检索速度，导致召回结果不准确。	向量检索方法的选择直接影响着召回结果的准确性，不合适的检索方法可能导致召回波动。
数据更新导致分布变化	知识库的内容会随着时间推移而更新，导致文档向量的分布发生变化。新加入的文档可能与原有文档的分布存在差异，从而影响相似度计算的结果。	原先设定的阈值可能不再适用，需要进行调整。

三、工程化调参体系设计

为了解决上述问题，我们需要建立一套工程化的调参体系，该体系应具备以下特点：

可观测性： 能够实时监控召回指标 (例如召回率、准确率)，并对相似度分布进行可视化分析。
可配置性： 能够灵活调整相似度阈值，并支持多种调参策略。
自动化： 能够根据数据变化自动调整阈值，降低人工干预成本。
可解释性： 能够解释阈值调整的原因，并提供调整建议。

下面我们将详细介绍该体系的各个组成部分。

1. 数据预处理与向量化

数据预处理是向量化的前提，直接影响着向量表示的质量。我们需要对知识库文档进行清洗、去噪、分词等处理。例如，可以使用以下 Python 代码进行文本清洗：

import re
from nltk.corpus import stopwords
from nltk.stem import PorterStemmer

def clean_text(text):
    """
    文本清洗函数，包括去除 HTML 标签、标点符号、停用词、词干提取等。
    """
    text = re.sub(r'<[^>]+>', '', text)  # 去除 HTML 标签
    text = re.sub(r'[^ws]', '', text)  # 去除标点符号
    text = text.lower()  # 转换为小写

    stop_words = set(stopwords.words('english'))
    words = text.split()
    words = [w for w in words if not w in stop_words]  # 去除停用词

    stemmer = PorterStemmer()
    words = [stemmer.stem(w) for w in words]  # 词干提取

    return ' '.join(words)

# 示例
text = "<p>This is an example <b>text</b> with some <i>HTML</i> tags and punctuation.</p>"
cleaned_text = clean_text(text)
print(f"原始文本：{text}")
print(f"清洗后的文本：{cleaned_text}")

然后，选择合适的向量模型，将清洗后的文本转换为向量表示。例如，可以使用 Sentence Transformers：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-mpnet-base-v2') # 选择合适的模型

def embed_text(text):
    """
    将文本转换为向量表示
    """
    embeddings = model.encode(text)
    return embeddings

# 示例
text = "This is a sample sentence."
embeddings = embed_text(text)
print(f"文本：{text}")
print(f"向量维度：{embeddings.shape}")

2. 向量检索与相似度计算

选择合适的向量检索方法，例如 HNSW (Hierarchical Navigable Small World) 算法，可以在保证检索速度的同时，尽可能提高检索精度。可以使用 Faiss 库来实现 HNSW 索引：

import faiss
import numpy as np

def build_index(embeddings, dimension):
    """
    构建 Faiss HNSW 索引
    """
    index = faiss.IndexHNSWFlat(dimension, 32)  # 32 是 M 值，控制连接数
    index.init_level_memory()
    index.add(embeddings)
    return index

def search_index(index, query_embedding, top_k=5):
    """
    在 Faiss 索引中搜索
    """
    D, I = index.search(query_embedding.reshape(1, -1), top_k)  # D 是距离，I 是索引
    return D, I

# 示例
dimension = 768  # Sentence Transformers all-mpnet-base-v2 的向量维度
num_vectors = 1000
embeddings = np.random.rand(num_vectors, dimension).astype('float32')  # 模拟 embeddings

index = build_index(embeddings, dimension)

query_embedding = np.random.rand(dimension).astype('float32')  # 模拟查询向量
D, I = search_index(index, query_embedding)

print(f"相似度分数：{D}")
print(f"检索到的索引：{I}")

3. 相似度阈值调优策略

固定阈值： 这是最简单的策略，设定一个固定的阈值，例如 0.7。但如前所述，这种策略容易受到数据分布不均的影响。
自适应阈值： 根据查询的特点，动态调整阈值。例如，可以根据查询的长度、关键词数量等特征来调整阈值。
百分比阈值： 选取相似度分数最高的 Top N 个文档，无论其绝对分数如何。这种策略适用于需要保证一定召回率的场景。
基于统计的阈值： 统计一定时间窗口内查询的相似度分布，并根据分布的统计特征 (例如均值、标准差) 来动态调整阈值。例如，可以将阈值设置为均值减去一个标准差。
强化学习： 将阈值调整问题建模为一个强化学习问题，通过与环境交互，学习最优的阈值调整策略。

下面是一个使用百分比阈值的 Python 示例：

def apply_percentage_threshold(similarity_scores, top_percentage=0.1):
  """
  应用百分比阈值，选取相似度分数最高的 top_percentage 的文档。

  Args:
    similarity_scores: 相似度分数列表。
    top_percentage: 百分比阈值，例如 0.1 表示选取前 10% 的文档。

  Returns:
    被选中的文档的索引列表。
  """
  num_docs = len(similarity_scores)
  num_to_select = int(num_docs * top_percentage)
  if num_to_select == 0 and num_docs > 0:
      num_to_select = 1 # 至少选择一个文档

  # 获取排序后的索引
  sorted_indices = np.argsort(similarity_scores)[::-1] # 从大到小排序

  # 选择 top N 个索引
  selected_indices = sorted_indices[:num_to_select]

  return selected_indices

# 示例
similarity_scores = np.random.rand(100)  # 模拟相似度分数
selected_indices = apply_percentage_threshold(similarity_scores, top_percentage=0.2)
print(f"选中的文档索引：{selected_indices}")

4. 监控与评估

我们需要建立一套完善的监控体系，实时监控召回指标 (例如召回率、准确率) 和相似度分布。可以使用以下指标来评估检索效果：

召回率 (Recall)： 相关文档被正确召回的比例。
准确率 (Precision)： 召回的文档中，相关文档的比例。
F1 值： 召回率和准确率的调和平均数，综合衡量检索效果。
NDCG (Normalized Discounted Cumulative Gain)： 考虑了文档相关性等级的排序指标。

可以使用以下 Python 代码来计算召回率和准确率：

def calculate_recall_precision(relevant_docs, retrieved_docs):
    """
    计算召回率和准确率。

    Args:
        relevant_docs: 相关的文档索引列表。
        retrieved_docs: 检索到的文档索引列表。

    Returns:
        召回率和准确率。
    """
    relevant_set = set(relevant_docs)
    retrieved_set = set(retrieved_docs)

    true_positives = len(relevant_set.intersection(retrieved_set))
    recall = true_positives / len(relevant_set) if len(relevant_set) > 0 else 0
    precision = true_positives / len(retrieved_set) if len(retrieved_set) > 0 else 0

    return recall, precision

# 示例
relevant_docs = [1, 3, 5, 7, 9]
retrieved_docs = [1, 2, 3, 4, 5]

recall, precision = calculate_recall_precision(relevant_docs, retrieved_docs)
print(f"召回率：{recall}")
print(f"准确率：{precision}")

同时，我们需要对相似度分布进行可视化分析，例如绘制直方图或箱线图，以便了解相似度分数的整体情况。可以使用 Matplotlib 库来进行可视化：

import matplotlib.pyplot as plt

def visualize_similarity_scores(similarity_scores):
    """
    可视化相似度分数分布。

    Args:
        similarity_scores: 相似度分数列表。
    """
    plt.hist(similarity_scores, bins=20)  # 绘制直方图，分成 20 个 bins
    plt.xlabel("相似度分数")
    plt.ylabel("文档数量")
    plt.title("相似度分数分布")
    plt.show()

# 示例
similarity_scores = np.random.rand(1000)  # 模拟相似度分数
visualize_similarity_scores(similarity_scores)

5. 自动化调优

在监控和评估的基础上，我们可以进一步实现自动化调优。例如，可以根据召回率和准确率的变化趋势，自动调整相似度阈值。可以使用以下策略：

PID 控制： 将召回率或准确率作为控制目标，使用 PID 控制器自动调整阈值。
贝叶斯优化： 使用贝叶斯优化算法，寻找最优的阈值组合，以最大化 F1 值或 NDCG。
A/B 测试： 同时运行多个不同的阈值策略，并根据实际效果选择最优的策略。

四、一个完整的工程化调参流程示例

数据准备： 收集知识库文档，并进行清洗和向量化。
初始阈值设定： 根据经验或初步实验，设定一个初始的相似度阈值。
在线监控： 实时监控召回率、准确率和相似度分布。
问题诊断： 如果发现召回率或准确率出现明显波动，则需要进一步诊断问题。例如，可以分析导致波动的原因，并检查数据质量或向量模型是否存在问题。
阈值调整： 根据诊断结果，选择合适的阈值调整策略，并进行调整。
效果评估： 调整阈值后，需要重新评估召回率和准确率，以确认调整是否有效。
迭代优化： 不断重复上述步骤，持续优化阈值，以达到最佳的检索效果。

五、代码整合示例

以下代码整合了上述部分功能，展示了一个简化的 RAG 检索链路调参示例：

import re
import numpy as np
import faiss
import matplotlib.pyplot as plt
from sentence_transformers import SentenceTransformer
from nltk.corpus import stopwords
from nltk.stem import PorterStemmer

# 1. 数据预处理与向量化
def clean_text(text):
    text = re.sub(r'<[^>]+>', '', text)
    text = re.sub(r'[^ws]', '', text)
    text = text.lower()
    stop_words = set(stopwords.words('english'))
    words = text.split()
    words = [w for w in words if not w in stop_words]
    stemmer = PorterStemmer()
    words = [stemmer.stem(w) for w in words]
    return ' '.join(words)

model = SentenceTransformer('all-mpnet-base-v2')

def embed_text(text):
    embeddings = model.encode(text)
    return embeddings

# 2. 向量检索与相似度计算
def build_index(embeddings, dimension):
    index = faiss.IndexHNSWFlat(dimension, 32)
    index.init_level_memory()
    index.add(embeddings)
    return index

def search_index(index, query_embedding, top_k=5):
    D, I = index.search(query_embedding.reshape(1, -1), top_k)
    return D, I

# 3. 相似度阈值调优策略 - 固定阈值
def apply_fixed_threshold(similarity_scores, threshold=0.7):
    selected_indices = np.where(similarity_scores >= threshold)[0]
    return selected_indices

# 4. 监控与评估
def calculate_recall_precision(relevant_docs, retrieved_docs):
    relevant_set = set(relevant_docs)
    retrieved_set = set(retrieved_docs)
    true_positives = len(relevant_set.intersection(retrieved_set))
    recall = true_positives / len(relevant_set) if len(relevant_set) > 0 else 0
    precision = true_positives / len(retrieved_set) if len(retrieved_set) > 0 else 0
    return recall, precision

def visualize_similarity_scores(similarity_scores):
    plt.hist(similarity_scores, bins=20)
    plt.xlabel("相似度分数")
    plt.ylabel("文档数量")
    plt.title("相似度分数分布")
    plt.show()

# 模拟数据
documents = [
    "This is document 1 about cats.",
    "Document 2 discusses dogs.",
    "Cats and dogs are common pets in document 3.",
    "Document 4 talks about birds.",
    "This document 5 is about the weather."
]
cleaned_documents = [clean_text(doc) for doc in documents]
embeddings = np.array([embed_text(doc) for doc in cleaned_documents]).astype('float32')
dimension = embeddings.shape[1]
index = build_index(embeddings, dimension)

# 模拟查询
query = "Tell me about cats."
cleaned_query = clean_text(query)
query_embedding = embed_text(cleaned_query).astype('float32')

# 检索
D, I = search_index(index, query_embedding)
similarity_scores = D[0]
retrieved_indices = I[0]

# 应用固定阈值
selected_indices = apply_fixed_threshold(similarity_scores, threshold=0.7)

# 评估
relevant_docs = [0, 2]  # 假设文档 0 和 2 与查询相关
retrieved_docs = retrieved_indices[selected_indices]
recall, precision = calculate_recall_precision(relevant_docs, retrieved_docs)

print(f"检索到的文档索引：{retrieved_docs}")
print(f"召回率：{recall}")
print(f"准确率：{precision}")

# 可视化
visualize_similarity_scores(similarity_scores)

六、未来方向

结合上下文的阈值调整： 考虑查询的上下文信息 (例如用户历史行为、查询意图)，更加精准地调整阈值。
多模态信息融合： 融合文本、图像、音频等多模态信息，提高相似度计算的准确性。
可解释性阈值调整： 提供阈值调整的原因和依据，增强系统的透明度和可信度。

工程调优是持续的过程

检索链路的工程调优不是一蹴而就的过程，需要持续地监控、评估和优化。通过建立完善的工程化调参体系，我们可以有效地解决相似度阈值不稳定导致的召回波动问题，提升 RAG 系统的整体性能。

希望今天的分享对大家有所帮助。

阈值不稳定是常见挑战，工程调优是关键

RAG 系统中的相似度阈值不稳定是常见的工程挑战，需要通过建立可观测、可配置、自动化的调参体系来解决。持续的监控、评估和优化是提升 RAG 系统性能的关键。

RAG 检索链路中相似度阈值不稳定导致召回波动的工程化调参体系

发表回复 取消回复

发表回复取消回复