OpenAI Embeddings 怎样按批量输入组织请求

当待向量化的文本从几十条增长到几千条时，最稳妥的做法不是把它们拼成一段超长字符串，而是把多条字符串作为同一个 Embeddings 请求的 input 数组，并在本地控制每批的数量和 token 预算。这样每个向量仍然对应一条原始记录，某一批失败时也只需要处理这一批。

官方地址：https://developers.openai.com/api/docs/guides/embeddings

多条文本应该怎样进入同一个 Embeddings 请求

OpenAI Embeddings 的 input 可以是一个字符串，也可以是字符串数组。数组中的每个元素代表一条独立文本，返回结果会在 data 中给出对应的 embedding 项。这个模型更适合“多条记录分别向量化”，而不是把标题、正文和其他记录拼在一起后失去记录边界。

官方 API reference 给出了两个需要同时考虑的输入边界：所有 embedding 模型单条输入最多 8192 tokens；单次请求中所有输入合计最多 300000 tokens。下面的 64 条只是一个便于控制失败范围的起始批次，不是硬性规定。短文本可以适当增大，长文本或限流敏感的任务则应该减小。

先按可控数量切出批次

先把空字符串过滤掉，再用固定的 max_items 切片。这里不在示例中假装用字符数精确代替 token 数；如果文本长度差异很大，应在切批前接入与模型匹配的 tokenizer，或者降低单批数量，给合计 token 留出余量。

from openai import OpenAI

client = OpenAI()


def split_batches(texts: list[str], max_items: int = 64):
    """把非空文本切成小批次，控制一次失败影响的记录范围。"""
    # 过滤空白输入，避免把无意义文本发送到接口。
    cleaned = [text.strip() for text in texts if text and text.strip()]
    for start in range(0, len(cleaned), max_items):
        # max_items 是本地工程策略，不代表接口的固定批量上限。
        yield cleaned[start:start + max_items]


def embed_texts(texts: list[str]) -> list[list[float]]:
    """按批次提交文本，并按返回 index 放回连续结果列表。"""
    results: list[list[float] | None] = [None] * len(
        [text for text in texts if text and text.strip()]
    )
    offset = 0

    for batch in split_batches(texts):
        try:
            response = client.embeddings.create(
                model="text-embedding-3-small",
                input=batch,
                encoding_format="float",
            )
            # 返回 index 是批次内位置，必须加上本批在全量结果中的偏移量。
            for item in response.data:
                results[offset + item.index] = item.embedding
            offset += len(batch)
        except Exception:
            # 真实任务中记录 batch 起点并重试当前批次，不要悄悄跳过。
            raise

    # 让调用方尽早发现批次回填不完整，而不是把 None 写入向量库。
    if any(vector is None for vector in results):
        raise RuntimeError("embedding result is incomplete")
    return [vector for vector in results if vector is not None]

这段代码有两个容易忽略的点。第一，空文本过滤后，结果数组的长度必须以清洗后的文本为准。第二，item.index 是当前批次内的索引，所以要加上 offset 才能定位到全量结果中的位置。生产环境可以把当前批次的起点、请求参数和异常类型写入任务日志，重试时只重试该批次。

为什么要用返回的 index 还原记录顺序

向量不是脱离业务记录独立存在的。写入数据库或向量存储时，通常还需要保留文章 ID、商品 ID或文档版本。最简单的安全做法是为每条输入保留一个业务主键，并用批次内位置建立临时映射；收到响应后，以 data.index 找回原记录，而不是直接假定返回数组永远按照你的输入顺序排列。

失败批次和 dimensions 怎么处理

批量请求的好处是减少请求次数，代价是单次失败会影响一组记录。因此建议把每批当成独立任务：保存批次起止位置，成功后再提交下游；失败时只重试当前批次，并设置有限次数与退避。不要在异常分支里继续把未知长度的结果写入向量存储。

text-embedding-3-small 默认返回 1536 维，text-embedding-3-large 默认返回 3072 维；官方指南说明，dimensions 可以用于 text-embedding-3 及更新模型，以换取更小的存储与计算开销。这个参数一旦用于生产索引，就应该和向量库的 schema、相似度计算及后续查询保持一致，不能只在写入时临时改变。

def build_request(batch: list[str], dimensions: int | None = None) -> dict:
    """构造一个可记录、可重试的请求参数。"""
    # 请求前再次检查空批次，避免把无效任务交给远端接口。
    if not batch:
        raise ValueError("batch must not be empty")

    payload = {
        "model": "text-embedding-3-small",
        "input": batch,
        "encoding_format": "float",
    }
    if dimensions is not None:
        # dimensions 只应在选定模型和下游 schema 都支持时设置。
        payload["dimensions"] = dimensions
    return payload

常见问题

把多条文本拼成一个字符串和传入字符串数组有什么区别？

拼成一个字符串只会得到一个输入对应的向量；传入字符串数组则让每个元素保持独立记录边界，并在响应中得到多个 embedding 项。

批次越大越省请求吗？

通常会减少请求次数，但会增加单次失败影响范围，也更容易触碰总 token、限流或超时边界。应按文本长度和重试成本选择，而不是只追求更大的数组。

能不能只按字符串长度判断 token？

不能把字符数当成精确 token 数。文本差异明显时应使用匹配模型的 tokenizer，或降低批次并为合计 token 预留余量。

dimensions 设小后还能和旧向量混用吗？

不要默认可以。向量长度和模型配置应作为索引契约的一部分；切换维度前先规划重建、分区或版本化策略。

批量 Embeddings 的核心不是把数组做大，而是保持三条边界：输入记录不混淆、每批额度可控、返回索引能回到业务主键。沿着这三条边界组织代码，后续无论接入检索、聚类还是推荐，都更容易定位失败和迁移模型。