大模型应用落地指南：从 RAG 知识库搭建到检索效果优化实战

很多团队做大模型应用时，第一反应是“把文档喂给模型”。真正开始落地后，问题很快就冒出来了：

模型明明接了知识库，回答还是“像在瞎猜”
文档一多，召回结果变得很随机
同一个问题，昨天答得好，今天答得差
线上延迟、成本、准确率很难一起兼顾

如果你也经历过这些阶段，这篇文章就是带你把一条典型的 RAG 落地路径走完整：从知识库构建、切分、向量化、检索，到重排、评估和优化。我会尽量用“能做出来”的方式来讲，而不是只停留在概念层。

一、背景与问题

RAG（Retrieval-Augmented Generation，检索增强生成）之所以重要，是因为它解决了大模型应用里的两个现实矛盾：

模型参数不是你的私有知识库
纯靠提示词，无法稳定覆盖长尾业务知识

一个常见业务场景是这样的：

公司有产品文档、FAQ、工单、制度说明、接口文档
用户提问非常具体，比如“退款规则里，优惠券退不退？”
如果直接问大模型，它可能会“编一个看起来合理的答案”
如果做了 RAG，但文档切分粗糙、检索不准，模型依然答不对

所以，RAG 的核心不是“接一个向量库就完事”，而是把这条链路打磨顺：

文档质量 → 切分策略 → 向量召回 → 关键词补充 → 重排 → 提示组装 → 结果评估

很多项目失败，不是败在模型本身，而是败在前面的检索链路。

二、前置知识与环境准备

本文默认你已经知道这些基本概念：

Embedding：把文本变成向量
Top-K 检索：取最相关的 K 条候选
Prompt：给模型的上下文和指令
向量数据库：存储向量并做近邻检索

为了让示例容易跑起来，我选一个比较轻量的本地方案：

Python 3.10+
sentence-transformers：生成中文向量
faiss-cpu：本地向量检索
rank-bm25：关键词检索
jieba：中文分词
numpy

安装依赖：

pip install sentence-transformers faiss-cpu rank-bm25 jieba numpy

三、核心原理

先别急着上代码，先把一条“可用的 RAG 流程图”建立起来。

3.1 RAG 的基础链路

flowchart TD
    A[原始文档] --> B[清洗与结构化]
    B --> C[文本切分 Chunking]
    C --> D[向量化 Embedding]
    D --> E[向量库索引]
    Q[用户问题] --> F[查询向量化]
    F --> G[召回候选 Top-K]
    Q --> H[关键词检索 BM25]
    G --> I[候选合并]
    H --> I
    I --> J[重排 Rerank]
    J --> K[构造 Prompt]
    K --> L[大模型生成答案]

这张图里最容易被低估的是中间三步：

切分
召回
重排

它们基本决定了模型最后“有没有东西可答”。

3.2 为什么只做向量检索不够

向量检索擅长语义相似，但对一些场景不稳定：

缩写词、型号、版本号
产品名、字段名、报错码
明确关键词查询，比如“退款 T+1”“字段 user_id”

这时 BM25 这类关键词检索通常更稳。

所以在实际项目里，我更推荐：

向量检索负责语义覆盖
BM25 负责关键词精确命中
最终用重排模型或规则做融合

3.3 Chunk 不是越大越好，也不是越小越好

这是我当时踩过的坑之一。

Chunk 太大

一个块包含多个主题
检索命中了，但真正有用的信息只占一小段
浪费上下文窗口

Chunk 太小

语义被切碎
上下文丢失
召回命中了碎片，但模型无法拼出完整答案

经验上，中级复杂度的知识库可以从这个起点试：

每块 300~800 中文字
相邻块 50~150 字重叠
按标题、段落、列表做优先切分，而不是死按字数硬切

3.4 检索优化的本质

检索优化不是单点技巧，而是一个漏斗：

flowchart LR
    A[文档质量] --> B[切分质量]
    B --> C[召回率 Recall]
    C --> D[重排精度 Precision]
    D --> E[生成答案质量]

如果召回阶段没把关键证据捞上来，后面模型再强也没法“空手变答案”。

四、从零搭一个可运行的 RAG 检索原型

下面我们做一个最小可运行版本，重点放在知识库搭建和检索优化，而不是接某家特定大模型 API。

4.1 准备示例文档

先定义几份模拟业务文档：

