Asynchrone LLM API-aanroepen in Python: een uitgebreide gids

Als ontwikkelaars en datawetenschappers hebben we vaak te maken met het interactief gebruiken van deze krachtige modellen via API’s. Echter, naarmate onze toepassingen complexer en groter worden, wordt de behoefte aan efficiënte en prestatiegerichte API-interacties cruciaal. Hier komt asynchrone programmering om de hoek kijken, waardoor we de doorvoer kunnen maximaliseren en de latentie kunnen minimaliseren bij het werken met LLM-API’s.

In deze uitgebreide gids zullen we de wereld van asynchrone LLM API-aanroepen in Python verkennen. We zullen alles behandelen, van de basisprincipes van asynchrone programmering tot geavanceerde technieken voor het afhandelen van complexe workflows. Aan het einde van dit artikel zult u een solide begrip hebben van hoe u asynchrone programmering kunt gebruiken om uw LLM-geactiveerde toepassingen te verbeteren.

Voordat we ons richten op de specifieke details van asynchrone LLM API-aanroepen, laten we eerst een solide basis leggen in asynchrone programmeringsconcepten.

Asynchrone programmering maakt het mogelijk om meerdere bewerkingen uit te voeren zonder de hoofdthread van uitvoering te blokkeren. In Python wordt dit voornamelijk bereikt via de asyncio-module, die een framework biedt voor het schrijven van concurrerende code met behulp van coroutines, event loops en futures.

Sleutelconcepten:

• Coroutines: Functies gedefinieerd met async def die kunnen worden gepauzeerd en hervat.
• Event Loop: Het centrale uitvoeringsmechanisme dat asynchrone taken beheert en uitvoert.
• Awaitables: Objecten die kunnen worden gebruikt met het await-sleutelwoord (coroutines, taken, futures).

Hier is een eenvoudig voorbeeld om deze concepten te illustreren:

import asyncio

<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # Simuleer een I/O-bewerking</p>
<p>print(f"Hallo, {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>

<p>asyncio.run(main())</p>

In dit voorbeeld definiëren we een asynchrone functie greet die een I/O-bewerking simuleert met asyncio.sleep(). De main-functie gebruikt asyncio.gather() om meerdere begroetingen tegelijk uit te voeren. Ondanks de vertraging door de slaap, worden alle drie de begroetingen afgedrukt na ongeveer 1 seconde, waarmee de kracht van asynchrone uitvoering wordt gedemonstreerd.

De noodzaak van asynchrone LLM API-aanroepen

Wanneer we werken met LLM-API’s, komen we vaak scenario’s tegen waarin we meerdere API-aanroepen moeten doen, hetzij opeenvolgend of parallel. Traditionele synchrone code kan leiden tot aanzienlijke prestatiebeperkingen, vooral bij het omgaan met hoog-latentiebewerkingen zoals netwerkverzoeken naar LLM-diensten.

Overweeg een scenario waarin we samenvattingen moeten genereren voor 100 verschillende artikelen met behulp van een LLM-API. Met een synchrone benadering zou elke API-aanroep worden geblokkeerd totdat deze een antwoord ontvangt, wat mogelijk enkele minuten kan duren om alle verzoeken te voltooien. Een asynchrone benadering daarentegen stelt ons in staat om meerdere API-aanroepen tegelijk te initiëren, waardoor de totale uitvoertijd aanzienlijk wordt verminderd.

Uw omgeving instellen

Om te beginnen met asynchrone LLM API-aanroepen, moet u uw Python-omgeving instellen met de nodige bibliotheken. Hier is wat u nodig heeft:

• Python 3.7 of hoger (voor native asyncio-ondersteuning)
• aiohttp: Een asynchrone HTTP-clientbibliotheek
• openai: De officiële OpenAI Python-client (als u OpenAI’s GPT-modellen gebruikt)
• langchain: Een framework voor het bouwen van toepassingen met LLM’s (optioneel, maar aanbevolen voor complexe workflows)

U kunt deze afhankelijkheden installeren met pip:

<p>pip install aiohttp openai langchain
&lt;div class=&quot;relative flex flex-col rounded-lg&quot;&gt;

Basisasynchrone LLM API-aanroepen met asyncio en aiohttp

Laten we beginnen met het maken van een eenvoudige asynchrone aanroep naar een LLM-API met behulp van aiohttp. We gebruiken OpenAI’s GPT-3.5-API als voorbeeld, maar de concepten zijn van toepassing op andere LLM-API’s.

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>"Leg quantumcomputing uit in eenvoudige bewoordingen.",</p>
<p>"Schrijf een haiku over kunstmatige intelligentie.",</p>
<p>"Beschrijf het proces van fotosynthese."</p>
<p>]</p>

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

