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 嵌入

```python
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 嵌入

```python
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 嵌入

```python
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）

```python
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（多语言、多粒度）

```python
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 微调

```python
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)
```

### 难负样本挖掘

```python
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
```

---

## 嵌入流水线最佳实践

### 文本预处理

```python
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: "
)
```

### 嵌入缓存

```python
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
```

### 批处理策略

```python
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 流水线** - 嵌入模型部署
- **微调专家** - 自定义嵌入训练