PAYATHON 2026

Weaviate 使用 OpenAI ada 批量导入时触发 429 限流

AI 小顾

结论

这是 OpenAI Embeddings API 的请求频率限制，与 JSON、Pandas 或 Weaviate schema 本身无关。当前额度只有 3 requests/min，而 Weaviate 导入每个对象时都要调用 OpenAI 生成向量。批量接口只负责在客户端批量提交对象，不能绕过 OpenAI 的限流。

可以这样处理：

  1. 为 OpenAI API 账户配置计费，并申请更高额度。这是最直接的办法。
  2. 如果额度仍是 3 RPM，将导入速度控制在每 21～25 秒一个对象。
  3. 遇到 429 时执行指数退避，并按照 Retry-After 指定的时间等待后重试。
  4. 使用确定性 UUID，让重试可以安全执行，避免生成重复数据。
  5. 记录导入失败的对象，等导入结束后单独补偿。
  6. 只向量化需要参与语义搜索的字段，通常是 title 和 description。
  7. 将检索和生成分开处理。ada 只负责向量化，最终回答仍需要聊天或生成模型。

5000 条数据在 3 RPM 下的理论最短耗时约为：

[ 5000 \div 3 = 1666.7\text{ 分钟}\approx27.8\text{ 小时} ]

实际导入还会受到网络延迟、失败重试和查询向量化的影响，通常需要更长时间。

调整 schema

skip 应使用布尔值，不能写成字符串 "false"。author 和 url 通常不需要参与语义向量，可以直接跳过。否则，URL 和作者名等内容可能会稀释 description 的语义。

下面继续沿用问题中的旧版 Weaviate schema 格式：

schema = {
    "classes": [
        {
            "class": "PersonalLibrary",
            "description": "Information from Library Books",
            "vectorizer": "text2vec-openai",
            "moduleConfig": {
                "text2vec-openai": {
                    "model": "ada",
                    "type": "text",
                }
            },
            "properties": [
                {
                    "name": "description",
                    "dataType": ["text"],
                    "description": "The information",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": False,
                            "vectorizePropertyName": False,
                        }
                    },
                },
                {
                    "name": "title",
                    "dataType": ["text"],
                    "description": "The title",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": False,
                            "vectorizePropertyName": False,
                        }
                    },
                },
                {
                    "name": "author",
                    "dataType": ["text"],
                    "description": "The author",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": True,
                            "vectorizePropertyName": False,
                        }
                    },
                },
                {
                    "name": "url",
                    "dataType": ["text"],
                    "description": "The url",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": True,
                            "vectorizePropertyName": False,
                        }
                    },
                },
                {
                    "name": "path",
                    "dataType": ["text"],
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": True,
                            "vectorizePropertyName": False,
                        }
                    },
                },
            ],
        }
    ]
}

ada 和 text-embedding-ada-002 都属于较早的模型配置。新建系统时，应根据当前 Weaviate 和 OpenAI API 版本选择受支持的 embedding 模型。Weaviate Python Client v3 和 v4 的 schema 写法也不同，不能直接混用。

在 3 RPM 下进行节流导入

限额只有 3 RPM 时，主动控制为一次导入一个对象，比一次提交大量对象更容易管理。下面的示例适用于问题中使用的旧版客户端调用方式。

import time
import random
import pandas as pd
import weaviate
from weaviate.util import generate_uuid5

CLASS_NAME = "PersonalLibrary"
MIN_INTERVAL_SECONDS = 21
MAX_RETRIES = 8

df = pd.read_json("dummy4.json")

last_request_at = 0.0
failed_rows = []

for index, row in df.iterrows():
    properties = {
        "title": str(row.get("title", "")),
        "description": str(row.get("description", "")),
        "author": str(row.get("author", "")),
        "url": str(row.get("url", "")),
        "path": "https://books.mypage.com/explore/" + str(row["id"]),
    }

    # 同一本书始终生成相同 UUID，重试时不会不断创建重复对象。
    object_uuid = generate_uuid5({
        "class": CLASS_NAME,
        "source_id": str(row["id"]),
    })

    imported = False

    for attempt in range(MAX_RETRIES):
        elapsed = time.monotonic() - last_request_at
        if elapsed < MIN_INTERVAL_SECONDS:
            time.sleep(MIN_INTERVAL_SECONDS - elapsed)

        try:
            last_request_at = time.monotonic()

            client.data_object.create(
                data_object=properties,
                class_name=CLASS_NAME,
                uuid=object_uuid,
            )

            imported = True
            print(f"Imported row={index}, id={row['id']}")
            break

        except Exception as exc:
            message = str(exc)
            is_rate_limit = (
                "429" in message
                or "rate limit" in message.lower()
            )

            if not is_rate_limit:
                print(f"Non-retryable error at row={index}: {message}")
                break

            # 21、42、84……秒，并加入少量随机抖动。
            delay = min(
                MIN_INTERVAL_SECONDS * (2 ** attempt),
                300,
            )
            delay += random.uniform(0, 3)

            print(
                f"Rate limited at row={index}; "
                f"retry={attempt + 1}/{MAX_RETRIES}; "
                f"sleep={delay:.1f}s"
            )
            time.sleep(delay)

    if not imported:
        failed_rows.append({
            "index": int(index),
            "id": str(row["id"]),
            "properties": properties,
            "uuid": str(object_uuid),
        })

