Wywołania Asynchroniczne API LLM w Pythonie: Kompleksowy Przewodnik

Jako deweloperzy i naukowcy danych, często musimy wchodzić w interakcje z tymi potężnymi modelami za pomocą API. Jednak gdy nasze aplikacje rosną w złożoności i skali, potrzeba wydajnych i efektywnych interakcji API staje się kluczowa. To właśnie tutaj programowanie asynchroniczne błyszczy, pozwalając nam maksymalizować przepływność i minimalizować opóźnienia podczas pracy z API LLM.

W tym kompleksowym przewodniku, będziemy eksplorować świat asynchronicznych wywołań API LLM w Pythonie. Pokryjemy wszystko, od podstaw programowania asynchronicznego po zaawansowane techniki obsługi złożonych przepływów pracy. Po przeczytaniu tego artykułu, będziesz miał solidne zrozumienie, jak wykorzystać programowanie asynchroniczne, aby zwiększyć wydajność swoich aplikacji zasilanych przez LLM.

Przed tym, jak zagłębimy się w szczegóły asynchronicznych wywołań API LLM, ustanówmy solidną podstawę w pojęciach programowania asynchronicznego.

Programowanie asynchroniczne pozwala na wykonywanie wielu operacji jednocześnie bez blokowania głównego wątku wykonywania. W Pythonie, jest to głównie osiągane za pomocą modułu asyncio, który zapewnia ramy dla pisania kodu współbieżnego za pomocą koreutin, pętli zdarzeń i przyszłości.

Kluczowe pojęcia:

• Koreutyny: Funkcje zdefiniowane z async def, które mogą być wstrzymane i wznowione.
• Pętla zdarzeń: Centralny mechanizm wykonywania, który zarządza i uruchamia zadania asynchroniczne.
• Obiekty oczekujące: Obiekty, które mogą być używane z kluczem await (koreutyny, zadania, przyszłości).

Oto prosty przykład, który ilustruje te pojęcia:

import asyncio

<p>async def greet(name):</p>
<p> await asyncio.sleep(1) # Symulacja operacji wejścia/wyjścia</p>
<p> print(f"Cześć, {name}!")</p>

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

asyncio.run(main())

W tym przykładzie, definiujemy funkcję asynchroniczną greet, która symuluje operację wejścia/wyjścia za pomocą asyncio.sleep(). Funkcja main używa asyncio.gather(), aby uruchomić wiele powitań jednocześnie. Pomimo opóźnienia spowodowanego przez asyncio.sleep(), wszystkie trzy powitania zostaną wydrukowane po około 1 sekundzie, demonstrując moc wykonywania asynchronicznego.

Potrzeba Asynchroniczności w Wywołaniach API LLM

Podczas pracy z API LLM, często spotykamy się z sytuacjami, w których musimy wykonać wiele wywołań API, albo sekwencyjnie, albo równolegle. Tradycyjny kod synchroniczny może prowadzić do znacznych wąskich gardeł wydajności, szczególnie podczas radzenia sobie z operacjami o wysokiej latencji, takimi jak żądania sieciowe do usług LLM.

Rozważmy sytuację, w której musimy wygenerować podsumowania dla 100 różnych artykułów za pomocą API LLM. Z podejściem synchronicznym, każde wywołanie API zablokuje się, aż do otrzymania odpowiedzi, potencjalnie zajmując kilka minut, aby ukończyć wszystkie żądania. Podejście asynchroniczne pozwala nam natomiast zainicjować wiele wywołań API jednocześnie, znacznie redukując całkowity czas wykonywania.

Konfigurowanie Środowiska

Aby rozpocząć pracę z asynchronicznymi wywołaniami API LLM, musisz skonfigurować swoje środowisko Pythona z niezbędnymi bibliotekami. Oto, co będziesz potrzebować:

• Python 3.7 lub nowszy (dla natywnego wsparcia asyncio)
• aiohttp: Asynchroniczna biblioteka klienta HTTP
• openai: Oficjalna biblioteka klienta OpenAI dla Pythona (jeśli używasz modeli GPT OpenAI)
• langchain: Ramy dla budowania aplikacji z LLM (opcjonalnie, ale zalecane dla złożonych przepływów pracy)

Możesz zainstalować te zależności za pomocą pip:

<p>pip install aiohttp openai langchain
&lt;div class=&quot;relative flex flex-col rounded-lg&quot;&gt;

Podstawowe Asynchroniczne Wywołania API LLM z asyncio i aiohttp

