知识库查询 —— 重排序节点

本文档详细介绍知识库查询流程中的重排序节点（rerank）的设计与实现。该节点使用 Reranker 模型对 RRF 融合结果和网络搜索结果进行精排，并通过断崖检测算法实现动态 TopK 截断。

学习理念：重排序是 RAG 流程的"质检环节"——RRF 融合后的结果还是基于粗略的排名分数，Reranker 用交叉编码器对每个(query, doc)对深度交互打分，把真正相关的文档排到前面。核心思路：召回阶段求"全"（多路检索），精排阶段求"准"（交叉编码器）。

海外对标：RerankNode 的"BGE-Reranker 精排 + 断崖检测动态截断"方案对标 Cohere Rerank API、LangChain 的 ContextualCompressionRetriever、以及 LlamaIndex 的 SentenceTransformerRerank。断崖检测算法在 Elasticsearch 的 rescore 阶段也有类似实现（window_size 动态截断）。

本节 AI 替代率：~80% | 人工干预率：~20%

角色	能力范围
🤖 AI 擅长	FlagReranker 调用、多源文档合并、compute_score 批量打分、排序逻辑、断崖检测算法代码、测试代码
👤 人类需理解	断崖检测阈值调优（gap_abs=0.5 / gap_ratio=0.25 的取值含义）、min_topk/max_topk 的设定策略

---

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习 Reranker / 断崖检测
🟢	§2 核心概念	~90%	~10%	Bi-Encoder vs Cross-Encoder / BGE-Reranker / 断崖检测
🟡	§3 整体流程	~90%	~10%	节点位置 + 输入输出
🟠	§4 分步实现	~75%	~25%	Step5(Reranker打分) + Step6(断崖检测) 是核心
🔴	§4.4 主代码	~70%	~30%	RerankNode + _rerank + _cliff_cutoff
🟢	§5 测试运行	~95%	~5%	看预期输出的断崖截断效果
🟡	§6 总结	~90%	~10%	设计要点回顾

---

技术栈健康度标签体系

技术	健康度	建议
BGE-Reranker-Large	🔥 巅峰	中英双语重排序模型，560M 参数。基于 XLM-RoBERTa。FlagReranker + compute_score 是标准 API。
Cross-Encoder	🔥 巅峰	交叉编码器架构，Query+Doc 联合编码。比 Bi-Encoder（双塔）精度高但速度慢，适合精排阶段。
断崖检测 (Cliff Detection)	🟢 稳定	基于相邻得分差的动态截断算法。gap_abs + gap_ratio 双阈值设计成熟可靠。
FlagEmbedding	🔥 巅峰	BAAI 开源的嵌入和重排序工具库，BGE 系列模型的官方推理框架。

体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。

---

中英文对照表

English	中文	本质
Bi-Encoder	双塔模型	Query 和 Document 独立编码，速度快但精度一般，适合召回阶段
Cross-Encoder	交叉编码器	Query 和 Document 联合编码，精度高但速度慢，适合精排阶段
Reranker	重排序器	使用交叉编码器对候选文档重新打分的模型
Cliff Detection	断崖检测	寻找相邻得分断崖式下跌的位置进行动态截断
Dynamic TopK	动态截断	根据得分分布自动决定保留多少文档
Score Normalization	得分归一化	不同模型得分尺度不一致时的标准化处理
Graceful Degradation	优雅降级	模型失败时返回原序，确保系统仍可运行
Multi-Source Merging	多源合并	将本地和网络搜索的结果统一为相同文档格式

---

💡 程序员比喻

• RerankNode 就像 Code Review 的第二轮——第一轮（RRF）是 CI 自动检查（所有人提的 PR 都有份），第二轮（Reranker）是资深工程师逐行精读（交叉编码器）。
• Bi-Encoder vs Cross-Encoder 就像 git diff --stat vs git diff——前者只看改了哪些文件（独立编码），后者逐行看 diff 内容（交互编码）。
• 断崖检测 就像 git log --oneline 的 commit message 质量——前几个 commit 写得好好的，突然来了个"fix bug"，直接从这里截断。
• 多源文档统一格式 就像 docker-compose.yml 的 services 定义——不管是用什么语言写的服务，统一成 image:port:env 格式。
• 降级处理 就像 try/except 包着 model.predict()——模型挂了就返回 None score，系统照样 work，只是精度降了。

---

1. 任务目标

本节课将实现知识库查询流程中的 重排序节点（rerank）。

