skillhub-137-rag-architect/references/embedding-models.md

嵌入模型

---

模型对比矩阵

模型	维度	最大 Token	优势	提供商
text-embedding-3-large	3072（或 256-3072）	8191	质量最佳，维度灵活	OpenAI
text-embedding-3-small	1536（或 256-1536）	8191	性价比高，质量良好	OpenAI
embed-english-v3.0	1024	512	压缩效果优秀，速度快	Cohere
embed-multilingual-v3.0	1024	512	支持 100+ 种语言	Cohere
voyage-large-2	1536	16000	长上下文，代码感知	Voyage AI
voyage-code-2	1536	16000	代码检索专用	Voyage AI
BGE-large-en-v1.5	1024	512	开源，质量高	BAAI
BGE-M3	1024	8192	多语言、多粒度	BAAI
E5-large-v2	1024	512	基准测试表现强劲	Microsoft
GTE-large	1024	512	通用性良好	Alibaba
all-MiniLM-L6-v2	384	256	快速、轻量	Sentence Transformers
nomic-embed-text-v1.5	768	8192	长上下文，开源权重	Nomic AI

---

各模型适用场景

OpenAI text-embedding-3-large

最佳用途：
- 对精度要求最高的生产级 RAG
- 有质量 SLA 的企业级应用
- 维度要求灵活（可降低维度以节省成本）
- 英文及主要语言

避免场景：
- 对成本敏感的高量级应用
- 气隙或离线部署环境
- 无微调预算的专有领域

OpenAI text-embedding-3-small

最佳用途：
- 性价比优先的生产部署
- 良好的质量价格比
- 通用检索任务
- 利用 API 简洁性快速原型开发

避免场景：
- 需要最高精度
- 专有技术领域
- 必须使用开源方案

Cohere embed-v3

最佳用途：
- 多语言应用（100+ 种语言）
- 面向搜索优化的检索（search_document/search_query 类型）
- 内置压缩（int8/binary 量化）
- 有成本约束的生产环境

避免场景：
- 超长文档（512 token 限制）
- 代码密集的检索任务

Voyage AI

最佳用途：
- 代码检索与技术文档
- 长上下文文档（16K tokens）
- 领域特定微调选项
- 法律/金融专用模型

避免场景：
- 预算受限的项目
- 简单的通用检索

BGE / E5（开源）

最佳用途：
- 自托管部署
- 气隙环境
- 零成本（无 API 费用）
- 自定义领域微调

避免场景：
- 团队没有 GPU 基础设施
- 需要零维护
- 需要最高的开箱即用质量

---

OpenAI 嵌入

from openai import OpenAI

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

def get_embedding(
    text: str,
    model: str = "text-embedding-3-small",
    dimensions: int | None = None
) -> list[float]:
    """获取嵌入向量，可选维度缩减。"""
    params = {"input": text, "model": model}
    if dimensions:
        params["dimensions"] = dimensions

    response = client.embeddings.create(**params)
    return response.data[0].embedding

# 单个嵌入
embedding = get_embedding("How do I install the software?")

# 批量嵌入（更高效）
def get_embeddings_batch(
    texts: list[str],
    model: str = "text-embedding-3-small",
    dimensions: int | None = None
) -> list[list[float]]:
    """批量嵌入多个文本。"""
    params = {"input": texts, "model": model}
    if dimensions:
        params["dimensions"] = dimensions

    response = client.embeddings.create(**params)
    # 按索引排序以保持顺序
    return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]

embeddings = get_embeddings_batch(["text1", "text2", "text3"])

# 维度缩减（节省成本/存储空间）
# text-embedding-3-large: 3072 -> 1024（节省 66% 存储空间）
reduced_embedding = get_embedding(
    "Installation guide...",
    model="text-embedding-3-large",
    dimensions=1024  # 从 3072 缩减
)

维度权衡

原始维度	缩减后	质量损失	存储节省
3072	1536	~1-2%	50%
3072	1024	~2-4%	67%
3072	512	~5-8%	83%
3072	256	~10-15%	92%

---

Cohere 嵌入

import cohere

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

# 文档嵌入（用于索引）
doc_embeddings = co.embed(
    texts=["Installation guide content...", "Configuration steps..."],
    model="embed-english-v3.0",
    input_type="search_document",  # 用于被索引的文档
    truncate="END"
).embeddings

# 查询嵌入（用于搜索）
query_embedding = co.embed(
    texts=["how to install"],
    model="embed-english-v3.0",
    input_type="search_query",  # 用于搜索查询
).embeddings[0]

# 多语言
multilingual_embedding = co.embed(
    texts=["Comment installer le logiciel?"],  # 法语
    model="embed-multilingual-v3.0",
    input_type="search_query"
).embeddings[0]

