RAG 系统踩坑记:从原型到生产的那些坑
去年我做了一个 RAG 系统,从原型到上线,踩了一堆坑。文档切分、向量检索、LLM 脑补答案、文档更新、成本失控,每个环节都有坑。分享实战经验,让后来人少走弯路。
# RAG 系统踩坑记:从原型到生产的那些坑
去年我做了一个 RAG(检索增强生成)系统,从原型到上线,踩了一堆坑。想把这些经验写出来,让后来人少走点弯路。
先说说背景
我们想做一个内部知识库问答系统。公司有几百份技术文档、会议纪要、项目文档,散落在各个地方。员工想找某个技术问题的答案,得翻半天。
老板说:"整个 RAG 吧,让 AI 帮他们找。"
我想着,RAG 不是很成熟了吗?LangChain 一把梭,应该很快就能搞定。
然后我就掉坑里了。
坑一:文档切分没那么简单
教科书上说,RAG 的流程是:
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 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:
因为 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 天..."
纯属胡说八道。但用户看了觉得挺合理,就信了。
解决方案:加引用 + 不确定性提示
我的做法是:
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)
这样做,至少能保证数据的一致性。但实现复杂度又上了一个台阶。
坑五:成本失控
这是我最后才意识到的坑。
每次用户提问,都要:
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 的文本内容,把元数据也存上:
这些信息可以帮助你做更精细的检索和排序。
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:如果问题里包含特殊字符,检索会挂。得去修了。*
VkingAI