用权重加 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。

步骤：

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

  2. 封装数据结构 → list：

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

  3. 使用 RRF 算法进行数据处理（list）→ rrf_chunks[5]。

  4. 修改 state = rrf_chunks。

  5. 返回 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 的三个理由：

  1. 适配语言（多语言支持友好）。
  2. 性能的问题（重排效果更好）。
  3. 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 内容，你是如何处理的？

  1. 换模型 → 需要重新评估。

  2. 压缩比较内容：把 content 用 LLM 压缩后再比较打分（选择该方案）。

    1
    2
    508 - 问题的 token = 答案 token
    答案 token > token -> 启动压缩 -> 调用模型

  3. 冷处理 → 不管。

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
1
2
3
4
5
6
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
1
2
3
4
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：

1
2
3
4
5
6
# 重排序模型配置
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 常量配置

1
2
3
4
5
6
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 业务步骤

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

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

    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。

  3. 组装问题和答案的列表（rewritten_query, reranker_list）→ question_answer_pair_list [[], [], []]：

    • 获取问题。

    • 判断问题 token 的长度。

    • 循环 reranker_list 获取答案 text：

      • 判断 text 的长度。

        • 超长 → 模型压缩 → 调用模型。

      • 装数据 pair。

    • 返回结果。

  4. reranker 模型打分 + 排序：

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

  5. 动态 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。

步骤（流式 / 非流式）：

1
2
3
4
非流式 -> 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，直接返回，不调用模型、不处理图片。

  1. 判断 state 是否有 answer，有就直接返回（state）→ 用 bool 判断有没有 answer：

    • 没有 → return false。
    • 有 → 流式 push_to_session(session_id, delta, {delta: answer})；非流式 return True，state => answer。

  2. 上一次返回的值是 false（即没有 answer），继续往下走。

  3. answer_prompt_text = 声明一个提示词拼接的方法（state）进行拼接。

  4. 调用模型处理字符串 answer 回答的问题（state, answer_prompt_text）：

    1
    2
    流式   llm - stream  -> push ... -> state.answer
    非流式 llm - invoke  -> state -> answer

    state[answer] = ...

  5. 提取图片存储到 state（reranked 里的 text ![]()、mcp 的 url → 判断是不是图片，写入 state）：

    • url → 看后缀名。

    • text → 网络搜索的 markdown ![xxxx](http://xxx)：

      1
      r"\!\[.*?\]\((.*?)\)"   # findall

    • image_urls → state。

  6. 保存聊天记录（回答）。

  7. 返回 state。

1
return state