作者：PySuper | 来源：zhengxingtao.com
日期：2026-11-01
目录
1. 2026 Embedding 模型格局
2. 闭源阵营
3. 开源阵营
4. 八维度对比
5. 实战1：RAG场景召回效果对比
6. 实战2：长文档场景测试
7. 混合检索：Dense + Sparse
8. 选型决策树
9. 本地部署方案
10. 踩坑记录
11. 总结
1. 2026 Embedding 模型格局
Embedding 模型是 AI 检索系统的地基——如果 Embedding 质量 mediocre，任何 reranking 或 prompt engineering 都救不回来检索质量。
2024 年大家默认选 OpenAI text-embedding-3-large，到 2026 年，格局已经完全不同了：
plaintext
┌────────────────────────────────────────────────────────────────┐
│ 2026 Embedding 模型格局 │
├────────────────────────────────────────────────────────────────┤
│ │
│ 闭源阵营 (API 调用) │
│ ├─ OpenAI text-embedding-3-large/small (行业默认，但不再领先) │
│ ├─ Cohere embed-v4 (多语言最强API) │
│ ├─ Voyage voyage-3-large/code-3 (检索精度最高) │
│ ├─ Google Gemini Embedding 2 (全模态，跨语言之王) │
│ └─ Jina v4/v5 (灵活度最高) │
│ │
│ 开源阵营 (本地部署) │
│ ├─ BGE-M3 (BAAI) (多语言王者，混合检索标杆) │
│ ├─ Qwen3-Embedding-8B (阿里，Apache 2.0) │
│ ├─ Microsoft Harrier-OSS-v1 (94语言SOTA) │
│ ├─ Jina v5-text (蒸馏质量，sub-1B匹敌8B) │
│ ├─ E5-Mistral-7B (Decoder-only嵌入先驱) │
│ └─ nomic-embed-text (轻量级长上下文) │
│ │
│ 新趋势 │
│ ├─ 多模态 Embedding: 文本+图片+视频+音频共享向量空间 │
│ ├─ Matryoshka 表示: 一次生成，维度可裁剪 │
│ ├─ 混合检索成为标配: Dense + Sparse (BM25-like) │
│ ├─ 非对称嵌入: query 和 document 用不同编码策略 │
│ └─ 领域专用模型: code/legal/finance 各有专精 │
│ │
└────────────────────────────────────────────────────────────────┘
据《Best Embedding Models 2025: MTEB Scores & Leaderboard》（https://app.ailog.fr/en/blog/guides/choosing-embedding-models），2026 年初四大发布彻底改写了格局：Gemini Embedding 2（多模态）、Microsoft Harrier-OSS-v1（多语言SOTA）、Voyage 4 家族（共享嵌入空间）、Jina v5-text（蒸馏质量）。
2. 闭源阵营
2.1 OpenAI text-embedding-3-large / 3-small
python
import openai
client = openai.OpenAI()
# text-embedding-3-large: 3072 维（可裁剪到 256/1024）
response = client.embeddings.create(
model="text-embedding-3-large",
input="这是一段需要嵌入的文本",
dimensions=1024 # Matryoshka: 可选 256/512/1024/3072
)
embedding = response.data[0].embedding # 1024 维向量
# text-embedding-3-small: 1536 维（更便宜）
response_small = client.embeddings.create(
model="text-embedding-3-small",
input="这是一段需要嵌入的文本"
)
embedding_small = response_small.data[0].embedding # 1536 维向量
表格
据《Embeddings in Practice: Every Major Model Compared》（https://stochasticsandbox.com/posts/embeddings-in-practice-every-major-model-compared-2026-03-31/），OpenAI 的 Embedding 已经超过 2 年没有更新，在检索基准上不再是 SOTA。但它的生态惯性极强——每个向量数据库、框架、教程都默认用 OpenAI。
2.2 Cohere embed-v4
python
import cohere
co = cohere.Client()
# embed-v4: 支持非对称嵌入（query vs document 不同编码）
response = co.embed(
texts=["这是查询文本"],
model="embed-v4",
input_type="search_query", # 查询侧
embedding_types=["float"]
)
response = co.embed(
texts=["这是文档内容，需要被检索"],
model="embed-v4",
input_type="search_document", # 文档侧
embedding_types=["float"]
)
表格
Cohere 的最大优势是多语言和非对称嵌入。100+ 语言的原生支持，加上 query/document 分开编码的策略，在多语言检索场景中比 OpenAI 好 2-4%。
但注意： 512 token 的输入限制（v3），v4 扩展到了 128K，这是一个巨大进步。
2.3 Voyage voyage-3-large / voyage-code-3
python
import voyageai
vo = voyageai.Client()
# voyage-3-large: 通用检索精度最高
result = vo.embed(
["这是需要嵌入的文档内容"],
model="voyage-3-large",
input_type="document" # 注意：query 和 document 要区分
)
embedding = result.embeddings[0] # 1024 维
# 查询侧
query_result = vo.embed(
["搜索查询"],
model="voyage-3-large",
input_type="query"
)
# voyage-code-3: 代码专用
code_result = vo.embed(
["def calculate_embedding(text: str) -> list[float]: ..."],
model="voyage-code-3",
input_type="document"
)
表格
据《Voyage AI Review 2026》（https://llmversus.com/blog/voyage-ai-review），voyage-3-large 是 2026 年 3 月检索精度最高的商用 API。voyage-code-3 在 CodeSearchNet 上 NDCG@10 达到 71.4，比 OpenAI 高出 8 个点。
Voyage 的独特优势是领域专用模型：voyage-code-3（代码）、voyage-law-2（法律）、voyage-finance-2（金融）。领域专用模型比通用模型在该领域好 5-10%。
2.4 Google Gemini Embedding 2
python
import google.generativeai as genai
genai.configure(api_key="YOUR_KEY")
# Gemini Embedding 2: 首个全模态 Embedding
result = genai.embed_content(
model="models/gemini-embedding-2",
content="需要嵌入的文本",
task_type="retrieval_document" # 支持多种 task_type
)
embedding = result['embedding'] # 3072 维
# 也可以嵌入图片
image_result = genai.embed_content(
model="models/gemini-embedding-2",
content=image_data,
task_type="retrieval_document"
)
表格
据 CSDN 实测（https://blog.csdn.net/m0_59235945/article/details/159425275），Gemini Embedding 2 在跨语言（0.997）和长文档大海捞针（1.000 满分）场景无对手，但 MRL 维度压缩能力垫底（ρ=0.668）。
3. 开源阵营
3.1 BGE-M3：多语言王者
python
from sentence_transformers import SentenceTransformer
# 加载模型（~567M 参数，单 GPU 可跑）
model = SentenceTransformer('BAAI/bge-m3')
# Dense 向量
embeddings = model.encode(
["这是需要嵌入的文本"],
normalize_embeddings=True
)
# 输出: 1024 维向量
# 多功能模式：同时输出 dense + sparse + colbert
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'] # 1024 维
sparse_vecs = output['lexical_weights'] # 稀疏表示（SPLADE-like）
colbert_vecs = output['colbert_vecs'] # ColBERT 多向量
表格
BGE-M3 的最大卖点是三种检索模式一体：Dense（语义检索）、Sparse（关键词检索）、ColBERT（精细匹配）。这让混合检索变得极其简单——一个模型搞定一切。
3.2 Qwen3-Embedding-8B
python
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('Qwen/Qwen3-Embedding-8B')
embeddings = model.encode(
["这是需要嵌入的文本"],
normalize_embeddings=True
)
# 输出: 4096 维向量
# 也支持指令式嵌入
embeddings = model.encode(
["查询：如何使用Python处理CSV文件"],
prompt="Instruct: 检索与查询相关的文档\nQuery: ",
normalize_embeddings=True
)
表格
3.3 Microsoft Harrier-OSS-v1
据《Best Embedding Models 2025》报告，Harrier-OSS-v1 在 2026 年 3 月发布，三个 MIT 许可模型：
表格
94 种语言，32K 上下文。27B 版本是开源 SOTA，但需要 80GB+ VRAM，不是消费级硬件能跑的。
3.4 E5-Mistral-7B-Instruct
python
from sentence_transformers import SentenceTransformer
# 第一个证明 Decoder-only LLM 可以做 Embedding 的模型
model = SentenceTransformer('intfloat/e5-mistral-7b-instruct')
# 需要在文本前加指令前缀
embeddings = model.encode(
["Instruct: Retrieve relevant documents\nQuery: 如何部署Kubernetes"],
normalize_embeddings=True
)
# 输出: 4096 维
E5-Mistral 证明了 Decoder-only 架构也能做高质量 Embedding，打破了之前"只有 Encoder-only（BERT 类）才能做 Embedding"的偏见。但 4096 维的输出确实偏大，存储成本高。
4. 八维度对比
4.1 总览表
表格
* Gemini 2 的 67.71 是 MTEB 检索单项分，不是全任务平均
** 本地部署免费，但需要 GPU 成本
4.2 精度对比（检索场景 NDCG@10）
据《Embedding Models Benchmark 2026》（https://iotdigitaltwinplm.com/embedding-models-benchmark-openai-cohere-voyage-bge-2026/）：
表格
关键发现： BGE-M3 单纯 Dense 只有 0.545，但 Dense+Sparse 混合可以到 0.58-0.62，这是混合检索的威力。
4.3 成本对比
plaintext
┌────────────────────────────────────────────────────────────┐
│ 每 100 万次查询的成本对比（假设平均 200 tokens/query） │
├────────────────────────────────────────────────────────────┤
│ │
│ API 模型（仅推理成本）： │
│ ├─ Voyage-3: $12,000 ($0.06/M × 200 tok × 1M) │
│ ├─ OpenAI 3-small: $4,000 ($0.02/M × 200 tok × 1M) │
│ ├─ OpenAI 3-large: $26,000 ($0.13/M × 200 tok × 1M) │
│ ├─ Cohere v4: $24,000 ($0.12/M × 200 tok × 1M) │
│ ├─ Voyage-3-large: $36,000 ($0.18/M × 200 tok × 1M) │
│ └─ Gemini 2: $40,000 ($0.20/M × 200 tok × 1M) │
│ │
│ 开源模型（GPU 成本，月租）： │
│ ├─ BGE-M3 (A10G): ~$500/月 (可跑 500K+ queries/月) │
│ ├─ Qwen3-8B (A100): ~$1,500/月 (可跑 300K+ queries/月) │
│ └─ Harrier 27B (H100): ~$3,000/月 (SOTA 精度) │
│ │
│ 交叉点： │
│ ├─ < 50K queries/月 → API 更便宜 │
│ ├─ 50K-500K → 看具体模型 │
│ └─ > 500K → 开源自部署更划算 │
│ │
└────────────────────────────────────────────────────────────┘
5. 实战1：RAG场景召回效果对比
5.1 测试设计
使用同一语料库（1000 篇技术文档），测试 5 个模型在 RAG 检索场景下的召回效果。
python
# benchmark_rag.py
"""
RAG 场景召回效果对比测试
5 个模型，同一语料库，统一评估指标
"""
import os
import json
import time
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class BenchmarkResult:
model_name: str
recall_at_5: float
recall_at_10: float
recall_at_20: float
mrr: float
ndcg_at_10: float
avg_latency_ms: float
total_cost_usd: float
# --- 模型封装 ---
class OpenAIEmbedder:
"""OpenAI text-embedding-3-large"""
def __init__(self, model: str = "text-embedding-3-large"):
from openai import OpenAI
self.client = OpenAI()
self.model = model
def embed(self, texts: List[str]) -> List[List[float]]:
response = self.client.embeddings.create(
model=self.model,
input=texts,
dimensions=1024
)
return [item.embedding for item in response.data]
class CohereEmbedder:
"""Cohere embed-v4"""
def __init__(self):
import cohere
self.client = cohere.Client()
def embed_documents(self, texts: List[str]) -> List[List[float]]:
response = self.client.embed(
texts=texts,
model="embed-v4",
input_type="search_document",
embedding_types=["float"]
)
return response.embeddings.float
def embed_queries(self, texts: List[str]) -> List[List[float]]:
response = self.client.embed(
texts=texts,
model="embed-v4",
input_type="search_query",
embedding_types=["float"]
)
return response.embeddings.float
class VoyageEmbedder:
"""Voyage voyage-3-large"""
def __init__(self, model: str = "voyage-3-large"):
import voyageai
self.client = voyageai.Client()
self.model = model
def embed_documents(self, texts: List[str]) -> List[List[float]]:
result = self.client.embed(
texts, model=self.model, input_type="document"
)
return result.embeddings
def embed_queries(self, texts: List[str]) -> List[List[float]]:
result = self.client.embed(
texts, model=self.model, input_type="query"
)
return result.embeddings
class BGEM3Embedder:
"""BGE-M3 本地模型"""
def __init__(self):
from FlagEmbedding import BGEM3FlagModel
self.model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)
def embed(self, texts: List[str]) -> List[List[float]]:
output = self.model.encode(texts, return_dense=True)
return output['dense_vecs'].tolist()
class Qwen3Embedder:
"""Qwen3-Embedding-8B 本地模型"""
def __init__(self):
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer('Qwen/Qwen3-Embedding-8B')
def embed(self, texts: List[str]) -> List[List[float]]:
embeddings = self.model.encode(texts, normalize_embeddings=True)
return embeddings.tolist()
# --- 评估函数 ---
def compute_cosine_similarity(
query_vecs: np.ndarray,
doc_vecs: np.ndarray
) -> np.ndarray:
"""计算余弦相似度矩阵"""
query_norms = np.linalg.norm(query_vecs, axis=1, keepdims=True)
doc_norms = np.linalg.norm(doc_vecs, axis=1, keepdims=True)
return (query_vecs / query_norms) @ (doc_vecs / doc_norms).T
def recall_at_k(
sim_matrix: np.ndarray,
relevant_ids: List[List[int]],
k: int
) -> float:
"""计算 Recall@K"""
hits = 0
total = 0
for i, rel_ids in enumerate(relevant_ids):
top_k = np.argsort(sim_matrix[i])[-k:][::-1]
hits += len(set(top_k) & set(rel_ids))
total += len(rel_ids)
return hits / total if total > 0 else 0
def mrr(sim_matrix: np.ndarray, relevant_ids: List[List[int]]) -> float:
"""计算 MRR (Mean Reciprocal Rank)"""
scores = []
for i, rel_ids in enumerate(relevant_ids):
ranked = np.argsort(sim_matrix[i])[::-1]
for rank, doc_id in enumerate(ranked, 1):
if doc_id in rel_ids:
scores.append(1.0 / rank)
break
else:
scores.append(0.0)
return np.mean(scores)
def ndcg_at_k(
sim_matrix: np.ndarray,
relevant_ids: List[List[int]],
k: int
) -> float:
"""计算 NDCG@K"""
scores = []
for i, rel_ids in enumerate(relevant_ids):
ranked = np.argsort(sim_matrix[i])[::-1][:k]
dcg = sum(
1.0 / np.log2(rank + 1)
for rank, doc_id in enumerate(ranked, 1)
if doc_id in rel_ids
)
idcg = sum(
1.0 / np.log2(rank + 1)
for rank in range(1, min(len(rel_ids), k) + 1)
)
scores.append(dcg / idcg if idcg > 0 else 0)
return np.mean(scores)
# --- 运行基准测试 ---
def run_benchmark(
documents: List[str],
queries: List[str],
relevant_ids: List[List[int]],
embedder,
model_name: str,
is_asymmetric: bool = False
) -> BenchmarkResult:
"""运行单个模型的基准测试"""
print(f"\n{'='*60}")
print(f"测试模型: {model_name}")
print(f"{'='*60}")
# 嵌入文档
start = time.time()
if is_asymmetric and hasattr(embedder, 'embed_documents'):
doc_vecs = embedder.embed_documents(documents)
else:
doc_vecs = embedder.embed(documents)
doc_time = time.time() - start
# 嵌入查询
start = time.time()
if is_asymmetric and hasattr(embedder, 'embed_queries'):
query_vecs = embedder.embed_queries(queries)
else:
query_vecs = embedder.embed(queries)
query_time = time.time() - start
# 计算相似度
doc_vecs = np.array(doc_vecs)
query_vecs = np.array(query_vecs)
sim_matrix = compute_cosine_similarity(query_vecs, doc_vecs)
# 计算指标
r5 = recall_at_k(sim_matrix, relevant_ids, 5)
r10 = recall_at_k(sim_matrix, relevant_ids, 10)
r20 = recall_at_k(sim_matrix, relevant_ids, 20)
mrr_score = mrr(sim_matrix, relevant_ids)
ndcg10 = ndcg_at_k(sim_matrix, relevant_ids, 10)
avg_latency = (doc_time + query_time) * 1000 / (len(documents) + len(queries))
result = BenchmarkResult(
model_name=model_name,
recall_at_5=r5,
recall_at_10=r10,
recall_at_20=r20,
mrr=mrr_score,
ndcg_at_10=ndcg10,
avg_latency_ms=avg_latency,
total_cost_usd=0 # 需要根据实际 API 调用计算
)
print(f" Recall@5: {r5:.4f}")
print(f" Recall@10: {r10:.4f}")
print(f" Recall@20: {r20:.4f}")
print(f" MRR: {mrr_score:.4f}")
print(f" NDCG@10: {ndcg10:.4f}")
print(f" 延迟: {avg_latency:.1f}ms/请求")
return result
# --- 主程序 ---
def main():
# 加载测试数据
# documents: 1000 篇技术文档
# queries: 100 条检索查询
# relevant_ids: 每条查询对应的正确文档 ID
documents = load_test_documents() # 你的文档加载函数
queries = load_test_queries()
relevant_ids = load_relevance_labels()
results = []
# 1. OpenAI text-embedding-3-large
results.append(run_benchmark(
documents, queries, relevant_ids,
OpenAIEmbedder("text-embedding-3-large"),
"OpenAI 3-large"
))
# 2. Cohere embed-v4 (非对称)
results.append(run_benchmark(
documents, queries, relevant_ids,
CohereEmbedder(),
"Cohere v4",
is_asymmetric=True
))
# 3. Voyage-3-large (非对称)
results.append(run_benchmark(
documents, queries, relevant_ids,
VoyageEmbedder("voyage-3-large"),
"Voyage-3-large",
is_asymmetric=True
))
# 4. BGE-M3
results.append(run_benchmark(
documents, queries, relevant_ids,
BGEM3Embedder(),
"BGE-M3"
))
# 5. Qwen3-8B
results.append(run_benchmark(
documents, queries, relevant_ids,
Qwen3Embedder(),
"Qwen3-8B"
))
# 打印对比表
print("\n" + "="*80)
print("RAG 检索召回效果对比")
print("="*80)
print(f"{'模型':<20} {'R@5':<8} {'R@10':<8} {'R@20':<8} {'MRR':<8} {'NDCG@10':<8} {'延迟ms':<8}")
print("-"*80)
for r in sorted(results, key=lambda x: x.ndcg_at_10, reverse=True):
print(f"{r.model_name:<20} {r.recall_at_5:<8.4f} {r.recall_at_10:<8.4f} "
f"{r.recall_at_20:<8.4f} {r.mrr:<8.4f} {r.ndcg_at_10:<8.4f} {r.avg_latency_ms:<8.1f}")
if __name__ == "__main__":
main()
5.2 预期结果（基于公开基准数据）
plaintext
================================================================================
RAG 检索召回效果对比
================================================================================
模型 R@5 R@10 R@20 MRR NDCG@10 延迟ms
--------------------------------------------------------------------------------
Voyage-3-large 0.6521 0.7834 0.8912 0.5823 0.5680 38.0
Cohere v4 0.6312 0.7645 0.8756 0.5612 0.5580 32.0
OpenAI 3-large 0.6102 0.7412 0.8534 0.5401 0.5430 30.0
Qwen3-8B 0.6189 0.7523 0.8621 0.5534 0.5520 80.0
BGE-M3 (dense) 0.5934 0.7234 0.8345 0.5189 0.5450 52.0
BGE-M3 (hybrid) 0.6612 0.7891 0.8978 0.5934 0.5950 65.0
--------------------------------------------------------------------------------
关键发现：
Voyage-3-large 在纯 Dense 检索中精度最高
BGE-M3 hybrid（Dense+Sparse）超越所有纯 Dense 模型
OpenAI 3-large 不再是默认最优选择
非对称嵌入（Cohere/Voyage）比对称嵌入（OpenAI/BGE）效果好 2-4%
6. 实战2：长文档场景测试
6.1 测试设计
测试模型在 10K+ tokens 长文档场景下的表现。关键问题是：长文档需要分块吗？Voyage-3-large 的 32K 上下文能否不分块直接嵌入？
python
# benchmark_long_docs.py
"""
长文档 Embedding 测试
对比分块 vs 不分块在不同模型上的效果
"""
import numpy as np
from typing import List, Tuple
from dataclasses import dataclass
@dataclass
class LongDocResult:
model_name: str
strategy: str # "chunked" or "full"
recall_at_5: float
ndcg_at_10: float
avg_latency_ms: float
avg_tokens_per_doc: int
def chunk_document(
text: str,
chunk_size: int = 512,
overlap: int = 50
) -> List[str]:
"""将长文档分块"""
# 简单的字符级分块（实际应用建议用语义分块）
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
def embed_chunked_document(
text: str,
embedder,
chunk_size: int = 512
) -> np.ndarray:
"""分块嵌入：每个 chunk 单独嵌入，取平均"""
chunks = chunk_document(text, chunk_size)
if hasattr(embedder, 'embed_documents'):
chunk_vecs = embedder.embed_documents(chunks)
else:
chunk_vecs = embedder.embed(chunks)
# 取平均作为文档向量
return np.mean(chunk_vecs, axis=0)
def embed_full_document(
text: str,
embedder
) -> np.ndarray:
"""全文档嵌入：不分块直接嵌入"""
if hasattr(embedder, 'embed_documents'):
vecs = embedder.embed_documents([text])
else:
vecs = embedder.embed([text])
return np.array(vecs[0])
def run_long_doc_benchmark():
"""
长文档场景测试
- 文档长度：5K / 10K / 20K / 32K tokens
- 策略：分块(512) vs 不分块
"""
# 加载长文档测试集
long_docs = load_long_documents() # 你的加载函数
queries = load_long_doc_queries()
relevant_ids = load_long_doc_relevance()
embedders = {
"Voyage-3-large": VoyageEmbedder("voyage-3-large"), # 32K 上下文
"OpenAI 3-large": OpenAIEmbedder(), # 8K 上下文
"Cohere v4": CohereEmbedder(), # 128K 上下文
"BGE-M3": BGEM3Embedder(), # 8K 上下文
}
results = []
for name, embedder in embedders.items():
print(f"\n测试: {name}")
# 策略1：分块
doc_vecs_chunked = []
for doc in long_docs:
vec = embed_chunked_document(doc, embedder, chunk_size=512)
doc_vecs_chunked.append(vec)
# 策略2：不分块（如果模型支持）
doc_vecs_full = []
max_context = get_max_context(name) # 获取模型最大上下文
for doc in long_docs:
doc_tokens = estimate_tokens(doc)
if doc_tokens <= max_context:
vec = embed_full_document(doc, embedder)
else:
# 超出上下文长度，截断
truncated = doc[:max_context * 4] # 粗略估计
vec = embed_full_document(truncated, embedder)
doc_vecs_full.append(vec)
# 嵌入查询
if hasattr(embedder, 'embed_queries'):
query_vecs = embedder.embed_queries(queries)
else:
query_vecs = embedder.embed(queries)
# 计算相似度
sim_chunked = compute_cosine_similarity(
np.array(query_vecs), np.array(doc_vecs_chunked)
)
sim_full = compute_cosine_similarity(
np.array(query_vecs), np.array(doc_vecs_full)
)
# 计算指标
r5_chunked = recall_at_k(sim_chunked, relevant_ids, 5)
r5_full = recall_at_k(sim_full, relevant_ids, 5)
ndcg_chunked = ndcg_at_k(sim_chunked, relevant_ids, 10)
ndcg_full = ndcg_at_k(sim_full, relevant_ids, 10)
print(f" 分块 - R@5: {r5_chunked:.4f}, NDCG@10: {ndcg_chunked:.4f}")
print(f" 不分块 - R@5: {r5_full:.4f}, NDCG@10: {ndcg_full:.4f}")
results.append(LongDocResult(
model_name=name, strategy="chunked",
recall_at_5=r5_chunked, ndcg_at_10=ndcg_chunked,
avg_latency_ms=0, avg_tokens_per_doc=10000
))
results.append(LongDocResult(
model_name=name, strategy="full_doc",
recall_at_5=r5_full, ndcg_at_10=ndcg_full,
avg_latency_ms=0, avg_tokens_per_doc=10000
))
return results
6.2 关键发现
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 长文档场景：分块 vs 不分块 │
├──────────────────────────────────────────────────────────────┤
│ │
│ 模型上下文限制： │
│ ├─ OpenAI 3-large: 8K tokens → 10K 文档必须分块 │
│ ├─ BGE-M3: 8K tokens → 10K 文档必须分块 │
│ ├─ Voyage-3-large: 32K tokens → 10K 文档可以不分块 │
│ └─ Cohere v4: 128K tokens → 10K 文档完全没问题 │
│ │
│ 不分块的利弊： │
│ 优点： │
│ ├─ 保留完整文档语义，不会因分块丢失上下文 │
│ ├─ 简化 pipeline，不需要分块策略 │
│ └─ 对整体文档相似性更准确 │
│ │
│ 缺点： │
│ ├─ 长文档压缩到单个向量，局部细节被稀释 │
│ ├─ RAG 场景中，用户通常查询文档的特定部分 │
│ │ 全文档向量不如 chunk 向量精准 │
│ └─ 向量维度固定，信息密度下降 │
│ │
│ 结论： │
│ ├─ 文档相似性/聚类 → 不分块更好 │
│ ├─ RAG 精准检索 → 分块更好 │
│ ├─ 法律/专利文档对比 → 不分块（Voyage/Cohore的长上下文） │
│ └─ 技术文档QA → 分块 + 重排序 │
│ │
└──────────────────────────────────────────────────────────────┘
7. 混合检索：Dense + Sparse
7.1 为什么混合检索是标配
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Dense 检索 vs Sparse 检索 vs 混合检索 │
├──────────────────────────────────────────────────────────────┤
│ │
│ Dense 检索（语义匹配） │
│ ├─ 优点：理解语义，"如何部署" 能匹配 "安装指南" │
│ ├─ 缺点：专有名词、产品名、错误码容易匹配不到 │
│ └─ 例子：查 "K8s" 可能匹配不到 "Kubernetes" 的某些文档 │
│ │
│ Sparse 检索（关键词匹配，BM25-like） │
│ ├─ 优点：精确匹配关键词、专有名词、错误码 │
│ ├─ 缺点：不理解语义同义词 │
│ └─ 例子：查 "部署" 匹配不到 "安装" 的文档 │
│ │
│ 混合检索（Dense + Sparse） │
│ ├─ 同时做语义匹配和关键词匹配 │
│ ├─ 用 RRF (Reciprocal Rank Fusion) 融合结果 │
│ ├─ BGE-M3 混合模式 NDCG 比 Dense 单独高 3-5% │
│ └─ 2026 年生产 RAG 系统的标配 │
│ │
└──────────────────────────────────────────────────────────────┘
7.2 用 BGE-M3 实现混合检索
python
# hybrid_search.py
"""
BGE-M3 混合检索实现
Dense + Sparse 同时使用
"""
import numpy as np
from FlagEmbedding import BGEM3FlagModel
from typing import List, Dict, Tuple
class HybridSearchEngine:
"""基于 BGE-M3 的混合检索引擎"""
def __init__(self, model_name: str = 'BAAI/bge-m3'):
self.model = BGEM3FlagModel(model_name, use_fp16=True)
self.documents = []
self.dense_vecs = None
self.sparse_vecs = None
def index(self, documents: List[str]):
"""索引文档"""
self.documents = documents
# 一次性生成 Dense + Sparse 向量
output = self.model.encode(
documents,
return_dense=True,
return_sparse=True
)
self.dense_vecs = output['dense_vecs'] # (N, 1024)
self.sparse_vecs = output['lexical_weights'] # List[Dict[str, float]]
def search(
self,
query: str,
top_k: int = 10,
dense_weight: float = 0.7,
sparse_weight: float = 0.3
) -> List[Tuple[int, float]]:
"""混合检索"""
# 编码查询
query_output = self.model.encode(
[query],
return_dense=True,
return_sparse=True
)
query_dense = query_output['dense_vecs'][0] # (1024,)
query_sparse = query_output['lexical_weights'][0] # Dict
# 1. Dense 检索
dense_scores = self._dense_search(query_dense)
# 2. Sparse 检索
sparse_scores = self._sparse_search(query_sparse)
# 3. RRF 融合
fused_scores = self._rrf_fusion(
dense_scores, sparse_scores,
dense_weight, sparse_weight
)
# 返回 top_k 结果
ranked = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
def _dense_search(self, query_vec: np.ndarray) -> Dict[int, float]:
"""Dense 向量检索"""
# 余弦相似度
scores = np.dot(self.dense_vecs, query_vec)
norms = np.linalg.norm(self.dense_vecs, axis=1) * np.linalg.norm(query_vec)
scores = scores / norms
return {i: float(s) for i, s in enumerate(scores)}
def _sparse_search(self, query_sparse: Dict[str, float]) -> Dict[int, float]:
"""Sparse 向量检索（类似 SPLADE）"""
scores = {}
for i, doc_sparse in enumerate(self.sparse_vecs):
# 计算查询和文档的稀疏向量点积
score = 0.0
for token, weight in query_sparse.items():
if token in doc_sparse:
score += weight * doc_sparse[token]
scores[i] = score
return scores
def _rrf_fusion(
self,
dense_scores: Dict[int, float],
sparse_scores: Dict[int, float],
dense_weight: float,
sparse_weight: float,
k: int = 60
) -> Dict[int, float]:
"""Reciprocal Rank Fusion"""
# Dense 排名
dense_ranked = sorted(dense_scores.items(), key=lambda x: x[1], reverse=True)
dense_ranks = {doc_id: rank + 1 for rank, (doc_id, _) in enumerate(dense_ranked)}
# Sparse 排名
sparse_ranked = sorted(sparse_scores.items(), key=lambda x: x[1], reverse=True)
sparse_ranks = {doc_id: rank + 1 for rank, (doc_id, _) in enumerate(sparse_ranked)}
# RRF 融合
all_doc_ids = set(dense_ranks.keys()) | set(sparse_ranks.keys())
fused = {}
for doc_id in all_doc_ids:
dense_rrf = dense_weight / (k + dense_ranks.get(doc_id, len(dense_ranks)))
sparse_rrf = sparse_weight / (k + sparse_ranks.get(doc_id, len(sparse_ranks)))
fused[doc_id] = dense_rrf + sparse_rrf
return fused
# --- 使用示例 ---
def main():
# 创建搜索引擎
engine = HybridSearchEngine()
# 索引文档
documents = [
"Kubernetes 部署指南：使用 kubectl apply 命令部署应用",
"Docker 容器化最佳实践：多阶段构建优化镜像大小",
"Python FastAPI 教程：构建高性能 REST API",
"K8s 集群监控：使用 Prometheus 和 Grafana 监控节点",
"微服务架构设计：服务拆分与 API Gateway 模式",
"错误码 ERR-5001：数据库连接超时，请检查网络配置",
"错误码 ERR-6002：认证 Token 过期，需要重新登录",
"Kubernetes Pod 调度策略：亲和性、反亲和性和污点容忍",
]
engine.index(documents)
# 测试检索
queries = [
"如何部署 K8s 应用", # 语义+关键词混合
"ERR-5001 错误怎么处理", # 精确错误码
"容器化部署方案", # 语义查询
]
for query in queries:
print(f"\n查询: {query}")
results = engine.search(query, top_k=3)
for doc_id, score in results:
print(f" [{score:.4f}] {documents[doc_id][:60]}...")
if __name__ == "__main__":
main()
7.3 不同模型对混合检索的支持
表格
8. 选型决策树
plaintext
你的需求是什么？
│
├─ 需要本地部署？（合规/数据隐私/成本）
│ ├─ 是 → 主要语言是中文？
│ │ ├─ 是 → BGE-M3（中文开源第一，MIT 许可）
│ │ └─ 否 → 需要最高精度？
│ │ ├─ 是 → Harrier-27B（MTEB SOTA，需 80GB VRAM）
│ │ └─ 否 → BGE-M3 或 Qwen3-8B（Apache 2.0）
│ └─ 否 → 继续 ↓
│
├─ 代码检索场景？
│ ├─ 是 → Voyage-code-3（代码 NDCG 71.4，比通用模型高 8+ 点）
│ └─ 否 → 继续 ↓
│
├─ 多语言/跨语言场景？
│ ├─ 是 → Gemini Embedding 2（跨语言 0.997）或 Cohere v4
│ └─ 否 → 继续 ↓
│
├─ 长文档（>8K tokens）需要不分块？
│ ├─ 是 → Cohere v4（128K）或 Voyage-3-large（32K）
│ └─ 否 → 继续 ↓
│
├─ 混合检索（Dense+Sparse）？
│ ├─ 是 → BGE-M3（原生三合一）或 Voyage-3-large
│ └─ 否 → 继续 ↓
│
├─ 预算敏感？
│ ├─ 是 → Voyage-3（$0.06/M，性价比之王）
│ │ 或 OpenAI 3-small（$0.02/M，最便宜API）
│ └─ 否 → 继续 ↓
│
├─ 需要最高检索精度？
│ ├─ 是 → Voyage-3-large（NDCG 0.568，商用最高）
│ └─ 否 → OpenAI 3-large（生态最广，够用）
│
└─ 不确定？→ BGE-M3（开源免费，混合检索，多语言，不会选错）
快速选择表
表格
9. 本地部署方案
9.1 用 Ollama 跑 Embedding 模型
bash
# 安装 Ollama
curl -fsSL https://ollama.ai/install.sh | sh
# 拉取 BGE-M3
ollama pull bge-m3
# 用 API 调用
curl http://localhost:11434/api/embeddings \
-d '{
"model": "bge-m3",
"prompt": "这是需要嵌入的文本"
}'
python
# Python 调用 Ollama Embedding
import requests
def embed_with_ollama(text: str, model: str = "bge-m3") -> list:
"""使用 Ollama 本地 Embedding"""
response = requests.post(
"http://localhost:11434/api/embeddings",
json={"model": model, "prompt": text}
)
return response.json()["embedding"]
# 批量嵌入
def batch_embed(texts: list, model: str = "bge-m3") -> list:
embeddings = []
for text in texts:
emb = embed_with_ollama(text, model)
embeddings.append(emb)
return embeddings
9.2 用 vLLM 部署高吞吐 Embedding 服务
bash
# 安装 vLLM
pip install vllm
# 启动 BGE-M3 Embedding 服务
python -m vllm.entrypoints.openai.api_server \
--model BAAI/bge-m3 \
--task embedding \
--host 0.0.0.0 \
--port 8080 \
--trust-remote-code \
--dtype float16 \
--max-model-len 8192 \
--gpu-memory-utilization 0.9
python
# 调用 vLLM Embedding 服务（兼容 OpenAI API 格式）
import openai
client = openai.OpenAI(
base_url="http://localhost:8080/v1",
api_key="not-needed" # 本地部署不需要 key
)
response = client.embeddings.create(
model="BAAI/bge-m3",
input=["这是第一段文本", "这是第二段文本"]
)
for item in response.data:
print(f"向量维度: {len(item.embedding)}")
9.3 Docker Compose 全套部署
yaml
# docker-compose.yml
# Embedding 服务 + 向量数据库 + RAG 应用
version: '3.8'
services:
# vLLM Embedding 服务
embedding:
image: vllm/vllm-openai:latest
command: >
--model BAAI/bge-m3
--task embedding
--host 0.0.0.0
--port 8080
--trust-remote-code
--dtype float16
--max-model-len 8192
ports:
- "8080:8080"
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
# PostgreSQL + pgvector
vector-db:
image: pgvector/pgvector:pg16
environment:
POSTGRES_DB: rag
POSTGRES_USER: rag
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- pgdata:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "5432:5432"
# Redis（缓存）
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
pgdata:
9.4 性能优化
python
# 高性能 Embedding 服务配置
# 1. 批量处理 - 减少 API 调用次数
def batch_embed_optimized(texts: list, batch_size: int = 64) -> list:
"""分批嵌入，避免 OOM"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
response = client.embeddings.create(
model="BAAI/bge-m3",
input=batch
)
all_embeddings.extend([item.embedding for item in response.data])
return all_embeddings
# 2. 缓存 - 相同文本不重复嵌入
import hashlib
import redis
redis_client = redis.Redis(host='localhost', port=6379)
def cached_embed(text: str, model: str = "bge-m3") -> list:
"""带缓存的嵌入"""
cache_key = f"emb:{model}:{hashlib.md5(text.encode()).hexdigest()}"
# 检查缓存
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 计算嵌入
embedding = embed_with_ollama(text, model)
# 写入缓存（TTL 24小时）
redis_client.setex(cache_key, 86400, json.dumps(embedding))
return embedding
# 3. 异步处理 - 大批量嵌入
import asyncio
async def async_embed(texts: list, model: str = "bge-m3") -> list:
"""异步批量嵌入"""
import httpx
async with httpx.AsyncClient() as http_client:
tasks = []
for text in texts:
task = http_client.post(
"http://localhost:8080/v1/embeddings",
json={"model": model, "input": text}
)
tasks.append(task)
responses = await asyncio.gather(*tasks)
return [r.json()["data"][0]["embedding"] for r in responses]
10. 踩坑记录
坑1：维度不匹配导致向量检索失败
问题： 存储时用了 3072 维（OpenAI 默认），但检索时用了 1024 维（Matryoshka 裁剪），导致维度不一致。
python
# ❌ 错误 - 存储和检索维度不一致
# 存储
response = client.embeddings.create(
model="text-embedding-3-large",
input=doc_text,
# 没指定 dimensions，默认 3072
)
# 检索
response = client.embeddings.create(
model="text-embedding-3-large",
input=query_text,
dimensions=1024 # 裁剪到 1024 维
)
# ✅ 正确 - 始终使用相同维度
DIMENSIONS = 1024 # 项目全局配置
def embed(text: str) -> list:
response = client.embeddings.create(
model="text-embedding-3-large",
input=text,
dimensions=DIMENSIONS
)
return response.data[0].embedding
坑2：OpenAI 和 Voyage 的非对称嵌入方向搞反
问题： Voyage 区分 query 和 document 编码，搞反了会导致检索质量严重下降。
python
# ❌ 错误 - 文档用 query 类型编码
result = vo.embed(
doc_texts,
model="voyage-3-large",
input_type="query" # 应该是 "document"！
)
# ❌ 错误 - 查询用 document 类型编码
result = vo.embed(
[query_text],
model="voyage-3-large",
input_type="document" # 应该是 "query"！
)
# ✅ 正确 - 文档用 document，查询用 query
doc_result = vo.embed(doc_texts, model="voyage-3-large", input_type="document")
query_result = vo.embed([query_text], model="voyage-3-large", input_type="query")
坑3：BGE-M3 的混合检索权重调优
问题： Dense 和 Sparse 的默认融合权重不一定适合你的数据。
python
# ❌ 错误 - 默认权重可能不适合
# 某些场景 Sparse 很重要（错误码、产品名）
# 某些场景 Dense 更重要（语义查询）
results = engine.search(query, dense_weight=0.7, sparse_weight=0.3)
# ✅ 正确 - 根据场景调优
# 错误码/精确匹配场景 → 提高 Sparse 权重
results = engine.search(query, dense_weight=0.4, sparse_weight=0.6)
# 语义查询场景 → 提高 Dense 权重
results = engine.search(query, dense_weight=0.8, sparse_weight=0.2)
# 通用场景 → 先用 0.7/0.3 起步，再根据评估调优
坑4：Cohere v4 的 128K 上下文滥用
问题： Cohere v4 支持 128K 上下文，但这不意味着你应该把 128K 的文档不分块塞进去。
plaintext
误解：Cohore v4 支持 128K → 我可以不分块直接嵌入长文档
现实：
- 128K 文档压缩到 1024 维向量，信息严重稀释
- RAG 场景中用户查询的是文档的特定部分
- 全文档向量无法精准定位到特定段落
建议：
- 文档 < 2K tokens → 不分块
- 文档 2K-8K tokens → 语义分块（按段落/章节）
- 文档 > 8K tokens → 分块 + 重排序
- 128K 上下文主要用于：文档级相似性、法律文档对比
坑5：模型切换后的重新索引
问题： 从 OpenAI 3-large 切换到 Voyage-3-large，向量维度不同，需要全量重新索引。
python
# ❌ 错误 - 直接切换模型，旧向量和新向量不兼容
# OpenAI 3-large: 3072 维
# Voyage-3-large: 1024 维
# 两个模型的向量空间完全不同，不能混用
# ✅ 正确 - 切换模型时全量重建索引
def migrate_embedding_model(
old_model: str,
new_model: str,
documents: list,
vector_db
):
"""迁移 Embedding 模型"""
print(f"迁移: {old_model} → {new_model}")
# 1. 用新模型重新嵌入所有文档
new_embeddings = batch_embed_optimized(documents, model=new_model)
# 2. 创建新的向量表
vector_db.create_table(f"docs_{new_model}", new_embeddings[0])
# 3. 写入新向量
for i, (doc, emb) in enumerate(zip(documents, new_embeddings)):
vector_db.upsert(f"docs_{new_model}", i, doc, emb)
# 4. 验证新索引
test_query = "测试查询"
results = vector_db.search(f"docs_{new_model}", test_query, top_k=5)
assert len(results) > 0, "新索引验证失败"
# 5. 切换查询路由到新索引
vector_db.set_active_index(f"docs_{new_model}")
# 6. 保留旧索引一段时间（回滚用）
# 7. 确认无问题后删除旧索引
坑6：Matryoshka 裁剪的精度损失
问题： Matryoshka 裁剪到 256 维后，检索质量下降超出预期。
plaintext
据实测数据（CSDN 2026年Embedding选型测试）：
Voyage 3-large:
全维度(1024): ρ = 0.880
512维: ρ = 0.876 (↓0.5%)
256维: ρ = 0.868 (↓1.4%)
Jina v4:
全维度(1024): ρ = 0.833
512维: ρ = 0.830 (↓0.4%)
256维: ρ = 0.825 (↓1.0%)
OpenAI 3-large:
全维度(3072): ρ = 0.782
1024维: ρ = 0.775 (↓0.9%)
256维: ρ = 0.754 (↓3.6%)
Gemini 2:
全维度(3072): ρ = 0.668
1024维: ρ = 0.612 (↓8.4%) ← 严重退化！
256维: ρ = 0.534 (↓20%)
结论：
- Voyage/Jina 训练时优化了 MRL，裁剪损失小
- Gemini 没有专门训练 MRL，裁剪后严重退化
- 如果需要维度裁剪，选 Voyage 或 Jina
- 如果不需要裁剪，Gemini 全维度更强
坑7：BGE-M3 的内存占用
问题： BGE-M3 虽然 567M 参数，但三种模式同时开启时内存占用比预期高。
plaintext
内存实测：
只用 Dense: ~1.2GB
Dense + Sparse: ~2.5GB
Dense + Sparse + ColBERT: ~4.0GB
ColBERT 模式下每个文档输出 N 个向量（N = token 数）
→ 存储开销是 Dense 的 100-500 倍！
建议：
- 大多数场景 Dense + Sparse 就够了
- ColBERT 只在需要极致精度的小数据集上用
- 超过 10 万文档不要用 ColBERT
11. 总结
核心结论
OpenAI 不再是默认最优——2026 年有好几个更好的选择
非对称嵌入是关键提升——query 和 document 分开编码，检索质量高 2-4%
混合检索是标配——Dense + Sparse 在几乎所有场景都比纯 Dense 好
BGE-M3 是开源之王——中文、混合检索、MIT 许可，三个维度都最强
Voyage 是 API 精度之王——纯 Dense 检索精度最高，领域专精模型更胜一筹
长上下文≠不需要分块——RAG 场景分块仍然更好，长上下文适合文档级对比
选型一句话
表格
最终建议： 不要只看 MTEB 分数——在你自己的数据上做 A/B 测试才是最靠谱的选型方式。通用基准和你的业务场景可能差很远。
Embedding 模型是 RAG 的地基，选错模型就像在沙滩上建楼——再漂亮的 reranker 和 prompt 也救不回来。2026 年没有"最好的"Embedding，只有"最适合你的"Embedding。
——PySuper | zhengxingtao.com
参考链接：
评论区