关于作者
- 深耕领域：大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
- 技术栈：Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
- 工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案
「让 AI 交互更智能，让技术落地更高效」
欢迎技术探讨与项目合作，解锁大模型与智能交互的无限可能！
RAG 混合检索全流程详解：每个节点的选型决策指南
从文档解析到最终答案生成，RAG 混合检索系统涉及 8 个关键节点。每个节点的选型决策都直接影响最终效果。本文基于 2024-2025 年最新研究和生产实践，全面解析每个节点的技术选型。
学术会议推荐
如果您对 AI Agent、人工智能前沿技术有研究，欢迎投稿以下国际学术会议：
2026年人工智能与智慧生活国际学术会议（ICAISL 2026）
- 时间：2026年5月29-31日
- 地点：中国-广州 & 马来西亚
- 检索：EI Compendex, Scopus
- 征稿主题：人工智能核心技术与算法、智慧生活场景应用
第二届人工智能、人机交互与自然语言处理国际学术会议（ICAHN 2026）
- 时间：2026年5月22-24日
- 地点：中国-厦门
- 主办：北京信息科技大学
- 检索：EI Compendex, Scopus
- 征稿主题：人工智能、人机交互、自然语言处理
第二届人工智能与数字金融国际学术会议（AIDF 2026）
- 时间：2026年5月29-31日
- 地点：中国-武汉
- 出版：ACM International Conference Proceeding Series
- 检索：EI Compendex, Scopus
- 征稿主题：人工智能技术、AI在数字金融中的应用、金融科技创新
第二届人工智能与数字伦理国际学术会议（ICAIDE 2026）
- 时间：2026年5月22-24日
- 地点：中国-广州 & 新加坡
- 出版：IEEE（ISBN: 979-8-3315-9297-4）
- 检索：EI Compendex, Scopus, IEEE Xplore
- 征稿主题：人工智能、数字伦理、电子信息科学与技术
一、RAG 混合检索全景架构
二、节点一：文档解析 (Document Parsing)
文档解析是RAG系统的起点，解析质量直接决定后续所有环节的效果。一个糟糕的解析结果——表格错乱、段落断裂、公式丢失——会让再先进的嵌入模型和检索策略都无济于事。
2.1 解析工具对比
| 工具 | 支持格式 | 表格解析 | OCR | 图片提取 | 速度 | 复杂度 |
|---|---|---|---|---|---|---|
| PyPDF2 | 差 | 无 | 无 | 快 | 低 | |
| pdfplumber | 优秀 | 无 | 无 | 中 | 中 | |
| PyMuPDF | 中 | 无 | 有 | 快 | 中 | |
| Unstructured | 多格式 | 好 | 有 | 有 | 中 | 中 |
| Docling | 多格式 | 优秀 | 有 | 有 | 中 | 高 |
| Marker | 优秀 | 有 | 有 | 中 | 高 | |
| LlamaParse | 多格式 | 优秀 | 有 | 有 | 慢 | 低 |
2.2 解析工具详细分析
PyPDF2是Python生态中最老牌的PDF解析库，纯Python实现意味着它可以在任何环境运行，无需担心依赖问题。它的速度确实很快，内存占用也低，适合处理大量简单的文本文档。但它的短板也很明显：表格解析几乎是灾难级的，遇到稍微复杂的排版就会把内容搅成一团乱麻。如果你的文档主要是纯文本的PDF，比如小说、论文正文、简单的报告，PyPDF2是个轻量的选择。但只要涉及表格或复杂布局，就别指望它了。
pdfplumber在表格提取方面堪称一绝。它基于pdfminer.six开发，但针对表格场景做了深度优化。pdfplumber能够识别表格边框，将单元格内容准确提取出来，甚至保留表格的结构信息。对于财务报表、数据表格密集型的技术文档，pdfplumber是首选。它的缺点是不支持OCR，所以扫描件PDF对它来说是无解的。另外，它的速度中等，处理超大文档时可能需要一些耐心。
PyMuPDF（也叫fitz）是速度最快的PDF解析库之一，底层用C语言实现。除了文本提取，它还能提取图片、注释、链接等元素。表格解析能力中等，比PyPDF2强但不如pdfplumber。如果你的场景需要快速处理大量PDF且对表格精度要求不高，PyMuPDF是不错的选择。它还支持PDF的各种操作（合并、分割、水印等），是个全能型选手。
Unstructured是近年来异军突起的多格式解析工具。它最大的卖点是"统一接口"——无论你扔给它PDF、Word、PPT、HTML还是Markdown，它都能处理。Unstructured会自动检测文档类型，选择合适的解析策略。它支持OCR（需要配置Tesseract），能够处理扫描件和图片中的文字。对于格式杂乱的知识库，Unstructured能大大简化开发工作。代价是依赖较多，首次安装可能需要折腾一番。
Docling是IBM在2024年开源的文档解析工具，质量相当惊艳。它基于深度学习模型，能够理解文档的语义结构——标题、段落、表格、公式、代码块都能准确识别。Docling输出的Markdown格式非常干净，表格渲染效果接近原文档。它还支持OCR和多语言处理。缺点是相对较新，社区生态还在建设中，学习曲线也比传统工具陡峭一些。对于学术论文、技术文档这类复杂排版，Docling是目前最好的开源选择之一。
Marker专门为PDF转Markdown优化，特别擅长处理学术论文和书籍。它使用深度学习模型进行布局分析和OCR，输出的Markdown质量很高，公式和表格都能较好保留。如果你的目标是把PDF文档转换为结构化的Markdown存入知识库，Marker值得尝试。
LlamaParse是LlamaIndex推出的托管解析服务。它把复杂的解析工作放到云端，你只需调用API就能获得高质量的解析结果。LlamaParse支持多格式，OCR能力强大，表格提取效果优秀。最大的优势是零部署成本——不需要安装任何依赖，不用担心环境配置。代价是速度较慢（网络延迟+排队），而且需要付费。对于不想折腾基础设施的团队，LlamaParse是省心的选择。
2.3 选型决策建议
选择文档解析工具时，首先要明确你的文档类型和质量要求。如果全是简单的文本PDF，PyPDF2或PyMuPDF足够用；如果表格密集，pdfplumber是必选项；如果格式多样，Unstructured或Docling更合适；如果是扫描件，必须选择支持OCR的工具。
其次考虑部署约束。需要离线部署的场景，只能选择开源工具；如果可以接受云服务，LlamaParse能省去很多麻烦。
最后是成本权衡。开源工具免费但需要投入开发和维护成本；商业服务按量付费，长期使用成本可能很高。
2.4 代码实现
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from dataclasses import dataclass
@dataclass
class ParsedDocument:
"""解析后的文档"""
content: str
metadata: Dict[str, Any]
tables: List[Dict] = None
images: List[Dict] = None
class DocumentParser(ABC):
"""文档解析器基类"""
@abstractmethod
def parse(self, file_path: str) -> List[ParsedDocument]:
pass
class PyPDF2Parser(DocumentParser):
"""
PyPDF2 解析器
优点：
- 纯 Python 实现，无需外部依赖
- 速度快，适合批量处理
- 内存占用低
缺点：
- 表格解析能力差
- 不支持 OCR
- 复杂排版处理弱
适用场景：
- 纯文本 PDF
- 简单排版文档
- 对速度要求高的批量处理
"""
def parse(self, file_path: str) -> List[ParsedDocument]:
import PyPDF2
documents = []
with open(file_path, 'rb') as f:
reader = PyPDF2.PdfReader(f)
for i, page in enumerate(reader.pages):
text = page.extract_text()
documents.append(ParsedDocument(
content=text,
metadata={'page': i + 1, 'source': file_path}
))
return documents
class PDFPlumberParser(DocumentParser):
"""
pdfplumber 解析器
优点：
- 表格提取能力优秀
- 可保留表格结构
- 支持精确位置信息
缺点：
- 不支持 OCR
- 扫描件无法处理
- 速度中等
适用场景：
- 财务报表
- 数据表格密集型文档
- 需要保留表格结构的场景
"""
def parse(self, file_path: str) -> List[ParsedDocument]:
import pdfplumber
documents = []
with pdfplumber.open(file_path) as pdf:
for i, page in enumerate(pdf.pages):
# 提取文本
text = page.extract_text() or ""
# 提取表格
tables = page.extract_tables()
table_texts = []
for table in tables:
table_text = self._format_table(table)
table_texts.append(table_text)
# 合并
if table_texts:
text += "\n\n表格内容:\n" + "\n\n".join(table_texts)
documents.append(ParsedDocument(
content=text,
metadata={'page': i + 1, 'source': file_path, 'has_tables': len(tables) > 0},
tables=tables
))
return documents
def _format_table(self, table: List[List[str]]) -> str:
if not table:
return ""
lines = []
for row in table:
cells = [str(cell) if cell else "" for cell in row]
lines.append(" | ".join(cells))
return "\n".join(lines)
class UnstructuredParser(DocumentParser):
"""
Unstructured 解析器
优点：
- 支持多种格式（PDF、Word、PPT、HTML 等）
- 自动检测文档类型
- 支持 OCR（需配置）
- 保留文档结构
缺点：
- 依赖较多
- OCR 需要额外配置
- 速度较慢
适用场景：
- 多格式文档混合
- 需要统一处理接口
- 扫描件处理
"""
def __init__(self, enable_ocr: bool = False):
self.enable_ocr = enable_ocr
def parse(self, file_path: str) -> List[ParsedDocument]:
from unstructured.partition.auto import partition
elements = partition(
filename=file_path,
skip_infer_table_types=[] if self.enable_ocr else ['pdf'],
strategy='hi_res' if self.enable_ocr else 'fast'
)
documents = []
current_content = []
current_metadata = {'source': file_path}
for element in elements:
element_type = type(element).__name__
# 按段落分块
if element_type in ['Title', 'Header']:
if current_content:
documents.append(ParsedDocument(
content='\n'.join(current_content),
metadata=current_metadata
))
current_content = [str(element)]
current_metadata = {'source': file_path, 'type': element_type}
else:
current_content.append(str(element))
if current_content:
documents.append(ParsedDocument(
content='\n'.join(current_content),
metadata=current_metadata
))
return documents
class DoclingParser(DocumentParser):
"""
Docling 解析器（IBM 开源）
优点：
- IBM 开源，质量高
- 复杂布局处理优秀
- 表格解析能力强
- 支持 OCR
- 输出 Markdown 格式
缺点：
- 较新，生态待完善
- 依赖较多
- 学习曲线
适用场景：
- 复杂排版文档
- 学术论文
- 技术文档
- 需要高质量解析的场景
"""
def parse(self, file_path: str) -> List[ParsedDocument]:
from docling.document_converter import DocumentConverter
converter = DocumentConverter()
result = converter.convert(file_path)
# 导出为 Markdown
markdown_content = result.document.export_to_markdown()
# 按段落分割
paragraphs = markdown_content.split('\n\n')
documents = []
for i, para in enumerate(paragraphs):
if para.strip():
documents.append(ParsedDocument(
content=para.strip(),
metadata={'source': file_path, 'paragraph': i + 1}
))
return documents
2.3 解析选型决策
三、节点二：文档分块 (Chunking)
分块是RAG系统中最容易被低估的环节。很多人把精力花在嵌入模型和检索策略上，却忽视了分块质量对最终效果的巨大影响。一个糟糕的分块策略会把完整的语义单元切得支离破碎——“苹果公司发布了新产品"可能被切成"苹果"和"公司发布了新产品”，检索时用户问"苹果公司"却找不到相关内容。
3.1 分块策略对比
| 策略 | 原理 | 语义完整性 | 实现复杂度 | 计算成本 | 适用场景 |
|---|---|---|---|---|---|
| 固定长度 | 按字符/Token数切分 | 低 | 低 | 低 | 结构化文档 |
| 递归分块 | 按分隔符层级切分 | 中 | 中 | 低 | 通用文档 |
| 语义分块 | 基于嵌入相似度 | 高 | 高 | 高 | 高精度场景 |
| 层次分块 | Parent-Child 结构 | 高 | 高 | 中 | 需要上下文 |
| 句子窗口 | 句子+周围上下文 | 高 | 中 | 中 | 问答系统 |
| LLM分块 | LLM 智能分割 | 最高 | 最高 | 最高 | 复杂文档 |
3.2 分块策略详细分析
固定长度分块是最简单的策略：按字符数或Token数直接切分。比如每512个Token切一块，相邻块之间有100个Token的重叠。这种方法实现简单、速度快、块大小可控，但问题也很明显——它完全不考虑语义边界。一个句子可能被切成两半，一个段落可能被拆散。固定长度分块适合处理结构化程度高的文档，比如日志文件、代码片段，或者对分块质量要求不高的快速原型开发。实际使用时，建议块大小设置在256-512 tokens之间，重叠比例10%-20%。
递归分块是生产环境中最常用的策略。它的核心思想是"优先按大分隔符切，切不动再按小分隔符切"。典型的分隔符层级是：段落（\n\n）→ 句子（\n或句号）→ 词（空格）→ 字符。递归分块会先尝试按段落切分，如果某段太长，再按句子切，以此类推。这种方式在保留语义完整性和控制块大小之间取得了很好的平衡。LangChain的RecursiveCharacterTextSplitter就是这类策略的代表实现。对于大多数通用文档，递归分块是默认选择。
语义分块更进一步，它使用嵌入模型来判断语义边界。具体做法是：先把文档切成句子，计算每个句子的嵌入向量，然后计算相邻句子的相似度。当相似度突然下降时，说明话题发生了转换，这里就是一个自然的分块边界。语义分块能够自动检测主题边界，语义完整性最好，但代价是计算成本高（需要为每个句子计算嵌入），而且块大小不可控——可能产生很长的块，也可能产生很短的块。适合对检索精度要求极高、且能接受计算成本的场景。
层次分块（Parent-Child） 是为了解决一个常见的痛点：检索时需要精确匹配，但生成时需要完整上下文。它的做法是将文档切成两层：父块（比如2000 tokens）和子块（比如400 tokens）。检索时匹配子块，返回时却返回整个父块。这样既能保证检索的精确性，又能给LLM提供足够的上下文。层次分块特别适合法律文档、医疗文档这类需要完整上下文的场景。代价是存储成本翻倍，索引结构也更复杂。
句子窗口分块 是层次分块的一种变体。它以单个句子为单位进行索引，但每个句子都附带前后若干句作为"窗口"。检索时匹配单个句子，返回时返回整个窗口。这种方法在问答系统中效果很好，因为问题往往针对某个具体句子，但回答时需要周围上下文来理解完整含义。
LLM分块是最新也是最昂贵的方法：让大语言模型来决定如何分块。你可以让LLM识别文档的主题边界，或者让它为每个块生成摘要。LLM分块能够处理最复杂的文档结构，理解最微妙的语义边界，但成本和延迟也是最高的。通常只在对分块质量要求极高、且文档量不大的场景下使用。
3.3 分块参数调优建议
分块大小和重叠比例是两个关键参数，需要根据具体场景调优：
块大小的选择要考虑嵌入模型的限制和下游任务的特性。太小的块（比如64 tokens）会丢失上下文，检索结果碎片化；太大的块（比如2000 tokens）会稀释关键信息，增加噪声。一般来说，512 tokens是个不错的起点。如果你的文档段落较长，可以尝试800-1000 tokens；如果是问答场景，256-400 tokens可能更合适。
重叠比例的设置要平衡信息冗余和检索效率。重叠太少，边界处的关键信息可能丢失；重叠太多，会增加存储和计算成本。10%-20%的重叠比例在大多数场景下表现良好。对于语义敏感的场景（比如法律、医疗），可以适当增加到25%-30%。
3.4 代码实现
from abc import ABC, abstractmethod
from typing import List, Tuple
from dataclasses import dataclass
import numpy as np
@dataclass
class Chunk:
"""文档块"""
content: str
metadata: Dict
start_idx: int = 0
end_idx: int = 0
class ChunkingStrategy(ABC):
"""分块策略基类"""
@abstractmethod
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
pass
class FixedSizeChunker(ChunkingStrategy):
"""
固定长度分块
优点：
- 实现简单
- 块大小可控
- 计算成本低
- 适合并行处理
缺点：
- 可能截断语义
- 边界信息丢失
- 不考虑文档结构
最佳实践：
- 块大小：256-512 tokens
- 重叠：10-20%
适用场景：
- 结构化文档
- 对块大小有严格要求
- 快速原型开发
"""
def __init__(
self,
chunk_size: int = 512,
overlap: int = 100,
unit: str = "character" # character 或 token
):
self.chunk_size = chunk_size
self.overlap = overlap
self.unit = unit
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
metadata = metadata or {}
if self.unit == "token":
# Token 级别分块
tokens = self._tokenize(text)
chunks = []
start = 0
while start < len(tokens):
end = min(start + self.chunk_size, len(tokens))
chunk_tokens = tokens[start:end]
chunk_text = self._detokenize(chunk_tokens)
chunks.append(Chunk(
content=chunk_text,
metadata={**metadata, 'chunk_type': 'fixed_token'},
start_idx=start,
end_idx=end
))
start = end - self.overlap
else:
# 字符级别分块
chunks = []
start = 0
while start < len(text):
end = min(start + self.chunk_size, len(text))
chunk_text = text[start:end]
chunks.append(Chunk(
content=chunk_text,
metadata={**metadata, 'chunk_type': 'fixed_char'},
start_idx=start,
end_idx=end
))
start = end - self.overlap
return chunks
def _tokenize(self, text: str) -> List[str]:
"""简单分词"""
return text.split()
def _detokenize(self, tokens: List[str]) -> str:
"""合并 tokens"""
return ' '.join(tokens)
class RecursiveChunker(ChunkingStrategy):
"""
递归分块
优点：
- 保留语义边界
- 适应不同文档结构
- 平衡大小和语义
- 生产环境推荐
缺点：
- 需要调参
- 对分隔符依赖
- 可能产生不均匀块
最佳实践：
- 默认分隔符：["\n\n", "\n", "。", ".", " ", ""]
- 块大小：512 tokens
- 重叠：100 tokens
适用场景：
- 通用文档处理
- 生产环境默认选择
- 长文档处理
"""
def __init__(
self,
chunk_size: int = 512,
chunk_overlap: int = 100,
separators: List[str] = None
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.separators = separators or [
"\n\n", # 段落
"\n", # 行
"。", # 中文句号
"！", # 中文感叹号
"？", # 中文问号
".", # 英文句号
"!", # 英文感叹号
"?", # 英文问号
" ", # 空格
"" # 字符
]
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
metadata = metadata or {}
return self._recursive_chunk(text, metadata, self.separators)
def _recursive_chunk(
self,
text: str,
metadata: Dict,
separators: List[str]
) -> List[Chunk]:
# 如果文本已经足够小
if len(text) <= self.chunk_size:
return [Chunk(
content=text,
metadata={**metadata, 'chunk_type': 'recursive'},
start_idx=0,
end_idx=len(text)
)]
# 尝试按当前分隔符切分
for separator in separators:
if separator and separator in text:
splits = text.split(separator)
chunks = []
current_chunk = ""
for split in splits:
if len(current_chunk) + len(split) + len(separator) <= self.chunk_size:
current_chunk += split + separator
else:
if current_chunk:
chunks.append(Chunk(
content=current_chunk.strip(),
metadata={**metadata, 'chunk_type': 'recursive'},
start_idx=0,
end_idx=len(current_chunk)
))
current_chunk = split + separator
if current_chunk:
chunks.append(Chunk(
content=current_chunk.strip(),
metadata={**metadata, 'chunk_type': 'recursive'},
start_idx=0,
end_idx=len(current_chunk)
))
# 检查是否所有块都满足要求
if all(len(c.content) <= self.chunk_size for c in chunks):
return chunks
# 仍有块过大，递归处理
final_chunks = []
next_separators = separators[separators.index(separator) + 1:]
for c in chunks:
if len(c.content) > self.chunk_size:
if next_separators:
final_chunks.extend(
self._recursive_chunk(c.content, metadata, next_separators)
)
else:
final_chunks.append(c)
else:
final_chunks.append(c)
return final_chunks
# 无分隔符，按固定长度切分
return FixedSizeChunker(self.chunk_size, self.chunk_overlap).chunk(text, metadata)
class SemanticChunker(ChunkingStrategy):
"""
语义分块
优点：
- 最佳语义完整性
- 自动检测主题边界
- 适应文档内容
缺点：
- 计算成本高
- 需要嵌入模型
- 块大小不可控
- 可能产生过大块
最佳实践：
- 断点阈值：percentile 95 或 standard_deviation
- 最小块大小：100 字符
- 最大块大小：2000 字符
适用场景：
- 高精度检索
- 主题明确的文档
- 可接受计算成本
"""
def __init__(
self,
embedding_model,
breakpoint_threshold_type: str = "percentile",
breakpoint_threshold_amount: float = 95,
min_chunk_size: int = 100,
max_chunk_size: int = 2000
):
self.embedding_model = embedding_model
self.breakpoint_threshold_type = breakpoint_threshold_type
self.breakpoint_threshold_amount = breakpoint_threshold_amount
self.min_chunk_size = min_chunk_size
self.max_chunk_size = max_chunk_size
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
metadata = metadata or {}
# 按句子切分
sentences = self._split_sentences(text)
if len(sentences) <= 1:
return [Chunk(
content=text,
metadata={**metadata, 'chunk_type': 'semantic'},
start_idx=0,
end_idx=len(text)
)]
# 计算句子嵌入
embeddings = self.embedding_model.encode(sentences)
# 计算相邻句子相似度
similarities = []
for i in range(len(embeddings) - 1):
sim = np.dot(embeddings[i], embeddings[i + 1])
similarities.append(sim)
# 计算断点阈值
if self.breakpoint_threshold_type == "percentile":
threshold = np.percentile(similarities, self.breakpoint_threshold_amount)
else: # standard_deviation
mean = np.mean(similarities)
std = np.std(similarities)
threshold = mean - self.breakpoint_threshold_amount * std
# 找断点
breakpoints = [0]
for i, sim in enumerate(similarities):
if sim < threshold:
breakpoints.append(i + 1)
breakpoints.append(len(sentences))
# 构建块
chunks = []
for i in range(len(breakpoints) - 1):
start = breakpoints[i]
end = breakpoints[i + 1]
chunk_text = " ".join(sentences[start:end])
# 检查块大小
if len(chunk_text) < self.min_chunk_size and i < len(breakpoints) - 2:
continue # 合并到下一个块
if len(chunk_text) > self.max_chunk_size:
# 递归切分
sub_chunks = RecursiveChunker(self.max_chunk_size, 100).chunk(chunk_text, metadata)
chunks.extend(sub_chunks)
continue
chunks.append(Chunk(
content=chunk_text,
metadata={**metadata, 'chunk_type': 'semantic'},
start_idx=start,
end_idx=end
))
return chunks
def _split_sentences(self, text: str) -> List[str]:
import re
# 中英文句子切分
sentences = re.split(r'(?<=[。！？.!?])\s*', text)
return [s.strip() for s in sentences if s.strip()]
class HierarchicalChunker(ChunkingStrategy):
"""
层次分块（Parent-Child）
优点：
- 检索时获取完整上下文
- 平衡精确性和完整性
- 支持多粒度检索
缺点：
- 存储成本高
- 实现复杂
- 索引维护复杂
适用场景：
- 需要上下文的问答
- 法律/医疗文档
- 技术文档检索
"""
def __init__(
self,
parent_size: int = 2000,
child_size: int = 400,
child_overlap: int = 50
):
self.parent_size = parent_size
self.child_size = child_size
self.child_overlap = child_overlap
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
metadata = metadata or {}
# 先切分为父块
parent_chunks = RecursiveChunker(self.parent_size, 200).chunk(text, metadata)
# 每个父块再切分为子块
all_chunks = []
for parent in parent_chunks:
child_chunks = RecursiveChunker(self.child_size, self.child_overlap).chunk(
parent.content,
{**metadata, 'parent_id': id(parent), 'parent_content': parent.content}
)
all_chunks.extend(child_chunks)
return all_chunks
class LateChunker(ChunkingStrategy):
"""
延迟分块（Late Chunking）
论文：2024 年最新研究
原理：
1. 先对整个文档编码，生成 token 级嵌入
2. 编码时利用自注意力捕获全局上下文
3. 编码后再按边界聚合为块级嵌入
优点：
- 块嵌入包含全局上下文
- 解决代词指代问题
- 提高长文档检索精度
缺点：
- 需要长上下文嵌入模型
- 计算成本高
- 实现复杂
适用场景：
- 长文档检索
- 复杂语义依赖
- 高精度需求
"""
def __init__(self, embedding_model, chunk_size: int = 512):
self.embedding_model = embedding_model
self.chunk_size = chunk_size
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
metadata = metadata or {}
# 需要支持长上下文的嵌入模型
# 如 Jina Embeddings v2 (8K context)
# 1. 对整个文档编码
# token_embeddings = self.embedding_model.encode_tokens(text)
# 2. 定义分块边界
# chunk_boundaries = self._define_boundaries(text, self.chunk_size)
# 3. 聚合 token 嵌入为块嵌入
# chunk_embeddings = self._aggregate_embeddings(token_embeddings, chunk_boundaries)
# 简化实现：使用递归分块
return RecursiveChunker(self.chunk_size, 100).chunk(text, metadata)
3.3 分块选型决策
四、节点三：嵌入模型 (Embedding Model)
嵌入模型是RAG系统的核心组件，它将文本转换为高维向量，决定了语义理解的上限。选型时需要权衡精度、成本、部署难度和语言支持等多个维度。
4.1 国际主流嵌入模型
OpenAI text-embedding-3系列是快速集成的首选。text-embedding-3-large在MTEB基准测试中得分64.6，3072维向量提供了丰富的语义表达。它的优势在于零部署成本、稳定的质量和多语言支持，但每百万token 0.13美元的价格在大规模应用中成本可观。对于成本敏感的场景，text-embedding-3-small以0.02美元/百万token的价格提供了62.3的MTEB分数，性价比突出。不过，使用OpenAI模型意味着数据需要传输到云端，对隐私要求高的场景需要慎重考虑。
Cohere embed-v4是企业级应用的标杆。65.2的MTEB分数使其在商业模型中名列前茅，更重要的是它支持向量压缩——可以将1024维向量压缩到更小维度，显著降低存储成本。Cohere的input_type参数允许针对不同场景（搜索文档、搜索查询、聚类等）优化嵌入效果，这种任务感知的设计在实际应用中能带来可观的提升。价格适中（0.10美元/百万token），适合有稳定预算的企业项目。
Voyage-3-large代表了当前开源/商业模型中的精度天花板。66.8的MTEB分数是目前公开基准测试中的最高水平，1536维向量在语义表达上更加精细。对于追求极致检索质量的场景，Voyage是值得考虑的选择。不过0.12美元/百万token的价格和相对较高的延迟需要在选型时权衡。
Jina-embeddings-v3的独特优势在于8K的长上下文支持。对于需要处理长文档的场景，Jina可以在一次编码中捕获更多上下文信息，避免了分块带来的语义断裂。64.0的MTEB分数虽然不是最高，但长上下文能力在某些场景下是无价的。
4.2 国内开源嵌入模型
国内开源嵌入模型近年来发展迅猛，在某些基准测试中已经超越国际同类产品。
Qwen-3-Embedding系列是2025年6月发布的最新力作，代表了国内嵌入模型的最高水平。旗舰模型Qwen-3-Embedding-8B在MTEB多语言排行榜上以70.58分位居第一，超越了此前的霸主Gemini-Embedding。更令人惊喜的是，轻量级的Qwen-3-Embedding-0.6B在MMTEB上取得了64.33分，相比同等规模的BGE-M3提升了7.9%，在中文CMTEB上更是领先9.9%。这意味着在相同计算资源下，Qwen-3能提供更精准的语义理解。
Qwen-3系列的另一大特色是指令感知架构。用户可以通过自定义指令来优化特定领域的嵌入效果，实验表明使用指令通常能带来1%-5%的性能提升。例如，对于法律文档检索，可以添加"为法律条文检索生成嵌入"这样的指令，让模型更准确地理解语义边界。模型支持100多种自然语言和编程语言，在代码检索任务上同样表现出色（MTEB Code得分80.68）。
BGE-M3是北京智源研究院推出的多语言嵌入模型，568M参数量在精度和效率之间取得了良好平衡。它最大的特色是"三合一"架构：同时支持密集检索（Dense）、稀疏检索（Sparse）和多向量检索（Multi-vector）。这意味着同一个模型可以适配不同的检索策略，在混合检索场景中尤为便利。BGE-M3支持100多种语言，8K上下文长度，对于多语言知识库是理想选择。在MTEB多语言评测中得分59.56，虽然略逊于Qwen-3，但其成熟度和社区支持更加完善。
BGE-large-zh是专门针对中文优化的版本，在中文检索任务上表现更为出色。如果你的知识库主要是中文内容，这个模型往往比多语言模型更精准。1024维向量在存储和检索效率上也相对友好。
GTE-Qwen系列是阿里巴巴推出的嵌入模型，基于Qwen-2架构。gte-Qwen2-7B-instruct在MTEB英文评测中达到71.62分，gte-Qwen2-1.5B-instruct也有67.12分的表现。GTE系列同样支持指令，可以根据任务类型动态调整嵌入策略。
text2vec系列是国内较早的开源嵌入模型，虽然性能不如上述新模型，但胜在轻量和成熟。对于资源受限或对精度要求不高的场景，text2vec-chinese仍是可用的选择。
4.3 嵌入模型选型对比
从精度角度看，Qwen-3-Embedding-8B > Voyage-3-large > Cohere embed-v4 > Qwen-3-Embedding-0.6B > OpenAI text-embedding-3-large > BGE-M3。但精度只是选型的维度之一。
从成本角度，开源模型（BGE、Qwen-3、GTE）免费但需要自行部署和维护；商业API模型（OpenAI、Cohere、Voyage）按量付费，省去运维成本但长期使用费用可观。对于日调用量百万级的场景，开源模型的经济优势明显。
从部署难度看，OpenAI和Cohere只需API调用，零运维；BGE和Qwen-3需要GPU服务器，但Hugging Face和Sentence Transformers提供了成熟的工具链，部署门槛并不高。对于没有GPU资源的团队，也可以使用云服务商的模型托管服务。
从语言支持看，Qwen-3、BGE-M3、GTE-Qwen都支持100多种语言，且在中文场景表现优异；OpenAI和Cohere的多语言支持也不错，但在中文专业术语上可能不如国内模型精准。
4.4 嵌入模型代码实现
from abc import ABC, abstractmethod
from typing import List
import numpy as np
class EmbeddingModel(ABC):
"""嵌入模型基类"""
@abstractmethod
def encode(self, texts: List[str]) -> np.ndarray:
pass
@property
@abstractmethod
def dimension(self) -> int:
pass
class OpenAIEmbedding(EmbeddingModel):
"""
OpenAI 嵌入模型
优点：
- 无需部署，API 调用
- 质量稳定
- 多语言支持
- 维度可调
缺点：
- 需要付费
- 数据隐私问题
- 网络延迟
- 依赖外部服务
成本：
- text-embedding-3-small: $0.02/1M tokens
- text-embedding-3-large: $0.13/1M tokens
适用场景：
- 快速原型
- 已使用 OpenAI 生态
- 对隐私要求不高
"""
def __init__(
self,
model_name: str = "text-embedding-3-large",
api_key: str = None,
dimensions: int = None
):
from openai import OpenAI
self.client = OpenAI(api_key=api_key)
self.model_name = model_name
self.dimensions = dimensions
self._dimensions = {
"text-embedding-3-small": 1536,
"text-embedding-3-large": 3072,
}
def encode(self, texts: List[str]) -> np.ndarray:
kwargs = {"input": texts, "model": self.model_name}
if self.dimensions:
kwargs["dimensions"] = self.dimensions
response = self.client.embeddings.create(**kwargs)
embeddings = [item.embedding for item in response.data]
return np.array(embeddings)
@property
def dimension(self) -> int:
if self.dimensions:
return self.dimensions
return self._dimensions.get(self.model_name, 1536)
class Qwen3Embedding(EmbeddingModel):
"""
Qwen-3 嵌入模型（阿里云开源）
优点：
- MTEB多语言排名第一（8B版本）
- 支持指令感知
- 100+语言支持
- 中文效果优异
- 完全免费开源
缺点：
- 需要自行部署
- 8B版本需要较大显存
- 0.6B版本精度略低
模型选择：
- Qwen3-Embedding-0.6B: 轻量级，适合资源受限场景
- Qwen3-Embedding-4B: 平衡选择
- Qwen3-Embedding-8B: 最高精度，需要24G+显存
适用场景：
- 中文为主的知识库
- 多语言检索
- 对精度要求高
- 需要私有化部署
"""
def __init__(
self,
model_name: str = "Qwen/Qwen3-Embedding-0.6B",
device: str = "cuda",
instruction: str = None
):
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(model_name, device=device)
self.instruction = instruction
self._dimension = self.model.get_sentence_embedding_dimension()
def encode(
self,
texts: List[str],
instruction: str = None
) -> np.ndarray:
instr = instruction or self.instruction
if instr:
texts = [f"Instruct: {instr}\nQuery: {t}" for t in texts]
return self.model.encode(
texts,
normalize_embeddings=True,
show_progress_bar=len(texts) > 100
)
@property
def dimension(self) -> int:
return self._dimension
class BGEEmbedding(EmbeddingModel):
"""
BGE 嵌入模型（北京智源开源）
优点：
- 完全免费
- 支持本地部署
- 多语言（100+）
- 质量接近商业模型
- 支持稀疏+密集向量
缺点：
- 需要自己部署
- 需要GPU加速
- 维护成本
模型选择：
- bge-m3: 多语言，支持dense/sparse/multi-vector
- bge-large-zh: 中文专用
- bge-large-en: 英文专用
适用场景：
- 需要开源模型
- 数据隐私要求
- 多语言场景
- 私有化部署
"""
def __init__(
self,
model_name: str = "BAAI/bge-m3",
device: str = "cuda",
normalize: bool = True
):
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(model_name, device=device)
self.normalize = normalize
self._dimension = self.model.get_sentence_embedding_dimension()
def encode(self, texts: List[str]) -> np.ndarray:
return self.model.encode(
texts,
normalize_embeddings=self.normalize,
show_progress_bar=len(texts) > 100
)
@property
def dimension(self) -> int:
return self._dimension
class CohereEmbedding(EmbeddingModel):
"""
Cohere 嵌入模型
优点：
- MTEB 分数高
- 支持压缩存储
- input_type 优化
- 企业级 SLA
缺点：
- 需要付费
- API 调用延迟
成本：
- $0.10/1M tokens
特色功能：
- embed-english-v3.0: 英文优化
- embed-multilingual-v3.0: 多语言
- 支持压缩到更小维度
适用场景：
- 企业级应用
- 需要存储压缩
- 多语言场景
"""
def __init__(
self,
model_name: str = "embed-v4",
api_key: str = None,
input_type: str = "search_document"
):
import cohere
self.client = cohere.Client(api_key)
self.model_name = model_name
self.input_type = input_type
def encode(
self,
texts: List[str],
input_type: str = None
) -> np.ndarray:
response = self.client.embed(
texts=texts,
model=self.model_name,
input_type=input_type or self.input_type
)
return np.array(response.embeddings)
@property
def dimension(self) -> int:
return 1024
4.5 嵌入模型选型决策
五、节点四：向量数据库 (Vector Database)
向量数据库是RAG系统的存储核心，负责向量数据的持久化、索引和检索。选型时需要考虑规模、性能、功能特性和运维成本等多个维度。一个不合适的向量数据库可能成为整个系统的瓶颈——要么撑不住数据量，要么查询太慢，要么缺少关键功能。
5.1 数据库对比
| 数据库 | 类型 | 规模上限 | 混合检索 | 部署方式 | 查询延迟 | 月成本(10M向量) |
|---|---|---|---|---|---|---|
| Milvus | 开源 | 十亿级 | 原生支持 | 云/自托管 | 75-150ms | $150-400 |
| Pinecone | 托管 | 百万级 | 支持 | 仅云 | <50ms | $200-500 |
| Qdrant | 开源 | 百万级 | 支持 | 云/自托管 | <70ms | $50-150 |
| Weaviate | 开源 | 百万级 | 优秀 | 云/自托管 | 50-100ms | $100-300 |
| Chroma | 嵌入式 | 十万级 | 基础 | 本地 | 100-200ms | 免费-$50 |
| pgvector | 扩展 | 十万级 | 基础 | 自托管 | 100-300ms | $100-200 |
5.2 数据库详细分析
Milvus是向量数据库领域的"重型武器"，专为大规模场景设计。它支持十亿级向量的存储和检索，原生支持混合检索（稀疏向量+密集向量），这对于RAG系统来说是个巨大的优势——你可以在同一个查询中同时使用语义检索和关键词检索。Milvus提供多种索引类型：HNSW适合高召回场景但内存消耗大，IVF_FLAT是平衡选择，IVF_PQ可以大幅降低内存但会有精度损失，GPU_CAGRA则利用GPU加速查询。Milvus的缺点是部署和运维复杂，需要配置多个组件（etcd、MinIO等），学习曲线陡峭。对于千万级以上向量的大规模生产环境，Milvus是最可靠的选择。
Pinecone是全托管向量数据库的代表，主打"零运维"。你不需要关心服务器配置、索引优化、数据分片，只需要调用API。Pinecone的查询延迟极低（<50ms），自动扩缩容，自动处理数据备份和恢复。对于不想投入运维资源的团队，Pinecone是省心的选择。但代价是成本较高，而且数据必须存储在Pinecone的云端，对数据隐私有要求的场景需要考虑。另外，Pinecone的规模上限相对较低（百万级），超大规模场景可能需要分片。
Qdrant是用Rust编写的向量数据库，性能和内存效率都非常出色。它支持百万级向量，查询延迟低（<70ms），同时提供了强大的过滤能力——你可以在向量检索的同时进行复杂的元数据过滤。Qdrant的部署简单，单机版开箱即用，也支持分布式部署。它的成本相对较低，是性价比很高的选择。缺点是生态不如Milvus成熟，大规模分布式部署的经验积累较少。对于百万级向量、需要复杂过滤的场景，Qdrant是理想选择。
Weaviate在混合检索方面表现突出。它内置了多种向量化模块，可以自动为文本、图片等数据生成向量，也支持自定义向量。Weaviate的GraphQL API设计优雅，查询灵活。它的混合检索能力是最大的卖点——可以同时使用向量检索和关键词检索，并自动融合结果。Weaviate还支持多租户，适合SaaS应用。缺点是资源占用较高，大规模场景下成本可能比Qdrant高。如果你的RAG系统需要强大的混合检索能力，Weaviate值得重点考虑。
Chroma是轻量级的嵌入式向量数据库，类似于SQLite之于关系型数据库。它完全在本地运行，无需任何服务器配置，几行代码就能启动。Chroma非常适合原型开发、本地测试、小规模应用。它的缺点也很明显：不支持分布式，性能有限，功能相对基础。如果你的向量数据在十万级以下，或者只是做实验和开发，Chroma是最简单的选择。Chroma也提供云服务，可以无缝从本地迁移到云端。
pgvector是PostgreSQL的向量扩展，让你可以在关系型数据库中存储和检索向量。如果你已经在使用PostgreSQL，pgvector是最自然的选择——不需要引入新的数据库，不需要学习新的API，向量数据和业务数据存储在一起。pgvector支持HNSW和IVFFlat索引，基本功能齐全。缺点是性能不如专用向量数据库，大规模场景下查询延迟较高。适合已有PostgreSQL技术栈、向量数据量不大（十万级以下）的场景。
5.3 选型决策建议
选择向量数据库时，首先要评估数据规模。十万级以下，Chroma或pgvector足够；百万级，Qdrant或Weaviate是主流选择；千万级以上，Milvus是首选。
其次考虑混合检索需求。如果你的RAG系统需要同时使用语义检索和关键词检索，Milvus和Weaviate的原生支持会让实现简单很多。其他数据库虽然也能实现混合检索，但需要自己处理多路召回和结果融合。
第三是部署约束。如果必须私有化部署，只能选择开源方案（Milvus、Qdrant、Weaviate、Chroma、pgvector）；如果能接受云服务，Pinecone和Chroma Cloud能省去运维麻烦。
最后是成本考量。开源方案虽然软件免费，但需要投入服务器和运维成本；托管服务按量付费，长期使用成本可能更高。需要根据实际调用量和数据规模做成本测算。
5.4 代码实现
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
import numpy as np
class VectorStore(ABC):
"""向量数据库基类"""
@abstractmethod
def create_collection(self, name: str, dimension: int, **kwargs):
pass
@abstractmethod
def insert(self, collection: str, vectors: np.ndarray, metadata: List[Dict]):
pass
@abstractmethod
def search(
self,
collection: str,
query_vector: np.ndarray,
top_k: int = 10,
filter: Dict = None
) -> List[Dict]:
pass
class MilvusVectorStore(VectorStore):
"""
Milvus 向量数据库
优点：
- 支持十亿级向量
- 原生混合检索（稀疏+密集）
- GPU 加速支持
- 分布式部署
- 多种索引类型
缺点：
- 部署复杂
- 运维成本高
- 学习曲线陡
索引选择：
- HNSW: 高召回，高内存
- IVF_FLAT: 平衡选择
- IVF_PQ: 低内存，有损
- GPU_CAGRA: GPU 加速
适用场景：
- 大规模生产环境
- 需要分布式部署
- 混合检索需求
"""
def __init__(self, uri: str = "http://localhost:19530"):
from pymilvus import MilvusClient
self.client = MilvusClient(uri=uri)
def create_collection(
self,
name: str,
dimension: int,
metric_type: str = "COSINE",
index_type: str = "HNSW",
**kwargs
):
if self.client.has_collection(name):
self.client.drop_collection(name)
self.client.create_collection(
collection_name=name,
dimension=dimension,
metric_type=metric_type,
auto_id=False,
**kwargs
)
def insert(
self,
collection: str,
vectors: np.ndarray,
metadata: List[Dict]
):
data = [
{"id": i, "vector": vectors[i].tolist(), **metadata[i]}
for i in range(len(vectors))
]
self.client.insert(collection_name=collection, data=data)
def search(
self,
collection: str,
query_vector: np.ndarray,
top_k: int = 10,
filter: Dict = None
) -> List[Dict]:
results = self.client.search(
collection_name=collection,
data=[query_vector.tolist()],
limit=top_k,
filter=filter
)
return results[0]
class QdrantVectorStore(VectorStore):
"""
Qdrant 向量数据库
优点：
- Rust 实现，性能优秀
- 内存效率高
- 强大的过滤能力
- 简单易用
- 成本低
缺点：
- 大规模需要分片
- 生态不如 Milvus
适用场景：
- 中等规模生产
- 性能敏感场景
- 成本敏感项目
- 需要复杂过滤
"""
def __init__(self, url: str = "http://localhost:6333"):
from qdrant_client import QdrantClient
self.client = QdrantClient(url=url)
def create_collection(
self,
name: str,
dimension: int,
metric_type: str = "Cosine",
**kwargs
):
from qdrant_client.models import Distance, VectorParams
distance_map = {
"COSINE": Distance.COSINE,
"L2": Distance.EUCLID,
"IP": Distance.DOT
}
self.client.recreate_collection(
collection_name=name,
vectors_config=VectorParams(
size=dimension,
distance=distance_map.get(metric_type, Distance.COSINE)
)
)
def insert(
self,
collection: str,
vectors: np.ndarray,
metadata: List[Dict]
):
from qdrant_client.models import PointStruct
points = [
PointStruct(id=i, vector=vectors[i].tolist(), payload=metadata[i])
for i in range(len(vectors))
]
self.client.upsert(collection_name=collection, points=points)
def search(
self,
collection: str,
query_vector: np.ndarray,
top_k: int = 10,
filter: Dict = None
) -> List[Dict]:
results = self.client.search(
collection_name=collection,
query_vector=query_vector.tolist(),
limit=top_k,
query_filter=filter
)
return [
{"id": r.id, "score": r.score, "payload": r.payload}
for r in results
]
class WeaviateVectorStore(VectorStore):
"""
Weaviate 向量数据库
优点：
- 混合检索最优秀
- GraphQL API
- 内置向量化模块
- 多租户支持
- 语义理解强
缺点：
- 资源占用较高
- 学习曲线
- 大规模成本高
适用场景：
- 混合检索需求强
- 多租户应用
- 需要内置向量化
"""
def __init__(self, url: str = "http://localhost:8080"):
import weaviate
self.client = weaviate.Client(url)
def create_collection(
self,
name: str,
dimension: int = None,
**kwargs
):
class_obj = {
"class": name,
"vectorizer": "none", # 我们自己提供向量
}
self.client.schema.create_class(class_obj)
def insert(
self,
collection: str,
vectors: np.ndarray,
metadata: List[Dict]
):
with self.client.batch as batch:
for i, (vector, meta) in enumerate(zip(vectors, metadata)):
batch.add_data_object(
data_object=meta,
class_name=collection,
uuid=str(i),
vector=vector.tolist()
)
def search(
self,
collection: str,
query_vector: np.ndarray,
top_k: int = 10,
filter: Dict = None
) -> List[Dict]:
query = (
self.client.query
.get(collection, ["content"])
.with_near_vector({"vector": query_vector.tolist()})
.with_limit(top_k)
.with_additional(["distance", "id"])
)
if filter:
query = query.with_where(filter)
results = query.do()
return results["data"]["Get"][collection]
class ChromaVectorStore(VectorStore):
"""
Chroma 向量数据库
优点：
- 零配置启动
- Python 原生
- 嵌入式部署
- 完全免费
- 开发友好
缺点：
- 不适合大规模
- 性能有限
- 功能较少
适用场景：
- 原型开发
- 本地测试
- 小规模应用
- 学习研究
"""
def __init__(self, persist_directory: str = "./chroma_db"):
import chromadb
self.client = chromadb.PersistentClient(path=persist_directory)
self.collections = {}
def create_collection(self, name: str, dimension: int = None, **kwargs):
self.collections[name] = self.client.get_or_create_collection(name)
def insert(
self,
collection: str,
vectors: np.ndarray,
metadata: List[Dict]
):
self.collections[collection].add(
embeddings=vectors.tolist(),
metadatas=metadata,
ids=[str(i) for i in range(len(vectors))]
)
def search(
self,
collection: str,
query_vector: np.ndarray,
top_k: int = 10,
filter: Dict = None
) -> List[Dict]:
results = self.collections[collection].query(
query_embeddings=[query_vector.tolist()],
n_results=top_k,
where=filter
)
return [
{"id": id, "score": 1 - dist, "payload": meta}
for id, dist, meta in zip(
results['ids'][0],
results['distances'][0],
results['metadatas'][0]
)
]
5.3 向量数据库选型决策
六、节点五：检索策略 (Retrieval Strategy)
检索策略决定了如何从知识库中找到相关文档。不同的检索策略有不同的适用场景，选型错误会导致召回不足或噪声过多。理解每种策略的原理和适用场景，是构建高质量RAG系统的关键。
6.1 检索策略对比
| 策略 | 召回率 | 精确率 | 延迟 | 复杂度 | 适用场景 |
|---|---|---|---|---|---|
| 纯向量检索 | 中 | 高 | 低 | 低 | 语义匹配 |
| 纯关键词检索(BM25) | 中 | 中 | 低 | 低 | 精确匹配 |
| 混合检索 | 高 | 高 | 中 | 中 | 通用场景 |
| 多查询检索 | 最高 | 中 | 高 | 高 | 模糊查询 |
| HyDE | 高 | 高 | 高 | 高 | 概念查询 |
6.2 检索策略详细分析
纯向量检索（Dense Retrieval） 是目前RAG系统的主流选择。它将查询和文档都编码为高维向量，通过向量相似度（通常是余弦相似度或欧氏距离）来衡量相关性。向量检索的优势在于语义理解——用户问"如何提高销售业绩"，能找到"提升营收的方法"这类语义相近但词汇不同的文档。它还支持跨语言检索，用中文查询可以找到英文文档。但向量检索也有明显的短板：对精确匹配（如产品型号、人名、代码片段）效果不佳，对专业术语的理解依赖嵌入模型的训练数据，而且嵌入向量一旦生成就固定了，无法动态调整。
关键词检索（BM25） 是传统信息检索的经典算法，在搜索引擎领域应用了几十年。它基于词频和逆文档频率计算相关性，对精确匹配非常有效。用户搜索"iPhone 15 Pro Max"，BM25能精准找到包含这个确切词组的文档。BM25的优势还包括可解释性强（你可以看到哪些词贡献了相关性分数）、无需训练模型、对专业术语天然支持。但它的缺点同样明显：无法理解语义——"苹果手机"和"iPhone"对BM25来说是两个完全不同的词，无法处理同义词，对长查询效果差。
混合检索（Hybrid Retrieval） 结合了向量检索和关键词检索的优势，是目前生产环境推荐的默认选择。它同时进行两路检索：向量检索负责语义匹配，关键词检索负责精确匹配，然后将两路结果融合。融合策略通常使用RRF（倒数排名融合）或加权融合。混合检索的召回率最高，因为它不会遗漏任何一路能找到的相关文档。代价是计算成本增加（需要维护两套索引），延迟略高（两路检索并行或串行），实现复杂度也更高。但对于大多数知识库场景，混合检索的收益远大于成本。
多查询检索（Multi-Query Retrieval） 是为了解决用户查询模糊的问题。很多时候，用户不知道如何准确描述自己的需求，或者使用的词汇与文档中的表述不一致。多查询检索的做法是：先用LLM将原始查询改写成多个语义相近但表述不同的查询，然后对每个查询分别检索，最后合并结果。比如用户问"怎么修电脑"，系统可能生成"计算机故障排除方法"、“PC维修指南”、"电脑问题解决方案"等多个查询，大大提高召回覆盖面。多查询检索的代价是LLM调用成本和延迟增加，适合查询模糊、召回要求高的场景。
HyDE（Hypothetical Document Embeddings） 是一种巧妙的方法，专门解决概念性查询的问题。它的思路是：很多用户查询是概念性的，比如"如何做好项目管理"，直接用这个查询去检索可能效果不好。HyDE的做法是先让LLM生成一个"假设文档"——即假设存在一个完美回答这个问题的文档，然后对这个假设文档进行嵌入，用假设文档的向量去检索真实文档。这种方法在概念性查询场景下效果显著，但增加了LLM调用成本和延迟。
查询路由（Query Routing） 是一种更高级的策略，根据查询类型选择不同的检索路径。比如，识别到查询是关于代码的，就路由到代码检索分支；识别到是关于概念的，就路由到语义检索分支。查询路由需要先对查询进行分类，然后选择最合适的检索策略。这种方法在多领域知识库中效果很好，但需要维护多套检索系统。
6.3 检索策略选型建议
对于通用知识库，混合检索是默认选择，它在召回率和精确率之间取得了最佳平衡。
如果知识库主要是技术文档、代码、API文档，关键词检索可能比向量检索更重要，可以考虑提高BM25的权重。
如果用户查询普遍模糊，或者知识库覆盖多个领域，多查询检索能显著提高召回率。
如果查询主要是概念性的，比如"如何提升团队效率"、“最佳实践是什么”，HyDE值得尝试。
如果资源有限，纯向量检索是最简单的起点，后续可以根据效果逐步引入混合检索。
6.4 代码实现
from abc import ABC, abstractmethod
from typing import List, Tuple, Dict
import numpy as np
class RetrievalStrategy(ABC):
"""检索策略基类"""
@abstractmethod
def retrieve(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
pass
class DenseRetrieval(RetrievalStrategy):
"""
纯向量检索
优点：
- 语义理解能力强
- 支持模糊匹配
- 跨语言检索
- 向量可预计算
缺点：
- 精确匹配能力弱
- 专业术语效果差
- 对嵌入模型依赖
适用场景：
- 语义相似性检索
- 跨语言检索
- 概念性查询
"""
def __init__(self, embedding_model, vector_store):
self.embedding_model = embedding_model
self.vector_store = vector_store
def retrieve(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
query_vector = self.embedding_model.encode([query])[0]
results = self.vector_store.search(
collection="documents",
query_vector=query_vector,
top_k=top_k
)
return [(r['payload']['content'], r['score']) for r in results]
class SparseRetrieval(RetrievalStrategy):
"""
关键词检索 (BM25)
优点：
- 精确匹配能力强
- 无需训练模型
- 可解释性强
- 专业术语效果好
缺点：
- 无语义理解
- 词汇鸿沟问题
- 对长查询效果差
适用场景：
- 精确关键词检索
- 专业术语检索
- 代码/ID 检索
"""
def __init__(self, documents: List[str]):
self.documents = documents
self._build_index()
def _build_index(self):
"""构建 BM25 索引"""
# 简化实现，实际使用 rank_bm25 库
from rank_bm25 import BM25Okapi
tokenized_docs = [doc.split() for doc in self.documents]
self.bm25 = BM25Okapi(tokenized_docs)
def retrieve(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
tokenized_query = query.split()
scores = self.bm25.get_scores(tokenized_query)
top_indices = np.argsort(scores)[::-1][:top_k]
return [(self.documents[i], scores[i]) for i in top_indices]
class HybridRetrieval(RetrievalStrategy):
"""
混合检索
优点：
- 兼顾语义和精确匹配
- 召回率最高
- 适用场景广
缺点：
- 计算成本高
- 需要融合策略
- 调参复杂
适用场景：
- 通用知识库
- 企业搜索
- RAG 系统
"""
def __init__(
self,
dense_retriever,
sparse_retriever,
fusion_method: str = "rrf",
rrf_k: int = 60,
dense_weight: float = 0.5
):
self.dense_retriever = dense_retriever
self.sparse_retriever = sparse_retriever
self.fusion_method = fusion_method
self.rrf_k = rrf_k
self.dense_weight = dense_weight
def retrieve(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
# 向量检索
dense_results = self.dense_retriever.retrieve(query, top_k * 2)
# 关键词检索
sparse_results = self.sparse_retriever.retrieve(query, top_k * 2)
# 融合
if self.fusion_method == "rrf":
return self._rrf_fusion(dense_results, sparse_results, top_k)
else:
return self._weighted_fusion(dense_results, sparse_results, top_k)
def _rrf_fusion(
self,
dense_results: List[Tuple[str, float]],
sparse_results: List[Tuple[str, float]],
top_k: int
) -> List[Tuple[str, float]]:
"""RRF 融合"""
rrf_scores: Dict[str, float] = {}
for rank, (doc, _) in enumerate(dense_results, 1):
rrf_scores[doc] = rrf_scores.get(doc, 0) + 1 / (self.rrf_k + rank)
for rank, (doc, _) in enumerate(sparse_results, 1):
rrf_scores[doc] = rrf_scores.get(doc, 0) + 1 / (self.rrf_k + rank)
sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
return sorted_results[:top_k]
def _weighted_fusion(
self,
dense_results: List[Tuple[str, float]],
sparse_results: List[Tuple[str, float]],
top_k: int
) -> List[Tuple[str, float]]:
"""加权融合"""
# 归一化分数
def normalize(results):
if not results:
return {}
scores = [s for _, s in results]
min_s, max_s = min(scores), max(scores)
if max_s == min_s:
return {doc: 1.0 for doc, _ in results}
return {doc: (s - min_s) / (max_s - min_s) for doc, s in results}
dense_norm = normalize(dense_results)
sparse_norm = normalize(sparse_results)
all_docs = set(dense_norm.keys()) | set(sparse_norm.keys())
final_scores = {}
for doc in all_docs:
d_score = dense_norm.get(doc, 0)
s_score = sparse_norm.get(doc, 0)
final_scores[doc] = self.dense_weight * d_score + (1 - self.dense_weight) * s_score
sorted_results = sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
return sorted_results[:top_k]
class MultiQueryRetrieval(RetrievalStrategy):
"""
多查询检索
优点：
- 提高召回覆盖面
- 处理模糊查询
- 减少漏召回
缺点：
- 增加 LLM 调用成本
- 延迟增加
- 可能引入噪声
适用场景：
- 用户查询模糊
- 需要高召回率
- 复杂语义查询
"""
def __init__(self, base_retriever, llm_client, num_queries: int = 3):
self.base_retriever = base_retriever
self.llm_client = llm_client
self.num_queries = num_queries
def retrieve(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
# 生成查询变体
queries = self._generate_queries(query)
# 对每个查询召回
all_results = {}
for q in queries:
results = self.base_retriever.retrieve(q, top_k * 2)
for doc, score in results:
if doc not in all_results:
all_results[doc] = []
all_results[doc].append(score)
# 合并分数
final_scores = {
doc: max(scores)
for doc, scores in all_results.items()
}
sorted_results = sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
return sorted_results[:top_k]
def _generate_queries(self, original_query: str) -> List[str]:
prompt = f"""Generate {self.num_queries} different versions of the following query to improve retrieval:
Original query: {original_query}
Generate {self.num_queries} semantically similar but differently phrased queries:"""
response = self.llm_client.generate(prompt)
queries = [q.strip() for q in response.strip().split('\n') if q.strip()]
return [original_query] + queries[:self.num_queries]
七、节点六：融合策略 (Fusion Strategy)
当使用混合检索或多查询检索时，会得到多路检索结果，如何将这些结果合并成一个最终排序列表？这就是融合策略要解决的问题。融合策略的选择直接影响最终检索效果，一个好的融合策略能让多路检索的优势真正发挥出来。
7.1 融合策略对比
| 策略 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| RRF | 基于排名倒数融合 | 无需调参、鲁棒性强 | 忽略分数信息 | 通用场景 |
| 加权融合 | 分数加权求和 | 可控权重、可解释 | 需要归一化、需调参 | 有明确优先级 |
| 学习排序 | 机器学习模型 | 性能最优 | 需要标注数据 | 有标注数据 |
7.2 融合策略详细分析
RRF（Reciprocal Rank Fusion，倒数排名融合） 是目前最流行的融合方法，原因很简单：它几乎不需要调参，效果却出奇地好。RRF的核心思想是：不关心文档的具体分数，只关心它的排名位置。公式是 RRF(d) = Σ 1/(k + rank(d))，其中k是一个常数（通常设为60）。一个文档在每路结果中的排名越靠前，它获得的RRF分数就越高。比如，一个文档在向量检索中排第1，在BM25中排第3，它的RRF分数就是 1/(60+1) + 1/(60+3) ≈ 0.031。RRF的优势在于它天然处理了不同检索系统分数尺度不一致的问题——向量检索的相似度分数可能在0.7-0.9之间，BM25的分数可能在10-30之间，直接加权融合会很麻烦，但RRF完全不受影响。RRF对异常值也很鲁棒，一个文档在一路检索中排名很差不会对最终结果产生太大影响。缺点是它完全忽略了分数信息，只看排名，可能丢失一些有价值的信号。
加权融合（Weighted Fusion） 是更直观的方法：对每路检索的分数进行归一化，然后按权重加权求和。公式是 Score(d) = Σ w_i * normalize(S_i(d))。加权融合的优势在于可解释性和可控性——你可以明确地说"向量检索权重0.7，BM25权重0.3"，如果发现BM25效果更好，可以动态调整权重。但加权融合有一个棘手的问题：分数归一化。不同检索系统的分数分布差异很大，简单的min-max归一化可能不够好，需要根据实际数据调整归一化策略。另外，权重的设置需要实验调优，没有通用的最佳值。
学习排序（Learning to Rank） 是用机器学习模型来学习最优的融合策略。你可以提取各种特征（文档在每路检索中的排名、分数、文档本身的特征等），训练一个排序模型（如LambdaMART、RankNet）来预测最终的相关性。学习排序在理论上能达到最优效果，但代价是需要标注数据——你需要人工标注一批查询-文档对的相关性标签。对于有标注数据的场景，学习排序是值得投入的方向；对于大多数没有标注数据的场景，RRF是更务实的选择。
7.3 融合策略选型建议
对于大多数场景，RRF是首选。它简单、有效、无需调参，k=60在几乎所有场景下都表现良好。
如果你明确知道某路检索更重要（比如向量检索效果明显好于BM25），可以使用加权融合，给向量检索更高的权重。
如果你有标注数据，或者愿意投入资源标注数据，学习排序能带来额外的性能提升。
实际项目中，建议先用RRF建立基线，然后根据效果决定是否需要更复杂的融合策略。
7.4 代码实现
class FusionStrategy(ABC):
"""融合策略基类"""
@abstractmethod
def fuse(
self,
results_list: List[List[Tuple[str, float]]]
) -> List[Tuple[str, float]]:
pass
class RRFFusion(FusionStrategy):
"""
倒数排名融合 (Reciprocal Rank Fusion)
公式：RRF(d) = Σ 1/(k + rank(d))
优点：
- 无需分数归一化
- 参数少（k=60 通用）
- 对异常值鲁棒
- 实现简单
缺点：
- 忽略分数信息
- 只考虑排名
最佳实践：
- k=60 在大多数场景下表现良好
适用场景：
- 不同检索系统分数尺度差异大
- 需要快速部署
- 多路检索结果融合
"""
def __init__(self, k: int = 60):
self.k = k
def fuse(
self,
results_list: List[List[Tuple[str, float]]]
) -> List[Tuple[str, float]]:
rrf_scores: Dict[str, float] = {}
for results in results_list:
for rank, (doc, _) in enumerate(results, 1):
rrf_scores[doc] = rrf_scores.get(doc, 0) + 1.0 / (self.k + rank)
return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
class WeightedFusion(FusionStrategy):
"""
加权融合
公式：Score(d) = Σ w_i * normalize(S_i(d))
优点：
- 可精确控制权重
- 支持动态调整
- 可解释性强
缺点：
- 需要分数归一化
- 权重需要调参
- 对异常分数敏感
最佳实践：
- 使用 minmax 或 softmax 归一化
- 根据业务优先级设置权重
适用场景：
- 明确知道哪路检索更重要
- 需要动态调整权重
- 有足够数据调参
"""
def __init__(
self,
weights: List[float] = None,
normalize_method: str = "minmax"
):
self.weights = weights or [0.5, 0.5]
self.normalize_method = normalize_method
def fuse(
self,
results_list: List[List[Tuple[str, float]]]
) -> List[Tuple[str, float]]:
# 归一化每路分数
normalized_results = []
for results in results_list:
if not results:
normalized_results.append({})
continue
docs, scores = zip(*results)
normalized_scores = self._normalize(list(scores))
normalized_results.append(dict(zip(docs, normalized_scores)))
# 加权融合
all_docs = set()
for results in normalized_results:
all_docs.update(results.keys())
final_scores = {}
for doc in all_docs:
score = 0.0
for i, results in enumerate(normalized_results):
weight = self.weights[i] if i < len(self.weights) else 0.0
score += weight * results.get(doc, 0.0)
final_scores[doc] = score
return sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
def _normalize(self, scores: List[float]) -> List[float]:
scores = np.array(scores)
if self.normalize_method == "minmax":
min_s, max_s = scores.min(), scores.max()
if max_s == min_s:
return [1.0] * len(scores)
return ((scores - min_s) / (max_s - min_s)).tolist()
elif self.normalize_method == "softmax":
exp_scores = np.exp(scores - scores.max())
return (exp_scores / exp_scores.sum()).tolist()
return scores.tolist()
八、节点七：重排序 (Reranking)
重排序是RAG系统中提升检索精度的关键环节。召回阶段返回的候选文档往往存在噪声，重排序模型通过更精细的语义匹配，将最相关的文档排在前面。选型时需要在精度、速度和成本之间找到平衡。
8.1 重排序模型架构解析
重排序模型的核心区别在于Query和Document的交互方式，这直接决定了精度和效率的权衡。
Cross-Encoder（交叉编码器） 是最精准的架构。它将Query和Document拼接后一起输入模型，通过深度自注意力机制实现两者的充分交互。这种"面对面"的交互方式能够捕获最细微的语义关联，比如"苹果"在科技语境和水果语境下的区别。代价是计算成本高昂——每个Query-Document对都需要完整的前向传播，无法预计算Document表示。在实际应用中，Cross-Encoder通常只对召回阶段返回的Top 20-50个候选进行重排序，否则延迟会难以接受。
Bi-Encoder（双编码器） 则是效率优先的选择。Query和Document分别独立编码为向量，通过向量相似度计算相关性。这种架构允许预先计算并存储所有Document的向量，查询时只需编码Query向量然后做向量检索。但独立编码意味着Query和Document之间没有深度交互，精度自然不如Cross-Encoder。Bi-Encoder更适合召回阶段，而非精排阶段。
Late Interaction（迟交互） 架构试图在精度和效率之间找到中间地带。ColBERT是典型代表，它为每个Token生成向量表示，Query和Document的匹配通过Token级别的MaxSim操作实现。这种方式保留了细粒度的匹配能力，同时Document的Token向量可以预计算。ColBERT的存储成本较高（每个Token一个向量），但查询效率远优于Cross-Encoder。
LLM Reranker利用大语言模型的理解能力进行重排序。可以通过Pointwise方式（让LLM为每个文档打分）、Listwise方式（让LLM直接排序）或Pairwise方式（让LLM比较文档对）实现。LLM的优势在于能够理解复杂的语义关系和用户意图，比如"性价比高的手机"这种需要推理的查询。但成本和延迟是最大的障碍——每次重排序都需要多次LLM调用，适合候选数很少且对精度要求极高的场景。
8.2 国际主流重排序模型
Cohere Rerank是商业重排序服务的标杆。最新的Rerank v3.5支持4096 token的长文档，能够处理半结构化数据（JSON），在多语言场景下表现优异。Cohere的独特优势在于其企业级SLA和持续优化——模型会定期更新，用户无需关心模型选择和部署。价格按次计费，对于中等规模的业务场景成本可控。
Jina Reranker v2是开源领域的明星产品。它针对Agent RAG场景优化，支持函数调用和代码搜索，处理速度比前代快6倍。Jina Reranker v2支持100多种语言，在跨语言检索场景下表现出色。作为开源模型，它可以完全私有化部署，满足数据隐私要求。
MonoT5和RankT5是基于T5的重排序模型，采用Seq2Seq架构生成相关性分数。这类模型在学术基准测试中表现稳定，社区支持完善，是研究和实验的好选择。
8.3 国内开源重排序模型
国内重排序模型的发展同样令人瞩目，在多语言和中文场景下已经达到国际领先水平。
BGE-Reranker系列是北京智源研究院推出的重排序模型，与BGE-Embedding形成完整的技术栈。bge-reranker-v2-m3是轻量级多语言版本，基于XLM-RoBERTa架构，参数量适中，推理速度快（在MacBook M1上仅需0.5秒），同时支持100多种语言。对于大多数生产场景，v2-m3是性价比最高的选择。
bge-reranker-v2-gemma是基于Gemma-2B的更大版本，精度更高但推理速度明显下降（相同硬件上需要20-30秒）。如果你的场景对精度要求极高且能接受较高延迟，可以考虑这个版本。智源还发布了bge-reranker-v2.5-gemma2-lightweight，通过Token压缩和层级轻量化操作，在保持较好性能的同时显著降低资源消耗。
BCE-Reranker（BCE: Bilingual and Cross-lingual Embedding） 是网易有道推出的重排序模型，专注于中英双语场景。bce-reranker-base_v1在中英文检索任务上表现出色，特别适合中英混合的知识库。模型相对轻量，部署成本低。
Qwen3-Reranker是阿里云与Qwen-3-Embedding同期发布的重排序模型，同样支持指令感知。它基于Qwen-3语言模型微调，继承了Qwen系列强大的语义理解能力。在MTEB重排序基准测试中，Qwen3-Reranker-0.6B超越了此前的顶尖模型，展现了小参数模型的潜力。对于已经使用Qwen生态的团队，Qwen3-Reranker是自然的选择。
GTE-Reranker是阿里巴巴GTE系列的重排序版本，与GTE-Embedding配套使用。在中英文场景下都有不错的表现，且与阿里云的云服务集成便利。
8.4 重排序模型选型对比
从精度角度看，基于大语言模型的Reranker（如bge-reranker-v2-gemma、Qwen3-Reranker）> Cross-Encoder（如bge-reranker-v2-m3、BCE-Reranker）> Bi-Encoder。但精度不是唯一考量。
从速度角度看，轻量级Cross-Encoder（bge-reranker-v2-m3）可以在毫秒级完成重排序，而基于LLM的Reranker可能需要数秒。对于实时性要求高的场景（如在线客服），速度往往是硬约束。
从语言支持看，BGE-M3、Qwen3-Reranker、Jina Reranker v2都支持100多种语言；BCE-Reranker在中英双语场景优化；如果主要是英文场景，Cohere Rerank和MonoT5也是好选择。
从部署成本看，开源模型（BGE、BCE、Qwen3、Jina）免费但需要GPU资源；商业API（Cohere）按量付费，省去运维成本。对于日调用量大的场景，自建服务的成本优势明显。
8.5 重排序模型代码实现
class Reranker(ABC):
"""重排序器基类"""
@abstractmethod
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10
) -> List[Tuple[str, float]]:
pass
class BGEReranker(Reranker):
"""
BGE 重排序器（北京智源开源）
模型选择：
- bge-reranker-v2-m3: 轻量多语言，速度快，推荐首选
- bge-reranker-v2-gemma: 高精度，速度慢，适合离线场景
- bge-reranker-large: 第一代模型，稳定成熟
优点：
- 多语言支持（100+）
- 开源免费
- 社区活跃
- 与BGE-Embedding配套
缺点：
- 需要自行部署
- gemma版本需要较大显存
最佳实践：
- 候选数控制在 20-50
- 使用 GPU 加速
- 中文场景优先选择
适用场景：
- 召回后的精排
- 多语言知识库
- 私有化部署
"""
def __init__(
self,
model_name: str = "BAAI/bge-reranker-v2-m3",
device: str = "cuda"
):
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name, device=device)
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10
) -> List[Tuple[str, float]]:
pairs = [(query, doc) for doc in candidates]
scores = self.model.predict(pairs)
results = list(zip(candidates, scores))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
class BCEReranker(Reranker):
"""
BCE 重排序器（网易有道开源）
优点：
- 中英双语优化
- 轻量高效
- 开源免费
缺点：
- 语言支持有限
- 社区相对较小
适用场景：
- 中英混合知识库
- 资源受限环境
"""
def __init__(
self,
model_name: str = "maidalun1020/bce-reranker-base_v1",
device: str = "cuda"
):
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name, device=device)
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10
) -> List[Tuple[str, float]]:
pairs = [(query, doc) for doc in candidates]
scores = self.model.predict(pairs)
results = list(zip(candidates, scores))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
class Qwen3Reranker(Reranker):
"""
Qwen-3 重排序器（阿里云开源）
优点：
- MTEB重排序基准领先
- 支持指令感知
- 多语言支持
- 与Qwen-3-Embedding配套
缺点：
- 需要较大显存
- 相对较新
适用场景：
- 追求最高精度
- 已使用Qwen生态
- 多语言场景
"""
def __init__(
self,
model_name: str = "Qwen/Qwen3-Reranker-0.6B",
device: str = "cuda",
instruction: str = None
):
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name, device=device)
self.instruction = instruction
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10,
instruction: str = None
) -> List[Tuple[str, float]]:
instr = instruction or self.instruction
if instr:
pairs = [(f"{instr} [SEP] {query}", doc) for doc in candidates]
else:
pairs = [(query, doc) for doc in candidates]
scores = self.model.predict(pairs)
results = list(zip(candidates, scores))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
class CohereReranker(Reranker):
"""
Cohere 重排序器（商业API）
优点：
- 企业级SLA
- 持续优化更新
- 长文档支持（4096 tokens）
- 半结构化数据处理
缺点：
- 需要付费
- 数据隐私问题
- 网络延迟
适用场景：
- 企业级应用
- 快速集成
- 多语言场景
"""
def __init__(self, api_key: str, model_name: str = "rerank-v3.5"):
import cohere
self.client = cohere.Client(api_key)
self.model_name = model_name
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10
) -> List[Tuple[str, float]]:
response = self.client.rerank(
query=query,
documents=candidates,
model=self.model_name,
top_n=top_k
)
return [
(candidates[r.index], r.relevance_score)
for r in response.results
]
class LLMReranker(Reranker):
"""
LLM 重排序器
优点：
- 理解复杂语义
- 可解释性强
- 灵活性高
缺点：
- 成本最高
- 延迟最长
- 不稳定
适用场景：
- 复杂判断场景
- 需要解释性
- 候选数少
"""
def __init__(self, llm_client):
self.llm = llm_client
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 10
) -> List[Tuple[str, float]]:
prompt = f"""Rank the following documents by relevance to the query.
Query: {query}
Documents:
{chr(10).join([f'{i+1}. {doc[:200]}...' for i, doc in enumerate(candidates)])}
Return the document numbers in order of relevance (most relevant first), comma-separated:"""
response = self.llm.generate(prompt)
import re
numbers = re.findall(r'\d+', response)
results = []
for i, num in enumerate(numbers[:top_k]):
idx = int(num) - 1
if 0 <= idx < len(candidates):
results.append((candidates[idx], 1.0 / (i + 1)))
return results
九、节点八：上下文构建 (Context Building)
检索到的文档如何组织成LLM能够理解的上下文？这看似简单，实则大有学问。上下文构建的质量直接影响LLM的生成效果——太长会超出上下文窗口，太短会丢失关键信息，组织混乱会让LLM难以理解。
9.1 上下文构建策略
| 策略 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 简单拼接 | 直接拼接文档 | 简单无信息损失 | 可能超长 | 短文档 |
| 摘要压缩 | LLM压缩 | 节省token | 有信息损失 | 长文档 |
| 上下文注入 | 添加文档上下文 | 完整性高 | 增加长度 | 需要上下文 |
| 动态选择 | 根据相关性选择 | 高效 | 复杂 | 大规模 |
9.2 上下文构建策略详细分析
简单拼接是最直接的方法：把检索到的文档按顺序拼接起来，中间用分隔符隔开。它的优点是实现简单、无信息损失——检索到的内容原封不动地传给LLM。缺点也很明显：可能超出LLM的上下文窗口限制，或者虽然没超限但token消耗巨大。简单拼接适合文档较短、检索结果不多的场景。实际使用时，通常会设置一个max_tokens参数，超过就截断。但截断策略需要考虑：是从前往后截（保留前面的文档），还是从后往前截（保留后面的文档），还是按相关性分数截？通常按相关性分数排序后从前往后截是合理的选择。
摘要压缩是为了解决文档过长的问题。它的做法是用LLM对检索到的文档进行压缩，提取与查询相关的关键信息。比如，检索到10篇文档，每篇2000字，直接拼接就是20000字；用LLM压缩后，每篇可能只剩200字的关键信息，总共2000字。摘要压缩能大幅节省token，但代价是信息损失——LLM压缩过程中可能遗漏重要细节。另外，压缩本身需要LLM调用，增加了成本和延迟。摘要压缩适合文档很长、token预算有限的场景，但需要权衡信息完整性和成本。
上下文注入（Contextual Retrieval） 是Anthropic提出的一种方法，专门解决分块导致的上下文丢失问题。当文档被切成小块后，每个块可能失去原有的上下文——比如一个块是"该公司2023年营收增长50%“，但"该公司"是谁？上下文注入的做法是：在嵌入之前，先用LLM为每个块生成一段上下文描述，比如"这是关于苹果公司2023年财务报告的内容”，然后把这个上下文和原始块拼接在一起进行嵌入。检索时，用户能更准确地找到相关内容。上下文注入能显著提高检索精度，但需要在索引阶段做预处理，增加了初始构建成本。
动态选择是一种更智能的方法。它不只是简单拼接，而是根据查询的相关性动态决定包含哪些内容。比如，如果查询是"苹果公司的财务状况"，系统可能会优先选择财务相关的段落，跳过产品介绍的部分。动态选择可以基于相关性分数、文档类型、查询意图等多种因素。这种方法能更高效地利用有限的上下文窗口，但实现复杂度较高。
9.3 上下文构建最佳实践
控制上下文长度是关键。不同LLM有不同的上下文窗口限制：GPT-4 Turbo是128K tokens，Claude 3是200K tokens，但实际使用中不建议用满。过长的上下文会增加成本、降低响应速度，甚至影响LLM的理解能力。一般建议控制在4000-8000 tokens之间。
文档排序很重要。检索系统返回的结果通常已经按相关性排序，直接按这个顺序拼接是合理的。但有时也可以考虑其他因素：比如文档的新鲜度（优先使用更新的文档）、来源权威性（优先使用官方文档）等。
添加文档元信息能帮助LLM理解上下文。比如在每个文档前加上"[来源：产品手册，第3章]"这样的标注，让LLM知道信息的来源，生成回答时可以引用。
9.4 代码实现
class ContextBuilder(ABC):
"""上下文构建器基类"""
@abstractmethod
def build(
self,
query: str,
retrieved_docs: List[Tuple[str, float]],
max_tokens: int = 4000
) -> str:
pass
class SimpleContextBuilder(ContextBuilder):
"""
简单拼接
优点：
- 实现简单
- 无信息损失
缺点：
- 可能超出长度限制
- 无优先级
适用场景：
- 短文档
- 快速原型
"""
def build(
self,
query: str,
retrieved_docs: List[Tuple[str, float]],
max_tokens: int = 4000
) -> str:
context_parts = []
current_length = 0
for doc, score in retrieved_docs:
doc_length = len(doc.split()) # 简化估算
if current_length + doc_length > max_tokens:
break
context_parts.append(doc)
current_length += doc_length
return "\n\n".join(context_parts)
class ContextualContextBuilder(ContextBuilder):
"""
上下文注入（Anthropic 方法）
原理：在嵌入前为每个块添加文档上下文
优点：
- 提高检索精度
- 保留全局上下文
缺点：
- 增加 token 消耗
- 需要预处理
适用场景：
- 需要上下文的问答
- 长文档检索
"""
def __init__(self, llm_client):
self.llm = llm_client
def build(
self,
query: str,
retrieved_docs: List[Tuple[str, float]],
max_tokens: int = 4000
) -> str:
context_parts = []
for doc, score in retrieved_docs:
# 为每个文档生成上下文说明
context_explanation = self._generate_context(doc)
context_parts.append(f"Context: {context_explanation}\n\nContent: {doc}")
return "\n\n---\n\n".join(context_parts)
def _generate_context(self, doc: str) -> str:
prompt = f"""Provide a brief context (1-2 sentences) for the following document chunk:
{doc[:500]}
Context:"""
return self.llm.generate(prompt)
class CompressedContextBuilder(ContextBuilder):
"""
摘要压缩
优点：
- 节省 token
- 聚焦关键信息
缺点：
- 可能丢失细节
- 增加 LLM 调用
适用场景：
- 长文档
- token 预算有限
"""
def __init__(self, llm_client):
self.llm = llm_client
def build(
self,
query: str,
retrieved_docs: List[Tuple[str, float]],
max_tokens: int = 4000
) -> str:
# 合并所有文档
combined_docs = "\n\n".join([doc for doc, _ in retrieved_docs])
# 压缩
prompt = f"""Summarize the following documents to answer the query, keeping only relevant information.
Query: {query}
Documents:
{combined_docs}
Summarized context (keep under {max_tokens // 4} tokens):"""
return self.llm.generate(prompt)
十、完整 RAG 混合检索流水线
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
@dataclass
class RAGConfig:
"""RAG 配置"""
# 解析配置
parser_type: str = "unstructured"
enable_ocr: bool = False
# 分块配置
chunk_strategy: str = "recursive"
chunk_size: int = 512
chunk_overlap: int = 100
# 嵌入配置
embedding_provider: str = "bge"
embedding_model: str = "BAAI/bge-m3"
# 向量数据库配置
vector_store: str = "milvus"
collection_name: str = "knowledge_base"
# 检索配置
retrieval_strategy: str = "hybrid"
fusion_method: str = "rrf"
rrf_k: int = 60
# 重排序配置
enable_reranking: bool = True
reranker_model: str = "BAAI/bge-reranker-v2-m3"
rerank_candidates: int = 50
# 生成配置
max_context_tokens: int = 4000
top_k: int = 10
class RAGPipeline:
"""
RAG 混合检索完整流水线
流程：
1. 文档解析 → 2. 分块 → 3. 嵌入 → 4. 存储
5. 查询处理 → 6. 多路召回 → 7. 融合 → 8. 重排序 → 9. 上下文构建
"""
def __init__(self, config: RAGConfig):
self.config = config
self._setup_components()
def _setup_components(self):
"""初始化组件"""
# 解析器
self.parser = self._create_parser()
# 分块器
self.chunker = self._create_chunker()
# 嵌入模型
self.embedding_model = self._create_embedding_model()
# 向量数据库
self.vector_store = self._create_vector_store()
# 检索器
self.retriever = self._create_retriever()
# 重排序器
self.reranker = self._create_reranker() if self.config.enable_reranking else None
# 上下文构建器
self.context_builder = SimpleContextBuilder()
def ingest(self, file_paths: List[str], show_progress: bool = True):
"""
导入文档
流程：解析 → 分块 → 嵌入 → 存储
"""
all_chunks = []
all_metadata = []
for file_path in file_paths:
if show_progress:
print(f"Processing: {file_path}")
# 解析
docs = self.parser.parse(file_path)
# 分块
for doc in docs:
chunks = self.chunker.chunk(doc.content, doc.metadata)
all_chunks.extend([c.content for c in chunks])
all_metadata.extend([c.metadata for c in chunks])
if show_progress:
print(f"Total chunks: {len(all_chunks)}")
# 嵌入
if show_progress:
print("Generating embeddings...")
embeddings = self.embedding_model.encode(all_chunks)
# 存储
if show_progress:
print("Storing to vector database...")
self.vector_store.insert(
collection=self.config.collection_name,
vectors=embeddings,
metadata=[{'content': c, **m} for c, m in zip(all_chunks, all_metadata)]
)
if show_progress:
print("Done!")
def query(
self,
query: str,
top_k: int = None,
return_context: bool = True
) -> Dict[str, Any]:
"""
查询
流程：检索 → 融合 → 重排序 → 上下文构建
"""
top_k = top_k or self.config.top_k
# 检索
if self.config.retrieval_strategy == "hybrid":
results = self.retriever.retrieve(query, self.config.rerank_candidates)
else:
results = self.retriever.retrieve(query, top_k)
# 重排序
if self.reranker and len(results) > top_k:
candidates = [doc for doc, _ in results]
results = self.reranker.rerank(query, candidates, top_k)
# 构建上下文
context = None
if return_context:
context = self.context_builder.build(
query,
results,
self.config.max_context_tokens
)
return {
"query": query,
"results": results[:top_k],
"context": context
}
def _create_parser(self):
"""创建解析器"""
if self.config.parser_type == "unstructured":
return UnstructuredParser(enable_ocr=self.config.enable_ocr)
elif self.config.parser_type == "pdfplumber":
return PDFPlumberParser()
elif self.config.parser_type == "docling":
return DoclingParser()
else:
return PyPDF2Parser()
def _create_chunker(self):
"""创建分块器"""
if self.config.chunk_strategy == "recursive":
return RecursiveChunker(
self.config.chunk_size,
self.config.chunk_overlap
)
elif self.config.chunk_strategy == "semantic":
return SemanticChunker(
self.embedding_model,
breakpoint_threshold_type="percentile"
)
elif self.config.chunk_strategy == "hierarchical":
return HierarchicalChunker()
else:
return FixedSizeChunker(
self.config.chunk_size,
self.config.chunk_overlap
)
def _create_embedding_model(self):
"""创建嵌入模型"""
if self.config.embedding_provider == "openai":
return OpenAIEmbedding(model_name=self.config.embedding_model)
elif self.config.embedding_provider == "qwen3":
return Qwen3Embedding(model_name=self.config.embedding_model)
elif self.config.embedding_provider == "bge":
return BGEEmbedding(model_name=self.config.embedding_model)
elif self.config.embedding_provider == "cohere":
return CohereEmbedding(model_name=self.config.embedding_model)
def _create_vector_store(self):
"""创建向量数据库"""
if self.config.vector_store == "milvus":
return MilvusVectorStore()
elif self.config.vector_store == "qdrant":
return QdrantVectorStore()
elif self.config.vector_store == "weaviate":
return WeaviateVectorStore()
elif self.config.vector_store == "chroma":
return ChromaVectorStore()
def _create_retriever(self):
"""创建检索器"""
dense_retriever = DenseRetrieval(
self.embedding_model,
self.vector_store
)
if self.config.retrieval_strategy == "hybrid":
sparse_retriever = SparseRetrieval([])
return HybridRetrieval(
dense_retriever,
sparse_retriever,
fusion_method=self.config.fusion_method,
rrf_k=self.config.rrf_k
)
return dense_retriever
def _create_reranker(self):
"""创建重排序器"""
return BGEReranker(model_name=self.config.reranker_model)
# 使用示例
def demo_rag_pipeline():
"""RAG 流水线示例"""
config = RAGConfig(
parser_type="unstructured",
chunk_strategy="recursive",
chunk_size=512,
chunk_overlap=100,
embedding_provider="qwen3",
embedding_model="Qwen/Qwen3-Embedding-0.6B",
vector_store="milvus",
retrieval_strategy="hybrid",
fusion_method="rrf",
enable_reranking=True,
reranker_model="BAAI/bge-reranker-v2-m3",
top_k=10
)
pipeline = RAGPipeline(config)
pipeline.ingest([
"docs/product_manual.pdf",
"docs/faq.docx"
])
result = pipeline.query("如何重置密码？")
print(f"Query: {result['query']}")
print(f"Top results: {len(result['results'])}")
print(f"Context length: {len(result['context']) if result['context'] else 0}")
十一、选型决策总表
11.1 场景推荐配置
| 场景 | 解析器 | 分块 | 嵌入模型 | 向量数据库 | 检索策略 | 重排序 |
|---|---|---|---|---|---|---|
| 企业知识库（中文） | Unstructured | 递归(512) | Qwen3-Embedding-0.6B | Milvus | 混合+RRF | BGE-reranker-v2-m3 |
| 企业知识库（多语言） | Unstructured | 递归(512) | BGE-M3 | Milvus | 混合+RRF | BGE-reranker-v2-m3 |
| 电商搜索 | Docling | 语义 | OpenAI | Qdrant | 路由 | 可选 |
| 技术文档 | pdfplumber | 层次 | Qwen3-Embedding-0.6B | Weaviate | 混合 | BGE-reranker-v2-m3 |
| 客服问答 | Unstructured | 递归(256) | OpenAI-small | Chroma | 多查询 | 快速重排 |
| 法律检索 | Docling | 语义 | Cohere | Milvus | 混合 | LLM重排 |
| 原型开发 | PyPDF2 | 固定(512) | OpenAI-small | Chroma | 向量 | 无 |
| 私有化部署 | Unstructured | 递归(512) | Qwen3-Embedding-0.6B | Qdrant | 混合 | BCE-Reranker |
11.2 性能基准参考
| 配置 | 召回率@10 | NDCG@10 | 延迟 | 月成本(10万文档) |
|---|---|---|---|---|
| 基础配置 | 62% | 0.48 | 50ms | $50 |
| 标准配置 | 75% | 0.58 | 100ms | $200 |
| 高级配置 | 85% | 0.67 | 200ms | $500 |
| 私有化配置（国产模型） | 78% | 0.61 | 80ms | GPU成本 |
十二、总结
RAG混合检索系统的每个节点都有多种选型，关键在于根据实际场景做出合理决策：
文档解析的选择相对直观：表格密集型文档用pdfplumber，它能精准提取表格结构；通用文档用Unstructured，支持多格式且有OCR能力；复杂排版如学术论文用Docling，IBM开源的方案在布局理解上更胜一筹。
文档分块在生产环境中推荐递归分块作为默认选择，它在语义完整性和实现复杂度之间取得了良好平衡。如果对检索精度有极高要求，可以考虑语义分块，但需要承担更高的计算成本。层次分块（Parent-Child）适合需要完整上下文的场景，比如法律或医疗文档检索。
嵌入模型的选型在2025年有了更多优秀选择。国内开源模型已经达到国际领先水平：Qwen-3-Embedding-8B在MTEB多语言排行榜位居第一，0.6B版本在中文场景下性价比极高；BGE-M3的三合一架构（Dense+Sparse+Multi-vector）在混合检索场景中非常便利。如果选择商业API，OpenAI适合快速集成，Cohere适合企业级应用，Voyage适合追求极致精度。
向量数据库的选择主要看规模：十万级以下用Chroma或pgvector，百万级用Qdrant或Weaviate，十亿级用Milvus。混合检索需求强的话，Weaviate和Milvus的原生支持更好。
检索策略上，混合检索（Dense+Sparse）是通用场景的最佳选择，RRF融合简单有效。对于用户查询模糊的场景，多查询检索能显著提高召回覆盖面。
重排序模型方面，国内开源方案同样出色：BGE-reranker-v2-m3是生产环境的首选，速度快且多语言支持好；追求更高精度可以用bge-reranker-v2-gemma或Qwen3-Reranker；中英双语场景BCE-Reranker是轻量高效的选择。
上下文构建相对简单：短文档直接拼接，长文档考虑压缩或摘要。
最后，一个实用的建议：从简单配置开始，用真实数据验证效果，再根据瓶颈逐步优化。没有放之四海而皆准的最优配置，只有最适合你场景的配置。对于中文为主的知识库，强烈推荐尝试Qwen-3-Embedding配合BGE-Reranker的组合，这是目前国产模型中的黄金搭档。