print(f"Finished. Failed objects: {len(failed_rows)}")

这里将间隔设为 21 秒，而不是精确的 20 秒，是为了给滑动时间窗口、网络延迟和时钟误差留出余量。如果导入期间还有其他程序使用同一个 OpenAI organization 和 API key，还需要进一步降低速度。

生产环境中不建议通过 except Exception 判断错误类型。应根据已安装的 Weaviate 客户端版本捕获具体异常，再从响应状态码或错误结构中识别 429。上面的宽泛捕获只是为了说明节流和重试流程。

使用 batch 时的处理方式

提高 OpenAI 限额后，可以继续使用 Weaviate batch，但要注意以下几点：

  • batch 大小和并发数不能超过下游 OpenAI 的承受能力。
  • 不能只看 Python 调用是否抛出异常，还要检查每个对象的批量响应。
  • 如果响应中包含 429、rate limit 或向量化失败，应将对应对象加入重试队列。
  • 重试时继续使用原来的 UUID。
  • 不要无限重试。超过重试上限后，应保存失败清单。

在 3 RPM 下，即使设置 batch_size=3，也不能马上连续提交下一批。每批之后至少要等待约 60 秒，滚动窗口仍可能使处于边界位置的请求收到 429。因此，一次导入一个对象，并间隔约 21 秒，通常更容易控制。

更高效的替代方案：自行生成向量

还可以由应用自行生成向量：

  1. 应用直接调用 OpenAI Embeddings API。
  2. 在应用侧统一处理限流、批处理和重试。
  3. 取得向量后，将对象和向量一起写入 Weaviate。
  4. Weaviate class 使用自带向量模式，避免导入时再次调用 OpenAI。

示意代码如下。具体参数需要根据已安装的 OpenAI SDK 和 Weaviate 客户端版本调整：

text_to_embed = "\n".join([
    str(row.get("title", "")),
    str(row.get("description", "")),
])

embedding_response = openai_client.embeddings.create(
    model="YOUR_SUPPORTED_EMBEDDING_MODEL",
    input=text_to_embed,
)

vector = embedding_response.data[0].embedding

client.data_object.create(
    data_object=properties,
    class_name="PersonalLibrary",
    uuid=object_uuid,
    vector=vector,
)

采用这种架构后，应用可以完全掌控限流和失败重试，也可以保存已经生成的向量，避免 Weaviate 写入失败后重复消耗 embedding 请求。

不过，仅凭上述错误无法判断 3 RPM 是只限制请求数，还是同时存在 token 限制，具体要以账户实际返回的限额为准。部分 Embeddings API 和 SDK 支持在一次请求中传入多个字符串，但是否适用于当前账户、模型和 Weaviate 集成，也不能仅凭该错误确定。如果由应用自行调用 API，可以在模型和限额允许的情况下，将多条文本组成一个 input 数组，以减少请求次数。

导入后的检索与回答流程

text2vec-openai 或 embedding 模型只负责生成向量，不会自动给出最终答案。完整流程如下：

用户问题
  → 生成问题向量
  → 在 Weaviate 中检索最相关的图书 description
  → 将检索结果作为上下文发送给生成模型
  → 生成有依据的自然语言回答

例如，先执行语义检索：

{
  Get {
    PersonalLibrary(
      nearText: {
        concepts: ["用户的问题"]
      }
      limit: 5
    ) {
      title
      author
      description
      url
      path
      _additional {
        distance
      }
    }
  }
}

然后将结果整理成提示词：

context = "\n\n".join(
    f"书名：{item['title']}\n"
    f"作者：{item['author']}\n"
    f"内容：{item['description']}"
    for item in search_results
)

prompt = f"""
请只根据下面提供的图书资料回答问题。
如果资料不足以得出答案，请明确说明资料不足，不要猜测。

资料：
{context}

问题：
{user_question}
"""

最后，将提示词发送给选定的聊天或生成模型。也可以使用与当前 Weaviate 版本兼容的生成模块完成类似流程，但检索数量、上下文长度、引用方式和无答案策略仍需由应用明确控制。

注意事项

  • 批量导入不等于只发出一次 OpenAI 请求。限流发生在 Weaviate 调用向量化服务的阶段。
  • API key 所属 organization 的总流量是共享的，其他服务发出的请求也会占用 3 RPM。
  • 除 RPM 外，账户还可能受到 TPM、每日额度或余额限制。
  • 不要对所有错误无限重试。429、临时网络错误和部分 5xx 可以重试；schema 错误和字段类型错误等问题应直接修复。
  • 应提前清洗 NaN、空 description 和异常长文本，避免生成无意义的向量或触发 token 限制。
  • 导入完成后，要核对 Weaviate 中的对象总数，并检查对象是否包含向量。不能仅凭脚本运行结束判断导入成功。
  • 使用确定性 UUID 后，补偿导入可以安全地改为 create-or-update/upsert，具体方法名取决于客户端版本。
  • 5000 条数据在 3 RPM 下可以导入，但会耗费很长时间。如果这是正式数据集，提高 API 限额或自行批量生成向量通常更实际。

备注：内容仅供参考。