该节点使用 Reranker 模型 对 RRF 融合结果和网络搜索结果进行精排，并通过 断崖检测算法 实现动态 TopK 截断，筛选出最相关的文档用于最终答案生成。

学完本节你将掌握：

• Reranker 模型的工作原理（交叉编码器 vs 双塔模型）
• BGE-Reranker 模型的使用方法
• 断崖检测（Cliff Detection）动态截断算法
• 多源文档格式统一与合并技巧

---

2. 核心概念扫盲

2.1 为什么需要重排序？

RRF 融合虽然合并了多路检索结果，但存在局限性：

重排序的作用：

• 使用专门的相关性模型进行精排
• 统一评估所有来源的文档（本地 + 网络）
• 过滤低质量文档，提高答案生成质量

2.2 双塔模型 vs 交叉编码器

检索系统中有两种主流的文本匹配架构：

双塔模型（Bi-Encoder）

    Query                Document
      │                     │
      v                     v
  ┌───────┐            ┌───────┐
  │Encoder│            │Encoder│
  └───────┘            └───────┘
      │                     │
      v                     v
  [向量 Q]              [向量 D]
      │                     │
      └──────────┬──────────┘
                 │
                 v
            相似度计算
           (余弦/点积)

特点：

• Query 和 Document 独立编码
• Document 向量可预计算存储
• 速度快，适合召回阶段
• 交互信息有限，精度一般

交叉编码器（Cross-Encoder）

    Query + Document（拼接输入）
              │
              v
    ┌─────────────────────┐
    │   Transformer       │
    │   (BERT/RoBERTa)    │
    │                     │
    │  Q 和 D 深度交互     │
    └─────────────────────┘
              │
              v
         相关性得分

特点：

• Query 和 Document 联合编码
• 充分捕获交互信息
• 精度高，适合精排阶段
• 速度慢，无法预计算

Reranker 模型就是交叉编码器！

2.3 BGE-Reranker 模型

我们使用的 BGE-Reranker-Large 是智源研究院开源的中英双语重排序模型：

🟡 【P1 看注释就行】 Step1 获取查询——优先 rewritten_query，回退 original_query。

from FlagEmbedding import FlagReranker

reranker = FlagReranker(
    model_name_or_path="BAAI/bge-reranker-large",
    device="cuda",      # GPU 加速
    use_fp16=True       # 半精度推理
)

# 计算相关性得分
pairs = [
    ["什么是万用表？", "万用表是一种测量电压、电流、电阻的仪器"],
    ["什么是万用表？", "今天天气很好"]
]
scores = reranker.compute_score(pairs)
# 输出: [0.9234, 0.0156]  高分 = 高相关

模型特性：

特性	说明
基座模型	XLM-RoBERTa-Large
参数量	560M
输入长度	最大 512 tokens
输出	相关性分数（越高越相关）
支持语言	中英双语

2.4 断崖检测算法

重排序后需要决定保留多少文档。传统做法是固定 TopK，但这不够灵活：

固定 TopK=5 的问题：

情况 1：前 3 篇高度相关，后 2 篇噪声
得分: [0.95, 0.92, 0.88, 0.12, 0.08]
                        ↑
                  应该在这里截断

情况 2：前 7 篇都相关
得分: [0.95, 0.91, 0.87, 0.83, 0.79, 0.75, 0.71]
                              ↑
                        固定截断会丢失有价值内容

断崖检测的思路： 寻找得分"断崖式下跌"的位置，在那里截断。

断崖检测示例：

得分: [0.95, 0.92, 0.88, 0.12, 0.08]
差值:      0.03   0.04   0.76   0.04
                         ↑
                   断崖！在此截断

结果: 保留前 3 篇

断崖检测公式：

🟡 【P1 看注释就行】 Step2 合并本地 RRF 结果——提取 content / chunk_id / title，标记 source="local"。跳过无效 content。

# 相邻得分差
gap = score[i] - score[i+1]

# 相对下降比例
rel = gap / (abs(score[i]) + 1e-6)

# 满足任一条件即为断崖
if gap >= gap_abs or rel >= gap_ratio:
    cutoff_at = i + 1

参数说明：

• gap_abs：绝对差值阈值（默认 0.5）
• gap_ratio：相对比例阈值（默认 0.25）
• min_topk：最少保留数量（默认 3）
• max_topk：最多保留数量（默认 10）

---

3. 重排序业务处理流程（总）

