Embedding 模型选型：OpenAI / BGE / Jina 效果与成本对比

本文是【AI 专题精讲】系列第 04 篇。 上一篇：RAG 大文件上传：分片上传、断点续传、进度追踪 | 下一篇：向量数据库选型：ChromaDB / pgvector / Pinecone / Milvus

---

这篇文章你会得到什么

前三篇我们搞定了文件解析、文档切片、大文件上传。现在文本已经切成了一段段 chunk，下一步就是把它们变成向量存进数据库，供后续检索。

这一步叫 Embedding——整条 RAG 链路的基石。

Embedding 模型选错了，后面做得再花哨也白搭：检索不准、召回率低、用户问一句”退货流程是什么”返回的是”公司简介”，那体验直接归零。

今天的目标：搞懂 Embedding 是什么、主流模型有哪些、怎么选、怎么评测。不光给你理论，还给你一个评测脚本，用自己的数据跑一把就知道哪个模型适合你。

---

Embedding 到底是什么

一句话：把文本变成一串数字（向量），语义相近的文本在向量空间里距离近。

"如何退货" → [0.12, -0.34, 0.56, ..., 0.08]   (1536维)
"退款流程" → [0.11, -0.33, 0.55, ..., 0.09]   (1536维)
"公司简介" → [-0.45, 0.23, -0.12, ..., 0.67]  (1536维)

“如何退货”和”退款流程”的向量非常接近（余弦相似度 > 0.9），而”公司简介”和它们的向量差距很大（余弦相似度 < 0.3）。

这就是 RAG 检索的核心原理：用户的问题也转成向量，然后在向量库里找最近的那几个 chunk 返回。

余弦相似度

两个向量的相似度通常用余弦相似度衡量：

import numpy as np

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

值域 [-1, 1]，越接近 1 越相似。实际使用中 > 0.8 可以认为是高度相关。

---

主流 Embedding 模型对比

1. OpenAI text-embedding-3

OpenAI 的第三代 Embedding 模型，分 small 和 large 两个版本：

from openai import OpenAI

client = OpenAI()

def get_openai_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(
        input=text,
        model=model,
    )
    return response.data[0].embedding

参数	text-embedding-3-small	text-embedding-3-large
维度	1536	3072
价格	$0.02 / 1M tokens	$0.13 / 1M tokens
MTEB 排名	中上	前列
中文效果	良好	优秀

特色功能：维度裁剪。 可以通过 dimensions 参数把 3072 维压缩到任意维度，比如 256 维，在牺牲少量精度的前提下大幅减少存储和计算成本：

response = client.embeddings.create(
    input="退货流程",
    model="text-embedding-3-large",
    dimensions=256,
)

2. BGE-M3（智源 BAAI）

国产开源模型，MTEB 中文排行榜长期霸榜，支持多语言、多粒度、多功能：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")

def get_bge_embedding(texts: list[str]) -> list[list[float]]:
    embeddings = model.encode(texts, normalize_embeddings=True)
    return embeddings.tolist()

参数	BGE-M3
维度	1024
价格	免费（本地部署）
最大长度	8192 tokens
中文效果	极佳（专门优化）
部署方式	HuggingFace / Ollama

BGE-M3 的杀手锏：同时支持 Dense、Sparse、ColBERT 三种检索模式，后面讲检索优化时会详细展开。

3. Jina Embeddings v3

Jina AI 出品，主打多语言和长上下文：

import httpx

JINA_API_KEY = "jina_xxxx"

def get_jina_embedding(texts: list[str]) -> list[list[float]]:
    response = httpx.post(
        "https://api.jina.ai/v1/embeddings",
        headers={"Authorization": f"Bearer {JINA_API_KEY}"},
        json={
            "model": "jina-embeddings-v3",
            "input": texts,
            "task": "retrieval.passage",
        },
    )
    data = response.json()
    return [item["embedding"] for item in data["data"]]

