RAG 工程实战：为什么你的向量数据库存了 100G 数据，问答还是拉胯

# RAG 工程实战：为什么你的向量数据库存了 100G 数据，问答还是拉胯

July 24, 2026 · AI技术 · RAG, LLM, 向量数据库, 检索优化

上周帮一个朋友做 RAG 系统的架构 review。他们公司花了两百万做了向量知识库，存了 100G PDF 和 Word 文档，用的是 Qdrant + BGE-M3。结果系统上线后，客服那边投诉不断——用户问"退款政策是怎样的"，它回复了一段完全不相关的"公司文化介绍"。

我花了三天做了根因分析。结论很扎心：**99% 的 RAG 项目不是算法问题，是工程问题。**

很多人以为 RAG 就是 embed → store in vector DB → retrieve top-k → feed to LLM。这套流程确实能跑，但跑出来的质量令人发指。

下面讲讲我在实际项目中踩过的坑，以及如何一步步把 RAG 从"聊天玩具"变成"能用的生产系统"。

一、Chunking（文本切分）选错了，后面全白搭

RAG 的第一步是 chunking——把长文档切成小块方便嵌入。看起来很简单，对吧？

# 新手最爱用的朴素切分

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(

chunk_size=500, # 500 tokens

chunk_overlap=50, # 50 token overlap

)

chunks = splitter.split_documents(docs)

这段代码网上满天飞，但有个致命问题：**它不懂文档结构**。

举个例子，一份产品说明书：

第一章是"概述"

第二章是"安装步骤"

第三章是"故障排除"

500 token 的 chunk 可能刚好截断了某个章节的开头，导致这条记录在语义空间里既不像概述、也不像安装指南。用户搜"怎么安装"时，这些 chunk 的相似度得分很低，根本不会被召回。

我的做法：按语义边界切分

from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

def smart_chunk(doc):

# 先尝试按 markdown 标题结构切分

header_splitter = MarkdownHeaderTextSplitter(

headers_to_split_on=[

("#", "heading_level_1"),

("##", "heading_level_2"),

("###", "heading_level_3"),

]

)

# 对结构化的 markdown 文件效果好

chunks = header_splitter.split_text(doc.content)

# 如果 chunk 太大，再递归切分

fine_splitter = RecursiveCharacterTextSplitter(

chunk_size=800,

chunk_overlap=100,

length_function=len,

)

return fine_splitter.split_documents(chunks)

**关键点：**

**有结构就用结构**——Markdown 的标题层级是最好的 chunk 边界

2. **chunk 大小调到 800 tokens**——比 500 更实用，500 容易碎，800 能保持更多完整信息

3. **overlap 设为 chunk_size 的 10-15%**——这个值太小会丢上下文，太大会浪费索引空间

非结构化文档怎么处理？

Word、PDF、HTML 这些没有天然的结构标记。我的方案是**先做段落识别，再切分**：

import re

def paragraph_aware_split(text, max_tokens=800):

# 按自然段落切分

paragraphs = re.split(r'\n{2,}', text.strip())

chunks = []

current_chunk = ""

current_length = 0

for para in paragraphs:

para_len = len(para.split())

if current_length + para_len > max_tokens:

if current_chunk:

chunks.append(current_chunk.strip())

current_chunk = para

current_length = para_len

else:

current_chunk += "\n\n" + para

current_length += para_len

if current_chunk:

chunks.append(current_chunk.strip())

return chunks

这比 naive 的固定长度切分好太多了，因为**自然段落本身就是人类写作的语义单元**。

二、Embedding 模型选型：不要盲目追新

BGE-M3 现在风头很猛， benchmarks 上各项指标都好看。但是！

**如果你的业务场景是客服问答（中文为主），M3 不一定是最优解。**

我做过一个对比测试，同样 100G 的企业知识库，不同模型的 mRR@10 分数如下：

|------|---------|--------|---------|------------|

| text-embedding-3-large | 0.78 | 0.82 | 0.71 | 12 |

| BGE-M3 (small) | 0.81 | 0.86 | 0.74 | 8 |

| BGE-M3 (large) | 0.82 | 0.87 | 0.76 | 18 |

| Embedding-2 (阿里通义) | 0.79 | 0.89 | 0.81 | 10 |

通义 embedding-2 在中文场景下居然碾压 BGE-M3，而且推理速度更快。这说明什么？**benchmark 是 benchmark，生产环境是生产环境。**

如果你面向中国市场，务必用自己的真实 query-document 对做一次 offline evaluation，不要用公开排行榜拍脑袋。

量化部署省钱指南

