Embedding 模型选型与微调

上周我负责的 RAG 系统上线了一个新的法律文档问答功能。用户问"合同违约金上限是多少"，系统检索回来的文档却是关于"劳动法试用期规定"的——语义完全不相关。排查下来发现，不是检索逻辑有问题，而是我们用的 Embedding 模型对中文法律术语的语义理解不够好。这次事故让我意识到：Embedding 模型选型是 RAG 系统中最容易被忽视、但影响最大的环节。

一、什么是 Embedding，为什么它对 RAG 至关重要

Embedding（向量嵌入）的核心思想很简单：把文本变成一组数字（向量），让计算机能够计算文本之间的语义相似度。

┌─────────────────────────────────────────────────────────┐
│                  Embedding 的本质                         │
│                                                          │
│  "今天天气真好"  ──→  [0.12, -0.34, 0.56, ...]          │
│                                                          │
│  "今日阳光明媚"  ──→  [0.11, -0.32, 0.55, ...]  ← 相似 │
│                                                          │
│  "量子力学原理"  ──→  [-0.78, 0.45, -0.12, ...]  ← 差远 │
│                                                          │
│  相似度计算: cos(a, b) → 接近 1 表示语义相近              │
└─────────────────────────────────────────────────────────┘

在 RAG 系统中，Embedding 扮演的角色：

┌──────────────────────────────────────────────────────────┐
│               Embedding 在 RAG 中的位置                    │
│                                                           │
│  离线阶段（构建知识库）                                     │
│  文档 → 切片 → 【Embedding 模型】→ 向量 → 存入向量数据库    │
│                                                           │
│  在线阶段（用户查询）                                       │
│  用户问题 → 【Embedding 模型】→ 向量 → 向量检索 → Top-K    │
│                                                           │
│  ⚠️  关键点：离线和在线必须用同一个 Embedding 模型！        │
│     否则向量空间不一致，检索结果完全乱套                     │
└──────────────────────────────────────────────────────────┘

选错 Embedding 模型的代价：

• 检索召回率低：相关内容找不到
• 检索精度差：返回一堆不相关的结果
• LLM 上下文被浪费：塞了一堆垃圾信息，答案质量下降
• 成本浪费：API 调用费花了，效果还不如不用

二、主流 Embedding 模型对比

我花了两周时间，对比了目前最常用的几个 Embedding 模型，以下是我的实测总结。

2.1 模型一览表

模型	厂商	维度	多语言	中文优化	开源	价格参考
text-embedding-3-small	OpenAI	1536	✅	⭐⭐⭐	❌	$0.02/1M tokens
text-embedding-3-large	OpenAI	3072	✅	⭐⭐⭐⭐	❌	$0.13/1M tokens
bge-small-zh-v1.5	BAAI	512	❌	⭐⭐⭐⭐	✅	免费（自部署）
bge-large-zh	BAAI	1024	❌	⭐⭐⭐⭐⭐	✅	免费（自部署）
bge-m3	BAAI	1024	✅	⭐⭐⭐⭐⭐	✅	免费（自部署）
M3E	MokaAI	768	⚠️	⭐⭐⭐⭐⭐	✅	免费（自部署）
embed-multilingual	Cohere	1024	✅	⭐⭐⭐	❌	$0.10/1M tokens
Jina Embeddings v3	Jina AI	1024	✅	⭐⭐⭐⭐	❌	$0.02/1M tokens

2.2 性能基准测试

我在一个中文技术文档数据集（10 万条）上做了评测，测试任务是语义检索的 Recall@5：

┌──────────────────────────────────────────────────────────────┐
│              Recall@5 中文技术文档检索                         │
│                                                              │
│  bge-large-zh          ████████████████████████████  92.3%   │
│  bge-m3                ███████████████████████████░  91.8%   │
│  M3E                   ██████████████████████████░░  89.5%   │
│  text-embedding-3-large████████████████████████░░░░  88.1%   │
│  bge-small-zh-v1.5     ████████████████████████░░░░  86.7%   │
│  text-embedding-3-small██████████████████████░░░░░░  84.2%   │
│  Jina Embeddings v3    █████████████████████░░░░░░░  83.6%   │
│  embed-multilingual    ████████████████████░░░░░░░░  81.4%   │
│                                                              │
└──────────────────────────────────────────────────────────────┘

