Modely a platformy AI
Asynchronní LLM API volání v Pythonu: Komplexní průvodce
Přidejte Unite.AI mezi své preferované zdroje na Google
Jako vývojáři a datoví vědci se často setkáváme s potřebou interagovat s těmito mocnými modely prostřednictvím API. Avšak s růstem složitosti a rozsahu našich aplikací se potřeba efektivní a performantní interakce s API stává kritickou. Zde se ukazuje asynchronní programování, které nám umožňuje maximalizovat propustnost a minimalizovat latenci při práci s LLM API.
V tomto komplexním průvodci prozkoumáme svět asynchronních LLM API volání v Pythonu. Pokryjeme vše od základů asynchronního programování až po pokročilé techniky pro zpracování složitých pracovních postupů. Na konci tohoto článku budete mít pevné pochopení toho, jak využít asynchronní programování k posílení vašich LLM-poháněných aplikací.
Než se ponoříme do specifik asynchronních LLM API volání, založme pevný základ v konceptech asynchronního programování.
Asynchronní programování umožňuje provádět více operací současně bez blokování hlavního vlákna provádění. V Pythonu je toto primárně dosaženo prostřednictvím modulu asyncio, který poskytuje rámec pro psaní konkurujících kódů pomocí korepetitorů, událostí a budoucností.
Klíčové koncepty:
- Korepetitory: Funkce definované pomocí async def, které lze pozastavit a obnovit.
- Událost: Centrální mechanismus provádění, který spravuje a spouští asynchronní úkoly.
- Budoucnosti: Objekty, které lze použít s klíčovým slovem await (korepetitory, úkoly, budoucnosti).
Zde je jednoduchý příklad, který ilustruje tyto koncepty:
import asyncio
<p>async def pozdrav(jmeno):</p>
<p>await asyncio.sleep(1) # Simulujte I/O operaci</p>
<p>print(f"Ahoj, {jmeno}!")</p>
<p>async def main():</p>
<p>await asyncio.gather(</p>
<p>pozdrav("Alice"),</p>
<p>pozdrav("Bob"),</p>
<p>pozdrav("Charlie")</p>
<p>)</p>
asyncio.run(main())
V tomto příkladu definujeme asynchronní funkci pozdrav, která simuluje I/O operaci pomocí asyncio.sleep(). Funkce main používá asyncio.gather() k současném spuštění více pozdravů. Přes zpoždění způsobené asyncio.sleep() budou všechny tři pozdravy vytisknuty po zhruba jedné sekundě, což demonstruje sílu asynchronního provádění.
Potřebnost asynchronních LLM API volání
Při práci s LLM API se často setkáváme s scénáři, kdy je třeba provést více API volání, buď v pořadí nebo paralelně. Tradiční synchronní kód může vést k významným výkonovým úzkým místům, zejména při zpracování operací s vysokou latencí, jako jsou síťové požadavky na LLM služby.
Zvažte scénář, kdy je třeba vygenerovat souhrny pro 100 různých článků pomocí LLM API. Synchronní přístup by vedl k tomu, že by každé API volání blokovalo, dokud by neobdrželo odpověď, potenciálně trvající několik minut, aby se dokončily všechny požadavky. Asynchronní přístup umožňuje iniciovat více API volání současně, což dramaticky snižuje celkovou dobu provádění.
Nastavení prostředí
Abyste mohli začít s asynchronními LLM API voláními, musíte nastavit své Pythonové prostředí s nezbytnými knihovnami. Zde je to, co budete potřebovat:
- Python 3.7 nebo vyšší (pro nativní podporu asyncio)
- aiohttp: Asynchronní HTTP klient knihovna
- openai: Oficiální OpenAI Python klient (pokud používáte OpenAI GPT modely)
- langchain: Rámec pro budování aplikací s LLM (volitelný, ale doporučený pro složitější pracovní postupy)
Tyto závislosti můžete nainstalovat pomocí pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Základní asynchronní LLM API volání s asyncio a aiohttp
Začneme jednoduchým asynchronním voláním LLM API pomocí aiohttp. Použijeme OpenAI GPT-3.5 API jako příklad, ale koncepty platí i pro jiná LLM API.
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def vygeneruj_text(prompt, client):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>async def main():</p>
<p>prompts = [</p>
<p>"Explikujte kvantové počítání v jednoduchých termínech.",</p>
<p>"Napište haiku o umělém inteligenci.",</p>
<p>"Popište proces fotosyntézy."</p>
<p>]</p>
<p>async with AsyncOpenAI() as client:</p>
<p>úkoly = [vygeneruj_text(prompt, client) for prompt in prompts]</p>
<p>výsledky = await asyncio.gather(*úkoly)</p>
<p>for prompt, výsledek in zip(prompts, výsledky):</p>
<p>print(f"Prompt: {prompt}\nOdpověď: {výsledek}\n")</p>
asyncio.run(main())
Tento přístup umožňuje odeslat více požadavků na LLM API současně, což dramaticky snižuje celkovou dobu zpracování všech promptů.
Pokročilé techniky: Batching a řízení souběžnosti
Zatímco předchozí příklad demonstruje základy asynchronních LLM API volání, reálné aplikace často vyžadují složitější přístupy. Prozkoumejte dvě důležité techniky: batchování požadavků a řízení souběžnosti.
Batchování požadavků: Při zpracování velkého počtu promptů je často efektivnější batchovat je do skupin místo odesílání jednotlivých požadavků pro každý prompt. To snižuje režii více API volání a může vést k lepšímu výkonu.
import asyncio
from openai import AsyncOpenAI
<p>async def zpracuj_batch(batch, client):</p>
<p>responses = await asyncio.gather(*[</p>
<p>client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>) for prompt in batch</p>
<p>])</p>
<p>return [response.choices[0].message.content for response in responses]</p>
<p>async def main():</p>
<p>prompts = [f"Řekněte mi fakt o čísle {i}" for i in range(100)]</p>
<p>batch_size = 10</p>
<p>async with AsyncOpenAI() as client:</p>
<p>výsledky = []</p>
<p>for i in range(0, len(prompts), batch_size):</p>
<p>batch = prompts[i:i+batch_size]</p>
<p>batch_výsledky = await zpracuj_batch(batch, client)</p>
<p>výsledky.extend(batch_výsledky)</p>
<p>for prompt, výsledek in zip(prompts, výsledky):</p>
<p>print(f"Prompt: {prompt}\nOdpověď: {výsledek}\n")</p>
asyncio.run(main())
Řízení souběžnosti: Při asynchronním programování je důležité řídit úroveň souběžnosti, aby se předešlo přetížení API serveru nebo překročení limitů. Můžeme použít asyncio.Semaphore pro tento účel.
import asyncio
from openai import AsyncOpenAI
<p>async def vygeneruj_text(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>async def main():</p>
<p>prompts = [f"Řekněte mi fakt o čísle {i}" for i in range(100)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:</p>
<p>úkoly = [vygeneruj_text(prompt, client, semaphore) for prompt in prompts]</p>
<p>výsledky = await asyncio.gather(*úkoly)</p>
<p>for prompt, výsledek in zip(prompts, výsledky):</p>
<p>print(f"Prompt: {prompt}\nOdpověď: {výsledek}\n")</p>
asyncio.run(main())
V tomto příkladu používáme semaphore k omezení počtu současných požadavků na 5, aby se předešlo přetížení API serveru.
Řízení chyb a opakovaných pokusů v asynchronních LLM voláních
Při práci s externími API je důležité implementovat robustní řízení chyb a opakovaných pokusů. Zlepšeme náš kód, aby zpracovával běžné chyby a implementoval exponenciální backoff pro opakované pokusy.
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))</p>
<p>async def vygeneruj_text_s_opakovanim(prompt, client):</p>
<p>try:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>except Exception as e:</p>
<p>print(f"Chyba nastala: {e}")</p>
<p>raise APIError("Nepodařilo se vygenerovat text")</p>
<p>async def zpracuj_prompt(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>try:</p>
<p>výsledek = await vygeneruj_text_s_opakovanim(prompt, client)</p>
<p>return prompt, výsledek</p>
<p>except APIError:</p>
<p>return prompt, "Nepodařilo se vygenerovat odpověď po opakovaných pokusech."</p>
<p>async def main():</p>
<p>prompts = [f"Řekněte mi fakt o čísle {i}" for i in range(20)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:</p>
<p>úkoly = [zpracuj_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p>výsledky = await asyncio.gather(*úkoly)</p>
<p>for prompt, výsledek in výsledky:</p>
<p>print(f"Prompt: {prompt}\nOdpověď: {výsledek}\n")</p>
asyncio.run(main())
Tato vylepšená verze zahrnuje:
- Vlastní výjimku
APIErrorpro chyby související s API. - Funkci
vygeneruj_text_s_opakovanimdekorovanou pomocí@retryz knihovny tenacity, která implementuje exponenciální backoff. - Řízení chyb ve funkci
zpracuj_promptpro zachycení a报告 selhání.
Optimalizace výkonu: Streamování odpovědí
Pro generování dlouhých textů může streamování odpovědí výrazně zlepšit vnímaný výkon vaší aplikace. Místo čekání na celou odpověď můžete zpracovávat a zobrazovat části textu, jakmile jsou k dispozici.
import asyncio
from openai import AsyncOpenAI
<p>async def stream_text(prompt, client):</p>
<p>stream = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}],</p>
<p>stream=True</p>
<p>)</p>
<p>celá_odezva = ""</p>
<p>async for chunk in stream:</p>
<p>if chunk.choices[0].delta.content is not None:</p>
<p>obsah = chunk.choices[0].delta.content</p>
<p>celá_odezva += obsah</p>
<p>print(obsah, end='', flush=True)</p>
<p>print("\n")</p>
<p>return celá_odezva</p>
<p>async def main():</p>
<p>prompt = "Napište krátký příběh o cestujícím v čase."</p>
<p>async with AsyncOpenAI() as client:</p>
<p>výsledek = await stream_text(prompt, client)</p>
<p>print(f"Celá odpověď:\n{výsledek}")</p>
asyncio.run(main())
Tento příklad demonstruje, jak streamovat odpověď z API, tisknoucí každou část, jakmile je k dispozici. Tento přístup je zvláště užitečný pro chatové aplikace nebo jakékoli scénáře, kde chcete poskytnout uživatelům okamžitou zpětnou vazbu.
Budování asynchronních pracovních postupů s LangChain
Pro složitější LLM-poháněné aplikace poskytuje rámec LangChain vyšší úroveň abstrakce, která zjednodušuje proces řetězení více LLM volání a integrace dalších nástrojů. Prozkoumejte příklad použití LangChain s asynchronními schopnostmi:
Tento příklad ukazuje, jak LangChain lze použít k vytvoření složitějších pracovních postupů se streamováním a asynchronním prováděním. AsyncCallbackManager a StreamingStdOutCallbackHandler umožňují streamování vygenerovaného obsahu v reálném čase.
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 vygeneruj_příběh(téma):</p>
<p>llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p>prompt = PromptTemplate(</p>
<p>input_variables=["téma"],</p>
<p>template="Napište krátký příběh o {téma}."</p>
<p>)</p>
<p>chain = LLMChain(llm=llm, prompt=prompt)</p>
<p>return await chain.arun(téma=téma)</p>
<p>async def main():</p>
<p>témata = ["magický les", "budoucnost", "podvodní civilizace"]</p>
<p>úkoly = [vygeneruj_příběh(téma) for téma in témata]</p>
<p>asyncio.gather(*úkoly)</p>
<p>for téma, příběh in zip(témata, await asyncio.gather(*úkoly)):</p>
<p>print(f"\nTéma: {téma}\nPříběh: {příběh}\n{'='*50}\n")</p>
asyncio.run(main())
Serving asynchronních LLM aplikací s FastAPI
Abyste mohli své asynchronní LLM aplikace zpřístupnit jako webové služby, je FastAPI skvělou volbou díky své nativní podpoře asynchronních operací. Zde je příklad, jak vytvořit jednoduchý API endpoint pro generování textu:
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
<p>class GenerationRequest(BaseModel):</p>
<p>prompt: str</p>
<p>class GenerationResponse(BaseModel):</p>
<p>generated_text: str</p>
<p>@app.post("/generate", response_model=GenerationResponse)</p>
<p>async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": request.prompt}]</p>
<p>)</p>
<p>generated_text = response.choices[0].message.content</p>
<p># Simulujte一些 post-processing v pozadí</p>
<p>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):</p>
<p># Simulujte logging nebo další zpracování</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")</p>
<p>if __name__ == "__main__":</p>
<p>import uvicorn</p>
<p>uvicorn.run(app, host="0.0.0.0", port=8000)</p>
Tato FastAPI aplikace vytváří endpoint /generate, který přijímá prompt a vrací vygenerovaný text. Také demonstruje, jak použít background tasks pro další zpracování bez blokování odpovědi.
Nejlepší praktiky a běžné pasti
Při práci s asynchronními LLM API voláními mějte na paměti tyto nejlepší praktiky:
- Použijte connection pooling: Při odesílání více požadavků opakujte použití spojení, aby se snížila režie.
- Implementujte správné řízení chyb: Vždy zohledněte síťové problémy, chyby API a neočekávané odpovědi.
- Respektujte limity: Použijte semafory nebo jiné mechanismy pro řízení souběžnosti, aby se předešlo přetížení API.
- Monitorujte a logujte: Implementujte komplexní logování pro sledování výkonu a identifikaci problémů.
- Použijte streamování pro dlouhý obsah: Zlepšuje uživatelský zážitek a umožňuje zpracování částečných výsledků.