# 压缩嵌入（int8）
compressed = co.embed(
    texts=["Document content..."],
    model="embed-english-v3.0",
    input_type="search_document",
    embedding_types=["int8"]  # 比 float32 小 4 倍
).embeddings

Cohere 输入类型

类型	用途
search_document	待索引到向量数据库中的文档
search_query	用户搜索查询
classification	文本分类任务
clustering	文档聚类

---

Voyage AI 嵌入

import voyageai

vo = voyageai.Client(api_key="your-api-key")

# 通用嵌入
result = vo.embed(
    texts=["Installation guide for the software..."],
    model="voyage-large-2",
    input_type="document"
)
embeddings = result.embeddings

# 代码嵌入（专用）
code_result = vo.embed(
    texts=[
        "def install_package(name):\n    subprocess.run(['pip', 'install', name])",
        "How do I install packages in Python?"
    ],
    model="voyage-code-2",
    input_type="document"  # 搜索时使用 "query"
)

# 长上下文（最高 16K tokens）
long_doc_embedding = vo.embed(
    texts=[very_long_document],  # 最高 16K tokens
    model="voyage-large-2",
    input_type="document"
).embeddings[0]

---

开源模型（Sentence Transformers）

from sentence_transformers import SentenceTransformer

# 加载模型（首次使用时会下载）
model = SentenceTransformer("BAAI/bge-large-en-v1.5")

# 单个嵌入
embedding = model.encode("How do I install the software?")

# 批量编码（GPU 加速）
embeddings = model.encode(
    ["doc1", "doc2", "doc3"],
    batch_size=32,
    show_progress_bar=True,
    convert_to_numpy=True,
    normalize_embeddings=True  # 用于余弦相似度
)

# BGE 要求对查询使用指令前缀
query_embedding = model.encode(
    "Represent this sentence for searching relevant passages: How do I install?"
)

# GPU 加速
model = SentenceTransformer("BAAI/bge-large-en-v1.5", device="cuda")

# 多 GPU 编码
pool = model.start_multi_process_pool()
embeddings = model.encode_multi_process(
    sentences=large_corpus,
    pool=pool,
    batch_size=64
)
model.stop_multi_process_pool(pool)

BGE-M3（多语言、多粒度）

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

# 一次调用获取稠密、稀疏和 colbert 嵌入
output = model.encode(
    ["Installation guide in English", "Guide d'installation en francais"],
    return_dense=True,
    return_sparse=True,
    return_colbert_vecs=True
)

dense_embeddings = output["dense_vecs"]
sparse_embeddings = output["lexical_weights"]
colbert_embeddings = output["colbert_vecs"]

---

嵌入模型微调

何时微调

场景	建议
领域专属术语（法律、医学）	在领域语料上微调
检索精度较低（<80%）	使用难负样本进行微调
查询分布外	使用查询-文档对进行微调
成本优化	微调较小模型以匹配较大模型

使用 Sentence Transformers 微调

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader

# 准备训练数据
train_examples = [
    InputExample(
        texts=["query: how to install", "doc: Installation guide content..."],
        label=1.0  # 相关性分数
    ),
    InputExample(
        texts=["query: how to install", "doc: Unrelated content..."],
        label=0.0  # 负样本
    ),
]

# 加载基础模型
model = SentenceTransformer("BAAI/bge-base-en-v1.5")

# 创建数据加载器
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,
    output_path="./fine-tuned-model"
)

# 或使用 Multiple Negatives Ranking Loss（检索效果更佳）
train_examples_mnrl = [
    InputExample(texts=["query", "positive_doc", "negative_doc1", "negative_doc2"])
]
train_loss = losses.MultipleNegativesRankingLoss(model)

难负样本挖掘

from sentence_transformers import SentenceTransformer
from sentence_transformers.util import semantic_search
import torch

def mine_hard_negatives(
    queries: list[str],
    positives: list[str],
    corpus: list[str],
    model: SentenceTransformer,
    top_k: int = 10
) -> list[InputExample]:
    """从语料库中为每个查询-正样本对挖掘难负样本。"""

    query_embeddings = model.encode(queries, convert_to_tensor=True)
    corpus_embeddings = model.encode(corpus, convert_to_tensor=True)
    positive_set = set(positives)

    examples = []
    for i, query in enumerate(queries):
        # 查找与正样本相似但不相同的文档
        hits = semantic_search(
            query_embeddings[i:i+1],
            corpus_embeddings,
            top_k=top_k + 1
        )[0]

        hard_negatives = [
            corpus[hit["corpus_id"]]
            for hit in hits
            if corpus[hit["corpus_id"]] not in positive_set
        ][:3]  # 取前 3 个难负样本

        examples.append(InputExample(
            texts=[query, positives[i]] + hard_negatives
        ))

    return examples