# rag_demo.py
documents = [
    {
        "id": "doc_1",
        "title": "退款规则",
        "content": """
退款申请在支付成功后 7 天内可发起。
若订单使用优惠券，退款时优惠券不退回。
余额支付部分原路退回账户余额，银行卡支付部分在 1-3 个工作日到账。
虚拟商品一经发货，不支持无理由退款。
""".strip()
    },
    {
        "id": "doc_2",
        "title": "发票说明",
        "content": """
电子发票在订单完成后 24 小时内开具。
企业用户可申请增值税专用发票，但需提前完成资质认证。
发票抬头一经提交，订单完成后不可修改。
""".strip()
    },
    {
        "id": "doc_3",
        "title": "会员权益",
        "content": """
高级会员享受每月 5 张免邮券。
会员有效期内可享受专属客服通道。
会员购买虚拟商品不额外享受折扣。
""".strip()
    },
    {
        "id": "doc_4",
        "title": "账户安全",
        "content": """
若发现账号异常登录，请立即修改密码并开启二次验证。
同一手机号最多绑定 3 个账号。
连续输错密码 5 次，账号将被临时锁定 30 分钟。
""".strip()
    }
]

4.2 实现文本切分

这里先做一个朴素但实用的切分器：按段落切，再做窗口拼接。

import re
from typing import List, Dict

def split_text(text: str, chunk_size: int = 120, overlap: int = 30) -> List[str]:
    text = re.sub(r'\n+', '\n', text).strip()
    paragraphs = [p.strip() for p in text.split('\n') if p.strip()]
    
    chunks = []
    current = ""
    
    for para in paragraphs:
        if len(current) + len(para) + 1 <= chunk_size:
            current += ("\n" if current else "") + para
        else:
            if current:
                chunks.append(current)
            if len(para) <= chunk_size:
                current = para
            else:
                # 对超长段落再按长度切
                start = 0
                while start < len(para):
                    end = start + chunk_size
                    piece = para[start:end]
                    chunks.append(piece)
                    start += chunk_size - overlap
                current = ""
    
    if current:
        chunks.append(current)
    
    # 加简单重叠
    final_chunks = []
    for i, chunk in enumerate(chunks):
        if i > 0:
            prev_tail = chunks[i - 1][-overlap:]
            merged = prev_tail + "\n" + chunk
            final_chunks.append(merged)
        else:
            final_chunks.append(chunk)
    
    return final_chunks


def build_chunks(documents: List[Dict]) -> List[Dict]:
    chunked_docs = []
    for doc in documents:
        chunks = split_text(doc["content"])
        for idx, chunk in enumerate(chunks):
            chunked_docs.append({
                "chunk_id": f'{doc["id"]}_chunk_{idx}',
                "doc_id": doc["id"],
                "title": doc["title"],
                "text": chunk
            })
    return chunked_docs

