运行大型评估任务时的一个常见问题是遇到第三方 API 速率限制,这通常来自模型提供商。有几种方法可以处理速率限制。
使用 langchain RateLimiter(仅 Python)
如果您使用的是 langchain Python 聊天模型,您可以为模型添加速率限制器,以添加客户端控制,从而控制向模型提供商 API 发送请求的频率,避免速率限制错误。
from langchain.chat_models import init_chat_model
from langchain_core.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() method:
from langchain import init_chat_model
model_with_retry = init_chat_model("gpt-5.4-mini").with_retry(stop_after_attempt=6)
const model = await initChatModel("gpt-5.5", {
modelProvider: "openai",
});
const modelWithRetry = model.withRetry({ stopAfterAttept: 2 });
请参阅 langchain Python and JS API 参考文档了解更多。
如果没有 langchain
如果您没有使用 langchain 您可以使用其他库,例如 tenacity (Python) 或 backoff (Python) 实现指数退避重试,或者您可以从头开始实现。请参阅其中的一些示例 OpenAI 文档.
限制 max_concurrency
限制您对应用程序和评估器的并发调用次数是另一种减少模型调用频率的方式,从而避免速率限制错误。 max_concurrency 可以直接在 evaluate() / aevaluate() 函数上设置。通过有效地将数据集分割到各个线程中来实现并行评估。
from langsmith import aevaluate
results = await aevaluate(
...
max_concurrency=4,
)
await evaluate(..., {
...,
maxConcurrency: 4,
});