Copyright © 春秋博客

鄂ICP备19007640号-4
菜单
博客首页 / 人工智能
RAG检索增强生成搭建实践：从0到1全记录
发布日期: 2026/08/22 阅读总量: 0

一、被ChatGPT坑回工位的那个周五

2024年3月，我们给客户演示一个内部知识库问答系统，用的GPT-4。客户问「我们2024年的报销标准是多少」，GPT-4给出了一版2022年的数字，当场翻车。它没联网，训练数据里没有我们2024年1月更新的制度文件——这是大模型的硬伤。

我用脚趾头想都知道该上RAG了，但网上一搜全是概念科普，没有一篇能直接撸代码的。这篇东西是我花4周踩出来的，全文涉及的代码片段可以直接粘到你项目里跑。

二、三套方案：手搓、微调、RAG

搭建成本、维护成本、更新时效性、私有数据支持，四个维度对着比。

方案搭建成本维护成本更新时效私有数据结论
直接LLM低（API调用）零训练截止日不支持不可用
LoRA微调高（需要GPU）高（数据变了要重训）按周/月支持数据变化慢才值得
RAG中（向量库+嵌入）低（改文档即可）实时支持知识库场景最优

我们的场景是制度文档问答，文档每周都在更新，RAG是唯一合理的选择。

三、整体架构

我使用的具体版本：Python 3.11.7、LangChain 0.1.0、Qdrant 1.9.0、text-embedding-3-small（OpenAI）。

核心链路：文档加载 → 文本切块 → 向量化 → 存向量库（Qdrant）→ 用户查询 → 向量检索 + 关键词检索（BM25）→ 融合排序（RRF）→ 拼接上下文 → 生成（GPT-4-turbo）。

四、完整代码实现

4.1 环境准备

# 创建虚拟环境（Python 3.11.7）
python3.11 -m venv rag_env
source rag_env/bin/activate

# 安装依赖
pip install langchain==0.1.0 \
    langchain-openai==0.0.5 \
    qdrant-client==1.9.0 \
    fastapi==0.109.2 \
    uvicorn==0.27.0 \
    rank-bm25==0.2.2 \
    pypdf==4.0.1 \
    python-multipart==0.0.6

# 启动向量数据库（Docker方式）
docker run -d -p 6333:6333 -p 6334:6334 \
    --name qdrant \
    -v $(pwd)/qdrant_storage:/qdrant/storage \
    qdrant/qdrant:v1.9.0

4.2 初始化配置

# config.py
import os
from dotenv import load_dotenv
load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_API_BASE = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4-turbo-2024-04-09"
QDRANT_HOST = "localhost"
QDRANT_PORT = 6333
COLLECTION_NAME = "company_kb"

# 切块参数
CHUNK_SIZE = 512
CHUNK_OVERLAP = 64

# 检索参数
TOP_K_VECTOR = 5
TOP_K_BM25 = 5
RRF_K = 60

4.3 文档加载与切块

切块是RAG效果好坏的关键。我用的LangChain的RecursiveCharacterTextSplitter，它按分隔符优先级递归切分，比固定长度切块更合理。

# chunker.py
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 分隔符优先级从高到低
separators = ["\n\n", "\n", "。", "！", "？", "；", " ", ""]

def load_and_chunk(file_path: str):
    if file_path.endswith(".pdf"):
        loader = PyPDFLoader(file_path)
    else:  # .txt 或 .md
        loader = TextLoader(file_path, encoding="utf-8")
    
    docs = loader.load()
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=separators,
        length_function=len,
    )
    
    chunks = splitter.split_documents(docs)
    return chunks

4.4 向量化并入库

# vector_store.py
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from openai import OpenAI

client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_API_BASE)
qdrant = QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT)

def get_embedding(text: str, model: str = EMBEDDING_MODEL) -> list[float]:
    resp = client.embeddings.create(input=text, model=model)
    return resp.data[0].embedding

def create_collection():
    # text-embedding-3-small 输出 1536 维
    qdrant.recreate_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    )

def index_documents(chunks):
    create_collection()
    points = []
    for idx, chunk in enumerate(chunks):
        vec = get_embedding(chunk.page_content)
        points.append(PointStruct(
            id=idx,
            vector=vec,
            payload={
                "text": chunk.page_content,
                "source": chunk.metadata.get("source", "unknown"),
                "page": chunk.metadata.get("page", 0),
            }
        ))
        # 批量写入，减少API调用
        if len(points) >= 64:
            qdrant.upsert(collection_name=COLLECTION_NAME, points=points)
            points = []
    if points:  # 最后一批
        qdrant.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"已索引 {len(chunks)} 个chunk")

