知识库查询 —— RRF 融合节点

学习理念：RRF（Reciprocal Rank Fusion）是"三路检索"的收口节点——向量检索、HyDE 检索、知识图谱检索各自返回了一堆结果，RRF 把它们合并成一个有序列表。它的核心思路很简单：不看分数看排名，多路都命中 = 更相关。 这不是深度学习，是一个 30 行的数学公式。

海外对标：RRF 算法源自信息检索领域（TREC 2009），被 Google 搜索、Elasticsearch 的混合检索、以及 Milvus 内置的 RRFRanker 广泛采用。本项目的手写实现与 Milvus 内置 RRF 功能等价。

本节 AI 替代率：~85% | 人工干预率：~15%

角色	能力范围
🤖 AI 擅长	解释 RRF 数学原理、生成 RRF 代码骨架、处理多路结果格式兼容
👤 人类需理解	为什么看排名不看分数、k 值的作用、加权策略的设计思路

---

阅读指引

章节	阅读方式	说明
§1 任务目标	🔥 精读	理解 RRF 在查询流程中的位置
§2 核心概念	🔥 精读	RRF 公式 + k 值 + 加权策略（含新增对照表）
§3 节点位置与流程	🟢 略读	Mermaid 图替换了 JPG
§4 业务实现	🔥 精读	逐步骤代码 + 类图 + 时序图
§5 测试	🟢 略读	看预期输出即可
§6 总结 + 新增加载	🟢 略读	要点回顾 + 企业痛点

---

1. 任务目标

本节课将实现知识库查询流程中的 RRF 融合节点（rrf）。

该节点负责将多路检索结果（向量检索、HyDE 检索、知识图谱检索）通过 Reciprocal Rank Fusion（倒数排名融合） 算法进行融合，生成统一的排序结果。

学完本节你将掌握：

• RRF 算法的数学原理与直觉理解
• 多路检索结果融合的工程实现
• 加权融合策略的设计思路
• chunk_id 去重与合并的处理技巧

---

2. 核心概念扫盲

2.1 为什么需要融合排序？

在 multi_search 阶段，我们并行执行了多路检索：

原图为 JPG 图片 images/106...jpg，以下用 Mermaid 图补充：

问题： 不同检索方法的评分标准不同，无法直接比较：

• 向量检索：返回距离（越小越好）
• 知识图谱：返回相关度（无统一标准）
• 不同方法可能返回相同文档

解决方案： 使用 RRF 算法，基于排名位置而非原始分数进行融合。

2.2 RRF 算法原理

Reciprocal Rank Fusion（倒数排名融合） 是一种经典的排名融合算法。

核心公式

$$ RRF_score(d) = \sum_{i=1}^{n} \frac{weight_i}{k + rank_i(d)} $$

参数说明：

• d：待评分的文档
• n：检索路数
• weight_i：第 $i$ 路的权重
• rank_i(d)：文档 $d$ 在第 $i$ 路中的排名位置（从 1 开始）
• k：平滑常数（通常取 60）

理解

假设有文档 A，在三路检索中的排名为：

路径           排名      贡献分数 (k=60)
─────────────────────────────────────────
向量检索        1        1/(60+1) = 0.0164
HyDE 检索       3        1/(60+3) = 0.0159
知识图谱        2        0.7/(60+2) = 0.0113  (权重 0.7)
─────────────────────────────────────────
                         总分 = 0.0436

RRF 的优势：

特点	说明
无需标准化	只看排名，不看原始分数
抗噪声	平滑常数 k 防止头部排名过度主导
鼓励共识	多路命中的文档得分更高
惩罚离散	只在少数路径命中的文档得分较低

2.3 常数 k 的作用

k 值决定了排名差异对得分的影响程度：

排名位置对得分的影响（k=60）：

排名 1:  1/(60+1)  = 0.0164
排名 2:  1/(60+2)  = 0.0161  (仅下降 1.8%)
排名 10: 1/(60+10) = 0.0143  (下降 13%)
排名 50: 1/(60+50) = 0.0091  (下降 45%)

k 值选择：

