用 bge-m3 自托管 embedding 服务（替代 OpenAI text-embedding API）

起因

我们的 RAG 系统每天调 OpenAI text-embedding-3-small 几十万次，账单
$300+/月。而且内部知识库不能出公网。bge-m3 是智源开源的多语言 embedding
模型，质量在 MTEB 中文榜上经常排前 3，比 OpenAI 3-small 还好。
本地跑一张 4090 就够。

解决方案：FastAPI 包一层 OpenAI 兼容接口

让现有用 OpenAI Python SDK 的代码改一行 base_url 就能切。

装

uv add fastapi 'uvicorn[standard]' sentence-transformers
# 第一次启动会下载 bge-m3 模型 (~1.2 GB)

service.py

import logging
from contextlib import asynccontextmanager
from typing import List

import torch
from FlagEmbedding import BGEM3FlagModel
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

log = logging.getLogger('embedding')

class EmbeddingRequest(BaseModel):
    input: str | List[str]
    model: str = 'bge-m3'
    encoding_format: str | None = 'float'

class EmbeddingData(BaseModel):
    object: str = 'embedding'
    embedding: List[float]
    index: int

class EmbeddingResponse(BaseModel):
    object: str = 'list'
    data: List[EmbeddingData]
    model: str
    usage: dict

model: BGEM3FlagModel | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    log.info('loading bge-m3 ...')
    model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=torch.cuda.is_available())
    log.info('ready')
    yield

app = FastAPI(lifespan=lifespan)

@app.post('/v1/embeddings', response_model=EmbeddingResponse)
def embed(req: EmbeddingRequest):
    if model is None:
        raise HTTPException(503, 'model loading')
    texts = [req.input] if isinstance(req.input, str) else req.input
    if not texts:
        raise HTTPException(400, 'empty input')
    if len(texts) > 256:
        raise HTTPException(400, 'batch too large')

    out = model.encode(texts, batch_size=32, max_length=8192,
                       return_dense=True)['dense_vecs']
    data = [EmbeddingData(embedding=vec.tolist(), index=i)
            for i, vec in enumerate(out)]
    total_tokens = sum(len(t) for t in texts) // 4   # 粗估
    return EmbeddingResponse(
        data=data, model='bge-m3',
        usage={'prompt_tokens': total_tokens, 'total_tokens': total_tokens},
    )

@app.get('/health')
def health():
    return {'ok': model is not None}

起服务

uv run uvicorn service:app --host 0.0.0.0 --port 8001
# 或生产：
uv run gunicorn -k uvicorn.workers.UvicornWorker \
  -w 1 -b 0.0.0.0:8001 service:app

注意 -w 1：embedding 是 GPU 密集，多 worker 抢一张卡反而慢。
水平扩容用多机或者把模型放多张卡。

客户端：OpenAI SDK 直接用

from openai import OpenAI

client = OpenAI(
    base_url='http://localhost:8001/v1',
    api_key='local',   # 不校验，随便填
)
resp = client.embeddings.create(
    input=['你好世界', 'Hello world', '今日天气不错'],
    model='bge-m3',
)
for i, d in enumerate(resp.data):
    print(f'{i}: dim={len(d.embedding)} first 5={d.embedding[:5]}')

性能

我的 RTX 4090 上：

batch=32 文本（每条 ~200 字）：~80ms（≈ 400 texts/s 持续）
batch=1：~20ms

P95 延迟 < 100ms 在企业 RAG 场景完全够。OpenAI 3-small 平均 200-500ms
（网络），自托管反而更快。

效果

月度 embedding 账单 $300 → $0
P95 延迟 320ms → 80ms（无公网往返）
内部敏感文档不出公网
bge-m3 中文 MTEB 评测比 text-embedding-3-small 高 5-8 个点

踩过的坑

第一次冷启动慢：bge-m3 模型 1.2 GB 从 HuggingFace 拉。在国内
设 HF_ENDPOINT=https://hf-mirror.com。
OpenAI SDK 严格校验维度：bge-m3 输出 1024 维，OpenAI 3-small
是 1536。如果客户端代码硬编码 1536 校验，要么改客户端要么改 model
多输出 padding。
use_fp16=True 在某些老 GPU 数值溢出：T4 / 老 CUDA 上偶尔 NaN。
降到 use_fp16=False 慢 30% 但稳。
batch 太大显存爆：bge-m3 max_length=8192，batch=64 × 8192 token
可能 OOM。生产 batch_size=32 + max_length=512（多数 chunk 这个
长度够用）。
multi-process 共享 GPU 模型不共享内存：每个 worker 各自 load 一份
1.2GB。-w 1 单 worker 是正确选择；要并发用 async + 内部 batch
合并请求。