Apeluri API LLM Asincrone în Python: O Ghid Cuprinzător

Ca dezvoltatori și oameni de știință, adesea ne găsim în situația de a interacționa cu aceste modele puternice prin intermediul API-urilor. Cu toate acestea, pe măsură ce aplicațiile noastre cresc în complexitate și dimensiune, nevoia de interacțiuni eficiente și performante cu API-urile devine crucială. Aici, programarea asincronă strălucește, permițându-ne să maximizăm debitul și să minimizăm întârzierea atunci când lucrăm cu API-urile LLM.

În acest ghid cuprinzător, vom explora lumea apelurilor API LLM asincrone în Python. Vom acoperi totul, de la conceptele de bază ale programării asincrone până la tehnici avansate pentru gestionarea fluxurilor de lucru complexe. La sfârșitul acestui articol, veți avea o înțelegere solidă a modului de a utiliza programarea asincronă pentru a îmbunătăți aplicațiile dvs. LLM.

Înainte de a intra în detalii despre apelurile API LLM asincrone, să stabilim o bază solidă în conceptele de programare asincronă.

Programarea asincronă permite executarea mai multor operații în paralel, fără a bloca firul principal de execuție. În Python, acest lucru se realizează în principal prin intermediul modulului asyncio, care oferă un cadru pentru scrierea de cod concurent utilizând corutine, bucle de evenimente și obiecte viitoare.

Concepte cheie:

• Corutine: Funcții definite cu async def care pot fi oprite și reluate.
• Buclă de evenimente: Mecanismul central de execuție care gestionează și rulează sarcinile asincrone.
• Obiecte așteptate: Obiecte care pot fi utilizate cu cuvântul cheie await (corutine, sarcini, obiecte viitoare).

Iată un exemplu simplu pentru a ilustra aceste concepte:

import asyncio

<p>async def salut(nume):</p>
<p>await asyncio.sleep(1) # Simulează o operație I/O</p>
<p>print(f"Salut, {nume}!")</p>

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

asyncio.run(main())

În acest exemplu, definim o funcție asincronă salut care simulează o operație I/O cu asyncio.sleep(). Funcția main utilizează asyncio.gather() pentru a rula mai multe saluturi în paralel. În ciuda întârzierii de somn, toate trei saluturi vor fi tipărite după aproximativ 1 secundă, demonstrând puterea execuției asincrone.

Nevoia de asincronism în apelurile API LLM

Atunci când lucrăm cu API-urile LLM, adesea ne confruntăm cu scenarii în care trebuie să facem mai multe apeluri API, fie în secvență, fie în paralel. Codul sincron tradițional poate duce la blocaje semnificative de performanță, în special atunci când avem de-a face cu operații cu latență ridicată, cum ar fi solicitările de rețea către serviciile LLM.

Să considerăm un scenariu în care trebuie să generăm rezumate pentru 100 de articole diferite utilizând un API LLM. Cu o abordare sincronă, fiecare apel API ar bloca până primește un răspuns, posibil necesitând câteva minute pentru a finaliza toate solicitările. O abordare asincronă, pe de altă parte, ne permite să inițiem mai multe apeluri API în paralel, reducând semnificativ timpul de execuție total.

Configurarea mediului

Pentru a începe cu apelurile API LLM asincrone, trebuie să configurați mediul Python cu bibliotecile necesare. Iată ce aveți nevoie:

• Python 3.7 sau superior (pentru suport nativ asyncio)
• aiohttp: O bibliotecă client HTTP asincronă
• openai: Clientul oficial OpenAI pentru Python (dacă utilizați modelele GPT ale OpenAI)
• langchain: Un cadru pentru construirea de aplicații cu LLM (opțional, dar recomandat pentru fluxuri de lucru complexe)

Puteți instala aceste dependențe utilizând pip:

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

Apeluri API LLM asincrone de bază cu asyncio și aiohttp

Să începem prin a face un apel asincron simplu către un API LLM utilizând aiohttp. Vom utiliza API-ul GPT-3.5 al OpenAI ca exemplu, dar conceptele se aplică și altor API-uri LLM.

import asyncio
import aiohttp
from openai import AsyncOpenAI