Zacznijmy od wykonania prostego asynchronicznego wywołania do API LLM za pomocą aiohttp. Będziemy używać API GPT-3.5 OpenAI jako przykładu, ale pojęcia te mają zastosowanie do innych API LLM.

import asyncio
import aiohttp
from openai import AsyncOpenAI

<p>async def generate_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> "Wyjaśnij kwantową komputację w prostych słowach.",</p>
<p> "Napisz haiku o sztucznej inteligencji.",</p>
<p> "Opisz proces fotosyntezy."</p>
<p> ]</p>

<p>async with AsyncOpenAI() as client:</p>
<p> tasks = [generate_text(prompt, client) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nOdpowiedź: {result}\n")</p>

asyncio.run(main())

W tym przykładzie, definiujemy funkcję asynchroniczną generate_text, która wykonuje wywołanie do API OpenAI za pomocą klienta AsyncOpenAI. Funkcja main tworzy wiele zadań dla różnych prompty i używa asyncio.gather(), aby uruchomić je jednocześnie.

To podejście pozwala nam wysyłać wiele żądań do API LLM jednocześnie, znacznie redukując całkowity czas przetwarzania.

Zaawansowane Techniki: Partionowanie i Kontrola Współbieżności

Podczas gdy poprzedni przykład demonstruje podstawy asynchronicznych wywołań API LLM, aplikacje rzeczywiste często wymagają bardziej zaawansowanych podejść. Rozważmy dwie ważne techniki: partionowanie żądań i kontrola współbieżności.

Partionowanie Żądań: Podczas radzenia sobie z dużą liczbą prompty, często jest bardziej efektywne partionowanie ich w grupy, zamiast wysyłania poszczególnych żądań dla każdego promptu. To redukuje nakład pracy związany z wieloma wywołaniami API i może prowadzić do lepszej wydajności.

import asyncio
from openai import AsyncOpenAI

<p>async def process_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"Powiedz mi fakt o liczbie {i}" for i in range(100)]</p>
<p> batch_size = 10</p>

<p>async with AsyncOpenAI() as client:</p>
<p> results = []</p>
<p> for i in range(0, len(prompts), batch_size):</p>
<p> batch = prompts[i:i+batch_size]</p>
<p> batch_results = await process_batch(batch, client)</p>
<p> results.extend(batch_results)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nOdpowiedź: {result}\n")</p>

asyncio.run(main())

Kontrola Współbieżności: Podczas gdy programowanie asynchroniczne pozwala na wykonywanie współbieżne, ważne jest kontrolowanie poziomu współbieżności, aby uniknąć przeciążenia serwera API lub przekroczenia limitów stawek. Możemy użyć asyncio.Semaphore do tego celu.

import asyncio
from openai import AsyncOpenAI

<p>async def generate_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"Powiedz mi fakt o liczbie {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> tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nOdpowiedź: {result}\n")</p>

asyncio.run(main())

W tym przykładzie, używamy semafora, aby ograniczyć liczbę współbieżnych żądań do 5, zapewniając, że nie przeciążymy serwera API.

Obsługa Błędów i Ponowienia w Asynchronicznych Wywołaniach LLM

Podczas pracy z zewnętrznymi API, ważne jest wdrożenie solidnej obsługi błędów i mechanizmów ponowienia. Rozważmy, jak możemy udoskonalić nasz kod, aby obsłużyć typowe błędy i wdrożyć wykładnicze ponowienie.

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 generate_text_with_retry(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"Wystąpił błąd: {e}")</p>
<p> raise APIError("Nie udało się wygenerować tekstu")</p>

<p>async def process_prompt(prompt, client, semaphore):</p>
<p> async with semaphore:</p>
<p> try:</p>
<p> result = await generate_text_with_retry(prompt, client)</p>
<p> return prompt, result</p>
<p> except APIError:</p>
<p> return prompt, "Nie udało się wygenerować odpowiedzi po wielu próbach."</p>

