RAG 进阶：查询改写、混合检索和重排

A5 的咖啡店问答只有 4 段资料，怎么检索都不会错。资料涨到几千段、用户问得很随意以后，RAG 答错的原因大多不在模型，而在检索没把对的段落找出来。这一课沿着一次提问的路径，逐段改进：

用户问题 → 查询改写 → 关键词检索 + 向量检索 → RRF 合并 → 重排 → 带引用回答

基础 RAG 为什么会漏

情况	例子	问题出在哪
用户和文档说法不同	问「能退钱吗」，文档写「退款政策」	向量能解决一部分，但口语和术语差得远时仍会漏
专有名词、编号	「SKU-2041 有货吗」「错误码 E103」	向量对精确字面不敏感，关键词检索更准
一句问了几件事	「积分怎么算，过期还能用吗，大杯加多少钱」	一个向量是几件事的平均，哪段都不太像
追问	上一句问营业时间，这一句「那周末呢」	只拿这一句去检索，根本不知道在问什么

排查顺序和 A5 一样：先打印检索结果。检索对了答案还错，再去改生成的提示词。

切块：按结构切，块里带上标题

A5 讲过块的大小和重叠，这里补几条容易忽略的：

按标题和段落切，不要按固定字数硬切。一个小节讲一件事，切在小节边界上，块的主题更集中。
块里带上标题路径。单独一句「自获得之日起 12 个月内有效」，看不出说的是积分还是优惠券；加上「会员手册 > 积分 > 有效期」，向量、关键词和模型都能看懂。
小节太长再按字数切，相邻块重叠 10%–20%。
FAQ 一问一答为一块；表格按行切，每块带上表头。

python

import re

def chunk_markdown(source, text, max_chars=400, overlap=60):
    """按 Markdown 标题切块，每块开头带标题路径；过长的小节按字数切并重叠。"""
    chunks, path, buf = [], [], []

    def flush():
        body = "\n".join(buf).strip()
        buf.clear()
        if not body:
            return
        title = " > ".join([source] + path)
        for start in range(0, len(body), max_chars - overlap):
            chunks.append({"source": title, "text": f"{title}\n{body[start:start + max_chars]}"})
            if start + max_chars >= len(body):
                break

    for line in text.splitlines():
        m = re.match(r"^(#{1,6})\s+(.*)", line)
        if m:
            flush()
            level = len(m.group(1))
            path = path[:level - 1] + [m.group(2).strip()]
        else:
            buf.append(line)
    flush()
    return chunks

查询改写

检索前先让模型把用户的话改写一遍，产出三样东西：

独立问题：结合对话历史补全指代。「那周末呢」→「小蜂咖啡周末几点开门、几点关门？」
子查询：一句问了几件事，就拆成几条，每条单独检索。
关键词：专有名词、编号、数字，交给关键词检索。

python

import json, os
from openai import OpenAI

client = OpenAI(base_url="https://hivegpt.cn/v1", api_key=os.environ["HIVEGPT_API_KEY"])

REWRITE_SCHEMA = {
    "name": "query_rewrite",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "standalone_question": {"type": "string"},
            "sub_queries": {"type": "array", "items": {"type": "string"}},
            "keywords": {"type": "array", "items": {"type": "string"}},
        },
        "required": ["standalone_question", "sub_queries", "keywords"],
        "additionalProperties": False,
    },
}

def rewrite(history, question):
    chat = "\n".join(f"{m['role']}：{m['content']}" for m in history[-6:])   # 最近几轮就够
    resp = client.chat.completions.create(
        model="gpt-5.5",
        messages=[
            {"role": "system", "content": "把用户的新问题改写成检索用的查询：补全指代得到独立问题；一句问了几件事就拆成子查询，每条只问一件事；列出专有名词、编号等关键词。不要回答问题。"},
            {"role": "user", "content": f"对话记录：\n{chat}\n\n新问题：{question}"},
        ],
        response_format={"type": "json_schema", "json_schema": REWRITE_SCHEMA},
    )
    return json.loads(resp.choices[0].message.content)

改写多一次模型调用，会增加延迟。对话刚开始、问题又短又清楚时，可以跳过改写直接检索。

动手：改写一个追问

▶ 动手试试

系统提示词（这次请求一起发送，点开查看）

你负责把用户的新问题改写成检索用的查询，不要回答问题。结合对话记录补全指代，得到一个不看上下文也能看懂的独立问题；一句问了几件事就拆成子查询，每条只问一件事；列出专有名词、商品名、编号、数字等适合关键词检索的词。

登录后运行登录 HiveGPT 后每天有免费运行次数

示例输出（之前运行的结果）

{
  "standalone_question": "小蜂咖啡周末几点开门、几点关门？买燕麦拿铁可以用会员积分抵扣吗？",
  "sub_queries": ["小蜂咖啡周末营业时间", "会员积分抵扣规则", "燕麦拿铁能否用积分抵扣"],
  "keywords": ["周末", "营业时间", "燕麦拿铁", "积分", "抵扣"]
}