• k 较小（如 10）：头部排名差异影响大，适合高精度场景
• k 较大（如 60）：排名差异影响平滑，适合多路融合
• 实践中 k=60 是经过验证的经典选择

2.4 加权 RRF 的意义

不同检索路径的可靠性可能不同，通过权重调节：

sources = {
    "embedding": (docs, 1.0),   # 向量检索，权重 1.0
    "hyde":      (docs, 1.0),   # HyDE 检索，权重 1.0
    "kg":        (docs, 0.7),   # 知识图谱，权重 0.7
}

为什么知识图谱权重较低（0.7）？

• 知识图谱依赖实体提取质量
• 实体对齐可能引入噪声
• 图谱覆盖面可能不完整
• 降低权重可减少低质量结果的影响

新增：中英文对照表

English	中文	本质
Reciprocal Rank Fusion (RRF)	倒数排名融合	基于排名位置而非原始分数的多路融合算法
Rank	排名	文档在检索结果中的位置（第1名 = rank 1）
k (smoothing constant)	平滑常数	控制排名差异对得分的影响程度
Weight	权重	不同检索路径的相对重要性
Setdefault	去重保留	首次出现的文档版本保留，后续忽略
Chunk	切片/文档块	被检索的最小文档单元

新增：程序员比喻

RRF 像 Stack Overflow 的回答排序——不止看点赞数（分数），还看回答者在多个问题上的活跃度（多路命中）。一个人只在 C++ 问题的第 1 名出现一次，不如一个人在 C++、Python、Go 三个问题的前 10 都出现可信。

---

3. RRF 融合业务处理流程（总）

3.1 节点在流程中的位置

原图为 JPG 图片 images/107...jpg，以下用 Mermaid 图补充：

3.2 节点输入输出

---

4. RRF 融合业务处理流程（分）

4.1 目标

实现一个 RRF 融合节点，将三路检索结果合并为统一的排序列表，为后续重排序提供候选集。

4.2 需求分析

需求项	说明
多源输入	支持向量、HyDE、知识图谱三路输入
格式兼容	兼容不同上游节点的输出格式
加权融合	支持为不同路径配置不同权重
去重合并	相同 chunk_id 的文档合并得分
可配置	k 值、权重、最大结果数可配置
容错	某路为空时不影响整体流程

4.3 实现流程

4.3.0 📊 类图（新增）

4.3.1 实现流程图

原图为 JPG 图片 images/109...jpg，以下用 Mermaid 图补充：

4.3.2 📊 时序图（新增）

4.3.3 具体实现步骤

Step 1: 收集三路检索结果

从状态中获取各路检索结果，并配置对应权重：

🟡 【P1 看注释就行】 三路收集的代码套路固定——get → 处理→ log。看懂 sources 字典的结构即可。

config = get_config()

# 收集三路检索结果及其权重
sources = {
    "embedding": (
        self._extract_entities(state.get("embedding_chunks")),
        1.0  # 向量检索权重
    ),
    "hyde": (
        self._extract_entities(state.get("hyde_embedding_chunks")),
        1.0  # HyDE 检索权重
    ),
    "kg": (
        self._extract_entities(state.get("kg_chunks")),
        config.rrf_kg_weight  # 知识图谱权重（默认 0.7）
    ),
}

self.logger.info(
    f"RRF 输入: {', '.join(f'{k}={len(v[0])}' for k, v in sources.items())}"
)

数据结构示例：

# 向量检索结果（10 条）
embedding_chunks = [
    {"entity": {"chunk_id": "c1", "content": "..."}},
    {"entity": {"chunk_id": "c2", "content": "..."}},
    # ...
]

# HyDE 检索结果（5 条）
hyde_embedding_chunks = [
    {"entity": {"chunk_id": "c2", "content": "..."}},  # 与向量检索重复
    {"entity": {"chunk_id": "c3", "content": "..."}},
    # ...
]

# 知识图谱结果（若干条）
kg_chunks = [
    {"chunk_id": "c1", "content": "..."},  # 格式可能不同
    {"chunk_id": "c4", "content": "..."},
    # ...
]

Step 2: 格式规整化

上游节点输出格式可能不统一，需要规整化处理：

