Chiamate API LLM Asincrone in Python: Una Guida Completa

Come sviluppatori e scienziati dei dati, spesso ci troviamo a interagire con questi potenti modelli attraverso API. Tuttavia, man mano che le nostre applicazioni crescono in complessità e scala, la necessità di interazioni API efficienti e performanti diventa cruciale. È qui che la programmazione asincrona brilla, consentendoci di massimizzare il throughput e minimizzare la latenza quando lavoriamo con API LLM.

In questa guida completa, esploreremo il mondo delle chiamate API LLM asincrone in Python. Copriremo tutto, dalle basi della programmazione asincrona a tecniche avanzate per la gestione di flussi di lavoro complessi. Alla fine di questo articolo, avrai una solida comprensione di come sfruttare la programmazione asincrona per potenziare le tue applicazioni LLM.

Prima di addentrarci nei dettagli delle chiamate API LLM asincrone, stabiliremo una solida base nei concetti di programmazione asincrona.

La programmazione asincrona consente l’esecuzione di più operazioni in modo concorrente senza bloccare il thread principale di esecuzione. In Python, ciò si ottiene principalmente attraverso il modulo asyncio, che fornisce un framework per la scrittura di codice concorrente utilizzando coroutine, event loop e future.

Concetti chiave:

• Coroutine: Funzioni definite con async def che possono essere sospese e riprese.
• Event Loop: Il meccanismo di esecuzione centrale che gestisce e esegue attività asincrone.
• Awaitable: Oggetti che possono essere utilizzati con la parola chiave await (coroutine, task, future).

Ecco un semplice esempio per illustrare questi concetti:

import asyncio

<p>async def greet(name):</p>
<p> await asyncio.sleep(1) # Simula un'operazione di I/O</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 questo esempio, definiamo una funzione asincrona greet che simula un’operazione di I/O con asyncio.sleep(). La funzione main utilizza asyncio.gather() per eseguire più saluti in modo concorrente. Nonostante il ritardo di sleep, tutti e tre i saluti saranno stampati dopo circa 1 secondo, dimostrando il potere dell’esecuzione asincrona.

La necessità di Async nelle chiamate API LLM

Quando lavoriamo con API LLM, spesso ci troviamo di fronte a scenari in cui dobbiamo effettuare più chiamate API, sia in sequenza che in parallelo. Il codice sincrono tradizionale può portare a significativi collo di bottiglia delle prestazioni, specialmente quando si ha a che fare con operazioni ad alta latenza come le richieste di rete ai servizi LLM.

Consideriamo uno scenario in cui dobbiamo generare riassunti per 100 articoli diversi utilizzando un’API LLM. Con un approccio sincrono, ogni chiamata API bloccerebbe fino a quando non ricevesse una risposta, potenzialmente richiedendo diversi minuti per completare tutte le richieste. Un approccio asincrono, d’altra parte, ci consente di avviare più chiamate API in modo concorrente, riducendo drasticamente il tempo di esecuzione complessivo.

Configurazione dell’ambiente

Per iniziare a lavorare con le chiamate API LLM asincrone, è necessario configurare l’ambiente Python con le librerie necessarie. Ecco cosa ti servirà:

• Python 3.7 o successivo (per il supporto nativo ad asyncio)
• aiohttp: Una libreria client HTTP asincrona
• openai: Il client Python ufficiale di OpenAI (se si utilizzano i modelli GPT di OpenAI)
• langchain: Un framework per la costruzione di applicazioni con LLM (opzionale, ma consigliato per flussi di lavoro complessi)

Puoi installare queste dipendenze utilizzando pip:

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

Chiamate API LLM asincrone di base con asyncio e aiohttp

Iniziamo facendo una semplice chiamata asincrona a un’API LLM utilizzando aiohttp. Useremo l’API GPT-3.5 di OpenAI come esempio, ma i concetti si applicano anche ad altre API LLM.

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> "Spiega il calcolo quantistico in termini semplici.",</p>
<p> "Scrivi una haiku sull'intelligenza artificiale.",</p>
<p> "Descrivi il processo della fotosintesi."</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}\nRisposta: {result}\n")</p>

asyncio.run(main())

In questo esempio, definiamo una funzione asincrona generate_text che effettua una chiamata all’API di OpenAI utilizzando il client AsyncOpenAI. La funzione main crea più attività per diversi prompt e utilizza asyncio.gather() per eseguirle in modo concorrente.

Questo approccio ci consente di inviare più richieste all’API LLM contemporaneamente, riducendo notevolmente il tempo totale necessario per elaborare tutti i prompt.

Tecniche avanzate: batching e controllo della concorrenza

Mentre l’esempio precedente dimostra le basi delle chiamate API LLM asincrone, le applicazioni reali spesso richiedono approcci più sofisticati. Esploriamo due tecniche importanti: il batching delle richieste e il controllo della concorrenza.

