RAG 中，单靠向量或稀疏检索等粗筛手段，难以应对生产级高精度要求，且多路混合检索的分数存在打分尺度不统一等局限。为此，RAG 通常引入重排序精排模型作为过滤屏障。本文从介绍重排模型的基本概念与基础使用，到多源数据合并、动态截断算法以及字符预算控制等核心流水线的设计与实现。

重排序模型（Reranker）的概念

为什么需要重排序？（两阶段检索架构）

在知识库或商品库检索中，系统需要平衡速度与精度：

• 第一阶段：粗筛召回（Retrieval）
  • 职责：侧重召回率，从海量候选（如数万至数百万）中快速筛选出潜在相关的数十个候选文档。
  • 手段：使用 Embedding 向量检索或传统的 BM25 稀疏检索。
  • 特点：速度快（毫秒级，向量可提前离线入库），但可能存在语义泛化，对型号、否定词等具体细节不够敏感。
• 第二阶段：精排重排序（Reranking）
  • 职责：侧重准确率，对粗筛返回的候选文档进行细粒度的语义打分，过滤无关文档并进行精确排序。
  • 手段：使用重排序模型（如 qwen3-rerank）。
  • 特点：语义匹配更精准，但由于需要将查询与候选文档拼接后进行实时在线打分，计算开销较大（几十到上百毫秒），因此通常只处理第一阶段初筛出的少量候选（一般在 50 个以内）。

向量检索与重排模型的对比

从应用开发的视角，两者的核心指标和应用特征对比如下：

评估维度	向量检索 (Embedding)	重排模型 (Reranker)
工作方式	双方各自离线生成向量，在线检索时只进行向量相似度计算	将查询与文档拼接输入，由模型直接实时评估两者的相关性评分
计算速度	极快（毫秒级），支持大规模高并发检索	较慢（几十毫秒），计算耗时随候选文档数量线性增加
检索精度	一般，容易受常见词重叠影响而泛化，可能丢失具体细节	较高，能识别逻辑转折、关键型号、否定修饰等语义关系
分值可比性	跨查询可比性低，绝对相似度易随模型和语料分布漂移，难设固定阈值	跨查询可比性高，分数通常直接反映语义关联的置信度
适用范围	粗筛召回：从海量文档中快速定位一个候选子集	精排过滤：对初筛子集进行二次打分，以便进行硬卡点或截断

Reranker 的基础使用

输入与输出规范

不同于 Embedding 模型接收单段文本并返回高维向量，Reranker 模型：

• 输入：一个 (Query, Document) 的二元组对列表。
• 输出：一个对应的标量相关性分数列表。
• 分数数值特性：Cross-Encoder 模型输出的原始相关性打分通常是一个无界的 logit 值。一般在经过 Sigmoid 函数变换后，被压缩映射到 0 到 1 之间，用来代表该 Document 相对于 Query 的相关性置信度度量（Confidence Score）。

基础使用与异步封装示例

在 Python 中调用重排模型（例如阿里云的 qwen3-rerank）时，如果直接使用同步调用，在并发 RAG 应用中会阻塞 asyncio 的事件循环。因此，我们可以使用 asyncio.to_thread 将其封装为异步客户端类。

下面是基于 DashScope SDK 的通用重排序客户端异步封装逻辑：

import asyncio
from http import HTTPStatus
from typing import List
import dashscope

class DashScopeRerankerClient:
    """
    通用 DashScope 重排序模型异步客户端包装器
    """
    def __init__(self, api_key: str, model: str = "qwen3-rerank"):
        self.api_key = api_key
        self.model = model

    async def rerank_documents(self, query: str, documents: List[str]) -> List[float]:
        """
        异步调用 DashScope Rerank 模型评估文档相关性。
        """
        if not documents:
            return []

        # 使用 asyncio.to_thread 将同步 SDK 调用包装进线程池异步执行
        response = await asyncio.to_thread(
            dashscope.TextReRank.call,
            model=self.model,
            query=query,
            documents=documents,
            api_key=self.api_key
        )

        if response.status_code != HTTPStatus.OK:
            raise RuntimeError(f"Rerank API 调用失败: {response.message}")

        # 获取打分并按输入 documents 的原始索引顺序对齐
        scores = [0.0] * len(documents)
        for result in response.output.results:
            idx = result.index
            if 0 <= idx < len(documents):
                scores[idx] = float(result.relevance_score)

        return scores

适用场景与工程痛点

适用场景

