RAG 实战完全指南:为什么你的检索增强问答总是一本正经胡说八道

2024 年,我给公司做一个内部知识库问答系统。我们有几百份产品文档、API 手册、运维 SOP,散落在 Confluence、Git 仓库、各种 Word 文件里,新人入职找一份资料,像大海捞针。我当时的想法很直接:接一个大模型,做成一个"问答机器人",不就解决了?可大模型本身,根本不知道我们公司内部的任何东西。于是我用了 RAG(检索增强生成)——把所有文档切碎、存进一个向量库;用户提问时,先去向量库里【检索】出最相关的几个文档片段,再把这些片段连同用户的问题,一起发给大模型,让它"看着资料回答"。第一版做出来,我给老板演示,问它"我们的订单接口怎么分页?",它答得头头是道,老板很满意。可真正上线给同事用,投诉立刻就来了:有人问"测试环境数据库密码在哪",它一本正经地【编了一个】;有人问"报销流程几步",它把三份不同文档的内容【缝在了一起】、张冠李戴;还有人问一个在文档里明明写得清清楚楚的问题,它却回答"未找到相关信息"。我起初以为是模型不够强,想换个更贵的模型。后来,我盯着它每次【检索回来的片段】看,愣住了:用户问"报销流程",检索回来的五段里,只有一段是真的相关,另外四段,是"流程""额度""出差申请"这些【字面上沾点边、内容上八竿子打不着】的东西。模型不是不行——是我喂给它的"参考资料"本身就错了。它只是忠实地,基于一堆错料,编出了一个像模像样的答案。这件事逼着我把 RAG 到底分几步、文档该怎么切、embedding 怎么选、检索怎么提高召回、prompt 又该怎么拼,彻底理清了一遍。本文是这份梳理的完整复盘。

问题背景:一个"一本正经胡说八道"的知识库问答

需求:把几百份散落的内部文档,做成一个能问答的机器人
我的方案:RAG —— 文档切碎存进向量库,提问时先检索相关
          片段,再连同问题发给大模型"看着资料回答"

demo 时挺好,上线后投诉不断:
- ★★ 问"测试环境数据库密码" —— 它一本正经【编了一个】
- ★★ 问"报销流程几步" —— 把三份文档内容【缝在一起】、
     张冠李戴
- ★ 问一个文档里写得清清楚楚的问题 —— 它说"未找到相关信息"

★ 我以为是模型不够强,想换个更贵的模型 —— 方向错了。

★★ 真相:我盯着它每次【检索回来的片段】看 —— 用户问
   "报销流程",检索回的五段里只有一段真相关,另外四段是
   字面沾边、内容无关的东西。模型不是不行,是我喂给它的
   "参考资料"本身就错了 —— 它只是忠实地基于错料编答案。

★ 本文要做的:把 RAG 分几步、文档怎么切、embedding 怎么
  选、检索怎么提召回、prompt 怎么拼,彻底讲透。

为什么你的 RAG 总是答非所问:先看清它到底分几步

# === ★ 想修好 RAG,先得知道它是一条几个环节的"流水线" ===

# === ★ RAG 不是一个动作,是一条三段式流水线 ===
# ★ ★ 很多人(包括当初的我)以为 RAG 就是"检索一下再回答"
#   这一个动作。错。它是一条至少分【三大段】的流水线:
#  - ★【索引(Indexing)】:离线阶段。把文档切块、算
#    embedding、存进向量库。这步【在用户提问之前】就做完了。
#  - ★【检索(Retrieval)】:在线阶段。用户一提问,把问题
#    也转成向量,去向量库里捞出最相关的若干个片段。
#  - ★【生成(Generation)】:把捞回的片段 + 用户问题,
#    拼成一个 prompt,发给大模型,让它生成最终答案。

# === ★★ 关键认知:这是一条"木桶",最短的板决定成败 ===
# ★ ★ 这三段是【严格串行】的:索引切坏了,检索就检不准;
#   检索捞回一堆垃圾,生成就只能基于垃圾编答案。
# ★ ★★ 所以,RAG 答非所问,问题【几乎从不在最后那个
#   大模型】。我当初想"换个更强的模型",方向就是错的 ——
#   模型再强,你喂它一堆错的参考资料,它也只会把错的答案
#   编得【更可信】而已。问题,几乎总是出在前面的索引和
#   检索环节。

# === ★ 一个最朴素、最有效的排查手段 ===
# ★ ★ RAG 出了问题,别急着调 prompt、别急着换模型。第一步,
#   永远是:把【检索环节实际捞回来的那几个片段】,原原本本
#   打印出来,自己【用人眼读一遍】。
# ★ ★★ 你只需要问自己一个问题:"如果我是模型,只拿着这
#   几段文字,我答得出用户的问题吗?"
#  - 答不出 -> 问题在【检索/索引】,往前查;
#  - 答得出、但模型还是答错了 -> 问题才在【生成】环节。
# ★ 我那次的事故,就是这样定位的:打印出来一看,五段里
#   四段是垃圾 —— 根本不用再怀疑模型。

