国内团队接入 Embedding API,难点通常不在 SDK。真正消耗时间的是链路是否稳定、费用怎么付、发票怎么开,以及几十万条文本批量向量化跑到一半失败时怎么办。Embedding 请求短、并发高、总量大,和聊天 API 的选型逻辑并不相同。

本文比较三条常见路线——官方直连、自建代理、国内 AI 网关——并给出批量任务的可靠性设计、故障切换边界和上线核验清单。具体模型、维度与计费以站内模型价格为准。

说明:OpenAI 及其他模型、服务名称是其各自权利人的商标,本文与相关厂商无隶属关系。跨境网络、数据出境与支付结算需遵守所在地法律法规及平台服务条款,落地前请咨询法务与合规人员。

一、三条路线先摆清楚

官方直连是直接调用模型厂商的接口。链路最短,文档与功能同步快,适合已有跨境网络、海外付款渠道和报销流程的团队。代价是网络质量、美元账单、汇率和发票都要自己处理。

自建代理是在合规服务器上部署反向代理或网关,把鉴权、日志、重试和限流收口。控制力强,但服务器、证书、升级和值班都要自己承担;若只为省事而搭,运维成本往往高于预期。

国内 AI 网关提供 OpenAI 兼容入口,人民币结算,通常带控制台、用量统计和发票。接入成本低,便于统一多个模型;需要确认服务主体、数据条款、限额规则和故障响应。

维度 官方直连 自建代理 国内 AI 网关
网络与接入 依赖跨境链路,需自行优化 线路可自选,需自行运维 国内入口,通常开箱可用
付款与发票 美元结算,报销链路较长 取决于账号与服务器主体 人民币结算,可开发票
批量吞吐 受账号限额约束 可做队列,但仍受上游限制 看网关限额与调度能力
数据边界 自行评估出境与条款 数据位置自行负责 需审查服务商条款与资质
运维成本 中,主要是网络与账号 高,服务器、证书、监控 低,控制台集中管理
模型覆盖 以单一厂商为主 可自行扩展 通常覆盖多家模型

没有一条路线适合所有团队。只用一家模型、网络和付款顺畅,直连最省事;需要固定出口与审计,选自建代理;没有运维人力、又要多模型统一账单,再评估国内网关。

二、网络测试:Embedding 要看吞吐,不只看延迟

聊天 API 关心首 Token 延迟,Embedding 关心的是批量吞吐和稳定性。一次请求可能包含几十到上百条文本,响应体是成百上千个浮点数,链路问题会以超时、截断和 5xx 的形式出现。

测试时不要只发一句“你好”。用真实长度的文档块,按生产批量组装请求,至少观察四项:单批耗时、P50/P95 延迟、错误率和晚高峰表现。还要确认代理或网关的请求体大小限制、连接复用和超时设置;大请求被中间层截断时,报错信息往往并不直观。

长任务要能断点续跑。把输入按内容哈希编号,成功一批就落盘一批,失败只重跑缺失部分,而不是整库重来。

三、付款、合同与发票

国内团队常见的卡点是付款主体。个人卡充值、代付和报销凭证不一致,财务往往无法入账。选方案前先问清楚:合同和谁签、结算币种是什么、能开什么发票、预付款是否有有效期、退款规则如何。

按量计费还要确认账单口径:按输入 Token 计费还是按请求数计费,失败请求是否计入,重试是否重复计费,用量能否按项目或 Key 拆分。把这些写进采购对比表,再用同一批数据实测,才能算出真实成本。具体费用以模型价格和合同为准。

四、批量任务的可靠性设计

向量化通常是离线批处理,最怕跑到 80% 失败后无法续跑。建议按下面的顺序设计:

  1. 给每条文本生成内容哈希,作为幂等键,重复提交只覆盖不重复入库。
  2. 先按 Token 估算分批,同时受单请求条数、单条长度和总 Token 三个限制约束。
  3. 并发从低到高逐步加,观察 429 比例,不要一上来就开满。
  4. 对 429 和 5xx 做指数退避加随机抖动,限制最大重试次数。
  5. 每批完成后记录成功条数、Token 用量和耗时,失败样本单独落盘。
  6. 入库前校验向量维度、空向量和 NaN,异常数据隔离而不是静默写入。
  7. 全量结束后抽样比对,确认条数、维度和索引统计一致。