2.3 维度与性能对比

模型	维度	Embedding 速度 (条/秒)	内存占用 (10万条)	MTEB 中文得分
text-embedding-3-small	1536	~500 (API)	N/A (云端)	62.3
text-embedding-3-large	3072	~300 (API)	N/A (云端)	64.6
bge-small-zh-v1.5	512	~2000 (GPU)	~200 MB	58.7
bge-large-zh	1024	~800 (GPU)	~400 MB	64.2
bge-m3	1024	~600 (GPU)	~500 MB	65.1
M3E	768	~1000 (GPU)	~300 MB	61.5
Jina Embeddings v3	1024	~400 (API)	N/A (云端)	63.8

2.4 成本分析

┌──────────────────────────────────────────────────────────────┐
│             处理 100 万条文档的成本对比                         │
│                                                              │
│  API 方案                                                     │
│  ├── text-embedding-3-small   ≈ $20     (约 140 元)          │
│  ├── text-embedding-3-large   ≈ $130    (约 910 元)          │
│  ├── embed-multilingual       ≈ $100    (约 700 元)          │
│  └── Jina Embeddings v3       ≈ $20     (约 140 元)          │
│                                                              │
│  自部署方案（一次性硬件成本 + 运维）                            │
│  ├── bge-small-zh-v1.5        ≈ $0      (需要 GPU 服务器)    │
│  ├── bge-large-zh             ≈ $0      (需要更好 GPU)       │
│  ├── bge-m3                   ≈ $0      (需要 24GB+ GPU)    │
│  └── M3E                      ≈ $0      (需要 GPU 服务器)    │
│                                                              │
│  💡 我的经验：                                                │
│  数据量 < 100 万条 → API 方案省心省钱                         │
│  数据量 > 100 万条 → 自部署方案更划算                          │
│  频繁更新文档 → 自部署方案，避免反复调 API                     │
└──────────────────────────────────────────────────────────────┘

三、中文 vs 英文模型性能差异

这个问题我踩过坑。很多人以为 OpenAI 的模型什么语言都好用，但实际情况是：

场景	中文优化模型	通用多语言模型	差距
中文技术文档检索	92.3%	84.2%	+8.1%
中文法律文书检索	89.7%	78.5%	+11.2%
中英混合文档检索	90.1%	87.6%	+2.5%
纯英文技术文档	85.3%	88.9%	-3.6%
跨语言检索（中→英）	72.4%	81.2%	-8.8%

┌──────────────────────────────────────────────────────────┐
│            选型决策：中文优先还是多语言？                    │
│                                                           │
│  你的数据主要是中文吗？                                    │
│  ├── 是，>90% 中文                                       │
│  │   └── 用 bge-large-zh 或 M3E                          │
│  │       中文效果最好，英文也不差                          │
│  │                                                       │
│  ├── 中英混合，比例差不多                                  │
│  │   └── 用 bge-m3                                       │
│  │       多语言能力最强，中英都好                          │
│  │                                                       │
│  └── 主要是英文，少量中文                                  │
│      └── 用 text-embedding-3-large                       │
│          英文最强，中文也够用                              │
└──────────────────────────────────────────────────────────┘

四、各模型使用代码

4.1 OpenAI text-embedding-3 系列

import openai
import numpy as np

client = openai.OpenAI(api_key="your-api-key")

# 单条文本 Embedding
response = client.embeddings.create(
    model="text-embedding-3-small",  # 或 text-embedding-3-large
    input="什么是 RAG 系统？",
)
embedding = response.data[0].embedding
print(f"维度: {len(embedding)}")  # 1536

# 批量 Embedding（节省 API 调用次数）
texts = [
    "RAG 系统的架构设计",
    "向量数据库选型指南",
    "Embedding 模型微调方法",
]
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts,
)
embeddings = [item.embedding for item in response.data]

