如何高效调用ChatGPT Embedding API：从原理到生产环境实践

1次阅读

共计 2112 个字符，预计需要花费 6 分钟才能阅读完成。

Embedding 简单来说就是把文字转换成计算机能理解的数字向量。就像我们给每个单词或句子分配一个独特的身份证号码，但这个号码不是简单的 12345，而是一串有意义的数字组合（通常是 1536 维的浮点数数组）。

核心价值：让计算机能够 ” 理解 ” 文本的语义，比如知道 ” 猫 ” 和 ” 喵星人 ” 的向量很接近
典型应用：
语义搜索（不用完全匹配关键词也能找到相关内容）
文本分类（自动判断评论是好评还是差评）
推荐系统（找到相似内容推荐给用户）

在实际项目中踩过的坑：

API 限流：免费账号每分钟只有 3 次调用，付费账号也可能触达上限
长文本处理：超过 8192token 会被拒绝，需要手动拆分
结果不一致：同样内容多次调用可能得到略有不同的向量
成本控制：按 token 计费，长文档处理可能产生意外账单
延迟问题：网络波动导致响应时间不稳定

import openai
from typing import List, Optional
import backoff  # 需要安装：pip install backoff

@backoff.on_exception(backoff.expo, 
                     openai.error.RateLimitError,
                     max_tries=5)
def get_embedding(text: str, 
                 model: str = "text-embedding-3-small") -> Optional[List[float]]:
    """
    获取单个文本的 embedding 向量
    :param text: 输入文本（建议不超过 8192token）:param model: 模型版本（推荐使用最新 3 系列）:return: 1536 维向量或 None（失败时）"""
    try:
        response = openai.Embedding.create(input=[text],
            model=model
        )
        return response['data'][0]['embedding']
    except Exception as e:
        print(f"Embedding 生成失败: {str(e)}")
        return None

from concurrent.futures import ThreadPoolExecutor
import numpy as np

def batch_embedding(texts: List[str], 
                   batch_size: int = 32) -> List[List[float]]:
    """
    批量处理文本列表（利用多线程加速）:param texts: 文本列表
    :param batch_size: 每批处理数量（根据 API 限流调整）:return: 向量列表
    """
    results = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            vectors = list(executor.map(get_embedding, batch))
            results.extend([v for v in vectors if v is not None])
    return results

from functools import lru_cache
import hashlib

@lru_cache(maxsize=10_000)
def cached_embedding(text: str) -> List[float]:
    """
    带缓存的 embedding 获取（基于文本内容哈希）注意：不同模型版本需要单独缓存
    """
    return get_embedding(text)

def text_hash(text: str) -> str:
    """生成文本唯一标识"""
    return hashlib.md5(text.encode()).hexdigest()

测试环境：AWS t3.xlarge 实例，Python 3.8