Introdução Este artigo ensinará como executar avaliações usando qualquer modelo LLM sem sucumbir à temida exceção “OpenAI Rate Limit”. Começaríamos por: Definindo o que significa limite de taxa e usando uma analogia Entendendo o que são limites de taxa OpenAI Explicando como o Processo Rate-Limit foi desenvolvido Explicando a implementação do código Resumindo a estratégia utilizada na Implementação do Código O que é limitação de taxa (e explicação de analogia)? Até agora, a explicação da Cloudflare é a melhor que já vi: a limitação de taxa é uma estratégia para limitar o tráfego de rede. Ele limita a frequência com que alguém pode repetir uma ação dentro de um determinado período de tempo – por exemplo, tentar fazer login em uma conta. Simplificando, imagine ser mãe de quatro filhos que amam mel. Da última vez, o mel acabou antes do esperado. Agora, você configurou um cronômetro para contar até dez mil e deu a cada criança a oportunidade de comer um pouco de mel. O cronômetro representa o limite de taxa, pois impõe um tempo de espera específico antes que eles possam ter mais mel. Depois de explicar o conceito, vamos entender os limites de taxa do OpenAI e discutir como implementei uma lógica de limite de taxa para gerenciar o R/TPM (solicitação/token por minuto) do OpenAI usando Python. Compreendendo os limites de taxa OpenAI A OpenAI estabeleceu certas restrições sobre o número de solicitações que podem ser feitas para seus modelos de IA em um minuto. Estas limitações são diferentes para cada modelo de IA fornecido pela OpenAI. Para a versão gratuita: Para o modelo gpt-3.5-turbo, os usuários podem fazer 40.000 solicitações de token por dia ou 3 chamadas de API por minuto. Para o nível 1: Para o modelo gpt-3.5-turbo, os usuários têm permissão para 60.000 solicitações de token ou 3.500 chamadas de API por minuto. Para o modelo gpt-4, o limite é de 10.000 solicitações de token ou 500 chamadas de API por minuto. Para o gpt-4-turbo-preview, os usuários podem fazer 150.000 solicitações de token ou 500 chamadas de API por minuto. Para o gpt-4-vision-preview, os usuários podem fazer 10.000 solicitações de token e/ou 500 chamadas de API por minuto. Consulte a para obter mais informações sobre limites de taxa de outros níveis. documentação A razão para essas restrições inclui: Garantir que os serviços funcionem de forma harmoniosa e responsiva, uma vez que tarefas complexas como as executadas pelos modelos de IA requerem recursos substanciais. Gerenciar a demanda de todos os usuários e garantir que a infraestrutura disponível, como servidores e GPUs, possa atender às solicitações sem ficar sobrecarregada. Preparação para picos de uso e manutenção de operação eficiente sob alta demanda. Espera-se que essas limitações permaneçam consistentes no futuro próximo. Explicando o processo de limite de taxa O processo (veja a imagem abaixo) envolve permitir que os usuários executem avaliações LLM a partir da UI e configurem parâmetros de limite de taxa para seus aplicativos LLM sem a necessidade de escrever a lógica por conta própria. Isto é conseguido através de uma função que prepara e invoca o lote. Cada chamada no lote invoca a função , que por sua vez invoca a função final ( ) com o mecanismo de nova tentativa. run_with_retry invoke_app Tenho certeza de que você pode escrever a lógica do código em qualquer linguagem de sua escolha depois de dar uma olhada no processo acima. Independentemente disso, vou mostrar como fiz o meu. Para obter mais informações e contexto, trabalho principalmente como engenheiro de software back-end na Agenta. é uma plataforma de desenvolvedor LLM de código aberto ponta a ponta que fornece as ferramentas para engenharia e gerenciamento imediatos, ⚖️ avaliação, anotação humana e 🚀 implantação. Tudo sem impor quaisquer restrições à sua escolha de estrutura, biblioteca ou modelo. permite que desenvolvedores e equipes de produto colaborem na construção de aplicativos LLM de nível de produção em menos tempo. Agenta Agenta Queríamos dar aos usuários a capacidade de definir a configuração de limitação de taxa de suas avaliações LLM na interface do usuário para que possam ignorar a exceção de limitação de taxa do provedor LLM. Observando o diagrama do processo, a primeira coisa a implementar é a lógica de preparação e invocação do lote (de chamadas LLM). É importante validar a configuração do limite de taxa e usar um modelo de validação de dados para definir o limite de taxa de execução do LLM. O modelo abaixo lida com o parâmetro necessário para que a chamada em lote funcione. 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) A função usa os seguintes parâmetros: batch_invoke uri: O URL da sua aplicação LLM. testset_data: os dados do conjunto de testes que seu aplicativo LLM precisa processar. parâmetros: os parâmetros para seu aplicativo LLM. rate_limit_config: A configuração do limite de taxa (conforme visto na interface acima para criar uma nova avaliação). 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 Depois de preparar e invocar o lote, a próxima etapa envolve a execução da lógica . Essa implementação personalizada inclui funcionalidade de limitação de taxa e gerencia a invocação do aplicativo llm, tentando novamente após atingir o atraso definido. A espera exponencial, uma técnica que tenta novamente uma operação com um tempo de espera crescente exponencialmente, é empregada até que uma contagem máxima de novas tentativas seja atingida. 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)) O uso de : é importante lidar com uma exceção mesmo depois de ela ter esgotado seu máximo de tentativas. Dessa forma, você permite que todos os dados que está tentando processar sejam executados e, então, pode determinar o que falhou e o que foi aprovado. AppOutput A etapa final é invocar o aplicativo, usando do aplicativo LLM para determinar como invocá-lo com um único ponto de dados. openapi_parameters A função make_payload não deve preocupar você. Ele constrói a carga para invocar o aplicativo LLM com base em seus parâmetros . 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") E isso completa o processo. Resumo A estratégia exponencial de backoff no código funciona da seguinte maneira: a função batch_invoke divide os dados do conjunto de testes em lotes menores com tamanho configurável. Cada lote é processado sequencialmente. Processamento em lote: dentro de cada lote, cada ponto de dados é processado pela função . Esta função tenta invocar o aplicativo para o ponto de dados. Se a invocação falhar devido a erros de rede específicos (tempo limite, problemas de conexão), a função tentará novamente com atraso. Esse atraso é inicialmente definido para um valor configurável ( ) e é duplicado para cada nova tentativa subsequente dentro do mesmo lote. Invocações individuais com nova tentativa: run_with_retry retry_delay Essa abordagem ajuda a evitar sobrecarregar o servidor de aplicativos com solicitações repetidas após uma falha. Dá ao servidor tempo para se recuperar e permite que a fila de solicitações pendentes seja limpa antes de tentar novamente. A estratégia também inclui um número máximo configurável de novas tentativas por ponto de dados para evitar loops infinitos. Um atraso entre lotes ( ) também está incluído para evitar exceder os limites de taxa definidos pelo servidor de aplicativos. delay_between_batches Espero que isso resuma tudo o que você aprendeu no artigo de hoje. Por favor, deixe-me saber se você tem alguma dúvida!