这套流程不复杂,但能避免“看起来跑完了、其实缺了一批”的问题。

五、限流、重试与故障切换

Embedding 接口通常按请求数和 Token 两个维度限流。触发限流时返回 429,网关过载时可能出现 5xx 或超时。重试策略要区分错误类型:429 和瞬时 5xx 可以退避重试,401、403 和参数错误重试没有意义,应先修配置。

import os
import time
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["ZSQH_API_KEY"],
    base_url=os.environ["ZSQH_BASE_URL"],
    timeout=60.0,
    max_retries=0,
)

def embed_batch(texts, model, attempts=4):
    for i in range(attempts):
        try:
            resp = client.embeddings.create(model=model, input=texts)
            vectors = [item.embedding for item in resp.data]
            dims = {len(v) for v in vectors}
            if len(dims) != 1 or len(vectors) != len(texts):
                raise ValueError("维度或条数不一致")
            return vectors, resp.usage
        except Exception as exc:
            if i == attempts - 1:
                raise
            wait = min(2 ** i, 30) + 0.5 * (i + 1)
            print(type(exc).__name__, "retry in", wait)
            time.sleep(wait)

vectors, usage = embed_batch(
    ["第一段文本", "第二段文本"],
    os.environ["EMBEDDING_MODEL"],
)
print(len(vectors), len(vectors[0]), usage)

故障切换有一个容易被忽略的边界:不同 Embedding 模型的向量不能混用。即使维度相同,向量空间也不一致,把两个模型的向量写进同一个索引,检索结果会明显变差。因此切换上游时,优先确认是否为同一模型、同一版本;确需换模型,应建立新索引、全量重算、对比效果后再切换,而不是在故障时临时改模型名。

六、上线前的核验清单

  • 维度:与索引配置一致,批量返回条数与输入一致。
  • 基线:固定一组测试文本,与直连结果比较余弦相似度,通常应非常接近。
  • 用量:响应中的 usage 与后台账单能否对上,批量请求要逐批核对。
  • 限流:并发上限、429 比例、重试后成功率是否可接受。
  • 超时:最大批次在晚高峰能否在超时时间内完成。
  • 数据:输入是否含个人信息,服务方的保留、日志和删除策略是否说得清。
  • 退出:Key 能否轮换,余额能否退款,旧索引和回退路线是否保留。

任何一项说不清,都建议先小流量试跑,不要直接把生产索引迁过去。

七、按约束选型

先看网络与付款能否长期稳定,再看是否需要固定出口与审计,最后看模型范围和账单能力。需要统一入口、人民币结算和多模型对比,可以在中商起航 AI 网关创建测试 Key,先用小批量数据跑通,再逐步放量;可选模型与计费方式见模型价格。

相关阅读:

常见问题

Embedding API 和聊天 API 的访问方案有什么不同?

聊天 API 更关注首 Token 延迟和流式稳定性;Embedding 是批量、短响应、高总量,更关注吞吐、单请求上限和失败续跑。选型时要把批量压测和断点续跑纳入方案。

国内直连 Embedding API 稳定吗?

取决于本地出口、时段和账号限额,不能一概而论。建议用真实批量和真实并发连续观察,记录错误率与 P95 延迟,再与网关方案做对照。不要仅凭一次成功请求下结论。

批量向量化中断了怎么办?

用内容哈希做幂等键,按批落盘,失败只补缺失部分。重跑前先核对已入库条数与维度,避免重复写入或把半批数据当成完整结果。

切换网关后旧向量还能用吗?

只要仍是同一模型、同一版本,通常可以继续使用;换模型则必须重建索引。迁移前用固定样本比较新旧向量的余弦相似度,并保留旧索引作为回退。

发票和用量能按项目拆分吗?

取决于服务方。签约前确认能否按项目、Key 或模型导出用量,账单口径是否包含失败和重试请求,以及发票内容能否匹配采购要求。