掌柜智库项目 Day 10｜RRF 融合、Reranker 重排与结果处理

用权重加 RRF 思路精炼两路向量检索结果，再引入 bge-reranker-large 对多源候选统一打分与动态 TopK 截取，最后通过 output_answer 节点生成答案并正则提取图片地址。

本篇要点

node_rrf 节点：用“权重 + RRF”把两路向量检索结果再次精炼
reranker 节点：面向多源候选统一打分与动态截取
MTEB 模型对比、bge-reranker-large 加载与 compute_score 打分
token 长度计算与重排序模型上下文超长处理
output_answer 节点：流式 / 非流式返回、正则提取 image_urls、保存聊天记录

内容回顾

1、检索节点

node_search_embedding：rewritten_query 的向量与 chunks 的 content（item_name + title + content）做混合 + 条件检索，生成 embedding_chunks。
node_search_embedding_hyde：先用大模型基于原问题生成假设性标准回答，再用“问题 + 答案”的向量复用混合检索链路，生成第二路增强召回 hyde_embedding_chunks。
node_web_search_mcp：通过 MCP 调用百炼联网搜索工具补全数据，生成 web_search_docs。

2、节点参数取向

数据库内检索两路都带 expr="item_name in [item_names]" 的 item_name 前置过滤。

今天内容

注意： 问题和建设性问题检索，包装的数据类型 key 用 chunk_id，评估代码也依赖 chunk_id。

1、节点 node_rrf

作用： 把向量数据库中的两路数据再次精炼，使用权重 + RRF 的思路。

参数：

使用：embedding_chunks、hyde_embedding_chunks。
生成：rrf_chunks。

步骤：

获取并且校验参数（state）→ embedding_chunks、hyde_embedding_chunks。

封装数据结构 → list：

1	[(权重, list -> 有顺序 -> 向量数据库查询 -> 分高), (权重, list)]

使用 RRF 算法进行数据处理（list）→ rrf_chunks[5]。
修改 state = rrf_chunks。
返回 state。

2、reranker 节点

作用： 对 rrf_chunks 粗排序的结果 + 非数据库中 mcp 网络搜索的结果进行最终打分 + 动态截取，产出最终候选者。

为什么在这里需要 reranker 模型？ 因为这两路数据没有统一的排名参照物！Reranker 的职责就是：reranker -> 打分 -> 算法截取 -> 最终的候选者。

参数：

使用：rrf_chunks、web_search_docs、rewritten_query。
生成：reranked_docs。

2.1 MTEB 模型对比

MTEB 是评估 embedding 模型的常用榜单，下面是几款模型的对比数据：

模型名称	维度	序列长度	平均值（56）	回溯（15）	聚类（11）	成对分类（3）	重排序（4）	STS（10）	总结（1）	分类（12）
BAAI/bge-large-en-v1.5	1024	512	64.23	54.29	46.08	87.12	60.03	83.11	31.61	75.97
BAAI/bge-base-en-v1.5	768	512	63.55	53.25	45.77	86.55	58.86	82.4	31.07	75.53
BAAI/bge-small-en-v1.5	384	512	62.17	51.68	43.82	84.92	58.36	81.59	30.12	74.14

2.2 为什么使用 bge-reranker-large

选择 bge-reranker-large 的三个理由：

适配语言（多语言支持友好）。
性能的问题（重排效果更好）。
milvus 兼容——与 bge-m3 配套使用。

2.3 上下文长度问题

重排序模型的输入是 [问题, 答案] 这样的一对：

1 2	"max_position_embeddings": 514 // -> 514 - 2 = 512 token "model_type": "xlm-roberta" // -> -4 = 508 token

打分时的格式是：\s问题\s \s答案\s。

面试题：重排序模型的上下文可能短于 content 内容，你是如何处理的？
换模型 → 需要重新评估。
压缩比较内容：把 content 用 LLM 压缩后再比较打分（选择该方案）。
1
2
508 - 问题的 token = 答案 token
答案 token > token -> 启动压缩 -> 调用模型
冷处理 → 不管。

2.4 Python 中如何加载模型

bge-m3 与 bge-reranker-large 都有官方实现与 Milvus 集成两种来源，这里选择官方提供的方式（数据要存储到 milvus 向量数据库，兼容性更好）。

步骤 1：下载模型的权重文件

1	D:\project_dir\ai_0318_knowledge_base\app\shared\tool\download_reranker.py

步骤 2：加载模型客户端才能使用

1	uv add FlagEmbedding

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

reranker.compute_score([问题, 答案])            # => score
reranker.compute_score([[问题, 答案], [问题, 答案]])  # => [scores]

注意： 每一对不超过 508 即可。

2.5 常用 API 与 token 长度计算

打分：

1 2	reranker.compute_score([问题, 答案]) # => score reranker.compute_score([[问题, 答案], [问题, 答案]]) # => [scores]

如何计算模型下的 token 的长度：

模型 → 分词器 → 字符串 → token。

1 2	"1231312" -> qwen => 15 "1231312" -> reranker => 12