# === 小结 ===
# ★ RAG 不是一个动作,是一条至少分三大段的流水线:索引
#   (离线 —— 文档切块、算 embedding、存进向量库,在用户
#   提问前就做完)、检索(在线 —— 用户提问把问题转成向量
#   去向量库捞最相关的若干片段)、生成(把捞回的片段+用户
#   问题拼成 prompt 发给大模型生成答案)。★★ 关键认知:
#   这三段严格串行,是一条木桶,最短的板决定成败 —— 索引
#   切坏检索就检不准、检索捞回垃圾生成就只能基于垃圾编;
#   所以 RAG 答非所问问题几乎从不在最后那个大模型,想"换
#   个更强的模型"方向就错了,模型再强喂它错的参考资料它
#   也只会把错答案编得更可信。★ 最朴素有效的排查手段:
#   出问题别急着调 prompt 换模型,第一步永远是把检索实际
#   捞回的那几个片段原样打印出来用人眼读一遍,问自己"如果
#   我是模型只拿这几段我答得出吗" —— 答不出问题在检索/
#   索引往前查,答得出但模型还错了问题才在生成环节。

# ★ RAG 三段式流水线的骨架 —— 看清每一步在干什么
# 索引(离线)、检索(在线)、生成(在线)

# ========== 第一段:索引 Indexing —— 离线,提问前就做完 ==========
def build_index(documents: list[str]):
    chunks = []
    for doc in documents:
        chunks.extend(split_into_chunks(doc))   # ★ 切块(下一节详谈)
    # ★ 每个 chunk 算出一个 embedding 向量,连同原文存进向量库
    vectors = embed_batch(chunks)
    vector_store.add(chunks, vectors)

# ========== 第二段:检索 Retrieval —— 用户一提问就触发 ==========
def retrieve(question: str, top_k: int = 5) -> list[str]:
    q_vector = embed_one(question)               # ★ 问题也转成向量
    # ★ 在向量库里捞出最相似的 top_k 个片段
    hits = vector_store.search(q_vector, top_k=top_k)
    return [h.text for h in hits]

# ========== 第三段:生成 Generation —— 拼 prompt,发给模型 ==========
def generate(question: str, contexts: list[str]) -> str:
    prompt = build_prompt(question, contexts)    # ★ 拼 prompt(后面详谈)
    return call_llm(prompt)

# ========== 串起来:一次完整的 RAG 问答 ==========
def answer(question: str) -> str:
    contexts = retrieve(question)
    # ★★ 排查 RAG 的第一手段:把检索回的片段【原样打印】出来,
    #    用人眼读 —— "拿着这几段,我自己答得出问题吗?"
    for i, c in enumerate(contexts):
        print(f'--- 检索片段 {i + 1} ---\n{c}\n')
    return generate(question, contexts)

# ★★ 三段严格串行:索引切坏 -> 检索检不准 -> 生成基于垃圾编。
#    答非所问,问题几乎总在前两段,而不在最后那个大模型。

文档切块(Chunking):切坏了,后面全白搭

# === ★ 索引的第一步是切块,也是最被低估的一步 ===

# === ★ 为什么文档非切不可 ===
# ★ ★ 你不能把一份三十页的文档整个塞进向量库。原因有二:
#  - ★ 检索要"精准":用户问一个具体问题,你该返回那
#    【相关的一小段】,而不是把整篇三十页都丢给模型;
#  - ★ 模型上下文有限:你也塞不下那么多。
# ★ 所以必须把长文档,切成一个个【大小适中、语义完整】的
#   片段(chunk)。

# === ★★ 最常见、也最致命的错误:按固定字数硬切 ===
# ★ ★ 新手最容易写出的切法:"每 500 个字符切一刀"。这是
#   一场灾难。
# ★ ★★ 因为它对文字的【语义结构】一无所知。它会把一句
#   完整的话从中间劈开;会把一个表格切得七零八落;会把
#   "问题"和紧跟其后的"答案"分到两个 chunk 里。结果:
#   每个 chunk 都成了【语义残缺】的碎片 —— 基于残缺碎片
#   算出的 embedding,自然也是有偏的,检索焉能准?
# ★ 我那次"报销流程"检索回一堆垃圾,根子就在这:文档被
#   按字数粗暴切碎,"报销流程共三步"这句话的上下文,被
#   切得支离破碎。

# === ★ 正确的思路:按"结构"和"语义"切 ===
# ★ ★ 好的切块,要尽量【尊重文档本身的结构】:
#  - ★ 按 Markdown 的标题层级(#、##)切 —— 一个小节,
#    天然就是一个语义完整的单元;
#  - ★ 按段落切 —— 段落之间是天然的语义边界;
#  - ★ 对 FAQ 文档,一个"问题+答案"必须【绑死在一个
#    chunk】里,绝不能拆开。
# ★ 核心原则:每个 chunk,都应该是一个【能独立读懂】的
#   小单元。

# === ★★ chunk 大小,和"重叠"该怎么定 ===
# ★ ★ chunk 太大:一段里塞了好几个主题,embedding 被"平均"
#   得很模糊,检索不精准;chunk 太小:语义不完整,而且
#   检索回来的信息太碎。要在两者间权衡(常见量级:几百
#   个 token)。
# ★ ★★ 一个关键技巧:相邻 chunk 之间,留一点【重叠
#   (overlap)】。比如每个 chunk 的开头,带上一段它前一个
#   chunk 的结尾。这样,即使一个关键句不幸落在切割线上,
#   它至少能【完整地存在于某一个 chunk】里,不至于被劈成
#   两半、两边都不完整。