把新问题换成「大杯呢？」「那上海的店呢？」，看独立问题补得对不对。

混合检索：BM25 + 向量

两种检索各有所长：

向量检索擅长意思相近、说法不同：「几点开门」能找到「营业时间」。
**关键词检索（BM25）**擅长精确字面：型号、错误码、人名、数字。中文要先分词，常用 jieba。

两路分别取前 30 个，再合并。BM25 分数和余弦相似度的量纲不同，不能直接相加；常用的合并方法是 RRF（Reciprocal Rank Fusion，倒数排名融合），只看名次：

RRF(d) = Σ 1 / (k + rank_r(d))      对每一路检索结果 r 求和

rank_r(d) 是文档 d 在第 r 路结果里的名次（从 1 开始），没出现就不计分；k 通常取 60，用来压低「只在一路里排第一」的优势。比如某段在 BM25 里排第 1、向量里排第 5，得分 1/61 + 1/65 ≈ 0.0318；另一段只在向量里排第 2，得分 1/62 ≈ 0.0161。两路都靠前的段落排在前面。

向量接口用哪个分组

HiveGPT 的 /v1/embeddings 需要「GPT-按量」分组的 Key，订阅类分组不支持 Embedding。也可以继续用 A5 里本地的 BAAI/bge-small-zh-v1.5，把下面的 embed 换掉即可，其余代码不变。

python

# pip install openai numpy rank_bm25 jieba
import jieba
import numpy as np
from rank_bm25 import BM25Okapi

def embed(texts):
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)   # 资料多时分批，每批几十条
    vecs = np.array([d.embedding for d in resp.data])
    return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)

CHUNKS = chunk_markdown("会员手册", open("handbook.md", encoding="utf-8").read())
TEXTS = [c["text"] for c in CHUNKS]
bm25 = BM25Okapi([jieba.lcut(t) for t in TEXTS])
chunk_vecs = embed(TEXTS)

def keyword_search(query, n=30):
    scores = bm25.get_scores(jieba.lcut(query))
    return [int(i) for i in np.argsort(-scores)[:n] if scores[i] > 0]

def vector_search(query, n=30):
    scores = chunk_vecs @ embed([query])[0]
    return [int(i) for i in np.argsort(-scores)[:n]]

def rrf(rankings, k=60):
    scores = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)

def hybrid_search(queries, keywords=(), n=30):
    rankings = []
    for q in queries:                       # 独立问题和每条子查询都检索一遍
        rankings.append(vector_search(q, n))
        rankings.append(keyword_search(q + " " + " ".join(keywords), n))
    return rrf(rankings)[:n]

重排：先多取，再精选

混合检索要的是别漏，所以多取一些（20–50 段）；交给模型回答的只要最相关的 3–5 段。中间这一步就是重排：把问题和每一段放在一起逐一判断相关程度，比只比较两个向量准得多。

两种做法：

重排模型（cross-encoder），例如开源的 BAAI/bge-reranker-v2-m3，支持中文，可以本地跑，一次给几十段打分：

python

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
scores = reranker.predict([(question, TEXTS[i]) for i in candidates])

用大模型重排：不用部署新模型，还能顺便给出理由。代价是多一次调用，段落多时 token 不少，每段截到 300 字左右再送进去。

python

RERANK_SCHEMA = {
    "name": "rerank",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "ranking": {"type": "array", "items": {
                "type": "object",
                "properties": {"id": {"type": "integer"}, "score": {"type": "integer"}, "reason": {"type": "string"}},
                "required": ["id", "score", "reason"],
                "additionalProperties": False,
            }},
        },
        "required": ["ranking"],
        "additionalProperties": False,
    },
}

def llm_rerank(question, candidates):
    passages = "\n".join(f"[{j}] {TEXTS[i][:300]}" for j, i in enumerate(candidates, start=1))
    resp = client.chat.completions.create(
        model="gpt-5.5",
        messages=[
            {"role": "system", "content": "按和问题的相关程度给每段打分（0–10），从高到低排列。只看段落能不能回答问题，不看字面像不像。"},
            {"role": "user", "content": f"问题：{question}\n\n候选段落：\n{passages}"},
        ],
        response_format={"type": "json_schema", "json_schema": RERANK_SCHEMA},
    )
    ranking = json.loads(resp.choices[0].message.content)["ranking"]
    return [{"chunk": candidates[r["id"] - 1], "score": r["score"]}
            for r in ranking if 1 <= r["id"] <= len(candidates)]   # 丢掉越界的编号

动手：给 5 段候选重排

第 3 段讲的是优惠券过期，字面上和问题很像，向量检索常把它排得很靠前。看看重排能不能把它压下去：