3.1 节点在流程中的位置

                    multi_search
                         │
        ┌────────────────┼────────────────┬────────────────┐
        │                │                │                │
        v                v                v                v
  search_embedding  search_hyde     query_kg        web_search
        │                │                │                │
        └────────────────┴────────────────┘                │
                         │                                  │
                         v                                  │
                        rrf  ───────────────────────────────│
                         │                                  │
                         │              ┌───────────────────┘
                         │              │
                         v              v
                    rrf_chunks    web_search_docs
                         │              │
                         └──────┬───────┘
                                │
                                v
                    ┌───────────────────────┐
                    │     ★ rerank ★        │
                    │                       │
                    │  1. 合并多源文档       │
                    │  2. Reranker 精排     │
                    │  3. 断崖检测截断       │
                    └───────────────────────┘
                                │
                                v
                          reranked_docs
                                │
                                v
                          answer_output

3.2 节点输入输出

┌─────────────────────────────────────────────────────────────┐
│                        RerankNode                            │
├─────────────────────────────────────────────────────────────┤
│  输入:                                                       │
│    state["rrf_chunks"]        # RRF 融合后的本地文档          │
│    state["web_search_docs"]   # 网络搜索结果                  │
│    state["rewritten_query"]   # 重写后的查询（或原始查询）     │
│                                                              │
│  输出:                                                       │
│    state["reranked_docs"]     # 重排序后的文档列表            │
│      - text: 文档内容                                        │
│      - score: 相关性得分                                     │
│      - source: 来源（local/web）                             │
│      - chunk_id: 本地文档 ID                                 │
│      - url: 网络文档链接                                     │
��─────────────────────────────────────────────────────────────┘

---

4. 重排序业务处理流程（分）

4.1 目标

实现一个重排序节点，将 RRF 融合结果和网络搜索结果统一精排，并通过断崖检测动态筛选最相关的文档。

4.2 需求分析

需求项	说明
多源合并	合并本地 RRF 结果和网络搜索结果
格式统一	不同来源的文档统一为相同结构
精确排序	使用 Reranker 模型计算相关性得分
动态截断	通过断崖检测自动确定保留数量
降级处理	Reranker 失败时返回原序
来源追溯	保留文档来源标识（local/web）

4.3 实现流程

4.3.1 实现流程图

┌─────────────────────────────────────────────────────────────┐
│                        RerankNode                            │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 1: 获取查询文本                                         │
│   - 优先使用 rewritten_query                                 │
│   - 降级使用 original_query                                  │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 2: 合并本地 RRF 结果                                    │
│   - 遍历 rrf_chunks                                          │
│   - 提取 content、chunk_id、title                            │
│   - 标记 source = "local"                                    │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 3: 合并网络搜索结果                                     │
│   - 遍历 web_search_docs                                     │
│   - 提取 snippet、url、title                                 │
│   - 标记 source = "web"                                      │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 4: 构建 Query-Document 对                               │
│   - 为每个文档创建 [query, text] 对                          │
│   - 准备输入 Reranker 模型                                   │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 5: Reranker 计算得分                                    │
│   - 调用 FlagReranker.compute_score()                        │
│   - 为每个文档添加 score 字段                                 │
│   - 按得分降序排列                                            │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 6: 断崖检测动态截断                                     │
│   - 从 min_topk 位置开始检测                                 │
│   - 计算相邻得分差和相对比例                                  │
│   - 发现断崖立即截断                                          │
└─────────────────────────────────────────────────────────────┘
                              │
                              v
┌─────────────────────────────────────────────────────────────┐
│ Step 7: 返回重排序结果                                       │
│   - 记录日志（处理前后数量）                                  │
│   - 返回 reranked_docs                                       │
└─────────────────────────────────────────────────────────────┘

4.3.2 具体实现步骤

Step 1: 获取查询文本

从状态中获取用于重排序的查询文本：

🟡 【P1 看注释就行】 Step3 合并网络搜索结果——提取 snippet，标记 source="web"。注意降级 content。

question = state.get("rewritten_query") or state.get("original_query", "")

为什么优先使用 rewritten_query？

原始查询: "这块主板怎么修？"
重写查询: "主板维修方法和常见故障排查步骤"

重写后的查询更完整，与文档的匹配效果更好。

Step 2: 合并本地 RRF 结果

遍历 RRF 融合结果，转换为统一格式：

🟡 【P1 看注释就行】 _make_doc_item——统一文档结构工具方法。注意 chunk_id / doc_id 兼容两个字段名。

doc_items = []

