从原理走到工程：RAG实战笔记

作者: 醉月思发布日期: 2024-07-03 分类: 技术实践

上周帮一个知识库项目做 RAG 调优，先把"答不上来"和"胡编"两类失败复现出来，再往下改切分和检索。

背景：为什么需要 RAG

那个项目要做一个内部知识库，把各种文档、邮件、会议纪要做进一个系统，让员工可以自然语言查询。

一开始的方案是直接调用大模型 API，把问题丢过去。很快发现问题：模型不知道公司内部的事情，回答全是"我没有这个信息"或者胡编乱造。买定制化服务？成本太高，而且数据还要传出去。

这时候就轮到 RAG（Retrieval-Augmented Generation）上场了。思路很简单：先把文档切分成小块，向量化后存进向量数据库；用户提问时，先把问题也向量化，在向量库里找最相关的内容片段，再把这些片段和问题一起丢给大模型，让它基于真实信息回答。

说起来简单，真做起来全是细节。

文档处理：切分比你想的麻烦

第一步是文档处理。拿到一堆 PDF、Word、Markdown，需要先转成文本，然后切成合理大小的片段。

切分是第一个坑。很多人一开始用固定长度切，比如每段 500 个字符，结果切出来的内容要么太碎语义不完整，要么太大检索不准。

我用的是一个比较实用主义的方案：先按段落、标题、章节这些自然边界切，再检查每段长度，超过阈值就递归往下切。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, UnstructuredMarkdownLoader
import re

def load_and_split(file_path, chunk_size=500, chunk_overlap=50):
    # 根据文件类型选择 loader
    if file_path.endswith('.pdf'):
        loader = PyPDFLoader(file_path)
    elif file_path.endswith('.md'):
        loader = UnstructuredMarkdownLoader(file_path)
    else:
        raise ValueError(f"不支持的文件类型: {file_path}")

    documents = loader.load()

    # 使用递归字符分割器，优先按自然边界切分
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", ";", "，", " ", ""]
    )

    splits = text_splitter.split_documents(documents)

    # 给每个片段添加元数据，方便后续追溯
    for i, split in enumerate(splits):
        split.metadata['chunk_id'] = f"{file_path}_{i}"
        split.metadata['source'] = file_path

    return splits

这里有几个经验值：

chunk_size 设在 400-800 之间比较合适。太小了上下文信息不够，太大了检索不精准。
chunk_overlap 设 50-100，确保语义边界不会硬切断。
分离器里的 separators 顺序很重要，优先用大的自然边界（段落、句子），最后才是字符。

踩过一个坑：有些 PDF 的表格被切得面目全非，本来是一行的数据被拆成几段，检索出来完全看不懂。后来加了专门的表格识别逻辑，遇到表格就整块保留。

向量数据库：选型要看实际场景

向量数据库选了不少。一开始试过 Pinecone，云服务省事，但数据要传出去，有些场景不适合；也试过 Milvus，功能强大但部署成本高，本地跑起来资源占用不小。

最后用了 ChromaDB。本地部署简单，Python API 友好，轻量场景够用。

import chromadb
from chromadb.config import Settings
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 初始化向量数据库
client = chromadb.PersistentClient(path="./chroma_db")
collection_name = "knowledge_base"

# 使用 OpenAI 的嵌入模型
embeddings = OpenAIEmbeddings(
    openai_api_key="your-api-key",
    model="text-embedding-3-small"  # 比大模型便宜，效果够用
)

# 创建或加载向量库
vectorstore = Chroma(
    client=client,
    collection_name=collection_name,
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# 批量添加文档
def add_documents(splits, batch_size=100):
    for i in range(0, len(splits), batch_size):
        batch = splits[i:i+batch_size]
        vectorstore.add_documents(batch)
        print(f"已添加 {i + len(batch)}/{len(splits)} 个文档片段")

这里有个性能问题：向量化很慢，几千个文档能跑半天。后来做了个简单的批量处理，每 100 个一批，加上进度打印，至少知道还要等多久。

还有一个坑：向量数据库的索引如果不重建，检索效果会越来越差。我写了个简单的重建逻辑，每周跑一次。

def rebuild_index():
    # 备份当前数据
    backup_dir = f"./chroma_db_backup_{datetime.now().strftime('%Y%m%d')}"
    shutil.copytree("./chroma_db", backup_dir)

    # 重建索引
    client = chromadb.PersistentClient(path="./chroma_db")
    collection = client.get_collection("knowledge_base")

    # 获取所有数据
    all_data = collection.get(include=['documents', 'metadatas', 'embeddings'])

    # 删除并重建集合
    client.delete_collection("knowledge_base")
    collection = client.create_collection("knowledge_base")

    # 重新添加数据
    collection.add(
        documents=all_data['documents'],
        metadatas=all_data['metadatas'],
        embeddings=all_data['embeddings'],
        ids=[f"doc_{i}" for i in range(len(all_data['documents']))]
    )

检索增强：相似度不够用时

核心的检索逻辑也踩了不少坑。最开始用最简单的相似度检索，只取 top 5，结果很多问题回答得不好。

问题出在几个地方：

有些问题的关键词被模型换了个说法，相似度就掉下去了
有些问题需要多方面的信息，但相似度检索可能只偏向某个方面
有些文档片段虽然相似度不高，但实际内容更相关

后来用了混合检索：相似度 + 关键词匹配，再加上结果重排序。

from langchain.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_community.document_compressors.cross_encoder import CrossEncoderReranker

def create_ensemble_retriever(vectorstore, documents, top_k=10):
    # 相似度检索
    vector_retriever = vectorstore.as_retriever(search_kwargs={"k": top_k * 2})

    # BM25 关键词检索
    bm25_retriever = BM25Retriever.from_documents(documents)
    bm25_retriever.k = top_k * 2

    # 组合两种检索
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.5, 0.5]  # 可以调整权重
    )

    # 使用交叉编码器重排序
    cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
    compressor = CrossEncoderReranker(compressor=cross_encoder, top_n=top_k)

    compression_retriever = ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=ensemble_retriever
    )

    return compression_retriever

