介绍 本文将教您如何使用任何 LLM 模型运行评估，而不会屈服于可怕的“OpenAI 速率限制”异常。我们从以下开始： 定义速率限制的含义并使用类比 了解 OpenAI 速率限制是什么 解释如何制定速率限制流程 解释代码实现 总结代码实现中使用的策略 什么是速率限制（以及类比解释）？ 到目前为止，Cloudflare 的解释是我见过的最好的：速率限制是限制网络流量的策略。它限制了某人在特定时间范围内重复操作的频率 - 例如，尝试登录帐户。 简而言之，想象一下你是四个孩子的母亲，他们都喜欢蜂蜜。上次蜂蜜用完的时间比预期的要早。现在，您已经设置了一个计时器，数到一万，并让每个孩子轮流吃一些蜂蜜。计时器代表速率限制，因为它会强制执行特定的等待时间，然后才能获得更多蜂蜜。 解释了这个概念后，让我们了解 OpenAI 速率限制并讨论如何使用 Python 实现速率限制逻辑来管理 OpenAI 的 R/TPM（每分钟请求/令牌）。 了解 OpenAI 速率限制 OpenAI 对一分钟内对其 AI 模型发出的请求数量设置了一定的限制。对于 OpenAI 提供的每个 AI 模型，这些限制都不同。 对于免费版本： 对于gpt-3.5-turbo模型，用户每天可以发出40,000个令牌请求或每分钟3个API调用。 对于 1 级： 对于 gpt-3.5-turbo 模型，每分钟允许用户 60,000 个令牌请求或 3,500 个 API 调用。 对于 gpt-4 模型，限制为每分钟 10,000 个令牌请求或 500 个 API 调用。 对于 gpt-4-turbo-preview，用户每分钟可以发出 150,000 个令牌请求或 500 个 API 调用。 对于 gpt-4-vision-preview，用户每分钟可以发出 10,000 个令牌请求和/或 500 个 API 调用。 有关其他层速率限制的更多信息，请参阅 。 文档 这些限制的原因包括： 确保服务平稳且快速响应，因为人工智能模型执行的复杂任务需要大量资源。 管理所有用户的需求，并确保可用的基础设施（例如服务器和 GPU）可以处理请求而不会过载。 为使用量激增做好准备，并在高需求下保持高效运行。 预计这些限制在可预见的未来将保持一致。 解释速率限制过程 该过程（见下图）涉及使用户能够从 UI 运行 LLM 评估并为其 LLM 应用程序配置速率限制参数，而无需自己编写逻辑。 这是通过准备和调用批处理的函数来实现的。批处理中的每个调用都会调用 函数，该函数又会使用重试机制调用最终函数 (   )。  run_with_retry invoke_app 我相信在看完上述过程后，您可以用您选择的任何语言编写代码逻辑。不管怎样，我会向你展示我是如何做到的。有关更多背景和背景，我主要在 Agenta 担任后端软件工程师。   是一个开源端到端 LLM 开发者平台，为您提供用于快速工程和管理、⚖️ 评估、人工注释和 🚀 部署的工具。所有这些都不会对您选择的框架、库或模型施加任何限制。  允许开发人员和产品团队在更短的时间内协作构建由 LLM 驱动的生产级应用程序。 Agenta Agenta 我们希望让用户能够从 UI 配置其 LLM 评估速率限制配置，以便他们可以绕过其 LLM 提供商速率限制例外。  查看流程图，首先要实现的是准备和调用批处理（LLM 调用）的逻辑。验证速率限制配置并使用数据验证模型来定义 LLM 运行速率限制非常重要。下面的模型处理批量调用运行所需的 参数。 rate_limit_config   from pydantic import BaseModel, Field class LLMRunRateLimit(BaseModel): batch_size: int = Field(default=10) max_retries: int = Field(default=3) retry_delay: int = Field(default=3) delay_between_batches: int = Field(default=5)   函数采用以下参数： batch_invoke  uri：您的 LLM 申请的 URL。  testset_data：您的LLM申请需要处理的测试集数据。 参数：您的 LLM 申请的参数。  rate_limit_config：速率限制配置（如上面创建新评估的界面所示）。   async def batch_invoke( uri: str, testset_data: List[Dict], parameters: Dict, rate_limit_config: Dict ) -> List[AppOutput]: """ Invokes the LLm app in batches, processing the testset data. Args: uri (str): The URI of the LLm app. testset_data (List[Dict]): The testset data to be processed. parameters (Dict): The parameters for the LLm app. rate_limit_config (Dict): The rate limit configuration. Returns: List[AppOutput]: The list of app outputs after running all batches. """ batch_size = rate_limit_config[ "batch_size" ] # Number of testset to make in each batch max_retries = rate_limit_config[ "max_retries" ] # Maximum number of times to retry the failed llm call retry_delay = rate_limit_config[ "retry_delay" ] # Delay before retrying the failed llm call (in seconds) delay_between_batches = rate_limit_config[ "delay_between_batches" ] # Delay between batches (in seconds) list_of_app_outputs: List[AppOutput] = [] # Outputs after running all batches openapi_parameters = await get_parameters_from_openapi(uri + "/openapi.json") async def run_batch(start_idx: int): print(f"Preparing {start_idx} batch...") end_idx = min(start_idx + batch_size, len(testset_data)) for index in range(start_idx, end_idx): try: batch_output: AppOutput = await run_with_retry( uri, testset_data[index], parameters, max_retries, retry_delay, openapi_parameters, ) list_of_app_outputs.append(batch_output) print(f"Adding outputs to batch {start_idx}") except Exception as exc: import traceback traceback.print_exc() print( f"Error processing batch[{start_idx}]:[{end_idx}] ==> {str(exc)}" ) # Schedule the next batch with a delay next_batch_start_idx = end_idx if next_batch_start_idx < len(testset_data): await asyncio.sleep(delay_between_batches) await run_batch(next_batch_start_idx) # Start the first batch await run_batch(0) return list_of_app_outputs 准备并调用批处理后，下一步涉及执行 逻辑。此自定义实现包括速率限制功能并管理 llm 应用程序的调用，在达到设置的延迟后重试。指数退避是一种以指数增加的等待时间重试操作的技术，直到达到最大重试计数为止。 run_with_retry   async def run_with_retry( uri: str, input_data: Any, parameters: Dict, max_retry_count: int, retry_delay: int, openapi_parameters: List[Dict], ) -> AppOutput: """ Runs the specified app with retry mechanism. Args: uri (str): The URI of the app. input_data (Any): The input data for the app. parameters (Dict): The parameters for the app. max_retry_count (int): The maximum number of retries. retry_delay (int): The delay between retries in seconds. openapi_parameters (List[Dict]): The OpenAPI parameters for the app. Returns: AppOutput: The output of the app. """ retries = 0 last_exception = None while retries < max_retry_count: try: result = await invoke_app(uri, input_data, parameters, openapi_parameters) return result except (httpx.TimeoutException, httpx.ConnectTimeout, httpx.ConnectError) as e: last_exception = e print(f"Error in evaluation. Retrying in {retry_delay} seconds:", e) await asyncio.sleep(retry_delay) retries += 1 # If max retries reached, return the last exception return AppOutput(output=None, status=str(last_exception))   的使用：即使在已用完最大重试次数后处理异常也很重要。这样，您可以允许您尝试处理的所有数据运行，然后您可以确定哪些数据失败了，哪些数据通过了。 AppOutput 最后一步是调用应用程序，使用 LLM 应用程序的 来确定如何使用单个数据点调用它。 openapi_parameters  make_payload 函数不应该让您担心。它根据其 参数构造用于调用 LLM 应用程序的有效负载。 OpenAPI   async def invoke_app( uri: str, datapoint: Any, parameters: Dict, openapi_parameters: List[Dict] ) -> AppOutput: """ Invokes an app for one datapoint using the openapi_parameters to determine how to invoke the app. Args: uri (str): The URI of the app to invoke. datapoint (Any): The data to be sent to the app. parameters (Dict): The parameters required by the app taken from the db. openapi_parameters (List[Dict]): The OpenAPI parameters of the app. Returns: AppOutput: The output of the app. Raises: httpx.HTTPError: If the POST request fails. """ url = f"{uri}/generate" payload = await make_payload(datapoint, parameters, openapi_parameters) async with httpx.AsyncClient() as client: try: logger.debug(f"Invoking app {uri} with payload {payload}") response = await client.post( url, json=payload, timeout=httpx.Timeout(timeout=5, read=None, write=5) ) response.raise_for_status() llm_app_response = response.json() app_output = ( llm_app_response["message"] if isinstance(llm_app_response, dict) else llm_app_response ) return AppOutput(output=app_output, status="success") except: return AppOutput(output="Error", status="error") 整个过程就这样结束了。 概括 代码中的退避指数策略的工作原理如下：  batch_invoke 函数将测试集数据拆分为具有可配置大小的较小批次。每个批次均按顺序处理。 批处理： 在每个批次中，每个数据点都由 函数处理。此函数尝试调用数据点的应用程序。如果由于特定网络错误（超时、连接问题）导致调用失败，该函数会延迟重试。此延迟最初设置为可配置值 (   )，并针对同一批次中的每次后续重试尝试加倍。 单独调用重试： run_with_retry retry_delay 此方法有助于避免失败后重复请求导致应用程序服务器过载。它为服务器提供了恢复时间，并允许在重试之前清除待处理请求的队列。 该策略还包括每个数据点的可配置最大重试次数，以防止无限循环。还包括批次之间的延迟 (   )，以避免超出应用程序服务器设置的速率限制。 delay_between_batches 我希望这总结了您在今天的文章中学到的所有内容。请让我知道，如果你有任何问题！

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

Available for Hire: Part-time / Freelance Technical Writer

2021 - HackerNoon Contributor of the Year - API

I now write about my personal lens on life, experience, and the/my ever-evolving self.

Backend Software Engineer

該音頻是用故事的原始語言製作的！

OpenAI 的速率限制：LLM 评估指数退避指南

About Author

註釋

標籤

这篇文章刊登在

Related Stories

Claude Sonnet 3.5 系统提示泄漏：法医分析

使用这 18 种开发工具来提高你的工作效率 🚀🔥

Floki 的 Valhalla 成为印度环斯里兰卡赛事联合赞助商

扬帆起航：利用数据湖开发生产级 RAG 应用程序

Claude Sonnet 3.5 系统提示泄漏：法医分析

使用这 18 种开发工具来提高你的工作效率 🚀🔥

Floki 的 Valhalla 成为印度环斯里兰卡赛事联合赞助商

扬帆起航：利用数据湖开发生产级 RAG 应用程序

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps