Asynkrona LLM API-anrop i Python: A Comprehensive Guide

Som utvecklare och DTA-forskare finner vi ofta att vi behöver interagera med dessa kraftfulla modeller genom API:er. Men när våra applikationer växer i komplexitet och skala, blir behovet av effektiva och presterande API-interaktioner avgörande. Det är här asynkron programmering lyser, vilket gör att vi kan maximera genomströmningen och minimera latens när vi arbetar med LLM API:er.

I den här omfattande guiden 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. I slutet av den här artikeln har du en gedigen förståelse för hur du kan utnyttja asynkron programmering för att överladda dina LLM-drivna applikationer.

Innan vi dyker in i detaljerna för asynkrona LLM API-anrop, låt oss etablera en solid grund i asynkrona programmeringskoncept.

Asynkron programmering tillåter flera operationer att utföras samtidigt utan att blockera huvudtråden för exekvering. I Python uppnås detta främst genom asyncio modul, som tillhandahåller ett ramverk för att skriva samtidig kod med hjälp av koroutiner, händelseloopar och terminer.

Nyckelbegrepp:

• Coroutiner: Funktioner definierade med asynkron def som kan pausas och återupptas.
• EventLoop: Den centrala exekveringsmekanismen som hanterar och kör asynkrona uppgifter.
• Väntar: Objekt som kan användas med nyckelordet await (coroutines, tasks, futures).

Här är ett enkelt exempel för att illustrera dessa begrepp:

import asyncio

async def greet(name):
    await asyncio.sleep(1)  # Simulate an I/O operation
    print(f"Hello, {name}!")

async def main():
    await asyncio.gather(
        greet("Alice"),
        greet("Bob"),
        greet("Charlie")
    )

asyncio.run(main())

I det här exemplet definierar vi en asynkron funktion greet som simulerar en I/O-operation med asyncio.sleep(). De main funktion använder asyncio.gather() för att köra flera hälsningar samtidigt. Trots sömnfördröjningen kommer alla tre hälsningarna att skrivas ut efter cirka 1 sekund, vilket visar kraften i asynkron exekvering.

Behovet av asynkronisering 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 prestandaflaskhalsar, särskilt när man hanterar operationer med hög latens som nätverksbegäranden 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 ett synkront tillvägagångssätt skulle varje API-anrop blockeras tills det får ett svar, vilket kan ta flera minuter att slutföra alla förfrågningar. Ett asynkront tillvägagångssätt, å andra sidan, tillåter oss att initiera flera API-anrop samtidigt, vilket dramatiskt minskar den totala exekveringstiden.

Ställa in din miljö

För att komma igång med asynkrona LLM API-anrop måste 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 inbyggt asyncio-stöd)
• aiohttp: Ett asynkront HTTP-klientbibliotek
• openai: Den officiella OpenAI Python-klient (om du använder OpenAI:s GPT-modeller)
• långkedja: Ett ramverk för att bygga applikationer med LLM (valfritt, men rekommenderas för komplexa arbetsflöden)

Du kan installera dessa beroenden med hjälp av pip:

pip install aiohttp openai langchain
<div class="relative flex flex-col rounded-lg">

Basic Async 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 aiohttp. Vi kommer att använda OpenAI:s GPT-3.5 API som ett exempel, men koncepten gäller även för andra LLM API:er.

import asyncio
import aiohttp
from openai import AsyncOpenAI

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

async def main():
    prompts = [
        "Explain quantum computing in simple terms.",
        "Write a haiku about artificial intelligence.",
        "Describe the process of photosynthesis."
    ]
    
    async with AsyncOpenAI() as client:
        tasks = [generate_text(prompt, client) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

I det här exemplet definierar vi en asynkron funktion generate_text som gör ett anrop till OpenAI API med AsyncOpenAI-klienten. De main funktionen skapar flera uppgifter för olika uppmaningar och användningsområden asyncio.gather() att köra dem samtidigt.

Detta tillvägagångssätt tillåter oss att skicka flera förfrågningar till LLM API samtidigt, vilket avsevärt minskar den totala tiden som krävs för att behandla alla meddelanden.

Avancerade tekniker: Batchning och samtidighetskontroll

Även om det föregående exemplet visar grunderna för asynkrona LLM API-anrop, kräver verkliga applikationer ofta mer sofistikerade tillvägagångssätt. Låt oss utforska två viktiga tekniker: batchförfrågningar och kontroll av samtidighet.

Batchförfrågningar: När man hanterar ett stort antal uppmaningar är det ofta mer effektivt att gruppera dem i grupper snarare än att skicka individuella förfrågningar för varje prompt. Detta minskar omkostnaderna för flera API-anrop och kan leda till bättre prestanda.

import asyncio
from openai import AsyncOpenAI

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]

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(100)]
    batch_size = 10
    
    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)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Samtidighetskontroll: Även om asynkron programmering tillåter samtidig exekvering, är det viktigt att kontrollera samtidighetsnivån för att undvika att API-servern överbelastas eller hastighetsgränser överskrids. Vi kan använda asyncio.Semaphore för detta ändamål.

