Advanced-RAG

发表于 | 更新于 | 分类于 | 阅读 次
改写、多路召回、精排串成可归因流水线；每一跳必须证明增益大于时延。

Naive 的漏检与噪声在准确率敏感场景不可接受。根因是单查询、单路、无精排。本方法用改写、多路召回和重排拉高可达性与排序。代价是链路变长，需用阶段指标分别归因。它不是「再换一个更大的嵌入模型」的同义词。

本文属于 RAG 工程框架中的「2 索引与召回」环节，聚焦「Advanced RAG」方法。

定位

维度 内容
角色 召回路径的可插拔流水线
输入 → 输出 query → 改写后的多路 hits → 精排后的 Top-n
默认组合 LlamaIndex / Haystack：改写 + Hybrid + reranker
何时不用 POC、流量极低且 Naive 已够；不要一上来就上 Agent

核心机制

$$
T=\mathrm{Rerank}\bigl(\bigcup_i \mathrm{Retrieve}(q’_i)\bigr),\quad {q’_i}=\mathrm{Rewrite}(q)
$$

只在 $\Delta\mathrm{Recall}$ 或 $\Delta\mathrm{nDCG}$ 覆盖额外时延时保留该跳。

实现路径与心智：把 Naive 的单跳拆成可插拔段：改写 $q\to{q’}$ → 多路取并集 → 精排 → 可选压缩 → 生成。底层心智：每一跳都必须用冻结指标证明「增益 > 时延/费用」，否则只是把链路变长。归因时一次只动一段。

优缺点

  • 优点：漏检和噪声可分开打。
  • 缺点：调参面大；不冻结指标就无法归因。

契约与走通样例

输入

1
{"query": "WGS 文库怎么建"}

中间量

改写得 ["whole-genome sequencing library prep", "WGS 文库构建"]。单路 Recall@5=$0.40$；并集后 $0.72$；精排后 nDCG@5 从 $0.51$ 到 $0.68$。时延 $2.8\times$ Naive，仍低于门禁。

输出

1
[{"rank": 1, "doc_id": "P-WGS-03-C2", "score": 0.91, "channel": "dense+rerank"}]

社区实现

LlamaIndex / Haystack 流水线。风险：把切分/检索/生成缠在一个 RAG 类里，冻结评测做不到。

工程落地

最小可运行示例

复制为 .py 后直接运行（仅标准库）。生产流水线对应 LlamaIndex / Haystack：改写器 + 多路 retriever + Cross-encoder。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
"""改写 → 多路并集 → 按原问权重精排。每次只开一段才能归因。"""
from __future__ import annotations

from collections import Counter


class SynonymRewriter:
    """词表改写。生产对应 LLM query transform / HyDE。"""

    def expand(self, query: str) -> list[str]:
        mapping = {"WGS": "whole-genome sequencing library prep"}
        extra = [mapping[w] for w in query.split() if w in mapping]
        return [query, *extra]


class KeywordRetriever:
    """一路词项召回。生产对应 BM25 或 dense 中的一路。"""

    def __init__(self, corpus: dict[str, str], name: str) -> None:
        self.corpus, self.name = corpus, name

    def search(self, query: str, k: int) -> list[str]:
        q = set(query.lower().split())
        scored = sorted(
            self.corpus,
            key=lambda i: -len(q & set(self.corpus[i].lower().split())),
        )
        return scored[:k]


class OverlapReranker:
    """用原问对并集重排。生产对应 bge-reranker。"""

    def rank(self, query: str, pool: list[str], corpus: dict[str, str]) -> list[str]:
        q = Counter(query.lower().split())
        def score(i: str) -> int:
            return sum(q[t] for t in corpus[i].lower().split())
        return sorted(set(pool), key=score, reverse=True)


def advanced_retrieve(query: str, rewriter: SynonymRewriter, retrievers: list[KeywordRetriever], reranker: OverlapReranker, corpus: dict[str, str], pool_k: int = 2, top_n: int = 1) -> list[str]:
    """输入原问；输出精排后的 doc_id。"""
    queries = rewriter.expand(query)
    pool = [doc_id for r in retrievers for q in queries for doc_id in r.search(q, pool_k)]
    return reranker.rank(query, pool, corpus)[:top_n]


if __name__ == "__main__":
    corpus = {
        "P-WGS-03-C2": "whole-genome sequencing library prep adapter ligation",
        "P-unrelated": "western blot loading control",
    }
    print(
        advanced_retrieve(
            "WGS 文库怎么建",
            SynonymRewriter(),
            [KeywordRetriever(corpus, "bm25"), KeywordRetriever(corpus, "dense")],
            OverlapReranker(),
            corpus,
        )
    )

参数

参数 起点 影响
改写条数 2～4 过多主题漂移
pool_k 50 精排池宽度
top_n 8 进生成的条数

失效—信号—螺丝

  • 一次改三跳：分数涨了不知道谁的功。螺丝：一次只开一段。
  • 改写漂移：Recall 降。螺丝：冻结评测，砍改写条数。
  • 精排池不含金标准：nDCG 不动。螺丝：先抬召回。

规模（100 篇生物学 PDF）

建库与 Naive 同量级；GPU 16–24 GB（多模型）；磁盘 3–8 GB。单次查询时延常为 Naive 的 2–6×。

-------------本文结束感谢您的阅读-------------