---

嵌入流水线最佳实践

文本预处理

import re
from typing import Callable

def clean_for_embedding(text: str) -> str:
    """在嵌入前清理文本。"""
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text)
    # 移除无意义的特殊字符
    text = re.sub(r'[^\w\s\.\,\!\?\-\:\;\(\)]', '', text)
    # 截断到合理长度（取决于模型）
    text = text[:8000]  # 为分词扩展预留空间
    return text.strip()

def preprocess_for_embedding(
    text: str,
    prefix: str = "",
    max_length: int = 8000
) -> str:
    """预处理并可选添加前缀（适用于指令微调模型）。"""
    cleaned = clean_for_embedding(text)
    prefixed = f"{prefix}{cleaned}" if prefix else cleaned
    return prefixed[:max_length]

# 针对查询的 BGE 风格前缀
query_text = preprocess_for_embedding(
    "how to install",
    prefix="Represent this sentence for searching relevant passages: "
)

嵌入缓存

import hashlib
import json
from functools import lru_cache
from pathlib import Path

class EmbeddingCache:
    """基于磁盘的嵌入缓存。"""

    def __init__(self, cache_dir: str = ".embedding_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def _hash_key(self, text: str, model: str) -> str:
        content = f"{model}:{text}"
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, text: str, model: str) -> list[float] | None:
        key = self._hash_key(text, model)
        cache_file = self.cache_dir / f"{key}.json"
        if cache_file.exists():
            return json.loads(cache_file.read_text())
        return None

    def set(self, text: str, model: str, embedding: list[float]) -> None:
        key = self._hash_key(text, model)
        cache_file = self.cache_dir / f"{key}.json"
        cache_file.write_text(json.dumps(embedding))

# 使用示例
cache = EmbeddingCache()

def get_embedding_cached(text: str, model: str = "text-embedding-3-small") -> list[float]:
    cached = cache.get(text, model)
    if cached:
        return cached

    embedding = get_embedding(text, model)  # 调用 API
    cache.set(text, model, embedding)
    return embedding

批处理策略

from typing import Iterator
import asyncio
from openai import AsyncOpenAI

def batch_texts(texts: list[str], batch_size: int = 100) -> Iterator[list[str]]:
    """生成文本批次。"""
    for i in range(0, len(texts), batch_size):
        yield texts[i:i + batch_size]

async def get_embeddings_async(
    texts: list[str],
    model: str = "text-embedding-3-small",
    batch_size: int = 100,
    max_concurrent: int = 5
) -> list[list[float]]:
    """带并发控制的异步批量嵌入。"""
    client = AsyncOpenAI()
    semaphore = asyncio.Semaphore(max_concurrent)

    async def embed_batch(batch: list[str]) -> list[list[float]]:
        async with semaphore:
            response = await client.embeddings.create(
                input=batch,
                model=model
            )
            return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]

    batches = list(batch_texts(texts, batch_size))
    results = await asyncio.gather(*[embed_batch(b) for b in batches])

    # 展平结果
    return [emb for batch_result in results for emb in batch_result]

---

模型选择流程图

开始
  │
  ├─ 需要离线/自托管？
  │   └─ 是 → BGE-large 或 E5-large（开源）
  │
  ├─ 需要多语言？
  │   └─ 是 → Cohere embed-multilingual-v3 或 BGE-M3
  │
  ├─ 代码/技术文档？
  │   └─ 是 → Voyage-code-2
  │
  ├─ 长文档（超过 8K tokens）？
  │   └─ 是 → Voyage-large-2 或 nomic-embed-text
  │
  ├─ 成本是首要考虑？
  │   └─ 是 → text-embedding-3-small（缩减维度）
  │
  ├─ 需要最高质量？
  │   └─ 是 → text-embedding-3-large
  │
  └─ 默认 → text-embedding-3-small（最佳平衡）

---

快速参考

任务	推荐模型
生产级 RAG（英文）	text-embedding-3-small/large
多语言	Cohere embed-multilingual-v3
代码检索	Voyage-code-2
自托管	BGE-large-en-v1.5
长文档	Voyage-large-2、nomic-embed-text
原型开发	all-MiniLM-L6-v2（快速、免费）
最高质量	text-embedding-3-large
成本优化	text-embedding-3-small @ 512 维

相关技能

• RAG 架构师 - 向量数据库集成
• Python 高手 - 异步嵌入流水线
• ML 流水线 - 嵌入模型部署
• 微调专家 - 自定义嵌入训练