• 多路召回融合后的过滤：当系统从稠密检索、稀疏检索、Web 检索等多条链路中召回了候选文档，由于各路原始评分的计算逻辑和数值范围不同，通过 Reranker 对所有候选进行统一尺度下的二次打分。
• 过滤语义泛化的无关文档：在商品匹配或精密文献检索场景中，过滤由于词语重叠而被向量数据库赋予高分的低相关性切片。

为什么不能只用简单的 Top-K 截断？

常见的 RAG 系统在重排序后，如果只采用固定的 Top-K 截断，可能会面临以下局限：

1. 噪声引入：当用户的提问在知识库中没有相关内容时，粗筛检索仍会返回候选，固定 Top-K 截断会将无关文档送给大模型。
2. 相关文档的遗漏：如果某次查询中高度相关的文档数量多于 K 篇，固定截断会导致部分关键上下文流失。
3. 低相关文档混入：如果候选文档中仅有少数几篇相关，其余均为低相关性文档，固定取 Top-K 会引入不必要的长尾噪声。

因此，为了解决这些痛点，在 Reranker 排序后，需要设计绝对阈值卡点与动态截断算法。

通用重排流水线设计

为了在实际工程中解耦并保持模块设计的通用性，我们可以将“多源数据合并”、“动态截断”和“大模型上下文预算控制”整合设计为一个通用的重排处理流程。

多源候选文档合并

在实际检索链路中，经常需要把来自本地知识库和外部网络等多条渠道的数据合并到一起进行统一重排。

以下为通用的多源数据规范化合并函数：

from typing import List, Dict, Any