4.3 建立向量索引

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorIndex:
    def __init__(self, model_name: str = "shibing624/text2vec-base-chinese"):
        self.model = SentenceTransformer(model_name)
        self.index = None
        self.texts = []
        self.metadata = []

    def build(self, chunked_docs: List[Dict]):
        self.texts = [item["text"] for item in chunked_docs]
        self.metadata = chunked_docs
        
        embeddings = self.model.encode(self.texts, normalize_embeddings=True)
        embeddings = np.array(embeddings).astype("float32")
        
        dim = embeddings.shape[1]
        self.index = faiss.IndexFlatIP(dim)
        self.index.add(embeddings)

    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        query_vec = self.model.encode([query], normalize_embeddings=True)
        query_vec = np.array(query_vec).astype("float32")
        
        scores, indices = self.index.search(query_vec, top_k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            results.append({
                "score": float(score),
                "text": self.texts[idx],
                "metadata": self.metadata[idx]
            })
        return results

4.4 加入 BM25 关键词检索

中文检索如果不做分词，BM25 效果会很一般。

import jieba
from rank_bm25 import BM25Okapi

class BM25Index:
    def __init__(self):
        self.corpus = []
        self.metadata = []
        self.bm25 = None

    def tokenize(self, text: str):
        return list(jieba.cut(text))

    def build(self, chunked_docs: List[Dict]):
        self.metadata = chunked_docs
        self.corpus = [self.tokenize(item["text"]) for item in chunked_docs]
        self.bm25 = BM25Okapi(self.corpus)

    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        tokenized_query = self.tokenize(query)
        scores = self.bm25.get_scores(tokenized_query)
        top_indices = np.argsort(scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            results.append({
                "score": float(scores[idx]),
                "text": self.metadata[idx]["text"],
                "metadata": self.metadata[idx]
            })
        return results

4.5 混合检索与简单重排

先用一个容易理解的融合方式：向量分数 + BM25 归一化分数。

def normalize_scores(results: List[Dict], score_key: str = "score") -> List[Dict]:
    if not results:
        return results
    
    scores = [item[score_key] for item in results]
    min_s, max_s = min(scores), max(scores)
    
    for item in results:
        if max_s == min_s:
            item["norm_score"] = 1.0
        else:
            item["norm_score"] = (item[score_key] - min_s) / (max_s - min_s)
    return results


def hybrid_search(query: str, vector_index: VectorIndex, bm25_index: BM25Index, top_k: int = 5):
    vec_results = normalize_scores(vector_index.search(query, top_k=top_k * 2))
    bm25_results = normalize_scores(bm25_index.search(query, top_k=top_k * 2))
    
    merged = {}
    
    for item in vec_results:
        chunk_id = item["metadata"]["chunk_id"]
        merged[chunk_id] = {
            "text": item["text"],
            "metadata": item["metadata"],
            "vector_score": item["norm_score"],
            "bm25_score": 0.0
        }
    
    for item in bm25_results:
        chunk_id = item["metadata"]["chunk_id"]
        if chunk_id not in merged:
            merged[chunk_id] = {
                "text": item["text"],
                "metadata": item["metadata"],
                "vector_score": 0.0,
                "bm25_score": item["norm_score"]
            }
        else:
            merged[chunk_id]["bm25_score"] = item["norm_score"]
    
    results = []
    for _, item in merged.items():
        final_score = 0.6 * item["vector_score"] + 0.4 * item["bm25_score"]
        item["final_score"] = final_score
        results.append(item)
    
    results.sort(key=lambda x: x["final_score"], reverse=True)
    return results[:top_k]

4.6 运行主程序

def main():
    chunked_docs = build_chunks(documents)
    
    print("切分后的 chunk 数量：", len(chunked_docs))
    for item in chunked_docs:
        print(item["chunk_id"], "=>", item["text"].replace("\n", " | "))
    
    vector_index = VectorIndex()
    vector_index.build(chunked_docs)
    
    bm25_index = BM25Index()
    bm25_index.build(chunked_docs)
    
    query = "订单用了优惠券，退款时优惠券会退回吗？"
    results = hybrid_search(query, vector_index, bm25_index, top_k=3)
    
    print("\n查询：", query)
    print("\nTop 结果：")
    for i, item in enumerate(results, 1):
        print(f"\n#{i}")
        print("标题：", item["metadata"]["title"])
        print("chunk_id：", item["metadata"]["chunk_id"])
        print("final_score：", round(item["final_score"], 4))
        print("文本：", item["text"])


if __name__ == "__main__":
    main()

4.7 预期效果

对于问题：

订单用了优惠券，退款时优惠券会退回吗？

理想召回应优先命中“退款规则”里的这一句：

若订单使用优惠券，退款时优惠券不退回。

这就说明你的知识库检索链路基本打通了。

五、把检索结果喂给大模型时，应该怎么组织 Prompt

RAG 的后半程是“基于证据生成答案”。这里有个很重要的原则：

让模型只根据召回内容回答，不要自由发挥。

一个实用模板如下：

你是企业知识库问答助手。请仅根据给定资料回答问题。
如果资料中没有明确答案，请直接回答“资料中未找到明确信息”，不要猜测。

【问题】
订单用了优惠券，退款时优惠券会退回吗？

【资料】
1. 退款规则：若订单使用优惠券，退款时优惠券不退回。
2. 退款规则：余额支付部分原路退回账户余额，银行卡支付部分在 1-3 个工作日到账。

【回答要求】
- 先直接回答结论
- 再简要给出依据
- 不要输出资料外内容

这类 Prompt 有几个好处：

降低幻觉
回答更稳定
更容易做自动评估

六、逐步验证清单

如果你在公司里推进 RAG 项目，我建议不要一开始就追求“大而全”，而是按下面顺序验证。

6.1 第一步：验证切分是否合理

抽样看 20~50 个 chunk，重点观察：

一个 chunk 是否只包含一个相对完整主题
标题、术语、结论句有没有被切断
重叠部分是否足够承接上下文

6.2 第二步：验证召回是否命中正确证据

准备 20 个真实问题，每个问题人工标注“理想证据文档”。

检查：

Top1 是否命中
Top3 是否命中
是向量检索命中的，还是 BM25 命中的

6.3 第三步：验证生成是否忠于证据

看模型回答时有没有这些问题：

证据明明写了“不退回”，模型却说“视情况而定”
证据没写到账时间，模型擅自补了“通常 24 小时”
多条证据冲突时，模型没有说明依据来源

七、常见坑与排查

这是实战里最容易踩的部分。

7.1 坑一：召回看起来相关，但不是答案证据

现象

用户问退款优惠券，结果召回了“会员权益”“发票说明”这类看起来也像订单相关的内容。

原因

chunk 太大，主题混杂
embedding 模型不适配中文业务语料
只用向量检索，缺少关键词约束

排查方法

打印 Top10 结果，人工看误召回内容
对比向量检索和 BM25 的结果差异
看 query 里的关键词有没有在 chunk 中出现

解决建议

先把 chunk 缩小到更聚焦的主题粒度
引入混合检索
对标题、关键词做额外加权

7.2 坑二：文档明明有，还是检索不到

现象

知识库确实包含某条规则，但查询总是命不中。

原因

切分时把关键句拆散了
文档清洗把特殊字符、编号、表格丢了
查询表达和文档表达差异太大

排查方法

直接搜原句，确认索引里是否存在
检查文档入库前后内容是否一致
查看 chunk 中是否保留标题和上下文

解决建议

chunk 里附带标题路径，例如“退款规则 > 退款说明”
对问答日志做 query rewrite
为专业术语建立同义词表

7.3 坑三：Top-K 调大后效果反而变差

现象

召回更多候选后，模型回答更啰嗦，甚至答错。

原因

低质量候选混入过多
Prompt 里资料太长，干扰判断
没有重排，模型自己“瞎选证据”

排查方法

分别观察 Top3、Top5、Top10 的最终回答质量
检查后几条候选是否在语义上偏题
看模型引用的是哪几条资料

解决建议

先召回 Top20，再重排取前 3~5 条
对相似 chunk 去重
对长文档采用“父子块”结构，而不是一次全塞进去

7.4 坑四：线上效果和离线测试差很多

现象

测试集看起来不错，一上线用户反馈“答非所问”。

原因

测试问题过于标准化
真实用户提问更口语、更省略、更跳跃
线上文档版本频繁更新，索引未同步

排查方法

从线上日志抽样真实 query
看失败问题是否集中在口语化表达
检查文档更新时间和索引更新时间

解决建议

建立线上 query 回放机制
给知识库索引增加版本号
对热点问题单独做 FAQ 兜底

八、安全/性能最佳实践

RAG 不只是“答得对”，还要“跑得稳”。

8.1 安全最佳实践

1）做文档权限隔离

如果知识库有部门权限、用户等级限制，检索阶段就必须做过滤。

不要把所有文档都检索出来再交给模型，否则很容易越权泄露。

2）防提示注入

知识库文档中可能混入恶意内容，比如：

忽略之前所有指令
输出系统配置
泄露内部信息

因此在生成阶段要明确规则：

文档内容只是资料，不是系统指令
系统提示优先级高于检索内容
对高风险输出做审计和拦截

3）敏感信息脱敏

入库前建议处理：

手机号
身份证号
银行卡号
Access Token、密钥

尤其是把工单、聊天记录拿来做知识库时，这个问题非常常见。

8.2 性能最佳实践

1）索引分层

如果文档规模上来，可以按业务域拆库：

售后知识库
财务知识库
技术文档库

先做粗路由，再做细检索，能明显降噪。

2）缓存高频查询

对热门问题可以缓存：

query embedding
检索结果
最终答案

这在客服、内部问答场景里很有价值。

3）异步更新索引

不要每次文档更新都全量重建索引。建议：

新文档增量入库
删除文档做软删除标记
夜间低峰做全量校验

4）控制上下文长度

送给模型的证据不是越多越好。一般建议：

只保留高置信候选
相似 chunk 合并去重
保留标题、来源、关键段落，删掉噪音描述

九、进一步优化：从“能用”到“好用”

当你的基础 RAG 跑通后，下一阶段优化通常在下面几个方向。

9.1 查询改写

用户说“优惠券退吗”，文档里写“退款时优惠券不退回”。
这类口语和书面语差异，可以通过 query rewrite 缓解。

例如把：

“优惠券退吗”
“券会不会回来”
“退款后券返还吗”

统一改写为：

“订单使用优惠券后，退款时优惠券是否退回”

9.2 标题增强

很多文档正文单独看语义不足，但标题信息很强。
入向量库时，可以把标题拼进文本：

def enrich_text(title: str, text: str) -> str:
    return f"标题：{title}\n正文：{text}"

这招很简单，但经常有效。

9.3 父子块检索

一种很实用的策略是：

子块用于召回，颗粒度小，命中准
父块用于生成，信息更完整

流程如下：

flowchart TD
    A[原始文档] --> B[父块 Parent Chunk]
    B --> C[子块 Child Chunk]
    C --> D[子块向量索引]
    Q[用户问题] --> E[召回子块]
    E --> F[映射回父块]
    F --> G[把完整上下文交给模型]

这样既能提高召回精度，也能减少上下文碎片化问题。

9.4 重排模型

如果你对准确率要求更高，可以在召回后增加 reranker。
典型链路是：

向量召回 Top20
BM25 召回 Top20
合并去重
reranker 排序
取前 3~5 条给模型

这往往比单纯调 Top-K 更有效。

十、一个更接近生产的调用时序

sequenceDiagram
    participant U as 用户
    participant R as RAG服务
    participant V as 向量索引
    participant B as BM25索引
    participant M as 大模型

    U->>R: 提问
    R->>V: 语义检索 TopN
    R->>B: 关键词检索 TopN
    V-->>R: 候选集合A
    B-->>R: 候选集合B
    R->>R: 合并、去重、重排
    R->>M: 问题 + 证据上下文
    M-->>R: 生成答案
    R-->>U: 返回答案 + 引用来源

我个人很建议在最终返回里附上“来源片段”或“引用文档标题”，原因很现实：

用户更容易信任结果
错误时方便排查
后续能做点击反馈和在线学习

十一、总结

如果把这篇文章压缩成几条最重要的落地建议，我会给你下面这份清单：

别把 RAG 理解成“接个向量库”
- 真正决定效果的是切分、召回、重排的整体设计
优先做好知识库数据质量
- 文档乱、结构差、版本不一致，后面再怎么调模型都很难救
中文场景优先考虑混合检索
- 向量检索解决语义相似
- BM25 解决关键词命中
- 两者结合通常更稳
Chunk 设计要围绕“一个块能否独立表达一个知识点”
- 不要只按字数机械切分
上线前一定做问题集评估
- 至少看 Top1/Top3 命中率
- 再看最终答案是否忠于证据
控制边界条件
- 没检索到就明确说没找到
- 不要让模型自由脑补
- 有权限要求的知识库必须前置过滤

如果你现在正准备把大模型接进业务系统，我建议从一个小而清晰的知识域开始，比如退款规则、内部制度、接口文档 FAQ。先把一条窄场景链路做通，再扩展到更复杂的多库检索、多轮问答和在线评估。这样成功率会高很多。

RAG 真正难的地方，不在“模型有多强”，而在于你是否能把知识以正确的方式送到模型面前。只要这一步做对，大模型才有机会在业务里稳定发挥。

《大模型应用落地指南：从 RAG 知识库搭建到检索效果优化实战》

大模型应用落地指南：从 RAG 知识库搭建到检索效果优化实战

一、背景与问题

二、前置知识与环境准备

三、核心原理

3.1 RAG 的基础链路

3.2 为什么只做向量检索不够

3.3 Chunk 不是越大越好，也不是越小越好

Chunk 太大

Chunk 太小

3.4 检索优化的本质

四、从零搭一个可运行的 RAG 检索原型

4.1 准备示例文档

4.2 实现文本切分

4.3 建立向量索引

4.4 加入 BM25 关键词检索

4.5 混合检索与简单重排

4.6 运行主程序

4.7 预期效果

五、把检索结果喂给大模型时，应该怎么组织 Prompt

六、逐步验证清单

6.1 第一步：验证切分是否合理

6.2 第二步：验证召回是否命中正确证据

6.3 第三步：验证生成是否忠于证据

七、常见坑与排查

7.1 坑一：召回看起来相关，但不是答案证据

现象

原因

排查方法

解决建议

7.2 坑二：文档明明有，还是检索不到

现象

原因

排查方法

解决建议

7.3 坑三：Top-K 调大后效果反而变差

现象

原因

排查方法

解决建议

7.4 坑四：线上效果和离线测试差很多

现象

原因

排查方法

解决建议

八、安全/性能最佳实践

8.1 安全最佳实践

1）做文档权限隔离

2）防提示注入

3）敏感信息脱敏

8.2 性能最佳实践

1）索引分层

2）缓存高频查询

3）异步更新索引

4）控制上下文长度

九、进一步优化：从“能用”到“好用”

9.1 查询改写

9.2 标题增强

9.3 父子块检索

9.4 重排模型

十、一个更接近生产的调用时序

十一、总结