AI-modeller og plattformer
Asynkronne LLM API-forespørsler i Python: En omfattende guide
Legg til Unite.AI blant dine foretrukne kilder på Google
Som utviklere og dataforskere, finner vi ofte oss selv i situasjoner der vi må interagere med disse kraftfulle modellene gjennom API-er. Men når våre applikasjoner vokser i kompleksitet og skala, blir behovet for effektive og performante API-interaksjoner kritisk. Dette er der asynkron programmering skinner, og lar oss maksimere gjennomstrømming og minimere forsinkelse når vi arbeider med LLM-API-er.
I denne omfattende guiden, vil vi utforske verden av asynkronne LLM API-forespørsler i Python. Vi vil dekke alt fra grunnleggende asynkron programmering til avanserte tekniker for å håndtere komplekse arbeidsflyter. Ved slutten av denne artikkelen, vil du ha en solid forståelse av hvordan du kan utnytte asynkron programmering til å gi en ekstra boost til dine LLM-drevne applikasjoner.
Før vi dykker inn i detaljene om asynkronne LLM API-forespørsler, la oss etablere en solid grunn i asynkron programmeringskonsepter.
Asynkron programmering lar flere operasjoner bli utført samtidig uten å blokkere hovedtråden av eksekvering. I Python, oppnås dette hovedsakelig gjennom asyncio-modulen, som gir en ramme for å skrive samtidig kode med korutiner, hendelsesløkker og fremtidige hendelser.
Nøkkelkonsepter:
- Korutiner: Funksjoner definert med async def som kan bli pause og gjenopptatt.
- Hendelsesløkke: Den sentrale eksekveringsmekanisme som håndterer og kjører asynkronne oppgaver.
- Avventende objekter: Objekter som kan bli brukt med await-nøkkelordet (korutiner, oppgaver, fremtidige hendelser).
Her er et enkelt eksempel for å illustrere disse konseptene:
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Simuler en I/O-operasjon
print(f"Hei, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Alice"),
greet("Bob"),
greet("Charlie")
)</p>
asyncio.run(main())
I dette eksemplet, definerer vi en asynkron funksjon greet som simulerer en I/O-operasjon med asyncio.sleep(). Funksjonen main bruker asyncio.gather() for å kjøre flere hilsninger samtidig. Til tross for forsinkelsen, vil alle tre hilsningene bli skrevet ut etter omtrent 1 sekund, og demonstrerer kraften av asynkron eksekvering.
Behovet for asynkronitet i LLM API-forespørsler
Når vi arbeider med LLM-API-er, møter vi ofte situasjoner der vi må sende flere API-forespørsler, enten i sekvens eller parallelt. Tradisjonell synkron kode kan føre til betydelige ytelsesbottlenecks, spesielt når vi har å gjøre med høyforsinkelsesoperasjoner som nettverksforespørsler til LLM-tjenester.
La oss se på et scenario der vi må generere sammenfatninger for 100 forskjellige artikler med en LLM-API. Med en synkron tilnærming, ville hver API-forespørsel blokkere til den fikk en respons, potensielt tatt flere minutter å fullføre alle forespørsler. En asynkron tilnærming, på den andre siden, lar oss initiere flere API-forespørsler samtidig, og reduserer dramatisk den totale eksekveringstiden.
Konfigurasjon av miljøet ditt
For å komme i gang med asynkronne LLM API-forespørsler, må du konfigurere Python-miljøet ditt med de nødvendige bibliotekene. Her er hva du trenger:
- Python 3.7 eller høyere (for innbygget asyncio-støtte)
- aiohttp: Et asynkront HTTP-klientbibliotek
- openai: Det offisielle OpenAI Python-klienten (hvis du bruker OpenAI sine GPT-modeller)
- langchain: Et rammeverk for å bygge applikasjoner med LLM-er (valgfritt, men anbefalt for komplekse arbeidsflyter)
Du kan installere disse avhengighetene med pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Grunnleggende asynkronne LLM API-forespørsler med asyncio og aiohttp
La oss starte med å sende en enkel asynkron forespørsel til en LLM-API med aiohttp. Vi vil bruke OpenAI sin GPT-3.5-API som eksempel, men konseptene gjelder for andre LLM-API-er også.
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 = [
"Forklar kvantecomputing på en enkel måte.",
"Skriv en haiku om kunstig intelligens.",
"Beskriv prosessen med fotosyntese."
]</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}\nRespons: {result}\n")</p>
asyncio.run(main())
I dette eksemplet, definerer vi en asynkron funksjon generate_text som sender en forespørsel til OpenAI-API-et med AsyncOpenAI-klienten. Funksjonen main oppretter flere oppgaver for forskjellige forespørsler og bruker asyncio.gather() for å kjøre dem samtidig.
Denne tilnærmingen lar oss sende flere forespørsler til LLM-API-et samtidig, og reduserer betydelig den totale tiden det tar å prosessere alle forespørsler.
Avanserte tekniker: Batching og konkurransekontroll
Mens det forrige eksemplet demonstrerer grunnleggende asynkronne LLM API-forespørsler, krever virkelige applikasjoner ofte mer avanserte tilnærminger. La oss utforske to viktige tekniker: batching av forespørsler og konkurransekontroll.
Batching av forespørsler: Når du har å gjøre med et stort antall forespørsler, er det ofte mer effektivt å batche dem inn i grupper enn å sende enkeltforespørsler for hver enkelt forespørsel. Dette reduserer overheaden av flere API-forespørsler og kan føre til bedre ytelse.
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"Fortell meg en faktum om tall {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}\nRespons: {result}\n")</p>
asyncio.run(main())
Konkurransekontroll: Mens asynkron programmering lar oss kjøre flere operasjoner samtidig, er det viktig å kontrollere nivået av konkurranse for å unngå å overbelaste API-tjenesten eller overskride ratelimitene. Vi kan bruke asyncio.Semaphore for dette formålet.
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"Fortell meg en faktum om tall {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}\nRespons: {result}\n")</p>
asyncio.run(main())
I dette eksemplet, bruker vi en semaphore for å begrense antallet samtidige forespørsler til 5, og sikre at vi ikke overbelaster API-tjenesten.
Feilhåndtering og gjentakelser i asynkronne LLM-forespørsler
Når du arbeider med eksterne API-er, er det viktig å implementere robust feilhåndtering og gjentakelsesmekanismer. La oss forbedre koden vår til å håndtere vanlige feil og implementere eksponentiell tilbakehold for gjentakelser.
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"Feil oppsto: {e}")
raise APIError("Kunne ikke generere tekst")</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, "Kunne ikke generere respons etter flere forsøk."</p>
<p>async def main():
prompts = [f"Fortell meg en faktum om tall {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}\nRespons: {result}\n")</p>
asyncio.run(main())
Denne forbedrede versjonen inkluderer:
- En egen
APIError-unntak for API-relaterte feil. - En
generate_text_with_retry-funksjon dekorert med@retryfra tenacity-biblioteket, som implementerer eksponentiell tilbakehold. - Feilhåndtering i
process_prompt-funksjonen for å fange og rapportere feil.
Optimering av ytelse: Strømming av responser
For langforminnholdsgenerering, kan strømming av responser betydelig forbedre ytelsen til din applikasjon. I stedet for å vente på hele responsen, kan du prosessere og vise deler av teksten mens de blir tilgjengelige.
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 = "Skriv en kort historie om en tidreisende vitenskapsmann."</p>
<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>
<p>print(f"Full respons:\n{result}")</p>
asyncio.run(main())
Dette eksemplet demonstrerer hvordan du kan strømme responsen fra API-et, og skrive ut hver del av teksten mens den blir tilgjengelig. Denne tilnærmingen er spesielt nyttig for chat-applikasjoner eller andre scenarioer der du ønsker å gi brukeren sanntids tilbakemelding.
Bygging av asynkronne arbeidsflyter med LangChain
For mer komplekse LLM-drevne applikasjoner, gir LangChain-rammeverket en høyere abstraksjonsnivå som forenkler prosessen med å kjede sammen flere LLM-forespørsler og integrere andre verktøy. La oss se på et eksempel på hvordan LangChain kan brukes med asynkron kapasitet:
Dette eksemplet viser hvordan LangChain kan brukes til å lage mer komplekse arbeidsflyter med strømming og asynkron eksekvering. AsyncCallbackManager og StreamingStdOutCallbackHandler muliggjør sanntidsstrømming av generert innhold.
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="Skriv en kort historie om {topic}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>
<p>async def main():
topics = ["en magisk skog", "en fremtidsby", "en undervannsivilisasjon"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"\nTema: {topic}\nHistorie: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Tjenesteytelse av asynkronne LLM-applikasjoner med FastAPI
For å gjøre din asynkronne LLM-applikasjon tilgjengelig som en webtjeneste, er FastAPI et godt valg på grunn av sin innbygde støtte for asynkron operasjoner. Her er et eksempel på hvordan du kan lage en enkel API-endepunkt for tekstgenerering:
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># Simuler noen post-prosessering i bakgrunnen
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):
# Simuler logging eller annen prosessering
await asyncio.sleep(2)
print(f"Logget: Prompt '{prompt}' genererte tekst av lengde {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Denne FastAPI-applikasjonen lager et endepunkt /generate som aksepterer en forespørsel og returnerer generert tekst. Den demonstrerer også hvordan du kan bruke bakgrunnsoppgaver for ekstra prosessering uten å blokkere responsen.
Beste praksis og vanlige feil
Når du arbeider med asynkronne LLM API-forespørsler, husk på disse beste praksisene:
- Bruk tilkoblingspooling: Når du sender flere forespørsler, gjenbruk tilkoblinger for å redusere overhead.
- Implementer ordentlig feilhåndtering: Alltid ta hensyn til nettverksproblemer, API-feil og uventede responser.
- Respekter ratelimitene: Bruk semaforer eller andre konkurransekontrollmekanismer for å unngå å overbelaste API-tjenesten.
- Overvåk og logg: Implementer omfattende logging for å spore ytelse og identifisere problemer.
- Bruk strømming for langforminnhold: Det forbedrer brukeropplevelsen og lar deg prosessere deler av teksten mens den blir tilgjengelig.