# === 小结 ===
# ★ 索引第一步是切块,也是最被低估的一步。文档非切不可:
#   检索要精准(用户问具体问题该返回相关的一小段而非整篇)、
#   模型上下文有限,所以必须把长文档切成大小适中、语义完整
#   的片段。★★ 最常见也最致命的错误是按固定字数硬切(每
#   500 字符切一刀):它对文字的语义结构一无所知,会把一句
#   完整的话从中间劈开、把表格切碎、把问题和紧跟的答案分到
#   两个 chunk —— 每个 chunk 都成了语义残缺的碎片,基于残缺
#   碎片算的 embedding 有偏检索焉能准。★ 正确思路是按结构
#   和语义切:按 Markdown 标题层级切(一个小节天然是语义
#   完整单元)、按段落切(段落间是天然语义边界)、FAQ 文档
#   一个"问题+答案"必须绑死在一个 chunk 里;核心原则是每个
#   chunk 都应是能独立读懂的小单元。★★ chunk 大小要权衡:
#   太大一段塞好几个主题 embedding 被平均得模糊、太小语义
#   不完整信息太碎;关键技巧是相邻 chunk 间留一点重叠
#   (overlap),即使关键句落在切割线上也能完整存在于某个
#   chunk 里不被劈成两半。

# ★ 切块对比:固定字数硬切(灾难) vs 按结构切(正确)
import re

# === ★★ 反例:按固定字数硬切 —— 对语义结构一无所知 ===
def bad_chunking(text: str, size: int = 500) -> list[str]:
    # ★★ 致命:它会把句子、表格、"问题+答案"从中间劈开
    return [text[i:i + size] for i in range(0, len(text), size)]

# === ★ 正例:按 Markdown 标题结构切 —— 尊重语义边界 ===
def split_by_heading(markdown: str) -> list[str]:
    # ★ 用 ## 标题作为切割点 —— 一个小节天然是语义完整单元
    sections = re.split(r'\n(?=#{1,3}\s)', markdown)
    return [s.strip() for s in sections if s.strip()]

# === ★ 对过长的小节,再按段落细切,并保留 overlap 重叠 ===
def split_with_overlap(text: str, max_chars: int = 800,
                       overlap: int = 120) -> list[str]:
    paragraphs = [p for p in text.split('\n\n') if p.strip()]
    chunks, buf = [], ''
    for para in paragraphs:
        if len(buf) + len(para) <= max_chars:
            buf = buf + '\n\n' + para if buf else para
        else:
            chunks.append(buf)
            # ★★ 关键:新 chunk 开头带上一段上个 chunk 的结尾
            #    —— 关键句落在切割线上也不会两边都残缺
            tail = buf[-overlap:] if len(buf) > overlap else buf
            buf = tail + '\n\n' + para
    if buf:
        chunks.append(buf)
    return chunks

# === ★ 组合:先按结构切,过长的再按段落切 ===
def smart_chunking(markdown: str) -> list[str]:
    result = []
    for section in split_by_heading(markdown):
        if len(section) <= 800:
            result.append(section)              # ★ 小节够小,整段就是一个 chunk
        else:
            result.extend(split_with_overlap(section))
    return result
# ★★ 每个 chunk 都该能"独立读懂" —— 这是检索准不准的地基

Embedding 与向量检索:把"意思"变成可计算的坐标

# === ★ 切好块了,下一步:怎么让机器"按意思"找文档 ===

# === ★ 先想清楚:为什么不能用关键词去匹配 ===
# ★ ★ 传统搜索是【关键词匹配】:用户搜"报销",就找含
#   "报销"二字的文档。它的死穴是:用户问"出差花的钱怎么
#   要回来",文档里写的是"差旅费用报销流程" —— 一个共同
#   的字都没有,关键词匹配【直接失效】。
# ★ ★ RAG 要解决的,正是这种"意思相同、说法不同"的匹配。

# === ★★ Embedding:把一段文字,变成一串数字坐标 ===
# ★ ★ embedding 模型,能把任意一段文字,转换成一个【高维
#   向量】(比如 1024 个浮点数组成的数组)。
# ★ ★★ 这个转换的神奇之处在于:它让【意思相近】的文字,
#   在这个高维空间里,位置也【靠得近】。"出差花的钱怎么
#   要回来"和"差旅费用报销流程",字面毫无交集,但它们
#   的向量,会落在空间里很接近的两个点上。
# ★ 于是,"按意思找文档"这个模糊的需求,就变成了一道
#   精确的数学题:【在向量空间里,找离问题最近的那些点】。

# === ★ 相似度:怎么衡量两个向量"有多近" ===
# ★ ★ 最常用的是【余弦相似度】—— 它衡量的是两个向量的
#   【方向】有多一致,值域 -1 到 1,越接近 1 越相似。
# ★ 向量库的核心工作,就是:给定问题向量,快速从几十万个
#   文档向量里,算出余弦相似度最高的 top_k 个。

# === ★★ 选 embedding 模型:三个必须留意的点 ===
# ★ ★ ① 中文能力:很多 embedding 模型是英文语料为主训练
#   的,中文场景下,务必选【中文或多语言】表现好的模型。
# ★ ★ ② 一致性铁律:【索引时】和【检索时】,必须用
#   【同一个 embedding 模型】。用 A 模型索引、B 模型检索,
#   两套向量根本不在同一个空间里,算出的相似度【毫无意义】
#   —— 这是一个很隐蔽、却会让整个 RAG 彻底失灵的坑。
# ★ ★ ③ 维度与成本:维度越高,通常越精准,但存储和计算
#   开销也越大。按你的数据量权衡。

