← 返回博客
·AI技术

RAG 系统踩坑记:从原型到生产的那些坑

去年我做了一个 RAG 系统,从原型到上线,踩了一堆坑。文档切分、向量检索、LLM 脑补答案、文档更新、成本失控,每个环节都有坑。分享实战经验,让后来人少走弯路。

#RAG#向量检索#LLM#生产部署

# RAG 系统踩坑记:从原型到生产的那些坑

去年我做了一个 RAG(检索增强生成)系统,从原型到上线,踩了一堆坑。想把这些经验写出来,让后来人少走点弯路。

先说说背景

我们想做一个内部知识库问答系统。公司有几百份技术文档、会议纪要、项目文档,散落在各个地方。员工想找某个技术问题的答案,得翻半天。

老板说:"整个 RAG 吧,让 AI 帮他们找。"

我想着,RAG 不是很成熟了吗?LangChain 一把梭,应该很快就能搞定。

然后我就掉坑里了。

坑一:文档切分没那么简单

教科书上说,RAG 的流程是:

  • 把文档切成 chunk
  • 2. 把 chunk 向量化

    3. 存到向量数据库

    4. 用户提问时,检索相关 chunk

    5. 把 chunk 喂给 LLM 生成答案

    看起来很简单对吧?但第一步"把文档切成 chunk"就搞死我了。

    固定大小切分?不行

    最开始我用的是最简单的策略:按固定大小切分,比如每个 chunk 500 个 token,重叠 50 个 token。

    from langchain.text_splitter import RecursiveCharacterTextSplitter

    text_splitter = RecursiveCharacterTextSplitter(

    chunk_size=500,

    chunk_overlap=50

    )

    documents = text_splitter.split_documents(raw_docs)

    然后问题来了。有份技术文档讲的是"用户认证流程",从"用户输入密码"到"返回 JWT token",是一个完整的流程。但我的切分把它切成了三截:

  • Chunk 1: "...用户输入密码,系统会..."
  • Chunk 2: "...验证密码是否正确,如果正确则..."
  • Chunk 3: "...生成 JWT token,返回给客户端..."
  • 用户问"我们的用户认证流程是怎样的?",检索的时候可能只命中了 Chunk 2,然后 LLM 给出的答案就是残缺的。

    按语义切分?也不行

    然后我试了按语义切分,用 embedding 计算相邻句子的相似度,相似度低的地方就切分。

    from langchain.text_splitter import SemanticChunker

    from langchain.embeddings import OpenAIEmbeddings

    embeddings = OpenAIEmbeddings()

    semantic_splitter = SemanticChunker(

    embeddings,

    breakpoint_threshold_type="percentile"

    )

    documents = semantic_splitter.create_documents([text])

    这个方法好一点,但还是有问题。有些文档里,不同章节的话题跳跃很大,但语义相似度却很高(因为用了相同的术语)。结果该切的地方没切,不该切的地方切了。

    最终方案:混合策略

    我最后的方案是:

  • 先按文档的天然结构切分(标题、段落)
  • 2. 如果某个块太大(超过 800 token),再用语义切分

    3. 如果语义切分后太小(小于 100 token),就合并到相邻的块

    def hybrid_split(document):

    # 按标题切分

    sections = split_by_headings(document)

    chunks = []

    for section in sections:

    if len(section) > 800:

    # 语义切分

    sub_chunks = semantic_split(section)

    chunks.extend(sub_chunks)

    else:

    chunks.append(section)

    # 合并太小的 chunk

    chunks = merge_small_chunks(chunks, min_size=100)

    return chunks

    这个方案不完美,但比之前好多了。关键是,你得根据你的文档特点来调整策略,没有万能的方案。

    坑二:向量检索的准确率不够

    原型阶段,我用的是简单的向量相似度检索:

    from langchain.vectorstores import Chroma

    db = Chroma.from_documents(documents, embeddings)

    retriever = db.as_retriever(search_kwargs={"k": 3})

    检索 top-3 相关的 chunk,然后喂给 LLM。

    问题是什么?向量相似度不等于语义相关性。

    比如用户问"如何部署到生产环境?",向量检索可能返回:

  • Chunk A: "在开发环境中,你可以使用 docker-compose 来部署..."(相似度 0.85)
  • Chunk B: "生产环境的配置需要注意以下几点..."(相似度 0.82)
  • Chunk C: "部署前需要运行的测试脚本..."(相似度 0.78)
  • 但真正有用的可能是另一个 chunk:

  • Chunk D: "生产环境部署的完整步骤:1. 备份数据库 2. 拉取最新代码..."(相似度 0.75)
  • 因为 Chunk D 里有很多部署相关的关键词,但"生产环境"这个词出现得少,所以向量相似度反而低了。

    解决方案:混合检索

    我加了一个关键词检索(BM25),跟向量检索的结果做融合:

    from rank_bm25 import BM25Okapi

    import numpy as np

    # 向量检索

    vector_results = vector_db.similarity_search(query, k=10)

    # 关键词检索

    tokenized_corpus = [doc.page_content.split() for doc in all_docs]

    bm25 = BM25Okapi(tokenized_corpus)

    tokenized_query = query.split()

    bm25_scores = bm25.get_scores(tokenized_query)

    bm25_results = sorted(zip(all_docs, bm25_scores), key=lambda x: x[1], reverse=True)[:10]

    # 融合

    from rankfusion import reciprocal_rank_fusion

    merged_results = reciprocal_rank_fusion(

    [vector_results, [doc for doc, _ in bm25_results]]

    )

    这样准确率提升了不少。但实现起来复杂了很多,而且你要维护两套索引。

    坑三:LLM 会"脑补"答案

    这是最要命的。即使检索到的 chunk 里没有明确答案,LLM 也会根据上下文"推理"出一个答案。而且它说得特别自信。

    比如有次我问"我们公司的年假政策是什么?",检索到的 chunk 里只有一句话:"员工福利包括年假、病假和产假。"

    然后 LLM 生成了一段长长的回答:"根据公司政策,入职满一年的员工享有 10 天年假,满三年的享有 15 天..."

    纯属胡说八道。但用户看了觉得挺合理,就信了。

    解决方案:加引用 + 不确定性提示

    我的做法是:

  • 让 LLM 在回答时必须引用原文
  • 2. 如果检索到的 chunk 里没有明确答案,就让 LLM 说"根据现有文档,无法找到明确答案"

    prompt = f"""

    根据以下参考资料回答问题。回答时必须引用原文,格式为 [来源X]。

    如果参考资料中没有明确答案,请回答"根据现有文档,无法找到相关信息"。

    参考资料:

    {context}

    问题:{query}

    """

    response = llm(prompt)

    这样至少用户能知道答案从哪来的,可以去核实。

    但这样做也有副作用:用户体验变差了。有些人就想要一个直接了当的答案,不想看引用。

    这是个权衡。

    坑四:文档更新是噩梦

    文档是会更新的。今天这份技术文档是 v1.0,明天可能就变成 v1.1 了。

    但向量数据库里的 embedding 是静态的。如果文档更新了,你得重新切分、重新向量化、重新建索引。

    我们最开始的做法是:每天凌晨全量重建索引。但这样做有两个问题:

  • 重建索引要跑很久(我们有几万份文档)
  • 2. 如果某份文档删除了,你还得从向量数据库里删掉对应的 chunk,这很难做

    解决方案:版本化 + 增量更新

    我最后的方案是:

  • 每份文档都有一个版本号
  • 2. 每次更新文档,都生成一个新的版本

    3. 向量数据库里存的是"文档 ID + 版本号"

    4. 定期清理旧版本的 embedding

    class DocumentStore:

    def update_document(self, doc_id, content):

    # 生成新版本

    new_version = self.get_latest_version(doc_id) + 1

    # 切分、向量化

    chunks = self.split_document(content)

    embeddings = self.embed_chunks(chunks)

    # 存入向量数据库

    for chunk, embedding in zip(chunks, embeddings):

    self.vector_db.add(

    embedding=embedding,

    metadata={

    'doc_id': doc_id,

    'version': new_version,

    'content': chunk

    }

    )

    # 标记旧版本为过期(不直接删除,避免检索时出错)

    self.mark_old_versions_as_stale(doc_id, new_version)

    这样做,至少能保证数据的一致性。但实现复杂度又上了一个台阶。

    坑五:成本失控

    这是我最后才意识到的坑。

    每次用户提问,都要:

  • 调用 embedding API 把问题向量化(便宜,但也有成本)
  • 2. 调用向量数据库做检索(便宜)

    3. 调用 LLM 生成答案(贵!)

    我们用的 GPT-4,每次生成答案大概要消耗 2000-3000 个 token(包括 context)。按 OpenAI 的定价,这就是几分钱美元。

    听起来不多对吧?但我们有几百个员工,每天几千次查询,一个月下来就是几千美元。

    解决方案:缓存 + 换便宜的模型

    我做的两件事:

  • **缓存相似问题的答案**:如果新问题跟历史问题的相似度超过某个阈值,直接返回历史答案
  • def get_answer(query):

    # 先看缓存

    similar_queries = find_similar_queries(query, threshold=0.95)

    if similar_queries:

    return cache.get(similar_queries[0])

    # 否则正常检索 + 生成

    answer = generate_answer(query)

    cache.set(query, answer)

    return answer

    2. **用更便宜的模型**:对于简单的问题,用 GPT-3.5 或者开源模型(比如 Llama 3)就够了

    def choose_model(query):

    complexity = estimate_complexity(query)

    if complexity < 0.5:

    return "gpt-3.5-turbo"

    else:

    return "gpt-4"

    这样成本降了大概 60%。

    一些其他的经验

    1. Chunk 大小不是越大越好

    有些人觉得,chunk 越大,上下文越完整,效果越好。但其实不是。

    chunk 太大,会引入太多无关信息,反而干扰 LLM。而且 token 消耗也更大。

    我试下来,300-800 token 是个比较合理的范围。

    2. 元数据很重要

    别只存 chunk 的文本内容,把元数据也存上:

  • 文档标题
  • 文档来源(URL、文件路径)
  • 最后更新时间
  • 作者
  • 这些信息可以帮助你做更精细的检索和排序。

    metadata = {

    'source': 'technical_docs/auth.md',

    'title': '用户认证流程',

    'last_updated': '2024-01-15',

    'author': '张三',

    'tags': ['authentication', 'jwt', 'security']

    }

    3. 评估很重要,但很难做

    你想知道你的 RAG 系统效果怎么样,但不能只看"用户满不满意"。

    我尝试做了一个评估框架:

  • 准备一组标准问题 + 标准答案
  • 定期跑这些问题的准确率、召回率
  • 跟踪用户反馈(点赞/点踩)
  • 但这样做成本很高,而且标准答案也不一定对。最后我们还是主要靠用户反馈来评估。

    总结

    RAG 系统看起来简单,但要做到生产级别,坑很多。

    关键的是:

  • **别迷信原型**:原型能跑 ≠ 生产能用
  • 2. **文档质量决定上限**:如果你的文档本身就很烂,RAG 也救不了你

    3. **持续迭代**:RAG 不是"一次性做好"的系统,需要不断优化

    最后说一句:如果你只是想做个简单的问答系统,试试 Perplexity API 或者 Vercel 的 AI SDK,可能比自己搭 RAG 更划算。

    但如果你想深度定制,那欢迎入坑。记得多留点预算和时间。


    *写这篇文章的时候,我又去看了眼我们的 RAG 系统,发现有个 bug:如果问题里包含特殊字符,检索会挂。得去修了。*