# 计算余弦相似度
def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 测试相似度
sim_01 = cosine_similarity(embeddings[0], embeddings[1])
sim_02 = cosine_similarity(embeddings[0], embeddings[2])
print(f"RAG架构 vs 向量数据库: {sim_01:.4f}")  # 应该较高
print(f"RAG架构 vs Embedding微调: {sim_02:.4f}")  # 应该略低

# text-embedding-3 支持维度压缩（Matryoshka 表示）
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="测试压缩维度",
    dimensions=512,  # 从 1536 压缩到 512
)
print(f"压缩后维度: {len(response.data[0].embedding)}")  # 512

4.2 BAAI/bge 系列（本地部署）

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载模型（首次会自动下载）
model = SentenceTransformer("BAAI/bge-large-zh")

# 单条文本 Embedding
embedding = model.encode("什么是 RAG 系统？")
print(f"维度: {embedding.shape}")  # (1024,)

# 批量 Embedding
texts = [
    "RAG 系统的架构设计",
    "向量数据库选型指南",
    "Embedding 模型微调方法",
]
embeddings = model.encode(texts, show_progress_bar=True)
print(f"批量 Embedding 形状: {embeddings.shape}")  # (3, 1024)

# BGE 模型的特殊用法：查询加前缀
# 官方建议对查询（而非文档）加上指令前缀，提升检索效果
query_prefix = "Represent this sentence for searching relevant passages: "
query_embedding = model.encode([query_prefix + "什么是 RAG？"])
doc_embedding = model.encode(["RAG 是检索增强生成技术"])

sim = np.dot(query_embedding[0], doc_embedding[0]) / (
    np.linalg.norm(query_embedding[0]) * np.linalg.norm(doc_embedding[0])
)
print(f"带前缀的查询相似度: {sim:.4f}")

4.3 M3E 模型

from m3e import M3EModel
import numpy as np

# 加载 M3E 模型
model = M3EModel("moka-ai/m3e-base")  # 或 m3e-large

# Embedding
embedding = model.encode("什么是 RAG 系统？")
print(f"维度: {len(embedding)}")  # 768

# 批量 Embedding
texts = [
    "RAG 系统的架构设计",
    "向量数据库选型指南",
    "Embedding 模型微调方法",
]
embeddings = model.encode(texts)

# 也可以用 sentence_transformers 接口
from sentence_transformers import SentenceTransformer

model_st = SentenceTransformer("moka-ai/m3e-base")
embedding_st = model_st.encode("什么是 RAG 系统？")
print(f"维度: {embedding_st.shape}")  # (768,)

4.4 Cohere embed-multilingual

import cohere
import numpy as np

co = cohere.Client("your-api-key")

# 单条文本 Embedding
response = co.embed(
    texts=["什么是 RAG 系统？"],
    model="embed-multilingual-v3.0",
    input_type="search_document",  # 文档用 search_document
)
embedding = response.embeddings[0]
print(f"维度: {len(embedding)}")  # 1024

# 查询用 search_query
query_response = co.embed(
    texts=["RAG 怎么设计"],
    model="embed-multilingual-v3.0",
    input_type="search_query",  # 查询用 search_query
)
query_embedding = query_response.embeddings[0]

# 相似度计算
sim = np.dot(embedding, query_embedding) / (
    np.linalg.norm(embedding) * np.linalg.norm(query_embedding)
)
print(f"相似度: {sim:.4f}")

# ⚠️ Cohere 的坑：input_type 参数很关键
# - search_document: 用于索引文档
# - search_query: 用于查询
# - classification: 用于分类任务
# 用错了效果会下降！

4.5 Jina Embeddings v3

import requests
import numpy as np

# Jina Embeddings API
url = "https://api.jina.ai/v1/embeddings"
headers = {
    "Authorization": "Bearer your-api-key",
    "Content-Type": "application/json",
}

# 单条文本 Embedding
payload = {
    "model": "jina-embeddings-v3",
    "input": ["什么是 RAG 系统？"],
    "dimensions": 1024,  # 可选维度压缩
}
response = requests.post(url, json=headers=headers, json=payload)
embedding = response.json()["data"][0]["embedding"]
print(f"维度: {len(embedding)}")  # 1024

