Asynchrone LLM-API-Anrufe in Python: Ein umfassender Leitfaden

Als Entwickler und Data-Scientists stoßen wir oft auf die Notwendigkeit, mit diesen leistungsstarken Modellen über APIs zu interagieren. Wenn jedoch unsere Anwendungen an Komplexität und Umfang zunehmen, wird die Notwendigkeit effizienter und leistungsfähiger API-Interaktionen immer wichtiger. Hier kommt die asynchrone Programmierung ins Spiel, die es uns ermöglicht, die Durchsatzrate zu maximieren und die Latenz bei der Arbeit mit LLM-APIs zu minimieren.

In diesem umfassenden Leitfaden werden wir die Welt der asynchronen LLM-API-Anrufe in Python erkunden. Wir werden alles von den Grundlagen der asynchronen Programmierung bis hin zu fortgeschrittenen Techniken für die Verarbeitung komplexer Workflows behandeln. Am Ende dieses Artikels werden Sie ein solides Verständnis dafür haben, wie Sie die asynchrone Programmierung nutzen können, um Ihre LLM-basierten Anwendungen zu beschleunigen.

Bevor wir uns auf die spezifischen asynchronen LLM-API-Anrufe einlassen, sollten wir ein solides Fundament in asynchronen Programmierkonzepten schaffen.

Asynchrone Programmierung ermöglicht es, mehrere Operationen gleichzeitig ohne Blockierung des Hauptthreads auszuführen. In Python wird dies hauptsächlich durch das Modul “asyncio” erreicht, das einen Rahmen für die Erstellung von konkurrierendem Code mit Hilfe von Coroutinen, Ereignisschleifen und Futures bietet.

Schlüsselkonzepte:

• Coroutinen: Funktionen, die mit async def definiert sind und die pausiert und fortgesetzt werden können.
• Ereignisschleife: Der zentrale Ausführungsmechanismus, der asynchrone Aufgaben verwaltet und ausführt.
• Wartbare Objekte: Objekte, die mit dem Schlüsselwort await verwendet werden können (Coroutinen, Aufgaben, Futures).

Hier ist ein einfaches Beispiel, um diese Konzepte zu veranschaulichen:

import asyncio

<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # Simuliere eine I/O-Operation</p>
<p>print(f"Hello, {name}!")</p>

<p>async def main():</p>
<p>await asyncio.gather(</p>
<p>greet("Alice"),</p>
<p>greet("Bob"),</p>
<p>greet("Charlie")</p>
<p>)</p>

asyncio.run(main())

In diesem Beispiel definieren wir eine asynchrone Funktion greet, die eine I/O-Operation mit asyncio.sleep() simuliert. Die main-Funktion verwendet asyncio.gather(), um mehrere Grüße gleichzeitig auszuführen. Trotz der Verzögerung durch asyncio.sleep() werden alle drei Grüße nach etwa einer Sekunde gedruckt, was die Leistungsfähigkeit der asynchronen Ausführung demonstriert.

Die Notwendigkeit von Async in LLM-API-Anrufen

Wenn wir mit LLM-APIs arbeiten, stoßen wir oft auf Szenarien, in denen wir mehrere API-Anrufe sequenziell oder parallel ausführen müssen. Traditionelle synchrone Code kann zu erheblichen Leistungsengpässen führen, insbesondere bei Operationen mit hoher Latenz wie Netzwerkanfragen an LLM-Dienste.

Betrachten wir ein Szenario, in dem wir Zusammenfassungen für 100 verschiedene Artikel mit einer LLM-API erstellen müssen. Mit einem synchronen Ansatz würde jeder API-Anruf blockieren, bis er eine Antwort erhält, was möglicherweise mehrere Minuten dauern könnte, um alle Anfragen abzuschließen. Ein asynchroner Ansatz ermöglicht es uns hingegen, mehrere API-Anrufe gleichzeitig auszuführen, was die Gesamtausführungszeit erheblich reduziert.