🟡 【P1 看注释就行】 格式兼容代码是工具方法，理解 doc.get("entity") or doc 这一行就够了。

@staticmethod
def _extract_entities(state_list) -> List[Dict[str, Any]]:
    """将上游节点输出统一规整为 entity 字典列表。

    兼容格式：
    - {"entity": {...}, "distance": ...} → 取 entity
    - {..., "chunk_id": ...} → 直接使用
    """
    out: List[Dict[str, Any]] = []
    for doc in (state_list or []):
        if not doc or not hasattr(doc, "get"):
            continue
        # 如果有 entity 字段，取 entity；否则直接使用
        out.append(doc.get("entity") or doc)
    return out

格式转换示例：

输入格式 1（向量检索）:
{"entity": {"chunk_id": "c1", "content": "..."}, "distance": 0.15}
                    ↓
输出: {"chunk_id": "c1", "content": "..."}

输入格式 2（知识图谱）:
{"chunk_id": "c1", "content": "..."}
                    ↓
输出: {"chunk_id": "c1", "content": "..."}  (不变)

Step 3: 构建得分映射表

遍历每路结果，计算并累加 RRF 得分：

🔥 【P0 必须要学】 这就是 RRF 的核心——三行公式。理解它等于理解了整篇文档。

score_map: Dict[str, float] = {}  # chunk_id -> 累计得分

for rank_list, weight in source_weights:
    for pos, item in enumerate(rank_list, start=1):
        chunk_id = item.get("chunk_id")
        if not chunk_id:
            continue

        # RRF 公式: weight / (k + rank)
        contribution = weight / (k + pos)
        score_map[chunk_id] = score_map.get(chunk_id, 0.0) + contribution

计算过程示例：

假设 k=60，知识图谱权重=0.7

文档 c1 的得分计算：
─────────────────────────────────────────────────────
路径            排名    权重    贡献分数
─────────────────────────────────────────────────────
embedding       1       1.0     1.0/(60+1) = 0.01639
hyde            3       1.0     1.0/(60+3) = 0.01587
kg              2       0.7     0.7/(60+2) = 0.01129
─────────────────────────────────────────────────────
                              总分 = 0.04355

文档 c2 的得分计算：
─────────────────────────────────────────────────────
路径            排名    权重    贡献分数
─────────────────────────────────────────────────────
embedding       2       1.0     1.0/(60+2) = 0.01613
hyde            1       1.0     1.0/(60+1) = 0.01639
kg              -       -       0（未命中）
─────────────────────────────────────────────────────
                              总分 = 0.03252

Step 4: 构建文档映射表

同时维护 chunk_id 到完整文档的映射：

🟡 【P1 看注释就行】 setdefault 是 Python 标准技巧，看懂这行的作用即可。

chunk_map: Dict[str, Dict] = {}  # chunk_id -> 完整文档

for rank_list, weight in source_weights:
    for pos, item in enumerate(rank_list, start=1):
        chunk_id = item.get("chunk_id")
        if not chunk_id:
            continue

        # setdefault: 只在首次遇到时记录，避免覆盖
        chunk_map.setdefault(chunk_id, item)

为什么使用 setdefault？

# 场景：同一文档在多路中出现，内容可能略有不同

# 向量检索返回（先遇到）
{"chunk_id": "c1", "content": "原始内容", "distance": 0.1}

# 知识图谱返回（后遇到）
{"chunk_id": "c1", "content": "原始内容", "score": 0.9}

# 使用 setdefault 保留第一次遇到的版本
# 这样可以保持结果一致性

Step 5: 排序与截断

将得分映射转换为排序列表：

🟡 【P1 看注释就行】 Python 标准排序写法，看懂 key 和 reverse 即可。

# 构建 (文档, 得分) 列表
merged = [
    (chunk_map[cid], score)
    for cid, score in score_map.items()
]

# 按得分降序排列
merged.sort(key=lambda x: x[1], reverse=True)

# 截取前 max_results 条
if max_results:
    merged = merged[:max_results]

排序结果示例：

排序前（按 chunk_id 遍历顺序）：
[
    (c1_doc, 0.04355),
    (c2_doc, 0.03252),
    (c3_doc, 0.02100),
    (c4_doc, 0.01800),
    ...
]

