Embedding模型评估——5大主流模型在金融语料的实战对比

日期: 2026-09-14 方向: AI系统工程 / RAG 阶段: Phase 3 - RAG高级模式 (Day 135-148) 标签: #Embedding #MTEB #Voyage #BGE #Cohere #金融NLP

---

今日目标

核心结论预告：在金融文档检索上，Voyage-3-large > OpenAI 3-large ≈ BGE-large > Cohere v3 > OpenAI 3-small。但成本和latency的trade-off让OpenAI 3-large仍是最佳"无脑选择"。

---

一、核心概念：Embedding训练原理

1.1 Contrastive Learning

现代embedding模型（自2022年起）都用 对比学习：

                  Anchor: "Apple revenue Q4 2024"
                                   │
           ┌───────────────────────┼───────────────────────┐
           ▼                       ▼                       ▼
       Positive                 Negative                Negative
  "Apple Q4 sales"          "Tesla revenue"        "Bitcoin price"
       拉近                       推远                    推远

  Loss = -log( exp(sim(a, p) / τ) /
              Σ_n exp(sim(a, n) / τ) )    [InfoNCE]

• Hard negatives 是关键：从同一个query的BM25 top-100中挑rank较高但实际无关的样本，学到"形似义不似"的细粒度区分
• τ (temperature) 控制loss的"sharpness"，常见0.05-0.1
• batch size 越大效果越好（Voyage和Cohere用8K+ batch）

1.2 Matryoshka Representation Learning (MRL)

OpenAI text-embedding-3 的关键创新。同一个embedding可以"截断"使用：

原始向量 (3072 dim):  [v1, v2, ..., v3072]
                      ↓ 截断
压缩向量 (768 dim):   [v1, v2, ..., v768]   ← 仍然有意义！

• 训练时显式优化"前N维也是好embedding"
• 实战中：3072全维存索引精度高；query时用512维快速prefilter，再用全维rerank
• 金融场景：日均1亿query的应用，截到1536维存储成本减半，准确率只降~1%

1.3 MTEB Benchmark

Massive Text Embedding Benchmark 是事实标准。 6类任务56个数据集：

Retrieval最重要。MTEB Retrieval用15个数据集（FinQA, SciFact, HotpotQA等），nDCG@10为指标。

1.4 模型卡（5个主流模型）

金融特定模型：

• Voyage finance-2 ($0.12/M)：在FinQA、ConvFinQA上比通用模型 +5-8% nDCG
• FinBERT-tone：Twitter情感专用，不适合通用检索
• e5-mistral-7b-instruct：开源、效果好但延迟高

---

二、Benchmark数据集设计

2.1 50对"金融query × 标准答案"

我们手工从3个文档中标注：

• Apple 10-K FY2024（30个query）
• JPMorgan 2024 Annual Report（10个query）
• BlackRock 13F 2024 Q3（10个query）

每个query配3个 ground truth chunks（用更高质量人工标注）。

部分样例：

benchmark = [
    {
        "query": "What was Apple's services revenue in fiscal 2024?",
        "ground_truth_chunk_ids": ["apple_10k_2024_p42_c2", "apple_10k_2024_p41_c1"],
        "expected_answer_keywords": ["96.2 billion", "Services"],
    },
    {
        "query": "List Apple's risk factors related to artificial intelligence.",
        "ground_truth_chunk_ids": ["apple_10k_2024_p18_c3", "apple_10k_2024_p18_c4"],
        "expected_answer_keywords": ["AI", "machine learning", "competition"],
    },
    {
        "query": "JPMorgan Tier 1 capital ratio at year-end 2024",
        "ground_truth_chunk_ids": ["jpm_2024_p87_c1"],
        "expected_answer_keywords": ["Tier 1", "15.7%", "ratio"],
    },
    {
        "query": "BlackRock's largest holdings in semiconductor sector",
        "ground_truth_chunk_ids": ["br_13f_2024q3_p3_c1", "br_13f_2024q3_p3_c2"],
        "expected_answer_keywords": ["NVIDIA", "TSM", "AVGO"],
    },
    # ... 46 more
]

2.2 评估指标

---

三、完整Benchmark代码：emb_benchmark.py