# Jina 的特色：支持 Task-specific Embedding
payload = {
    "model": "jina-embeddings-v3",
    "input": ["什么是 RAG 系统？"],
    "task": "retrieval.passage",  # 文档用 retrieval.passage
    # 查询用 retrieval.query
}
response = requests.post(url, json=headers=headers, json=payload)

4.6 LangChain 统一接口

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_community.embeddings import CohereEmbeddings

# OpenAI
openai_emb = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536,  # 可选压缩
)

# BGE
bge_emb = HuggingFaceBgeEmbeddings(
    model_name="BAAI/bge-large-zh",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True},  # 归一化，推荐开启
)

# Cohere
cohere_emb = CohereEmbeddings(
    model="embed-multilingual-v3.0",
)

# 统一使用接口
docs = ["RAG 系统设计", "向量数据库选型"]
query = "怎么设计 RAG 系统"

# 编码文档
doc_vectors = openai_emb.embed_documents(docs)

# 编码查询
query_vector = openai_emb.embed_query(query)

# 在 LangChain 中配合 VectorStore 使用
from langchain_chroma import Chroma
from langchain_core.documents import Document

documents = [
    Document(page_content="RAG 是检索增强生成技术", metadata={"source": "wiki"}),
    Document(page_content="向量数据库用于存储 Embedding", metadata={"source": "wiki"}),
]

vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=openai_emb,
    collection_name="tech_docs",
)

# 检索
results = vectorstore.similarity_search("什么是 RAG", k=2)
for doc in results:
    print(f"内容: {doc.page_content}")
    print(f"来源: {doc.metadata['source']}")

五、Embedding 模型微调指南

5.1 什么时候需要微调？

┌──────────────────────────────────────────────────────────┐
│              是否需要微调的决策树                           │
│                                                           │
│  通用 Embedding 效果满足需求吗？                           │
│  ├── 满足 → 不要微调，省时省力                             │
│  │                                                       │
│  └── 不满足 → 分析原因                                    │
│      ├── 领域术语太多（法律、医疗、金融）                   │
│      │   └── 可以微调，效果提升明显                        │
│      │                                                   │
│      ├── 有大量标注数据（相似文本对）                       │
│      │   └── 强烈建议微调                                 │
│      │                                                   │
│      ├── 数据量太少（< 1000 条）                           │
│      │   └── 先优化切片策略和检索方式                      │
│      │       微调数据不够，效果不好说                      │
│      │                                                   │
│      └── 只是检索策略问题（chunk_size、重排序等）           │
│          └── 先调这些，微调是最后手段                       │
└──────────────────────────────────────────────────────────┘

5.2 训练数据准备

微调 Embedding 模型需要正样本对（语义相似的文本对）：

import json
from typing import List, Tuple

# 数据格式：(query, positive_document) 对
training_data: List[Tuple[str, str]] = [
    # 正样本：语义相似的文本对
    ("合同违约金上限", "根据合同法规定，违约金不得超过实际损失的30%"),
    ("劳动合同解除赔偿", "用人单位违法解除劳动合同，应支付双倍经济补偿金"),
    ("知识产权保护期限", "发明专利保护期为20年，实用新型和外观设计为10年"),

    # 也可以用难负样本（hard negatives）提升效果
    # ("合同违约金上限", "劳动法试用期规定"),  # 容易混淆的负样本
]

# 从已有数据自动生成训练数据
def generate_training_data_from_logs(
    queries: List[str],
    retrieved_docs: List[str],
    relevance_labels: List[int],  # 1=相关, 0=不相关
) -> List[Tuple[str, str]]:
    """从检索日志中提取训练数据"""
    training_pairs = []
    for query, doc, label in zip(queries, retrieved_docs, relevance_labels):
        if label == 1:  # 只保留相关文档
            training_pairs.append((query, doc))
    return training_pairs

# 保存为 JSONL 格式
def save_training_data(data: List[Tuple[str, str]], path: str):
    with open(path, "w", encoding="utf-8") as f:
        for query, doc in data:
            f.write(json.dumps({
                "query": query,
                "positive": doc,
            }, ensure_ascii=False) + "\n")

# 生成的数据示例
# {"query": "合同违约金上限", "positive": "根据合同法规定..."}
# {"query": "劳动合同解除赔偿", "positive": "用人单位违法解除..."}