<p>async def main():</p>
<p> prompts = [f"Powiedz mi fakt o liczbie {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> tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in results:</p>
<p> print(f"Prompt: {prompt}\nOdpowiedź: {result}\n")</p>

asyncio.run(main())

Ta ulepszona wersja zawiera:

• Niestandardowy wyjątek APIError dla błędów związanych z API.
• Funkcję generate_text_with_retry ozdobioną dekoratorem @retry z biblioteki tenacity, wdrażając wykładnicze ponowienie.
• Obsługę błędów w funkcji process_prompt, aby przechwytywać i zgłaszać niepowodzenia.

Optymalizacja Wydajności: Transmisja Strumieniowa

Dla generacji długich treści, transmisja strumieniowa może znacznie poprawić postrzeganą wydajność Twojej aplikacji. Zamiast czekać na całą odpowiedź, możesz przetwarzać i wyświetlać fragmenty tekstu, gdy stają się one dostępne.

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>full_response = ""</p>
<p> async for chunk in stream:</p>
<p> if chunk.choices[0].delta.content is not None:</p>
<p> content = chunk.choices[0].delta.content</p>
<p> full_response += content</p>
<p> print(content, end="", flush=True)</p>

<p>print("\n")</p>
<p> return full_response</p>

<p>async def main():</p>
<p> prompt = "Napisz krótką historię o podróżującym w czasie naukowcu."</p>

<p>async with AsyncOpenAI() as client:</p>
<p> result = await stream_text(prompt, client)</p>

<p>print(f"Pełna odpowiedź:\n{result}")</p>

asyncio.run(main())

Ten przykład demonstruje, jak transmitować strumieniowo odpowiedź z API, drukując każdy fragment, gdy tylko jest dostępny. To podejście jest szczególnie przydatne w aplikacjach czatu lub w każdej sytuacji, w której chcesz zapewnić użytkownikowi natychmiastową informację zwrotną.

Budowanie Asynchronicznych Przepływów Pracy z LangChain

Dla bardziej złożonych aplikacji zasilanych przez LLM, ramy LangChain zapewniają wysoki poziom abstrakcji, który upraszcza proces łączenia wielu wywołań LLM i integracji z innymi narzędziami. Rozważmy przykład użycia LangChain z możliwościami asynchronicznymi:

Ten przykład pokazuje, jak LangChain może być użyty do tworzenia bardziej złożonych przepływów pracy z transmisją strumieniową i wykonywaniem asynchronicznym. AsyncCallbackManager i StreamingStdOutCallbackHandler umożliwiają transmisję strumieniową wygenerowanego tekstu w czasie rzeczywistym.

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):</p>
<p> llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p> prompt = PromptTemplate(</p>
<p> input_variables=["topic"],</p>
<p> template="Napisz krótką historię o {topic}."</p>
<p> )</p>
<p> chain = LLMChain(llm=llm, prompt=prompt)</p>
<p> return await chain.arun(topic=topic)</p>

<p>async def main():</p>
<p> topics = ["magiczny las", "miasto przyszłości", "podwodna cywilizacja"]</p>
<p> tasks = [generate_story(topic) for topic in topics]</p>
<p> stories = await asyncio.gather(*tasks)</p>

<p>for topic, story in zip(topics, stories):</p>
<p> print(f"\nTemat: {topic}\nHistoria: {story}\n{'='*50}\n")</p>

asyncio.run(main())

Uruchamianie Asynchronicznych Aplikacji LLM z FastAPI

Aby udostępnić swoją asynchroniczną aplikację LLM jako usługę internetową, FastAPI jest doskonałym wyborem ze względu na jego natywne wsparcie dla operacji asynchronicznych. Oto przykład, jak utworzyć prosty punkt końcowy API dla generacji tekstu:

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># Symulacja pewnych post-procesów w tle</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> # Symulacja logowania lub dodatkowej obróbki</p>
<p> await asyncio.sleep(2)</p>
<p> print(f"Zalogowano: Prompt '{prompt}' wygenerował tekst o długości {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>

Ta aplikacja FastAPI tworzy punkt końcowy /generate, który akceptuje prompt i zwraca wygenerowany tekst. Pokazuje również, jak używać zadań w tle do dodatkowej obróbki bez blokowania odpowiedzi.

Najlepsze Praktyki i Powszechne Pułapki

Podczas pracy z asynchronicznymi wywołaniami API LLM, pamiętaj o tych najlepszych praktykach:

1. Używaj puli połączeń: Podczas wykonywania wielu żądań, ponownie używaj połączeń, aby zmniejszyć nakład pracy.
2. Wdrażaj odpowiednią obsługę błędów: Zawsze uwzględniaj problemy z siecią, błędy API i nieoczekiwane odpowiedzi.
3. Szanhuj limity stawek: Używaj semaforów lub innych mechanizmów kontroli współbieżności, aby uniknąć przeciążenia serwera API.
4. Monitoruj i rejestruj: Wdrażaj kompleksywne logowanie, aby śledzić wydajność i identyfikować problemy.
5. Używaj transmisji strumieniowej dla długich treści: Poprawia to doświadczenie użytkownika i pozwala na wczesną obróbkę częściowych wyników.