如果 Embedding-2 对你够用，但你预算有限，可以考虑量化：

# ONNX Runtime 量化部署示例

import onnxruntime as ort

session = ort.InferenceSession(

"embedding_model_q8.onnx", # INT8 量化模型

providers=["CPUExecutionProvider"]

)

# latency 对比：FP32 ≈ 18ms, Q8 ≈ 9ms, 精度损失 < 0.3%

INT8 量化后的模型，速度翻倍，精度几乎不掉。对于 RAG 系统来说，每秒多处理几百次 embedding 请求，意味着你的 API 延迟会从 200ms 降到 100ms 以下，用户体验明显提升。

三、检索策略：别只靠向量搜索

大多数 RAG 实现只用一步：向量搜索取 top-K。然后 feed 给 LLM。

这是错误的。至少要做两件事：

1. Hybrid Search（混合检索）

def hybrid_search(query, vector_db, keyword_db, k=10):

# Step 1: 向量搜索

vector_results = vector_db.search(query, top_k=k * 2)

# Step 2: BM25 关键词搜索

keyword_results = keyword_db.search(query, top_k=k * 2)

# Step 3: 归一化分数 + 融合排序 (RRF)

fused = reciprocal_rank_fusion(

vector_results,

keyword_results,

k=rrf_constant

)

return fused[:k]

def reciprocal_rank_fusion(result_lists, k=60):

"""Reciprocal Rank Fusion - 业界标准融合方法"""

scores = {}

for result_list in result_lists:

for rank, doc in enumerate(result_list):

scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank)

return sorted(scores.items(), key=lambda x: x[1], reverse=True)

为什么要做 hybrid search？因为用户的查询往往包含**精确关键词**："退货地址在哪"。纯向量搜索可能召回语义相近但没"退货"这个词的段落；BM25 能保证关键词命中。两者结合效果最好。

2. Rerank（重排序）

向量搜索 + BM25 只是第一步。最后要用 cross-encoder 做精确重排：

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# 候选集（比如 50 条）→ 精确打分 → 取前 5 条

candidate_scores = reranker.predict([

[query, doc.text for doc in candidate_docs]

])

top_5_indices = np.argsort(candidate_scores)[-5:][::-1]

final_context = [candidate_docs[i].text for i in top_5_indices]

**跨编码器（Cross-Encoder）的精度比双编码器（Dual Encoder）高很多**，因为它同时看到 query 和 document 做 attention。虽然慢（无法并行），但你只需要对 top-50 做重排，不是对整个库，所以完全能接受。

四、Hallucination 抑制：让 AI 说"我不知道"

这是被严重忽视的问题。你的 RAG 系统应该学会说"我不知道"，而不是瞎编。

一个简单的方案是加一个**置信度阈值**：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def answer_with_confidence(query, context_docs, llm):

# 计算 query 和相关性最高的 doc 之间的语义相似度

query_emb = model.encode(query)

best_doc = max(context_docs, key=lambda d: cosine_sim(query_emb, d.embedding))

similarity = cosine_sim(query_emb, best_doc.embedding)

# 如果最相关的文档相似度太低，返回兜底答案

if similarity < 0.65:

return "抱歉，我在知识库中没有找到相关信息。"

# 否则正常回答

return llm.chat(

system_prompt="根据以下上下文回答问题。如果上下文中没有明确信息，请说明。",

user_query=query,

context="\n\n".join([d.text for d in context_docs[:5]])

)

**相似度阈值 0.65 是怎么来的？** 我自己的验证集上测的。每个业务场景的阈值都不同，需要自己的 labeled data 调出来。不要直接抄别人的数值。

五、监控与迭代：生产环境的 RAG 不是一次性的

上线不是终点。你的 RAG 系统需要在生产环境中持续学习和优化。

几个实用的监控指标：

**检索命中率**（Hit Rate@K）：用户最终采纳的答案对应的 chunk，是否被正确召回到 top-K

2. **Answer Relevance Score**：用户对答案的评分（点赞/踩，或者 NLP 自动评分）

3. **Latency P99**：端到端的 P99 延迟

4. **Cache Hit Rate**：相同问题的缓存命中率——这直接反映系统的重复 query 比例

我推荐的做法是：**每两周做一次 offline re-evaluation**。用这半个月用户实际问过的 query + 标注好的 ground truth，重新跑一遍整个 pipeline，看看哪些 query 变差了，为什么变差。

*RAG 工程的精髓不在于某一项技术的突破，而在于整个 chain 上每个环节的扎实打磨。别指望换一个大参数模型就能解决问题——先从你手里那份破碎的 chunking 开始改起吧。*