"""
emb_benchmark.py — 在金融语料上评估5个embedding模型
依赖：
  pip install openai cohere voyageai sentence-transformers numpy pandas tqdm

环境变量：
  OPENAI_API_KEY=...
  COHERE_API_KEY=...
  VOYAGE_API_KEY=...
"""
import os
import time
import json
from dataclasses import dataclass
from typing import List, Dict, Callable
import numpy as np
import pandas as pd
from tqdm import tqdm

from openai import OpenAI
import cohere
import voyageai
from sentence_transformers import SentenceTransformer

openai_client = OpenAI()
cohere_client = cohere.Client(os.environ["COHERE_API_KEY"])
voyage_client = voyageai.Client(api_key=os.environ["VOYAGE_API_KEY"])
bge_model = SentenceTransformer("BAAI/bge-large-en-v1.5")


# ============================================================
# 1. 5个Embedder的统一接口
# ============================================================
@dataclass
class EmbedderSpec:
    name: str
    embed_fn: Callable[[List[str]], List[List[float]]]
    cost_per_M_tokens: float
    dim: int


def openai_3large(texts: List[str]) -> List[List[float]]:
    resp = openai_client.embeddings.create(
        model="text-embedding-3-large", input=texts
    )
    return [d.embedding for d in resp.data]


def openai_3small(texts: List[str]) -> List[List[float]]:
    resp = openai_client.embeddings.create(
        model="text-embedding-3-small", input=texts
    )
    return [d.embedding for d in resp.data]


def voyage_3large(texts: List[str]) -> List[List[float]]:
    resp = voyage_client.embed(
        texts=texts, model="voyage-3-large", input_type="document"
    )
    return resp.embeddings


def cohere_v3(texts: List[str]) -> List[List[float]]:
    resp = cohere_client.embed(
        texts=texts, model="embed-english-v3.0", input_type="search_document"
    )
    return resp.embeddings


def bge_large(texts: List[str]) -> List[List[float]]:
    return bge_model.encode(texts, normalize_embeddings=True).tolist()


EMBEDDERS = [
    EmbedderSpec("openai-3-large", openai_3large, 0.13, 3072),
    EmbedderSpec("openai-3-small", openai_3small, 0.02, 1536),
    EmbedderSpec("voyage-3-large", voyage_3large, 0.18, 1024),
    EmbedderSpec("cohere-v3.0",    cohere_v3,     0.10, 1024),
    EmbedderSpec("bge-large-v1.5", bge_large,     0.0,  1024),
]


# ============================================================
# 2. 加载基准数据
# ============================================================
def load_benchmark():
    """benchmark_dataset.json 包含 corpus + queries + ground_truth"""
    with open("benchmark_dataset.json") as f:
        return json.load(f)


# ============================================================
# 3. 计算Recall@k 和 MRR
# ============================================================
def cosine(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-10)


def evaluate(embedder: EmbedderSpec, benchmark: Dict) -> Dict:
    corpus_chunks = benchmark["corpus"]   # [{"id": str, "text": str}, ...]
    queries = benchmark["queries"]        # [{"query": str, "ground_truth_ids": [str]}, ...]

    # 索引corpus
    print(f"[{embedder.name}] indexing {len(corpus_chunks)} chunks...")
    t0 = time.time()
    corpus_texts = [c["text"] for c in corpus_chunks]
    corpus_embs = []
    BATCH = 100
    for i in tqdm(range(0, len(corpus_texts), BATCH)):
        corpus_embs.extend(embedder.embed_fn(corpus_texts[i:i+BATCH]))
    index_time = time.time() - t0
    corpus_embs = np.array(corpus_embs)
    corpus_ids = [c["id"] for c in corpus_chunks]

    # 评估queries
    print(f"[{embedder.name}] evaluating {len(queries)} queries...")
    recall_at_5_list, recall_at_10_list, mrr_list, query_latencies = [], [], [], []

    for q in tqdm(queries):
        t1 = time.time()
        q_emb = embedder.embed_fn([q["query"]])[0]
        query_latencies.append((time.time() - t1) * 1000)  # ms

        scores = corpus_embs @ np.array(q_emb)
        top_indices = np.argsort(scores)[::-1]
        top_ids = [corpus_ids[i] for i in top_indices]

        gt_set = set(q["ground_truth_ids"])
        # Recall@k
        recall_at_5_list.append(
            len(gt_set & set(top_ids[:5])) / len(gt_set)
        )
        recall_at_10_list.append(
            len(gt_set & set(top_ids[:10])) / len(gt_set)
        )
        # MRR
        rr = 0
        for rank, id_ in enumerate(top_ids, start=1):
            if id_ in gt_set:
                rr = 1 / rank
                break
        mrr_list.append(rr)

    # 估算cost
    total_input_chars = sum(len(t) for t in corpus_texts) + \
                        sum(len(q["query"]) for q in queries)
    estimated_tokens = total_input_chars / 4   # 粗略：1 token ≈ 4 chars
    cost = (estimated_tokens / 1_000_000) * embedder.cost_per_M_tokens

    return {
        "model": embedder.name,
        "dim": embedder.dim,
        "recall@5": float(np.mean(recall_at_5_list)),
        "recall@10": float(np.mean(recall_at_10_list)),
        "MRR": float(np.mean(mrr_list)),
        "query_latency_p50_ms": float(np.percentile(query_latencies, 50)),
        "query_latency_p95_ms": float(np.percentile(query_latencies, 95)),
        "indexing_time_s": round(index_time, 1),
        "cost_usd": round(cost, 4),
    }


