AI融合RAG实践笔记

先复现,再谈优化。

去年做了一个知识问答系统,当时觉得向量检索 + LLM 就是银弹,但跑了一段时间发现几个很难忽视的问题:

有些问题明明知识库里就有,但向量相似度死活排不到前面。

问题从哪里开始

去年做了一个知识问答系统,当时觉得向量检索 + LLM 就是银弹,但跑了一段时间发现几个很难忽视的问题:

有些问题明明知识库里就有,但向量相似度死活排不到前面。比如问"系统报错代码 50003 怎么处理",而知识库里写的是"错误码 50003 对应的服务异常处理流程",语义确实接近,但就是经常被其他似是而非的文档挤下去。

还有时候用户问具体参数,比如"timeout 参数最大能设多少",向量检索可能返回一段整体配置说明,而答案其实藏在某个 API 文档的参数列表里。

更头疼的是专业术语和品牌名称,向量模型对"Oracle 数据库"和"Oracle 公司"、“PostgreSQL"和"Postgres"这类相似但不同的概念经常搞混。

这些问题不是简单的"模型不够强"或"向量维度不够多”,而是单一检索方式的天生局限。向量检索擅长语义匹配,但缺少对关键词、结构、上下文的综合判断。

融合RAG是什么

说人话就是:不要只依赖一种检索方式,把多种检索结果按规则或权重混在一起,给 LLM 提供更丰富的上下文。

常见的融合方式包括:

  • 关键词检索 + 向量检索:用 BM25 或 Elasticsearch 的全文搜索找精确匹配,用向量检索找语义相关,然后合并去重、重排序
  • 多路向量检索:对不同领域的数据建立多个向量库,比如技术文档一套向量、产品手册一套向量,检索时按场景调用不同来源
  • 结构化检索 + 非结构化检索:对数据库里的结构化数据用 SQL 查询,对文档内容用向量检索,两部分结果拼在一起
  • 多模型检索:用不同的嵌入模型(如 BGE、OpenAI text-embedding-3、Cohere)分别检索,把结果融合

关键是"融合"不是简单的并集,要有重排序和权重分配,否则只是把垃圾混在一起而已。

我的实践场景

我们系统里的知识库大致分三类:

  1. 技术文档:API 说明、架构设计、故障排查,约 1.5 万篇
  2. 产品手册:用户指南、配置参数、功能说明,约 8000 篇
  3. 历史问答:以前处理过的工单和答疑记录,约 2000 条

原来的做法是把所有内容灌到一个向量库里,用户提问时只做一次向量检索,取 top-10 传给 LLM。

问题就出在这里:技术文档和产品手册的写作风格、术语密度完全不同,用同一个嵌入模型和同一个检索阈值,很多场景下要么召回太少要么噪音太多。

融合方案设计

第一步:拆分知识库

先把知识库按类型拆成三个独立的向量索引:

# 向量索引配置
index_configs = {
    "technical_docs": {
        "embedding_model": "BAAI/bge-large-zh-v1.5",
        "chunk_size": 512,
        "overlap": 64,
        "metadata_fields": ["category", "version", "component"]
    },
    "product_manuals": {
        "embedding_model": "BAAI/bge-base-zh-v1.5",
        "chunk_size": 256,
        "overlap": 32,
        "metadata_fields": ["product", "module", "language"]
    },
    "qa_history": {
        "embedding_model": "text-embedding-3-small",
        "chunk_size": 384,
        "overlap": 48,
        "metadata_fields": ["resolved_by", "satisfaction", "timestamp"]
    }
}

不同索引用不同模型和切分策略,技术文档密度高、专业术语多,用大模型;产品手册更偏通俗,用基模型就够;历史问答可以直接用 OpenAI 的小模型。

第二步:多路检索

提问时同时发起多个检索请求:

async def multi_retrieve(query: str, top_k: int = 8):
    # 并发检索
    tasks = [
        vector_search("technical_docs", query, top_k),
        vector_search("product_manuals", query, top_k),
        vector_search("qa_history", query, top_k),
        keyword_search(query, top_k // 2)  # 关键词检索少取一些
    ]
    results = await asyncio.gather(*tasks)
    return results

每个索引独立检索,按类型取合适的结果数量。关键词检索可以少取一些,主要是兜底和补位。

下面这张图展示了从用户提问到最终融合结果的完整流程,方便你把几个步骤串起来看:

flowchart TD A[用户提问] --> B[并发多路检索] B --> B1[技术文档向量检索] B --> B2[产品手册向量检索] B --> B3[历史问答向量检索] B --> B4[关键词检索] B1 --> C[标记来源权重] B2 --> C B3 --> C B4 --> C C --> D[去重处理] D --> E[交叉编码器重排序] E --> F[最终评分排序] F --> G[组装上下文] G --> H[传给 LLM 生成答案] style A fill:#e3f2fd,stroke:#2196f3 style H fill:#e8f5e9,stroke:#4caf50

第三步:结果融合与重排序

拿到多路结果后,需要统一评分并重新排序。这里我用了三个策略:

def fuse_results(retrieval_results: List[List[Dict]], query: str) -> List[Dict]:
    # 1. 标记来源并打基础权重
    labeled = []
    source_weights = {
        "technical_docs": 1.0,
        "product_manuals": 1.0,
        "qa_history": 0.8,
        "keyword": 0.9
    }

    for i, source_results in enumerate(retrieval_results):
        source_name = ["technical_docs", "product_manuals", "qa_history", "keyword"][i]
        for doc in source_results:
            doc["source"] = source_name
            doc["base_weight"] = source_weights[source_name]
            labeled.append(doc)

    # 2. 去重:同一文档的多个片段只保留分数最高的
    deduped = deduplicate_by_doc_id(labeled)

    # 3. 重排序:用交叉编码器打分
    reranked = rerank_with_cross_encoder(query, deduped)

    # 4. 最终评分 = 重排序分数 × 来源权重
    for doc in reranked:
        doc["final_score"] = doc["rerank_score"] * doc["base_weight"]

    # 按最终分数排序,取 top-k
    return sorted(reranked, key=lambda x: x["final_score"], reverse=True)[:8]

来源权重是根据经验调的,历史问答的质量不稳定,权重稍低;关键词检索作为兜底,权重略高于技术文档但低于其他向量检索。

第四步:上下文组装

给 LLM 的上下文需要保留来源信息,方便它判断哪些内容更可信:

def build_context(fused_results: List[Dict]) -> str:
    context_parts = []
    for i, doc in enumerate(fused_results, 1):
        source_label = {
            "technical_docs": "【技术文档】",
            "product_manuals": "【产品手册】",
            "qa_history": "【历史问答】",
            "keyword": "【关键词检索】"
        }[doc["source"]]

        context_parts.append(
            f"{source_label}(相关度 {doc['final_score']:.2f}):\n"
            f"{doc['content']}"
        )

    return "\n\n".join(context_parts)

这样 LLM 在生成答案时能知道哪些内容来自官方文档,哪些来自历史经验,哪些是关键词匹配兜底的。

踩过的坑

坑一:去重策略不当导致信息丢失

刚开始对同一文档的多个切片做了硬去重,只保留相似度最高的那个。结果发现有些复杂问题需要文档的不同段落拼起来才能回答完整,去重后反而丢信息。

后来改成按文档 ID 去重,但保留每个文档得分最高的前 2 个片段,且确保这些片段在原文中至少相隔一定距离(避免连续片段被同时选中)。

坑二:重排序延迟太大

用 BGE-reranker-large 做交叉编码器,单次重排序 50 个文档需要 2-3 秒,直接把问答延迟拉高到不可接受。

优化办法有两个:

  • 先用轻量模型做粗排(比如 BGE-reranker-base),保留前 20 个再用大模型精排
  • 对历史问答和关键词检索的结果不做交叉编码,只对向量检索结果精排

坑三:来源权重不好调

一开始所有来源权重都设成 1.0,结果关键词检索返回的噪音太多,因为 BM25 容易匹配到不相关的短词。后来把关键词权重降到 0.7,又发现一些专有名词或错误码查询完全找不到结果。

最终折衷方案是:根据问题类型动态调整权重。如果问题包含明显的专业术语或数字编号,提高关键词权重;如果是自然语言描述的问题,降低关键词权重。

坑四:上下文太长导致 LLM 跑偏

融合后的结果通常比单一检索多,有时候 top-8 的内容加起来超过 4000 token,传给 LLM 后反而让它注意力分散,无法聚焦核心答案。

解决办法是限制总 token 数,当上下文超过阈值时,按最终分数截断,或者只保留分数明显高于平均值的片段。

结果和边界

改用融合 RAG 后,几个核心指标有明显改善:

  • 准确率从 58% 提升到 72%(人工评估 500 个样本)
  • 关键参数、错误码类问题的召回率从 41% 提升到 68%
  • 用户反馈"答非所问"的工单减少约 30%

但也有几个边界问题始终没解决:

  • 跨文档推理依然困难:比如需要对比两个产品的配置差异,融合检索能找到两份文档,但 LLM 不一定能准确提炼对比点
  • 最新内容的时效性:向量索引更新有延迟,如果是当天发布的紧急故障说明,还是得靠关键词检索兜底
  • 多轮对话的上下文传递:融合检索只处理当前问题,没有利用对话历史,用户说"那个方法不行"时,系统不知道"那个方法"指什么

这些问题不是融合 RAG 能单独解决的,需要结合上下文检索、时间衰减、对话状态管理等更复杂的机制。

一些实用建议

如果你的场景也在考虑融合 RAG,这几条经验可以参考:

  1. 先确认单一检索的瓶颈在哪里:是召回不够全,还是排序不够准,还是上下文不够丰富?不同问题对应不同融合策略
  2. 不要一开始就上太多路数:先试向量 + 关键词双路,效果不够再考虑多模型或多索引
  3. 重排序很有用,但要注意延迟:轻量粗排 + 重量精排是常见折衷方案
  4. 来源权重宁可保守一点:宁可少召回一些,也不要让低质量内容冲淡高价值信息
  5. 上线后要持续监控:哪些来源经常被选中、哪些类型的问题经常答错,这些数据能帮你调优权重和策略

结语

融合 RAG 本质上是在"越多越好"和"越准越好"之间找平衡。单一检索简洁但视野有限,多路检索视野宽但需要更精细的控制。

这件事没有什么万能配方,更多是踩坑和调优的过程。但如果你也遇到过"明明有答案就是检索不到"的窘境,融合 RAG 至少提供了一个值得尝试的方向。

至于最终效果能提升多少,取决于你的数据质量、问题类型和用户预期。技术方案能改善上限,但很多细节还得靠你自己去磨。


参考阅读

  • Fusion RAG 相关论文和实践(需要你自己搜一下,我就不硬补了)
  • BGE 系列嵌入和重排序模型:https://github.com/FlagOpen/FlagEmbedding
  • Elasticsearch BM25 实现细节

版权声明: 本文首发于 指尖魔法屋-AI融合RAG实践笔记https://blog.thinkmoon.cn/post/381-ai-fusion-rag-single-fusion-practice/) 转载或引用必须申明原指尖魔法屋来源及源地址!