<p>async def genereaza_text(prompt, client):</p>
<p>răspuns = await client.chat.completions.create(</p>
<p>model=&quot;gpt-3.5-turbo&quot;,</p>
<p>mesaje=[{&quot;rol&quot;: &quot;utilizator&quot;, &quot;conținut&quot;: prompt}]</p>
<p>)</p>
<p>return răspuns.alegeri[0].mesaj.conținut</p>

<p>async def main():</p>
<p>prompturi = [</p>
<p>&quot;Explicați calculul cuantic în termeni simpli.&quot;,</p>
<p>&quot;Scrieți un haiku despre inteligența artificială.&quot;,</p>
<p>&quot;Descrieți procesul de fotosinteză.&quot;</p>
<p>]</p>

<p>async with AsyncOpenAI() as client:</p>
<p>sarcini = [genereaza_text(prompt, client) for prompt in prompturi]</p>
<p>rezultate = await asyncio.gather(*sarcini)</p>

<p>for prompt, rezultat in zip(prompturi, rezultate):</p>
<p>print(f&quot;Prompt: {prompt}\nRăspuns: {rezultat}\n&quot;)</p>

asyncio.run(main())

În acest exemplu, definim o funcție asincronă genereaza_text care face un apel către API-ul OpenAI utilizând clientul AsyncOpenAI. Funcția main creează mai multe sarcini pentru diferite prompturi și utilizează asyncio.gather() pentru a rula aceste sarcini în paralel.

Această abordare ne permite să trimitem mai multe solicitări către API-ul LLM simultan, reducând semnificativ timpul total de execuție.

Tehnici avansate: Împachetarea și controlul concurenței

În timp ce exemplul anterior demonstrează conceptele de bază ale apelurilor API LLM asincrone, aplicațiile din lumea reală adesea necesită abordări mai sofisticate. Să explorăm două tehnici importante: împachetarea solicitărilor și controlul concurenței.

Împachetarea solicitărilor: Atunci când avem de-a face cu un număr mare de prompturi, este adesea mai eficient să le împachetăm în grupuri decât să trimitem solicitări individuale pentru fiecare prompt. Acest lucru reduce suprasarcina multiplelor apeluri API și poate duce la o performanță mai bună.

import asyncio
from openai import AsyncOpenAI

<p>async def proceseaza_pachet(pachet, client):</p>
<p>răspunsuri = await asyncio.gather([</p>
<p>client.chat.completions.create(</p>
<p>model=&quot;gpt-3.5-turbo&quot;,</p>
<p>mesaje=[{&quot;rol&quot;: &quot;utilizator&quot;, &quot;conținut&quot;: prompt}]</p>
<p>) for prompt in pachet</p>
<p>])</p>
<p>return [răspuns.alegeri[0].mesaj.conținut for răspuns in răspunsuri]</p>

<p>async def main():</p>
<p>prompturi = [f&quot;Spuneți-mi un fapt despre numărul {i}&quot; for i in range(100)]</p>
<p>dimensiunea_pachetului = 10</p>

<p>async with AsyncOpenAI() as client:</p>
<p>rezultate = []</p>
<p>for i in range(0, len(prompturi), dimensiunea_pachetului):</p>
<p>pachet = prompturi[i:i+dimensiunea_pachetului]</p>
<p>rezultate_pachet = await proceseaza_pachet(pachet, client)</p>
<p>rezultate.extend(rezultate_pachet)</p>

<p>for prompt, rezultat in zip(prompturi, rezultate):</p>
<p>print(f&quot;Prompt: {prompt}\nRăspuns: {rezultat}\n&quot;)</p>

asyncio.run(main())

Controlul concurenței: Deși programarea asincronă permite execuția concurentă, este important să controlați nivelul de concurență pentru a evita supraîncărcarea serverului API sau depășirea limitelor de rată. Putem utiliza asyncio.Semaphore pentru acest scop.

import asyncio
from openai import AsyncOpenAI

<p>async def genereaza_text(prompt, client, semafor):</p>
<p>async with semafor:</p>
<p>răspuns = await client.chat.completions.create(</p>
<p>model=&quot;gpt-3.5-turbo&quot;,</p>
<p>mesaje=[{&quot;rol&quot;: &quot;utilizator&quot;, &quot;conținut&quot;: prompt}]</p>
<p>)</p>
<p>return răspuns.alegeri[0].mesaj.conținut</p>