参数	Jina Embeddings v3
维度	1024（默认）
价格	$0.02 / 1M tokens
最大长度	8192 tokens
中文效果	优秀
特色	区分 query/passage 角色

Jina 的独特设计：区分 retrieval.query 和 retrieval.passage 两种 task。存文档时用 passage，查询时用 query，模型会针对不同角色做优化。

4. 其他值得关注的模型

模型	厂商	维度	价格	特点
text-embedding-v3	阿里通义	1024/2048	¥0.7 / 1M tokens	国内 API 访问快
Embedding-V1	百度文心	1024	¥0.2 / 1K tokens	百度生态
nomic-embed-text	Nomic AI	768	免费（本地）	轻量，适合嵌入式
mxbai-embed-large	Mixedbread	1024	免费（本地）	Ollama 直接跑

---

选型五维度

维度一：中文效果

如果你的知识库主要是中文内容，这是最重要的维度。

我的实测排名（主观，基于企业文档场景）：

BGE-M3 ≈ Jina v3 > OpenAI large > 通义 v3 > OpenAI small

BGE-M3 和 Jina v3 对中文的理解明显更好，尤其是专业术语和长句。

维度二：成本

模型	10 万条文档（约 5000 万 tokens）	月均查询（10 万次）
OpenAI small	$1	$0.2
OpenAI large	$6.5	$1.3
Jina v3	$1	$0.2
通义 v3	¥35	¥7
BGE-M3（本地）	电费	电费

成本差距可以到 10 倍以上。 如果你有 GPU 服务器，本地部署 BGE-M3 几乎零成本。

维度三：速度

API 调用受网络影响大，这里只看本地推理速度（RTX 4090）：

模型	1000 条文本 Embedding 时间
BGE-M3	~3 秒
nomic-embed-text	~1.5 秒
OpenAI small（API）	~8 秒（含网络延迟）

本地模型在速度上碾压 API 调用，尤其是批量处理场景。

维度四：维度与存储

维度越高，表达能力越强，但存储和检索成本也越高：

维度	单条向量大小	10 万条占用
256	1 KB	100 MB
768	3 KB	300 MB
1024	4 KB	400 MB
1536	6 KB	600 MB
3072	12 KB	1.2 GB

对大部分中文 RAG 场景，1024 维是甜蜜点——效果够好，存储可控。

维度五：部署复杂度

模型	部署方式	最低要求
OpenAI	API 调用	无（需翻墙或代理）
Jina	API 调用	无
通义	API 调用	无（国内直连）
BGE-M3	HuggingFace / Ollama	8GB 显存 GPU
nomic-embed-text	Ollama	4GB 内存即可（CPU）

---

统一封装：支持多模型切换

生产环境不要把模型写死。封装一个统一接口，方便随时切换：

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class EmbeddingResult:
    embeddings: list[list[float]]
    model: str
    dimensions: int
    token_usage: int


class EmbeddingProvider(ABC):
    @abstractmethod
    def embed(self, texts: list[str]) -> EmbeddingResult:
        ...


class OpenAIEmbedding(EmbeddingProvider):
    def __init__(self, model: str = "text-embedding-3-small", dimensions: int | None = None):
        from openai import OpenAI
        self.client = OpenAI()
        self.model = model
        self.dimensions = dimensions

    def embed(self, texts: list[str]) -> EmbeddingResult:
        kwargs = {"input": texts, "model": self.model}
        if self.dimensions:
            kwargs["dimensions"] = self.dimensions

        response = self.client.embeddings.create(**kwargs)

        return EmbeddingResult(
            embeddings=[item.embedding for item in response.data],
            model=self.model,
            dimensions=len(response.data[0].embedding),
            token_usage=response.usage.total_tokens,
        )


class BGEEmbedding(EmbeddingProvider):
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer(model_name)
        self.model_name = model_name

    def embed(self, texts: list[str]) -> EmbeddingResult:
        embeddings = self.model.encode(texts, normalize_embeddings=True)
        return EmbeddingResult(
            embeddings=embeddings.tolist(),
            model=self.model_name,
            dimensions=embeddings.shape[1],
            token_usage=0,
        )