排序后（按得分降序）：
[
    (c1_doc, 0.04355),  # 最高分
    (c2_doc, 0.03252),
    (c3_doc, 0.02100),
    (c4_doc, 0.01800),
    ...
]

截断后（max_results=10）：
[前 10 条结果]

Step 6: 输出结果

记录日志并返回融合结果：

🟡 【P1 看注释就行】 输出代码是固定模式——提取 → 日志 → return。

# 提取文档列表（不含得分）
rrf_chunks = [doc for doc, _ in rrf_results]

self.logger.info(f"RRF 融合完成，返回 {len(rrf_chunks)} 条结果")

# 记录分数范围（便于调试）
if rrf_results:
    scores = [s for _, s in rrf_results]
    self.logger.info(f"分数范围: [{min(scores):.6f}, {max(scores):.6f}]")

return {"rrf_chunks": rrf_chunks}

4.4 代码实现

🟡 【P1 看注释就行】 完整代码是前面 Step 1~6 的汇总。如果 Step 1~6 看懂了，这里直接过。

完整的节点实现代码：

"""RRF 融合排序节点

使用 Reciprocal Rank Fusion 算法融合多路检索结果。
"""

from typing import List, Dict, Any, Tuple

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


class RrfNode(BaseNode):
    """RRF 融合排序节点。

    流程: 收集三路检索结果 → 带权重 RRF 融合 → 按得分降序返回
    """

    name = "rrf"

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

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

        # Step 1: 收集三路检索结果
        sources = {
            "embedding": (
                self._extract_entities(state.get("embedding_chunks")),
                1.0
            ),
            "hyde": (
                self._extract_entities(state.get("hyde_embedding_chunks")),
                1.0
            ),
            "kg": (
                self._extract_entities(state.get("kg_chunks")),
                config.rrf_kg_weight
            ),
        }

        self.logger.info(
            f"RRF 输入: {', '.join(f'{k}={len(v[0])}' for k, v in sources.items())}"
        )

        # Step 2-5: 执行 RRF 融合
        source_weights = list(sources.values())
        rrf_results = self._reciprocal_rank_fusion(
            source_weights,
            k=config.rrf_k,
            max_results=config.rrf_max_results,
        )

        # Step 6: 输出结果
        rrf_chunks = [doc for doc, _ in rrf_results]
        self.logger.info(f"RRF 融合完成，返回 {len(rrf_chunks)} 条结果")

        if rrf_results:
            scores = [s for _, s in rrf_results]
            self.logger.info(f"分数范围: [{min(scores):.6f}, {max(scores):.6f}]")

        return {"rrf_chunks": rrf_chunks}

    # ================================================================== #
    #                      RRF 算法实现                                    #
    # ================================================================== #

    @staticmethod
    def _reciprocal_rank_fusion(
        source_weights: List[Tuple[List[Dict], float]],
        k: int = 60,
        max_results: int = None,
    ) -> List[Tuple[Dict, float]]:
        """带权重的 RRF 融合。

        公式: score(d) = Σ weight_i / (k + rank_i(d))

        Args:
            source_weights: [(文档列表, 权重), ...]
            k: RRF 常数，值越大则排名差异对得分的影响越平滑。
            max_results: 返回前 N 个，None 表示全部。

        Returns:
            [(文档, 得分), ...] 按得分降序。
        """
        # Step 3: 构建得分映射表
        score_map: Dict[str, float] = {}
        # Step 4: 构建文档映射表
        chunk_map: Dict[str, Dict] = {}

        for rank_list, weight in source_weights:
            for pos, item in enumerate(rank_list, start=1):
                chunk_id = item.get("chunk_id")
                if not chunk_id:
                    continue
                # RRF 公式
                score_map[chunk_id] = score_map.get(chunk_id, 0.0) + weight / (k + pos)
                chunk_map.setdefault(chunk_id, item)

        # Step 5: 排序与截断
        merged = sorted(
            [(chunk_map[cid], score) for cid, score in score_map.items()],
            key=lambda x: x[1],
            reverse=True,
        )

        return merged[:max_results] if max_results else merged

    # ================================================================== #
    #                      工具方法                                        #
    # ================================================================== #

    @staticmethod
    def _extract_entities(state_list) -> List[Dict[str, Any]]:
        """将上游节点输出统一规整为 entity 字典列表。

        兼容格式：
        - {"entity": {...}, "distance": ...} → 取 entity
        - {..., "chunk_id": ...} → 直接使用
        """
        out: List[Dict[str, Any]] = []
        for doc in (state_list or []):
            if not doc or not hasattr(doc, "get"):
                continue
            out.append(doc.get("entity") or doc)
        return out