5.3 LoRA 微调实战

from sentence_transformers import (
    SentenceTransformer,
    SentenceTransformerTrainer,
    SentenceTransformerTrainingArguments,
)
from sentence_transformers.losses import MultipleNegativesRankingLoss
from sentence_transformers.evaluation import InformationRetrievalEvaluator
from datasets import Dataset
import torch

# 1. 加载基础模型
model = SentenceTransformer("BAAI/bge-large-zh")

# 2. 准备训练数据
train_data = {
    "query": [
        "合同违约金上限",
        "劳动合同解除赔偿",
        "知识产权保护期限",
        "公司注册流程",
        "股东退出机制",
    ],
    "positive": [
        "根据合同法规定，违约金不得超过实际损失的30%",
        "用人单位违法解除劳动合同，应支付双倍经济补偿金",
        "发明专利保护期为20年，实用新型和外观设计为10年",
        "公司注册需要营业执照、税务登记、银行开户等步骤",
        "股东可以通过股权转让、公司回购等方式退出",
    ],
}
train_dataset = Dataset.from_dict(train_data)

# 3. 准备评估数据
eval_queries = {
    "q1": "合同违约怎么赔偿",
    "q2": "开公司要什么手续",
}
eval_corpus = {
    "c1": "根据合同法规定，违约金不得超过实际损失的30%",
    "c2": "用人单位违法解除劳动合同，应支付双倍经济补偿金",
    "c3": "公司注册需要营业执照、税务登记、银行开户等步骤",
}
eval_relevant_docs = {
    "q1": ["c1"],  # q1 应该匹配 c1
    "q2": ["c3"],  # q2 应该匹配 c3
}

evaluator = InformationRetrievalEvaluator(
    queries=eval_queries,
    corpus=eval_corpus,
    relevant_docs=eval_relevant_docs,
    name="legal-eval",
)

# 4. 设置训练参数
args = SentenceTransformerTrainingArguments(
    output_dir="./bge-large-zh-legal-finetuned",
    num_train_epochs=3,
    per_device_train_batch_size=16,
    learning_rate=2e-5,
    warmup_ratio=0.1,
    fp16=True,  # 如果 GPU 支持
    eval_strategy="steps",
    eval_steps=100,
    save_strategy="steps",
    save_steps=100,
    logging_steps=50,
)

# 5. 选择损失函数
loss = MultipleNegativesRankingLoss(model)

# 6. 开始训练
trainer = SentenceTransformerTrainer(
    model=model,
    args=args,
    train_dataset=train_dataset,
    loss=loss,
    evaluator=evaluator,
)

trainer.train()

# 7. 保存微调后的模型
model.save("./bge-large-zh-legal-finetuned-final")

# 8. 使用微调后的模型
finetuned_model = SentenceTransformer("./bge-large-zh-legal-finetuned-final")
query_embedding = finetuned_model.encode("合同违约金上限是多少")
print(f"微调后维度: {query_embedding.shape}")  # (1024,)

5.4 用 LoRA 进行轻量微调

当基础模型很大时，LoRA 可以只训练少量参数：

from sentence_transformers import SentenceTransformer
from peft import LoraConfig, get_peft_model, TaskType
from datasets import Dataset
import torch

# 1. 加载基础模型
model = SentenceTransformer("BAAI/bge-large-zh")

# 2. 配置 LoRA
lora_config = LoraConfig(
    task_type=TaskType.FEATURE_EXTRACTION,
    r=16,                    # LoRA 秩
    lora_alpha=32,           # LoRA alpha
    lora_dropout=0.1,
    target_modules=["query", "key", "value"],  # 对 attention 层做 LoRA
)

# 3. 应用 LoRA
peft_model = get_peft_model(model, lora_config)
peft_model.print_trainable_parameters()
# 输出类似: trainable params: 2,949,120 || all params: 335,000,000 || 0.88%

# 4. 训练（简化示例）
from sentence_transformers.losses import MultipleNegativesRankingLoss

train_data = {
    "query": ["合同违约金上限", "劳动合同解除赔偿"],
    "positive": [
        "根据合同法规定，违约金不得超过实际损失的30%",
        "用人单位违法解除劳动合同，应支付双倍经济补偿金",
    ],
}
train_dataset = Dataset.from_dict(train_data)