import asyncio
from openai import AsyncOpenAI

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

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(100)]
    max_concurrent_requests = 5
    semaphore = asyncio.Semaphore(max_concurrent_requests)
    
    async with AsyncOpenAI() as client:
        tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

I det här exemplet använder vi en semafor för att begränsa antalet samtidiga förfrågningar till 5, för att säkerställa att vi inte överväldiga API-servern.

Felhantering och återförsök i Async LLM-samtal

När du arbetar med externa API:er är det avgörande att implementera robusta felhanterings- och försöksmekanismer. Låt oss förbättra vår kod för att hantera vanliga fel och implementera exponentiell backoff för återförsök.

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

class APIError(Exception):
    pass

@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"Error occurred: {e}")
        raise APIError("Failed to generate text")

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, "Failed to generate response after multiple attempts."

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(20)]
    max_concurrent_requests = 5
    semaphore = asyncio.Semaphore(max_concurrent_requests)
    
    async with AsyncOpenAI() as client:
        tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in results:
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Denna förbättrade version innehåller:

• En anpassad APIError undantag för API-relaterade fel.
• A generate_text_with_retry funktion dekorerad med @retry från tenacity-biblioteket och implementerar exponentiell backoff.
• Felhantering i process_prompt funktion för att fånga upp och rapportera fel.

Optimera prestanda: Strömmande svar

För generering av innehåll i långa former kan strömmande svar avsevärt förbättra den upplevda prestandan för din applikation. Istället för att vänta på hela svaret kan du bearbeta och visa textbitar när de blir tillgängliga.

import asyncio
from openai import AsyncOpenAI

async def stream_text(prompt, client):
    stream = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    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)
    
    print("\n")
    return full_response

async def main():
    prompt = "Write a short story about a time-traveling scientist."
    
    async with AsyncOpenAI() as client:
        result = await stream_text(prompt, client)
    
    print(f"Full response:\n{result}")

asyncio.run(main())

Det här exemplet visar hur man strömmar svaret från API:t och skriver ut varje del när den kommer. Detta tillvägagångssätt är särskilt användbart för chattapplikationer eller andra scenarier där du vill ge feedback i realtid till användaren.

Bygga asynkrona arbetsflöden med LangChain

För mer komplexa LLM-drivna applikationer LangChain ramverk ger en abstraktion på hög nivå som förenklar processen med att koppla ihop flera LLM-samtal och integrera andra verktyg. Låt oss titta på ett exempel på hur man använder LangChain med asynkfunktioner:

Det här exemplet visar hur LangChain kan användas för att skapa mer komplexa arbetsflöden med streaming och asynkron exekvering. De 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

async def generate_story(topic):
    llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
    prompt = PromptTemplate(
        input_variables=["topic"],
        template="Write a short story about {topic}."
    )
    chain = LLMChain(llm=llm, prompt=prompt)
    return await chain.arun(topic=topic)

async def main():
    topics = ["a magical forest", "a futuristic city", "an underwater civilization"]
    tasks = [generate_story(topic) for topic in topics]
    stories = await asyncio.gather(*tasks)
    
    for topic, story in zip(topics, stories):
        print(f"\nTopic: {topic}\nStory: {story}\n{'='*50}\n")

asyncio.run(main())

Betjänar Async LLM-applikationer med FastAPI

För att göra din async LLM-applikation tillgänglig som en webbtjänst är FastAPI ett utmärkt val på grund av dess inbyggda stöd för asynkrona operationer. Här är ett exempel på hur man skapar en enkel API-slutpunkt för textgenerering:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

class GenerationRequest(BaseModel):
    prompt: str

class GenerationResponse(BaseModel):
    generated_text: str

@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
    
    # Simulate some post-processing in the background
    background_tasks.add_task(log_generation, request.prompt, generated_text)
    
    return GenerationResponse(generated_text=generated_text)

async def log_generation(prompt: str, generated_text: str):
    # Simulate logging or additional processing
    await asyncio.sleep(2)
    print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Denna FastAPI-applikation skapar en slutpunkt /generate som accepterar en uppmaning och returnerar genererad text. Den visar också hur man använder bakgrundsuppgifter för ytterligare bearbetning utan att blockera svaret.

Bästa praxis och vanliga fallgropar

När du arbetar med asynkrona LLM API:er bör du ha dessa bästa metoder i åtanke:

1. Använd anslutningspoolning: När du gör flera förfrågningar, återanvänd anslutningar för att minska omkostnader.
2. Implementera korrekt felhantering: Ta alltid hänsyn till nätverksproblem, API-fel och oväntade svar.
3. Respektera hastighetsgränser: Använd semaforer eller andra samtidighetskontrollmekanismer för att undvika att överväldiga API:et.
4. Övervaka och logga: Implementera omfattande loggning för att spåra prestanda och identifiera problem.
5. Använd streaming för långformat innehåll: Det förbättrar användarupplevelsen och möjliggör tidig bearbetning av delresultat.