RAG系统中的知识准备与索引构建：深入刨析细节
·
一、引言：RAG系统的知识准备阶段
检索增强生成（Retrieval-Augmented Generation, RAG）已成为当前大语言模型应用的核心架构模式。RAG通过将外部知识检索与生成模型相结合，有效解决了大语言模型的幻觉问题、知识时效性问题和领域专业性不足等挑战。
在RAG系统的整体架构中，知识准备与索引构建阶段是决定系统性能的基石。这个阶段通常发生在离线环节，主要包括以下核心步骤：
二、数据集构建与准备
2.1 数据源类型与选择策略
RAG系统的数据来源多样化，需要根据应用场景合理选择和组合：
2.1.1 本地文档库
- 适用场景：企业内部知识管理、技术文档问答、合规文档检索
- 特点：私有性强、领域专业性高、更新频率可控
- 常见格式：PDF、Word、Excel、PPT、Markdown等
- 数据规模：根据企业规模，从数千到数百万文档不等
2.1.2 公开知识库（Wikipedia等）
- 适用场景：通用知识问答、百科类应用、教育辅助
- 特点：覆盖面广、质量相对可靠、持续更新
- 获取方式：API接口、数据集下载、爬虫采集
- 数据规模：Wikipedia英文版包含约600万篇文章
2.1.3 实时联网搜索
- 适用场景：新闻资讯、实时信息查询、动态知识更新
- 特点：时效性强、覆盖最新信息
- 技术挑战：延迟控制、内容质量筛选、去重处理
2.1.4 推荐系统知识
- 适用场景：电商推荐、内容推荐、个性化服务
- 特点：用户行为驱动、需要结构化数据支持
- 数据类型：用户画像、商品属性、交互历史
2.2 数据规模与质量要求
数据量级考量：
- 小规模（<10K chunks）：适合原型验证，可使用简单的向量库如Chroma
- 中等规模（10K-1M chunks）：需要考虑检索效率，推荐使用FAISS或Qdrant
- 大规模（>1M chunks）：需要分布式架构，考虑Milvus或Pinecone
数据质量保障：
- 去重处理：使用MinHash或语义相似度检测近似重复
- 噪声清理：移除页眉页脚、HTML标签、特殊字符
- 格式标准化：统一编码格式、日期格式、数字格式
- 错误纠正：拼写检查、语法修正
2.3 数据标注需求
是否需要标注取决于应用目标：
无需标注的场景：
- 纯检索应用（文档搜索、知识库查询）
- 使用预训练embedding模型即可满足需求
- 数据量大且标注成本高
需要标注的场景：
- 构建评估数据集（Question-Answer pairs）
- 微调embedding模型以适应特定领域
- 优化检索相关性（相关性标注）
评估数据集构建方法：
根据研究表明，可以通过以下方式高效构建评估数据集：
- 人工标注：小规模高质量数据集（通常100-1000个样本）
- LLM生成：使用GPT-4等模型自动生成Question-Context-Answer三元组
- 混合方式：LLM生成后人工审核和筛选
典型的RAG评估数据集格式：
{
"question": "What is the capital of France?",
"context": ["France is a country in Western Europe. Its capital is Paris..."],
"ground_truth": "Paris",
"metadata": {
"source": "geography_knowledge_base.pdf",
"difficulty": "easy"
}
}
三、文档解析：从多格式到结构化文本
文档解析是RAG管道中的第一道关键工序，直接影响后续所有环节的质量。不同文档类型需要采用不同的解析策略。
3.1 主流文档解析工具对比
根据2024年的综合评测研究，以下是主流工具的性能对比：
| 工具 | 类型 | 优势 | 劣势 | 最佳场景 |
|---|---|---|---|---|
| PyMuPDF | 开源 | 速度快（10x+），F1分数高（0.97+） | 科学文档表现较弱 | 金融、法律、手册类文档 |
| pypdfium2 | 开源 | 准确率高，保持词序好 | 内存消耗较大 | 结构化商业文档 |
| Unstructured | 开源 | 支持多格式（PDF/Word/HTML/PPT） | 复杂表格识别较弱 | ETL管道、通用文档处理 |
| MinerU | 开源 | 公式识别强，支持多语言OCR（84种） | 计算资源消耗大 | 学术论文、技术文档 |
| Marker | 开源 | PDF转Markdown优秀，OCR支持 | 复杂表格对齐问题 | 技术文档、博客内容 |
| Docling (IBM) | 开源 | 模块化设计，读取顺序保留好 | 相对较新，生态较小 | 企业文档管理 |
| LlamaParse | 商业 | AI驱动，表格提取准确 | 需付费，API限制 | 复杂结构文档 |
性能数据参考（基于DocLayNet数据集）：
- PyMuPDF在金融文档类别F1得分：0.98
- pypdfium在专利文档F1得分：0.969
- Nougat（深度学习方法）在科学文档上显著超越规则方法
3.2 PyMuPDF深度解析
PyMuPDF是最广泛使用的PDF解析库之一，其优势在于：
核心特性：
- 高性能：C++内核，Python绑定，处理速度是纯Python库的10倍以上
- 精确提取：保持文本原始结构和阅读顺序
- 丰富功能：支持文本、图片、表格、元数据提取
与RAG框架集成：
# LangChain集成
from langchain_community.document_loaders import PyMuPDFLoader
loader = PyMuPDFLoader("document.pdf")
documents = loader.load()
# LlamaIndex集成
from llama_index.readers.file import PyMuPDFReader
loader = PyMuPDFReader()
documents = loader.load(file_path="document.pdf")
PyMuPDF4LLM：专为LLM优化的封装：
import pymupdf4llm
# 转换为Markdown格式
md_text = pymupdf4llm.to_markdown("input.pdf")
# 支持分页chunk
data = pymupdf4llm.to_markdown("input.pdf", page_chunks=True)
# data[0]将包含第一页的文本和元数据
关键配置参数：
page_chunks=True：按页分块，便于后续chunkingwrite_images=True：提取并保存图片image_format="png"：图片格式选择dpi=300：图片分辨率
3.3 MinerU：AI驱动的高精度解析
MinerU是上海人工智能实验室开发的新一代文档解析工具，特别适合学术和技术文档。
技术架构：
MinerU的四阶段处理流程：
- 文档预处理：使用PyMuPDF读取PDF，提取元数据（语言类型、页面尺寸、可解析性）
- 内容解析：
- 版面分析：检测文本、标题、图片、表格、公式区域
- OCR识别：84种语言支持
- 公式识别：转换为LaTeX格式
- 表格识别：转换为HTML格式
- 后处理：去除页眉页脚、按位置排序、拼接内容
- 格式转换：生成Markdown或JSON
MinerU输出格式选择：Markdown vs JSON
Markdown格式（推荐用于RAG）：
# 文档标题
## 章节1
这是正文内容...
$$E = mc^2$$
| 列1 | 列2 |
|-----|-----|
| 数据1 | 数据2 |
适用场景：
- ✅ 直接用于LLM上下文输入
- ✅ 易于人类阅读和审核
- ✅ 保持文档层级结构
- ✅ 便于基于标题的语义分块
JSON格式（推荐用于二次开发）：
{
"pdf_info": {
"para_blocks": [
{
"type": "text",
"bbox": [x0, y0, x1, y1],
"content": "文本内容",
"page_idx": 0
},
{
"type": "table",
"bbox": [x0, y0, x1, y1],
"content": "<table>...</table>",
"page_idx": 0
}
]
}
}
适用场景：
- ✅ 需要精确位置信息
- ✅ 构建自定义索引结构
- ✅ 多模态RAG（图文混合）
- ✅ 需要保留完整元数据
性能对比（基于实际测试）：
- MinerU比规则方法在科学文档上准确率提升40%
- 公式识别准确率（UniMERNet）：95%+
- 处理速度（GPU加速）：提升50%+
3.4 Office文档处理策略
Word文档（DOCX）：
# 使用python-docx
from docx import Document
doc = Document("document.docx")
text = "\n".join([para.text for para in doc.paragraphs])
# 使用Unstructured
from unstructured.partition.docx import partition_docx
elements = partition_docx("document.docx")
Excel文档（XLSX）：
import pandas as pd
# 读取所有sheet
xlsx_file = pd.ExcelFile("data.xlsx")
for sheet_name in xlsx_file.sheet_names:
df = pd.read_excel(xlsx_file, sheet_name)
# 转换为文本或保留结构化数据
PowerPoint文档（PPTX）：
from pptx import Presentation
prs = Presentation("presentation.pptx")
for slide in prs.slides:
for shape in slide.shapes:
if hasattr(shape, "text"):
print(shape.text)
HTML文档：
from bs4 import BeautifulSoup
import requests
response = requests.get(url)
soup = BeautifulSoup(response.content, 'html.parser')
# 移除脚本和样式
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
四、文档分块：平衡语义完整性与检索效率
文档分块是RAG系统中最具挑战性的环节之一。分块质量直接影响检索精度和生成质量。
4.1 分块策略分类与原理
4.1.1 固定长度分块（Fixed-size Chunking）
原理：按照固定的字符数、词数或token数切分文本。
实现示例：
from langchain.text_splitters import CharacterTextSplitter
splitter = CharacterTextSplitter(
chunk_size=1000, # 每块1000字符
chunk_overlap=200, # 重叠200字符
separator="\n\n" # 优先按段落分割
)
chunks = splitter.split_text(text)
优势：
- 实现简单，计算开销低
- 块大小一致，便于批处理
- 适合大规模数据快速处理
劣势：
- 可能切断句子或完整概念
- 忽略文档结构和语义边界
- 在表格、代码等结构化内容上表现差
适用场景：快速原型、非结构化文本、资源受限环境
关键参数选择：
- Chunk size：建议250 tokens（约1000字符）作为起点
- Overlap：通常设置为chunk_size的10-20%
- Token计算：使用embedding模型的tokenizer（如tiktoken）
4.1.2 递归字符分块（Recursive Character Splitting）
原理：使用分隔符层次结构，从大到小递归分割。
实现示例：
from langchain.text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " ", ""] # 分隔符优先级
)
chunks = splitter.split_documents(documents)
分隔符优先级逻辑：
- 首先尝试用
\n\n（段落）分割 - 如果块仍然过大，用
\n（行）分割 - 继续用句号分割
- 最后用空格分割
- 仍然过大则强制按字符切分
优势：
- 保持文档自然结构
- 灵活处理不同类型文本
- 比固定长度更智能
劣势：
- 计算复杂度略高
- 仍可能破坏特殊结构（表格、代码）
适用场景：通用文档处理、博客文章、技术文档
4.1.3 基于文档结构的分块
Markdown文档：
from langchain.text_splitters import MarkdownTextSplitter
splitter = MarkdownTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = splitter.split_text(markdown_text)
特点：按照Markdown标题层级（#, ##, ###）分块，保持章节完整性。
HTML文档：
from langchain.text_splitters import HTMLHeaderTextSplitter
headers_to_split_on = [
("h1", "Header 1"),
("h2", "Header 2"),
("h3", "Header 3"),
]
splitter = HTMLHeaderTextSplitter(headers_to_split_on)
chunks = splitter.split_text(html_text)
代码文档：
from langchain.text_splitters import RecursiveCharacterTextSplitter, Language
# Python代码专用splitter
python_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON,
chunk_size=500,
chunk_overlap=50
)
chunks = python_splitter.split_text(python_code)
代码分块的AST方法：
import ast
def chunk_python_by_ast(code):
tree = ast.parse(code)
chunks = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
# 提取完整函数或类定义
chunk = ast.get_source_segment(code, node)
chunks.append({
'type': type(node).__name__,
'name': node.name,
'code': chunk
})
return chunks
4.1.4 语义分块（Semantic Chunking）
原理：使用embedding模型计算句子间语义相似度，当相似度下降超过阈值时分块。
实现示例：
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
splitter = SemanticChunker(
embeddings,
breakpoint_threshold_type="percentile", # 或"standard_deviation"
breakpoint_threshold_amount=75
)
chunks = splitter.split_text(text)
算法流程：
- 将文本分割为句子
- 计算相邻句子的embedding
- 计算句子对之间的余弦相似度
- 当相似度低于阈值时插入分块边界
- 合并相邻的高相似度句子
LlamaIndex实现：
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding
embed_model = OpenAIEmbedding()
splitter = SemanticSplitterNodeParser(
buffer_size=1,
breakpoint_percentile_threshold=95,
embed_model=embed_model
)
nodes = splitter.get_nodes_from_documents(documents)
关键参数：
buffer_size：决策时考虑的上下文窗口（句子数）breakpoint_percentile_threshold：相似度阈值（百分位）- 高阈值（95%）：只在主题明显变化时分块，产生更大的块
- 低阈值（75%）：对细微变化敏感，产生更多更小的块
优势：
- 保持语义完整性
- 自适应内容变化
- 检索相关性高
劣势：
- 计算开销大（需要embedding）
- 处理速度慢
- 块大小不固定，可能超出token限制
适用场景：高质量要求、技术文档、学术论文
4.1.5 滑动窗口分块（Sliding Window）
原理：固定大小的窗口在文本上滑动，相邻块之间有重叠。
实现示例：
def sliding_window_chunk(text, window_size=200, overlap=50):
words = text.split()
chunks = []
step = window_size - overlap
for i in range(0, len(words), step):
chunk = " ".join(words[i:i+window_size])
chunks.append(chunk)
return chunks
优势：
- 确保上下文连续性
- 减少边界信息丢失
- 提高召回率
劣势：
- 存储冗余
- 可能检索到重复内容
- 需要去重策略
重叠率建议：10-20%的窗口大小
4.2 特殊内容的分块策略
4.2.1 表格处理：保持结构完整性
挑战：
- 行列关系断裂
- 列名丢失
- 跨页表格
解决方案1：表格完整保留
def preserve_tables(text):
# 检测表格区域（使用正则或库）
tables = extract_tables(text)
for table in tables:
# 将整个表格作为一个chunk
# 添加上下文信息
chunk = {
'type': 'table',
'content': table.to_html(), # 或to_markdown()
'caption': table.caption,
'headers': table.headers
}
解决方案2：表格文本化
def tabularize_table(df):
"""将表格转换为描述性文本"""
descriptions = []
for _, row in df.iterrows():
desc = f"{row['Column1']} 的 {row['Column2']} 是 {row['Value']}"
descriptions.append(desc)
return "\n".join(descriptions)
解决方案3：专用表格检索器
from langchain.retrievers import MultiVectorRetriever
# 为表格生成摘要
table_summaries = llm.generate_summaries(tables)
# 存储原始表格，检索时使用摘要
retriever = MultiVectorRetriever(
vectorstore=vectorstore,
docstore=docstore,
id_key="doc_id"
)
基于页面的分块策略：
许多文档表格设计时会遵循页面边界，因此按页分块可以有效保留表格：
import pymupdf
def chunk_by_page(pdf_path):
doc = pymupdf.open(pdf_path)
chunks = []
for page_num, page in enumerate(doc):
text = page.get_text()
tables = page.find_tables()
chunk = {
'page': page_num,
'text': text,
'tables': [t.extract() for t in tables]
}
chunks.append(chunk)
return chunks
4.2.2 代码块处理
挑战：
- 语法结构被破坏
- 依赖关系丢失
- 缩进和格式被打乱
解决方案：语法感知分块
def chunk_code_by_functions(code, language='python'):
if language == 'python':
import ast
tree = ast.parse(code)
chunks = []
for node in ast.iter_child_nodes(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
source = ast.get_source_segment(code, node)
chunks.append({
'type': 'function' if isinstance(node, ast.FunctionDef) else 'class',
'name': node.name,
'code': source,
'lineno': node.lineno
})
return chunks
保留依赖关系：
def add_imports_to_chunk(code_chunk, full_code):
"""为代码片段添加必要的import"""
import_lines = [line for line in full_code.split('\n')
if line.strip().startswith('import') or
line.strip().startswith('from')]
return '\n'.join(import_lines) + '\n\n' + code_chunk
4.2.3 跨段引用和上下文链
问题：分块后，引用关系断裂
Chunk 1: "根据上述研究..." # 上述研究在哪？
Chunk 2: "该方法的优势在于..." # 该方法是什么？
解决方案1：上下文增强
def add_context_to_chunk(chunks, context_window=1):
"""为每个chunk添加前后文信息"""
enhanced_chunks = []
for i, chunk in enumerate(chunks):
context = []
# 添加前文
if i > 0:
context.append(f"[前文]: {chunks[i-1][-200:]}") # 取前文末尾200字符
# 主体内容
context.append(f"[主体]: {chunk}")
# 添加后文
if i < len(chunks) - 1:
context.append(f"[后文]: {chunks[i+1][:200]}") # 取后文开头200字符
enhanced_chunks.append("\n".join(context))
return enhanced_chunks
解决方案2：层次化chunk索引
class HierarchicalChunk:
def __init__(self, content, parent=None, children=None):
self.content = content
self.parent = parent # 父chunk（更大粒度）
self.children = children or [] # 子chunk（更小粒度）
self.metadata = {}
# 检索时可以同时返回父子context
解决方案3：元数据丰富
chunk_with_metadata = {
'content': chunk_text,
'metadata': {
'document': 'research_paper.pdf',
'section': '3.2 Methodology',
'page': 5,
'previous_heading': 'Experimental Setup',
'next_heading': 'Results'
}
}
4.3 重叠窗口与去重策略
4.3.1 重叠的必要性与权衡
为什么需要重叠：
- 减少边界信息丢失：重要信息可能恰好在chunk边界
- 提供上下文连续性：帮助模型理解完整语境
- 提高召回率：同一信息出现在多个chunk中
重叠带来的问题：
- 存储冗余：相同内容被多次存储
- 检索冗余：可能返回高度相似的多个chunk
- 答案冲突：不同chunk可能导致不一致的回答
重叠率建议：
- 固定长度分块：10-20%
- 语义分块：可以更小或不需要
- 技术文档：15-20%（保证上下文）
- 简单FAQ：5-10%（减少冗余）
4.3.2 去重策略实现
策略1：基于哈希的精确去重
import hashlib
def deduplicate_by_hash(chunks):
seen_hashes = set()
unique_chunks = []
for chunk in chunks:
# 标准化文本（去除空白、转小写）
normalized = ' '.join(chunk.lower().split())
chunk_hash = hashlib.sha256(normalized.encode()).hexdigest()
if chunk_hash not in seen_hashes:
seen_hashes.add(chunk_hash)
unique_chunks.append(chunk)
return unique_chunks
策略2：基于相似度的模糊去重
from sentence_transformers import SentenceTransformer, util
def deduplicate_by_similarity(chunks, threshold=0.85, model_name='all-MiniLM-L6-v2'):
model = SentenceTransformer(model_name)
embeddings = model.encode(chunks)
unique_chunks = []
unique_embeddings = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
is_duplicate = False
for unique_emb in unique_embeddings:
similarity = util.cos_sim(embedding, unique_emb).item()
if similarity > threshold:
is_duplicate = True
break
if not is_duplicate:
unique_chunks.append(chunk)
unique_embeddings.append(embedding)
return unique_chunks
策略3：Locality-Sensitive Hashing (LSH)
from datasketch import MinHash, MinHashLSH
def deduplicate_by_lsh(chunks, threshold=0.8):
lsh = MinHashLSH(threshold=threshold, num_perm=128)
minhashes = {}
for i, chunk in enumerate(chunks):
# 创建MinHash
m = MinHash(num_perm=128)
for word in chunk.split():
m.update(word.encode('utf8'))
# 检查是否有相似chunk
result = lsh.query(m)
if not result:
lsh.insert(f"chunk_{i}", m)
minhashes[f"chunk_{i}"] = chunk
return list(minhashes.values())
策略4：检索时去重
def diversify_retrieval_results(retrieved_chunks, threshold=0.9):
"""在检索结果中移除高度相似的chunk"""
model = SentenceTransformer('all-MiniLM-L6-v2')
if not retrieved_chunks:
return retrieved_chunks
# 计算embeddings
embeddings = model.encode([c['content'] for c in retrieved_chunks])
diversified = [retrieved_chunks[0]]
diversified_embeddings = [embeddings[0]]
for i in range(1, len(retrieved_chunks)):
current_emb = embeddings[i]
is_similar = False
for prev_emb in diversified_embeddings:
if util.cos_sim(current_emb, prev_emb).item() > threshold:
is_similar = True
break
if not is_similar:
diversified.append(retrieved_chunks[i])
diversified_embeddings.append(current_emb)
return diversified
策略5：答案冲突解决
当多个相似chunk给出不同答案时：
def resolve_conflicts(answers, chunks, query):
"""使用LLM解决答案冲突"""
# 方法1：选择最相关的chunk
relevance_scores = compute_relevance(chunks, query)
best_answer = answers[relevance_scores.index(max(relevance_scores))]
# 方法2：使用LLM综合多个答案
prompt = f"""
Query: {query}
我检索到了多个相关段落，它们给出的答案略有不同：
{format_answers(answers)}
请综合这些信息，给出一个准确的答案。如果答案确实存在矛盾，请指出。
"""
unified_answer = llm.generate(prompt)
return unified_answer
五、向量化：将文本转换为数学表示
向量化是连接文本空间和检索空间的桥梁。高质量的向量表示是RAG系统检索准确性的关键。
5.1 Embedding模型原理
什么是Embedding：
Embedding是将文本映射到高维向量空间的过程，使得语义相似的文本在向量空间中距离更近。
文本: "Paris is the capital of France"
↓ Embedding Model
向量: [0.23, -0.45, 0.67, ..., 0.12] # 通常768或1536维
核心思想：
- 维度（Dimension）：向量的长度，如768D、1536D
- 语义空间：相似文本聚集，不同文本分离
- 距离度量：余弦相似度、欧氏距离、内积
Embedding模型架构演进：
- Word2Vec时代（2013）
- CBOW、Skip-gram
- 词级别embedding
- 无法处理上下文
- BERT时代（2018）
- 上下文感知
- 双向transformer
- [CLS] token作为句子表示
- Sentence-BERT（2019）
- 专门为句子级别任务优化
- 孪生网络结构
- 对比学习
- 当前SOTA模型（2024-2025）
- BGE系列（BAAI）
- E5系列（Microsoft）
- Instructor-XL
- Text-Embedding-3（OpenAI）
5.2 主流Embedding模型对比
5.2.1 基于MTEB排行榜的评估
Massive Text Embedding Benchmark (MTEB) 是评估embedding模型的标准基准，包含8类任务、58个数据集。
2025年MTEB Top模型（Retrieval任务）：
| 模型 | 开发者 | 维度 | MTEB平均分 | Retrieval分数 | 类型 |
|---|---|---|---|---|---|
| gte-Qwen2-7B-instruct | Alibaba | 3584 | 72.05 | 63.03 | 开源 |
| SFR-Embedding-Mistral | Salesforce | 4096 | 71.24 | 59.00 | 开源 |
| voyage-multilingual-2 | Voyage AI | 1024 | 68.77 | 60.15 | 商业 |
| text-embedding-3-large | OpenAI | 3072 | 64.59 | 55.44 | 商业 |
| bge-large-en-v1.5 | BAAI | 1024 | 64.23 | 54.29 | 开源 |
| bge-m3 | BAAI | 1024 | 63.48 | 53.72 | 开源 |
| e5-mistral-7b-instruct | Microsoft | 4096 | 61.89 | 56.63 | 开源 |
| text-embedding-ada-002 | OpenAI | 1536 | 60.99 | 49.25 | 商业 |
关键发现：
- BGE模型表现优异：在多个实际应用中，BGE系列在特定领域常超过排行榜更高的模型
- 模型大小权衡：7B模型（如E5-Mistral）虽然性能更好，但推理速度慢10倍以上
- OpenAI Ada-002已过时：虽然使用广泛，但性能已被开源模型超越
5.2.2 开源模型深度解析
BGE (BAAI General Embedding)：
from sentence_transformers import SentenceTransformer
# BGE模型使用
model = SentenceTransformer('BAAI/bge-large-en-v1.5')
# 编码文档
documents = ["Document 1 text", "Document 2 text"]
doc_embeddings = model.encode(documents, normalize_embeddings=True)
# 编码查询（需要加前缀）
queries = ["What is RAG?"]
query_embeddings = model.encode(
[f"Represent this sentence for searching relevant passages: {q}" for q in queries],
normalize_embeddings=True
)
# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity(query_embeddings, doc_embeddings)
BGE特点：
- 支持多种语言（en, zh, multilingual）
- 需要查询指令前缀
- 性能/成本比优秀
- 支持Matryoshka表示学习（可调整维度）
E5 (Text Embeddings by Weakly-Supervised Contrastive Pre-training)：
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('intfloat/e5-large-v2')
# E5也需要前缀
docs = [f"passage: {doc}" for doc in documents]
queries = [f"query: {q}" for q in queries]
doc_embeddings = model.encode(docs)
query_embeddings = model.encode(queries)
E5特点：
- Microsoft开源
- 训练数据来自弱监督对比学习
- 多种尺寸（small/base/large）
- E5-Mistral-7B达到SOTA水平
Instructor-XL：
from InstructorEmbedding import INSTRUCTOR
model = INSTRUCTOR('hkunlp/instructor-xl')
# 可以自定义指令
instruction = "Represent the scientific document for retrieval:"
embeddings = model.encode([[instruction, doc] for doc in documents])
Instructor特点：
- 可自定义任务指令
- 适应性强（同一模型适配不同任务）
- 模型较大（4.96GB）
5.2.3 商业模型API调用
OpenAI Embeddings：
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
response = client.embeddings.create(
model="text-embedding-3-large",
input=documents,
dimensions=1024 # 可选：支持降维
)
embeddings = [item.embedding for item in response.data]
定价（截至2025）：
- text-embedding-3-large: $0.00013 / 1K tokens
- text-embedding-3-small: $0.00002 / 1K tokens
- text-embedding-ada-002: $0.0001 / 1K tokens
Cohere Embeddings：
import cohere
co = cohere.Client('your-api-key')
response = co.embed(
texts=documents,
model='embed-multilingual-v3.0',
input_type='search_document' # 或'search_query'
)
embeddings = response.embeddings
Cohere特点：
- 支持多语言（100+）
- 区分文档和查询类型
- embed-v3支持压缩和multi-vector
Voyage AI：
import voyageai
vo = voyageai.Client(api_key="your-api-key")
embeddings = vo.embed(
documents,
model="voyage-2",
input_type="document" # 或"query"
)
5.3 Embedding模型选型决策树
选型考虑因素：
- 领域适配性：
- 通用领域：BGE、E5、OpenAI
- 科研文献：Instructor、SciBERT
- 医疗领域：BioBERT、PubMedBERT
- 法律文本：Legal-BERT
- 代码：CodeBERT、GraphCodeBERT
- 语言支持：
- 单语言（英文）：BGE-large-en、E5-large
- 多语言：BGE-m3、multilingual-e5、Cohere
- 中文优化：BGE-large-zh、text2vec-base-chinese
- 性能需求：
- 低延迟：小模型（all-MiniLM, BGE-small）
- 高准确率：大模型（E5-Mistral-7B, Instructor-XL）
- 平衡选择：BGE-large（1024维）
- 成本考虑：
- 自部署：开源模型（BGE, E5）
- 初期成本：GPU硬件（T4/A10/A100）
- 运行成本：电费、维护
- API调用：按token计费
- 无硬件成本
- 按需扩展
- 数据隐私需考虑
- 自部署：开源模型（BGE, E5）
5.4 Embedding模型评估与微调
5.4.1 领域评估
构建领域评估数据集：
# 评估数据格式
eval_dataset = [
{
'query': 'How to implement RAG?',
'positive': 'RAG implementation involves...', # 相关文档
'negative': 'Machine learning is...' # 不相关文档
},
# 更多样本...
]
def evaluate_embedding_model(model, eval_dataset):
from sklearn.metrics import roc_auc_score
scores = []
labels = []
for item in eval_dataset:
query_emb = model.encode(item['query'])
pos_emb = model.encode(item['positive'])
neg_emb = model.encode(item['negative'])
pos_score = cosine_similarity([query_emb], [pos_emb])[0][0]
neg_score = cosine_similarity([query_emb], [neg_emb])[0][0]
scores.extend([pos_score, neg_score])
labels.extend([1, 0])
auc = roc_auc_score(labels, scores)
return auc
检索质量指标：
def compute_retrieval_metrics(query_embeddings, doc_embeddings, relevance_labels, k=10):
"""
relevance_labels: 二值矩阵，[i,j]=1表示query i与doc j相关
"""
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity(query_embeddings, doc_embeddings)
# NDCG@k
from sklearn.metrics import ndcg_score
ndcg = ndcg_score(relevance_labels, similarities, k=k)
# MRR (Mean Reciprocal Rank)
mrr = 0
for i, sim_scores in enumerate(similarities):
# 找到第一个相关文档的排名
ranked_indices = np.argsort(sim_scores)[::-1]
for rank, idx in enumerate(ranked_indices, 1):
if relevance_labels[i, idx] == 1:
mrr += 1 / rank
break
mrr /= len(similarities)
# Recall@k
recall_at_k = 0
for i, sim_scores in enumerate(similarities):
top_k_indices = np.argsort(sim_scores)[::-1][:k]
relevant_retrieved = relevance_labels[i, top_k_indices].sum()
total_relevant = relevance_labels[i].sum()
recall_at_k += relevant_retrieved / total_relevant if total_relevant > 0 else 0
recall_at_k /= len(similarities)
return {
'ndcg@10': ndcg,
'mrr': mrr,
'recall@10': recall_at_k
}
5.4.2 Embedding模型微调
使用Sentence-Transformers微调：
from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
# 准备训练数据
train_examples = [
InputExample(texts=['query', 'positive_doc'], label=1.0),
InputExample(texts=['query', 'negative_doc'], label=0.0),
]
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
# 加载预训练模型
model = SentenceTransformer('BAAI/bge-large-en-v1.5')
# 定义损失函数
train_loss = losses.CosineSimilarityLoss(model)
# 微调
model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=3,
warmup_steps=100,
output_path='./fine_tuned_model'
)
使用对比学习（Contrastive Learning）：
from sentence_transformers import losses
# MultipleNegativesRankingLoss
# 一个query对应一个正样本和多个负样本
train_loss = losses.MultipleNegativesRankingLoss(model)
# TripletLoss
# (anchor, positive, negative)三元组
train_loss = losses.TripletLoss(
model,
triplet_margin=0.5
)
领域适配最佳实践：
- 收集500-1000个领域相关的(query, document)对
- 使用预训练模型初始化
- 小学习率微调（1e-5到5e-5）
- 在验证集上早停
- 评估检索性能提升
5.5 构建向量索引
基本流程：
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
# 1. 加载embedding模型
model = SentenceTransformer('BAAI/bge-large-en-v1.5')
# 2. 对文档进行向量化
documents = ["Doc 1", "Doc 2", ...]
embeddings = model.encode(documents, show_progress_bar=True)
# 3. 标准化（推荐）
embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
# 4. 构建FAISS索引
dimension = embeddings.shape[1] # 如1024
index = faiss.IndexFlatIP(dimension) # 内积（余弦相似度）
index.add(embeddings)
# 5. 保存索引
faiss.write_index(index, "vector_index.faiss")
# 6. 保存文档映射
import pickle
with open("documents.pkl", "wb") as f:
pickle.dump(documents, f)
批量处理大规模数据：
def batch_encode_documents(documents, model, batch_size=128):
"""批量编码文档以节省内存"""
all_embeddings = []
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
batch_embeddings = model.encode(
batch,
batch_size=batch_size,
show_progress_bar=True,
convert_to_numpy=True
)
all_embeddings.append(batch_embeddings)
return np.vstack(all_embeddings)
增量索引更新：
class IncrementalIndex:
def __init__(self, model, index_path="vector_index.faiss"):
self.model = model
self.index_path = index_path
self.index = faiss.read_index(index_path) if os.path.exists(index_path) else None
self.documents = []
def add_documents(self, new_documents):
"""增量添加新文档"""
new_embeddings = self.model.encode(new_documents)
if self.index is None:
dimension = new_embeddings.shape[1]
self.index = faiss.IndexFlatIP(dimension)
self.index.add(new_embeddings)
self.documents.extend(new_documents)
# 保存更新
self.save()
def save(self):
faiss.write_index(self.index, self.index_path)
with open("documents.pkl", "wb") as f:
pickle.dump(self.documents, f)
多语言embedding处理：
# 使用多语言模型
model = SentenceTransformer('BAAI/bge-m3')
# 混合语言文档
multilingual_docs = [
"This is English text",
"这是中文文本",
"C'est du texte français"
]
embeddings = model.encode(multilingual_docs)
# 同一向量空间中，可以跨语言检索
六、向量数据库选型
向量数据库是RAG系统的存储和检索引擎，其选择直接影响系统的性能、可扩展性和维护成本。
6.1 向量数据库 vs 向量库
关键区别：
| 特性 | 向量库（如FAISS） | 向量数据库（如Milvus） |
|---|---|---|
| CRUD操作 | ❌ 难以修改索引数据 | ✅ 完整的增删改查 |
| 持久化 | 需手动序列化 | 内置持久化机制 |
| 分布式 | ❌ 单机运行 | ✅ 支持分布式部署 |
| 并发 | 有限 | 高并发支持 |
| 过滤 | 基础 | 丰富的元数据过滤 |
| 监控 | 无 | 完善的监控和日志 |
| 适用场景 | 研究原型、小规模 | 生产环境、大规模 |
6.2 主流向量数据库对比
6.2.1 开源方案
Milvus
架构特点：
- 云原生设计，存储计算分离
- 支持多种索引类型（HNSW, IVF, DiskANN）
- GPU加速支持
- Kubernetes原生
适用场景：
- 10亿级别向量规模
- 需要高可用和容错
- 有Kubernetes基础设施
- 团队有运维能力
性能数据：
- 吞吐量：Recall=0.9时最高（相比Weaviate、Qdrant）
- 索引构建：略慢于Weaviate，但索引更大（1.2M向量，100维约1.5GB）
- 查询延迟：P95约20-40ms
示例代码：
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
# 连接Milvus
connections.connect("default", host="localhost", port="19530")
# 定义schema
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535)
]
schema = CollectionSchema(fields, "RAG collection")
# 创建collection
collection = Collection("rag_docs", schema)
# 创建HNSW索引
index_params = {
"metric_type": "IP", # 内积
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)
# 插入数据
collection.insert([embeddings, texts])
# 搜索
collection.load()
results = collection.search(
data=[query_embedding],
anns_field="embedding",
param={"metric_type": "IP", "params": {"ef": 50}},
limit=10
)
Weaviate
架构特点：
- GraphQL API
- 内置向量化模块（可选）
- 混合搜索（向量+关键词BM25）
- 知识图谱能力
适用场景：
- 需要向量+结构化数据结合
- GraphQL生态
- 快速原型到生产
- 需要混合搜索
性能数据：
- 索引构建：最快
- 索引大小：最小
- 吞吐量：中等（Recall>0.95时追上Milvus）
示例代码：
import weaviate
client = weaviate.Client("http://localhost:8080")
# 定义schema
schema = {
"classes": [{
"class": "Document",
"vectorizer": "none", # 使用外部embedding
"properties": [
{"name": "content", "dataType": ["text"]},
{"name": "source", "dataType": ["string"]}
]
}]
}
client.schema.create(schema)
# 插入数据（批量）
with client.batch as batch:
for doc, embedding in zip(documents, embeddings):
batch.add_data_object(
{"content": doc, "source": "file.pdf"},
"Document",
vector=embedding
)
# 混合搜索
result = client.query.get("Document", ["content", "source"]) \
.with_hybrid(query="What is RAG?", alpha=0.5) \
.with_limit(10) \
.do()
Qdrant
架构特点：
- Rust编写，性能优异
- 丰富的过滤能力
- Payload存储和搜索
- 轻量级部署
适用场景：
- 需要复杂元数据过滤
- 成本敏感项目
- 边缘部署
- 中小规模（<1亿向量）
性能数据：
- 查询延迟：低（Rust性能优势）
- 内存占用：紧凑
- 吞吐量：与Milvus、Weaviate相当
示例代码：
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
client = QdrantClient("localhost", port=6333)
# 创建collection
client.create_collection(
collection_name="documents",
vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
)
# 插入数据
points = [
PointStruct(
id=i,
vector=embedding.tolist(),
payload={"text": doc, "source": "file.pdf"}
)
for i, (doc, embedding) in enumerate(zip(documents, embeddings))
]
client.upsert(collection_name="documents", points=points)
# 搜索（带过滤）
results = client.search(
collection_name="documents",
query_vector=query_embedding,
limit=10,
query_filter={
"must": [
{"key": "source", "match": {"value": "file.pdf"}}
]
}
)
Chroma
架构特点：
- 开发者友好
- 轻量级（SQLite后端）
- LangChain/LlamaIndex深度集成
- 适合原型开发
适用场景：
- 快速原型验证
- 小规模应用（<100K向量）
- 本地开发环境
- 教学演示
局限性：
- 可扩展性有限（SQLite限制）
- 性能基准数据较少
- 不适合生产级大规模部署
示例代码：
import chromadb
client = chromadb.Client()
# 创建collection
collection = client.create_collection("documents")
# 添加数据
collection.add(
embeddings=embeddings.tolist(),
documents=documents,
metadatas=[{"source": "file.pdf"}] * len(documents),
ids=[str(i) for i in range(len(documents))]
)
# 查询
results = collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=10
)
6.2.2 商业方案
Pinecone
特点：
- 完全托管（Serverless）
- 自动扩缩容
- 多区域部署
- 高可用SLA
定价模型：
- Serverless：按请求和存储计费
- Pod-based：固定资源包月付费
适用场景：
- 不想管理基础设施
- 需要全球分布
- SLA要求严格
- 快速上线
示例代码：
import pinecone
pinecone.init(api_key="your-api-key")
# 创建索引
pinecone.create_index(
name="rag-index",
dimension=1024,
metric="cosine"
)
index = pinecone.Index("rag-index")
# 批量upsert
index.upsert(vectors=[
(str(i), embedding.tolist(), {"text": doc})
for i, (doc, embedding) in enumerate(zip(documents, embeddings))
])
# 查询
results = index.query(
vector=query_embedding.tolist(),
top_k=10,
include_metadata=True
)
Zilliz Cloud（Milvus托管版）
特点：
- 基于Milvus
- 完全托管
- 更低成本（相比Pinecone）
- 多云支持
MongoDB Atlas Vector Search
特点：
- 集成在MongoDB中
- 向量+文档数据库统一
- 成熟的生态系统
- 适合已有MongoDB用户
6.2.3 SQL数据库扩展
pgvector (PostgreSQL)
特点：
- PostgreSQL扩展
- SQL生态完整
- 事务支持
- 适合混合查询
局限性：
- 百万级向量后性能下降
- 不如专用向量数据库优化
示例代码：
-- 安装扩展
CREATE EXTENSION vector;
-- 创建表
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
content TEXT,
embedding vector(1024)
);
-- 创建HNSW索引
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
-- 插入数据
INSERT INTO documents (content, embedding) VALUES
('Document text', '[0.1, 0.2, ...]');
-- 查询
SELECT content, embedding <=> '[query_embedding]' AS distance
FROM documents
ORDER BY embedding <=> '[query_embedding]'
LIMIT 10;
6.3 向量数据库选型决策矩阵
| 需求场景 | 推荐方案 | 理由 |
|---|---|---|
| 快速原型 | Chroma | 易用、轻量、LangChain集成 |
| 中小规模生产（<10M） | Qdrant或Weaviate | 性能好、易部署、功能完整 |
| 大规模生产（>10M） | Milvus或Pinecone | 分布式、高性能、可扩展 |
| 已有PostgreSQL | pgvector | 无需额外基础设施 |
| 无运维能力 | Pinecone或Zilliz Cloud | 完全托管 |
| 混合搜索需求 | Weaviate或Elasticsearch | BM25+向量原生支持 |
| 成本敏感 | 开源方案（Milvus/Qdrant） | 自部署，仅硬件成本 |
| 多租户隔离 | Pinecone或Weaviate | 命名空间支持好 |
6.4 实际部署建议
小规模部署（<100K向量）：
# docker-compose.yml for Qdrant
version: '3.8'
services:
qdrant:
image: qdrant/qdrant:latest
ports:
- "6333:6333"
volumes:
- ./qdrant_storage:/qdrant/storage
中等规模部署（100K-10M向量）：
- 使用Kubernetes部署Milvus或Weaviate
- 配置持久卷（PVC）
- 设置副本和负载均衡
- 监控（Prometheus + Grafana）
大规模部署（>10M向量）：
- Milvus分布式集群
- 分片和副本策略
- 混合存储（内存+SSD+对象存储）
- 读写分离
- 缓存层（Redis）
混合部署策略：
# 小数据用Chroma原型，大规模迁移到Milvus
class VectorStoreAdapter:
def __init__(self, store_type='chroma'):
if store_type == 'chroma':
self.store = ChromaStore()
elif store_type == 'milvus':
self.store = MilvusStore()
def search(self, query_vector, top_k=10):
return self.store.search(query_vector, top_k)
# 迁移脚本
def migrate_chroma_to_milvus(chroma_collection, milvus_collection):
"""从Chroma迁移到Milvus"""
# 从Chroma导出
data = chroma_collection.get()
# 导入Milvus
milvus_collection.insert([
data['embeddings'],
data['documents'],
data['metadatas']
])
七、索引结构：从原理到实践
索引结构决定了向量检索的速度和准确性。理解不同索引算法的原理对于优化RAG系统至关重要。
7.1 索引算法分类
近似最近邻搜索（ANN）算法分类：
主流算法对比：
| 算法 | 类型 | 召回率 | 速度 | 内存 | 适用场景 |
|---|---|---|---|---|---|
| Flat | 精确搜索 | 100% | 慢 | 高 | 小规模/<10K |
| IVF | 量化+倒排 | 90-95% | 快 | 中 | 通用，百万级 |
| HNSW | 图 | 95-99% | 极快 | 高 | 高召回要求 |
| LSH | 哈希 | 70-85% | 快 | 低 | 大规模，低精度可容忍 |
| PQ | 量化 | 80-90% | 快 | 低 | 内存受限 |
| IVF-PQ | 混合 | 85-95% | 快 | 低 | 平衡方案 |
7.2 HNSW (Hierarchical Navigable Small World)
7.2.1 算法原理
HNSW是目前性能最优的ANN算法之一，核心思想是构建一个多层的proximity graph。
NSW（Navigable Small World）基础：
- 每个节点连接到M个最近邻
- 使用贪心算法搜索：从入口节点开始，每次移动到更接近目标的邻居
- "小世界"特性：任意两节点间平均路径长度短
层次化（Hierarchical）：
Layer 2: o--------o (稀疏，长距离跳转)
| |
Layer 1: o--o--o--o--o (中等密度)
| | | | |
Layer 0: o-o-o-o-o-o-o-o (密集，精确搜索)
- 顶层：稀疏，快速定位大致区域
- 底层：密集，精确找到最近邻
- 搜索时从顶层开始，逐层向下
插入算法：
def insert_element(element, M, ef_construction):
"""
M: 每层最大连接数
ef_construction: 构建时的搜索宽度
"""
# 1. 随机确定元素的层数
level = random_level() # 指数分布
# 2. 从顶层搜索到该元素应在的层
entry_point = top_level_entry_point
for lc in range(top_level, level, -1):
# 在当前层找最近邻
nearest = search_layer(element, entry_point, ef=1, layer=lc)
entry_point = nearest
# 3. 在目标层及以下各层插入并建立连接
for lc in range(level, -1, -1):
# 找ef_construction个候选
candidates = search_layer(element, entry_point, ef=ef_construction, layer=lc)
# 选择M个最佳邻居建立双向连接
neighbors = select_neighbors(candidates, M)
add_connections(element, neighbors, layer=lc)
# 向下搜索
entry_point = neighbors[0]
搜索算法：
def search(query, K, ef):
"""
K: 返回的最近邻数量
ef: 搜索时的候选集大小（ef >= K）
"""
# 1. 从顶层开始
entry_point = top_level_entry_point
# 2. 贪心搜索到底层
for layer in range(top_level, 0, -1):
entry_point = search_layer(query, entry_point, ef=1, layer=layer)
# 3. 在底层精确搜索
candidates = search_layer(query, entry_point, ef=ef, layer=0)
return top_K(candidates, K)
def search_layer(query, entry_points, ef, layer):
visited = set()
candidates = priority_queue() # 最大堆
results = priority_queue() # 最小堆
for point in entry_points:
dist = distance(query, point)
candidates.push((-dist, point))
results.push((dist, point))
visited.add(point)
while candidates:
current_dist, current = candidates.pop()
if -current_dist > results.top()[0]:
break
# 遍历邻居
for neighbor in get_neighbors(current, layer):
if neighbor not in visited:
visited.add(neighbor)
dist = distance(query, neighbor)
if dist < results.top()[0] or len(results) < ef:
candidates.push((-dist, neighbor))
results.push((dist, neighbor))
if len(results) > ef:
results.pop()
return results
7.2.2 FAISS中的HNSW实现
import faiss
import numpy as np
# 创建HNSW索引
d = 1024 # 维度
M = 32 # 每层连接数（推荐16-64）
index = faiss.IndexHNSWFlat(d, M)
# 设置构建参数
index.hnsw.efConstruction = 200 # 构建时的搜索宽度
# 添加数据
index.add(embeddings)
# 设置搜索参数
index.hnsw.efSearch = 50 # 搜索时的宽度（影响召回率和速度）
# 搜索
D, I = index.search(query_embeddings, k=10)
参数调优：
| 参数 | 影响 | 推荐值 |
|---|---|---|
| M | 连接数，越大召回率越高，内存越大 | 16-32（通用），64（高召回） |
| efConstruction | 构建质量，越大索引质量越好 | 200-400 |
| efSearch | 搜索质量，越大召回率越高 | 50-100（平衡），200+（高召回） |
内存估算：
- HNSW内存 ≈ N × d × 4 bytes（向量）+ N × M × 4 bytes（连接）
- 1M个1024维向量，M=32：约4GB（向量）+ 128MB（图结构）= 4.1GB
7.2.3 HNSW优势与局限
优势：
- ✅ 召回率高（95-99%）
- ✅ 查询速度快（对数复杂度）
- ✅ 数据无关（性能不依赖数据分布）
- ✅ 支持动态插入
局限：
- ❌ 内存占用大
- ❌ 构建时间长
- ❌ 不支持GPU加速（大部分实现）
优化策略：
- 结合PQ压缩：
# HNSW + PQ
m = 8 # 子向量数
bits = 8 # 每个子向量的bit数
index = faiss.IndexHNSWPQ(d, M, m, bits)
- 使用IVF预过滤：
# IVF + HNSW
quantizer = faiss.IndexHNSWFlat(d, 32)
index = faiss.IndexIVFFlat(quantizer, d, nlist=1000)
7.3 IVF (Inverted File Index)
7.3.1 算法原理
IVF通过聚类将向量空间划分为多个区域（Voronoi cells），查询时只搜索最近的几个区域。
构建过程：
def build_ivf_index(vectors, nlist=100):
"""
nlist: 聚类中心数量
"""
# 1. 使用K-means聚类
centroids = kmeans(vectors, k=nlist)
# 2. 为每个向量分配到最近的聚类中心
inverted_lists = [[] for _ in range(nlist)]
for i, vector in enumerate(vectors):
nearest_centroid = argmin([distance(vector, c) for c in centroids])
inverted_lists[nearest_centroid].append(i)
return centroids, inverted_lists
搜索过程：
def search_ivf(query, centroids, inverted_lists, nprobe=10, k=10):
"""
nprobe: 搜索的聚类数量
"""
# 1. 找到nprobe个最近的聚类中心
centroid_distances = [distance(query, c) for c in centroids]
nearest_clusters = argsort(centroid_distances)[:nprobe]
# 2. 在这些聚类中的向量中搜索
candidates = []
for cluster_id in nearest_clusters:
for vector_id in inverted_lists[cluster_id]:
candidates.append(vector_id)
# 3. 精确计算距离并返回top-k
distances = [distance(query, vectors[vid]) for vid in candidates]
top_k_indices = argsort(distances)[:k]
return [candidates[i] for i in top_k_indices]
Voronoi图示例：
C1 •
/ | \
/ | \
v1 v2 v3
C2 •
/ | \
v4 v5 v6
Query → 先找最近的C1和C2 → 只在v1-v6中搜索
7.3.2 FAISS中的IVF实现
import faiss
d = 1024
nlist = 1000 # 聚类数量
# 训练聚类中心
quantizer = faiss.IndexFlatL2(d)
index = faiss.IndexIVFFlat(quantizer, d, nlist)
# 训练（需要训练数据）
index.train(train_embeddings)
# 添加数据
index.add(embeddings)
# 搜索（设置nprobe）
index.nprobe = 10
D, I = index.search(query_embeddings, k=10)
IVF变种：
IVF-Flat：
- 存储原始向量
- 精确距离计算
- 内存占用：N × d × 4 bytes
IVF-PQ：
- 向量量化压缩
- 近似距离计算
- 内存占用：N × (m × bits / 8) bytes
# IVF-PQ
m = 8 # 将向量分成8个子向量
bits = 8 # 每个子向量用8bit编码
index = faiss.IndexIVFPQ(quantizer, d, nlist, m, bits)
index.train(train_embeddings)
index.add(embeddings)
7.3.3 IVF参数调优
nlist（聚类数量）：
- 太小：每个聚类太大，搜索慢
- 太大：聚类精度低，召回率下降
- 推荐：√N 到 4√N
- 1M向量：nlist=1000-4000
- 10M向量：nlist=3000-10000
nprobe（搜索聚类数）：
- nprobe=1：最快，召回率低
- nprobe=nlist：退化为暴力搜索
- 推荐：10-50（平衡）
召回率 vs 速度权衡：
nprobe=1 → 召回率 ~70%, 速度 100x
nprobe=10 → 召回率 ~85%, 速度 10x
nprobe=50 → 召回率 ~95%, 速度 2x
nprobe=100 → 召回率 ~98%, 速度 1.2x
实验建议：
def tune_ivf_parameters(index, queries, ground_truth):
results = []
for nprobe in [1, 5, 10, 20, 50, 100]:
index.nprobe = nprobe
start = time.time()
D, I = index.search(queries, k=10)
latency = time.time() - start
recall = compute_recall(I, ground_truth)
results.append({
'nprobe': nprobe,
'recall': recall,
'latency': latency,
'qps': len(queries) / latency
})
return results
7.4 LSH (Locality-Sensitive Hashing)
7.4.1 算法原理
LSH通过哈希函数将相似的向量映射到相同的桶（bucket）中，查询时只需搜索同一桶内的向量。
核心思想：
- 传统哈希：相似输入→不同输出
- LSH：相似输入→相同输出（高概率）
余弦LSH（适用于归一化向量）：
import numpy as np
class CosineLSH:
def __init__(self, d, num_tables=10, hash_size=8):
"""
d: 向量维度
num_tables: 哈希表数量（提高召回率）
hash_size: 每个哈希函数的位数
"""
self.d = d
self.num_tables = num_tables
self.hash_size = hash_size
# 生成随机超平面
self.random_planes = [
np.random.randn(hash_size, d)
for _ in range(num_tables)
]
# 哈希表
self.tables = [defaultdict(list) for _ in range(num_tables)]
def hash_vector(self, v, table_id):
"""将向量哈希为bit串"""
projections = np.dot(self.random_planes[table_id], v)
# 大于0为1，小于0为0
hash_code = ''.join(['1' if p > 0 else '0' for p in projections])
return hash_code
def insert(self, vector, vector_id):
"""插入向量"""
for i in range(self.num_tables):
hash_code = self.hash_vector(vector, i)
self.tables[i][hash_code].append(vector_id)
def query(self, query_vector, k=10):
"""查询最近邻"""
candidates = set()
# 从所有哈希表中收集候选
for i in range(self.num_tables):
hash_code = self.hash_vector(query_vector, i)
candidates.update(self.tables[i].get(hash_code, []))
# 精确计算距离
if len(candidates) == 0:
return []
distances = [(vid, cosine_distance(query_vector, vectors[vid]))
for vid in candidates]
distances.sort(key=lambda x: x[1])
return [vid for vid, _ in distances[:k]]
MinHash（适用于集合相似度）：
from datasketch import MinHash
def create_minhash(text, num_perm=128):
"""为文本创建MinHash签名"""
m = MinHash(num_perm=num_perm)
for word in text.split():
m.update(word.encode('utf8'))
return m
# 估计Jaccard相似度
mh1 = create_minhash("the quick brown fox")
mh2 = create_minhash("the fast brown fox")
similarity = mh1.jaccard(mh2)
7.4.2 LSH优劣势
优势：
- ✅ 内存效率高（只存哈希码）
- ✅ 查询速度快（O(1)哈希查找）
- ✅ 支持高维数据
- ✅ 可调节召回率/速度权衡
劣势：
- ❌ 召回率相对较低（70-85%）
- ❌ 需要多个哈希表提高召回
- ❌ 参数调优复杂
- ❌ 不适合需要高精度的场景
适用场景：
- 超大规模数据集（>100M）
- 对召回率要求不高的场景
- 实时推荐系统
- 去重检测
7.4.3 FAISS中的LSH
import faiss
d = 1024
nbits = 16 # 每个向量编码为16bit
index = faiss.IndexLSH(d, nbits)
index.add(embeddings)
# 搜索
D, I = index.search(query_embeddings, k=10)
7.5 索引算法对比实验
实验设置：
- 数据集：1M个1024维向量
- 查询：1000个查询向量
- 硬件：CPU: Intel Xeon, 64GB RAM
- 指标：召回率@10, QPS, 内存
结果：
| 算法 | 索引时间 | 内存 | QPS | Recall@10 |
|---|---|---|---|---|
| Flat | - | 4GB | 10 | 100% |
| IVF(nlist=1000, nprobe=10) | 5min | 4GB | 500 | 85% |
| IVF(nprobe=50) | 5min | 4GB | 150 | 95% |
| HNSW(M=32, ef=50) | 30min | 4.2GB | 1000 | 96% |
| HNSW(ef=100) | 30min | 4.2GB | 600 | 98% |
| LSH(ntables=10, nbits=16) | 2min | 500MB | 800 | 75% |
| IVF-PQ(nlist=1000, m=8) | 10min | 1GB | 400 | 88% |
结论：
- 需要高召回（>95%）：选择HNSW
- 内存受限：选择IVF-PQ或LSH
- 平衡方案：IVF-Flat (nprobe=10-50)
- 极大规模：LSH或分层索引
7.6 混合索引策略
两阶段检索：
def two_stage_retrieval(query, coarse_index, fine_index,
coarse_k=100, final_k=10):
"""
Stage 1: 用快速索引（如IVF）粗筛
Stage 2: 用精确方法重排
"""
# 粗筛
_, coarse_ids = coarse_index.search(query, k=coarse_k)
candidate_vectors = [vectors[i] for i in coarse_ids[0]]
# 精确重排
distances = [np.dot(query, v) for v in candidate_vectors]
top_k_indices = np.argsort(distances)[::-1][:final_k]
return [coarse_ids[0][i] for i in top_k_indices]
分层索引：
class HierarchicalIndex:
def __init__(self):
# 粗粒度：文档级别
self.doc_index = faiss.IndexIVFFlat(...)
# 细粒度：chunk级别
self.chunk_index = faiss.IndexHNSWFlat(...)
def search(self, query):
# 先找相关文档
doc_ids = self.doc_index.search(query, k=5)
# 在文档内搜索chunks
relevant_chunks = []
for doc_id in doc_ids:
chunks = self.get_chunks_from_doc(doc_id)
chunk_results = self.chunk_index.search(query, k=2)
relevant_chunks.extend(chunk_results)
return relevant_chunks
八、常见问题与解决方案
问题1：内存溢出（OOM）
解决方案：
def batch_process_large_corpus(documents, batch_size=1000):
"""批量处理大规模语料"""
index = None
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
# 处理batch
embeddings = model.encode(batch)
if index is None:
d = embeddings.shape[1]
index = faiss.IndexHNSWFlat(d, 32)
index.add(embeddings)
# 显式清理
del embeddings
import gc
gc.collect()
return index
问题2：索引构建时间过长
解决方案：
# 1. 使用GPU加速
model = SentenceTransformer('model', device='cuda')
# 2. 多进程处理
from multiprocessing import Pool
def embed_chunk(chunk):
return model.encode(chunk)
with Pool(8) as pool:
embeddings = pool.map(embed_chunk, document_chunks)
# 3. 选择更快的索引类型
# HNSW（慢）→ IVF（快）
问题3：检索质量不佳
诊断与优化：
def diagnose_retrieval_quality(index, queries, ground_truth):
"""诊断检索质量问题"""
results = {
'low_similarity_scores': [],
'no_results': [],
'wrong_results': []
}
for query, truth in zip(queries, ground_truth):
query_emb = model.encode([query])
D, I = index.search(query_emb, k=10)
# 检查相似度分数
if D[0][0] < 0.5: # 余弦相似度阈值
results['low_similarity_scores'].append(query)
# 检查是否检索到正确结果
if truth not in I[0]:
results['wrong_results'].append({
'query': query,
'expected': truth,
'got': I[0].tolist()
})
# 给出建议
if results['low_similarity_scores']:
print("⚠ 建议：")
print(" - 检查embedding模型是否适配领域")
print(" - 考虑微调embedding模型")
print(" - 优化分块策略")
if results['wrong_results']:
print("⚠ 建议：")
print(" - 增加索引参数（如HNSW的ef、IVF的nprobe）")
print(" - 检查分块是否保留了关键信息")
print(" - 考虑使用混合搜索")
return results
参考文献
核心论文
- RAG原始论文: Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks (Lewis et al., 2020) - https://arxiv.org/abs/2005.11401
- HNSW算法: Efficient and robust approximate nearest neighbor search using Hierarchical Navigable Small World graphs (Malkov & Yashunin, 2018) - https://arxiv.org/abs/1603.09320
- MinerU技术报告: MinerU: An Open-Source Solution for Precise Document Extraction - https://arxiv.org/abs/2409.18839
开源工具与框架
- LangChain文档: RAG Tutorial - https://python.langchain.com/docs/tutorials/rag/
- PyMuPDF官方文档: LLM & RAG Integration - https://pymupdf.readthedocs.io/en/latest/rag.html
- MinerU项目: GitHub Repository - https://github.com/opendatalab/MinerU
- FAISS库: Facebook AI Similarity Search - https://github.com/facebookresearch/faiss
评估与基准测试
- MTEB排行榜: Massive Text Embedding Benchmark - https://huggingface.co/spaces/mteb/leaderboard
- PDF解析工具评测: A Comparative Study of PDF Parsing Tools (2024) - https://arxiv.org/abs/2410.09871
- RAG评估指南: RAGAS Framework - https://docs.ragas.io/
向量数据库
- Milvus官方文档: Vector Database Documentation - https://milvus.io/docs
- Weaviate文档: Vector Search Engine - https://weaviate.io/developers/weaviate
- Qdrant文档: Vector Database - https://qdrant.tech/documentation/
- Pinecone学习中心: Vector Indexes Guide - https://www.pinecone.io/learn/
技术博客与教程
- Databricks: Chunking Strategies for RAG Applications - https://community.databricks.com/t5/technical-blog/
- Hugging Face: Advanced RAG Cookbook - https://huggingface.co/learn/cookbook/
- Microsoft: RAG Chunking Phase Guide - https://learn.microsoft.com/en-us/azure/architecture/ai-ml/guide/rag/
- Stack Overflow: Chunking in RAG Applications - https://stackoverflow.blog/2024/12/27/breaking-up-is-hard-to-do-chunking-in-rag-applications/
学术资源
- Recent advances in text embeddings: A Comprehensive Review (2024) - https://arxiv.org/abs/2406.01607
- Locality-Sensitive Hashing: Theory and Applications (Andoni & Indyk) - Original LSH paper
- Product Quantization: for Approximate Nearest Neighbor Search (Jégou et al., 2011)
行业报告
- Gartner: Vector Database Market Overview (2024)
- Databricks: The Big Book of MLOps - RAG章节
- DeepLearning.AI: RAG Course Materials - https://www.deeplearning.ai/courses/retrieval-augmented-generation-rag/
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
23
0- 0
已为社区贡献5条内容
所有评论(0)