Einführung In diesem Artikel erfahren Sie, wie Sie Auswertungen mit einem beliebigen LLM-Modell durchführen können, ohne der gefürchteten „OpenAI Rate Limit“-Ausnahme zu unterliegen. Wir würden beginnen mit: Definieren, was Ratenbegrenzung bedeutet, und Verwendung einer Analogie Verstehen, was OpenAI-Ratenlimits sind Erklären, wie der Rate-Limit-Prozess entwickelt wurde Erläutern der Code-Implementierung Zusammenfassung der bei der Code-Implementierung verwendeten Strategie Was ist Ratenbegrenzung (und Erklärung der Analogie)? Bisher ist die Erklärung von Cloudflare die beste, die ich je gesehen habe: Rate-Limiting ist eine Strategie zur Begrenzung des Netzwerkverkehrs. Es gibt eine Obergrenze dafür, wie oft jemand eine Aktion innerhalb eines bestimmten Zeitraums wiederholen kann – beispielsweise den Versuch, sich bei einem Konto anzumelden. Um es einfach auszudrücken: Stellen Sie sich vor, Sie wären Mutter von vier Kindern, die alle Honig lieben. Beim letzten Mal ging der Honig früher als erwartet aus. Jetzt haben Sie einen Timer eingestellt, der bis zu zehntausend zählt, und jedem Kind die Möglichkeit gegeben, etwas Honig zu essen. Der Timer stellt die Mengenbegrenzung dar, da er eine bestimmte Wartezeit vorschreibt, bevor sie mehr Honig haben können. Nachdem wir das Konzept erklärt haben, wollen wir uns mit den Ratenbegrenzungen von OpenAI befassen und besprechen, wie ich eine Ratenbegrenzungslogik implementiert habe, um OpenAIs R/TPM (Anfrage/Token pro Minute) mit Python zu verwalten. Grundlegendes zu OpenAI-Ratenbeschränkungen OpenAI hat bestimmte Beschränkungen für die Anzahl der Anfragen festgelegt, die man innerhalb einer Minute an seine KI-Modelle stellen kann. Diese Einschränkungen sind für jedes von OpenAI bereitgestellte KI-Modell unterschiedlich. Für die kostenlose Version: Für das gpt-3.5-turbo-Modell können Benutzer 40.000 Token-Anfragen pro Tag oder 3 API-Aufrufe pro Minute stellen. Für die Stufe 1: Für das Modell gpt-3.5-turbo sind Benutzern 60.000 Token-Anfragen oder 3.500 API-Aufrufe pro Minute gestattet. Für das gpt-4-Modell liegt die Grenze bei 10.000 Token-Anfragen oder 500 API-Aufrufen pro Minute. Für die gpt-4-turbo-preview können Benutzer 150.000 Token-Anfragen oder 500 API-Aufrufe pro Minute durchführen. Für die gpt-4-vision-preview können Benutzer 10.000 Token-Anfragen und/oder 500 API-Aufrufe pro Minute durchführen. Weitere Informationen zu den Ratenbegrenzungen anderer Stufen finden Sie in den . Dokumenten Der Grund für diese Einschränkungen ist unter anderem: Sicherstellen, dass die Dienste reibungslos und reaktionsschnell funktionieren, da komplexe Aufgaben, wie sie von den KI-Modellen ausgeführt werden, erhebliche Ressourcen erfordern. Verwalten Sie die Nachfrage aller Benutzer und stellen Sie sicher, dass die verfügbare Infrastruktur, wie z. B. ihre Server und GPUs, die Anfragen verarbeiten kann, ohne überlastet zu werden. Vorbereitung auf Nutzungsspitzen und Aufrechterhaltung eines effizienten Betriebs bei hoher Nachfrage. Es wird erwartet, dass diese Einschränkungen auf absehbare Zeit bestehen bleiben. Erklären des Rate-Limit-Prozesses Der Prozess (siehe Abbildung unten) beinhaltet, dass Benutzer LLM-Bewertungen über die Benutzeroberfläche ausführen und Ratenbegrenzungsparameter für ihre LLM-Apps konfigurieren können, ohne die Logik selbst schreiben zu müssen. Dies wird durch eine Funktion erreicht, die den Stapel vorbereitet und aufruft. Jeder Aufruf im Batch ruft die Funktion auf, die wiederum die letzte Funktion ( ) mit dem Wiederholungsmechanismus aufruft. run_with_retry invoke_app Ich bin zuversichtlich, dass Sie die Code-Logik in jeder Sprache Ihrer Wahl schreiben können, nachdem Sie sich den obigen Prozess angesehen haben. Egal, ich zeige euch, wie ich es gemacht habe. Für mehr Hintergrund und Kontext: Ich arbeite hauptsächlich als Backend-Softwareentwickler bei Agenta. ist eine Open-Source-End-to-End-LLM-Entwicklerplattform, die Ihnen die Tools für schnelles Engineering und Management, ⚖️ Evaluierung, menschliche Annotation und 🚀 Bereitstellung bietet. Und das alles ohne Einschränkungen bei der Wahl des Frameworks, der Bibliothek oder des Modells. ermöglicht Entwicklern und Produktteams die Zusammenarbeit bei der Erstellung produktionstauglicher LLM-basierter Anwendungen in kürzerer Zeit. Agenta Agenta Wir wollten Benutzern die Möglichkeit geben, die Ratenbegrenzungskonfiguration ihrer LLM-Bewertungen über die Benutzeroberfläche zu konfigurieren, damit sie die Ratenbegrenzungsausnahme ihres LLM-Anbieters umgehen können. Betrachtet man das Prozessdiagramm, so muss zunächst die Logik zum Vorbereiten und Aufrufen des Stapels (von LLM-Aufrufen) implementiert werden. Es ist wichtig, die Ratenbegrenzungskonfiguration zu validieren und ein Datenvalidierungsmodell zur Definition der LLM-Laufratenbegrenzung zu verwenden. Das folgende Modell verarbeitet den Parameter , der für die Funktion des Batch-Aufrufs erforderlich ist. 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) Die Funktion akzeptiert die folgenden Parameter: batch_invoke uri: Die URL Ihrer LLM-Anwendung. testset_data: Die Testset-Daten, die Ihre LLM-Anwendung verarbeiten muss. Parameter: Die Parameter für Ihre LLM-Anwendung. rate_limit_config: Die Ratenbegrenzungskonfiguration (wie in der obigen Schnittstelle zum Erstellen einer neuen Auswertung zu sehen). 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 Nach der Vorbereitung und dem Aufruf des Batches umfasst der nächste Schritt die Ausführung der Logik. Diese benutzerdefinierte Implementierung umfasst Funktionen zur Ratenbegrenzung und verwaltet den Aufruf der llm-App, indem sie es erneut versucht, nachdem die festgelegte Verzögerung erreicht ist. Exponentielles Backoff, eine Technik, die einen Vorgang mit einer exponentiell zunehmenden Wartezeit wiederholt, wird verwendet, bis eine maximale Anzahl von Wiederholungen erreicht ist. 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)) Die Verwendung von : Es ist wichtig, eine Ausnahme auch dann zu behandeln, wenn die maximale Anzahl an Wiederholungsversuchen aufgebraucht ist. Auf diese Weise ermöglichen Sie die Ausführung aller Daten, die Sie verarbeiten möchten, und können dann feststellen, was fehlgeschlagen ist und was erfolgreich war. AppOutput Der letzte Schritt besteht darin, die App aufzurufen. Dabei werden die der LLM-App verwendet, um zu bestimmen, wie sie mit einem einzelnen Datenpunkt aufgerufen werden soll. openapi_parameters Die Funktion make_payload sollte Sie nicht beunruhigen. Es erstellt die Nutzlast zum Aufrufen der LLM-App basierend auf seinen Parametern. 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") Und das rundet den Prozess ab. Zusammenfassung Die exponentielle Backoff-Strategie im Code funktioniert wie folgt: Die Funktion „batch_invoke“ teilt die Testsatzdaten in kleinere Stapel mit konfigurierbarer Größe auf. Jede Charge wird nacheinander verarbeitet. Stapelverarbeitung: Innerhalb jedes Stapels wird jeder Datenpunkt von der Funktion verarbeitet. Diese Funktion versucht, die App für den Datenpunkt aufzurufen. Wenn der Aufruf aufgrund bestimmter Netzwerkfehler (Zeitüberschreitungen, Verbindungsprobleme) fehlschlägt, versucht die Funktion mit einer Verzögerung erneut. Diese Verzögerung ist zunächst auf einen konfigurierbaren Wert ( ) eingestellt und wird für jeden weiteren Wiederholungsversuch innerhalb desselben Stapels verdoppelt. Einzelne Aufrufe mit Wiederholung: run_with_retry retry_delay Dieser Ansatz trägt dazu bei, eine Überlastung des App-Servers durch wiederholte Anfragen nach einem Fehler zu vermeiden. Dies gibt dem Server Zeit, sich zu erholen, und ermöglicht das Leeren der Warteschlange ausstehender Anforderungen vor einem erneuten Versuch. Die Strategie umfasst außerdem eine konfigurierbare maximale Anzahl von Wiederholungsversuchen pro Datenpunkt, um Endlosschleifen zu verhindern. Eine Verzögerung zwischen den Batches ( ) ist ebenfalls enthalten, um eine Überschreitung der vom App-Server festgelegten Ratengrenzen zu vermeiden. delay_between_batches Ich hoffe, dass dies alles zusammenfasst, was Sie im heutigen Artikel gelernt haben. Bitte lassen Sie mich wissen, wenn Sie Fragen haben!