4.5 混个检索：向量 + BM25 + RRF融合

只用向量检索会出现一个坑：涉及专有名词或精确代码时，向量检索往往不如传统关键词检索。采用「向量 + BM25」混合检索，用RRF算法融合排序。

# hybrid_search.py
import math
from collections import defaultdict
from rank_bm25 import BM25Okapi
import jieba
from qdrant_client import QdrantClient

qdrant = QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT)

def search_vector(query: str) -> list[dict]:
    vec = get_embedding(query)
    results = qdrant.search(
        collection_name=COLLECTION_NAME,
        query_vector=vec,
        limit=TOP_K_VECTOR,
        score_threshold=0.35,  # 低于0.35的不要
    )
    return [{"id": hit.id, "score": hit.score, "text": hit.payload["text"]} for hit in results]

def search_bm25(query: str, all_docs: list[str]) -> list[dict]:
    # 中文分词后构建 BM25 索引
    tokenized_corpus = [list(jieba.cut(doc)) for doc in all_docs]
    bm25 = BM25Okapi(tokenized_corpus)
    tokenized_query = list(jieba.cut(query))
    scores = bm25.get_scores(tokenized_query)
    
    results = []
    for idx, score in enumerate(scores):
        if score > 0:
            results.append({"id": idx, "score": score, "text": all_docs[idx]})
    # 按分数降序、取前5
    results.sort(key=lambda x: x["score"], reverse=True)
    return results[:TOP_K_BM25]

def rrf_fusion(vector_results: list[dict], bm25_results: list[dict], k: int = 60) -> list[dict]:
    """RRF: RRF(d) = Σ 1/(k + rank_i(d))"""
    scores = defaultdict(float)
    rank_map = {}
    
    for rank, item in enumerate(vector_results):
        scores[item["id"]] += 1.0 / (k + rank + 1)
        rank_map[item["id"]] = item["text"]
    
    for rank, item in enumerate(bm25_results):
        scores[item["id"]] += 1.0 / (k + rank + 1)
        rank_map[item["id"]] = item["text"]
    
    # 融合排序结果
    fused = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [{"id": pid, "score": score, "text": rank_map[pid]} for pid, score in fused]

4.6 完整RAG链路：检索 → 拼接上下文 → 生成

# rag_pipeline.py
from config import *
import json

SYSTEM_PROMPT = """你是企业知识库助手，回答必须基于以下资料片段。
如果资料中没有答案，明确告知「资料中未找到相关信息」，不要编造。
引用来源时要标注[文档名:页码]。"""

def build_context(fused_results: list[dict], max_chars: int = 3000) -> str:
    context = ""
    for item in fused_results:
        text = item["text"].strip()
        if len(context) + len(text) > max_chars:
            continue
        context += text + "\n\n"
    return context

def rag_answer(query: str) -> dict:
    # 1. 混合检索
    vector_results = search_vector(query)
    all_docs = [doc["text"] for doc in qdrant.scroll(collection_name=COLLECTION_NAME)[0]]
    bm25_results = search_bm25(query, all_docs)
    fused_results = rrf_fusion(vector_results, bm25_results)
    
    if not fused_results:
        return {"answer": "数据库中未找到相关资料。", "sources": []}
    
    # 2. 拼装上下文
    context = build_context(fused_results)
    
    # 3. 调用LLM生成
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"请根据以下资料回答问题：\n\n【资料】\n{context}\n\n【问题】{query}"}
    ]
    
    resp = client.chat.completions.create(
        model=LLM_MODEL,
        messages=messages,
        temperature=0.1,  # 知识库问答用低温，减少幻觉
    )
    
    answer = resp.choices[0].message.content
    
    # 4. 返回答案 + 来源
    sources = [{"text": item["text"][:100], "score": item["score"]} for item in fused_results[:3]]
    return {"answer": answer, "sources": sources}

4.7 API接口：FastAPI 部署成服务

# api.py
from fastapi import FastAPI
from pydantic import BaseModel
import time
from rag_pipeline import rag_answer

app = FastAPI(title="RAG Knowledge Base API")

class QueryRequest(BaseModel):
    query: str
    session_id: str = "default"

class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    latency_ms: int

@app.post("/api/rag", response_model=QueryResponse)
def answer_question(req: QueryRequest):
    t0 = time.time()
    result = rag_answer(req.query)
    latency = int((time.time() - t0) * 1000)
    return QueryResponse(
        answer=result["answer"],
        sources=result["sources"],
        latency_ms=latency,
    )

