💡 LLM 成本优化 / Embedding 成本

🔮 Embedding 成本

Embedding 看起来便宜，但量大了也是真金白银。本地部署 BGE/E5 可以零成本，API 方案也有选择空间。

Embedding 方案对比

成本计算

# 10万篇文档，平均 500 tokens/篇
# 索引一次 + 每日更新 1% + 每天 1000 次查询

documents = 100_000
avg_tokens = 500
daily_updates = documents * 0.01  # 1000
daily_queries = 1_000
avg_query_tokens = 50

# 方案1: OpenAI text-embedding-3-small
index_cost = documents * avg_tokens * 0.02 / 1_000_000  # $1.00
daily_update_cost = daily_updates * avg_tokens * 0.02 / 1_000_000  # $0.01
daily_query_cost = daily_queries * avg_query_tokens * 0.02 / 1_000_000  # $0.001
monthly_api = (daily_update_cost + daily_query_cost) * 30  # $0.33
print(f"OpenAI small: 初始${index_cost:.2f} + 月均${monthly_api:.2f}")

# 方案2: 本地 BGE-M3
# 初始索引: 本地 GPU/CPU 推理，0 API 成本
# 月度: 0
# 硬件: 需要 GPU 或接受 CPU 较慢的速度
print(f"本地 BGE-M3: 初始$0 + 月均$0 (需GPU/CPU)")

# 方案3: 混合——用本地 BGE 做索引，OpenAI 做查询（如果延迟要求高）
print(f"混合方案: 本地索引$0 + 查询月均${daily_query_cost * 30:.2f}")

# 结论：10万文档量级，Embedding 成本本身不高
# 但百万文档以上，API 方案的成本开始显著

本地 Embedding 部署

# 使用 sentence-transformers 部署本地 Embedding
# pip install sentence-transformers

from sentence_transformers import SentenceTransformer
import numpy as np

class LocalEmbeddingService:
    """本地 Embedding 服务"""
    
    # 推荐模型
    MODELS = {
        'bge-m3': 'BAAI/bge-m3',                 # 多语言，推荐中文场景
        'bge-large-zh': 'BAAI/bge-large-zh-v1.5', # 中文专用
        'bge-large-en': 'BAAI/bge-large-en-v1.5', # 英文
        'e5-large': 'intfloat/e5-large-v2',       # 通用
        'gte-large': 'thenlper/gte-large',        # 通用
    }
    
    def __init__(self, model_name: str = 'bge-m3', device: str = None):
        """初始化本地 Embedding 模型
        
        Args:
            model_name: 模型名称或 HuggingFace ID
            device: 'cuda', 'cpu', 或 None（自动选择）
        """
        hf_id = self.MODELS.get(model_name, model_name)
        self.model = SentenceTransformer(hf_id, device=device)
        self.model_name = model_name
    
    def embed(self, texts: list[str], batch_size: int = 32) -> list[list[float]]:
        """生成 Embedding"""
        # BGE 模型需要加前缀
        if 'bge' in self.model_name.lower():
            texts = [f"Represent this sentence: {t}" for t in texts]
        
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=False,
            normalize_embeddings=True  # L2 归一化，直接用于余弦相似度
        )
        return embeddings.tolist()
    
    def embed_query(self, query: str) -> list[float]:
        """查询 Embedding（BGE 需要不同前缀）"""
        if 'bge' in self.model_name.lower():
            query = f"Represent this sentence for searching: {query}"
        
        embedding = self.model.encode(
            [query],
            normalize_embeddings=True
        )
        return embedding[0].tolist()
    
    def similarity(self, query: str, documents: list[str], top_k: int = 5) -> list[dict]:
        """查询最相似的文档"""
        query_emb = self.embed_query(query)
        doc_embs = self.embed(documents)
        
        # 余弦相似度（已归一化，直接点积）
        similarities = np.dot(doc_embs, query_emb)
        
        # 排序
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        return [
            {'index': int(i), 'score': float(similarities[i]), 'text': documents[i]}
            for i in top_indices
        ]

