如何处理模型速率限制

运行大型评估任务时，一个常见问题是遇到第三方 API 速率限制，通常来自模型提供商。有几种方法可以处理速率限制。

使用 langchain RateLimiter（仅 Python）

如果你在应用或评估器中使用 langchain Python 聊天模型，可以为模型添加速率限制器，在客户端控制向模型提供商 API 发送请求的频率，以避免速率限制错误。

from langchain.chat_models import init_chat_model
from langchain.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,  # <-- Super slow! We can only make a request once every 10 seconds!!
    check_every_n_seconds=0.1,  # Wake up every 100 ms to check whether allowed to make a request,
    max_bucket_size=10,  # Controls the maximum burst size.
)

model = init_chat_model("gpt-5.5", rate_limiter=rate_limiter)

def app(inputs: dict) -> dict:
    response = model.invoke(...)
    ...

def evaluator(inputs: dict, outputs: dict, reference_outputs: dict) -> dict:
    response = model.invoke(...)
    ...

有关如何配置速率限制器的更多信息，请参阅 langchain 文档。

使用指数退避重试

处理速率限制错误的一种非常常见的方法是使用指数退避重试。指数退避重试是指对失败的请求反复重试，并在每次重试之间以（指数级）增加的等待时间间隔。这一过程会持续到请求成功或达到最大请求次数为止。

使用 langchain

如果你使用 langchain 组件，可以通过 .with_retry(...) / .withRetry() 方法为所有模型调用添加重试：

Python

from langchain import init_chat_model

model_with_retry = init_chat_model("gpt-5.4-mini").with_retry(stop_after_attempt=6)

TypeScript

import { initChatModel } from "langchain";

const model = await initChatModel("gpt-5.5", {
    modelProvider: "openai",
});

const modelWithRetry = model.withRetry({ stopAfterAttept: 2 });

更多信息请参阅 langchain 的 Python 和 JS API 参考。

不使用 langchain

如果你不使用 langchain，可以使用 tenacity（Python）或 backoff（Python）等其他库来实现指数退避重试，也可以从头实现。有关如何实现的示例，请参阅 OpenAI 文档。

限制 max_concurrency

限制对应用和评估器的并发调用数量，是降低模型调用频率、从而避免速率限制错误的另一种方法。max_concurrency 可以直接在 evaluate() / aevaluate() 函数上设置。这会通过将数据集有效拆分到多个线程来实现并行评估。

Python

from langsmith import aevaluate

results = await aevaluate(
    ...
    max_concurrency=4,
)

TypeScript

import { evaluate } from "langsmith/evaluation";

await evaluate(..., {
  ...,
  maxConcurrency: 4,
});

---