class JinaEmbedding(EmbeddingProvider):
    def __init__(self, api_key: str, model: str = "jina-embeddings-v3"):
        self.api_key = api_key
        self.model = model

    def embed(self, texts: list[str]) -> EmbeddingResult:
        import httpx
        response = httpx.post(
            "https://api.jina.ai/v1/embeddings",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "model": self.model,
                "input": texts,
                "task": "retrieval.passage",
            },
            timeout=30,
        )
        data = response.json()
        embeddings = [item["embedding"] for item in data["data"]]

        return EmbeddingResult(
            embeddings=embeddings,
            model=self.model,
            dimensions=len(embeddings[0]),
            token_usage=data.get("usage", {}).get("total_tokens", 0),
        )


def create_embedding_provider(provider: str, **kwargs) -> EmbeddingProvider:
    providers = {
        "openai": OpenAIEmbedding,
        "bge": BGEEmbedding,
        "jina": JinaEmbedding,
    }
    if provider not in providers:
        raise ValueError(f"不支持的 provider: {provider}，可选: {list(providers.keys())}")
    return providers[provider](**kwargs)

使用方式：

# 开发环境用 OpenAI（方便调试）
provider = create_embedding_provider("openai")

# 生产环境切本地模型（零成本）
provider = create_embedding_provider("bge")

# 调用完全一致
result = provider.embed(["如何退货", "退款流程", "公司地址"])
print(f"维度: {result.dimensions}, Token: {result.token_usage}")

---

实测评测：同一批数据跑三个模型

光看参数没用，得拿自己的数据跑。下面是一个评测脚本，原理：

1. 准备一批”问题 → 正确 chunk”对（Ground Truth）
2. 用不同模型把所有 chunk Embed
3. 对每个问题检索 Top-K，看正确答案排在第几
4. 算 Hit Rate 和 MRR

评测数据准备

test_cases = [
    {
        "query": "如何申请退货",
        "relevant_chunks": ["chunk_id_12", "chunk_id_15"],
    },
    {
        "query": "年假有多少天",
        "relevant_chunks": ["chunk_id_42"],
    },
    {
        "query": "报销流程是什么",
        "relevant_chunks": ["chunk_id_67", "chunk_id_68"],
    },
]

# 所有知识库 chunks
all_chunks = [
    {"id": "chunk_id_1", "content": "公司成立于2010年..."},
    {"id": "chunk_id_12", "content": "退货申请需在签收后7天内提交..."},
    # ... 更多 chunks
]

评测脚本

import numpy as np
from dataclasses import dataclass


@dataclass
class EvalResult:
    model_name: str
    hit_rate_at_3: float
    hit_rate_at_5: float
    hit_rate_at_10: float
    mrr: float
    avg_latency_ms: float
    total_tokens: int