# 本地 RRF 结果
for doc in (state.get("rrf_chunks") or []):
    # 跳过无效文档
    if not isinstance(doc, dict) or not doc.get("content"):
        continue

    doc_items.append(self._make_doc_item(
        text=doc["content"],               # 文档内容
        chunk_id=doc.get("chunk_id") or doc.get("id"),  # 文档 ID
        title=doc.get("title", ""),        # 标题
        source="local",                    # 来源标记
    ))

输入数据示例：

🔥 【P0 必须要学】 Step5 Reranker 计算得分——核心逻辑：reranker.compute_score(pairs) 批量计算 (query, doc) 对的相关性得分。注意异常处理：Exception 捕获后降级返回原序（score: None），确保模型失败时系统不崩溃。get_reranker_model() 是单例模式，避免重复加载模型。

rrf_chunks = [
    {
        "chunk_id": "chunk_001",
        "title": "万用表使用手册",
        "content": "测量电压时，将旋钮转到V档位..."
    },
    {
        "chunk_id": "chunk_002",
        "title": "电子元件检测",
        "content": "使用蜂鸣档可以快速检测短路..."
    }
]

Step 3: 合并网络搜索结果

遍历网络搜索结果，同样转换为统一格式：

🔥 【P0 必须要学】 Step6 断崖检测动态截断——核心算法：从 min_topk-1 位置开始，计算相邻得分的 gap（绝对差值）和 rel（相对比例）。任一条件触发即截断。注意 1e-6 防止除零。min_topk=3 确保至少保留 3 篇，max_topk=10 限制最大保留数。

# 网络搜索结果
for doc in (state.get("web_search_docs") or []):
    # 优先使用 snippet，降级使用 content
    text = (doc.get("snippet") or doc.get("content") or "").strip()
    if not text:
        continue

    doc_items.append(self._make_doc_item(
        text=text,                          # 摘要内容
        title=doc.get("title", "").strip(), # 网页标题
        url=doc.get("url", "").strip(),     # 网页链接
        source="web",                       # 来源标记
    ))

self.logger.info(f"合并文档: {len(doc_items)} 篇")

统一文档结构：

🟡 【P1 看注释就行】 Step7 返回结果——reranked_docs 包含 text / score / source / chunk_id / url 等字段。

@staticmethod
def _make_doc_item(
    text: str, source: str = "",
    chunk_id=None, title: str = "", url: str = "",
) -> Dict[str, Any]:
    """构建统一的文档结构。"""
    return {
        "text": text,           # 文档内容（必填）
        "source": source,       # 来源: local / web
        "chunk_id": chunk_id,   # 本地文档 ID
        "doc_id": chunk_id,     # 兼容字段
        "title": title,         # 标题
        "url": url,             # 网页链接（仅 web）
    }

Step 4: 构建 Query-Document 对

准备 Reranker 模型的输入格式：

🔥 【P0 必须要学】 RerankNode 完整代码——重点理解 4 步流程：(1)_merge_docs 多源合并 (2)_rerank 批量打分 + 异常降级 (3)_cliff_cutoff 断崖截断 (4)返回统一格式。Reranker 工具模块的单例模式（get_reranker_model）从环境变量读取模型路径、设备、FP16。

pairs = [[question, item["text"]] for item in doc_items]

数据格式示例：

🟢 【P2 后面可以查】 测试代码——mock 4 篇（2 local + 2 web），看断崖检测截断为 2 篇。

question = "主板短路怎么检测？"

pairs = [
    ["主板短路怎么检测？", "测量电压时，将旋钮转到V档位..."],
    ["主板短路怎么检测？", "使用蜂鸣档可以快速检测短路..."],
    ["主板短路怎么检测？", "主板通电前先打各主供电电感的对地阻值..."],
    ["主板短路怎么检测？", "苹果发布新款手机，A系列芯片性能提升..."],
]

Step 5: Reranker 计算得分

调用 Reranker 模型计算相关性得分：

try:
    # 获取 Reranker 模型（单例）
    reranker = get_reranker_model()

    # 计算所有 query-doc 对的相关性得分
    scores = reranker.compute_score(pairs)

    # 将得分附加到文档
    scored = [
        {**item, "score": float(s)}
        for item, s in zip(doc_items, scores)
    ]

    # 按得分降序排列
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored

except Exception as e:
    # 降级处理：返回原序，score 设为 None
    self.logger.error(f"重排序失败，降级为原序: {e}")
    return [{**item, "score": None} for item in doc_items]