# === 小结 ===
# ★ 切好块下一步是让机器按意思找文档。传统搜索是关键词
#   匹配,死穴是用户问"出差花的钱怎么要回来"、文档写"差旅
#   费用报销流程"一个共同字都没有就直接失效;RAG 要解决的
#   正是这种"意思相同说法不同"的匹配。★★ Embedding 模型
#   能把任意一段文字转换成一个高维向量,神奇之处是让意思
#   相近的文字在高维空间里位置也靠得近 —— 于是"按意思找
#   文档"这个模糊需求变成一道精确数学题:在向量空间里找离
#   问题最近的那些点。★ 衡量两向量多近最常用余弦相似度
#   (衡量方向多一致,值域 -1 到 1 越接近 1 越相似),向量
#   库核心工作就是给定问题向量快速从几十万文档向量里算出
#   相似度最高的 top_k 个。★★ 选 embedding 模型三个必须
#   留意:①中文能力(很多模型英文语料为主、中文场景务必选
#   中文或多语言表现好的);②一致性铁律 —— 索引时和检索时
#   必须用同一个 embedding 模型,用 A 索引 B 检索两套向量
#   不在同一空间相似度毫无意义,是会让整个 RAG 彻底失灵的
#   隐蔽坑;③维度与成本 —— 维度越高通常越精准但开销越大。

# ★ embedding + 向量检索:把"按意思找"变成一道数学题
import numpy as np

# ★★ 一致性铁律:索引和检索,自始至终用【同一个】模型实例
#    —— 用 A 模型索引、B 模型检索,两套向量不在同一空间,
#    算出的相似度毫无意义,整个 RAG 会彻底失灵
EMBED_MODEL = load_embedding_model('bge-large-zh')   # ★ 选中文表现好的模型

def embed_one(text: str) -> np.ndarray:
    vec = EMBED_MODEL.encode(text)
    # ★ 归一化后,点积就等于余弦相似度,后续计算更省事
    return vec / np.linalg.norm(vec)

def embed_batch(texts: list[str]) -> np.ndarray:
    return np.array([embed_one(t) for t in texts])

# === ★ 余弦相似度:衡量两个向量"方向"有多一致 ===
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    # 两个向量都已归一化 -> 点积即余弦相似度,值域 -1~1
    return float(np.dot(a, b))

# === ★ 一个最小可用的向量库:存向量,按相似度检索 ===
class SimpleVectorStore:
    def __init__(self):
        self.texts: list[str] = []
        self.vectors: np.ndarray | None = None

    def add(self, texts: list[str], vectors: np.ndarray):
        self.texts.extend(texts)
        self.vectors = (vectors if self.vectors is None
                        else np.vstack([self.vectors, vectors]))

    def search(self, query_vec: np.ndarray, top_k: int = 5):
        # ★★ 核心:一次矩阵乘法,算出问题与所有文档的相似度
        scores = self.vectors @ query_vec
        # ★ 取分数最高的 top_k 个
        idx = np.argsort(scores)[::-1][:top_k]
        return [(self.texts[i], float(scores[i])) for i in idx]
# ★★ 生产环境会换成 FAISS / Milvus 这类专业向量库,但
#    "算相似度、取 top_k"这个内核,是完全一样的

检索召回:top-k、混合检索、重排,一个都不能少

# === ★ 检索这一环,是 RAG 答得准不准的"主战场" ===

# === ★ top_k:到底捞回几个片段 ===
# ★ ★ top_k 设太小(比如 1):万一最相关的那段没排在第一,
#   你就彻底错过了答案 —— 召回不足。
# ★ ★ top_k 设太大(比如 20):捞回一大堆,里面混进大量
#   "勉强沾边"的噪音,既挤占模型的上下文,又干扰它判断。
# ★ 实践:先取一个稍大的 top_k(如 20)做【粗筛】,再靠
#   后面的"重排"精挑细选 —— 这是下面要讲的关键套路。

# === ★★ 纯向量检索的盲区:它对"精确词"不敏感 ===
# ★ ★ 向量检索擅长"按意思找",但它有个明显盲区:对那些
#   【必须精确匹配】的词 —— 产品型号 X-2000、错误码
#   ERR_4011、人名、专有缩写 —— 它反而【不灵】。因为
#   "X-2000"和"X-3000"语义上很接近,向量分不清。
# ★ ★★ 解法:【混合检索(hybrid search)】。同时跑两路:
#   一路【向量检索】(管"意思相近"),一路传统的【关键词
#   检索】(如 BM25,管"精确命中那个词")。再把两路的
#   结果【融合】起来。意思和精确词,这下都不漏。

# === ★★ 最关键的一招:重排(Rerank) ===
# ★ ★ 向量检索的相似度,是"粗"的 —— 它快,但不够精准。
#   于是有了这个黄金套路:【粗筛 + 精排】两阶段。
#  - ★ 第一阶段(粗筛):用向量检索,快速捞回 top 20~50
#    个【候选】。图的是快、是召回率高(别漏)。
#  - ★ 第二阶段(精排):用一个专门的【重排模型
#    (reranker)】,把"问题"和这 20 个候选【逐一配对】,
#    给出一个【精准得多】的相关性打分,重新排序。
# ★ ★★ reranker 为什么更准?向量检索是把问题和文档【各自】
#   编码成向量再比对;而 reranker 是把【问题和文档拼在一起】
#   送进模型,让它"盯着这一对"做判断 —— 信息没有损失,
#   自然准得多。代价是慢,所以只用它给少量候选精排。
# ★ 取重排后的 top 3~5,才是真正喂给大模型的"参考资料"。