<p>for prompt, resultaat in zip(prompts, resultaten):</p>
<p>print(f"Prompt: {prompt}\nAntwoord: {resultaat}\n")</p>

asyncio.run(main())

In dit voorbeeld definiëren we een asynchrone functie generate_text die een aanroep doet naar de OpenAI-API met behulp van de AsyncOpenAI-client. De main-functie maakt meerdere taken voor verschillende prompts en gebruikt asyncio.gather() om deze tegelijk uit te voeren.

Deze aanpak stelt ons in staat om meerdere verzoeken naar de LLM-API te sturen zonder dat we op de antwoorden hoeven te wachten, waardoor de totale verwerkingstijd aanzienlijk wordt verminderd.

Geavanceerde technieken: batchen en concurrencybeheer

Terwijl het vorige voorbeeld de basis van asynchrone LLM API-aanroepen demonstreert, vereisen echte toepassingen vaak meer geavanceerde benaderingen. Laten we twee belangrijke technieken verkennen: batchen van aanvragen en concurrencybeheer.

Batchen van aanvragen: wanneer u te maken heeft met een groot aantal prompts, is het vaak efficiënter om deze in batches te groeperen in plaats van afzonderlijke aanvragen te doen voor elke prompt. Dit vermindert de overhead van meerdere API-aanroepen en kan leiden tot betere prestaties.

import asyncio
from openai import AsyncOpenAI

<p>async def process_batch(batch, client):</p>
<p>antwoorden = 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 [antwoord.choices[0].message.content for antwoord in antwoorden]</p>

<p>async def main():</p>
<p>prompts = [f"Vertel me een feit over nummer {i}" for i in range(100)]</p>
<p>batch_size = 10</p>

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

<p>for prompt, resultaat in zip(prompts, resultaten):</p>
<p>print(f"Prompt: {prompt}\nAntwoord: {resultaat}\n")</p>

asyncio.run(main())

Concurrencybeheer: terwijl asynchrone programmering het mogelijk maakt om taken tegelijk uit te voeren, is het belangrijk om het niveau van concurrency te controleren om te voorkomen dat de API-server wordt overbelast of dat rate limits worden overschreden. We kunnen asyncio.Semaphore gebruiken voor dit doel.

import asyncio
from openai import AsyncOpenAI

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