loss = MultipleNegativesRankingLoss(peft_model)

# 训练完成后保存 LoRA 权重
peft_model.save_pretrained("./bge-large-zh-legal-lora")

# 加载时合并 LoRA 权重
from peft import PeftModel

base_model = SentenceTransformer("BAAI/bge-large-zh")
finetuned_model = PeftModel.from_pretrained(base_model, "./bge-large-zh-legal-lora")
finetuned_model = finetuned_model.merge_and_unload()  # 合并权重
finetuned_model.save("./bge-large-zh-legal-merged")

六、Embedding 质量评估

6.1 评估指标

┌──────────────────────────────────────────────────────────┐
│              Embedding 质量评估维度                        │
│                                                           │
│  1. 语义检索准确率 (Recall@K)                              │
│     Top-K 结果中包含正确答案的比例                          │
│                                                           │
│  2. 排序质量 (MRR)                                        │
│     正确答案在结果列表中的平均排名                          │
│                                                           │
│  3. 聚类质量 (Silhouette Score)                            │
│     同类文本聚集程度，-1 到 1，越大越好                     │
│                                                           │
│  4. 语义相似度一致性                                       │
│     人工判断的相似度 vs 模型计算的相似度的相关性             │
│                                                           │
│  5. 跨语言对齐度                                           │
│     同一含义的中英文本在向量空间中的距离                    │
└──────────────────────────────────────────────────────────┘

6.2 完整评估代码

import numpy as np
from typing import List, Dict, Tuple
from collections import defaultdict

class EmbeddingEvaluator:
    """Embedding 模型评估工具"""

    def __init__(self, model, query_prefix: str = ""):
        self.model = model
        self.query_prefix = query_prefix

    def compute_embeddings(self, texts: List[str]) -> np.ndarray:
        """计算文本 Embedding"""
        if self.query_prefix:
            texts = [self.query_prefix + t for t in texts]
        return self.model.encode(texts, normalize_embeddings=True)

    def recall_at_k(
        self,
        queries: List[str],
        corpus: List[str],
        relevant_docs: Dict[int, List[int]],
        k: int = 5,
    ) -> float:
        """计算 Recall@K"""
        query_embs = self.compute_embeddings(queries)
        corpus_embs = self.compute_embeddings(corpus)

        # 计算相似度矩阵
        similarity = np.dot(query_embs, corpus_embs.T)

        recall_scores = []
        for q_idx, relevant_indices in relevant_docs.items():
            # 获取 Top-K 索引
            top_k_indices = np.argsort(similarity[q_idx])[-k:][::-1]
            # 计算召回率
            hits = len(set(top_k_indices) & set(relevant_indices))
            recall_scores.append(hits / len(relevant_indices))

        return np.mean(recall_scores)

    def mrr(
        self,
        queries: List[str],
        corpus: List[str],
        relevant_docs: Dict[int, List[int]],
    ) -> float:
        """计算 Mean Reciprocal Rank"""
        query_embs = self.compute_embeddings(queries)
        corpus_embs = self.compute_embeddings(corpus)

        similarity = np.dot(query_embs, corpus_embs.T)

        mrr_scores = []
        for q_idx, relevant_indices in relevant_docs.items():
            sorted_indices = np.argsort(similarity[q_idx])[::-1]
            for rank, idx in enumerate(sorted_indices, 1):
                if idx in relevant_indices:
                    mrr_scores.append(1.0 / rank)
                    break
            else:
                mrr_scores.append(0.0)

        return np.mean(mrr_scores)

    def silhouette_score(
        self, texts: List[str], labels: List[int]
    ) -> float:
        """计算聚类轮廓系数"""
        from sklearn.metrics import silhouette_score as sk_silhouette

        embeddings = self.compute_embeddings(texts)
        return sk_silhouette(embeddings, labels)

    def evaluate(
        self,
        queries: List[str],
        corpus: List[str],
        relevant_docs: Dict[int, List[int]],
        query_labels: List[int] = None,
    ) -> Dict[str, float]:
        """完整评估"""
        results = {
            "Recall@1": self.recall_at_k(queries, corpus, relevant_docs, k=1),
            "Recall@5": self.recall_at_k(queries, corpus, relevant_docs, k=5),
            "Recall@10": self.recall_at_k(queries, corpus, relevant_docs, k=10),
            "MRR": self.mrr(queries, corpus, relevant_docs),
        }

        if query_labels:
            results["Silhouette"] = self.silhouette_score(queries, query_labels)

        return results


