AI-mallit ja alustat
Asynkroniset LLM API -kutsut Pythonissa: Kattava opas
Lisää Unite.AI suosikkilähteisiisi Google-palvelussa
Kehittäjienä ja data-tieteilijöinä usein tarvitsemme vuorovaikutusta näiden voimakkaiden mallien kanssa API:n kautta. Kuitenkin sovellusten monimutkaisuuden ja mittakaavan kasvaessa tehokkaiden ja suorituskykyisten API-vuorovaikutusten tarve muodostuu olennaiseksi. Tässä async-programmoinnin avulla voimme maksimoida läpimenoaikaa ja minimoida viivettä LLM API:iden kanssa työskennellessä.
Tässä kattavassa oppaassa tutustumme asynkronisten LLM API -kutsujen maailmaan Pythonissa. Käymme läpi kaiken async-programmoinnin perusteista edistyneisiin tekniikoihin monimutkaisten työvirran käsittelyyn. Oppaan lopussa sinulla on vankka ymmärrys siitä, miten async-programmoinnin avulla voit tehostaa LLM-pohjaisia sovelluksia.
Ennen kuin syventymme asynkronisten LLM API -kutsujen yksityiskohtiin, perustelemme vankan perustan async-programmoinnin käsitteille.
Asynkroninen ohjelmointi sallii useiden toimintojen suorittamisen rinnakkain ilman pääsuoritussäikeen estoamista. Pythonissa tämä saavutetaan pääasiassa asyncio-moduulin avulla, joka tarjoaa kehyksen rinnakkaisen koodin kirjoittamiseen korutinojen, tapahtumaloopin ja tulevaisuuden avulla.
Avainkäsitteet:
- Korutinit: Funktiot, jotka määritellään async def -avainsanalla ja voidaan keskeyttää ja jatkaa.
- Tapahtumalooppi: Keskinen suoritusmekanismi, joka hallinnoi ja suorittaa asynkronisia tehtäviä.
- Odottavat: Objektit, joita voidaan käyttää await -avainsanalla (korutinit, tehtävät, tulevaisuus).
Tässä on yksinkertainen esimerkki, joka havainnollistaa nämä käsitteet:
import asyncio
<p>async def tervehdi(nimi):
await asyncio.sleep(1) # Simuloi I/O -operaatio
print(f"Hei, {nimi}!")</p>
<p>async def paa():
await asyncio.gather(
tervehdi("Aada"),
tervehdi("Bertta"),
tervehdi("Cecilia")
)</p>
asyncio.run(paa())
Tässä esimerkissä määrittelemme asynkronisen funktion tervehdi, joka simuloimalla I/O -operaation asyncio.sleep() . paa -funktiota käytetään asyncio.gather() suorittamaan useita tervehdyksiä rinnakkain. Vaikka on viivästystä, kaikki kolme tervehdystä tulostetaan noin yhden sekunnin kuluttua, osoittaen asynkronisen suorituksen voiman.
Asynkronisten LLM API -kutsujen tarve
Kun työskentelemme LLM API:iden kanssa, kohtaamme usein tilanteita, joissa on tehtävä useita API -kutsuja joko peräkkäin tai rinnakkain. Perinteinen synkroninen koodi voi johtaa merkittäviin suorituskyvyn pullonkauloihin, erityisesti silloin, kun on kyse suurviiveisistä operaatioista, kuten verkkopyyntöjen lähettämisestä LLM -palveluihin.
Oletetaan, että meidän on luotava yhteenvetoja 100:lle eri artikkelille LLM API:n avulla. Synkronisella lähestymistavalla jokainen API -kutsu estäisi suorituksen, kunnes vastaus saadaan, mikä voisi kestää useita minuutteja. Asynkroninen lähestymistapa sallii useiden API -kutsujen aloittamisen rinnakkain, mikä vähentää merkittävästi koko suoritusajan.
Ympäristön määrittäminen
Asynkronisten LLM API -kutsujen käyttämiseksi sinun on asennettava Python -ympäristösi tarvittavilla kirjastoilla. Tässä on mitä tarvitset:
- Python 3.7 tai uudempi (alkuperäinen asyncio -tuki)
- aiohttp: Asynkroninen HTTP -asiakas kirjasto
- openai: Virallinen OpenAI Python -asiakas (jos käytät OpenAI:n GPT -malleja)
- langchain: Keinotekoisen älykkyyden sovellusten kehitykseen tarkoitettu kehys (vapaaehtoinen, mutta suositeltava monimutkaisten työvirtojen käsittelyyn)
Voit asentaa nämä riippuvuudet pip:in avulla:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Perussääntöiset asynkroniset LLM API -kutsut asyncio: n ja aiohttp: n avulla
Aloita yksinkertaisella asynkronisella API -kutsulla LLM: ään aiohttp: n avulla. Käytämme OpenAI:n GPT-3.5 API: ä esimerkkinä, mutta käsitteet ovat sovellettavissa muihin LLM API: iin.
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def luo_teksti(viesti, asiakas):
vastaus = await asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": viesti}]
)
return vastaus.valinnat[0].viesti.sisalto</p>
<p>async def paa():
viestit = [
"Selita kvanttilaskenta yksinkertaisilla sanoilla.",
"Kirjoita haiku tekolylle.",
"Kuvaa fotosynteesin prosessi."
]</p>
<p>async with AsyncOpenAI() as asiakas:
tehtavat = [luo_teksti(viesti, asiakas) for viesti in viestit]
tulokset = await asyncio.gather(*tehtavat)</p>
<p>for viesti, tulos in zip(viestit, tulokset):
print(f"Viesti: {viesti}\nVastaus: {tulos}\n")</p>
asyncio.run(paa())
Tässä esimerkissä määrittelemme asynkronisen funktion luo_teksti, joka tekee kutsun OpenAI API:lle AsyncOpenAI -asiakkaan avulla. paa -funktio luo useita tehtäviä eri viesteille ja käyttää asyncio.gather() suorittaakseen ne rinnakkain.
Tämä lähestymistapa sallii useiden pyyntöjen lähettämisen LLM API:lle samanaikaisesti, mikä vähentää merkittävästi koko käsittelyaikaa.
Edistyneet tekniikat: Eräkäsittely ja rinnakkaisuuden hallinta
Vaikka edellinen esimerkki osoittaa asynkronisten LLM API -kutsujen perusteet, todelliset sovellukset vaativat usein monimutkaisempia lähestymistapoja. Tutustumme kahteen tärkeään tekniikkaan: pyyntöjen eräkäsittelyyn ja rinnakkaisuuden hallintaan.
Eräkäsittely: Kun on kyse suuresta määrästä viestejä, on usein tehokkaampaa käsitellä niitä erissä sen sijaan, että lähetettäisiin yksittäisiä pyyntöjä kullekin viestille. Tämä vähentää pyyntöjen määrää ja johtaa usein parempaan suorituskykyyn.
import asyncio
from openai import AsyncOpenAI
<p>async def prosessoi_era(era, asiakas):
vastaukset = await asyncio.gather(*[
asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": viesti}]
) for viesti in era
])
return [vastaus.valinnat[0].viesti.sisalto for vastaus in vastaukset]</p>
<p>async def paa():
viestit = [f"Kerro minulle jotain numerosta {i}" for i in range(100)]
era_koko = 10</p>
<p>async with AsyncOpenAI() as asiakas:
tulokset = []
for i in range(0, len(viestit), era_koko):
era = viestit[i:i+era_koko]
era_tulokset = await prosessoi_era(era, asiakas)
tulokset.extend(era_tulokset)</p>
<p>for viesti, tulos in zip(viestit, tulokset):
print(f"Viesti: {viesti}\nVastaus: {tulos}\n")</p>
asyncio.run(paa())
Rinnakkaisuuden hallinta: Vaikka asynkroninen ohjelmointi sallii rinnakkaisen suorituksen, on tärkeää hallita rinnakkaisuuden tasoa estääkseen API -palvelimen ylikuormittumisen tai ylittääkseen siirtomäärät. Voimme käyttää asyncio.Semaphorea tähän tarkoitukseen.
import asyncio
from openai import AsyncOpenAI
<p>async def luo_teksti(viesti, asiakas, semafori):
async with semafori:
vastaus = await asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": viesti}]
)
return vastaus.valinnat[0].viesti.sisalto</p>
<p>async def paa():
viestit = [f"Kerro minulle jotain numerosta {i}" for i in range(100)]
max_rinnakkaiset_pyynto = 5
semafori = asyncio.Semaphore(max_rinnakkaiset_pyynto)</p>
<p>async with AsyncOpenAI() as asiakas:
tehtavat = [luo_teksti(viesti, asiakas, semafori) for viesti in viestit]
tulokset = await asyncio.gather(*tehtavat)</p>
<p>for viesti, tulos in zip(viestit, tulokset):
print(f"Viesti: {viesti}\nVastaus: {tulos}\n")</p>
asyncio.run(paa())
Tässä esimerkissä käytetään semaforia rajoittamaan rinnakkaisia pyyntöjä 5:een, estäen API -palvelimen ylikuormittumisen.
Virheiden käsittely ja uudelleenkutsut asynkronisissa LLM -kutsuissa
Kun työskentelet ulkoisten API:iden kanssa, on tärkeää toteuttaa tehokas virheiden käsittely ja uudelleenkutsujen mekanismit. Parannetaan koodiamme käsitelläkseen yleisiä virheitä ja toteuttaakseen eksponentiaalisen takaisinkutsun.
import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
class APIVirhe(Exception):
pass
<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def luo_teksti_uudelleen(viesti, asiakas):
try:
vastaus = await asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": viesti}]
)
return vastaus.valinnat[0].viesti.sisalto
except Exception as e:
print(f"Virhe tapahtui: {e}")
raise APIVirhe("Tekstin luominen epäonnistui")</p>
<p>async def prosessoi_viesti(viesti, asiakas, semafori):
async with semafori:
try:
tulos = await luo_teksti_uudelleen(viesti, asiakas)
return viesti, tulos
except APIVirhe:
return viesti, "Tekstin luominen epäonnistui useita kertoja."</p>
<p>async def paa():
viestit = [f"Kerro minulle jotain numerosta {i}" for i in range(20)]
max_rinnakkaiset_pyynto = 5
semafori = asyncio.Semaphore(max_rinnakkaiset_pyynto)</p>
<p>async with AsyncOpenAI() as asiakas:
tehtavat = [prosessoi_viesti(viesti, asiakas, semafori) for viesti in viestit]
tulokset = await asyncio.gather(*tehtavat)</p>
<p>for viesti, tulos in tulokset:
print(f"Viesti: {viesti}\nVastaus: {tulos}\n")</p>
asyncio.run(paa())
Tässä parannetussa versiossa on:
- Mukautettu
APIVirhe-poikkeus API:hen liittyville virheille. - Funktio
luo_teksti_uudelleen, joka on koristeltu@retry:lla tenacity -kirjastosta, toteuttaen eksponentiaalisen takaisinkutsun. - Virheiden käsittely
prosessoi_viesti-funktiota käyttäen, joka catchaa ja raportoi epäonnistumiset.
Suorituskyvyn optimointi: Vastauksen virtaus
Pitkien sisältöjen luomisessa vastauksen virtaus voi parantaa sovelluksen havaittua suorituskykyä huomattavasti. Sen sijaan, että odottaisit koko vastausta, voit prosessoida ja näyttää tekstinpätkiä, kun ne ovat saatavilla.
import asyncio
from openai import AsyncOpenAI
<p>async def virtaa_teksti(viesti, asiakas):
virta = await asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": viesti}],
virtaa=True
)</p>
<p>tayden_vastaus = ""
async for palanen in virta:
if palanen.valinnat[0].delta.sisalto is not None:
sisalto = palanen.valinnat[0].delta.sisalto
tayden_vastaus += sisalto
print(sisalto, end='', flush=True)</p>
<p>print("\n")
return tayden_vastaus</p>
<p>async def paa():
viesti = "Kirjoita lyhyt tarina aikamatkustavasta tieteilijasta."</p>
<p>async with AsyncOpenAI() as asiakas:
tulos = await virtaa_teksti(viesti, asiakas)</p>
<p>print(f"Tayden vastaus:\n{tulos}")</p>
asyncio.run(paa())
Tässä esimerkissä osoitetaan, miten vastausta voidaan virtata API:sta, tulostamalla jokainen palanen, kun se on saatavilla. Tämä lähestymistapa on erityisen hyödyllinen chat-sovelluksissa tai missä tahansa, kun haluat tarjota käyttäjille reaaliaikaisen palautteen.
Asynkronisten työvirtojen luominen LangChain:in avulla
Monimutkaisempien LLM-pohjaisien sovellusten kohdalla LangChain -kehys tarjoaa korkean tason abstraktion, joka yksinkertaa useiden LLM-kutsujen ketjuttamista ja muiden työkalujen integroimista. Tutustumme esimerkkiin LangChainin käytöstä asynkronisilla ominaisuuksilla:
Tämä esimerkki osoittaa, miten LangChain voidaan käyttää monimutkaisempien työvirtojen luomiseen virtauksen ja asynkronisen suorituksen avulla. AsyncCallbackManager ja StreamingStdOutCallbackHandler mahdollistavat luodun sisällön reaaliaikaisen virtauksen.
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 luo_tarina(aihe):
llm = OpenAI(lämpötila=0.7, virtaa=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
viesti = PromptTemplate(
input_muuttujat=["aihe"],
malli="Kirjoita lyhyt tarina {aihe}sta."
)
ketju = LLMChain(llm=llm, viesti=viesti)
return await ketju.arun(aihe=aihe)</p>
<p>async def paa():
aiheet = ["maaginen metsä", "tulevaisuuden kaupunki", "vesi-alainen sivilisaatio"]
tehtavat = [luo_tarina(aihe) for aihe in aiheet]
tarinat = await asyncio.gather(*tehtavat)</p>
<p>for aihe, tarina in zip(aiheet, tarinat):
print(f"\nAihe: {aihe}\nTarina: {tarina}\n{'='*50}\n")</p>
asyncio.run(paa())
Asynkronisten LLM-sovellusten tarjoaminen FastAPI: n avulla
Jotta voit tehdä asynkronisen LLM-sovelluksesi saataville web-palveluna, FastAPI on erinomainen valinta sen alkuperäisen asynkronisen toiminnan tukemisen vuoksi. Tässä on esimerkki yksinkertaisen API -päätepisteen luomisesta tekstin luontia varten:
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
asiakas = AsyncOpenAI()
<p>class LuontiPyto(BaseModel):
viesti: str</p>
<p>class LuontiVastaus(BaseModel):
luotu_teksti: str</p>
<p>@app.post("/luo", response_model=LuontiVastaus)
async def luo_teksti(pyto: LuontiPyto, taustatehtavat: BackgroundTasks):
vastaus = await asiakas.chat.completions.create(
malli="gpt-3.5-turbo",
viestit=[{"rooli": "kayttaja", "sisaalto": pyto.viesti}]
)
luotu_teksti = vastaus.valinnat[0].viesti.sisalto</p>
<p># Simuloi jokin post-prosessointi taustalla
taustatehtavat.add_task(kirjaa_luonti, pyto.viesti, luotu_teksti)</p>
<p>return LuontiVastaus(luotu_teksti=luotu_teksti)</p>
<p>async def kirjaa_luonti(viesti: str, luotu_teksti: str):
# Simuloi kirjaamista tai lisää prosessointia
await asyncio.sleep(2)
print(f"Kirjattu: Viesti '{viesti}' loi {len(luotu_teksti)} merkin pituisen tekstin")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Tämä FastAPI-sovellus luo päätepisteen /luo, joka hyväksyy viestin ja palauttaa luodun tekstin. Se myös osoittaa, miten taustatehtäviä voidaan käyttää lisäprosessoinnissa ilman, että se estää vastausta.
Parhaat käytännöt ja yleiset ansat
Kun työskentelet asynkronisten LLM API -kutsujen kanssa, pidä seuraavat parhaat käytännöt mielessä:
- Käytä yhteyden jakoa: Kun teet useita pyyntöjä, käytä yhteyden jakoa vähentääksesi yhteyden muodostamiseen liittyvää kuormitusta.
- Toteuta virheiden käsittely: Varmista, että otat huomioon verkkovirheet, API-virheet ja odottamattomat vastaukset.
- Kunnioita siirtomääriä: Käytä semaforeja tai muita rinnakkaisuuden hallintamekanismeja estääksesi API-palvelimen ylikuormittumisen.
- Seuraa ja kirjaa: Toteuta kattava kirjaaminen seurataksesi suorituskykyä ja tunnistamaan ongelmat.
- Käytä virtausta pitkän sisällön kohdalla: Se parantaa käyttäjäkokemusta ja sallii osittaisen tuloksen prosessoinnin.