得分计算过程：

Query: "主板短路怎么检测？"

文档 1: "使用蜂鸣档可以快速检测短路..."
        → score = 0.9156 (高度相关)

文档 2: "主板通电前先打各主供电电感的对地阻值..."
        → score = 0.8734 (相关)

文档 3: "测量电压时，将旋钮转到V档位..."
        → score = 0.4521 (部分相关)

文档 4: "苹果发布新款手机..."
        → score = 0.0234 (不相关)

Step 6: 断崖检测动态截断

从 min_topk 位置开始检测得分断崖：

def _cliff_cutoff(
    self, scored_docs: List[Dict[str, Any]], config,
) -> List[Dict[str, Any]]:
    """断崖检测截断：相邻得分差距超过阈值时截断。"""
    if not scored_docs:
        return []

    # 确定截断范围
    max_topk = min(config.rerank_max_topk, len(scored_docs))
    min_topk = config.rerank_min_topk

    topk = max_topk  # 默认取最大值

    # 从 min_topk 位置开始检测
    for i in range(min_topk - 1, max_topk - 1):
        s1 = scored_docs[i].get("score")
        s2 = scored_docs[i + 1].get("score")

        # 跳过无得分的文档
        if s1 is None or s2 is None:
            continue

        # 计算差值
        gap = s1 - s2                        # 绝对差值
        rel = gap / (abs(s1) + 1e-6)         # 相对比例

        # 发现断崖，立即截断
        if gap >= config.rerank_gap_abs or rel >= config.rerank_gap_ratio:
            topk = i + 1
            self.logger.debug(
                f"断崖检测: 位置 {i+1}, gap={gap:.4f}, rel={rel:.4f}"
            )
            break

    return scored_docs[:topk]

断崖检测示例：

配置: gap_abs=0.5, gap_ratio=0.25, min_topk=3, max_topk=10

排序后得分: [0.92, 0.87, 0.45, 0.08, 0.02]
位置:        [1]   [2]   [3]   [4]   [5]

检测过程（从位置 3 开始）:
  位置 3→4: gap = 0.45 - 0.08 = 0.37
            rel = 0.37 / 0.45 = 0.82 (> 0.25)
            → 发现断崖！截断在位置 3

结果: 保留前 3 篇文档

Step 7: 返回重排序结果

记录处理日志并返回结果：

topk_docs = self._cliff_cutoff(scored_docs, config)
self.logger.info(f"重排序完成: {len(doc_items)} → {len(topk_docs)}")

return {"reranked_docs": topk_docs}

4.4 代码实现

完整的节点实现代码：

"""重排序节点

使用 Reranker 模型对 RRF 融合结果和网络搜索结果进行重排序，
并通过断崖检测实现动态 TopK 截断。
"""

from typing import List, Dict, Any

from knowledge.processor.query_process.base import BaseNode, setup_logging
from knowledge.processor.query_process.state import QueryGraphState
from knowledge.processor.query_process.config import get_config
from knowledge.tools.reranker_utils import get_reranker_model