# === 小结 ===
# ★ 检索是 RAG 答得准不准的主战场。★ top_k 设太小(如 1)
#   万一最相关那段没排第一就彻底错过答案、召回不足;设太大
#   (如 20)捞回一堆混进大量勉强沾边的噪音,既挤占上下文
#   又干扰判断 —— 实践是先取稍大 top_k 做粗筛再靠重排精挑。
# ★★ 纯向量检索的盲区:它擅长按意思找但对必须精确匹配的词
#   (产品型号 X-2000、错误码 ERR_4011、人名、缩写)不灵,
#   因为 X-2000 和 X-3000 语义很接近向量分不清;解法是混合
#   检索 —— 同时跑向量检索(管意思相近)和传统关键词检索
#   (如 BM25 管精确命中)再融合两路结果。★★ 最关键一招是
#   重排:向量检索相似度是粗的、快但不够精准,黄金套路是
#   粗筛+精排两阶段 —— 粗筛用向量检索快速捞回 top 20~50 个
#   候选(图快、召回率高别漏),精排用专门的重排模型把问题
#   和每个候选逐一配对给出精准得多的打分重新排序;reranker
#   更准是因为向量检索把问题和文档各自编码再比对、而
#   reranker 把问题和文档拼在一起送进模型盯着这一对判断
#   信息没损失,代价是慢所以只给少量候选精排;取重排后
#   top 3~5 才是真正喂给大模型的参考资料。

# ★ 检索召回:混合检索 + 重排 —— "粗筛 + 精排"两阶段
from rank_bm25 import BM25Okapi

# === 第一阶段(粗筛):混合检索 —— 向量 + 关键词,两路都捞 ===
def hybrid_recall(question: str, top_k: int = 20) -> list[str]:
    # ★ 第一路:向量检索 —— 管"意思相近"
    q_vec = embed_one(question)
    vec_hits = [t for t, _ in vector_store.search(q_vec, top_k=top_k)]

    # ★ 第二路:BM25 关键词检索 —— 管"精确命中那个词"
    #   产品型号、错误码这类必须精确匹配的词,向量检索抓不住
    bm25_scores = bm25_index.get_scores(tokenize(question))
    bm25_idx = sorted(range(len(bm25_scores)),
                      key=lambda i: bm25_scores[i], reverse=True)[:top_k]
    kw_hits = [all_chunks[i] for i in bm25_idx]

    # ★★ 融合两路结果并去重 —— 意思和精确词,都不漏
    merged, seen = [], set()
    for text in vec_hits + kw_hits:
        if text not in seen:
            seen.add(text)
            merged.append(text)
    return merged

# === 第二阶段(精排):用 reranker 给候选精准打分、重排 ===
RERANKER = load_reranker('bge-reranker-large')

def rerank(question: str, candidates: list[str],
           top_n: int = 4) -> list[str]:
    # ★★ 关键:把"问题 + 每个候选"【成对】送进 reranker
    #    —— 不是各自编码再比对,信息无损,所以准得多
    pairs = [(question, c) for c in candidates]
    scores = RERANKER.predict(pairs)
    ranked = sorted(zip(candidates, scores),
                    key=lambda x: x[1], reverse=True)
    # ★ 只取精排后最靠前的 top_n,这才是喂给大模型的参考资料
    return [c for c, _ in ranked[:top_n]]

# === 串起来:粗筛 20 个 -> 精排留 4 个 ===
def retrieve_v2(question: str) -> list[str]:
    candidates = hybrid_recall(question, top_k=20)   # ★ 粗筛:多召回,别漏
    return rerank(question, candidates, top_n=4)      # ★ 精排:精挑,去噪

把检索结果喂给模型:prompt 不是"一股脑塞进去"

# === ★ 检索准了,最后一步:怎么把片段交给大模型 ===

# === ★★ 别把检索结果"一股脑"拼进 prompt ===
# ★ ★ 最潦草的写法:把检索回的 N 段文字,直接首尾相连成
#   一大坨,塞进 prompt。这有几个问题:模型分不清【哪段
#   是哪段】、分不清【哪段更重要】、也无法【溯源】。
# ★ ★ 正解:把每个片段,用清晰的分隔【结构化】地列出来,
#   并给每段标上【编号和来源】(来自哪个文档)。结构清楚,
#   模型才好"看着资料"逐条对照。

# === ★ 一条铁律:必须明确指示"不知道就说不知道" ===
# ★ ★ 这是治理"一本正经胡说八道"的【最关键一句话】。你
#   必须在 prompt 里【斩钉截铁地】写明:"只能依据下面提供
#   的资料回答;如果资料里没有相关信息,就直接回答'根据
#   现有资料无法回答',严禁自行编造。"
# ★ ★★ 我那次"编出一个数据库密码"的事故,根子就是缺了
#   这句话。大模型天性是"有问必答、不会就编",你不【显式
#   地、强硬地】给它一个"我不知道"的出口,它就一定会去编。
#   这个出口,必须由你在 prompt 里亲手给它。

