Modèles et plateformes d’IA
Appels d’API LLM asynchrones en Python : un guide complet
Ajouter Unite.AI à vos sources préférées sur Google
En tant que développeurs et scientifiques de données, nous nous retrouvons souvent à interagir avec ces puissants modèles via des API. Cependant, à mesure que nos applications grandissent en complexité et en échelle, le besoin d’interactions d’API efficaces et performantes devient crucial. C’est là que la programmation asynchrone brille, nous permettant de maximiser le débit et de minimiser la latence lors du travail avec les API LLM.
Dans ce guide complet, nous allons explorer le monde des appels d’API LLM asynchrones en Python. Nous allons couvrir tout, des bases de la programmation asynchrone aux techniques avancées pour gérer des flux de travail complexes. À la fin de cet article, vous aurez une solide compréhension de la façon d’utiliser la programmation asynchrone pour booster vos applications alimentées par LLM.
Avant de plonger dans les détails des appels d’API LLM asynchrones, établissons une solide fondation dans les concepts de programmation asynchrone.
La programmation asynchrone permet l’exécution multiple d’opérations sans bloquer le fil d’exécution principal. En Python, cela est principalement réalisé via le module asyncio, qui fournit un cadre pour écrire du code concurrent en utilisant des coroutines, des boucles d’événements et des futures.
Concepts clés :
- Coroutines : Fonctions définies avec async def qui peuvent être suspendues et reprises.
- Boucle d’événements : Le mécanisme d’exécution central qui gère et exécute des tâches asynchrones.
- Objets attendables : Objets qui peuvent être utilisés avec le mot-clé await (coroutines, tâches, futures).
Voici un exemple simple pour illustrer ces concepts :
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Simule une opération d'E/S
print(f"Bonjour, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Alice"),
greet("Bob"),
greet("Charlie")
)</p>
asyncio.run(main())
Dans cet exemple, nous définissons une fonction asynchrone greet qui simule une opération d’E/S avec asyncio.sleep(). La fonction main utilise asyncio.gather() pour exécuter plusieurs salutations de manière concurrente. Malgré le délai de sommeil, les trois salutations seront imprimées après environ 1 seconde, démontrant la puissance de l’exécution asynchrone.
Besoin d’asynchronisme dans les appels d’API LLM
Lorsque nous travaillons avec des API LLM, nous rencontrons souvent des scénarios où nous devons effectuer plusieurs appels d’API, soit en séquence, soit en parallèle. Le code synchrone traditionnel peut entraîner des goulets d’étranglement de performance importants, en particulier lorsqu’il s’agit d’opérations à latence élevée comme les requêtes réseau vers les services LLM.
Considérons un scénario où nous devons générer des résumés pour 100 articles différents à l’aide d’une API LLM. Avec une approche synchrone, chaque appel d’API bloquerait jusqu’à ce qu’il reçoive une réponse, ce qui pourrait prendre plusieurs minutes pour traiter toutes les requêtes. Une approche asynchrone, en revanche, nous permet d’initier plusieurs appels d’API de manière concurrente, réduisant ainsi considérablement le temps d’exécution total.
Configuration de votre environnement
Pour commencer avec les appels d’API LLM asynchrones, vous devez configurer votre environnement Python avec les bibliothèques nécessaires. Voici ce dont vous aurez besoin :
- Python 3.7 ou supérieur (pour la prise en charge native d’asyncio)
- aiohttp : Une bibliothèque cliente HTTP asynchrone
- openai : Le client Python officiel d’OpenAI (si vous utilisez les modèles GPT d’OpenAI)
- langchain : Un cadre pour la construction d’applications avec des LLM (facultatif, mais recommandé pour les flux de travail complexes)
Vous pouvez installer ces dépendances à l’aide de pip :
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Appels d’API LLM asynchrones de base avec asyncio et aiohttp
Commençons par effectuer un appel d’API LLM asynchrone simple en utilisant aiohttp. Nous allons utiliser l’API GPT-3.5 d’OpenAI comme exemple, mais les concepts s’appliquent à d’autres API LLM.
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client):
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content</p>
<p>async def main():
prompts = [
"Expliquez le calcul quantique en termes simples.",
"Écrivez un haïku sur l'intelligence artificielle.",
"Décrivez le processus de photosynthèse."
]</p>
<p>async with AsyncOpenAI() as client:
tasks = [generate_text(prompt, client) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nResponse: {result}\n")</p>
asyncio.run(main())
Dans cet exemple, nous définissons une fonction asynchrone generate_text qui effectue un appel à l’API d’OpenAI en utilisant le client AsyncOpenAI. La fonction main crée plusieurs tâches pour différents prompts et utilise asyncio.gather() pour les exécuter de manière concurrente.
Cette approche nous permet d’envoyer plusieurs requêtes à l’API LLM simultanément, réduisant ainsi considérablement le temps total nécessaire pour traiter tous les prompts.
Techniques avancées : regroupement et contrôle de la concurrence
Alors que l’exemple précédent démontre les bases des appels d’API LLM asynchrones, les applications réelles nécessitent souvent des approches plus sophistiquées. Explorons deux techniques importantes : le regroupement des requêtes et le contrôle de la concurrence.
Regroupement des requêtes : Lorsque nous traitons un grand nombre de prompts, il est souvent plus efficace de les regrouper en lots plutôt que d’envoyer des requêtes individuelles pour chaque prompt. Cela réduit la surcharge des appels d’API multiples et peut améliorer les performances.
import asyncio
from openai import AsyncOpenAI
<p>async def process_batch(batch, client):
responses = await asyncio.gather(*[
client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
) for prompt in batch
])
return [response.choices[0].message.content for response in responses]</p>
<p>async def main():
prompts = [f"Dites-moi un fait sur le nombre {i}" for i in range(100)]
batch_size = 10</p>
<p>async with AsyncOpenAI() as client:
results = []
for i in range(0, len(prompts), batch_size):
batch = prompts[i:i+batch_size]
batch_results = await process_batch(batch, client)
results.extend(batch_results)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nResponse: {result}\n")</p>
asyncio.run(main())
Contrôle de la concurrence : Bien que la programmation asynchrone permette l’exécution concurrente, il est important de contrôler le niveau de concurrence pour éviter de submerger le serveur d’API ou de dépasser les limites de débit. Nous pouvons utiliser asyncio.Semaphore pour cela.
import asyncio
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client, semaphore):
async with semaphore:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content</p>
<p>async def main():
prompts = [f"Dites-moi un fait sur le nombre {i}" for i in range(100)]
max_concurrent_requests = 5
semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:
tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nResponse: {result}\n")</p>
asyncio.run(main())
Dans cet exemple, nous utilisons un sémaphore pour limiter le nombre de requêtes concurrentes à 5, garantissant ainsi que nous ne submergions pas le serveur d’API.
Gestion des erreurs et des réessais dans les appels LLM asynchrones
Lorsque nous travaillons avec des API externes, il est crucial de mettre en œuvre une gestion des erreurs robuste et des mécanismes de réessai. Améliorons notre code pour gérer les erreurs courantes et mettre en œuvre une rétroaction exponentielle pour les réessais.
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))
async def generate_text_with_retry(prompt, client):
try:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except Exception as e:
print(f"Erreur survenue: {e}")
raise APIError("Échec de la génération de texte")</p>
<p>async def process_prompt(prompt, client, semaphore):
async with semaphore:
try:
result = await generate_text_with_retry(prompt, client)
return prompt, result
except APIError:
return prompt, "Échec de la génération de réponse après plusieurs tentatives."</p>
<p>async def main():
prompts = [f"Dites-moi un fait sur le nombre {i}" for i in range(20)]
max_concurrent_requests = 5
semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:
tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in results:
print(f"Prompt: {prompt}\nResponse: {result}\n")</p>
asyncio.run(main())
Cette version améliorée inclut :
- Une exception
APIErrorpersonnalisée pour les erreurs liées à l’API. - Une fonction
generate_text_with_retrydécorée avec@retryde la bibliothèque tenacity, mettant en œuvre une rétroaction exponentielle. - Une gestion des erreurs dans la fonction
process_promptpour capturer et signaler les échecs.
Optimisation des performances : réponses en flux
Pour la génération de contenu à long terme, les réponses en flux peuvent améliorer considérablement les performances perçues de votre application. Au lieu d’attendre la réponse complète, vous pouvez traiter et afficher des morceaux de texte à mesure qu’ils deviennent disponibles.
import asyncio
from openai import AsyncOpenAI
<p>async def stream_text(prompt, client):
stream = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
stream=True
)</p>
<p>full_response = ""
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
print(content, end='', flush=True)</p>
<p>print("\n")
return full_response</p>
<p>async def main():
prompt = "Écrivez une histoire courte sur un scientifique voyageant dans le temps."</p>
<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>
<p>print(f"Réponse complète:\n{result}")</p>
asyncio.run(main())
Cet exemple démontre comment diffuser la réponse de l’API, imprimant chaque morceau à mesure qu’il arrive. Cette approche est particulièrement utile pour les applications de chat ou tout scénario où vous souhaitez fournir une rétroaction en temps réel à l’utilisateur.
Construire des flux de travail asynchrones avec LangChain
Pour des applications plus complexes alimentées par LLM, le cadre LangChain fournit une abstraction de niveau supérieur qui simplifie le processus de chaînage de plusieurs appels LLM et d’intégration d’autres outils. Explorons un exemple d’utilisation de LangChain avec des capacités asynchrones :
Cet exemple montre comment LangChain peut être utilisé pour créer des flux de travail plus complexes avec une exécution asynchrone et en flux. Le AsyncCallbackManager et le StreamingStdOutCallbackHandler permettent un affichage en temps réel du contenu généré.
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):
llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
prompt = PromptTemplate(
input_variables=["topic"],
template="Écrivez une histoire courte sur {topic}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>
<p>async def main():
topics = ["une forêt magique", "une ville futuriste", "une civilisation sous-marine"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"\nThème: {topic}\nHistoire: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Servir des applications LLM asynchrones avec FastAPI
Pour rendre votre application LLM asynchrone disponible en tant que service Web, FastAPI est un excellent choix en raison de son support natif pour les opérations asynchrones. Voici un exemple de création d’un point de terminaison d’API simple pour la génération de texte :
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
<p>class GenerationRequest(BaseModel):
prompt: str</p>
<p>class GenerationResponse(BaseModel):
generated_text: str</p>
<p>@app.post("/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": request.prompt}]
)
generated_text = response.choices[0].message.content</p>
<p># Simulez un traitement post-génération en arrière-plan
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):
# Simulez un enregistrement ou un traitement supplémentaire
await asyncio.sleep(2)
print(f"Enregistré: Prompt '{prompt}' a généré un texte de longueur {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Cette application FastAPI crée un point de terminaison /generate qui accepte un prompt et retourne le texte généré. Elle démontre également comment utiliser des tâches en arrière-plan pour un traitement supplémentaire sans bloquer la réponse.
Meilleures pratiques et pièges courants
Lorsque vous travaillez avec des appels d’API LLM asynchrones, gardez à l’esprit ces meilleures pratiques :
- Utilisez le regroupement de connexions : Lorsque vous effectuez plusieurs requêtes, réutilisez les connexions pour réduire la surcharge.
- Mettez en œuvre une gestion des erreurs appropriée : Prenez toujours en compte les problèmes de réseau, les erreurs d’API et les réponses inattendues.
- Respectez les limites de débit : Utilisez des sémaphores ou d’autres mécanismes de contrôle de la concurrence pour éviter de submerger le serveur d’API.
- Surveillez et enregistrez : Mettez en œuvre une journalisation complète pour suivre les performances et identifier les problèmes.
- Utilisez les réponses en flux pour le contenu à long terme : Cela améliore l’expérience utilisateur et permet un traitement précoce des résultats partiels.