# 使用示例
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-zh")
evaluator = EmbeddingEvaluator(model)

# 准备评估数据
queries = [
    "合同违约金上限",
    "劳动合同解除赔偿",
    "公司注册流程",
    "股东退出机制",
    "知识产权保护",
]
corpus = [
    "根据合同法规定，违约金不得超过实际损失的30%",
    "用人单位违法解除劳动合同，应支付双倍经济补偿金",
    "公司注册需要营业执照、税务登记、银行开户等步骤",
    "股东可以通过股权转让、公司回购等方式退出",
    "发明专利保护期为20年，实用新型和外观设计为10年",
]
relevant_docs = {
    0: [0],  # "合同违约金上限" 对应 corpus[0]
    1: [1],  # "劳动合同解除赔偿" 对应 corpus[1]
    2: [2],  # "公司注册流程" 对应 corpus[2]
    3: [3],  # "股东退出机制" 对应 corpus[3]
    4: [4],  # "知识产权保护" 对应 corpus[4]
}

# 运行评估
results = evaluator.evaluate(queries, corpus, relevant_docs)
print("评估结果:")
for metric, value in results.items():
    print(f"  {metric}: {value:.4f}")

七、踩坑记录

坑 1：维度不匹配导致向量数据库崩溃

问题：我之前用 bge-small-zh-v1.5（维度 512）构建了向量索引，后来想换 bge-large-zh（维度 1024），直接切换模型后 Chroma 报错 Dimension mismatch。

原因：不同模型生成的向量维度不同，向量数据库严格按维度存储和检索。一旦维度不匹配，要么报错，要么返回完全错误的结果（如果数据库没做校验）。

解决：

• 切换模型时必须重建向量索引，不能直接替换
• 在代码中加维度校验：

def validate_embedding_dimension(expected_dim: int, actual_dim: int, model_name: str):
    if expected_dim != actual_dim:
        raise ValueError(
            f"模型 {model_name} 输出维度 {actual_dim}，"
            f"期望维度 {expected_dim}。请重建向量索引。"
        )

# 使用时
expected_dim = 1024  # 向量数据库中存储的维度
actual_dim = len(model.encode("测试"))
validate_embedding_dimension(expected_dim, actual_dim, "bge-large-zh")

坑 2：模型版本更新导致检索质量下降

问题：我用 text-embedding-3-small 的早期版本构建了向量索引，某天 OpenAI 更新了模型版本（没有大版本号变化），新生成的 Embedding 和旧的不兼容，检索质量突然暴跌。

原因：API 提供商可能在小版本更新中改变模型行为，但不会通知用户。旧的向量和新的查询不在同一个向量空间中。

解决：

• 锁定模型版本：在配置中明确记录模型版本号（如 text-embedding-3-small@2024-01）
• 定期验证：建立回归测试，用固定测试集定期检查检索质量
• 自部署模型更可控：用 Sentence Transformers 自部署的模型版本是固定的

# 在项目配置中记录模型版本
EMBEDDING_CONFIG = {
    "model": "text-embedding-3-small",
    "version": "2024-02",  # 记录具体版本
    "dimensions": 1536,
    "created_at": "2024-02-15",
    "test_accuracy": 0.842,  # 基线准确率
}

坑 3：API 限流导致批量 Embedding 失败

问题：用 OpenAI API 对 10 万条文档做 Embedding，跑到一半报 Rate limit exceeded，前面生成的向量白费了。

解决：

• 加指数退避重试：

import time
import openai