<p>async def main():</p>
<p>prompts = [f"Vertel me een feit over nummer {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>taken = [generate_text(prompt, client, semaphore) for prompt in prompts]</p>
<p>resultaten = await asyncio.gather(*taken)</p>

<p>for prompt, resultaat in zip(prompts, resultaten):</p>
<p>print(f"Prompt: {prompt}\nAntwoord: {resultaat}\n")</p>

asyncio.run(main())

In dit voorbeeld gebruiken we een semaphore om het aantal gelijktijdige aanvragen te beperken tot 5, waardoor we voorkomen dat de API-server wordt overbelast.

Foutafhandeling en retries in asynchrone LLM-aanroepen

Wanneer u werkt met externe API’s, is het cruciaal om robuuste foutafhandeling en retry-mechanismeën te implementeren. Laten we ons code verbeteren om gemeenschappelijke fouten af te handelen en exponentiële backoff voor retries te implementeren.

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>antwoord = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return antwoord.choices[0].message.content</p>
<p>except Exception as e:</p>
<p>print(f"Fout opgetreden: {e}")</p>
<p>raise APIError("Kon tekst niet genereren")</p>

<p>async def process_prompt(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>try:</p>
<p>resultaat = await generate_text_with_retry(prompt, client)</p>
<p>return prompt, resultaat</p>
<p>except APIError:</p>
<p>return prompt, "Kon antwoord niet genereren na meerdere pogingen."</p>

<p>async def main():</p>
<p>prompts = [f"Vertel me een feit over nummer {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>taken = [process_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p>resultaten = await asyncio.gather(*taken)</p>

<p>for prompt, resultaat in resultaten:</p>
<p>print(f"Prompt: {prompt}\nAntwoord: {resultaat}\n")</p>

asyncio.run(main())

Deze verbeterde versie omvat:

• Een aangepaste APIError-uitzondering voor API-gerelateerde fouten.
• Een generate_text_with_retry-functie met de @retry-decorator van de tenacity-bibliotheek, die exponentiële backoff implementeert.
• Foutafhandeling in de process_prompt-functie om fouten te vangen en te melden.

Prestatie-optimalisatie: streaming-antwoorden

Voor langdurige inhoudsgeneratie kan het streamen van antwoorden de prestaties van uw toepassing aanzienlijk verbeteren. In plaats van te wachten op het volledige antwoord, kunt u chunks van tekst verwerken en weergeven zodra deze beschikbaar zijn.

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>volledig_antwoord = ""</p>
<p>async for chunk in stream:</p>
<p>if chunk.choices[0].delta.content is not None:</p>
<p>inhoud = chunk.choices[0].delta.content</p>
<p>volledig_antwoord += inhoud</p>
<p>print(inhoud, end="", flush=True)</p>

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

<p>async def main():</p>
<p>prompt = "Schrijf een kort verhaal over een tijdreizende wetenschapper."</p>

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

<p>print(f"Volledig antwoord:\n{resultaat}")</p>

asyncio.run(main())

Dit voorbeeld demonstreert hoe u antwoorden van de API kunt streamen, waarbij elke chunk wordt afgedrukt zodra deze beschikbaar is. Deze aanpak is vooral nuttig voor chat-toepassingen of elke situatie waarin u real-time feedback aan de gebruiker wilt bieden.

Asynchrone workflows bouwen met LangChain

Voor meer complexe LLM-geactiveerde toepassingen biedt het LangChain-framework een hoog niveau van abstractie dat het proces van het koppelen van meerdere LLM-aanroepen en het integreren van andere tools vereenvoudigt. Laten we een voorbeeld bekijken van het gebruik van LangChain met asynchrone mogelijkheden:

Dit voorbeeld laat zien hoe LangChain kan worden gebruikt om meer complexe workflows te maken met streaming en asynchrone uitvoering. De AsyncCallbackManager en StreamingStdOutCallbackHandler maken real-time streaming van gegenereerde inhoud mogelijk.

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(onderwerp):</p>
<p>llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p>prompt = PromptTemplate(</p>
<p>input_variables=["onderwerp"],</p>
<p>template="Schrijf een kort verhaal over {onderwerp}."</p>
<p>)</p>
<p>chain = LLMChain(llm=llm, prompt=prompt)</p>
<p>return await chain.arun(onderwerp=onderwerp)</p>

<p>async def main():</p>
<p>onderwerpen = ["een magisch bos", "een futuristische stad", "een onderwaterbeschaving"]</p>
<p>taken = [generate_story(onderwerp) for onderwerp in onderwerpen]</p>
<p>verhalen = await asyncio.gather(*taken)</p>

<p>for onderwerp, verhaal in zip(onderwerpen, verhalen):</p>
<p>print(f"\nOnderwerp: {onderwerp}\nVerhaal: {verhaal}\n{'='*50}\n")</p>

asyncio.run(main())

Asynchrone LLM-toepassingen serveren met FastAPI

Om uw asynchrone LLM-toepassing beschikbaar te maken als webdienst, is FastAPI een goede keuze vanwege de native ondersteuning voor asynchrone bewerkingen. Hier is een voorbeeld van hoe u een eenvoudig API-eindpunt kunt maken voor tekstgeneratie:

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>antwoord = 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 = antwoord.choices[0].message.content</p>

<p># Simuleer enkele post-processing in de achtergrond</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># Simuleer logging of extra verwerking</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")</p>

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

Deze FastAPI-toepassing maakt een eindpunt /generate dat een prompt accepteert en gegenereerde tekst teruggeeft. Het demonstreert ook hoe u achtergrondtaken kunt gebruiken voor extra verwerking zonder de reactie te blokkeren.

Beste praktijken en veelvoorkomende valkuilen

Terwijl u werkt met asynchrone LLM-API’s, houd deze beste praktijken in gedachten:

1. Gebruik connection pooling: wanneer u meerdere verzoeken doet, hergebruik verbindingen om overhead te verminderen.
2. Implementeer correcte foutafhandeling: neem altijd rekening met netwerkproblemen, API-fouten en onverwachte antwoorden.
3. Respecteer rate limits: gebruik semaforen of andere concurrencybeheermechanismen om te voorkomen dat de API-server wordt overbelast.
4. Bewaak en log: implementeer uitgebreide logging om prestaties te volgen en problemen te identificeren.
5. Gebruik streaming voor langdurige inhoud: dit verbetert de gebruikerservaring en stelt u in staat om partiële resultaten vroegtijdig te verwerken.