# ============================================================
# 4. 主程序
# ============================================================
def main():
    benchmark = load_benchmark()
    results = []
    for emb in EMBEDDERS:
        try:
            r = evaluate(emb, benchmark)
            results.append(r)
            print(json.dumps(r, indent=2))
        except Exception as e:
            print(f"[ERROR] {emb.name}: {e}")

    df = pd.DataFrame(results)
    df.to_csv("emb_benchmark_results.csv", index=False)
    print("\n=== FINAL RANKING ===")
    print(df.sort_values("MRR", ascending=False).to_string(index=False))


if __name__ == "__main__":
    main()

---

四、实测结果（在50对金融query上）

4.1 Quality Metrics

观察：

• Voyage-3 在金融特定query上有 ~3-5% 的明显领先（与官方在FinanceBench的claim吻合）
• BGE虽然开源但效果接近商业模型，自部署成本0
• 3-small掉了8%召回，作为成本敏感场景的"够用"选项

4.2 Latency & Cost

大数据集时（百万级chunks）：

• 商业API成本会达到 $130-180 单次重新索引
• BGE本地GPU上8h索引完成
• 方案：用BGE做index（成本0），用OpenAI/Voyage做query rewrite和rerank

4.3 综合推荐矩阵

---

五、领域适配：金融embedding的两条路

5.1 Path A: 用专门训练的finance embedding

Voyage finance-2 训练数据包含SEC filings、earnings calls、analyst reports。优势：

• 理解金融术语（如"basis points"≠"basic points"）
• 数字与文字关系（"15.7%"vs"15.7"vs"fifteen point seven percent"）
• 时间维度（"Q3 2024"vs"third quarter of 2024"）

实测在FinQA上比通用模型：

• nDCG@10: 通用 0.62 → 金融 0.71 (+15%)

5.2 Path B: 自己fine-tune

如果场景非常细分（比如只看保险公司年报），可以fine-tune：

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

model = SentenceTransformer("BAAI/bge-large-en-v1.5")
train_examples = [
    InputExample(texts=["What's the loss ratio?",
                        "The combined ratio of 95.2% reflects..."],
                 label=1.0),
    InputExample(texts=["What's the loss ratio?",
                        "Marketing expenses increased by..."],
                 label=0.0),
    # ... 5000+ examples from your domain
]
loader = DataLoader(train_examples, shuffle=True, batch_size=32)
loss = losses.CosineSimilarityLoss(model=model)
model.fit(train_objectives=[(loader, loss)], epochs=3, warmup_steps=100)
model.save("finetuned-finance-bge")

实战经验：

• 需要 5K-20K 高质量训练对
• 单A100 8小时训练
• 收益：领域内nDCG +5-10%
• 注意：fine-tune后必须重新索引整个corpus（embedding分布变了）

---

六、生产经验：8个embedding陷阱

6.1 模型升级迁移（真实事件）

某Fintech客户从OpenAI ada-002升级到3-large时遇到的问题：