# 部署为 API 服务
"""
from fastapi import FastAPI

app = FastAPI()
embedding_service = LocalEmbeddingService('bge-m3')

@app.post("/embed")
async def embed_texts(texts: list[str]):
    embeddings = embedding_service.embed(texts)
    return {"embeddings": embeddings}

@app.post("/search")
async def search(query: str, documents: list[str], top_k: int = 5):
    results = embedding_service.similarity(query, documents, top_k)
    return {"results": results}
"""

# 性能参考（CPU, bge-m3）
# 单条: ~20ms
# 批量32条: ~100ms
# 批量128条: ~300ms
# GPU (T4): ~5ms/条

混合检索减少 Embedding 调用

class HybridRetriever:
    """混合检索：关键词 + Embedding，减少 Embedding 调用"""
    
    def __init__(self, embedding_service, bm25_index=None):
        self.embedding = embedding_service
        self.bm25 = bm25_index  # BM25 关键词检索
    
    async def search(self, query: str, top_k: int = 10, 
                      alpha: float = 0.5) -> list[dict]:
        """混合检索
        
        Args:
            alpha: Embedding 权重 (0=纯BM25, 1=纯Embedding)
        """
        # Step 1: BM25 关键词检索（快速、免费）
        bm25_results = []
        if self.bm25:
            bm25_results = self.bm25.search(query, top_k=top_k * 3)
        
        # Step 2: Embedding 语义检索（更准确、有成本）
        # 只对 BM25 的候选集做 Embedding，而不是全量
        if bm25_results and alpha < 1.0:
            # 优化：只对候选集计算 Embedding 相似度
            candidate_texts = [r['text'] for r in bm25_results]
            query_emb = self.embedding.embed_query(query)
            doc_embs = self.embedding.embed(candidate_texts)
            
            # 融合分数
            for i, result in enumerate(bm25_results):
                semantic_score = float(np.dot(doc_embs[i], query_emb))
                bm25_score = result.get('score', 0)
                
                # 归一化后加权
                result['combined_score'] = (
                    alpha * semantic_score + 
                    (1 - alpha) * bm25_score
                )
            
            # 按 combined_score 排序
            bm25_results.sort(key=lambda x: x['combined_score'], reverse=True)
            return bm25_results[:top_k]
        
        # 纯 Embedding 检索
        return self.embedding.similarity(query, all_documents, top_k)

# 成本节省分析
# 纯 Embedding: 每次查询 = 1 query embedding + N document embeddings
# 混合: 每次查询 = 1 query embedding + K document embeddings (K << N)
# 节省: (N-K)/N 的 Embedding 调用

Embedding 缓存

import hashlib
import json

class EmbeddingCache:
    """Embedding 缓存——相同文本不重复调用"""
    
    def __init__(self, redis_client, embedding_service, ttl: int = 86400*30):
        self.redis = redis_client
        self.embedding = embedding_service
        self.ttl = ttl
    
    def _cache_key(self, text: str, model: str = None) -> str:
        """生成缓存键"""
        content = f"{model or 'default'}:{text}"
        return f"emb:{hashlib.md5(content.encode()).hexdigest()}"
    
    async def embed(self, texts: list[str], model: str = None) -> list[list[float]]:
        """带缓存的 Embedding"""
        results = [None] * len(texts)
        uncached_indices = []
        uncached_texts = []
        
        # 1. 检查缓存
        for i, text in enumerate(texts):
            key = self._cache_key(text, model)
            cached = await self.redis.get(key)
            if cached:
                results[i] = json.loads(cached)
            else:
                uncached_indices.append(i)
                uncached_texts.append(text)
        
        # 2. 批量获取未缓存的 Embedding
        if uncached_texts:
            new_embeddings = self.embedding.embed(uncached_texts)
            
            # 3. 缓存结果
            for idx, emb in zip(uncached_indices, new_embeddings):
                results[idx] = emb
                key = self._cache_key(texts[idx], model)
                await self.redis.set(key, json.dumps(emb), ex=self.ttl)
        
        return results

← 上一篇：Token 预算 | 下一篇：成本监控 → | 返回首页