tokenizer = reranker.tokenizer
# "1231312" = [101, 798, 109, 22, 43] => 101 -> token 码表 -> token
list_token_id = tokenizer.encode("字符串")
len(list_token_id)  # = token 数量

2.6 模型加载工具类与配置

确认配置文件 .env：

# 重排序模型配置
BGE_RERANKER_LARGE=D:\ai_models\modelscope_cache\models\rerank\BAAI\bge-reranker-large
#BGE_RERANKER_DEVICE=cuda:0
BGE_RERANKER_DEVICE=cpu
#BGE_RERANKER_FP16=1
BGE_RERANKER_FP16=0

加载工具类位置：

1	D:\project_dir\ai_0318_knowledge_base\app\shared\model\reranker_utils.py

封装到 infra，并提供一个方法获取重排序模型：

1
2
3

# 重排序模型对象获取
def reranker_model(self):
    return get_reranker_model()

2.7 常量配置

RERANK_MAX_TOPK: int = 10
RERANK_MIN_TOPK: int = 1
RERANK_GAP_RATIO: float = 2
RERANK_GAP_ABS: float = 2
RERANK_MAX_INPUT_TOKENS: int = 512
RERANK_SUMMARY_CHAR_RATIO: float = 1.3
RERANK_MIN_SUMMARY_CHARS: int = 50

2.8 业务步骤

获取并且校验参数（state） → rewritten_query、rrf_chunks、web_search_docs。

数据格式化处理：把两类数据统一成一种格式再交给模型。

1 2	rrf_chunks {chunk_id, title, parent_title, part, file_title, content, item_name, score, type, url} web_search_docs{snippt, title, url}

字段对齐关系：

rrf_chunks	web_search_docs	统一字段	说明
chunk_id	没有	chunk_id	数据库里才有的标识
content	snippt	text	回答参考内容
title	title	title	标题
score	没有	score	web 记 0；milvus 记 rrf 的分，之后由 reranker 打分
type	没有	type	mcp 记 web；数据库记 milvus
url	没用	url	图片地址

最终得到 reranker_list。

组装问题和答案的列表（rewritten_query, reranker_list）→ question_answer_pair_list [[], [], []]：
- 获取问题。
- 判断问题 token 的长度。
- 循环 reranker_list 获取答案 text：
  - 判断 text 的长度。
    - 超长 → 模型压缩 → 调用模型。
  - 装数据 pair。
- 返回结果。

reranker 模型打分 + 排序：

reranker.compute_score([question_answer_pair_list[问题, 答案]])  # -> [scores]
[scores] -> reranker_list {score: x}   # -> zip
sort 排序
reranker_list {text 没有压缩}  # -> 分数 + 排序

动态 TopK 截取数据：
1
2
3
reranker_list min max topk
1 2 3 4 5 6 7 8
断崖值 0.3
按分数排好序后，一旦相邻分数出现明显“断崖”（如 0.3 的落差）就提前截断；否则在 min、max 之间按 RERANK_MIN_TOPK / RERANK_MAX_TOPK 取值。

3、output_answer 节点

作用：

把 reranked_docs、context（切片拼接的上下文，回答的参考）+ history（聊天记录）+ item_names + question（重写的问题）交给模型生成答案，同时自己用正则从 context 中提取关联的图片地址。

建议修改提示词顺序：question 向上挪 —— question - context - item_names - history。

参数：

使用：reranked_docs、context（切片拼接上下文，回答的参考）+ history（聊天记录）+ item_names + question（重写的问题）。
生成：answer、image_urls。

步骤（流式 / 非流式）：

非流式 -> answer [llm]  image_urls [正则] -> state return state -> 接口
流式   -> answer sse -> queue -> content = llm.stream()
          -> sse_utils.push_to_session(session_id, delta, {delta: content})
          image_urls -> 接口 -> 最后 -> push_to_session(session_id, final, {image_urls: image...}) -> state

特殊情况： 没有识别出 item_name 时，state.answer = xxxxxx，直接返回，不调用模型、不处理图片。

判断 state 是否有 answer，有就直接返回（state）→ 用 bool 判断有没有 answer：
- 没有 → return false。
- 有 → 流式 push_to_session(session_id, delta, {delta: answer})；非流式 return True，state => answer。
上一次返回的值是 false（即没有 answer），继续往下走。
answer_prompt_text = 声明一个提示词拼接的方法（state）进行拼接。
调用模型处理字符串 answer 回答的问题（state, answer_prompt_text）：
1
2
流式 llm - stream -> push ... -> state.answer
非流式 llm - invoke -> state -> answer
state[answer] = ...
提取图片存储到 state（reranked 里的 text ![]()、mcp 的 url → 判断是不是图片，写入 state）：
- url → 看后缀名。
- text → 网络搜索的 markdown ![xxxx](http://xxx)：
  1
  r"\!\[.*?\]\((.*?)\)" # findall
- image_urls → state。
保存聊天记录（回答）。
返回 state。

1	return state