def merge_and_normalize_docs(local_docs: List[Dict[str, Any]], web_docs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    合并本地和网页等多源检索结果，统一映射为包含标题、内容、来源等通用字段的字典列表
    """
    normalized_docs = []

    # 1. 抽取并规范化本地文档
    for doc in local_docs:
        normalized_docs.append({
            "title": doc.get("title") or doc.get("item_name") or "",
            "content": doc.get("content") or "",
            "source": "local",
            "url": None
        })

    # 2. 抽取并规范化外部网页文档
    for doc in web_docs:
        normalized_docs.append({
            "title": doc.get("title") or "",
            "content": doc.get("snippet") or doc.get("content") or "",
            "source": "web",
            "url": doc.get("url")
        })

    return normalized_docs

截断算法

核心判定逻辑

截断的核心逻辑是：如果对相关性分数降序排列，在某一处出现了分数跌幅明显加大，那么该位置之后的所有文档大概率是低相关性的噪声。

为了防止过度截断并控制上下文长度，算法设定了四个控制参数：

1. RERANK_MAX_TOPK = 8：最多保留 8 篇文档，控制大模型输入长度。
2. RERANK_MIN_TOPK = 3：在发生分数下跌时，也保留至少 3 篇文档，避免因过度拦截而丢失必要信息。
3. RERANK_GAP_ABS = 0.35（绝对差门限）：若相邻两个文档的分数差值达到 0.35 或以上，触发截断。
4. RERANK_GAP_RATIO = 0.25（相对比例门限）：若相邻两个文档的分数降幅达到 25% 或以上，触发截断，主要用于防止在低分段发生大幅度相对降幅。

当相邻两个文档的分数满足“绝对差值达到 0.35”或“相对降幅达到 25%”的任一条件时，系统即判定发生了分值骤降，并在该位置进行截断，舍弃后续的所有候选文档。

阈值调优指南与实践建议

重排阶段的截断阈值并没有一成不变的行业标准，通常取决于所使用的具体重排序模型与业务场景对精准/召回的权衡。以下是几项实践建议：

1. 分值归一化是前提：不同重排模型的原始得分范围差异很大。例如某些 Cross-Encoder 模型输出无界的原始 logits（如 -10 到 10）。在应用绝对差和相对降幅截断前，建议通过 Sigmoid 函数 将得分压缩映射到 [0, 1] 之间，否则固定的阈值判定将失效。
2. 参数推荐范围（在分数归一化至 [0, 1] 下）：
   • 最大篇数（Max Top-K）：推荐设在 5 到 8 之间。由于大模型存在“Lost in the Middle”（易忽略上下文中间部分）的缺陷，且长文本会显著增加延时与 Token 成本，塞入过多的文档反而会降低生成质量。
   • 最小篇数（Min Top-K）：推荐设在 2 到 3 之间。这属于保守策略，即使后方的分数降幅较大，也保留最相关的 2~3 篇，以防下游大模型缺少基本上下文。
   • 绝对差值（GAP_ABS）：推荐设在 0.3 到 0.4 之间。在归一化的分值体系中，0.3 的差值代表了较显著的语义相关度下降；若设为 0.5 则可能偏大，导致过滤不够敏感。
   • 相对降幅（GAP_RATIO）：推荐设在 0.20 到 0.30 之间。这主要用于拦截在低分段发生的较大幅度相对降幅（例如从 0.45 跌到 0.32，绝对差仅 0.13，但相对降幅已达 28.8%），能够过滤不相关的长尾文档。

代码实现

下面是执行断崖式截断的通用纯函数逻辑：

def truncate_by_score_cliff(
    ranked_docs: List[Dict[str, Any]],
    min_top_k: int = 3,
    max_top_k: int = 8,
    gap_abs: float = 0.35,
    gap_ratio: float = 0.25
) -> List[Dict[str, Any]]:
    """
    根据相邻文档重排分数的降幅动态截断低相关性文档
    :param ranked_docs: 已按分数降序排列且带有 'score' 键的文档列表
    :return: 截断后的文档列表
    """
    if not ranked_docs:
        return []

    n = len(ranked_docs)
    upper_bound = min(max_top_k, n)
    lower_bound = min(min_top_k, upper_bound)
    
    cutoff_index = upper_bound

    # 扫描从下限到上限之间的可能断崖位置
    for i in range(lower_bound - 1, upper_bound - 1):
        s_curr = ranked_docs[i].get("score", 0.0)
        s_next = ranked_docs[i + 1].get("score", 0.0)

        delta_abs = s_curr - s_next
        delta_ratio = delta_abs / s_curr if s_curr > 0 else 0.0

        # 双重阈值门控判断
        if delta_abs >= gap_abs or delta_ratio >= gap_ratio:
            cutoff_index = i + 1
            break

    return ranked_docs[:cutoff_index]

运行场景演算示例

假设重排模型打分排序后得出以下五个文档的相关性得分：

• 案例 A（无显著降幅）

  • 分数序列：[0.92, 0.88, 0.85, 0.81, 0.78]
  • 演算：从第 3 个文档（分数 0.85）开始与第 4 个文档（分数 0.81）对比，绝对差值为 0.04（小于 0.35），相对降幅为 4.7%（小于 25%），未触发截断。
  • 最终结果：全部保留（不触发截断，默认受最大数量限制）。

• 案例 B（高分段分值骤降）

  • 分数序列：[0.95, 0.92, 0.88, 0.45, 0.20]
  • 演算：从第 3 个（分数 0.88）和第 4 个（分数 0.45）对比，绝对差值为 0.43（达到 0.35 的阈值），且相对降幅达到了 48.8%（达到 25% 的阈值），触发截断。
  • 最终结果：cutoff_index = 3，仅保留前 3 个文档 [0.95, 0.92, 0.88]，后续文档被过滤。

• 案例 C（低分段绝对差值小但相对降幅大）

  • 分数序列：[0.85, 0.70, 0.40, 0.28, 0.26]
  • 演算：从第 3 个（分数 0.40）和第 4 个（分数 0.28）对比，绝对差值为 0.12（小于 0.35），但相对降幅达到了 30%（达到 25% 的阈值），触发截断。
  • 最终结果：cutoff_index = 3，仅保留前 3 个，过滤掉低相关性的长尾文档。

下游生成字符预算截断控制

尽管精排和断崖式截断在“篇数”维度（Top-K）上对文档进行了过滤，但在下游大模型的答案生成节点中，仍面临文本长度溢出的风险（例如某些网页内容过于冗长，或多轮历史对话累加导致超出大模型的上下文窗口上限）。

为了在字符长度层面提供安全保障，可以在生成阶段引入基于字符预算的累加截断机制：

• 设定最大上下文预算门限：MAX_CONTEXT_CHARS = 12000。
• 按相关度高低扣减：逐篇格式化重排后的候选文档，如果当前累加的文档字符长度超出剩余预算，则从该处截断，舍弃后续相关度相对较低的文档。

字符预算格式化代码实现

from typing import Tuple

def format_docs_with_char_budget(documents: List[Dict[str, Any]], char_budget: int) -> Tuple[str, int]:
    """
    格式化重排序后的文档，在不超过字符预算的前提下拼装为大模型上下文
    :param documents: 经过重排和断崖截断后的文档列表
    :param char_budget: 字符预算上限
    :return: (格式化后的文本, 剩余可用字符数)
    """
    formatted_lines = []
    used_chars = 0
    
    for idx, doc in enumerate(documents, start=1):
        content = doc.get("content") or ""
        meta_tags = [f"[{idx}]"]
        
        for field, template in [
            ("source", "[source={}]"),
            ("url", "[url={}]"),
            ("title", "[title={}]"),
        ]:
            field_value = str(doc.get(field)).strip() if doc.get(field) is not None else ""
            if field_value and field_value != "None":
                meta_tags.append(template.format(field_value))

        relevance_score = doc.get("score")
        if relevance_score is not None:
            meta_tags.append(f"[score={float(relevance_score):.4f}]")

        doc_entry = " ".join(meta_tags) + "\n" + content

        # 如果加入此切片会导致超出预算，则舍弃此切片及后续切片
        if used_chars + len(doc_entry) > char_budget:
            break

        formatted_lines.append(doc_entry)
        used_chars += len(doc_entry) + 2  # 加上换行符开销

    formatted_context = "\n\n".join(formatted_lines) if formatted_lines else "无参考文档。"
    return formatted_context, char_budget - used_chars

完整端到端测试 Demo

为展示以上模块如何在实际的问答流水线中组合工作，我们使用了一个通用的旅游检索与重排作为演示用例：

import asyncio

async def main():
    # 模拟从不同源初筛出来的文档
    local_knowledge = [
        {
            "title": "北京旅行避坑指南",
            "content": "故宫门票需要提前7天在线实名预约，周一闭馆。建议从午门进，神武门出。"
        },
        {
            "title": "闲聊杂记",
            "content": "今天中午和朋友去胡同里吃了炸酱面，老北京味道确实很地道。"
        }
    ]

    web_search_results = [
        {
            "title": "2026年北京故宫游览全攻略",
            "url": "https://example.com/gugong-travel",
            "snippet": "游览故宫必须提前预约门票，旺季门票紧张，务必提前关注放票时间。刷身份证入园。"
        },
        {
            "title": "数码新品发布资讯",
            "url": "https://example.com/tech-news",
            "snippet": "某知名科技厂商发布全新主动降噪蓝牙耳机，音质表现优异。"
        }
    ]

    # 1. 多源文档合并与格式标准化
    raw_docs = merge_and_normalize_docs(local_knowledge, web_search_results)
    
    # 2. 定义 Query 并模拟重排模型打分 (实际应用中应调用 DashScopeRerankerClient)
    query = "如何预约故宫门票以及游览路线推荐？"
    
    # 模拟 Reranker 模型对四篇文档的打分结果：故宫相关得分高，无关内容得分低
    mock_scores = [0.912, 0.183, 0.885, 0.114]
    
    # 将打分回填并按分数降序排序
    ranked_docs = [{**doc, "score": score} for doc, score in zip(raw_docs, mock_scores)]
    ranked_docs.sort(key=lambda x: x["score"], reverse=True)

    print("=== 重排与排序结果 ===")
    for idx, d in enumerate(ranked_docs):
        print(f"Top {idx+1}: [{d['source']}] {d['title']} (得分: {d['score']:.4f})")

    # 3. 动态断崖式截断
    # 设定：最少保留2条，最多保留8条，发生 0.35 分数差即截断
    cutoff_docs = truncate_by_score_cliff(
        ranked_docs,
        min_top_k=2,
        max_top_k=8,
        gap_abs=0.35
    )

    print("\n=== 断崖式截断后保留结果 ===")
    for idx, d in enumerate(cutoff_docs):
        print(f"保留项 {idx+1}: [{d['source']}] {d['title']} (得分: {d['score']:.4f})")

    # 4. 生成阶段的字符预算控制 (假定总字符预算限制为 500)
    context_text, remaining_budget = format_docs_with_char_budget(cutoff_docs, char_budget=500)
    
    print("\n=== 送入大语言模型的最终上下文 ===")
    print(context_text)
    print(f"\n剩余字符预算: {remaining_budget}")

if __name__ == "__main__":
    asyncio.run(main())

通过这种规范化的接口设计，重排序模块降低了与特定检索框架的耦合度，具有较好的可扩展性，便于在不同 RAG 系统中复用。

总结

重排序模型（Reranker）通过 Cross-Encoder 架构，为 RAG 检索链路提供了相关性二次过滤手段。在检索系统构建中，为避免固定 Top-K 截断带来的局限性，可以通过以下方式进行优化：

1. 多源渠道的输入规范化与格式映射。
2. 结合绝对差与相对比例的双重截断算法。
3. 下游生成节点的字符预算控制，防范上下文长度超出大模型限制。

这些方法的结合使用，有助于在保证服务可用性的前提下，过滤不相关的长尾文档，为下游大模型提供高相关性的上下文。