# === ★ 让模型"标注引用来源" ===
# ★ ★ 在 prompt 里要求模型:回答时,注明它的每个结论是
#   依据【第几段资料】得出的。
# ★ ★ 好处是双份的:① 用户能【自行核实】,信任度大增;
#   ② 你排查问题时,一眼就能看出模型是"参考对了资料但
#   答错了",还是"压根就在瞎编"。

# === ★★ 当心:别让上下文"过载" ===
# ★ ★ 不是塞进 prompt 的资料越多越好。塞太多片段,会有
#   两个恶果:① 成本和延迟飙升;② "迷失在中间(lost in
#   the middle)"—— 大模型对一长串上下文里【中间部分】
#   的内容,注意力会明显下降,容易"看漏"。
# ★ ★ 所以前面"重排后只取 top 3~5"才如此重要 —— 给模型
#   的,要是【少而精】的几段,不是【多而杂】的一大堆。

# === 小结 ===
# ★ 检索准了最后一步是怎么把片段交给大模型。★★ 别把检索
#   结果一股脑拼进 prompt:最潦草的写法是把 N 段文字首尾
#   相连成一大坨塞进去,模型分不清哪段是哪段、哪段更重要、
#   也无法溯源;正解是把每个片段用清晰分隔结构化列出并标上
#   编号和来源。★ 一条铁律:必须明确指示"不知道就说不知道"
#   —— 这是治理"一本正经胡说八道"最关键的一句话,要在
#   prompt 里斩钉截铁写明"只能依据下面资料回答、资料里没有
#   就回答'根据现有资料无法回答'、严禁编造";"编出一个
#   数据库密码"的事故根子就是缺了这句话,大模型天性是有问
#   必答不会就编,你不显式强硬给它一个"我不知道"的出口它
#   就一定会编。★ 让模型标注引用来源:要求它注明每个结论
#   依据第几段资料,好处是用户能自行核实信任度大增、你排查
#   时一眼能看出是参考对了资料但答错还是压根瞎编。★★ 当心
#   别让上下文过载:不是塞的资料越多越好,塞太多成本延迟
#   飙升、还会"迷失在中间"(模型对长上下文中间部分注意力
#   明显下降容易看漏);所以"重排后只取 top 3~5"才如此
#   重要 —— 给模型的要少而精不是多而杂。

# ★ 拼 prompt:结构化列出片段 + 强制"不知道就说不知道"
PROMPT_TEMPLATE = '''你是一个严谨的知识库问答助手。请严格遵守以下规则:

1. 只能依据【参考资料】中的内容回答问题。
2. 如果参考资料里没有足够信息,必须直接回答"根据现有资料
   无法回答",严禁自行编造、严禁用你自己的知识补充。
3. 回答时,在每个结论后用 [资料N] 标注它依据的是第几段。

【参考资料】
{contexts}

【用户问题】
{question}

【你的回答】'''

def build_prompt(question: str, contexts: list[str]) -> str:
    # ★★ 别把片段首尾相连成一大坨 —— 要【结构化】编号 + 标来源
    blocks = []
    for i, ctx in enumerate(contexts, start=1):
        blocks.append(f'[资料{i}]\n{ctx}')
    contexts_text = '\n\n'.join(blocks)
    return PROMPT_TEMPLATE.format(contexts=contexts_text,
                                  question=question)

def answer_v2(question: str) -> str:
    # ★ 用前面"混合检索 + 重排"得到的少而精的片段
    contexts = retrieve_v2(question)

    # ★★ 兜底:检索结果为空,直接告知,别硬让模型"无中生有"
    if not contexts:
        return '根据现有资料无法回答这个问题。'

    prompt = build_prompt(question, contexts)
    # ★ 给模型的,是【少而精】的几段(top 3~5),不是一大堆
    #   —— 太多会让模型"迷失在中间",还徒增成本与延迟
    return call_llm(prompt)

工程坑:让 RAG 真正能上线的那些细节

# === ★ demo 跑通容易,真正稳定上线,还有几道坎 ===

# === ★★ 坑 1:RAG 没有评估,你就是在"盲调" ===
# ★ ★ 你改了切块策略、换了 embedding 模型 —— 到底变好了
#   还是变差了?如果只靠"随手问几个问题感觉一下",那你就是
#   在【盲调】:今天调好了 A 问题,可能悄悄弄坏了 B 问题。
# ★ ★★ 必须建一个【评估集】:几十到上百条"标准问题 + 期望
#   答案/期望命中的文档"。每次改动后,都用这个集合【自动
#   跑一遍】,量化地看两个指标 —— 检索的【命中率】(该召回
#   的片段有没有召回),和最终答案的【正确率】。没有评估集
#   的 RAG 优化,全是凭感觉。

# === ★ 坑 2:文档更新了,向量库不会自己知道 ===
# ★ ★ 索引是【离线】建好的。可你的文档会变 —— 改了、删了、
#   新增了。如果不管,向量库里就一直是【过时的旧内容】,
#   RAG 会信誓旦旦地告诉用户一个早已作废的答案。
# ★ ★ 必须设计【增量更新】机制:文档变动时,定位到对应的
#   chunk,把旧向量删掉、新向量加进去。别每次都全量重建 ——
#   那样又慢又贵。