def evaluate_embedding_model(
    provider: EmbeddingProvider,
    chunks: list[dict],
    test_cases: list[dict],
    top_k: int = 10,
) -> EvalResult:
    import time

    # 1. Embed 所有 chunks
    chunk_texts = [c["content"] for c in chunks]
    chunk_ids = [c["id"] for c in chunks]

    start = time.time()
    chunk_result = provider.embed(chunk_texts)
    embed_time = (time.time() - start) * 1000

    chunk_vectors = np.array(chunk_result.embeddings)

    # 归一化（余弦相似度 = 点积，前提是归一化）
    norms = np.linalg.norm(chunk_vectors, axis=1, keepdims=True)
    chunk_vectors = chunk_vectors / norms

    hits_at_3, hits_at_5, hits_at_10 = 0, 0, 0
    reciprocal_ranks = []
    total_tokens = chunk_result.token_usage

    # 2. 对每个 query 检索
    latencies = []
    for case in test_cases:
        query = case["query"]
        relevant_ids = set(case["relevant_chunks"])

        start = time.time()
        query_result = provider.embed([query])
        latencies.append((time.time() - start) * 1000)
        total_tokens += query_result.token_usage

        query_vec = np.array(query_result.embeddings[0])
        query_vec = query_vec / np.linalg.norm(query_vec)

        # 余弦相似度 = 点积（已归一化）
        scores = chunk_vectors @ query_vec
        top_indices = np.argsort(scores)[::-1][:top_k]
        top_ids = [chunk_ids[i] for i in top_indices]

        # Hit Rate
        if relevant_ids & set(top_ids[:3]):
            hits_at_3 += 1
        if relevant_ids & set(top_ids[:5]):
            hits_at_5 += 1
        if relevant_ids & set(top_ids[:10]):
            hits_at_10 += 1

        # MRR
        for rank, cid in enumerate(top_ids, 1):
            if cid in relevant_ids:
                reciprocal_ranks.append(1.0 / rank)
                break
        else:
            reciprocal_ranks.append(0.0)

    n = len(test_cases)
    return EvalResult(
        model_name=provider.__class__.__name__,
        hit_rate_at_3=hits_at_3 / n,
        hit_rate_at_5=hits_at_5 / n,
        hit_rate_at_10=hits_at_10 / n,
        mrr=sum(reciprocal_ranks) / n,
        avg_latency_ms=sum(latencies) / len(latencies),
        total_tokens=total_tokens,
    )

运行评测

providers = [
    ("OpenAI small", create_embedding_provider("openai", model="text-embedding-3-small")),
    ("OpenAI large", create_embedding_provider("openai", model="text-embedding-3-large")),
    ("BGE-M3", create_embedding_provider("bge")),
]

results = []
for name, provider in providers:
    result = evaluate_embedding_model(provider, all_chunks, test_cases)
    result.model_name = name
    results.append(result)

# 打印对比表
print(f"{'模型':<16} {'Hit@3':<8} {'Hit@5':<8} {'Hit@10':<8} {'MRR':<8} {'延迟(ms)':<10} {'Tokens':<8}")
print("-" * 70)
for r in results:
    print(f"{r.model_name:<16} {r.hit_rate_at_3:<8.2%} {r.hit_rate_at_5:<8.2%} {r.hit_rate_at_10:<8.2%} {r.mrr:<8.3f} {r.avg_latency_ms:<10.1f} {r.total_tokens:<8}")

我的实测结果（企业中文文档场景，约 500 chunks）

模型	Hit@3	Hit@5	Hit@10	MRR	延迟
BGE-M3	89%	94%	97%	0.85	3ms
Jina v3	86%	92%	96%	0.82	120ms
OpenAI large	84%	91%	95%	0.80	180ms
OpenAI small	78%	85%	92%	0.73	150ms

数据仅供参考，不同场景差异可能很大。建议用自己的数据跑一遍。

---

实用技巧

1. 批量 Embedding 的正确姿势

一次性传多条文本，比逐条调用高效得多：

# 错误：逐条调用，100 条文档要 100 次 API 请求
for chunk in chunks:
    embedding = provider.embed([chunk.content])

# 正确：批量调用，1 次请求搞定
batch_size = 100
for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i + batch_size]
    result = provider.embed([c.content for c in batch])

OpenAI 单次最多 2048 条输入，BGE 本地模型建议按显存调整 batch_size。

2. Query 和 Document 的区别

有些模型会区分”查询文本”和”文档文本”的 Embedding。比如 BGE 推荐在查询文本前加前缀：

# BGE 推荐的查询前缀
query_embedding = model.encode(
    ["Represent this sentence for searching relevant passages: " + query],
    normalize_embeddings=True,
)

# 文档不加前缀
doc_embeddings = model.encode(
    [chunk.content for chunk in chunks],
    normalize_embeddings=True,
)

Jina 则通过 task 参数区分：

# 存文档
jina_embed(texts, task="retrieval.passage")