Einrichten Ihrer Umgebung

Um asynchrone LLM-API-Anrufe zu verwenden, müssen Sie Ihre Python-Umgebung mit den erforderlichen Bibliotheken einrichten. Hier sind die Anforderungen:

• Python 3.7 oder höher (für native asyncio-Unterstützung)
• aiohttp: Eine asynchrone HTTP-Client-Bibliothek
• openai: Der offizielle OpenAI Python-Client (wenn Sie OpenAI-GPT-Modelle verwenden)
• langchain: Ein Framework für die Erstellung von Anwendungen mit LLMs (optional, aber empfohlen für komplexe Workflows)

Sie können diese Abhängigkeiten mit pip installieren:

<p>pip install aiohttp openai langchain</p>

Grundlegende asynchrone LLM-API-Anrufe mit asyncio und aiohttp

Lassen Sie uns beginnen, indem wir einen einfachen asynchronen Anruf an eine LLM-API mit aiohttp ausführen. Wir verwenden OpenAI-GPT-3.5-API als Beispiel, aber die Konzepte gelten auch für andere LLM-APIs.

import asyncio
import aiohttp
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>

<p>async def main():</p>
<p>prompts = [</p>
<p>"Erkläre Quantencomputing in einfachen Worten.",</p>
<p>"Schreibe ein Haiku über künstliche Intelligenz.",</p>
<p>"Beschreibe den Prozess der Photosynthese."</p>
<p>]</p>

<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [generate_text(prompt, client) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"Prompt: {prompt}\nAntwort: {result}\n")</p>

asyncio.run(main())

In diesem Beispiel definieren wir eine asynchrone Funktion generate_text, die einen Anruf an die OpenAI-API mit dem AsyncOpenAI-Client ausführt. Die main-Funktion erstellt mehrere Aufgaben für verschiedene Prompts und verwendet asyncio.gather(), um sie gleichzeitig auszuführen.

Dieser Ansatz ermöglicht es uns, mehrere Anfragen an die LLM-API gleichzeitig zu senden, was die Gesamtausführungszeit erheblich reduziert.

Fortgeschrittene Techniken: Batchverarbeitung und Konkurrenzkontrolle

Während das vorherige Beispiel die Grundlagen asynchroner LLM-API-Anrufe demonstriert, erfordern reale Anwendungen oft komplexere Ansätze. Lassen Sie uns zwei wichtige Techniken erkunden: Batchverarbeitung und Konkurrenzkontrolle.

Batchverarbeitung: Wenn Sie mit einer großen Anzahl von Prompts arbeiten, ist es oft effizienter, sie in Gruppen zu bündeln, anstatt einzelne Anfragen für jeden Prompt zu senden. Dies reduziert den Overhead mehrerer API-Anrufe und kann zu besserer Leistung führen.

import asyncio
from openai import AsyncOpenAI

<p>async def process_batch(batch, client):</p>
<p>responses = await asyncio.gather(*[</p>
<p>client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>) for prompt in batch</p>
<p>])</p>
<p>return [response.choices[0].message.content for response in responses]</p>

<p>async def main():</p>
<p>prompts = [f"Erzähle mir eine Tatsache über die Zahl {i}" for i in range(100)]</p>
<p>batch_size = 10</p>

<p>async with AsyncOpenAI() as client:</p>
<p>results = []</p>
<p>for i in range(0, len(prompts), batch_size):</p>
<p>batch = prompts[i:i+batch_size]</p>
<p>batch_results = await process_batch(batch, client)</p>
<p>results.extend(batch_results)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"Prompt: {prompt}\nAntwort: {result}\n")</p>

asyncio.run(main())

Konkurrenzkontrolle: Wenn Sie asynchrone Anrufe ausführen, ist es wichtig, die Konkurrenz zu kontrollieren, um den API-Server nicht zu überlasten oder Rate-Limits zu überschreiten. Wir können asyncio.Semaphore für diesen Zweck verwenden.