<p>async def main():</p>
<p>prompturi = [f&quot;Spuneți-mi un fapt despre numărul {i}&quot; for i in range(100)]</p>
<p>număr_solicitarilor_concurente = 5</p>
<p>semafor = asyncio.Semaphore(număr_solicitarilor_concurente)</p>

<p>async with AsyncOpenAI() as client:</p>
<p>sarcini = [genereaza_text(prompt, client, semafor) for prompt in prompturi]</p>
<p>rezultate = await asyncio.gather(*sarcini)</p>

<p>for prompt, rezultat in zip(prompturi, rezultate):</p>
<p>print(f&quot;Prompt: {prompt}\nRăspuns: {rezultat}\n&quot;)</p>

asyncio.run(main())

În acest exemplu, utilizăm un semafor pentru a limita numărul de solicitări concurente la 5, asigurându-ne că nu supraîncărcăm serverul API.

Gestionarea erorilor și reîncercările în apelurile LLM asincrone

Atunci când lucrăm cu API-uri externe, este crucial să implementăm o gestionare robustă a erorilor și mecanisme de reîncercare. Să îmbunătățim codul nostru pentru a gestiona erorile comune și pentru a implementa o reîncercare cu întârziere exponențială.

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

class EroareAPI(Exception):
pass

<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))</p>
<p>async def genereaza_text_cu_reîncercare(prompt, client):</p>
<p>try:</p>
<p>răspuns = await client.chat.completions.create(</p>
<p>model=&quot;gpt-3.5-turbo&quot;,</p>
<p>mesaje=[{&quot;rol&quot;: &quot;utilizator&quot;, &quot;conținut&quot;: prompt}]</p>
<p>)</p>
<p>return răspuns.alegeri[0].mesaj.conținut</p>
<p>except Exception as e:</p>
<p>print(f&quot;Eroare apărută: {e}&quot;)</p>
<p>raise EroareAPI(&quot;Nu am putut genera text&quot;)</p>

<p>async def proceseaza_prompt(prompt, client, semafor):</p>
<p>async with semafor:</p>
<p>try:</p>
<p>rezultat = await genereaza_text_cu_reîncercare(prompt, client)</p>
<p>return prompt, rezultat</p>
<p>except EroareAPI:</p>
<p>return prompt, &quot;Nu am putut genera răspuns după mai multe încercări.&quot;</p>

<p>async def main():</p>
<p>prompturi = [f&quot;Spuneți-mi un fapt despre numărul {i}&quot; for i in range(20)]</p>
<p>număr_solicitarilor_concurente = 5</p>
<p>semafor = asyncio.Semaphore(număr_solicitarilor_concurente)</p>

<p>async with AsyncOpenAI() as client:</p>
<p>sarcini = [proceseaza_prompt(prompt, client, semafor) for prompt in prompturi]</p>
<p>rezultate = await asyncio.gather(*sarcini)</p>

<p>for prompt, rezultat in rezultate:</p>
<p>print(f&quot;Prompt: {prompt}\nRăspuns: {rezultat}\n&quot;)</p>

asyncio.run(main())

Această versiune îmbunătățită include:

• O excepție personalizată EroareAPI pentru erori legate de API.
• O funcție genereaza_text_cu_reîncercare decorată cu @retry din biblioteca tenacity, care implementează o reîncercare cu întârziere exponențială.
• Gestionarea erorilor în funcția proceseaza_prompt pentru a prinde și raporta eșecurile.

Optimizarea performanței: Răspunsuri în flux

Pentru generarea de conținut de lungă durată, răspunsurile în flux pot îmbunătăți semnificativ performanța percepută a aplicației dvs. În loc de a aștepta întregul răspuns, puteți prelucra și afișa bucăți de text pe măsură ce devin disponibile.

import asyncio
from openai import AsyncOpenAI

<p>async def flux_text(prompt, client):</p>
<p>flux = await client.chat.completions.create(</p>
<p>model=&quot;gpt-3.5-turbo&quot;,</p>
<p>mesaje=[{&quot;rol&quot;: &quot;utilizator&quot;, &quot;conținut&quot;: prompt}],</p>
<p>flux=True</p>
<p>)</p>