def embed_with_retry(
    client: openai.OpenAI,
    texts: List[str],
    model: str = "text-embedding-3-small",
    max_retries: int = 5,
    batch_size: int = 100,
) -> List[List[float]]:
    """带重试的批量 Embedding"""
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]

        for attempt in range(max_retries):
            try:
                response = client.embeddings.create(
                    model=model,
                    input=batch,
                )
                all_embeddings.extend([item.embedding for item in response.data])
                break
            except openai.RateLimitError:
                wait_time = 2 ** attempt  # 1, 2, 4, 8, 16 秒
                print(f"限流，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            except openai.APIError as e:
                print(f"API 错误: {e}")
                if attempt == max_retries - 1:
                    raise

    return all_embeddings

# 使用
client = openai.OpenAI()
texts = ["文档1", "文档2", ...]  # 10万条
embeddings = embed_with_retry(client, texts, batch_size=100)

• 分片处理：将大任务拆分为小批次，中间结果持久化
• 预算控制：设置每日 API 调用预算上限

坑 4：中文特殊字符导致 Embedding 异常

问题：某些包含特殊 Unicode 字符的中文文本（如繁体字、数学公式符号），Embedding 结果出现 NaN 或异常值。

原因：模型的 tokenizer 对某些特殊字符的处理不一致，特别是数学符号（∑、∫）和罕见汉字。

解决：

• 做文本预处理，过滤或替换特殊字符
• 检查 Embedding 结果是否包含 NaN

import re
import numpy as np

def preprocess_text(text: str) -> str:
    """文本预处理，清理特殊字符"""
    # 移除控制字符
    text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
    # 移除数学符号（保留中文、英文、数字、标点）
    text = re.sub(r'[∑∫∏√∞≈≠≤≥±×÷∂∇]', '', text)
    # 规范化空白字符
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def safe_encode(model, texts: List[str]) -> np.ndarray:
    """安全的 Embedding 编码，包含 NaN 检查"""
    clean_texts = [preprocess_text(t) for t in texts]
    embeddings = model.encode(clean_texts)

    # 检查 NaN
    nan_mask = np.isnan(embeddings).any(axis=1)
    if nan_mask.any():
        nan_indices = np.where(nan_mask)[0]
        print(f"警告: 以下索引的 Embedding 包含 NaN: {nan_indices}")
        # 用零向量替代 NaN
        embeddings[nan_indices] = np.zeros(embeddings.shape[1])

    return embeddings

坑 5：模型加载太慢影响服务启动

问题：用 Sentence Transformers 加载 bge-large-zh 需要 5-10 秒，每次服务重启都很慢，用户能明显感知到延迟。

解决：

• 用模型缓存 + 预加载：

import torch
from sentence_transformers import SentenceTransformer
from functools import lru_cache

# 方案1：全局预加载
_model = None

def get_model(model_name: str = "BAAI/bge-large-zh") -> SentenceTransformer:
    global _model
    if _model is None:
        print(f"加载模型 {model_name}...")
        _model = SentenceTransformer(model_name)
        # 预热：用一条空文本跑一次，确保模型完全加载
        _model.encode(["warmup"])
        print("模型加载完成")
    return _model

# 方案2：服务启动时预加载（在 FastAPI/Flask 启动时调用）
@app.on_event("startup")
async def startup_event():
    get_model()  # 预加载到内存

# 方案3：用 ONNX Runtime 加速推理
from optimum.onnxruntime import ORTModelForFeatureExtraction

ort_model = ORTModelForFeatureExtraction.from_pretrained(
    "BAAI/bge-large-zh",
    export=True,
)
# 推理速度提升 2-3 倍

九、参考资料

• Xiao et al., "C-Pack: Packaged Resources To Advance General Chinese Embedding" (2023) — BGE 模型论文
• Chen et al., "BGE M3-Embedding: Multi-Lingual, Multi-Functionality, Multi-Granularity" (2024)
• MTEB Leaderboard：https://huggingface.co/spaces/mteb/leaderboard
• Sentence Transformers 文档：https://www.sbert.net/
• OpenAI Embedding 文档：https://platform.openai.com/docs/guides/embeddings
• LangChain Embedding 文档：https://python.langchain.com/docs/how_to/embed_text/
• Cohere Embed 文档：https://docs.cohere.com/docs/embeddings
• Jina Embeddings 文档：https://jina.ai/embeddings/