重排序这一步加进去后，检索效果明显提升，但也带来了性能问题。原来几百毫秒的检索变成了几秒，用户体验下降。

折中的方案是：简单问题不用重排序，只有当相似度分数不够高、或者检测到问题比较复杂时才启用重排序。

生成：如何让模型不胡说

检索搞定了，但生成的答案还是会有问题。有时候模型会基于检索到的内容编造细节，有时候会忽略重要信息。

试了几个方案：

在 prompt 里明确要求"只基于提供的上下文回答"
让模型先列出答案的依据，再给出答案
限制模型的回答长度，避免它过度发挥

最终一个比较实用的方案是：让模型生成答案的同时，把引用的来源也标出来。

from langchain.chains import RetrievalQAWithSourcesChain
from langchain_openai import ChatOpenAI

# 创建带引用的 QA 链
llm = ChatOpenAI(
    model_name="gpt-4o-mini",  # 比 GPT-4 便宜，质量够用
    temperature=0,  # 降低随机性
    openai_api_key="your-api-key"
)

chain = RetrievalQAWithSourcesChain.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={
        "prompt": """你是一个专业的知识库助手。请基于以下上下文回答问题。

上下文：
{context}

问题：{question}

请用中文回答，并在回答中用【来源：文档名】的形式标注信息来源。如果上下文中没有相关信息，请直接说"我没有找到相关信息"，不要编造答案。

答案："""
    }
)

# 使用
result = chain.invoke({"question": "公司的休假政策是什么？"})
print(result["answer"])
print("来源：", result["source_documents"])

这样用户不仅能看到答案，还能追溯到原文，信任度会高不少。

性能优化：快和准是矛盾

项目上线后发现，慢的问题还是存在。一次查询经常要 3-5 秒，用户等不及。

从几个层面优化：

向量层面：换用了更快的嵌入模型 text-embedding-3-small，速度提升 30%
检索层面：缓存常见问题的答案，命中率能到 20%
生成层面：对于简单问题，直接使用检索到的片段，不调用大模型

import hashlib
import json
from functools import lru_cache

# 简单的答案缓存
cache_file = "./answer_cache.json"

def load_cache():
    try:
        with open(cache_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    except:
        return {}

def save_cache(cache):
    with open(cache_file, 'w', encoding='utf-8') as f:
        json.dump(cache, f, ensure_ascii=False, indent=2)

def get_cache_key(question):
    return hashlib.md5(question.encode()).hexdigest()

def query_with_cache(question):
    cache = load_cache()
    cache_key = get_cache_key(question)

    if cache_key in cache:
        print("命中缓存")
        return cache[cache_key]

    # 实际查询
    result = chain.invoke({"question": question})
    cache[cache_key] = result
    save_cache(cache)

    return result

还有一些小的优化：比如批量向量化、异步请求、数据库连接池复用，加起来能再快一点。

但最后还是得承认：RAG 的速度受限于向量检索和大模型生成，想做到毫秒级响应很难。关键是要把复杂的查询控制在几秒内，简单查询尽量做到几百毫秒。

踩过的那些坑

这次实践踩过不少坑，记录下来省得下次再踩：

切分问题：切得太碎导致上下文不够，切得太粗导致检索不准。最后用了递归切分 + 自然边界的方案。
向量数据库索引：索引不重建会导致效果越来越差，需要定期重建。
相似度检索局限：纯相似度检索会有很多漏网之鱼，混合检索 + 重排序效果好但慢，需要根据场景平衡。
模型幻觉：再好的检索也可能被模型编造成分，需要严格限制模型的行为。
性能瓶颈：向量检索和模型生成都是慢操作，需要多层优化和缓存策略。
成本控制：嵌入模型和生成模型的调用成本都不低，需要合理选择模型和使用策略。

今天的经验

折腾了一圈，对 RAG 的看法更接地气了：它得跟着场景调，没有一劳永逸的架构。

效果和性能永远在打架，按业务取舍
用户反馈比闭门调参管用
持续迭代比一次搭完美流水线重要

项目跑了几个月，满意度还行，多轮对话、个性化、实时索引这些还排着队。技术没有终点，只有下一段要修的路。

版权声明: 本文首发于指尖魔法屋-从原理走到工程：RAG实战笔记（https://blog.thinkmoon.cn/post/138-rag-practice-vector-database/）转载或引用必须申明原指尖魔法屋来源及源地址！

从LangChain走到AutoGPT：AI Agent开发笔记

那时候 AI Agent 还没现在这么火，LangChain 也刚到 0.1 版本，我还在想这玩意儿是不是只是又一层封装。

📁 技术实践

📅 2024-04-02

AI编排：这次怎么落地的

从单一的模型调用到复杂的工作流编排，中间到底跨越了哪些坑，以及怎么才能尽量少踩一点。

📁 技术实践

📅 2024-08-22

AI多跳RAG折腾手记

比如： 1. 第一跳：查"项目A的架构设计"，得到微服务架构、事件驱动等信息 2. 第二跳：基于上一跳结果，查"项目B的架构设计"，得到类似结…

📁 技术实践

📅 2025-04-26

从原理走到工程：RAG实战笔记

背景：为什么需要 RAG

文档处理：切分比你想的麻烦

向量数据库：选型要看实际场景

检索增强：相似度不够用时

生成：如何让模型不胡说

性能优化：快和准是矛盾

踩过的那些坑

今天的经验

评论

相关阅读