import asyncio
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>

<p>async def main():</p>
<p>prompts = [f"Erzähle mir eine Tatsache über die Zahl {i}" for i in range(100)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>

<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"Prompt: {prompt}\nAntwort: {result}\n")</p>

asyncio.run(main())

In diesem Beispiel verwenden wir ein Semaphore, um die Anzahl der gleichzeitigen Anfragen auf 5 zu begrenzen, um den API-Server nicht zu überlasten.

Fehlerbehandlung und Wiederholungen in asynchronen LLM-Anrufen

Wenn wir mit externen APIs arbeiten, ist es wichtig, eine robuste Fehlerbehandlung und Wiederholungsmechanismen zu implementieren. Lassen Sie uns unseren Code erweitern, um häufige Fehler zu behandeln und exponentielle Wiederholungen zu implementieren.

import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

class APIError(Exception):
pass

<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))</p>
<p>async def generate_text_with_retry(prompt, client):</p>
<p>try:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>except Exception as e:</p>
<p>print(f"Fehler aufgetreten: {e}")</p>
<p>raise APIError("Fehler bei der Texterstellung")</p>

<p>async def process_prompt(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>try:</p>
<p>result = await generate_text_with_retry(prompt, client)</p>
<p>return prompt, result</p>
<p>except APIError:</p>
<p>return prompt, "Fehler bei der Texterstellung nach mehreren Versuchen."</p>

<p>async def main():</p>
<p>prompts = [f"Erzähle mir eine Tatsache über die Zahl {i}" for i in range(20)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>

<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in results:</p>
<p>print(f"Prompt: {prompt}\nAntwort: {result}\n")</p>

asyncio.run(main())

Diese erweiterte Version umfasst:

• Eine benutzerdefinierte APIError-Ausnahme für API-spezifische Fehler.
• Eine generate_text_with_retry-Funktion, die mit @retry aus der tenacity-Bibliothek dekoriert ist, um exponentielle Wiederholungen zu implementieren.
• Fehlerbehandlung in der process_prompt-Funktion, um Fehler zu erfassen und zu melden.

Leistungsoptimierung: Streamen von Antworten

Für die Erstellung von Langtexten kann das Streamen von Antworten die wahrgenommene Leistung Ihrer Anwendung erheblich verbessern. Anstatt auf die gesamte Antwort zu warten, können Sie Teile des Textes verarbeiten und anzeigen, sobald sie verfügbar sind.

import asyncio
from openai import AsyncOpenAI

<p>async def stream_text(prompt, client):</p>
<p>stream = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}],</p>
<p>stream=True</p>
<p>)</p>

<p>full_response = ""</p>
<p>async for chunk in stream:</p>
<p>if chunk.choices[0].delta.content is not None:</p>
<p>content = chunk.choices[0].delta.content</p>
<p>full_response += content</p>
<p>print(content, end="", flush=True)</p>

<p>print("\n")</p>
<p>return full_response</p>

<p>async def main():</p>
<p>prompt = "Schreibe eine kurze Geschichte über einen Zeitreisenden."</p>

<p>async with AsyncOpenAI() as client:</p>
<p>result = await stream_text(prompt, client)</p>

<p>print(f"Vollständige Antwort:\n{result}")</p>

asyncio.run(main())

In diesem Beispiel demonstrieren wir, wie Sie die Antwort von der API streamen und jeden Teil des Textes ausgeben, sobald er verfügbar ist. Dieser Ansatz ist besonders nützlich für Chat-Anwendungen oder Szenarien, in denen Sie dem Benutzer Echtzeit-Feedback liefern möchten.

Erstellen asynchroner Workflows mit LangChain

Für komplexere LLM-basierte Anwendungen bietet das LangChain-Framework eine höhere Abstraktionsebene, die den Prozess des Kettens mehrerer LLM-Anrufe und der Integration anderer Tools vereinfacht. Lassen Sie uns ein Beispiel für die Verwendung von LangChain mit asynchronen Fähigkeiten betrachten:

