Embedding 嵌入与重排序模型完全教程
零基础系统学习文本Embedding与重排序，从原理到生产级语义搜索架构实战
目录
- 第一章：Embedding 嵌入模型基础原理
- 第二章：主流Embedding模型对比与选型
- 第三章：多模态Embedding
- 第四章：重排序（Reranking）模型
- 第五章：语义搜索管线构建
- 第六章：Embedding微调与优化
- 第七章：向量压缩与量化
- 第八章：生产级检索架构设计
- 第九章：实战项目一——企业级语义搜索系统
- 第十章：实战项目二——智能推荐系统
- 第十一章：常见问题与故障排查
- 附录：学习资源与参考文献
第一章：Embedding 嵌入模型基础原理
1.1 什么是文本Embedding
文本Embedding（文本嵌入）是将自然语言文本映射到连续向量空间的技术。简单来说，它把一段文字转换为一组固定长度的数字（向量），使得语义相似的文本在向量空间中距离更近，语义不同的文本距离更远。
这个过程可以形式化描述为：
f: Text → ℝ^d
其中 Text 是任意长度的文本，ℝ^d 是 d 维实数向量空间（通常 d = 384 ~ 4096）。
Embedding的价值在于：
- 语义理解：捕获文本的深层语义，而非表面词汇匹配
- 计算效率：向量之间的相似度计算（余弦相似度、点积等）非常高效
- 通用性：同一个Embedding模型可以服务搜索、分类、聚类、推荐等多种下游任务
- 可组合性：向量可以进行加减运算，实现语义算术
# 最简单的Embedding示例
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
texts = [
"如何训练一只小狗",
"怎样教狗狗基本指令",
"今天天气真不错",
]
embeddings = model.encode(texts)
print(f"向量维度: {embeddings.shape}") # (3, 512)
# 计算语义相似度
from sklearn.metrics.pairwise import cosine_similarity
sim_matrix = cosine_similarity(embeddings)
print(f"'训练小狗' vs '教狗狗指令': {sim_matrix[0][1]:.4f}") # 高相似度
print(f"'训练小狗' vs '今天天气': {sim_matrix[0][2]:.4f}") # 低相似度
1.2 从One-Hot到稠密向量：表示学习的演进
在Embedding出现之前，文本表示经历了多个发展阶段：
One-Hot编码：每个词用一个长度等于词表大小的向量表示，只有对应位置为1，其余为0。缺点是维度极高（词表通常有数万到数十万词）且无法表达词之间的语义关系。
TF-IDF：通过词频和逆文档频率为每个词赋予权重，生成稀疏向量。虽然考虑了词的重要性，但仍然无法捕获语义。
稠密Embedding：用低维（通常数百维）的连续向量表示文本，通过训练让语义相近的文本在向量空间中聚集。这是当前主流的表示方式。
# 不同文本表示方式的对比
# 1. One-Hot（示意）
vocab = ["我", "喜欢", "吃", "苹果", "香蕉"]
one_hot_apple = [0, 0, 0, 1, 0]
one_hot_banana = [0, 0, 0, 0, 1]
# 余弦相似度 = 0，无法反映"苹果"和"香蕉"都是水果
# 2. TF-IDF
from sklearn.feature_extraction.text import TfidfVectorizer
tfidf = TfidfVectorizer()
tfidf_matrix = tfidf.fit_transform(["我喜欢吃苹果", "我喜欢吃香蕉", "今天天气好"])
# 稀疏向量，维度等于词表大小
# 3. 稠密Embedding
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-small-zh-v1.5")
dense_emb = model.encode(["苹果", "香蕉"])
# 稠密向量，维度512，"苹果"和"香蕉"距离很近
1.3 Word2Vec与GloVe：经典词向量时代
Word2Vec（2013年，Mikolov等人）是第一个广泛成功的词向量模型，包含两种训练方式：
- CBOW（Continuous Bag of Words）：用上下文词预测中心词
- Skip-gram：用中心词预测上下文词
Word2Vec最著名的发现是词向量的线性代数性质：
vector("King") - vector("Man") + vector("Woman") ≈ vector("Queen")
GloVe（2014年，Pennington等人）结合了全局矩阵分解和局部上下文窗口的优点，通过词共现矩阵学习词向量。
这些经典方法的局限在于：每个词只有一个固定的向量表示，无法处理一词多义（如"苹果"可以是水果也可以是公司）。
1.4 Transformer时代的句子Embedding
BERT（2018年）的出现彻底改变了Embedding的格局。与Word2Vec不同，BERT生成的是上下文相关的（contextualized）词表示——同一个词在不同句子中会有不同的向量。
从BERT到句子Embedding，主要经历了以下技术路线：
1. 简单平均/CLS池化：直接取BERT最后一层所有token的平均值或[CLS] token的输出作为句子表示。效果一般，因为BERT的预训练目标（MLM/NSP）并非为生成良好句子Embedding设计。
2. Sentence-BERT（SBERT）：在BERT基础上加入siamese或triplet网络结构，使用余弦相似度目标进行微调，显著提升了句子Embedding的质量。
3. 对比学习方法：如SimCSE，通过对比正样本对和负样本对来训练Embedding模型。
# Sentence-BERT的核心思想
from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
# 加载预训练模型
model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
# 准备训练数据（句子对 + 相似度标签）
train_examples = [
InputExample(texts=["如何退款", "怎样申请退款"], label=0.9),
InputExample(texts=["如何退款", "今天天气"], label=0.1),
]
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
train_loss = losses.CosineSimilarityLoss(model)
# 微调
model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=3,
warmup_steps=100,
)
1.5 对比学习与Embedding训练范式
对比学习（Contrastive Learning）是当前训练Embedding模型的主流范式。其核心思想是：
- 拉近正样本对（语义相似的文本对）在向量空间中的距离
- 推远负样本对（语义不同的文本对）在向量空间中的距离
InfoNCE损失函数是对比学习中最常用的损失：
import torch
import torch.nn.functional as F
def info_nce_loss(embeddings_a, embeddings_b, temperature=0.05):
"""
embeddings_a: [batch_size, dim] 正样本对的A侧
embeddings_b: [batch_size, dim] 正样本对的B侧
"""
# 归一化
embeddings_a = F.normalize(embeddings_a, dim=1)
embeddings_b = F.normalize(embeddings_b, dim=1)
# 计算相似度矩阵
similarity_matrix = torch.matmul(embeddings_a, embeddings_b.T) / temperature
# 对角线是正样本对
labels = torch.arange(similarity_matrix.size(0), device=similarity_matrix.device)
# 交叉熵损失
loss = F.cross_entropy(similarity_matrix, labels)
return loss
训练数据的构造方式决定了Embedding模型的特性：
| 数据来源 | 代表方法 | 优势 |
|---|---|---|
| 自然句子对 | Wiki、书籍对齐 | 通用语义理解 |
| 查询-文档对 | 搜索日志 | 搜索场景优化 |
| 合成数据 | LLM生成 | 可扩展、可控 |
| 弱监督 | 标题-正文 | 无需标注 |
第二章：主流Embedding模型对比与选型
2.1 Cohere Embed v3 深度解析
Cohere Embed v3 是目前商业Embedding模型中的佼佼者，在MTEB排行榜上长期位居前列。其核心特性包括：
多语言支持：支持超过100种语言，在跨语言检索场景中表现出色。
多任务适配：通过 input_type 参数区分不同任务场景：
search_document：索引文档时使用search_query：搜索查询时使用classification：分类任务使用clustering：聚类任务使用
维度灵活性：支持 256、512、1024、2048 维输出，通过 embedding_types 参数控制。
import cohere
import numpy as np
co = cohere.Client("YOUR_API_KEY")
# 基础Embedding生成
texts = [
"机器学习是人工智能的子领域",
"深度学习属于机器学习的一种方法",
"今天股市大涨",
]
response = co.embed(
texts=texts,
model="embed-multilingual-v3.0",
input_type="search_document",
embedding_types=["float"],
truncate="END",
)
embeddings = np.array(response.embeddings.float)
print(f"维度: {embeddings.shape[1]}") # 1024
# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
sim = cosine_similarity(embeddings)
print(f"ML vs DL: {sim[0][1]:.4f}") # 高
print(f"ML vs 股市: {sim[0][2]:.4f}") # 低
# 搜索场景：查询用search_query，文档用search_document
query_response = co.embed(
texts=["什么是深度学习"],
model="embed-multilingual-v3.0",
input_type="search_query",
)
query_emb = np.array(query_response.embeddings.float)
doc_response = co.embed(
texts=[
"深度学习是机器学习的分支，使用多层神经网络进行特征学习。",
"今天是星期三，天气晴朗。",
],
model="embed-multilingual-v3.0",
input_type="search_document",
)
doc_embs = np.array(doc_response.embeddings.float)
# 查询与文档的相似度
scores = cosine_similarity(query_emb, doc_embs)[0]
print(f"深度学习文档得分: {scores[0]:.4f}") # 高分
print(f"天气文档得分: {scores[1]:.4f}") # 低分
Cohere Embed v3 的压缩模式：支持二值压缩（binary）和 int8 量化，可以将存储需求降低 8~32 倍，同时保持相对较好的检索质量。
# 二值压缩模式
response = co.embed(
texts=["示例文本"],
model="embed-multilingual-v3.0",
embedding_types=["binary"],
)
# binary embeddings 占用空间仅为 float 的 1/32
binary_emb = response.embeddings.binary[0]
print(f"二值向量长度: {len(binary_emb)}") # 128 bytes (1024 bits)
2.2 BGE系列开源模型全解
BGE（BAAI General Embedding）是由智源研究院（BAAI）开发的开源Embedding模型系列，是国内最广泛使用的Embedding模型之一。
BGE系列模型矩阵：
| 模型 | 参数量 | 维度 | 特点 |
|---|---|---|---|
| bge-small-zh-v1.5 | 24M | 512 | 轻量级，适合端侧 |
| bge-base-zh-v1.5 | 102M | 768 | 平衡性能与效果 |
| bge-large-zh-v1.5 | 326M | 1024 | 最佳效果 |
| bge-m3 | 568M | 1024 | 多语言、多粒度、多功能 |
| bge-multilingual-gemma2 | 9.2B | 3584 | 多语言大模型 |
BGE模型的训练特点：
- 指令前缀（Instruction Prefix）：BGE在编码查询时需要添加指令前缀来区分任务类型
- 难负样本挖掘：使用检索模型挖掘难负样本进行对比学习训练
- 多阶段训练：先在大规模弱监督数据上预训练，再在高质量标注数据上微调
from sentence_transformers import SentenceTransformer
import numpy as np
# 加载BGE模型
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
# BGE的查询需要添加指令前缀
query_instruction = "为这个句子生成表示以用于检索中文文档："
queries = [
query_instruction + "什么是深度学习",
query_instruction + "如何训练神经网络",
]
# 文档不需要指令前缀
documents = [
"深度学习是机器学习的一个子领域，它使用多层人工神经网络来学习数据的层次化表示。",
"神经网络的训练过程包括前向传播、计算损失、反向传播和参数更新四个步骤。",
"今天北京的天气预报显示有小雨。",
"Python是一种广泛使用的编程语言。",
]
query_embeddings = model.encode(queries, normalize_embeddings=True)
doc_embeddings = model.encode(documents, normalize_embeddings=True)
# 计算相似度
scores = np.dot(query_embeddings, doc_embeddings.T)
for i, q in enumerate(queries):
print(f"\n查询: {q.replace(query_instruction, '')}")
ranked = np.argsort(scores[i])[::-1]
for rank, idx in enumerate(ranked):
print(f" #{rank+1} [{scores[i][idx]:.4f}] {documents[idx][:40]}...")
BGE-M3 的三大特性：
BGE-M3 是BGE系列中功能最全面的模型，支持：
- 多语言（Multi-Linguality）：覆盖100+语言
- 多粒度（Multi-Granularity）：支持从短句到长文档（最高8192 tokens）
- 多功能（Multi-Functionality）：同时支持稠密检索、稀疏检索和多向量检索
from FlagEmbedding import BGEM3FlagModel
model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)
sentences = [
"什么是机器学习",
"机器学习是人工智能的一个分支，通过算法让计算机从数据中学习模式。",
"今天股市表现如何",
]
# 同时获取三种表示
output = model.encode(
sentences,
return_dense=True,
return_sparse=True,
return_colbert_vecs=True,
)
# 稠密向量（用于语义检索）
dense_vecs = output["dense_vecs"]
print(f"稠密向量维度: {dense_vecs.shape[1]}") # 1024
# 稀疏向量（用于关键词检索，类似BM25）
sparse_vecs = output["lexical_weights"]
print(f"稀疏向量示例: {list(sparse_vecs[0].items())[:5]}")
# ColBERT多向量（用于精细交互式检索）
colbert_vecs = output["colbert_vecs"]
print(f"ColBERT向量形状: {colbert_vecs.shape}") # [batch, seq_len, 1024]
2.3 E5系列与GTE系列模型
E5（EmbEdding from bidirEctional Encoder rEpresentations） 由微软研究院开发，其核心创新在于使用文本对的弱监督数据进行预训练：
# E5模型使用 "query: " 和 "passage: " 前缀区分任务
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("intfloat/multilingual-e5-large")
# E5的前缀约定
queries = ["query: 什么是自然语言处理"]
passages = [
"passage: 自然语言处理(NLP)是计算机科学和人工智能的子领域，专注于计算机与人类语言之间的交互。",
"passage: 机器学习是人工智能的一个分支。",
]
q_emb = model.encode(queries, normalize_embeddings=True)
p_emb = model.encode(passages, normalize_embeddings=True)
scores = (q_emb @ p_emb.T)[0]
print(f"NLP文档得分: {scores[0]:.4f}")
print(f"ML文档得分: {scores[1]:.4f}")
GTE（General Text Embeddings） 由阿里通义实验室开发：
- GTE-base：基于BERT架构，参数量118M
- GTE-large：基于BERT架构，参数量335M
- GTE-Qwen2：基于Qwen2大模型架构，参数量1.5B~7B
GTE-Qwen2系列的优势在于它基于大语言模型架构，天然具备更强的语言理解能力，特别适合长文本和复杂语义场景。
# GTE-Qwen2 使用方式
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("Alibaba-NLP/gte-Qwen2-1.5B-instruct", trust_remote_code=True)
# 支持instruction
queries = ["query: 深度学习的应用场景有哪些"]
documents = [
"passage: 深度学习广泛应用于计算机视觉、自然语言处理、语音识别等领域。",
"passage: 传统机器学习算法包括决策树、支持向量机和随机森林。",
]
q_emb = model.encode(queries)
d_emb = model.encode(documents)
similarity = (q_emb @ d_emb.T)[0]
print(f"深度学习文档: {similarity[0]:.4f}")
print(f"传统ML文档: {similarity[1]:.4f}")
2.4 MTEB排行榜与模型评估方法
MTEB（Massive Text Embedding Benchmark） 是评估Embedding模型最权威的基准测试，覆盖8个任务类型、58个数据集、112种语言：
| 任务类型 | 说明 | 评估指标 |
|---|---|---|
| Bitext Mining | 双语句对挖掘 | F1 |
| Classification | 文本分类 | Accuracy |
| Clustering | 文本聚类 | V-measure |
| Pair Classification | 句对分类 | Average Precision |
| Reranking | 重排序 | MAP |
| Retrieval | 检索 | NDCG@10 |
| STS | 语义文本相似度 | Spearman相关系数 |
| Summarization | 摘要评估 | Spearman相关系数 |
# 使用MTEB评估自定义模型
import mteb
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
# 选择要评估的任务
tasks = mteb.get_tasks(tasks=["CMTEBRetrieval"], languages=["zho"])
evaluation = mteb.MTEB(tasks=tasks)
results = evaluation.run(model, output_folder="results/bge-large-zh")
# results 包含每个任务的详细评估结果
for result in results:
print(f"任务: {result.task_name}")
print(f"分数: {result.scores}")
2.5 模型选型决策指南
选择Embedding模型时需要考虑以下因素：
1. 部署约束
资源受限（端侧/边缘）→ bge-small-zh (24M) / all-MiniLM-L6-v2 (22M)
中等资源（单GPU）→ bge-large-zh (326M) / e5-large (335M)
充裕资源（多GPU）→ GTE-Qwen2-7B (7B) / bge-multilingual-gemma2 (9.2B)
无GPU/不想自建 → Cohere Embed v3 API
2. 语言需求
仅中文 → bge-large-zh-v1.5
中英双语 → bge-large-zh-v1.5 / e5-large
多语言 → bge-m3 / Cohere Embed v3 / multilingual-e5-large
3. 文本长度
短文本 (< 512 tokens) → 大多数模型都支持
中等文本 (512-4096) → bge-m3 / GTE-Qwen2
长文本 (4096-8192) → bge-m3 / nomic-embed-text-v1.5
超长文本 (> 8192) → 需要分块策略
4. 任务类型
语义搜索 → 使用 "search_query" / "search_document" 前缀的模型
分类/聚类 → Cohere (classification/clustering input_type) / 通用模型
代码搜索 → codebert / unixcoder / CodeRankEmbed
第三章：多模态Embedding
3.1 多模态Embedding的原理
多模态Embedding将不同模态的数据（文本、图像、音频等）映射到同一个向量空间中，使得不同模态但语义相似的内容在向量空间中距离接近。
核心思想是通过对比学习训练不同模态的编码器，使得匹配的图文对在向量空间中靠近，不匹配的对远离。
3.2 CLIP与ALIGN：图文对齐模型
CLIP（Contrastive Language-Image Pre-training） 由OpenAI于2021年发布，使用4亿图文对进行训练，是多模态Embedding的里程碑模型。
CLIP包含两个编码器：
- 图像编码器：通常是Vision Transformer（ViT）或ResNet
- 文本编码器：Transformer
两个编码器分别将图像和文本映射到同一维度的向量空间，通过对比学习对齐。
import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
# 加载CLIP模型
model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")
# 图像编码
image = Image.open("example.jpg")
inputs = processor(images=image, return_tensors="pt")
image_features = model.get_image_features(**inputs)
image_features = image_features / image_features.norm(dim=-1, keepdim=True)
# 文本编码
texts = ["一只可爱的猫咪", "一辆红色的汽车", "日落风景"]
text_inputs = processor(text=texts, return_tensors="pt", padding=True)
text_features = model.get_text_features(**text_inputs)
text_features = text_features / text_features.norm(dim=-1, keepdim=True)
# 计算图文相似度
similarity = (image_features @ text_features.T)[0]
for text, score in zip(texts, similarity):
print(f" {text}: {score:.4f}")
3.3 多模态Embedding实战
以CLIP为基础构建图文检索系统：
import torch
import numpy as np
from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import os
class MultimodalSearchEngine:
def __init__(self, model_name="openai/clip-vit-large-patch14"):
self.model = CLIPModel.from_pretrained(model_name)
self.processor = CLIPProcessor.from_pretrained(model_name)
self.image_index = [] # 存储图像路径
self.image_embeddings = None
def index_images(self, image_paths: list):
"""批量索引图像"""
all_embeddings = []
for path in image_paths:
image = Image.open(path)
inputs = processor(images=image, return_tensors="pt")
with torch.no_grad():
emb = self.model.get_image_features(**inputs)
emb = emb / emb.norm(dim=-1, keepdim=True)
all_embeddings.append(emb.numpy())
self.image_index.append(path)
self.image_embeddings = np.vstack(all_embeddings)
print(f"已索引 {len(image_paths)} 张图像")
def search_by_text(self, query: str, top_k: int = 5):
"""用文本搜索图像"""
inputs = self.processor(text=[query], return_tensors="pt", padding=True)
with torch.no_grad():
query_emb = self.model.get_text_features(**inputs)
query_emb = query_emb / query_emb.norm(dim=-1, keepdim=True)
scores = (query_emb.numpy() @ self.image_embeddings.T)[0]
top_indices = np.argsort(scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"path": self.image_index[idx],
"score": float(scores[idx]),
})
return results
def search_by_image(self, image_path: str, top_k: int = 5):
"""用图搜图"""
image = Image.open(image_path)
inputs = self.processor(images=image, return_tensors="pt")
with torch.no_grad():
query_emb = self.model.get_image_features(**inputs)
query_emb = query_emb / query_emb.norm(dim=-1, keepdim=True)
scores = (query_emb.numpy() @ self.image_embeddings.T)[0]
top_indices = np.argsort(scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"path": self.image_index[idx],
"score": float(scores[idx]),
})
return results
# 使用示例
engine = MultimodalSearchEngine()
engine.index_images(["cat.jpg", "dog.jpg", "car.jpg", "sunset.jpg"])
# 文本搜图
results = engine.search_by_text("一只小猫在睡觉")
for r in results:
print(f" {r['path']}: {r['score']:.4f}")
第四章：重排序（Reranking）模型
4.1 为什么需要重排序
在检索系统中，重排序是两阶段检索架构的关键组件：
第一阶段（召回/粗排）: 从海量文档中快速筛选出候选集（如Top 100~1000）
第二阶段（重排/精排）: 对候选集进行精细排序，输出最终结果（如Top 10）
为什么需要两阶段？因为：
- 效率约束：交叉编码器对每个查询-文档对都要进行完整的前向推理，无法对百万级文档逐一计算
- 质量提升：重排序模型可以捕捉查询和文档之间的细粒度交互，显著提升排序质量
- 互补性：双塔模型（Embedding检索）擅长快速召回，交叉编码器擅长精细排序
典型的数据流：
用户查询 → Embedding模型编码 → 向量检索（Top 100）→ 重排序模型（Top 10）→ 返回结果
4.2 交叉编码器与双塔模型的本质区别
双塔模型（Bi-Encoder）：
Query → [Encoder_Q] → q_emb ─┐
├→ cosine_similarity → score
Doc → [Encoder_D] → d_emb ─┘
- 查询和文档独立编码，可以预计算文档向量
- 适合大规模检索（向量索引可以预建）
- 交互信息有限，精度相对较低
交叉编码器（Cross-Encoder）：
[Query, Doc] → [Encoder] → score
- 查询和文档拼接后一起输入模型
- 可以捕获查询和文档之间的细粒度交互（如token级别的注意力）
- 精度高，但计算成本大，无法预计算
# 双塔模型 vs 交叉编码器的对比
from sentence_transformers import SentenceTransformer, CrossEncoder
import time
# 双塔模型
bi_encoder = SentenceTransformer("BAAI/bge-base-zh-v1.5")
# 交叉编码器
cross_encoder = CrossEncoder("BAAI/bge-reranker-v2-m3")
query = "如何学习机器学习"
documents = [
"机器学习入门指南：从零开始学习ML的基本概念和算法。",
"深度学习实战：使用PyTorch构建神经网络。",
"今天股市行情分析。",
"Python编程基础教程。",
"机器学习工程师面试题集锦。",
]
# 双塔模型：独立编码
start = time.time()
q_emb = bi_encoder.encode([query])
d_embs = bi_encoder.encode(documents)
bi_scores = (q_emb @ d_embs.T)[0]
bi_time = time.time() - start
# 交叉编码器：联合编码
pairs = [[query, doc] for doc in documents]
start = time.time()
cross_scores = cross_encoder.predict(pairs)
cross_time = time.time() - start
print(f"双塔模型耗时: {bi_time*1000:.1f}ms")
print(f"交叉编码器耗时: {cross_time*1000:.1f}ms")
print("\n双塔模型排序:")
for idx in bi_scores.argsort()[::-1]:
print(f" [{bi_scores[idx]:.4f}] {documents[idx][:30]}")
print("\n交叉编码器排序:")
for idx in cross_scores.argsort()[::-1]:
print(f" [{cross_scores[idx]:.4f}] {documents[idx][:30]}")
4.3 Cohere Rerank v3 深度解析
Cohere Rerank（现称为 Cohere Rerank）是目前最强大的商业重排序API之一。
核心特性：
- 多语言支持：覆盖100+语言
- 长文档处理：支持最长4096 tokens的文档
- 语义理解：基于交叉编码器架构，深度理解查询意图
- 返回解释：可以返回相关性分数和高亮片段
import cohere
co = cohere.Client("YOUR_API_KEY")
query = "企业如何实施数据治理策略"
documents = [
"数据治理是一套确保企业数据质量、安全性和可用性的管理框架。它包括数据标准制定、数据质量管理、元数据管理等核心组件。",
"数据治理实施步骤：1) 建立数据治理组织架构 2) 制定数据标准和政策 3) 部署数据质量管理工具 4) 建立数据血缘追踪。",
"人工智能在医疗领域的应用正在快速增长，包括疾病诊断、药物发现和医学影像分析。",
"企业数字化转型的关键步骤包括：业务流程数字化、数据资产化、智能化决策。",
"数据安全法规GDPR要求企业对个人数据的收集、存储和使用进行严格管理。",
]
# 使用Rerank进行重排序
results = co.rerank(
query=query,
documents=documents,
top_n=3,
model="rerank-v3.5",
return_documents=True,
)
print(f"查询: {query}\n")
for idx, result in enumerate(results.results):
print(f"#{idx+1} (相关度: {result.relevance_score:.4f})")
print(f" {result.document.text[:80]}...\n")
Cohere Rerank v3 的高级用法：
# 1. 结合Embedding检索和重排序的完整管线
def hybrid_search_with_rerank(query, documents, top_k_retrieve=20, top_k_rerank=5):
"""
两阶段检索：Embedding召回 + Rerank精排
"""
# 第一阶段：Embedding粗召回
import cohere
co = cohere.Client("YOUR_API_KEY")
# 生成文档Embedding
doc_response = co.embed(
texts=documents,
model="embed-multilingual-v3.0",
input_type="search_document",
)
# 生成查询Embedding
query_response = co.embed(
texts=[query],
model="embed-multilingual-v3.0",
input_type="search_query",
)
# 计算相似度并召回Top K
import numpy as np
doc_embs = np.array(doc_response.embeddings.float)
query_emb = np.array(query_response.embeddings.float)
scores = (query_emb @ doc_embs.T)[0]
top_indices = np.argsort(scores)[::-1][:top_k_retrieve]
candidate_docs = [documents[i] for i in top_indices]
# 第二阶段：Rerank精排
rerank_results = co.rerank(
query=query,
documents=candidate_docs,
top_n=top_k_rerank,
model="rerank-v3.5",
return_documents=True,
)
final_results = []
for result in rerank_results.results:
final_results.append({
"text": result.document.text,
"rerank_score": result.relevance_score,
"retrieve_score": float(scores[top_indices[result.index]]),
})
return final_results
4.4 开源重排序模型
对于需要本地部署的场景，以下是主要的开源重排序模型：
BGE Reranker系列：
from FlagEmbedding import FlagReranker
# BGE Reranker v2 M3
reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)
# 计算查询-文档对的相关性分数
pairs = [
["什么是深度学习", "深度学习是机器学习的一个分支，使用多层神经网络。"],
["什么是深度学习", "今天天气不错。"],
]
scores = reranker.compute_score(pairs)
print(f"相关文档得分: {scores[0]:.4f}")
print(f"不相关文档得分: {scores[1]:.4f}")
# 批量重排序
query = "如何提高代码质量"
documents = [
"代码审查是提高代码质量的重要手段，通过同行评审发现潜在问题。",
"单元测试覆盖率是衡量代码质量的关键指标之一。",
"今天是周末，适合出去玩。",
"使用静态代码分析工具可以自动发现代码中的安全隐患。",
"设计模式是解决常见软件设计问题的经验总结。",
]
pairs = [[query, doc] for doc in documents]
scores = reranker.compute_score(pairs, normalize=True)
# 按分数排序
ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
for rank, (idx, score) in enumerate(ranked):
print(f"#{rank+1} [{score:.4f}] {documents[idx][:40]}")
Jina Reranker：
import requests
def jina_rerank(query, documents, top_n=5):
"""使用Jina Reranker API"""
response = requests.post(
"https://api.jina.ai/v1/rerank",
headers={
"Authorization": "Bearer YOUR_JINA_API_KEY",
"Content-Type": "application/json",
},
json={
"model": "jina-reranker-v2-base-multilingual",
"query": query,
"documents": documents,
"top_n": top_n,
},
)
results = response.json()
return results["results"]
4.5 重排序实战：从粗排到精排
构建一个完整的多阶段排序管线：
from sentence_transformers import SentenceTransformer, CrossEncoder
import numpy as np
from typing import List, Dict
class TwoStageRetriever:
"""两阶段检索系统：双塔召回 + 交叉编码器重排"""
def __init__(
self,
bi_encoder_name: str = "BAAI/bge-base-zh-v1.5",
cross_encoder_name: str = "BAAI/bge-reranker-v2-m3",
):
self.bi_encoder = SentenceTransformer(bi_encoder_name)
self.cross_encoder = CrossEncoder(cross_encoder_name)
self.documents = []
self.doc_embeddings = None
def index(self, documents: List[str]):
"""索引文档"""
self.documents = documents
self.doc_embeddings = self.bi_encoder.encode(
documents,
normalize_embeddings=True,
show_progress_bar=True,
batch_size=32,
)
print(f"已索引 {len(documents)} 个文档")
def search(
self,
query: str,
retrieve_top_k: int = 50,
rerank_top_k: int = 10,
) -> List[Dict]:
"""两阶段检索"""
# 第一阶段：双塔召回
query_emb = self.bi_encoder.encode(
[query],
normalize_embeddings=True,
)
retrieve_scores = (query_emb @ self.doc_embeddings.T)[0]
retrieve_indices = np.argsort(retrieve_scores)[::-1][:retrieve_top_k]
candidate_docs = [self.documents[i] for i in retrieve_indices]
candidate_scores = [float(retrieve_scores[i]) for i in retrieve_indices]
# 第二阶段：交叉编码器重排
pairs = [[query, doc] for doc in candidate_docs]
rerank_scores = self.cross_encoder.predict(pairs)
# 合并结果
results = []
for i, (doc_idx, rerank_score) in enumerate(zip(retrieve_indices, rerank_scores)):
results.append({
"doc_id": int(doc_idx),
"text": self.documents[doc_idx],
"retrieve_score": candidate_scores[i],
"rerank_score": float(rerank_score),
})
# 按重排序分数排序
results.sort(key=lambda x: x["rerank_score"], reverse=True)
return results[:rerank_top_k]
第五章：语义搜索管线构建
5.1 语义搜索架构全景
一个完整的语义搜索系统包含以下组件：
┌──────────────────────────────────────────────────────────┐
│ 用户查询 │
└──────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────┐
│ 查询理解层（Query Understanding） │
│ - 查询改写 / 扩展 │
│ - 意图识别 │
│ - 实体识别 │
└──────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────┐
│ 检索层（Retrieval） │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────────┐ │
│ │ 稠密检索 │ │ 稀疏检索 │ │ 知识图谱检索 │ │
│ │(Embedding) │ │(BM25/TF-IDF)│ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬───────┘ │
│ └────────────┬───┴────────────────┘ │
│ ▼ │
│ 结果融合（Fusion） │
└──────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────┐
│ 重排序层（Reranking） │
│ - 交叉编码器精排 │
│ - 特征融合 │
└──────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────┐
│ 结果输出层 │
│ - 结果去重与过滤 │
│ - 摘要生成 │
│ - 个性化调整 │
└──────────────────────────────────────────────────────────┘
5.2 向量数据库选型与实战
主流向量数据库对比：
| 数据库 | 类型 | 特点 | 适用场景 |
|---|---|---|---|
| Milvus | 分布式 | 高可用、可扩展、丰富的索引类型 | 大规模生产环境 |
| Qdrant | 独立服务 | Rust实现、高性能、丰富的过滤 | 中大规模生产环境 |
| Weaviate | 独立服务 | GraphQL API、内置向量化 | 快速原型和生产 |
| ChromaDB | 嵌入式 | 简单易用、轻量级 | 原型开发和小规模 |
| Pinecone | 云服务 | 全托管、零运维 | 不想自建的团队 |
| pgvector | PostgreSQL扩展 | 与PostgreSQL集成 | 已有PG的团队 |
使用Milvus构建语义搜索：
from pymilvus import (
connections,
FieldSchema,
CollectionSchema,
DataType,
Collection,
utility,
)
from sentence_transformers import SentenceTransformer
import numpy as np
# 1. 连接Milvus
connections.connect("default", host="localhost", port="19530")
# 2. 定义Schema
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2048),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
]
schema = CollectionSchema(fields, description="语义搜索文档集合")
collection = Collection("semantic_search", schema)
# 3. 创建索引
index_params = {
"metric_type": "COSINE",
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 200},
}
collection.create_index("embedding", index_params)
# 4. 准备数据并插入
model = SentenceTransformer("BAAI/bge-base-zh-v1.5")
documents = [
"深度学习在图像识别领域的应用取得了巨大突破",
"自然语言处理技术使得机器能够理解和生成人类语言",
"强化学习在游戏AI和机器人控制中表现出色",
"数据预处理是机器学习项目中最重要的步骤之一",
"分布式训练可以加速大规模模型的训练过程",
]
embeddings = model.encode(documents, normalize_embeddings=True)
categories = ["CV", "NLP", "RL", "Data", "Infrastructure"]
# 插入数据
collection.insert([
documents, # text字段
embeddings.tolist(), # embedding字段
categories, # category字段
])
# 5. 加载集合到内存
collection.load()
# 6. 语义搜索
query = "计算机视觉有哪些进展"
query_emb = model.encode([query], normalize_embeddings=True)
search_params = {"metric_type": "COSINE", "params": {"ef": 100}}
results = collection.search(
data=query_emb.tolist(),
anns_field="embedding",
param=search_params,
limit=3,
output_fields=["text", "category"],
)
print(f"查询: {query}\n")
for hits in results:
for hit in hits:
print(f" [{hit.score:.4f}] [{hit.entity.get('category')}] {hit.entity.get('text')}")
使用Qdrant构建语义搜索：
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance,
VectorParams,
PointStruct,
Filter,
FieldCondition,
MatchValue,
)
from sentence_transformers import SentenceTransformer
import numpy as np
# 1. 连接Qdrant
client = QdrantClient(host="localhost", port=6333)
# 2. 创建集合
client.create_collection(
collection_name="documents",
vectors_config=VectorParams(
size=768,
distance=Distance.COSINE,
),
)
# 3. 准备数据
model = SentenceTransformer("BAAI/bge-base-zh-v1.5")
documents = [
{"text": "深度学习在CV领域进展迅速", "category": "AI"},
{"text": "NLP技术让机器理解语言", "category": "AI"},
{"text": "React是前端框架", "category": "Web"},
]
texts = [d["text"] for d in documents]
embeddings = model.encode(texts, normalize_embeddings=True)
# 4. 插入数据
points = [
PointStruct(
id=i,
vector=emb.tolist(),
payload=doc,
)
for i, (emb, doc) in enumerate(zip(embeddings, documents))
]
client.upsert(collection_name="documents", points=points)
# 5. 语义搜索
query = "人工智能最新进展"
query_emb = model.encode([query], normalize_embeddings=True)[0]
results = client.search(
collection_name="documents",
query_vector=query_emb.tolist(),
limit=3,
)
for result in results:
print(f" [{result.score:.4f}] {result.payload['text']}")
# 6. 带过滤的搜索（混合检索）
filtered_results = client.search(
collection_name="documents",
query_vector=query_emb.tolist(),
query_filter=Filter(
must=[
FieldCondition(
key="category",
match=MatchValue(value="AI"),
)
]
),
limit=3,
)
5.3 混合检索策略
混合检索（Hybrid Search）结合稠密检索（语义匹配）和稀疏检索（关键词匹配）的优势，是当前生产系统的最佳实践。
为什么需要混合检索？
- 稠密检索擅长语义理解，但可能忽略精确关键词匹配
- 稀疏检索（BM25）擅长精确匹配，但无法理解语义
- 两者互补，融合后效果通常优于任何单一方法
RRF（Reciprocal Rank Fusion）融合算法：
def reciprocal_rank_fusion(
rankings: list[list[int]],
k: int = 60,
) -> list[tuple[int, float]]:
"""
RRF融合多个排序列表
rankings: 每个检索方法返回的文档ID排序列表
k: 平滑参数（通常取60）
"""
rrf_scores = {}
for ranking in rankings:
for rank, doc_id in enumerate(ranking):
if doc_id not in rrf_scores:
rrf_scores[doc_id] = 0.0
rrf_scores[doc_id] += 1.0 / (k + rank + 1)
# 按RRF分数排序
sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
return sorted_results
# 混合检索实战
from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import numpy as np
class HybridSearcher:
def __init__(self, model_name="BAAI/bge-base-zh-v1.5"):
self.model = SentenceTransformer(model_name)
self.documents = []
self.doc_embeddings = None
self.bm25 = None
def index(self, documents: list[str]):
self.documents = documents
# 构建稠密索引
self.doc_embeddings = self.model.encode(
documents, normalize_embeddings=True
)
# 构建BM25索引
tokenized_docs = [list(doc) for doc in documents] # 简化分词
self.bm25 = BM25Okapi(tokenized_docs)
def search(self, query: str, top_k: int = 10, alpha: float = 0.7):
"""
alpha: 稠密检索权重（1-alpha为稀疏检索权重）
"""
# 稠密检索
query_emb = self.model.encode([query], normalize_embeddings=True)
dense_scores = (query_emb @ self.doc_embeddings.T)[0]
dense_ranking = np.argsort(dense_scores)[::-1].tolist()
# 稀疏检索（BM25）
query_tokens = list(query)
sparse_scores = self.bm25.get_scores(query_tokens)
sparse_ranking = np.argsort(sparse_scores)[::-1].tolist()
# RRF融合
fused = reciprocal_rank_fusion([dense_ranking[:50], sparse_ranking[:50]])
results = []
for doc_id, score in fused[:top_k]:
results.append({
"doc_id": doc_id,
"text": self.documents[doc_id],
"fused_score": score,
"dense_score": float(dense_scores[doc_id]),
"sparse_score": float(sparse_scores[doc_id]),
})
return results
# 使用示例
searcher = HybridSearcher()
searcher.index([
"深度学习是机器学习的子领域",
"机器学习算法包括决策树和SVM",
"Python是最流行的编程语言",
"神经网络的反向传播算法",
])
results = searcher.search("如何训练深度神经网络")
for r in results:
print(f" [{r['fused_score']:.4f}] {r['text']}")
5.4 检索增强生成（RAG）中的Embedding
RAG（Retrieval-Augmented Generation）是Embedding技术最重要的应用场景之一。在RAG系统中，Embedding的质量直接决定了检索质量，进而影响生成质量。
from sentence_transformers import SentenceTransformer
import numpy as np
class SimpleRAGSystem:
"""简单的RAG系统示例"""
def __init__(self, embedding_model="BAAI/bge-base-zh-v1.5"):
self.embedder = SentenceTransformer(embedding_model)
self.chunks = []
self.chunk_embeddings = None
self.metadata = []
def add_documents(self, documents: list[dict]):
"""
documents: [{"text": "...", "source": "...", "page": 1}, ...]
"""
for doc in documents:
# 简单的文本分块（实际生产中需要更精细的分块策略）
text = doc["text"]
chunks = self._split_text(text, chunk_size=500, overlap=50)
for chunk in chunks:
self.chunks.append(chunk)
self.metadata.append({
"source": doc.get("source", "unknown"),
"page": doc.get("page", 0),
})
# 编码所有块
self.chunk_embeddings = self.embedder.encode(
self.chunks,
normalize_embeddings=True,
show_progress_bar=True,
)
print(f"已索引 {len(self.chunks)} 个文本块")
def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
"""检索相关文本块"""
query_emb = self.embedder.encode([query], normalize_embeddings=True)
scores = (query_emb @ self.chunk_embeddings.T)[0]
top_indices = np.argsort(scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"text": self.chunks[idx],
"score": float(scores[idx]),
"metadata": self.metadata[idx],
})
return results
def _split_text(self, text, chunk_size=500, overlap=50):
"""简单的文本分块"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunks
def build_prompt(self, query: str, retrieved_chunks: list[dict]) -> str:
"""构建RAG提示词"""
context = "\n\n".join([
f"[来源: {c['metadata']['source']}] {c['text']}"
for c in retrieved_chunks
])
prompt = f"""基于以下检索到的参考资料回答用户问题。如果参考资料中没有相关信息，请说明。
参考资料:
{context}
用户问题: {query}
回答:"""
return prompt
分块策略对RAG效果的影响：
class TextChunker:
"""多种文本分块策略"""
@staticmethod
def fixed_size(text: str, chunk_size: int = 500, overlap: int = 50):
"""固定大小分块"""
chunks = []
for i in range(0, len(text), chunk_size - overlap):
chunks.append(text[i:i + chunk_size])
return chunks
@staticmethod
def sentence_based(text: str, max_chunk_size: int = 500):
"""基于句子的分块"""
import re
sentences = re.split(r'[。！？\n]', text)
chunks = []
current_chunk = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
if len(current_chunk) + len(sentence) <= max_chunk_size:
current_chunk += sentence + "。"
else:
if current_chunk:
chunks.append(current_chunk)
current_chunk = sentence + "。"
if current_chunk:
chunks.append(current_chunk)
return chunks
@staticmethod
def semantic_split(text: str, model, threshold: float = 0.5):
"""基于语义相似度的分块（相邻段落相似度低于阈值时切分）"""
import re
sentences = re.split(r'[。！？\n]', text)
sentences = [s.strip() for s in sentences if s.strip()]
if len(sentences) <= 1:
return [text]
embeddings = model.encode(sentences, normalize_embeddings=True)
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
sim = np.dot(embeddings[i], embeddings[i-1])
if sim < threshold:
chunks.append("。".join(current_chunk) + "。")
current_chunk = [sentences[i]]
else:
current_chunk.append(sentences[i])
if current_chunk:
chunks.append("。".join(current_chunk) + "。")
return chunks
第六章：Embedding微调与优化
6.1 何时需要微调Embedding模型
以下场景建议微调Embedding模型：
- 领域专业性强：医疗、法律、金融等领域术语，通用模型理解不足
- 特定任务优化：如代码搜索、产品匹配、FAQ检索等特定场景
- 语言特殊性：方言、行业黑话、新词等未被预训练覆盖的语言现象
- 质量不达标：在你的数据集上，现有模型的检索效果（如NDCG@10）不满足业务需求
微调前的评估清单：
from sentence_transformers import SentenceTransformer
import numpy as np
def evaluate_retrieval(model, queries, documents, relevance_labels):
"""
评估检索效果
queries: 查询列表
documents: 文档列表
relevance_labels: {query_idx: [relevant_doc_indices]}
"""
q_embs = model.encode(queries, normalize_embeddings=True)
d_embs = model.encode(documents, normalize_embeddings=True)
scores = q_embs @ d_embs.T
# 计算 MRR (Mean Reciprocal Rank)
mrr_scores = []
for q_idx in range(len(queries)):
ranked = np.argsort(scores[q_idx])[::-1]
relevant = set(relevance_labels.get(q_idx, []))
for rank, doc_idx in enumerate(ranked):
if doc_idx in relevant:
mrr_scores.append(1.0 / (rank + 1))
break
else:
mrr_scores.append(0.0)
mrr = np.mean(mrr_scores)
# 计算 NDCG@10
ndcg_scores = []
for q_idx in range(len(queries)):
ranked = np.argsort(scores[q_idx])[::-1][:10]
relevant = relevance_labels.get(q_idx, [])
dcg = sum(
1.0 / np.log2(rank + 2)
for rank, doc_idx in enumerate(ranked)
if doc_idx in relevant
)
idcg = sum(1.0 / np.log2(i + 2) for i in range(len(relevant)))
ndcg_scores.append(dcg / idcg if idcg > 0 else 0)
return {
"MRR": np.mean(mrr_scores),
"NDCG@10": np.mean(ndcg_scores),
}
6.2 Sentence-Transformers微调实战
方法一：使用标注数据微调
from sentence_transformers import (
SentenceTransformer,
InputExample,
losses,
evaluation,
)
from torch.utils.data import DataLoader
import json
# 1. 加载基础模型
model = SentenceTransformer("BAAI/bge-base-zh-v1.5")
# 2. 准备训练数据
# 格式：(query, positive_doc, negative_doc)
train_examples = [
InputExample(
texts=[
"如何申请退款",
"您可以在订单详情页点击'申请退款'按钮，填写退款原因后提交。",
"我们的营业时间是周一到周五9:00-18:00。",
],
),
InputExample(
texts=[
"密码忘了怎么办",
"请点击登录页面的'忘记密码'，通过手机号或邮箱重置密码。",
"商品将在下单后3个工作日内发货。",
],
),
# ... 更多训练样本
]
# 3. 创建DataLoader
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
# 4. 选择损失函数
# MultipleNegativesRankingLoss 适合三元组数据
train_loss = losses.MultipleNegativesRankingLoss(model)
# 5. 创建评估器（可选）
eval_queries = ["如何退款", "密码重置"]
eval_documents = ["退款流程说明", "密码重置方法", "发货时间"]
eval_relevance = {0: [0], 1: [1]}
evaluator = evaluation.InformationRetrievalEvaluator(
eval_queries,
eval_documents,
eval_relevance,
show_progress_bar=True,
)
# 6. 开始训练
model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=3,
warmup_steps=100,
evaluator=evaluator,
evaluation_steps=500,
output_path="./finetuned-bge-qa",
save_best_model=True,
)
print("微调完成！模型已保存到 ./finetuned-bge-qa")
方法二：使用合成数据微调（适合缺少标注数据的场景）
from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
import json
def generate_training_pairs_with_llm(documents, api_key):
"""
使用LLM为文档生成查询，构建训练对
"""
import cohere
co = cohere.Client(api_key)
training_pairs = []
for doc in documents:
response = co.generate(
model="command",
prompt=f"根据以下文档内容，生成3个用户可能会搜索的查询：\n\n文档：{doc}\n\n查询：",
max_tokens=200,
)
queries = response.generations[0].text.strip().split("\n")
for query in queries:
query = query.strip().lstrip("0123456789.-)")
if query:
training_pairs.append(InputExample(texts=[query, doc]))
return training_pairs
# 使用合成数据训练
model = SentenceTransformer("BAAI/bge-base-zh-v1.5")
# 假设我们有一批领域文档
domain_documents = [
"本产品支持7天无理由退货，需保持商品完好。",
"会员积分每100分可抵扣1元，积分有效期为一年。",
# ... 更多文档
]
# 生成训练对
train_examples = generate_training_pairs_with_llm(
domain_documents,
api_key="YOUR_API_KEY",
)
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
train_loss = losses.MultipleNegativesRankingLoss(model)
model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=1,
warmup_steps=50,
output_path="./synthetic-finetuned-bge",
)
6.3 领域自适应训练策略
三阶段训练法：
阶段1: 弱监督预训练
- 使用大规模领域文本对（如标题-正文、问题-回答）
- 目标：让模型适应领域词汇和表达
阶段2: 有监督微调
- 使用人工标注的查询-文档对
- 目标：优化检索精度
阶段3: 难负样本训练
- 使用阶段2的模型挖掘难负样本
- 目标：提升模型对细微差异的辨别能力
# 难负样本挖掘
import numpy as np
from sentence_transformers import SentenceTransformer
def mine_hard_negatives(model, queries, documents, top_k=10):
"""
挖掘难负样本：检索排名靠前但实际不相关的文档
"""
q_embs = model.encode(queries, normalize_embeddings=True)
d_embs = model.encode(documents, normalize_embeddings=True)
scores = q_embs @ d_embs.T
hard_negatives = {}
for q_idx in range(len(queries)):
ranked = np.argsort(scores[q_idx])[::-1]
# 排除前3个（可能是正样本），取4-10作为难负样本
hard_negatives[q_idx] = ranked[3:top_k].tolist()
return hard_negatives
# 使用难负样本构建训练数据
hard_negs = mine_hard_negatives(model, train_queries, train_documents)
train_examples_with_hard = []
for q_idx, doc_idx in positive_pairs:
neg_indices = hard_negs.get(q_idx, [])
for neg_idx in neg_indices[:3]: # 每个正样本配3个难负样本
train_examples_with_hard.append(
InputExample(texts=[
train_queries[q_idx],
train_documents[doc_idx],
train_documents[neg_idx],
])
)
第七章：向量压缩与量化
7.1 向量量化的基本原理
向量量化（Vector Quantization）是将高维浮点向量压缩为更紧凑表示的技术。在生产系统中，向量量化至关重要：
- 存储节省：10亿条768维float32向量需要约2.88TB，量化后可降至数百GB甚至几十GB
- 检索加速：压缩向量的距离计算更快，且能更好地利用CPU缓存
- 成本降低：减少向量数据库的硬件需求
7.2 标量量化与乘积量化
标量量化（Scalar Quantization, SQ）：将每个float32值量化为int8或其他低精度类型。
import numpy as np
def scalar_quantize(vectors: np.ndarray, bits: int = 8):
"""
标量量化：将float向量量化为整数
"""
# 计算每维的最小值和最大值
min_vals = vectors.min(axis=0)
max_vals = vectors.max(axis=0)
# 计算缩放因子
scale = (max_vals - min_vals) / (2**bits - 1)
scale[scale == 0] = 1 # 避免除零
# 量化
if bits == 8:
quantized = ((vectors - min_vals) / scale).astype(np.uint8)
elif bits == 16:
quantized = ((vectors - min_vals) / scale).astype(np.uint16)
return quantized, min_vals, scale
def scalar_dequantize(quantized, min_vals, scale):
"""反量化"""
return quantized.astype(np.float32) * scale + min_vals
# 示例
vectors = np.random.randn(10000, 768).astype(np.float32)
quantized, mins, scales = scalar_quantize(vectors, bits=8)
# 存储节省
original_size = vectors.nbytes
quantized_size = quantized.nbytes + mins.nbytes + scales.nbytes
print(f"原始大小: {original_size / 1024 / 1024:.1f} MB")
print(f"量化后: {quantized_size / 1024 / 1024:.1f} MB")
print(f"压缩比: {original_size / quantized_size:.1f}x")
乘积量化（Product Quantization, PQ）：将向量分成多个子空间，每个子空间独立聚类。
import numpy as np
class ProductQuantizer:
"""简单的乘积量化实现"""
def __init__(self, num_subspaces: int = 8, num_clusters: int = 256):
self.num_subspaces = num_subspaces
self.num_clusters = num_clusters
self.codebooks = None # 每个子空间的聚类中心
def fit(self, vectors: np.ndarray):
"""训练PQ码本"""
dim = vectors.shape[1]
assert dim % self.num_subspaces == 0
sub_dim = dim // self.num_subspaces
self.codebooks = []
for i in range(self.num_subspaces):
sub_vectors = vectors[:, i * sub_dim : (i + 1) * sub_dim]
# K-Means聚类
from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=self.num_clusters, n_init=1)
kmeans.fit(sub_vectors)
self.codebooks.append(kmeans.cluster_centers_)
def encode(self, vectors: np.ndarray) -> np.ndarray:
"""编码向量为PQ码"""
dim = vectors.shape[1]
sub_dim = dim // self.num_subspaces
n = vectors.shape[0]
codes = np.zeros((n, self.num_subspaces), dtype=np.uint8)
for i in range(self.num_subspaces):
sub_vectors = vectors[:, i * sub_dim : (i + 1) * sub_dim]
# 找最近的聚类中心
from sklearn.metrics.pairwise import euclidean_distances
dists = euclidean_distances(sub_vectors, self.codebooks[i])
codes[:, i] = np.argmin(dists, axis=1)
return codes
def decode(self, codes: np.ndarray) -> np.ndarray:
"""解码PQ码为近似向量"""
n = codes.shape[0]
sub_dim = self.codebooks[0].shape[1]
vectors = np.zeros((n, sub_dim * self.num_subspaces), dtype=np.float32)
for i in range(self.num_subspaces):
vectors[:, i * sub_dim : (i + 1) * sub_dim] = self.codebooks[i][codes[:, i]]
return vectors
# 使用示例
vectors = np.random.randn(100000, 768).astype(np.float32)
pq = ProductQuantizer(num_subspaces=8, num_clusters=256)
pq.fit(vectors)
codes = pq.encode(vectors)
print(f"原始大小: {vectors.nbytes / 1024 / 1024:.1f} MB")
print(f"PQ编码大小: {codes.nbytes / 1024 / 1024:.1f} MB")
print(f"压缩比: {vectors.nbytes / codes.nbytes:.0f}x")
7.3 二值化与Matryoshka表示
二值化（Binary Quantization）：将float向量的每个维度量化为1 bit（正数→1，负数→0），实现32倍压缩。
import numpy as np
def binary_quantize(vectors: np.ndarray) -> np.ndarray:
"""二值量化：float → bit"""
return np.packbits(vectors > 0, axis=1)
def hamming_distance(a: np.ndarray, b: np.ndarray) -> np.ndarray:
"""计算汉明距离（用位运算高效计算）"""
xor = np.bitwise_xor(a[:, None, :], b[None, :, :])
# 计算每个byte中1的个数
popcount = np.zeros(xor.shape[:2], dtype=np.int32)
for i in range(xor.shape[2]):
# 使用查表法计算popcount
byte = xor[:, :, i]
popcount += np.unpackbits(np.expand_dims(byte, -1), axis=-1).sum(axis=-1)
return popcount
# 示例
vectors = np.random.randn(10000, 768).astype(np.float32)
binary_codes = binary_quantize(vectors)
print(f"原始大小: {vectors.nbytes / 1024 / 1024:.1f} MB")
print(f"二值编码大小: {binary_codes.nbytes / 1024 / 1024:.1f} MB")
print(f"压缩比: {vectors.nbytes / binary_codes.nbytes:.0f}x")
Matryoshka表示学习（Matryoshka Representation Learning, MRL）：
MRL是一种训练Embedding模型的方法，使得模型的前d维子向量也能保持良好的检索质量。
import numpy as np
def matryoshka_search(query_emb, doc_embs, dimensions=[768, 256, 64]):
"""
Matryoshka搜索：支持不同精度的检索
"""
results = {}
for dim in dimensions:
# 使用前dim维进行检索
q_sub = query_emb[:, :dim]
d_sub = doc_embs[:, :dim]
# 归一化
q_norm = q_sub / np.linalg.norm(q_sub, axis=1, keepdims=True)
d_norm = d_sub / np.linalg.norm(d_sub, axis=1, keepdims=True)
scores = (q_norm @ d_norm.T)[0]
results[dim] = np.argsort(scores)[::-1]
return results
# 在实际应用中，可以先用低维快速筛选，再用高维精排
# 例如：先用64维召回Top 100，再用768维精排Top 10
7.4 量化实战：在精度与性能间取舍
import numpy as np
from sentence_transformers import SentenceTransformer
def benchmark_quantization(model_name, queries, documents, relevance):
"""
量化方案对比评估
"""
model = SentenceTransformer(model_name)
q_embs = model.encode(queries, normalize_embeddings=True)
d_embs = model.encode(documents, normalize_embeddings=True)
results = {}
# 1. 原始float32
scores = q_embs @ d_embs.T
results["float32"] = compute_ndcg(scores, relevance, k=10)
# 2. float16
q_f16 = q_embs.astype(np.float16)
d_f16 = d_embs.astype(np.float16)
scores_f16 = q_f16.astype(np.float32) @ d_f16.astype(np.float32).T
results["float16"] = compute_ndcg(scores_f16, relevance, k=10)
# 3. int8 标量量化
q_i8, q_min, q_scale = scalar_quantize_int8(q_embs)
d_i8, d_min, d_scale = scalar_quantize_int8(d_embs)
# 近似距离计算
scores_i8 = q_i8.astype(np.float32) @ d_i8.astype(np.float32).T
results["int8"] = compute_ndcg(scores_i8, relevance, k=10)
# 4. 二值量化
q_bin = binary_quantize(q_embs)
d_bin = binary_quantize(d_embs)
# 汉明距离
hamming = hamming_distance(q_bin, d_bin)
scores_bin = -hamming # 汉明距离越小越好
results["binary"] = compute_ndcg(scores_bin, relevance, k=10)
return results
第八章：生产级检索架构设计
8.1 大规模Embedding服务部署
在生产环境中，Embedding服务需要满足高吞吐、低延迟、高可用的要求。
部署架构选择：
小规模 (< 10 QPS)
→ 单机Python服务 (Flask/FastAPI)
中等规模 (10-1000 QPS)
→ GPU推理服务 (Triton/TorchServe) + 负载均衡
大规模 (> 1000 QPS)
→ 多GPU推理集群 + 模型分片 + 动态批处理
使用Triton Inference Server部署Embedding模型：
# model_repository/text_embedding/config.pbtxt
"""
name: "text_embedding"
platform: "python"
max_batch_size: 256
input [
{
name: "TEXT"
data_type: TYPE_STRING
dims: [1]
}
]
output [
{
name: "EMBEDDING"
data_type: TYPE_FP32
dims: [768]
}
]
instance_group [
{
count: 2
kind: KIND_GPU
}
]
dynamic_batching {
preferred_batch_size: [32, 64, 128]
max_queue_delay_microseconds: 100000
}
"""
# model_repository/text_embedding/1/model.py
"""
import triton_python_backend_utils as pb_utils
import numpy as np
from sentence_transformers import SentenceTransformer
import torch
class TritonPythonModel:
def initialize(self, args):
self.model = SentenceTransformer(
"BAAI/bge-base-zh-v1.5",
device="cuda" if torch.cuda.is_available() else "cpu",
)
def execute(self, requests):
responses = []
# 批量处理所有请求中的文本
all_texts = []
for request in requests:
texts = pb_utils.get_input_tensor_by_name(request, "TEXT").as_numpy()
all_texts.extend([t.decode("utf-8") for t in texts.flatten()])
# 批量编码
embeddings = self.model.encode(
all_texts,
normalize_embeddings=True,
batch_size=len(all_texts),
)
# 构建响应
idx = 0
for request in requests:
n = len(pb_utils.get_input_tensor_by_name(request, "TEXT").as_numpy().flatten())
emb = embeddings[idx:idx+n]
idx += n
out_tensor = pb_utils.Tensor("EMBEDDING", emb.astype(np.float32))
responses.append(pb_utils.InferenceResponse([out_tensor]))
return responses
"""
使用FastAPI构建Embedding服务：
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
from typing import List
import asyncio
from functools import lru_cache
import torch
app = FastAPI()
# 模型加载（全局单例）
@lru_cache()
def get_model():
return SentenceTransformer(
"BAAI/bge-base-zh-v1.5",
device="cuda" if torch.cuda.is_available() else "cpu",
)
class EmbeddingRequest(BaseModel):
texts: List[str]
normalize: bool = True
class EmbeddingResponse(BaseModel):
embeddings: List[List[float]]
dimensions: int
model: str
@app.post("/embed", response_model=EmbeddingResponse)
async def create_embeddings(request: EmbeddingRequest):
model = get_model()
if len(request.texts) > 1000:
raise HTTPException(400, "单次请求最多1000条文本")
# 异步编码
loop = asyncio.get_event_loop()
embeddings = await loop.run_in_executor(
None,
lambda: model.encode(
request.texts,
normalize_embeddings=request.normalize,
batch_size=64,
),
)
return EmbeddingResponse(
embeddings=embeddings.tolist(),
dimensions=embeddings.shape[1],
model="bge-base-zh-v1.5",
)
@app.get("/health")
async def health():
model = get_model()
return {"status": "healthy", "model": "bge-base-zh-v1.5"}
8.2 缓存与增量更新策略
Embedding缓存：对于重复查询或文档，缓存Embedding结果可以显著降低计算成本。
import hashlib
import json
from typing import Optional
import redis
import numpy as np
class EmbeddingCache:
"""基于Redis的Embedding缓存"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 86400 * 7):
self.redis = redis.from_url(redis_url)
self.ttl = ttl # 默认7天过期
def _make_key(self, text: str, model_name: str) -> str:
text_hash = hashlib.md5(text.encode()).hexdigest()
return f"emb:{model_name}:{text_hash}"
def get(self, text: str, model_name: str) -> Optional[np.ndarray]:
key = self._make_key(text, model_name)
data = self.redis.get(key)
if data:
return np.frombuffer(data, dtype=np.float32)
return None
def set(self, text: str, model_name: str, embedding: np.ndarray):
key = self._make_key(text, model_name)
self.redis.setex(key, self.ttl, embedding.astype(np.float32).tobytes())
def batch_get(self, texts: list[str], model_name: str) -> dict:
"""批量获取缓存"""
pipe = self.redis.pipeline()
keys = []
for text in texts:
key = self._make_key(text, model_name)
keys.append(key)
pipe.get(key)
results = pipe.execute()
cached = {}
for i, (key, data) in enumerate(zip(keys, results)):
if data:
cached[i] = np.frombuffer(data, dtype=np.float32)
return cached
def batch_set(self, texts: list[str], model_name: str, embeddings: np.ndarray):
"""批量设置缓存"""
pipe = self.redis.pipeline()
for text, emb in zip(texts, embeddings):
key = self._make_key(text, model_name)
pipe.setex(key, self.ttl, emb.astype(np.float32).tobytes())
pipe.execute()
class CachedEmbeddingModel:
"""带缓存的Embedding模型"""
def __init__(self, model_name: str = "BAAI/bge-base-zh-v1.5"):
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer(model_name)
self.cache = EmbeddingCache()
self.model_name = model_name
def encode(self, texts: list[str], normalize: bool = True) -> np.ndarray:
# 1. 查缓存
cached = self.cache.batch_get(texts, self.model_name)
# 2. 找出未缓存的文本
uncached_indices = []
uncached_texts = []
for i in range(len(texts)):
if i not in cached:
uncached_indices.append(i)
uncached_texts.append(texts[i])
# 3. 编码未缓存的文本
if uncached_texts:
new_embeddings = self.model.encode(
uncached_texts,
normalize_embeddings=normalize,
)
# 写入缓存
self.cache.batch_set(uncached_texts, self.model_name, new_embeddings)
else:
new_embeddings = np.array([])
# 4. 合并结果
result = np.zeros((len(texts), self.model.get_sentence_embedding_dimension()))
for i in range(len(texts)):
if i in cached:
result[i] = cached[i]
else:
uncached_idx = uncached_indices.index(i)
result[i] = new_embeddings[uncached_idx]
return result
8.3 监控与可观测性
import time
from dataclasses import dataclass, field
from typing import Dict, List
import threading
@dataclass
class EmbeddingMetrics:
"""Embedding服务监控指标"""
request_count: int = 0
total_latency_ms: float = 0
error_count: int = 0
cache_hits: int = 0
cache_misses: int = 0
batch_sizes: List[int] = field(default_factory=list)
_lock: threading.Lock = field(default_factory=threading.Lock)
def record_request(self, latency_ms: float, batch_size: int, from_cache: bool = False):
with self._lock:
self.request_count += 1
self.total_latency_ms += latency_ms
self.batch_sizes.append(batch_size)
if from_cache:
self.cache_hits += 1
else:
self.cache_misses += 1
def record_error(self):
with self._lock:
self.error_count += 1
def get_stats(self) -> dict:
with self._lock:
avg_latency = (
self.total_latency_ms / self.request_count
if self.request_count > 0
else 0
)
cache_hit_rate = (
self.cache_hits / (self.cache_hits + self.cache_misses)
if (self.cache_hits + self.cache_misses) > 0
else 0
)
avg_batch = (
sum(self.batch_sizes) / len(self.batch_sizes)
if self.batch_sizes
else 0
)
return {
"request_count": self.request_count,
"avg_latency_ms": round(avg_latency, 2),
"error_count": self.error_count,
"cache_hit_rate": round(cache_hit_rate, 4),
"avg_batch_size": round(avg_batch, 1),
"p99_latency_ms": round(self._percentile(99), 2),
}
def _percentile(self, p: int) -> float:
if not self.batch_sizes:
return 0
sorted_latencies = sorted(self.batch_sizes)
idx = int(len(sorted_latencies) * p / 100)
return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
8.4 安全与合规考量
在生产部署Embedding系统时，需要关注以下安全和合规问题：
- 数据隐私：用户输入的文本可能包含PII（个人可识别信息），在调用外部Embedding API时需注意数据脱敏
- 模型安全：防止对抗样本攻击，即恶意构造的输入导致Embedding质量严重下降
- 访问控制：Embedding API应有适当的认证和限流机制
- 数据主权：使用海外Embedding API（如Cohere）时需考虑数据出境合规问题
import re
from typing import List
class TextSanitizer:
"""文本脱敏处理"""
PATTERNS = {
"phone": (r"1[3-9]\d{9}", "***手机号***"),
"id_card": (r"\d{17}[\dXx]", "***身份证号***"),
"email": (r"[\w.+-]+@[\w-]+\.[\w.]+", "***邮箱***"),
"bank_card": (r"\d{16,19}", "***银行卡号***"),
}
@classmethod
def sanitize(cls, text: str) -> str:
"""对文本中的敏感信息进行脱敏"""
for name, (pattern, replacement) in cls.PATTERNS.items():
text = re.sub(pattern, replacement, text)
return text
@classmethod
def sanitize_batch(cls, texts: List[str]) -> List[str]:
return [cls.sanitize(t) for t in texts]
# 在调用外部API前进行脱敏
sanitizer = TextSanitizer()
raw_texts = ["请联系我 13812345678 或 email@test.com"]
safe_texts = sanitizer.sanitize_batch(raw_texts)
print(safe_texts) # ['请联系我 ***手机号*** 或 ***邮箱***']
第九章：实战项目一——企业级语义搜索系统
9.1 项目背景与需求分析
场景：某企业有大量内部文档（产品手册、FAQ、技术文档、会议纪要等），员工需要快速找到相关信息。
需求：
- 支持中文语义搜索
- 支持多类型文档（PDF、Word、Markdown）
- 搜索延迟 < 500ms
- 支持10万级文档规模
- 结果需要展示来源和上下文
9.2 系统架构设计
┌─────────────────────────────────────────────────────────────┐
│ 前端界面 │
│ (搜索框 + 结果展示 + 过滤器) │
└──────────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ API网关 (FastAPI) │
│ 查询处理 + 认证 + 限流 + 日志 │
└──────────────────────────┬──────────────────────────────────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────┐ ┌──────────────┐
│ Embedding服务 │ │ Rerank服务│ │ 文档处理服务 │
│ (GPU推理) │ │ (GPU推理) │ │ (文本提取/分块)│
└──────┬───────┘ └────┬─────┘ └──────┬───────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────┐
│ 向量数据库 (Milvus) │
│ + 元数据库 (PostgreSQL) │
└─────────────────────────────────────────────┘
9.3 完整代码实现
"""
企业级语义搜索系统 - 完整实现
"""
import os
import hashlib
import time
from typing import List, Dict, Optional
from dataclasses import dataclass
from pathlib import Path
import numpy as np
from fastapi import FastAPI, HTTPException, UploadFile, File
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer, CrossEncoder
from pymilvus import (
connections,
FieldSchema,
CollectionSchema,
DataType,
Collection,
)
import fitz # PyMuPDF for PDF extraction
import docx # python-docx for Word extraction
# ==================== 数据模型 ====================
@dataclass
class Document:
id: str
title: str
content: str
source: str
doc_type: str
chunk_index: int
metadata: dict
class SearchRequest(BaseModel):
query: str
top_k: int = 10
doc_type: Optional[str] = None
use_rerank: bool = True
class SearchResponse(BaseModel):
results: List[Dict]
query_time_ms: float
total_results: int
# ==================== 文档处理 ====================
class DocumentProcessor:
"""文档处理器：支持PDF、Word、Markdown"""
@staticmethod
def extract_text(file_path: str) -> str:
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
return DocumentProcessor._extract_pdf(file_path)
elif suffix in [".docx", ".doc"]:
return DocumentProcessor._extract_docx(file_path)
elif suffix in [".md", ".txt"]:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
else:
raise ValueError(f"不支持的文件格式: {suffix}")
@staticmethod
def _extract_pdf(file_path: str) -> str:
doc = fitz.open(file_path)
text_parts = []
for page in doc:
text_parts.append(page.get_text())
return "\n".join(text_parts)
@staticmethod
def _extract_docx(file_path: str) -> str:
doc = docx.Document(file_path)
return "\n".join([para.text for para in doc.paragraphs])
@staticmethod
def chunk_text(
text: str,
chunk_size: int = 500,
overlap: int = 50,
) -> List[str]:
"""智能文本分块"""
# 先按段落分割
paragraphs = text.split("\n")
chunks = []
current_chunk = ""
for para in paragraphs:
para = para.strip()
if not para:
continue
if len(current_chunk) + len(para) <= chunk_size:
current_chunk += para + "\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
# 如果单个段落超过chunk_size，按字符切分
if len(para) > chunk_size:
for i in range(0, len(para), chunk_size - overlap):
chunks.append(para[i:i + chunk_size])
else:
current_chunk = para + "\n"
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
# ==================== 搜索引擎 ====================
class SemanticSearchEngine:
"""语义搜索引擎核心"""
def __init__(
self,
embedding_model: str = "BAAI/bge-base-zh-v1.5",
reranker_model: str = "BAAI/bge-reranker-v2-m3",
milvus_host: str = "localhost",
milvus_port: int = 19530,
collection_name: str = "enterprise_docs",
):
# 加载模型
print(f"加载Embedding模型: {embedding_model}")
self.embedder = SentenceTransformer(embedding_model)
print(f"加载重排序模型: {reranker_model}")
self.reranker = CrossEncoder(reranker_model)
# 连接Milvus
connections.connect("default", host=milvus_host, port=milvus_port)
self.collection_name = collection_name
# 创建集合
self._setup_collection()
# 文档处理器
self.doc_processor = DocumentProcessor()
def _setup_collection(self):
"""创建Milvus集合"""
if self.collection_name in [c.name for c in Collection.list()]:
self.collection = Collection(self.collection_name)
return
fields = [
FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=64, is_primary=True),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=4096),
FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=32),
FieldSchema(name="chunk_index", dtype=DataType.INT64),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
]
schema = CollectionSchema(fields, description="企业文档语义搜索")
self.collection = Collection(self.collection_name, schema)
# 创建HNSW索引
index_params = {
"metric_type": "COSINE",
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 200},
}
self.collection.create_index("embedding", index_params)
self.collection.load()
def index_document(
self,
file_path: str,
title: str = "",
doc_type: str = "general",
):
"""索引单个文档"""
# 提取文本
text = self.doc_processor.extract_text(file_path)
if not title:
title = Path(file_path).stem
# 分块
chunks = self.doc_processor.chunk_text(text, chunk_size=500, overlap=50)
if not chunks:
print(f"警告: 文档 {file_path} 无有效内容")
return
# 生成Embedding
embeddings = self.embedder.encode(
chunks,
normalize_embeddings=True,
show_progress_bar=False,
)
# 生成文档ID
doc_hash = hashlib.md5(file_path.encode()).hexdigest()[:12]
# 准备数据
ids = [f"{doc_hash}_{i}" for i in range(len(chunks))]
titles = [title] * len(chunks)
sources = [file_path] * len(chunks)
doc_types = [doc_type] * len(chunks)
chunk_indices = list(range(len(chunks)))
# 插入Milvus
self.collection.insert([
ids,
chunks,
titles,
sources,
doc_types,
chunk_indices,
embeddings.tolist(),
])
self.collection.flush()
print(f"已索引文档: {title}, {len(chunks)} 个分块")
def index_directory(self, directory: str, doc_type: str = "general"):
"""批量索引目录中的所有文档"""
supported = [".pdf", ".docx", ".doc", ".md", ".txt"]
files = []
for ext in supported:
files.extend(Path(directory).rglob(f"*{ext}"))
print(f"找到 {len(files)} 个文档")
for file_path in files:
try:
self.index_document(str(file_path), doc_type=doc_type)
except Exception as e:
print(f"索引失败 {file_path}: {e}")
def search(
self,
query: str,
top_k: int = 10,
doc_type: Optional[str] = None,
use_rerank: bool = True,
retrieve_top_k: int = 50,
) -> List[Dict]:
"""语义搜索"""
start_time = time.time()
# 1. Embedding检索
query_emb = self.embedder.encode(
[query],
normalize_embeddings=True,
)
search_params = {"metric_type": "COSINE", "params": {"ef": 100}}
# 构建过滤条件
filter_expr = f'doc_type == "{doc_type}"' if doc_type else None
results = self.collection.search(
data=query_emb.tolist(),
anns_field="embedding",
param=search_params,
limit=retrieve_top_k,
output_fields=["text", "title", "source", "doc_type", "chunk_index"],
expr=filter_expr,
)
# 整理召回结果
candidates = []
for hits in results:
for hit in hits:
candidates.append({
"id": hit.id,
"text": hit.entity.get("text"),
"title": hit.entity.get("title"),
"source": hit.entity.get("source"),
"doc_type": hit.entity.get("doc_type"),
"chunk_index": hit.entity.get("chunk_index"),
"retrieve_score": float(hit.score),
})
# 2. 重排序
if use_rerank and candidates:
pairs = [[query, c["text"]] for c in candidates]
rerank_scores = self.reranker.predict(pairs)
for i, score in enumerate(rerank_scores):
candidates[i]["rerank_score"] = float(score)
candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
# 截取Top K
final_results = candidates[:top_k]
query_time = (time.time() - start_time) * 1000
return {
"results": final_results,
"query_time_ms": round(query_time, 2),
"total_results": len(candidates),
}
# ==================== API服务 ====================
app = FastAPI(title="企业语义搜索系统")
engine = None
@app.on_event("startup")
async def startup():
global engine
engine = SemanticSearchEngine()
@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
result = engine.search(
query=request.query,
top_k=request.top_k,
doc_type=request.doc_type,
use_rerank=request.use_rerank,
)
return SearchResponse(**result)
@app.post("/index/upload")
async def upload_and_index(file: UploadFile = File(...), doc_type: str = "general"):
"""上传并索引文档"""
# 保存临时文件
temp_path = f"/tmp/{file.filename}"
with open(temp_path, "wb") as f:
content = await file.read()
f.write(content)
try:
engine.index_document(temp_path, title=file.filename, doc_type=doc_type)
return {"status": "success", "message": f"文档 {file.filename} 索引成功"}
except Exception as e:
raise HTTPException(500, f"索引失败: {str(e)}")
finally:
os.remove(temp_path)
@app.get("/stats")
async def get_stats():
"""获取系统统计信息"""
return {
"total_documents": engine.collection.num_entities,
"collection_name": engine.collection_name,
}
# ==================== 启动入口 ====================
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
9.4 性能优化与测试
"""
性能测试脚本
"""
import time
import requests
import concurrent.futures
import statistics
BASE_URL = "http://localhost:8000"
# 测试查询集
TEST_QUERIES = [
"如何申请退款",
"产品保修政策是什么",
"数据安全合规要求",
"员工入职流程",
"API接口文档在哪里",
"如何配置SSO单点登录",
"系统架构设计规范",
"数据库备份策略",
"代码审查流程",
"监控告警配置方法",
]
def single_request(query: str) -> dict:
"""发送单个搜索请求"""
start = time.time()
response = requests.post(
f"{BASE_URL}/search",
json={"query": query, "top_k": 10, "use_rerank": True},
)
latency = (time.time() - start) * 1000
result = response.json()
return {
"query": query,
"latency_ms": latency,
"server_time_ms": result["query_time_ms"],
"results_count": result["total_results"],
}
def run_performance_test(concurrency: int = 10, num_requests: int = 100):
"""并发性能测试"""
queries = (TEST_QUERIES * (num_requests // len(TEST_QUERIES) + 1))[:num_requests]
latencies = []
server_times = []
with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
futures = [executor.submit(single_request, q) for q in queries]
for future in concurrent.futures.as_completed(futures):
result = future.result()
latencies.append(result["latency_ms"])
server_times.append(result["server_time_ms"])
# 统计结果
print(f"\n{'='*50}")
print(f"并发数: {concurrency}")
print(f"总请求数: {num_requests}")
print(f"{'='*50}")
print(f"端到端延迟:")
print(f" 平均: {statistics.mean(latencies):.1f}ms")
print(f" P50: {statistics.median(latencies):.1f}ms")
print(f" P95: {sorted(latencies)[int(len(latencies)*0.95)]:.1f}ms")
print(f" P99: {sorted(latencies)[int(len(latencies)*0.99)]:.1f}ms")
print(f" 最小: {min(latencies):.1f}ms")
print(f" 最大: {max(latencies):.1f}ms")
print(f"\n服务端处理时间:")
print(f" 平均: {statistics.mean(server_times):.1f}ms")
print(f" P50: {statistics.median(server_times):.1f}ms")
print(f"\nQPS: {num_requests / (sum(latencies)/1000/concurrency):.1f}")
if __name__ == "__main__":
# 预热
single_request("测试查询")
# 运行性能测试
run_performance_test(concurrency=5, num_requests=50)
run_performance_test(concurrency=10, num_requests=100)
run_performance_test(concurrency=20, num_requests=200)
第十章：实战项目二——智能推荐系统
10.1 推荐系统中的Embedding应用
Embedding在推荐系统中有多种应用方式：
- 内容表征：将商品/内容的文本描述编码为Embed向量，用于基于内容的推荐
- 用户画像：将用户历史行为序列编码为用户Embed向量，用于个性化推荐
- 跨模态对齐：将商品图片和文字描述映射到同一空间，实现多模态推荐
- 冷启动：利用Embedding的语义理解能力，解决新用户/新商品的冷启动问题
10.2 基于语义的商品推荐引擎
"""
基于Embedding的商品语义推荐系统
"""
import numpy as np
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Optional
import json
import time
class ProductRecommender:
"""基于语义的商品推荐引擎"""
def __init__(self, model_name: str = "BAAI/bge-base-zh-v1.5"):
self.model = SentenceTransformer(model_name)
self.products = []
self.product_embeddings = None
self.category_embeddings = {}
def load_products(self, products: List[Dict]):
"""
加载商品数据
products: [
{
"id": "001",
"name": "无线蓝牙耳机",
"description": "高品质无线蓝牙耳机，支持降噪...",
"category": "数码配件",
"price": 299,
"tags": ["蓝牙", "耳机", "降噪"],
},
...
]
"""
self.products = products
# 构建商品的文本表示（拼接多个字段）
product_texts = []
for p in products:
text = f"{p['name']}。{p['description']}"
if p.get("tags"):
text += f"。标签：{'、'.join(p['tags'])}"
product_texts.append(text)
# 批量编码
self.product_embeddings = self.model.encode(
product_texts,
normalize_embeddings=True,
show_progress_bar=True,
batch_size=64,
)
# 构建品类Embedding
categories = set(p.get("category", "") for p in products)
for cat in categories:
if cat:
cat_products = [
i for i, p in enumerate(products)
if p.get("category") == cat
]
if cat_products:
self.category_embeddings[cat] = np.mean(
self.product_embeddings[cat_products], axis=0
)
print(f"已加载 {len(products)} 个商品，{len(self.category_embeddings)} 个品类")
def recommend_by_product(
self,
product_id: str,
top_k: int = 10,
exclude_same_category: bool = False,
) -> List[Dict]:
"""基于商品的相似商品推荐（看了又看）"""
# 找到目标商品
target_idx = None
for i, p in enumerate(self.products):
if p["id"] == product_id:
target_idx = i
break
if target_idx is None:
raise ValueError(f"商品 {product_id} 不存在")
target = self.products[target_idx]
target_emb = self.product_embeddings[target_idx]
# 计算相似度
scores = np.dot(self.product_embeddings, target_emb)
# 排除自身和同品类（可选）
candidates = []
for i, score in enumerate(scores):
if i == target_idx:
continue
if exclude_same_category and self.products[i].get("category") == target.get("category"):
continue
candidates.append((i, float(score)))
# 排序
candidates.sort(key=lambda x: x[1], reverse=True)
results = []
for idx, score in candidates[:top_k]:
product = self.products[idx].copy()
product["similarity_score"] = score
results.append(product)
return results
def recommend_by_text(
self,
query: str,
top_k: int = 10,
category_filter: Optional[str] = None,
price_range: Optional[tuple] = None,
) -> List[Dict]:
"""基于自然语言的商品推荐（以文搜品）"""
query_emb = self.model.encode([query], normalize_embeddings=True)[0]
scores = np.dot(self.product_embeddings, query_emb)
candidates = []
for i, score in enumerate(scores):
product = self.products[i]
# 品类过滤
if category_filter and product.get("category") != category_filter:
continue
# 价格过滤
if price_range:
min_price, max_price = price_range
if product.get("price", 0) < min_price or product.get("price", 0) > max_price:
continue
candidates.append((i, float(score)))
candidates.sort(key=lambda x: x[1], reverse=True)
results = []
for idx, score in candidates[:top_k]:
product = self.products[idx].copy()
product["relevance_score"] = score
results.append(product)
return results
def recommend_for_user(
self,
user_history: List[str], # 用户浏览/购买的商品ID列表
top_k: int = 20,
diversity_factor: float = 0.3,
) -> List[Dict]:
"""基于用户历史的个性化推荐"""
# 获取用户历史商品的Embedding
history_indices = []
for pid in user_history:
for i, p in enumerate(self.products):
if p["id"] == pid:
history_indices.append(i)
break
if not history_indices:
return self._get_popular_products(top_k)
# 用户向量 = 历史商品向量的加权平均（越近的权重越高）
weights = np.linspace(0.5, 1.0, len(history_indices))
user_emb = np.average(
self.product_embeddings[history_indices],
axis=0,
weights=weights,
)
user_emb = user_emb / np.linalg.norm(user_emb)
# 计算相似度
scores = np.dot(self.product_embeddings, user_emb)
# 排除已交互的商品
candidates = []
for i, score in enumerate(scores):
if i in history_indices:
continue
candidates.append((i, float(score)))
candidates.sort(key=lambda x: x[1], reverse=True)
# 多样性重排（MMR）
if diversity_factor > 0:
candidates = self._mmr_rerank(
candidates, top_k=top_k * 2, lambda_param=1 - diversity_factor
)
results = []
for idx, score in candidates[:top_k]:
product = self.products[idx].copy()
product["recommend_score"] = score
results.append(product)
return results
def _mmr_rerank(
self,
candidates: List[tuple],
top_k: int,
lambda_param: float = 0.7,
) -> List[tuple]:
"""最大边际相关性（MMR）重排，提升多样性"""
if len(candidates) <= top_k:
return candidates
selected = [candidates[0]]
remaining = candidates[1:]
while len(selected) < top_k and remaining:
best_score = -1
best_idx = -1
for i, (idx, rel_score) in enumerate(remaining):
# 与已选集合的最大相似度
max_sim = max(
np.dot(self.product_embeddings[idx], self.product_embeddings[s[0]])
for s in selected
)
# MMR分数 = λ * 相关性 - (1-λ) * 最大冗余度
mmr_score = lambda_param * rel_score - (1 - lambda_param) * max_sim
if mmr_score > best_score:
best_score = mmr_score
best_idx = i
selected.append(remaining[best_idx])
remaining.pop(best_idx)
return selected
def _get_popular_products(self, top_k: int) -> List[Dict]:
"""冷启动：返回热门商品"""
# 简化：返回随机商品（实际应基于销量/热度）
import random
indices = random.sample(range(len(self.products)), min(top_k, len(self.products)))
results = []
for idx in indices:
product = self.products[idx].copy()
product["recommend_score"] = 0.0
results.append(product)
return results
# ==================== 使用示例 ====================
def demo():
# 模拟商品数据
products = [
{
"id": "001",
"name": "Sony WH-1000XM5 无线降噪耳机",
"description": "旗舰级主动降噪无线头戴式耳机，30小时续航，多点连接，Hi-Res音质。",
"category": "数码配件",
"price": 2499,
"tags": ["降噪", "头戴式", "蓝牙", "Sony"],
},
{
"id": "002",
"name": "AirPods Pro 2",
"description": "苹果真无线降噪耳机，自适应透明模式，个性化空间音频。",
"category": "数码配件",
"price": 1899,
"tags": ["降噪", "真无线", "苹果", "AirPods"],
},
{
"id": "003",
"name": "机械键盘 Cherry MX",
"description": "德国Cherry MX红轴机械键盘，全键无冲，RGB背光，适合游戏和办公。",
"category": "电脑外设",
"price": 699,
"tags": ["机械键盘", "Cherry", "游戏"],
},
{
"id": "004",
"name": "4K高清显示器 27英寸",
"description": "27英寸4K IPS显示器，HDR400，Type-C 65W反向充电，专业设计修图。",
"category": "电脑外设",
"price": 3299,
"tags": ["4K", "显示器", "设计"],
},
{
"id": "005",
"name": "Python编程入门到精通",
"description": "零基础Python编程教程，涵盖基础语法、数据分析、Web开发、机器学习。",
"category": "图书",
"price": 89,
"tags": ["Python", "编程", "教程"],
},
{
"id": "006",
"name": "深度学习实战指南",
"description": "使用PyTorch和TensorFlow进行深度学习项目实战，涵盖CV、NLP、推荐系统。",
"category": "图书",
"price": 109,
"tags": ["深度学习", "PyTorch", "AI"],
},
]
# 初始化推荐系统
recommender = ProductRecommender()
recommender.load_products(products)
print("\n" + "="*60)
print("1. 基于商品的相似推荐（看了又看）")
print("="*60)
similar = recommender.recommend_by_product("001", top_k=3)
print(f"与 Sony WH-1000XM5 相似的商品:")
for p in similar:
print(f" [{p['similarity_score']:.4f}] {p['name']} - ¥{p['price']}")
print("\n" + "="*60)
print("2. 基于自然语言的商品推荐")
print("="*60)
results = recommender.recommend_by_text("想买个降噪耳机通勤用", top_k=3)
print(f"'想买个降噪耳机通勤用' 推荐结果:")
for p in results:
print(f" [{p['relevance_score']:.4f}] {p['name']} - ¥{p['price']}")
print("\n" + "="*60)
print("3. 基于用户历史的个性化推荐")
print("="*60)
personalized = recommender.recommend_for_user(
user_history=["001", "002"], # 用户浏览过两款耳机
top_k=3,
)
print(f"浏览过两款耳机的用户推荐:")
for p in personalized:
print(f" [{p['recommend_score']:.4f}] {p['name']} - ¥{p['price']}")
if __name__ == "__main__":
demo()
10.3 冷启动与实时更新策略
冷启动问题的Embedding解决方案：
- 新商品冷启动：利用商品的文本描述（名称、类目、属性）生成Embedding，与已有商品进行相似度匹配
- 新用户冷启动：通过用户注册信息（如兴趣标签）或热门商品Embedding进行推荐
class ColdStartHandler:
"""冷启动处理器"""
def __init__(self, recommender: ProductRecommender):
self.recommender = recommender
def handle_new_product(self, product: Dict) -> List[Dict]:
"""新商品上架时，找到最相似的商品用于初始化推荐"""
# 编码新商品
text = f"{product['name']}。{product['description']}"
new_emb = self.recommender.model.encode([text], normalize_embeddings=True)[0]
# 找相似商品
scores = np.dot(self.recommender.product_embeddings, new_emb)
top_indices = np.argsort(scores)[::-1][:5]
similar_products = []
for idx in top_indices:
similar_products.append({
"product": self.recommender.products[idx],
"similarity": float(scores[idx]),
})
# 将新商品加入索引
self.recommender.products.append(product)
new_emb_reshaped = new_emb.reshape(1, -1)
self.recommender.product_embeddings = np.vstack([
self.recommender.product_embeddings,
new_emb_reshaped,
])
return similar_products
def handle_new_user(self, interests: List[str]) -> List[Dict]:
"""新用户注册时，基于兴趣标签推荐"""
# 编码兴趣标签
interest_emb = self.recommender.model.encode(
interests,
normalize_embeddings=True,
)
user_emb = np.mean(interest_emb, axis=0)
user_emb = user_emb / np.linalg.norm(user_emb)
# 推荐商品
scores = np.dot(self.recommender.product_embeddings, user_emb)
top_indices = np.argsort(scores)[::-1][:10]
results = []
for idx in top_indices:
product = self.recommender.products[idx].copy()
product["score"] = float(scores[idx])
results.append(product)
return results
第十一章：常见问题与故障排查
Q1: Embedding模型返回的向量全为0或NaN
可能原因：
- 输入文本为空或过长（超出模型最大长度）
- 模型权重加载失败
- 使用了错误的前缀/指令格式
解决方案：
# 检查输入
def validate_input(texts, max_length=512):
valid_texts = []
for t in texts:
if not t or not t.strip():
print(f"警告: 跳过空文本")
continue
if len(t) > max_length * 4: # 粗略估算字符/token比
print(f"警告: 文本过长({len(t)}字符)，将截断")
t = t[:max_length * 4]
valid_texts.append(t)
return valid_texts
# 检查输出
def check_embeddings(embeddings):
if np.any(np.isnan(embeddings)):
print("错误: Embedding包含NaN!")
if np.all(embeddings == 0):
print("错误: Embedding全为0!")
if not np.allclose(np.linalg.norm(embeddings, axis=1), 1.0, atol=0.01):
print("警告: Embedding未归一化")
Q2: 检索结果不相关，语义匹配效果差
可能原因：
- 模型选择不当（通用模型 vs 领域模型）
- 查询和文档使用了不同的编码方式（如查询加了前缀但文档没加）
- Embedding维度不够
- 分块策略不合理
解决方案：
# 1. 确保查询和文档使用一致的编码方式
# BGE: 查询加指令前缀，文档不加
query = "为这个句子生成表示以用于检索中文文档：什么是退款政策"
doc = "退款政策：购买后7天内可申请退款..."
# 2. 检查分块是否合理
# 不好的分块：把一句话切成两半
# 好的分块：按语义完整的段落切分
# 3. 尝试更大的模型
# bge-small → bge-base → bge-large
Q3: 检索速度慢，延迟高
可能原因：
- 向量维度太高
- 索引类型不适合数据规模
- 没有使用GPU加速
- 没有使用批量编码
解决方案：
# 1. 使用量化降低向量维度
# float32 (768维) → binary (96字节)
# 2. 选择合适的索引类型
# < 10K向量: 暴力搜索(FLAT)
# 10K-1M: HNSW
# > 1M: IVF_FLAT / IVF_PQ
# 3. 使用GPU加速
model = SentenceTransformer("BAAI/bge-base-zh-v1.5", device="cuda")
# 4. 批量编码
embeddings = model.encode(texts, batch_size=128) # 而不是逐条编码
Q4: 内存占用过高
可能原因：
- 模型太大
- 向量未压缩
- 所有向量加载到内存
解决方案：
# 1. 使用更小的模型
# bge-large (326M) → bge-base (102M) → bge-small (24M)
# 2. 向量量化
# float32 → float16 (2x) → int8 (4x) → binary (32x)
# 3. 使用内存映射
import numpy as np
embeddings = np.memmap("embeddings.bin", dtype="float32", mode="r", shape=(1000000, 768))
# 4. 分片加载
# 不把所有向量加载到内存，而是按需加载
Q5: 多语言检索效果不一致
可能原因：
- 模型的多语言能力不足
- 不同语言的文本长度差异导致分块不一致
- 缺少语言特定的前缀/指令
解决方案：
# 1. 使用多语言模型
# bge-m3, multilingual-e5-large, Cohere Embed v3
# 2. 语言检测后再分块
from langdetect import detect
def multilingual_chunk(text, chunk_size=500):
lang = detect(text)
if lang == "zh":
# 中文按字符数分块
return chunk_by_chars(text, chunk_size)
else:
# 英文按词数分块
return chunk_by_words(text, chunk_size // 4) # 英文单词平均4字符
Q6: 如何评估Embedding质量？
def evaluate_embedding_quality(model, test_data):
"""
test_data: [
{
"query": "...",
"positive": ["相关文档1", "相关文档2"],
"negative": ["不相关文档1", "不相关文档2"]
},
...
]
"""
mrr_scores = []
recall_at_10 = []
for item in test_data:
all_docs = item["positive"] + item["negative"]
q_emb = model.encode([item["query"]], normalize_embeddings=True)
d_embs = model.encode(all_docs, normalize_embeddings=True)
scores = (q_emb @ d_embs.T)[0]
ranked = np.argsort(scores)[::-1]
# MRR
for rank, idx in enumerate(ranked):
if idx < len(item["positive"]):
mrr_scores.append(1.0 / (rank + 1))
break
else:
mrr_scores.append(0.0)
# Recall@10
top10 = ranked[:10]
relevant_in_top10 = sum(1 for idx in top10 if idx < len(item["positive"]))
recall_at_10.append(relevant_in_top10 / len(item["positive"]))
return {
"MRR": np.mean(mrr_scores),
"Recall@10": np.mean(recall_at_10),
}
附录：学习资源与参考文献
核心论文
- Attention Is All You Need (Vaswani et al., 2017) - Transformer架构
- BERT: Pre-training of Deep Bidirectional Transformers (Devlin et al., 2019)
- Sentence-BERT (Reimers & Gurevych, 2019) - 句子Embedding
- SimCSE (Gao et al., 2021) - 对比学习句子Embedding
- BGE (Xiao et al., 2023) - 通用Embedding模型
- E5 (Wang et al., 2024) - 微软Embedding模型
- CLIP (Radford et al., 2021) - 多模态Embedding
- Cohere Embed v3 - 商业Embedding模型
开源工具库
| 工具 | 用途 | 链接 |
|---|---|---|
| sentence-transformers | Embedding训练与推理 | github.com/UKPLab/sentence-transformers |
| FlagEmbedding | BGE模型官方实现 | github.com/FlagOpen/FlagEmbedding |
| MTEB | Embedding评估基准 | github.com/embeddings-benchmark/mteb |
| Milvus | 分布式向量数据库 | github.com/milvus-io/milvus |
| Qdrant | 高性能向量数据库 | github.com/qdrant/qdrant |
| ChromaDB | 轻量级向量数据库 | github.com/chroma-core/chroma |
| Faiss | 向量检索库 | github.com/facebookresearch/faiss |
推荐学习路径
入门（1-2周）
├── 理解Embedding基本概念
├── 使用sentence-transformers生成Embedding
├── 构建简单的语义搜索Demo
└── 了解余弦相似度、欧氏距离等度量方式
进阶（2-4周）
├── 对比不同Embedding模型（BGE、E5、GTE）
├── 学习重排序模型的使用
├── 构建混合检索系统
├── 学习向量数据库的使用
└── 理解量化与压缩技术
高级（1-2月）
├── 微调领域Embedding模型
├── 设计生产级检索架构
├── 优化检索延迟和吞吐量
├── 构建RAG系统
└── 多模态Embedding应用
本教程持续更新中。 Embedding与重排序技术发展迅速，建议定期关注MTEB排行榜和各模型的最新版本。
