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

使用 langchain 速率限制器(仅限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,  # <-- 非常慢!我们每10秒只能发送一次请求!!
    check_every_n_seconds=0.1,  # 每100毫秒唤醒一次以检查是否允许发送请求,
    max_bucket_size=10,  # 控制最大突发大小。
)

model = init_chat_model("gpt-5.4", 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() 方法为所有模型调用添加重试:
from langchain import init_chat_model

model_with_retry = init_chat_model("gpt-5.4-mini").with_retry(stop_after_attempt=6)
有关更多信息,请参阅 langchain PythonJS API 参考。

不使用 langchain

如果您不使用 langchain,可以使用其他库(如 tenacity(Python)或 backoff(Python))来实现指数退避重试,或者您可以从头开始实现。有关如何操作的示例,请参阅 OpenAI 文档

限制 max_concurrency

限制您对应用程序和评估器进行的并发调用数量是减少模型调用频率的另一种方式,从而避免速率限制错误。max_concurrency 可以直接在 evaluate() / aevaluate() 函数上设置。这通过有效地将数据集分配到线程中来并行化评估。
from langsmith import aevaluate

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