AI-modeller och plattformar
Asynkrona LLM API-anrop i Python: En omfattande guide
Lägg till Unite.AI bland dina föredragna källor på Google
Som utvecklare och dataforskare behöver vi ofta interagera med dessa kraftfulla modeller genom API:er. Men när våra applikationer växer i komplexitet och omfattning blir behovet av effektiva och presterande API-interaktioner avgörande. Här är där asynkron programmering lyser, vilket gör att vi kan maximera genomströmningen och minimera latency när vi arbetar med LLM API:er.
I denna omfattande guide kommer vi att utforska världen av asynkrona LLM API-anrop i Python. Vi kommer att täcka allt från grunderna i asynkron programmering till avancerade tekniker för att hantera komplexa arbetsflöden. När du har läst den här artikeln kommer du att ha en solid förståelse för hur du kan utnyttja asynkron programmering för att förbättra dina LLM-aktiverade applikationer.
Innan vi dyker in i detaljerna om asynkrona LLM API-anrop, låt oss etablera en solid grund i asynkron programmeringskoncept.
Asynkron programmering tillåter flera operationer att utföras samtidigt utan att blockera huvudtråden för körning. I Python uppnås detta främst genom asyncio-modulen, som tillhandahåller en ram för att skriva samtidig kod med hjälp av koroutiner, händelseloopar och framtidsutsikter.
Nyckelkoncept:
- Koroutiner: Funktioner definierade med async def som kan pausas och återupptas.
- Händelseloop: Den centrala körningsmekanismen som hanterar och kör asynkrona uppgifter.
- Inväntbara: Objekt som kan användas med await-nyckelordet (koroutiner, uppgifter, framtidsutsikter).
Här är ett enkelt exempel för att illustrera dessa koncept:
import asyncio
<p>async def hälsa(namn):
await asyncio.sleep(1) # Simulerar en I/O-operation
print(f"Hej, {namn}!")</p>
<p>async def main():
await asyncio.gather(
hälsa("Alice"),
hälsa("Bob"),
hälsa("Charlie")
)</p>
asyncio.run(main())
I det här exemplet definierar vi en asynkron funktion hälsa som simulerar en I/O-operation med asyncio.sleep(). Funktionen main använder asyncio.gather() för att köra flera hälsningar samtidigt. Trots fördröjningen kommer alla tre hälsningar att skrivas ut efter ungefär 1 sekund, vilket demonstrerar kraften i asynkron körning.
Behovet av asynkronitet i LLM API-anrop
När vi arbetar med LLM API:er stöter vi ofta på scenarier där vi behöver göra flera API-anrop, antingen i sekvens eller parallellt. Traditionell synkron kod kan leda till betydande prestandabottleneck, särskilt när det gäller höglatensoperationer som nätverksförfrågningar till LLM-tjänster.
Tänk på ett scenario där vi behöver generera sammanfattningar för 100 olika artiklar med hjälp av ett LLM API. Med en synkron approach skulle varje API-anrop blockera tills det får ett svar, vilket potentiellt kan ta flera minuter att slutföra alla förfrågningar. En asynkron approach, å andra sidan, tillåter oss att initiera flera API-anrop samtidigt, vilket dramatiskt minskar den totala exekveringstiden.
Konfigurera din miljö
För att komma igång med asynkrona LLM API-anrop behöver du konfigurera din Python-miljö med de nödvändiga biblioteken. Här är vad du behöver:
- Python 3.7 eller högre (för nativ asyncio-stöd)
- aiohttp: Ett asynkront HTTP-klientbibliotek
- openai: Det officiella OpenAI Python-klientbiblioteket (om du använder OpenAI:s GPT-modeller)
- langchain: Ett ramverk för att bygga applikationer med LLM:er (valfritt, men rekommenderas för komplexa arbetsflöden)
Du kan installera dessa beroenden med hjälp av pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Grundläggande asynkrona LLM API-anrop med asyncio och aiohttp
Låt oss börja med att göra ett enkelt asynkront anrop till ett LLM API med hjälp av aiohttp. Vi kommer att använda OpenAI:s GPT-3.5 API som exempel, men koncepten gäller för andra LLM API:er också.
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def generera_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 = [
"Förklara kvantberäkning på ett enkelt sätt.",
"Skriv en haiku om artificiell intelligens.",
"Beskriv processen för fotosyntes."
]</p>
<p>async with AsyncOpenAI() as client:
tasks = [generera_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}\nSvar: {result}\n")</p>
asyncio.run(main())
I det här exemplet definierar vi en asynkron funktion generera_text som gör ett anrop till OpenAI API:t med hjälp av AsyncOpenAI-klienten. Funktionen main skapar flera uppgifter för olika prompts och använder asyncio.gather() för att köra dem samtidigt.
Detta tillvägagångssätt tillåter oss att skicka flera förfrågningar till LLM API:t samtidigt, vilket dramatiskt minskar den totala exekveringstiden.
Avancerade tekniker: Batching och koncurrencykontroll
Medan det föregående exemplet demonstrerar grunderna i asynkrona LLM API-anrop, kräver riktiga applikationer ofta mer sofistikerade tillvägagångssätt. Låt oss utforska två viktiga tekniker: batchning av förfrågningar och koncurrencykontroll.
Batchning av förfrågningar: När du har att göra med ett stort antal prompts är det ofta mer effektivt att batcha dem i grupper istället för att skicka enskilda förfrågningar för varje prompt. Detta minskar overheaden för flera API-anrop och kan leda till bättre prestanda.
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"Berätta om en sak om nummer {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}\nSvar: {result}\n")</p>
asyncio.run(main())
Koncurrencykontroll: Medan asynkron programmering tillåter samtidig körning, är det viktigt att kontrollera koncurrency-nivån för att undvika att överbelasta API-servern eller överskrida rategränser. Vi kan använda asyncio.Semaphore för detta ändamål.
import asyncio
from openai import AsyncOpenAI
<p>async def generera_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"Berätta om en sak om nummer {i}" for i in range(100)]
max_concurrent_requests = 5
semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:
tasks = [generera_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}\nSvar: {result}\n")</p>
asyncio.run(main())
I det här exemplet använder vi en semaphore för att begränsa antalet samtidiga förfrågningar till 5, vilket säkerställer att vi inte överbelastar API-servern.
Felhantering och återförsök i asynkrona LLM-anrop
När du arbetar med externa API:er är det viktigt att implementera robust felhantering och återförsöksmekanismer. Låt oss förbättra vår kod för att hantera vanliga fel och implementera exponentiell återförsök.
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 generera_text_med_återförsök(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"Ett fel uppstod: {e}")
raise APIError("Misslyckades med att generera text")</p>
<p>async def process_prompt(prompt, client, semaphore):
async with semaphore:
try:
result = await generera_text_med_återförsök(prompt, client)
return prompt, result
except APIError:
return prompt, "Misslyckades med att generera svar efter flera försök."</p>
<p>async def main():
prompts = [f"Berätta om en sak om nummer {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}\nSvar: {result}\n")</p>
asyncio.run(main())
Denna förbättrade version innehåller:
- Ett anpassat
APIError-undantag för API-relaterade fel. - En
generera_text_med_återförsök-funktion som är dekorerad med@retryfrån tenacity-biblioteket, som implementerar exponentiell återförsök. - Felhantering i
process_prompt-funktionen för att catcha och rapportera fel.
Optimering av prestanda: Strömmande svar
För generering av långformigt innehåll kan strömmande svar förbättra prestandan avsevärt. Istället för att vänta på hela svaret kan du bearbeta och visa delar av texten allteftersom de blir tillgängliga.
import asyncio
from openai import AsyncOpenAI
<p>async def strömma_text(prompt, client):
stream = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
stream=True
)</p>
<p>fullt_svar = ""
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
fullt_svar += content
print(content, end='', flush=True)</p>
<p>print("\n")
return fullt_svar</p>
<p>async def main():
prompt = "Skriv en kort berättelse om en tidsresande vetenskapsman."</p>
<p>async with AsyncOpenAI() as client:
result = await strömma_text(prompt, client)</p>
<p>print(f"Fullt svar:\n{result}")</p>
asyncio.run(main())
I det här exemplet demonstrerar vi hur man kan strömma svaret från API:t, skriva ut varje del allteftersom den anländer. Detta tillvägagångssätt är särskilt användbart för chattapplikationer eller när du vill ge användaren realtidsåterkoppling.
Bygga asynkrona arbetsflöden med LangChain
För mer komplexa LLM-aktiverade applikationer tillhandahåller LangChain-ramverket en högnivåabstraktion som förenklar processen att kedja samman flera LLM-anrop och integrera andra verktyg. Låt oss titta på ett exempel på hur man använder LangChain med asynkrona funktioner:
Detta exempel visar hur LangChain kan användas för att skapa mer komplexa arbetsflöden med strömmande körning och asynkron exekvering. AsyncCallbackManager och StreamingStdOutCallbackHandler möjliggör realtidsströmning av det genererade innehållet.
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 generera_berättelse(ämne):
llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
prompt = PromptTemplate(
input_variables=["ämne"],
template="Skriv en kort berättelse om {ämne}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(ämne=ämne)</p>
<p>async def main():
ämnen = ["en magisk skog", "en framtida stad", "en undervattenscivilisation"]
tasks = [generera_berättelse(ämne) for ämne in ämnen]
berättelser = await asyncio.gather(*tasks)</p>
<p>for ämne, berättelse in zip(ämnen, berättelser):
print(f"\nÄmne: {ämne}\nBerättelse: {berättelse}\n{'='*50}\n")</p>
asyncio.run(main())
Tjäna asynkrona LLM-applikationer med FastAPI
För att göra din asynkrona LLM-applikation tillgänglig som en webbtjänst är FastAPI ett utmärkt val på grund av dess nativa stöd för asynkrona operationer. Här är ett exempel på hur du kan skapa ett enkelt API-ändpunkt för textgenerering:
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># Simulera någon postbearbetning i bakgrunden
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):
# Simulera loggning eller ytterligare bearbetning
await asyncio.sleep(2)
print(f"Loggat: Prompt '{prompt}' genererade text med längd {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Denna FastAPI-applikation skapar en ändpunkt /generate som accepterar en prompt och returnerar genererad text. Den demonstrerar också hur man kan använda bakgrundsaktiviteter för ytterligare bearbetning utan att blockera svaret.
Bästa praxis och vanliga fallgropar
När du arbetar med asynkrona LLM API:er, håll dessa bästa praxis i åtanke:
- Använd anslutningspoolning: När du gör flera förfrågningar, återanvänd anslutningar för att minska overhead.
- Implementera ordentlig felhantering: Ta alltid hänsyn till nätverksproblem, API-fel och oväntade svar.
- Respektera rategränser: Använd semaforer eller andra koncurrency-kontrollmekanismer för att undvika att överbelasta API-servern.
- Övervaka och logga: Implementera omfattande loggning för att spåra prestanda och identifiera problem.
- Använd strömning för långformigt innehåll: Det förbättrar användarupplevelsen och tillåter tidig bearbetning av delar av svaret.