▶ 动手试试

系统提示词（这次请求一起发送，点开查看）

你是检索系统的重排器。按每段候选和问题的相关程度打分：10 表示能直接回答问题，0 表示无关。只看段落能不能回答问题，不看字面像不像。所有候选都要出现在 ranking 里，按分数从高到低排列；reason 用一句话说明。

登录后运行登录 HiveGPT 后每天有免费运行次数

示例输出（之前运行的结果）

{
  "ranking": [
    {"id": 2, "score": 10, "reason": "直接说明积分过期后清零、不能补发"},
    {"id": 1, "score": 4, "reason": "讲积分怎么积和抵扣，没提过期"},
    {"id": 5, "score": 2, "reason": "讲会员等级，和积分过期无关"},
    {"id": 3, "score": 1, "reason": "讲的是优惠券过期，不是积分"},
    {"id": 4, "score": 0, "reason": "营业时间，无关"}
  ]
}

引用来源，答不上就说没有

把重排后的几段带上编号和来源交给模型，沿用 A5 的要求：只根据资料回答，句末标 [1]、[2]，资料里没有就直说。再加两道程序里的检查：

分数门槛：重排后最高分都低于门槛（比如 5 分），就不调用生成，直接回答「资料里没有找到相关内容」，同时把这个问题记进日志。这些问题就是知识库要补的内容。
核对引用：回答里出现的编号必须是这次真的给过的段落，对不上的编号删掉或整条回答重试。

python

def answer(history, question):
    q = rewrite(history, question)
    candidates = hybrid_search([q["standalone_question"], *q["sub_queries"]], q["keywords"])
    top = [r for r in llm_rerank(q["standalone_question"], candidates) if r["score"] >= 5][:4]
    if not top:
        print("未命中：", question)                       # 记日志，补资料
        return "资料里没有找到相关内容。"
    context = "\n".join(f"[{n}]（{CHUNKS[r['chunk']]['source']}）{TEXTS[r['chunk']]}"
                        for n, r in enumerate(top, start=1))
    resp = client.chat.completions.create(model="gpt-5.5", messages=[
        {"role": "system", "content": "只根据「资料」回答，句末用 [1]、[2] 标出依据；资料里没有的内容，直接说「资料里没有提到」，不要猜。"},
        {"role": "user", "content": f"资料：\n{context}\n\n问题：{q['standalone_question']}"},
    ])
    return resp.choices[0].message.content

单独衡量检索

改了切块、加了混合检索，效果到底变好没有？不要只看几次回答的感觉，把检索单独拿出来测。

准备 30–100 个真实问题，每个标出「应该被找到的段落」。标注时记来源和一小段原文，不要只记序号，换了切块方式还能对上。
对每种检索方案算 recall@k：前 k 个结果里，找回了多少比例的相关段落，再对所有问题取平均。
每次只改一处，对比改动前后的 recall@5、recall@20。

python

TESTSET = [   # relevant 是相关块的序号，示例用；实际按上面第 1 条存来源和原文
    {"q": "周末几点关门", "relevant": {3}},
    {"q": "积分过期还能用吗", "relevant": {7, 8}},
]

def recall_at_k(search_fn, k=5):
    total = 0
    for case in TESTSET:
        got = set(search_fn(case["q"])[:k])
        total += len(got & case["relevant"]) / len(case["relevant"])
    return total / len(TESTSET)

print("向量", recall_at_k(vector_search))
print("混合", recall_at_k(lambda q: hybrid_search([q])))

recall@20 低，说明候选里就没有，要改切块、改写或检索；recall@20 高但 recall@5 低，说明找到了但排得靠后，该加重排。检索的指标达标以后，再去评估回答本身（H2 的方法）。

小结

基础 RAG 漏检，多半因为说法不同、专有名词、一句多问和追问；切块按结构切，块里带标题路径。
查询改写补全指代、拆子查询、提关键词；BM25 和向量两路检索用 RRF 合并，再对前 20–50 段重排。
引用要核对，分数不够就答「资料里没有」并记日志；用 recall@k 在标注集上单独衡量检索，每次只改一处。

下一课：一个任务拆给多个模型调用——工作流与多智能体：路由、并行和评审。

RAG 进阶：查询改写、混合检索和重排

基础 RAG 为什么会漏 ​

切块：按结构切，块里带上标题 ​

查询改写 ​

动手：改写一个追问 ​

混合检索：BM25 + 向量 ​

重排：先多取，再精选 ​

动手：给 5 段候选重排 ​

引用来源，答不上就说没有 ​

单独衡量检索 ​

小结 ​

基础 RAG 为什么会漏

切块：按结构切，块里带上标题

查询改写

动手：改写一个追问

混合检索：BM25 + 向量

重排：先多取，再精选

动手：给 5 段候选重排

引用来源，答不上就说没有

单独衡量检索

小结