Dieses Beispiel zeigt, wie LangChain verwendet werden kann, um komplexere Workflows mit Streaming und asynchroner Ausführung zu erstellen. Der AsyncCallbackManager und der StreamingStdOutCallbackHandler ermöglichen die Echtzeit-Verarbeitung des generierten Textes.

import asyncio
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.callbacks.manager import AsyncCallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

<p>async def generate_story(topic):</p>
<p>llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p>prompt = PromptTemplate(</p>
<p>input_variables=["topic"],</p>
<p>template="Schreibe eine kurze Geschichte über {topic}."</p>
<p>)</p>
<p>chain = LLMChain(llm=llm, prompt=prompt)</p>
<p>return await chain.arun(topic=topic)</p>

<p>async def main():</p>
<p>topics = ["ein magischer Wald", "eine futuristische Stadt", "eine Unterwasserzivilisation"]</p>
<p>tasks = [generate_story(topic) for topic in topics]</p>
<p>stories = await asyncio.gather(*tasks)</p>

<p>for topic, story in zip(topics, stories):</p>
<p>print(f"\nThema: {topic}\nGeschichte: {story}\n{'='*50}\n")</p>

asyncio.run(main())

Bereitstellen asynchroner LLM-Anwendungen mit FastAPI

Um Ihre asynchrone LLM-Anwendung als Webdienst bereitzustellen, ist FastAPI aufgrund seiner nativen Unterstützung für asynchrone Operationen eine gute Wahl. Hier ist ein Beispiel für die Erstellung eines einfachen API-Endpunkts für die Texterstellung:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

<p>class GenerationRequest(BaseModel):</p>
<p>prompt: str</p>

<p>class GenerationResponse(BaseModel):</p>
<p>generated_text: str</p>

<p>@app.post("/generate", response_model=GenerationResponse)</p>
<p>async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": request.prompt}]</p>
<p>)</p>
<p>generated_text = response.choices[0].message.content</p>

<p># Simulieren Sie einige Nachverarbeitungsschritte im Hintergrund</p>
<p>background_tasks.add_task(log_generation, request.prompt, generated_text)</p>

<p>return GenerationResponse(generated_text=generated_text)</p>

<p>async def log_generation(prompt: str, generated_text: str):</p>
<p># Simulieren Sie Protokollierung oder weitere Verarbeitung</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"Protokoll: Prompt '{prompt}' hat Text der Länge {len(generated_text)} generiert")</p>

<p>if __name__ == "__main__":</p>
<p>import uvicorn</p>
<p>uvicorn.run(app, host="0.0.0.0", port=8000)</p>

Diese FastAPI-Anwendung erstellt einen Endpunkt /generate, der einen Prompt annimmt und generierten Text zurückgibt. Sie demonstriert auch, wie Hintergrundaufgaben für weitere Verarbeitung verwendet werden können, ohne die Antwort zu blockieren.

Best Practices und häufige Fallstricke

Wenn Sie mit asynchronen LLM-API-Anrufen arbeiten, sollten Sie diese Best Practices beachten:

1. Verwenden Sie Verbindungspooling: Wenn Sie mehrere Anfragen ausführen, sollten Sie Verbindungen wiederverwenden, um den Overhead zu reduzieren.
2. Implementieren Sie ordnungsgemäße Fehlerbehandlung: Berücksichtigen Sie immer Netzwerkprobleme, API-Fehler und unerwartete Antworten.
3. Respektieren Sie Rate-Limits: Verwenden Sie Semaphore oder andere Konkurrenzkontrollmechanismen, um den API-Server nicht zu überlasten.
4. Überwachen und protokollieren Sie: Implementieren Sie umfassende Protokollierung, um Leistung und Probleme zu verfolgen.
5. Verwenden Sie Streaming für Langtexte: Dies verbessert die Benutzererfahrung und ermöglicht die frühe Verarbeitung von Teilergebnissen.