# === ★★ 坑 3:检索质量,要【持续监控】 ===
# ★ ★ 上线不是终点。要把每次检索的【重排最高分】记录下来。
#   如果某个查询,重排后最高分都【很低】—— 这是一个强烈
#   信号:知识库里【可能压根没有】能回答这个问题的内容。
# ★ ★ 把这些"低分查询"收集起来,就是一份现成的【知识库
#   补全清单】:它精确地告诉你,用户在问什么、而你的文档
#   还缺什么。

# === ★ 坑 4:别什么问题都指望 RAG ===
# ★ ★ RAG 的本领,是回答"答案明确写在某份文档里"的事实型
#   问题。它【不擅长】的是:需要跨大量文档做【统计、汇总、
#   推理】的问题(如"所有项目里延期的占比是多少")—— 这种
#   答案不在任何单一片段里,靠检索几段根本拼不出来。
# ★ 想清楚边界:这类问题,该走数据库查询、该走专门的数据
#   分析,而不是硬塞给 RAG。

# === 认知 ===
# ★ demo 跑通容易、真正稳定上线还有几道坎。★★ 坑 1 RAG
#   没有评估你就是在盲调:改了切块策略换了 embedding 模型
#   到底变好还是变差,只靠"随手问几个问题感觉一下"就是盲调
#   —— 今天调好 A 可能悄悄弄坏 B;必须建评估集(几十到上百
#   条标准问题+期望答案/期望命中文档),每次改动后自动跑一
#   遍量化看检索命中率和答案正确率。★ 坑 2 文档更新了向量
#   库不会自己知道:索引是离线建好的但文档会改会删会新增,
#   不管向量库就一直是过时旧内容、RAG 会信誓旦旦告诉用户
#   早已作废的答案;必须设计增量更新机制,文档变动时定位
#   对应 chunk 删旧向量加新向量,别每次全量重建。★★ 坑 3
#   检索质量要持续监控:把每次检索的重排最高分记录下来,
#   某查询重排后最高分都很低是强烈信号 —— 知识库可能压根
#   没有能回答它的内容;把这些低分查询收集起来就是现成的
#   知识库补全清单。★ 坑 4 别什么问题都指望 RAG:它的本领
#   是回答"答案明确写在某份文档里"的事实型问题,不擅长需要
#   跨大量文档做统计汇总推理的问题(如"所有项目延期占比"),
#   这种答案不在任何单一片段里检索几段拼不出来,该走数据库
#   查询、走专门数据分析,别硬塞给 RAG。

# ★ 坑1 解法:建评估集,每次改动后自动量化跑一遍
# 评估集:一批"标准问题 + 期望命中的文档 id + 期望答案要点"
EVAL_SET = [
    {'question': '报销流程一共几步?',
     'expect_doc': 'finance_reimburse.md',
     'expect_points': ['三步', '提交', '审批']},
    {'question': '订单接口怎么分页?',
     'expect_doc': 'api_order.md',
     'expect_points': ['page', 'size']},
    # ... 几十到上百条
]

def evaluate() -> dict:
    hit, correct = 0, 0
    for case in EVAL_SET:
        contexts = retrieve_v2(case['question'])
        # ★ 指标一:检索命中率 —— 该召回的文档,有没有被召回
        if any(case['expect_doc'] in c for c in contexts):
            hit += 1
        # ★ 指标二:答案正确率 —— 期望要点是否都出现在回答里
        reply = answer_v2(case['question'])
        if all(p in reply for p in case['expect_points']):
            correct += 1
    total = len(EVAL_SET)
    # ★★ 每次改切块/换模型/调 top_k 后都跑一遍,用数字说话
    #    —— 而不是"随手问两句感觉一下",那是盲调
    return {'检索命中率': hit / total, '答案正确率': correct / total}

# ★ 坑3 解法:监控检索质量,低分查询自动进"知识库补全清单"
def retrieve_with_monitor(question: str) -> list[str]:
    candidates = hybrid_recall(question, top_k=20)
    pairs = [(question, c) for c in candidates]
    scores = sorted(RERANKER.predict(pairs), reverse=True)

    top_score = scores[0] if scores else 0.0
    # ★★ 重排最高分都很低 -> 知识库可能压根没有相关内容
    if top_score < 0.3:
        log_low_score_query(question, top_score)   # 进补全清单
    return rerank(question, candidates, top_n=4)

命令速查

RAG 实战:三段流水线 + 一套提质量手段
=============================================================
三段式流水线(严格串行,木桶效应)
-------------------------------------------------------------
索引 Indexing   离线:文档切块 -> 算 embedding -> 存向量库
检索 Retrieval  在线:问题转向量 -> 向量库捞 top_k 片段
生成 Generation 在线:片段 + 问题拼 prompt -> 大模型回答
排查第一步      把检索回的片段【原样打印】,人眼读一遍

文档切块 Chunking
-------------------------------------------------------------
反例            按固定字数硬切 -> 句子/表格被劈开,语义残缺
正解            按标题层级 / 段落切,FAQ 的问答绑死在一起
技巧            相邻 chunk 留 overlap,关键句不被切割线劈开

Embedding 与检索
-------------------------------------------------------------
embedding       把文字变高维向量,意思近的向量也近
相似度          余弦相似度,值域 -1~1,越近 1 越相似
铁律            索引和检索必须用【同一个】embedding 模型