# ================================================================== #
#                        兼容 & 测试                                   #
# ================================================================== #

_node_instance = RrfNode()


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

---

5. 测试运行

5.1 运行 RRF 融合节点测试

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

5.2 测试代码

🟢 【P2 后面可以查】 测试代码量很大（~50 行），但价值最低。看懂 mock_state 的结构、跑一遍看预期输出即可。具体怎么 mock 的，需要时再回来翻。

if __name__ == "__main__":
    import json
    setup_logging()

    print("=" * 60)
    print("开始测试: RRF 融合节点 (RrfNode)")
    print("=" * 60)

    # 模拟三路检索结果
    # chunk_1 命中 3 路（最高分）
    # chunk_2 命中 2 路
    # chunk_3, chunk_4, chunk_5 各命中 1 路
    mock_state = {
        "embedding_chunks": [
            {"entity": {"chunk_id": "chunk_1", "content": "向量搜索结果#1"}},
            {"entity": {"chunk_id": "chunk_2", "content": "向量搜索结果#2"}},
            {"entity": {"chunk_id": "chunk_3", "content": "向量搜索结果#3"}},
        ],
        "hyde_embedding_chunks": [
            {"entity": {"chunk_id": "chunk_2", "content": "HyDE搜索结果#1"}},
            {"entity": {"chunk_id": "chunk_1", "content": "HyDE搜索结果#2"}},
            {"entity": {"chunk_id": "chunk_4", "content": "HyDE搜索结果#3"}},
        ],
        "kg_chunks": [
            {"chunk_id": "chunk_5", "content": "知识图谱结果#1"},
            {"chunk_id": "chunk_1", "content": "知识图谱结果#2"},
        ],
    }

    print("【输入状态】:")
    print(f"  embedding_chunks: {len(mock_state['embedding_chunks'])} 条")
    print(f"  hyde_embedding_chunks: {len(mock_state['hyde_embedding_chunks'])} 条")
    print(f"  kg_chunks: {len(mock_state['kg_chunks'])} 条")
    print("-" * 60)

    # 执行 RRF 融合
    result = node_rrf(mock_state)

    # 打印结果
    print("\n【融合结果】:")
    for i, chunk in enumerate(result["rrf_chunks"], 1):
        print(f"[{i}] {chunk.get('chunk_id')} - {chunk.get('content')}")

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

5.3 预期输出

============================================================
开始测试: RRF 融合节点 (RrfNode)
============================================================
【输入状态】:
  embedding_chunks: 3 条
  hyde_embedding_chunks: 3 条
  kg_chunks: 2 条
------------------------------------------------------------
[rrf] RRF 输入: embedding=3, hyde=3, kg=2
[rrf] RRF 融合完成，返回 5 条结果
[rrf] 分数范围: [0.011290, 0.043867]

【融合结果】:
[1] chunk_1 - 向量搜索结果#1
[2] chunk_2 - 向量搜索结果#2
[3] chunk_3 - 向量搜索结果#3
[4] chunk_4 - HyDE搜索结果#3
[5] chunk_5 - 知识图谱结果#1
------------------------------------------------------------
测试完成

分数计算验证（k=60, kg_weight=0.7）：

chunk_1: 命中 embedding(1), hyde(2), kg(2)
  = 1.0/(60+1) + 1.0/(60+2) + 0.7/(60+2)
  = 0.01639 + 0.01613 + 0.01129
  = 0.04381  ← 最高分

chunk_2: 命中 embedding(2), hyde(1)
  = 1.0/(60+2) + 1.0/(60+1)
  = 0.01613 + 0.01639
  = 0.03252