@app.get("/health")
def health():
    return {"status": "ok"}

五、效果数据：没有对比就没有发言权

测试使用20份内部制度文档（PDF），共356页、约87,000字，切成312个chunk。准备了50个常见问题，人工标注了标准答案和相关文档页码。

5.1 向量 vs 混合检索：准确率对比

检索方式命中率（top5含正确答案）准确率（top1）平均检索耗时
纯向量检索82%（41/50）58%（29/50）210ms
纯BM25关键词64%（32/50）36%（18/50）42ms
混合检索(RRF)96%（48/50）78%（39/50）255ms

混合检索在高频问题命中率上优势明显，top5命中率96%，比纯向量多成功7个问题。

5.2 RAG vs 裸GPT-4：回答正确率

方式正确率幻觉率平均生成耗时
裸GPT-436%（18/50）44%（22/50）2.1s
GPT-4 + RAG86%（43/50）8%（4/50）2.9s

加RAG后正确率提升50个百分点，幻觉率骤降至8%。额外耗时主要来自检索和上下文拼接。

5.3 切块大小对效果的影响

Chunk大小重叠词数检索准确率上下文token消耗
2563282%约600 tok
5126496%约1200 tok
102412878%约2400 tok

512是最优值。chunk太小上下文信息不完整，太大则引入噪声并烧token。

5.4 响应延迟分析（p95，50次请求）

环节平均耗时（ms）占比
文档加载（首次）—一次性
Embedding生成58016%
向量检索822.3%
BM25检索381.1%
RRF融合30.1%
LLM生成278080%

延迟瓶颈在LLM生成。如果使用4o-mini，生成耗时降至910ms，整体延迟可以控制在2s内。

六、避坑指南（我实际踩过的5个坑）

坑1：openai-python版本兼容性问题

openai库从0.x升级到1.x后接口完全变了，很多老教程用的是openai.Completion.create()，新版本是client.chat.completions.create()。装新版库前务必查好API签名。我在初版用了openai 0.28，代码全废。上面代码基于OpenAI Python SDK v1.x。

坑2：中文文档切块踩「语义腰斩」

直接用LangChain默认的文本切分把中文一句话从中间切开，导致检索时语义缺失。解决办法是设置separators=["\n\n", "\n", "。", "！", "？", "；"]，让切块边界落在句子结束后而不是单词中间。同时加64字符重叠，保持上下文连续性。

坑3：Qdrant版本升级带来的API不兼容

Qdrant 1.x版本的recreate_collection签名和0.x不同。1.9.0的写法是vectors_config=VectorParams(size=1536, distance=Distance.COSINE)，而0.x版本是vector_size=1536, distance=Distance.COSINE。两个版本API不兼容，报错信息还不明显，排查了很久。

坑4：chunk数量少时，向量检索「高分低能」

当公司只有十几份文档、300来个chunk时，向量检索经常出现高相关度分数但内容错误的情况。这就是低candidate池导致的结果。解决办法是降低score_threshold到0.3，同时引入BM25兜底。

坑5：pdf文本提取质量差

PyPDFLoader对扫描版PDF完全无解——你提取出来的是图片不是文字。如果有扫描件，需要先用OCR预处理（推荐paddleocr，但要下载模型包）。

另外PyPDFLoader对嵌入的表格结构还原很差，表格内容会乱序。建议先把重要表格转成markdown格式的文本文件再入库。

七、什么时候用RAG，什么时候还是去微调？

这是每次分享都会被问到的问题，直接给你判断标准：

  • 数据更新频率：按周更新的选RAG，按季度甚至按年更新的才考虑微调
  • 数据量：少于5000条文本的用RAG划算，几十万条的海量数据微调更有优势
  • 推理延迟：RAG在检索阶段多200ms左右延迟，微调不增加延迟
  • 硬件投入：RAG部署只需一台2核4G的机器就能跑，微调最好有4张以上的A100

八、后续优化方向

  • Embedding模型从text-embedding-3-small换成bge-m3或在开源模型上微调，中文场景能再提升2%-5%准确率
  • 针对用户意图判断增加路由层：输入是简单事实问题走BM25，复杂逻辑问题走混合检索
  • 引入重排序模型（bge-reranker-v2-m3）对检索回来的20个chunk精排，效果提升显著但会增加100ms左右延迟

代码你已经拿到了，回去跑通它，用自己的数据试一遍，上手会比看十篇概念文章都快。