AI-modeller og platforme
Asynkrone LLM API-kald i Python: En komplet vejledning
Føj Unite.AI til dine foretrukne kilder på Google
Som udviklere og datavidenskabsfolk har vi ofte brug for at interagere med disse kraftfulde modeller gennem API’er. Men når vores applikationer vokser i kompleksitet og omfang, bliver behovet for effektive og ydedygtige API-interaktioner afgørende. Det er her, asynkron programmering kommer til sin ret, og vi kan maksimere gennemstrømningen og minimere ventetiden, når vi arbejder med LLM-API’er.
I denne komplette vejledning vil vi udforske verden af asynkrone LLM API-kald i Python. Vi vil dække alt fra grundlæggende asynkron programmering til avancerede teknikker for håndtering af komplekse arbejdsgange. Til slut af denne artikel vil du have en solid forståelse af, hvordan du kan udnytte asynkron programmering til at give dit LLM-drevne applikationer et boost.
Før vi dykker ned i detaljerne om asynkrone LLM API-kald, lad os etablere en solid grund i asynkron programmeringskoncepter.
Asynkron programmering tillader flere operationer at udføres samtidigt uden at blokere hovedtråden af udførelse. I Python opnås dette primært gennem asyncio-modulen, der giver en ramme for at skrive koncurrent kode ved hjælp af korutiner, begivenhedsloop og fremtidsretninger.
Nøglekoncepter:
- Korutiner: Funktioner defineret med async def, der kan pauses og genoptages.
- Begivenhedsloop: Den centrale udførelse-mekanisme, der styrer og udfører asynkrone opgaver.
- VENTelige objekter: Objekter, der kan bruges med vent-ordet (korutiner, opgaver, fremtidsretninger).
Her er et simpelt eksempel for at illustrere disse koncepter:
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Simuler en I/O-operation
print(f"Hej, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Alice"),
greet("Bob"),
greet("Charlie")
)</p>
asyncio.run(main())
I dette eksempel definerer vi en asynkron funktion greet, der simulerer en I/O-operation med asyncio.sleep(). Funktionen main bruger asyncio.gather() til at udføre flere hilsner samtidigt. Trods sleep-forsinkelsen vil alle tre hilsner blive udskrevet efter ca. 1 sekund, hvilket demonstrerer asynkron udførelses kraft.
Behovet for asynkronitet i LLM API-kald
Når vi arbejder med LLM-API’er, møder vi ofte scenarier, hvor vi skal foretage flere API-kald, enten i sekvens eller parallelt. Traditionel synkron kode kan føre til betydelige ydelsesbottlenecks, især når vi har med høj-latens-operationer at gøre, som f.eks. netværksanmodninger til LLM-tjenester.
Overvej et scenarie, hvor vi skal generere sammenfatninger for 100 forskellige artikler ved hjælp af en LLM-API. Med en synkron tilgang ville hvert API-kald blokere, indtil det modtog en respons, hvilket potentielt kunne tage flere minutter at fuldføre alle anmodninger. En asynkron tilgang tillader os at initiere flere API-kald samtidigt, hvilket dramatisk reducerer den samlede udførelsestid.
Konfiguration af din miljø
For at komme i gang med asynkrone LLM API-kald skal du konfigurere din Python-miljø med de nødvendige biblioteker. Her er, hvad du skal bruge:
- Python 3.7 eller højere (for nativ asyncio-understøttelse)
- aiohttp: Et asynkront HTTP-klientbibliotek
- openai: Det officielle OpenAI Python-klient (hvis du bruger OpenAI’s GPT-modeller)
- langchain: En ramme for at bygge applikationer med LLM’er (valgfrit, men anbefalet til komplekse arbejdsgange)
Du kan installere disse afhængigheder ved hjælp af pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Grundlæggende asynkrone LLM API-kald med asyncio og aiohttp
Lad os starte med at lave et simpelt asynkront kald til en LLM-API ved hjælp af aiohttp. Vi skal bruge OpenAI’s GPT-3.5-API som eksempel, men koncepterne gælder for andre LLM-API’er.
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 = [
"Fortæl om kvantecomputering på en simpel måde.",
"Skriv et haiku om kunstig intelligens.",
"Beskriv processen 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}\nResponse: {result}\n")</p>
asyncio.run(main())
I dette eksempel definerer vi en asynkron funktion generate_text, der laver et kald til OpenAI’s API ved hjælp af AsyncOpenAI-klienten. Funktionen main opretter flere opgaver for forskellige prompts og bruger asyncio.gather() til at udføre dem samtidigt.
Denne tilgang tillader os at sende flere anmodninger til LLM-API’en samtidigt, hvilket betydeligt reducerer den samlede tid, der kræves for at behandle alle prompts.
Avancerede teknikker: Batchning og koncurrency-styring
Selvom det foregående eksempel demonstrerer grundlæggende asynkrone LLM API-kald, kræver virkelige applikationer ofte mere avancerede tilgange. Lad os udforske to vigtige teknikker: batchning af anmodninger og koncurrency-styring.
Batchning af anmodninger: Når vi har med en stor mængde prompts at gøre, er det ofte mere effektivt at batche dem i grupper i stedet for at sende enkeltanmodninger for hver prompt. Dette reducerer overhead’en af flere API-kald og kan føre til bedre ydelse.
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"Fortæl om tal {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())
Koncurrency-styring: Selvom asynkron programmering tillader samtidig udførelse, er det vigtigt at styre koncurrency-niveauet for at undgå at overbelaste API-serveren eller overskride rate-grænser. Vi kan bruge asyncio.Semaphore til dette formål.
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"Fortæl om tal {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())
I dette eksempel bruger vi en semaphore til at begrænse antallet af samtidige anmodninger til 5, hvilket sikrer, at vi ikke overbelaster API-serveren.
Fejlhåndtering og gentagelser i asynkrone LLM-kald
Når vi arbejder med eksterne API’er, er det afgørende at implementere robust fejlhåndtering og gentagelsesmekanismer. Lad os forbedre vores kode til at håndtere almindelige fejl og implementere eksponentiel tilbagefald for gentagelser.
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"Fejl opstået: {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 svar efter flere forsøg."</p>
<p>async def main():
prompts = [f"Fortæl om tal {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())
Denne forbedrede version inkluderer:
- En brugerdefineret
APIError-undtagelse for API-relaterede fejl. - En
generate_text_with_retry-funktion dekoreret med@retryfra tenacity-biblioteket, der implementerer eksponentiel tilbagefald. - Fejlhåndtering i
process_prompt-funktionen til at fange og rapportere fejl.
Optimering af ydelse: Streaming af svar
Til lange former for indholdsgenerering kan streaming af svar betydeligt forbedre brugeroplevelsen og tillade tidligere behandling af delvise resultater.
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 tidsrejsende videnskabsmand."</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 eksempel demonstrerer, hvordan man kan streame svaret fra API’en og udskrive hvert chunk, når det ankommer. Denne tilgang er især nyttig til chat-applikationer eller andre scenarier, hvor du ønsker at give brugeren tidlig feedback.
Bygning af asynkrone arbejdsgange med LangChain
Til mere komplekse LLM-drevne applikationer giver LangChain-rammen en højniveau-abstraktion, der forenkler processen med at kæde flere LLM-kald sammen og integrere andre værktøjer. Lad os se på et eksempel på, hvordan man kan bruge LangChain med asynkron kapacitet:
Dette eksempel viser, hvordan LangChain kan bruges til at oprette mere komplekse arbejdsgange med streaming og asynkron udførelse. AsyncCallbackManager og StreamingStdOutCallbackHandler aktiverer realtids-streaming af den genererede tekst.
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 skov", "en fremtidig by", "en undervandscivilisation"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"\nEmne: {topic}\nHistorie: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Tjeneste af asynkrone LLM-applikationer med FastAPI
For at gøre din asynkrone LLM-applikation tilgængelig som en webtjeneste er FastAPI et godt valg på grund af dets native understøttelse af asynkrone operationer. Her er et eksempel på, hvordan du kan oprette et simpelt API-endpoint 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 nogen post-behandling i baggrunden
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 yderligere behandling
await asyncio.sleep(2)
print(f"Logged: Prompt '{prompt}' generated tekst af længde {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Denne FastAPI-applikation opretter et endpoint /generate, der accepterer en prompt og returnerer genereret tekst. Den demonstrerer også, hvordan man kan bruge baggrundsopgaver til yderligere behandling uden at blokere svaret.
Bedste praksis og almindelige fælder
Når du arbejder med asynkrone LLM-API’er, husk følgende bedste praksis:
- Brug forbindelsespulje: Når du laver flere anmodninger, genbrug forbindelser til at reducere overhead.
- Implementer ordentlig fejlhåndtering: Tag altid hensyn til netværksproblemer, API-fejl og uventede svar.
- Respekter rate-grænser: Brug semaforer eller andre koncurrency-kontrollmekanismer til at undgå at overbelaste API-serveren.
- Overvåg og log: Implementer omfattende logging til at spore ydelse og identificere problemer.
- Brug streaming til lange former for indhold: Det forbedrer brugeroplevelsen og tillader tidligere behandling af delvise resultater.