Batching delle richieste: quando si ha a che fare con un gran numero di prompt, è spesso più efficiente raggrupparli in lotti piuttosto che inviare singole richieste per ogni prompt. Ciò riduce l’onere delle multiple chiamate API e può portare a migliori prestazioni.

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"Racconta un fatto sul numero {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}\nRisposta: {result}\n")</p>

asyncio.run(main())

Controllo della concorrenza: mentre la programmazione asincrona consente l’esecuzione concorrente, è importante controllare il livello di concorrenza per evitare di sovraccaricare il server API o superare i limiti di velocità. Possiamo utilizzare asyncio.Semaphore per questo scopo.

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"Racconta un fatto sul numero {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}\nRisposta: {result}\n")</p>

asyncio.run(main())

In questo esempio, utilizziamo un semaforo per limitare il numero di richieste concorrenti a 5, assicurandoci di non sovraccaricare il server API.

Gestione degli errori e ritentativi nelle chiamate LLM asincrone

Quando si lavora con API esterne, è cruciale implementare una gestione degli errori robusta e meccanismi di ritentativo. Miglioriamo il nostro codice per gestire errori comuni e implementare un ritentativo con backoff esponenziale.

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"Errore occorso: {e}")</p>
<p> raise APIError("Errore durante la generazione del testo")</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, "Errore durante la generazione della risposta dopo più tentativi."</p>

<p>async def main():</p>
<p> prompts = [f"Racconta un fatto sul numero {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}\nRisposta: {result}\n")</p>

asyncio.run(main())

Questo codice migliorato include:

• Un’eccezione personalizzata APIError per errori relativi all’API.
• Una funzione generate_text_with_retry decorata con @retry dalla libreria tenacity, che implementa un backoff esponenziale.
• Gestione degli errori nella funzione process_prompt per catturare e segnalare gli errori.

Ottimizzazione delle prestazioni: risposte in streaming

Per la generazione di contenuti di lunga forma, le risposte in streaming possono migliorare notevolmente le prestazioni percepiti della tua applicazione. Invece di attendere l’intera risposta, puoi elaborare e visualizzare porzioni di testo man mano che diventano disponibili.

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 = "Scrivi una storia breve su un scienziato che viaggia nel tempo."</p>

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

<p>print(f"Risposta completa:\n{result}")</p>

asyncio.run(main())

Questo esempio dimostra come ricevere la risposta dall’API in streaming, stampando ogni porzione di testo man mano che arriva. Questo approccio è particolarmente utile per applicazioni di chat o qualsiasi scenario in cui si desidera fornire un feedback in tempo reale all’utente.

Costruzione di flussi di lavoro asincroni con LangChain

Per applicazioni LLM più complesse, il framework LangChain fornisce un’astrazione di alto livello che semplifica il processo di concatenamento di più chiamate LLM e l’integrazione di altri strumenti. Vediamo un esempio di utilizzo di LangChain con capacità asincrone:

Questo esempio mostra come LangChain possa essere utilizzato per creare flussi di lavoro più complessi con esecuzione asincrona e streaming. Il AsyncCallbackManager e il StreamingStdOutCallbackHandler abilitano lo streaming in tempo reale del contenuto generato.

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="Scrivi una storia breve su {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 = ["una foresta magica", "una città futuristica", "una civiltà sottomarina"]</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"\nArgomento: {topic}\nStoria: {story}\n{'='*50}\n")</p>

asyncio.run(main())

Servire applicazioni LLM asincrone con FastAPI

Per rendere la tua applicazione LLM asincrona disponibile come servizio web, FastAPI è una scelta eccellente a causa del suo supporto nativo alle operazioni asincrone. Ecco un esempio di come creare un endpoint API semplice per la generazione di testo:

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> # Simula un'elaborazione post-richiesta in background</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> # Simula la registrazione o un'elaborazione aggiuntiva</p>
<p> await asyncio.sleep(2)</p>
<p> print(f"Registrato: Prompt '{prompt}' ha generato testo di lunghezza {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>

Questa applicazione FastAPI crea un endpoint /generate che accetta un prompt e restituisce il testo generato. Dimostra anche come utilizzare attività in background per ulteriori elaborazioni senza bloccare la risposta.

Migliori pratiche e errori comuni

Mentre lavori con le chiamate API LLM asincrone, tieni presente queste migliori pratiche:

1. Utilizza il pooling delle connessioni: quando effettui più richieste, riutilizza le connessioni per ridurre l’onere.
2. Implementa una gestione degli errori adeguata: considera sempre problemi di rete, errori API e risposte inattese.
3. Rispetta i limiti di velocità: utilizza semafori o altri meccanismi di controllo della concorrenza per evitare di sovraccaricare il server API.
4. Monitora e registra: implementa una registrazione completa per tracciare le prestazioni e identificare problemi.
5. Utilizza lo streaming per contenuti di lunga forma: migliora l’esperienza utente e consente l’elaborazione precoce dei risultati parziali.