# 查询
jina_embed([query], task="retrieval.query")

3. 长文本处理

大部分模型有长度限制（512 或 8192 tokens）。超长文本需要截断或分段：

def safe_embed(provider: EmbeddingProvider, texts: list[str], max_length: int = 8000) -> EmbeddingResult:
    processed = []
    for text in texts:
        if len(text) > max_length:
            processed.append(text[:max_length])
        else:
            processed.append(text)
    return provider.embed(processed)

更好的做法是在切片阶段就控制好 chunk 大小（第 02 篇的内容），确保不超过 Embedding 模型的最大长度。

4. 缓存 Embedding 结果

同一段文本反复 Embed 是浪费钱。生产环境一定要缓存：

import hashlib
import json
from pathlib import Path

CACHE_DIR = Path("./embedding_cache")
CACHE_DIR.mkdir(exist_ok=True)


def cached_embed(provider: EmbeddingProvider, texts: list[str]) -> EmbeddingResult:
    cache_key = hashlib.md5(
        json.dumps(texts, ensure_ascii=False).encode()
    ).hexdigest()

    cache_file = CACHE_DIR / f"{provider.__class__.__name__}_{cache_key}.json"

    if cache_file.exists():
        data = json.loads(cache_file.read_text())
        return EmbeddingResult(**data)

    result = provider.embed(texts)
    cache_file.write_text(json.dumps({
        "embeddings": result.embeddings,
        "model": result.model,
        "dimensions": result.dimensions,
        "token_usage": result.token_usage,
    }))

    return result

---

我的选型建议

场景 1：快速原型 / 个人项目

推荐 OpenAI text-embedding-3-small

理由：一行代码搞定，不用部署，便宜够用。中文效果虽不是最好但能接受。

场景 2：企业中文知识库

推荐 BGE-M3（本地部署）

理由：中文效果最好，免费，数据不出公司内网。需要一台 8GB+ 显存的 GPU 服务器。

场景 3：没有 GPU 但要好效果

推荐 Jina v3 API

理由：中文效果接近 BGE-M3，API 调用无需部署，价格和 OpenAI small 一样。

场景 4：预算有限 + 只有 CPU

推荐 Ollama + nomic-embed-text

ollama pull nomic-embed-text

import httpx

def ollama_embed(texts: list[str]) -> list[list[float]]:
    response = httpx.post(
        "http://localhost:11434/api/embed",
        json={"model": "nomic-embed-text", "input": texts},
    )
    return response.json()["embeddings"]

768 维，CPU 就能跑，效果能覆盖大部分场景。

决策流程图

你的知识库主要是中文吗？
├── 是 → 有 GPU 服务器吗？
│       ├── 是 → BGE-M3（最优解）
│       └── 否 → 数据能上云吗？
│               ├── 是 → Jina v3 API
│               └── 否 → Ollama + nomic-embed-text（CPU 跑）
└── 否（英文/多语言）→ OpenAI text-embedding-3-large

---

总结

1. Embedding 是 RAG 的基石——选错模型，后续再怎么优化检索都救不回来。
2. 中文场景首选 BGE-M3——开源免费、效果一流、本地部署数据安全。
3. 没有 GPU 就用 API——Jina v3 和 OpenAI small 价格相当，Jina 中文效果更好。
4. 务必用自己的数据评测——别信别人的 benchmark，跑一把 Hit Rate 和 MRR 才是王道。
5. 统一封装接口——EmbeddingProvider 抽象类让你随时切换模型，零改动。
6. 批量处理 + 缓存——这两个优化能省 80% 的成本和时间。

下一篇我们解决向量存哪的问题：ChromaDB、pgvector、Pinecone、Milvus 四个主流向量库，哪个适合你的场景？

---

下一篇预告：05 | 向量数据库选型：ChromaDB / pgvector / Pinecone / Milvus

---

讨论话题：你现在用的哪个 Embedding 模型？踩过什么坑？评论区聊聊。