chunk_3: 仅命中 embedding(3)
  = 1.0/(60+3)
  = 0.01587

chunk_4: 仅命中 hyde(3)
  = 1.0/(60+3)
  = 0.01587

chunk_5: 仅命中 kg(1)
  = 0.7/(60+1)
  = 0.01148  ← 最低分（虽然排名第1，但权重低且仅单路命中）

5.4 处理前后对比

对比项	处理前	处理后
embedding_chunks	3 条结果	不变
hyde_embedding_chunks	3 条结果	不变
kg_chunks	2 条结果	不变
rrf_chunks	不存在	5 条去重融合结果

数据结构变化：

# 处理前
state = {
    "embedding_chunks": [
        {"entity": {"chunk_id": "chunk_1", ...}},
        {"entity": {"chunk_id": "chunk_2", ...}},
        {"entity": {"chunk_id": "chunk_3", ...}},
    ],
    "hyde_embedding_chunks": [
        {"entity": {"chunk_id": "chunk_2", ...}},  # 与上面重复
        {"entity": {"chunk_id": "chunk_1", ...}},  # 与上面重复
        {"entity": {"chunk_id": "chunk_4", ...}},
    ],
    "kg_chunks": [
        {"chunk_id": "chunk_5", ...},
        {"chunk_id": "chunk_1", ...},  # 与上面重复
    ],
    # rrf_chunks 不存在
}

# 处理后
state = {
    # 原有字段不变...
    "rrf_chunks": [
        {"chunk_id": "chunk_1", "content": "..."},  # 得分最高
        {"chunk_id": "chunk_2", "content": "..."},
        {"chunk_id": "chunk_3", "content": "..."},
        {"chunk_id": "chunk_4", "content": "..."},
        {"chunk_id": "chunk_5", "content": "..."},  # 去重后共 5 条
    ]
}

---

6. 总结

6.1 节点功能概览

6.2 节点设计要点

要点 1：RRF 核心公式实现

# 公式: score(d) = Σ weight_i / (k + rank_i(d))
score_map[chunk_id] = score_map.get(chunk_id, 0.0) + weight / (k + pos)

关键细节：

• pos 从 1 开始（不是 0），符合排名语义
• 使用 get(..., 0.0) 实现累加
• k 取 60 是经过验证的经典值

要点 2：格式兼容处理

@staticmethod
def _extract_entities(state_list) -> List[Dict[str, Any]]:
    out = []
    for doc in (state_list or []):
        if not doc or not hasattr(doc, "get"):
            continue
        # 兼容两种格式
        out.append(doc.get("entity") or doc)
    return out

设计考量：

• 向量检索返回 {"entity": {...}, "distance": ...}
• 知识图谱返回 {"chunk_id": ..., "content": ...}
• 使用 doc.get("entity") or doc 统一处理

要点 3：去重策略

# 使用 setdefault 保留首次遇到的文档版本
chunk_map.setdefault(chunk_id, item)

为什么这样设计：

• 同一文档可能在多路中出现
• 不同路径返回的元数据可能略有不同
• 保留第一次遇到的版本，保持结果一致性

要点 4：加权策略的灵活性

sources = {
    "embedding": (docs, 1.0),   # 权重可配置
    "hyde":      (docs, 1.0),
    "kg":        (docs, config.rrf_kg_weight),  # 从配置读取
}

设计优势：

• 权重从配置中心读取，支持动态调整
• 不同场景可使用不同权重
• 新增检索路径只需添加一行

---

新增：企业痛点映射

痛点	传统方案	RRF 方案	效率提升
多路检索分数不可比	直接拼接，图谱高分碾压向量	只看排名不看分数	准确性提升 ~20%
重复文档需人工去重	手动逐条对比合并	setdefault 自动去重	零人工成本
不同检索路径质量不均	等权处理	加权 RRF（图谱 0.7）	灵活适配业务场景

---

新增：Git Commit 对应

对应的代码提交记录：

1bb9a6e day17-查询流程(rrf节点和rerank节点)编码实现

cd 代码/shopkeeper_brain
git checkout 1bb9a6e    # 查看当天的代码
git checkout master     # 回到最新