class RerankNode(BaseNode):
    """重排序节点。

    流程: 合并多源文档 → Reranker 计算相关性 → 断崖检测动态截断
    """

    name = "rerank"

    # ================================================================== #
    #                           主流程                                     #
    # ================================================================== #

    def process(self, state: QueryGraphState) -> QueryGraphState:
        config = get_config()

        # Step 1: 获取查询文本
        question = state.get("rewritten_query") or state.get("original_query", "")

        # Step 2-3: 合并文档
        doc_items = self._merge_docs(state)

        # Step 4-5: 重排序
        self.log_step("step_1", f"重排序 {len(doc_items)} 篇文档")
        scored_docs = self._rerank(question, doc_items)

        # Step 6: 动态 TopK 截断
        topk_docs = self._cliff_cutoff(scored_docs, config)

        # Step 7: 返回结果
        self.logger.info(f"重排序完成: {len(doc_items)} → {len(topk_docs)}")
        return {"reranked_docs": topk_docs}

    # ================================================================== #
    #                      文档合并                                        #
    # ================================================================== #

    def _merge_docs(self, state: QueryGraphState) -> List[Dict[str, Any]]:
        """合并本地 RRF 结果和网络搜索结果为统一格式。"""
        doc_items = []

        # Step 2: 本地 RRF 结果
        for doc in (state.get("rrf_chunks") or []):
            if not isinstance(doc, dict) or not doc.get("content"):
                continue
            doc_items.append(self._make_doc_item(
                text=doc["content"],
                chunk_id=doc.get("chunk_id") or doc.get("id"),
                title=doc.get("title", ""),
                source="local",
            ))

        # Step 3: 网络搜索结果
        for doc in (state.get("web_search_docs") or []):
            text = (doc.get("snippet") or doc.get("content") or "").strip()
            if not text:
                continue
            doc_items.append(self._make_doc_item(
                text=text,
                title=doc.get("title", "").strip(),
                url=doc.get("url", "").strip(),
                source="web",
            ))

        self.logger.info(f"合并文档: {len(doc_items)} 篇")
        return doc_items

    @staticmethod
    def _make_doc_item(
        text: str, source: str = "",
        chunk_id=None, title: str = "", url: str = "",
    ) -> Dict[str, Any]:
        """构建统一的文档结构。"""
        return {
            "text": text, "source": source,
            "chunk_id": chunk_id, "doc_id": chunk_id,
            "title": title, "url": url,
        }

    # ================================================================== #
    #                      Reranker 排序                                   #
    # ================================================================== #

    def _rerank(
        self, question: str, doc_items: List[Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        """计算相关性得分并排序，失败时降级返回原序。"""
        if not doc_items or not question:
            return []

        try:
            # Step 4: 构建 Query-Document 对
            reranker = get_reranker_model()
            pairs = [[question, item["text"]] for item in doc_items]

            # Step 5: 计算得分并排序
            scores = reranker.compute_score(pairs)
            scored = [
                {**item, "score": float(s)}
                for item, s in zip(doc_items, scores)
            ]
            scored.sort(key=lambda x: x["score"], reverse=True)
            return scored

        except Exception as e:
            self.logger.error(f"重排序失败，降级为原序: {e}")
            return [{**item, "score": None} for item in doc_items]

    # ================================================================== #
    #                    动态 TopK 截断                                    #
    # ================================================================== #

    def _cliff_cutoff(
        self, scored_docs: List[Dict[str, Any]], config,
    ) -> List[Dict[str, Any]]:
        """断崖检测截断：相邻得分差距超过阈值时截断。"""
        if not scored_docs:
            return []

        max_topk = min(config.rerank_max_topk, len(scored_docs))
        min_topk = config.rerank_min_topk

        topk = max_topk
        for i in range(min_topk - 1, max_topk - 1):
            s1 = scored_docs[i].get("score")
            s2 = scored_docs[i + 1].get("score")
            if s1 is None or s2 is None:
                continue

            gap = s1 - s2
            rel = gap / (abs(s1) + 1e-6)

            if gap >= config.rerank_gap_abs or rel >= config.rerank_gap_ratio:
                topk = i + 1
                self.logger.debug(
                    f"断崖检测: 位置 {i+1}, gap={gap:.4f}, rel={rel:.4f}"
                )
                break

        return scored_docs[:topk]


# 兼容原有调用方式
_node_instance = RerankNode()


def node_rerank(state: QueryGraphState) -> QueryGraphState:
    """兼容原有调用方式的入口函数。"""
    return _node_instance(state)

Reranker 工具模块（reranker_utils.py）：

import os
from FlagEmbedding import FlagReranker

_reranker_model = None


def get_reranker_model():
    """获取 Reranker 模型（单例模式）。"""
    global _reranker_model
    if _reranker_model is None:
        _reranker_model = FlagReranker(
            model_name_or_path=os.getenv("BGE_RERANKER_LARGE"),
            device=os.getenv("BGE_RERANKER_DEVICE"),
            use_fp16=os.getenv("BGE_RERANKER_FP16")
        )
    return _reranker_model

---

5. 测试运行

5.1 运行重排序节点测试

# 确保配置了 Reranker 模型路径
export BGE_RERANKER_LARGE="/path/to/bge-reranker-large"
export BGE_RERANKER_DEVICE="cuda"
export BGE_RERANKER_FP16="True"

# 运行测试
python -m knowledge.processor.query_process.nodes.rerank

5.2 测试代码

if __name__ == "__main__":
    from dotenv import load_dotenv
    load_dotenv()
    setup_logging()

    print("=" * 60)
    print("开始测试: 重排序节点 (RerankNode)")
    print("=" * 60)

    # 模拟输入状态
    # 包含 2 篇相关文档 + 2 篇不相关文档
    mock_state = {
        "rewritten_query": "怎么测这块主板的短路问题？",
        "rrf_chunks": [
            {
                "chunk_id": "local_1",
                "title": "主板维修手册",
                "content": "主板短路通常表现为通电后风扇转一下就停，"
                          "可以使用万用表的蜂鸣档测量。"
            },
            {
                "chunk_id": "local_2",
                "title": "闲聊",
                "content": "今天中午去吃猪脚饭吧，这块主板外观很漂亮。"
            },
        ],
        "web_search_docs": [
            {
                "url": "https://example.com/repair",
                "title": "短路查修指南",
                "snippet": "主板通电前先打各主供电电感的对地阻值，"
                          "阻值偏低就是短路。"
            },
            {
                "url": "https://example.com/news",
                "title": "科技新闻",
                "snippet": "苹果发布新款手机，A系列芯片性能提升20%。"
            },
        ],
    }

    print("【输入状态】:")
    print(f"  查询: {mock_state['rewritten_query']}")
    print(f"  本地文档: {len(mock_state['rrf_chunks'])} 篇")
    print(f"  网络文档: {len(mock_state['web_search_docs'])} 篇")
    print("-" * 60)

    # 执行重排序
    result = node_rerank(mock_state)

    # 打印结果
    print("\n【重排序结果】:")
    for i, doc in enumerate(result["reranked_docs"], 1):
        score = doc.get('score')
        score_str = f"{score:.4f}" if score is not None else "N/A"
        source = doc['source']
        text = doc['text'][:50]
        print(f"[{i}] score={score_str} | {source:5} | {text}...")

    print("-" * 60)
    print("测试完成")

5.3 预期输出

============================================================
开始测试: 重排序节点 (RerankNode)
============================================================
【输入状态】:
  查询: 怎么测这块主板的短路问题？
  本地文档: 2 篇
  网络文档: 2 篇
------------------------------------------------------------
[rerank] 合并文档: 4 篇
[rerank] [step_1] 重排序 4 篇文档
[rerank] 断崖检测: 位置 2, gap=0.4521, rel=0.5123
[rerank] 重排序完成: 4 → 2

【重排序结果】:
[1] score=0.9156 | local | 主板短路通常表现为通电后风扇转一下就停，可以使用万用表的蜂鸣档...
[2] score=0.8823 | web   | 主板通电前先打各主供电电感的对地阻值，阻值偏低就是短路...
------------------------------------------------------------
测试完成

5.4 处理前后对比

对比项	处理前	处理后
文档数量	4 篇（rrf 2 + web 2）	2 篇（断崖截断）
排序依据	RRF 位置 / 搜索顺序	Reranker 相关性得分
文档格式	不统一	统一结构
来源标识	无	source: local/web
相关性	混杂相关与不相关	仅保留高相关文档

数据结构变化：

# 处理前
state = {
    "rrf_chunks": [
        {"chunk_id": "local_1", "content": "主板短路..."},
        {"chunk_id": "local_2", "content": "今天中午..."},
    ],
    "web_search_docs": [
        {"url": "...", "snippet": "主板通电前..."},
        {"url": "...", "snippet": "苹果发布..."},
    ]
}

# 处理后
state = {
    ...,
    "reranked_docs": [
        {
            "text": "主板短路通常表现为...",
            "score": 0.9156,
            "source": "local",
            "chunk_id": "local_1",
            "title": "主板维修手册",
            "url": ""
        },
        {
            "text": "主板通电前先打各主供电...",
            "score": 0.8823,
            "source": "web",
            "chunk_id": None,
            "title": "短路查修指南",
            "url": "https://example.com/repair"
        }
    ]
}

---

6. 总结

6.1 节点功能概览

┌───────────────────��─────────────────────────────────────────┐
│                        RerankNode                            │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  核心功能: 使用 Reranker 模型精排并动态截断                    │
│                                                              │
│  输入:                                                       │
│    ├── state["rrf_chunks"]        RRF 融合结果               │
│    ├── state["web_search_docs"]   网络搜索结果               │
│    └── state["rewritten_query"]   查询文本                   │
│                                                              │
│  输出:                                                       │
│    └── state["reranked_docs"]     精排后的文档列表           │
│          ├── text      文档内容                              │
│          ├── score     相关性得分                            │
│          ├── source    来源 (local/web)                      │
│          ├── chunk_id  本地文档 ID                           │
│          └── url       网页链接                              │
│                                                              │
│  依赖:                                                       │
│    ├── FlagEmbedding.FlagReranker  重排序模型                │
│    ├── BGE_RERANKER_LARGE          模型路径                  │
│    ├── BGE_RERANKER_DEVICE         运行设备                  │
│    └── BGE_RERANKER_FP16           半精度模式                │
│                                                              │
│  特点:                                                       │
│    ├── 交叉编码器精排，精度高                                  │
│    ├── 断崖检测动态截断，自适应                                │
│    ├── 多源文档统一格式                                       │
│    └── 降级处理，容错性强                                      │
│                                                              │
└─────────────────────────────────────────────────────────────┘

6.2 节点设计要点

要点 1：交叉编码器 vs 双塔模型的选择

# 双塔模型（召回阶段）
# Query 和 Document 独立编码，可预计算
query_vec = encoder(query)
doc_vec = encoder(doc)        # 可离线计算存储
score = cosine(query_vec, doc_vec)

# 交叉编码器（精排阶段）
# Query 和 Document 联合编码，精度更高
pairs = [[query, doc1], [query, doc2], ...]
scores = reranker.compute_score(pairs)  # 需要在线计算

要点 2：多源文档格式统一

# 本地文档
{
    "chunk_id": "...",
    "content": "...",    # 内容字段名不同
    "title": "..."
}

# 网络文档
{
    "url": "...",
    "snippet": "...",    # 内容字段名不同
    "title": "..."
}

# 统一格式
{
    "text": "...",       # 统一使用 text
    "source": "local/web",
    "chunk_id": "...",
    "url": "...",
    "title": "..."
}

要点 3：断崖检测算法实现

def _cliff_cutoff(scored_docs, config):
    for i in range(min_topk - 1, max_topk - 1):
        gap = scored_docs[i]["score"] - scored_docs[i+1]["score"]
        rel = gap / (abs(scored_docs[i]["score"]) + 1e-6)

        # 绝对阈值 OR 相对阈值
        if gap >= gap_abs or rel >= gap_ratio:
            return scored_docs[:i+1]

    return scored_docs[:max_topk]

为什么需要两个阈值？

场景 1：高分区间断崖
  得分: [0.95, 0.92, 0.40, ...]
  gap = 0.52 > gap_abs=0.5  ✓ 触发截断

场景 2：低分区间断崖
  得分: [0.30, 0.28, 0.08, ...]
  gap = 0.20 < gap_abs=0.5  ✗
  rel = 0.20/0.28 = 0.71 > gap_ratio=0.25  ✓ 触发截断

要点 4：降级处理策略

try:
    reranker = get_reranker_model()
    scores = reranker.compute_score(pairs)
    # 正常处理...
except Exception as e:
    # 降级：返回原序，score 设为 None
    self.logger.error(f"重排序失败，降级为原序: {e}")
    return [{**item, "score": None} for item in doc_items]

为什么需要降级？

• Reranker 模型可能加载失败
• GPU 显存不足
• 输入文本过长
• 降级后系统仍可工作，只是精度下降

---

企业痛点映射

痛点	传统方案	Reranker 精排方案	效率提升
RRF 排序精度不够	排名分数粗糙，不相关工作可能排前面	交叉编码器深度交互打分	精排精度提升 ~40%（预估）
固定 TopK 丢失好结果	截断 5 篇后还有相关文档被丢弃	断崖检测动态截断，自适应保留	有用文档保留率提升 ~30%
多源文档格式不统一	本地和网络文档字段名不同，需手动拼接	_merge_docs 统一格式 + _make_doc_item 标准化	接入新来源只需加一个 if
Reranker 模型加载失败时全流程崩溃	模型挂了流程就停了	异常捕获 + 降级返回原序	流程可用性提升至 ~100%

---

Remote & Agent 应用场景价值

• Remote 场景价值：BGE-Reranker 模型需要 GPU 推理（通过 BGE_RERANKER_DEVICE 配置），远程团队成员需共享 GPU 资源。get_reranker_model() 的单例模式避免重复加载模型浪费显存。

• Agent 落地场景：RerankNode 可封装为"精排 Agent"——Agent 接收多源文档列表 + 查询 → 调用 Reranker 模型批量打分 → 断崖检测截断 → 返回精排结果。该 Agent 可作为"答案生成 Agent"的上游，确保只输入最相关的文档给 LLM。

---

Git Commit 对应

本节重排序节点对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议搜索 "rerank.py" / "reranker_utils.py" 相关提交>

cd shopkeeper_brain
git log --oneline --all -- knowledge/processor/query_process/nodes/rerank.py