<p>răspuns_complet = &quot;&quot;</p>
<p>async for bucățică in flux:</p>
<p>if bucățică.alegeri[0].delta.conținut is not None:</p>
<p>conținut = bucățică.alegeri[0].delta.conținut</p>
<p>răspuns_complet += conținut</p>
<p>print(conținut, end=&#039;&#039;, flush=True)</p>

<p>print(&quot;\n&quot;)</p>
<p>return răspuns_complet</p>

<p>async def main():</p>
<p>prompt = &quot;Scrieți o poveste scurtă despre un om de știință care călătorește în timp.&quot;</p>

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

<p>print(f&quot;Răspuns complet:\n{rezultat}&quot;)</p>

asyncio.run(main())

Acest exemplu demonstrează cum să fluxați răspunsul de la API, tipărind fiecare bucățică pe măsură ce sosește. Această abordare este deosebit de utilă pentru aplicații de chat sau orice scenariu în care doriți să oferiți feedback în timp real utilizatorului.

Construirea fluxurilor de lucru asincrone cu LangChain

Pentru aplicații LLM mai complexe, cadru LangChain oferă o abstracție de nivel înalt care simplifică procesul de înlănțuire a mai multor apeluri LLM și integrarea altor instrumente. Să explorăm un exemplu de utilizare a LangChain cu capacități asincrone:

Acest exemplu arată cum LangChain poate fi utilizat pentru a crea fluxuri de lucru mai complexe cu fluxare și execuție asincronă. AsyncCallbackManager și StreamingStdOutCallbackHandler permit fluxarea în timp real a conținutului generat.

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 genereaza_poveste(subiect):</p>
<p>llm = OpenAI(temperatura=0.7, flux=True, manager_callback=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p>prompt = PromptTemplate(</p>
<p>variabile_de_intrare=[&quot;subiect&quot;],</p>
<p>șablon=&quot;Scrieți o poveste scurtă despre {subiect}.&quot;</p>
<p>)</p>
<p>lanț = LLMChain(llm=llm, prompt=prompt)</p>
<p>return await lanț.arun(subiect=subiect)</p>

<p>async def main():</p>
<p>subiecte = [&quot;o pădure magică&quot;, &quot;un oraș futurist&quot;, &quot;o civilizație subacvatică&quot;]</p>
<p>sarcini = [genereaza_poveste(subiect) for subiect in subiecte]</p>
<p>povești = await asyncio.gather(*sarcini)</p>

<p>for subiect, poveste in zip(subiecte, povești):</p>
<p>print(f&quot;\nSubiect: {subiect}\nPoveste: {poveste}\n{&#039;=&#039;*50}\n&quot;)</p>

asyncio.run(main())

Servirea aplicațiilor LLM asincrone cu FastAPI

Pentru a face aplicația dvs. LLM asincronă disponibilă ca serviciu web, FastAPI este o alegere excelentă datorită suportului său nativ pentru operații asincrone. Iată un exemplu de creare a unui punct de terminare API simplu pentru generarea de text:

Acestă aplicație FastAPI creează un punct de terminare /generează care acceptă un prompt și returnează text generat. De asemenea, demonstrează cum să utilizați sarcini în background pentru prelucrări suplimentare fără a bloca răspunsul.

Practici recomandate și capcane comune

Pe măsură ce lucrați cu apelurile API LLM asincrone, țineți cont de aceste practici recomandate:

1. Folosiți pool-uri de conexiuni: Atunci când faceți mai multe solicitări, reutilizați conexiunile pentru a reduce suprasarcina.
2. Implementați o gestionare corespunzătoare a erorilor: Luați întotdeauna în considerare problemele de rețea, erorile API și răspunsurile neașteptate.
3. Respectați limitele de rată: Utilizați semafoare sau alte mecanisme de control al concurenței pentru a evita supraîncărcarea serverului API.
4. Monitorizați și înregistrați: Implementați o înregistrare cuprinzătoare pentru a urmări performanța și a identifica problemele.
5. Folosiți fluxarea pentru conținut de lungă durată: Îmbunătățește experiența utilizatorului și permite prelucrarea timpurie a rezultatelor parțiale.