检索召回(主战场)
-------------------------------------------------------------
混合检索        向量检索(意思) + BM25(精确词)两路融合
重排 rerank     粗筛 top20~50 -> reranker 精排 -> 留 top3~5
prompt          结构化标编号来源 + 强制"不知道就说不知道"

四个工程坑
-------------------------------------------------------------
坑1 没评估      建评估集,量化命中率/正确率,别凭感觉盲调
坑2 不更新      文档变了要增量更新向量库,否则答过时内容
坑3 不监控      记录重排最高分,低分查询 -> 知识库补全清单
坑4 滥用        统计/汇总型问题别用 RAG,走数据库查询

口诀:RAG 答非所问,问题几乎不在模型,在检索和切块
      给模型少而精的几段,并留一个"我不知道"的出口
      没有评估集的 RAG 优化,全是盲调

避坑清单

1. RAG 不是"检索一下再回答"一个动作,而是索引、检索、生成三段严格串行的流水线,任何一段拉胯整个就废——它是一只木桶,最短的板决定成败
2. RAG 答非所问,问题几乎从不在最后那个大模型;别急着换更贵的模型,第一步永远是把检索实际捞回的片段原样打印出来用人眼读一遍
3. 按固定字数硬切文档是一场灾难,它会把句子、表格、"问题+答案"从中间劈开;要按标题层级、段落等语义结构切,每个 chunk 都该能独立读懂
4. 相邻 chunk 之间要留一点重叠(overlap),即使关键句不幸落在切割线上,也能完整存在于某一个 chunk 里,不被劈成两边都残缺
5. 索引时和检索时必须用同一个 embedding 模型,用 A 模型索引、B 模型检索,两套向量根本不在同一空间,算出的相似度毫无意义
6. 纯向量检索对必须精确匹配的词(产品型号、错误码、人名)不敏感,要用混合检索——向量检索管"意思"、BM25 关键词检索管"精确命中"
7. 检索要用"粗筛 + 精排"两阶段:向量检索快速召回 top20~50 个候选,再用 reranker 重排模型精排留 top3~5,reranker 准是因为它把问题和文档成对送入
8. prompt 里必须斩钉截铁写明"资料里没有就回答无法回答、严禁编造"——大模型天性是不会就编,你不给它一个"我不知道"的出口它就一定会编
9. 给模型的参考资料要少而精(top3~5),塞太多会让模型"迷失在中间"、注意力下降看漏内容,还徒增成本和延迟
10. 没有评估集的 RAG 优化全是盲调;另外文档更新了要增量更新向量库,统计汇总型问题别硬塞给 RAG——它只擅长答案明确写在某份文档里的事实型问题

总结

这一趟把 RAG 彻底理清的过程,纠正了我一个特别根本、也特别有迷惑性的误解——我一直把"模型"当成了整个系统里【唯一聪明、也唯一需要操心】的那个部件。在我最初的脑子里,RAG 的逻辑简单得像一句话:文档丢进去,问题发进去,模型很聪明,它自己会从文档里找到答案。所以当系统胡说八道时,我下意识的反应,是去怀疑那个唯一"聪明"的部件——是不是模型不够强?换个更贵的?可当我真正把检索回来的片段一段段打印出来、用人眼读过之后,我才被狠狠点醒:模型一点没错,它甚至可以说是"尽职尽责"——我递给它五段参考资料,其中四段是垃圾,它就忠实地基于这四段垃圾,推理出了一个逻辑自洽、措辞专业的错误答案。它不是"不聪明",恰恰相反,它聪明得能把错料加工成一篇像样的文章。问题从来不在它,在我递给它的那个"信息袋子"。想通这件事,我对 RAG 的整个理解被重构了:RAG 这套东西,真正的工程量、真正的难点,百分之九十都【不在模型那一端】,而在模型之前的那条又长又琐碎的流水线上——文档怎么切才不破坏语义、embedding 模型选得对不对、检索怎么混合才不漏掉精确的词、捞回来的一堆候选怎么重排才能精准、最后那几段又该怎么结构化地、带着"不许编造"的硬约束递给模型。这每一个环节,都是默默无闻的、不性感的、纯粹的工程活,但 RAG 的成败,恰恰就钉死在这些环节上。模型,只是这条流水线最末端的一个消费者——你喂它什么品质的原料,它就产出什么品质的成品,这是一条冷酷而公平的规则。我也因此真正读懂了那句"垃圾进、垃圾出"的老话,在大模型时代为什么不仅没过时,反而变得更危险了:过去,垃圾输入往往得到一个一眼可辨的垃圾输出;而今天,大模型有本事把垃圾输入,加工成一个【看起来一点都不像垃圾】的、自信满满的输出——它让"错误"穿上了"专业"的外衣,反而更难被发现、更容易被采信。这件事给我的最终启发,早已超出了 RAG 本身:任何一个由"人/工程"和"模型"协作的 AI 系统,我都会先冷静地认清一个分工——模型负责"基于给定信息做推理与表达",而"给定的信息本身,品质如何、完整与否、有没有混入噪音",这个责任,百分之百在我、在我那条流水线身上,推卸不掉、也指望不上模型替我兜底。与其反复追问"模型够不够聪明",不如老老实实地回头检查:我喂给它的东西,到底干不干净。一个 AI 系统真正的水位,从来不是由它那个最聪明的部件决定的,而是由它最不起眼的那条数据流水线,决定的。

—— 别看了 · 2026