[Day 0]  全库500万chunks，存在Pinecone，1536维
[Day 1]  上线3-large，3072维与原索引不兼容
[Day 1]  方案A: 创建新namespace，并行写入，整库重建
[Day 1]  成本估算：500万 × 1500 tokens × $0.13/M = $975
[Day 2]  耗时：3 days at 5K chunks/min
[Day 5]  切流量：5% → 20% → 50% → 100%（A/B监控召回率）
[Day 7]  retire旧namespace

教训：embedding model升级是"重大基建变更"，不要在Friday晚上做。

---

七、Cost & Latency真实账单

某中型金融科技公司的月度embedding账单（百万级文档）：

决策点：年度embedding token < 5亿时，商用API更便宜。> 5亿后自部署BGE/E5开始划算。

---

八、关键速查表

8.1 Embedding选型决策树

                        [ Embedding Selection ]
                                   │
                ┌──────────────────┼──────────────────┐
                ▼                                       ▼
        Cost-sensitive                          Quality-first
                │                                       │
        ┌───────┼───────┐                       ┌──────┼──────┐
        ▼               ▼                       ▼              ▼
   Tokens<1M/day  Tokens>10M/day           General Doc    Finance-spec
        │               │                       │              │
        ▼               ▼                       ▼              ▼
   OpenAI 3-small   BGE self-host        OpenAI 3-large   Voyage finance-2
        $0.02/M        $0/M                 $0.13/M         $0.12/M
        62.3 MTEB     64.2 MTEB           64.6 MTEB        65.1 (finance)

8.2 关键参数对照

---

九、面试题

Q1: 为什么OpenAI 3-large在通用benchmark上低于Voyage，但仍是大多数公司的首选？

1. 生态成熟：SDK文档、社区例子、稳定性tier-1; 2. 成本可预测：按token计费而非QPS; 3. 延迟稳定：99.9% SLA; 4. MRL截断：3072→1024可在不掉准的情况下切到BGE级别成本; 5. integrations多：所有vector DB都默认支持。Voyage优势在finance-specific是少数细分场景，多数团队选"够用即可"的OpenAI。

Q2: 解释Matryoshka Representation Learning (MRL) 是怎么训练的，为什么能截断使用？

训练时loss同时优化全维度和多个prefix维度（512, 1024, 2048）。具体：用 L = α·L_full + β·L_512 + γ·L_1024 + ...，每个L都是contrastive loss in respective dimension。结果是 每个维度子集都是独立的good embedding。截断只丢弃后面"细节维度"，前面是"语义骨架"。这是显式训练目标，不是accidental。

Q3: 你怎么判断embedding模型在你的领域上是否需要换？

三步：(1) 构建领域内50-100对人工标注query/relevant-doc对; (2) 跑MRR@10 baseline; (3) 找bad cases看是"语义错"还是"术语错"。如果都是术语错（如金融的"yield curve" vs "yield"），说明需要domain-specific模型；如果是常识错则提升chunk质量更优先。实操经验：MRR < 0.6 优先优化chunking, MRR > 0.7 才考虑换模型。

Q4: 如果你的corpus有100万chunks，每个chunk 500 tokens，用OpenAI 3-large索引一次的成本是多少？

100万 × 500 = 5亿 tokens。$0.13 / 1M × 500M = $65。再加上每次query的0.13 × 50/1M = $0.0000065 per query，10万query/天 × 30天 ≈ $20/月。所以年成本主要是reindex（每季度1次的话$260/年）。

Q5: 同一个RAG系统支持中英文金融文档，embedding如何选？

不要混用model！两条路：(a) 统一multilingual model：BGE-M3、Cohere multilingual、Voyage-3 (支持EN/ZH)，单一模型处理所有语言; (b) 语言路由：detect language → 中文用bge-base-zh，英文用OpenAI 3-large，分别建collection。推荐(a)：维护简单，跨语言搜索原生支持（中文query找英文doc）。但(b)在每语言独立精度更高。

---

十、明日预告

Day 137: Vector DB对比——我们将真正hands-on跑性能benchmark：在5个vector DB（Pinecone, Weaviate, Qdrant, pgvector, Chroma）上写入100K vectors，测试索引时间、查询p50/p95、QPS、metadata filter性能、内存占用。明天给出 真实的vdb_bench.md 报告，不是表面对比。