FastAPI i LLM: async, streaming i równoległe zapytania do modelu

Gdy Twoja aplikacja wysyła pytanie do modelu językowego, serwer przez większość czasu nic nie liczy. Czeka na odpowiedź z sieci, czasem kilka sekund. Przy jednym użytkowniku to bez znaczenia, ale gdy takich zapytań jest sto naraz, sposób czekania decyduje o tym, czy aplikacja działa, czy się zawiesza. W tym artykule sprawdzimy to pomiarem. Zakładamy, że znasz podstawy z pierwszego artykułu o FastAPI, a samo wywołanie API modelu opisaliśmy w artykule API OpenAI w Pythonie.

Dlaczego FastAPI pasuje do pracy z modelami językowymi?

Zapytanie do modelu to typowe zadanie typu I/O: wysyłasz tekst, a potem czekasz. Framework asynchroniczny może w tym czasie obsługiwać inne zapytania zamiast stać w miejscu. FastAPI jest zbudowany wokół tej idei (standard ASGI), więc async def jest tu podstawowym sposobem pisania endpointów, a nie dodatkiem.

Trzy rzeczy sprawiają, że to dobry wybór do LLM:

• Równoległe oczekiwanie. Jeden proces obsługuje wiele zapytań czekających na model.
• Streaming. Możesz odsyłać odpowiedź kawałkami, w miarę jak model ją generuje.
• Walidacja przez Pydantic. Wejście i wyjście opisujesz modelami, jak w artykule o modelach wejścia i wyjścia.

Nie jest to jedyne rozwiązanie. Flask i Django również pozwalają na kod asynchroniczny, ale FastAPI jest od początku projektowany pod niego. Do porównania wszystkich trzech wrócimy w osobnym artykule.

Ile zapytań naraz obsłuży jeden proces FastAPI?

Sprawdziliśmy to prostym testem. Trzy endpointy udają model, który odpowiada po 2 sekundach:

import asyncio
import time

from fastapi import FastAPI

app = FastAPI()


@app.get("/async")
async def non_blocking():
    await asyncio.sleep(2)  # udaje oczekiwanie na model
    return {"ok": True}


@app.get("/blocking-in-async")
async def blocking_in_async():
    time.sleep(2)  # blokuje pętlę zdarzeń
    return {"ok": True}


@app.get("/sync")
def sync_endpoint():
    time.sleep(2)  # FastAPI uruchomi to w puli wątków
    return {"ok": True}

Następnie wysłaliśmy do każdego z nich kilkadziesiąt, kilkaset i tysiąc zapytań naraz (jeden proces uvicorn, laptop, bez żadnych optymalizacji):

To nie jest test wydajności do porównań między komputerami. Pokazuje proporcje:

• async def z await obsługuje 200 zapytań prawie tak szybko jak 20. Serwer czeka na wszystkie naraz. Przy tysiącu wzrasta czas, bo zaczyna się liczyć narzut samej obsługi zapytań i klienta testowego.
• Zwykłe def jest uruchamiane w puli wątków. Wystarcza przy kilkunastu zapytaniach, ale przy 200 widać kolejkę: pula ma ograniczoną liczbę wątków.
• async def z blokującym kodem to najgorszy przypadek: 20 zapytań po 2 sekundy dało 40 sekund, czyli wykonywanie jedno po drugim.

Liczby z naszego laptopa nie będą takie same na Twoim serwerze, ale kolejność wyników powinna się powtórzyć.

Dlaczego moje API z modelem odpowiada po kolei, a nie równolegle?

To pytanie, które wraca na Stack Overflow co chwilę (jedno z nich ma ok. 99 tysięcy wyświetleń). Przyczyna jest prawie zawsze ta sama: w endpoincie async def wywołujesz coś blokującego. Dla modelu językowego typowym winowajcą jest zwykłe requests.post(...) albo synchroniczny klient biblioteki. Pętla zdarzeń staje, a wszyscy czekają.

Masz dwie poprawne drogi:

• użyj klienta asynchronicznego i await w async def,
• albo zostaw zwykłe def z klientem synchronicznym, ale pamiętaj o limicie wątków.

Nie łącz tych podejść: async def ze zwykłym klientem to powód, dla którego test wyżej trwał 40 sekund zamiast 2.

Jak zbudować endpoint async do modelu?

Poniższy przykład używa asynchronicznego klienta OpenAI. Wywołanie wymaga klucza w zmiennej OPENAI_API_KEY; samo wywołanie opisaliśmy w artykule o API OpenAI.

import asyncio

from fastapi import FastAPI, HTTPException
from openai import APITimeoutError, AsyncOpenAI, RateLimitError
from pydantic import BaseModel, Field

app = FastAPI()
client = AsyncOpenAI(timeout=30)
limit = asyncio.Semaphore(20)
MODEL = "gpt-4o-mini"


class Question(BaseModel):
    text: str = Field(min_length=1, max_length=4000)


@app.post("/ask")
async def ask(question: Question):
    try:
        async with limit:
            response = await client.chat.completions.create(
                model=MODEL,
                messages=[{"role": "user", "content": question.text}],
            )
    except RateLimitError:
        raise HTTPException(status_code=503, detail="Model jest przeciążony, spróbuj za chwilę")
    except APITimeoutError:
        raise HTTPException(status_code=504, detail="Model nie odpowiedział na czas")
    return {"answer": response.choices[0].message.content}

Zwróć uwagę na trzy elementy, które odróżniają ten kod od prostego przykładu:

• AsyncOpenAI zamiast zwykłego klienta, żeby await naprawdę zwalniał pętlę.
• timeout=30, żeby jeden zawieszony model nie trzymał połączenia w nieskończoność.
• asyncio.Semaphore(20), czyli limit jednoczesnych wywołań modelu. O nim w następnej sekcji.

Po co ograniczać liczbę równoległych zapytań do modelu?

Skoro serwer potrafi czekać na tysiące odpowiedzi, dlaczego ustawiamy limit 20? Bo wąskim gardłem szybko przestaje być Twój serwer, a stają się nim limity dostawcy modelu. Gdy wyślesz za dużo zapytań naraz, usłyszysz błąd 429 (RateLimitError), a każde wywołanie kosztuje. Limit chroni Cię przed oboma problemami.

Sprawdziliśmy to z atrapą klienta, który odpowiada po sekundzie (jak w artykule o testowaniu kodu z API modelu):

Z limitem 20 zapytania ustawiają się w kolejkę po 20 sztuk, więc ostatnie czeka 5 sekund. To nie błąd, tylko świadomy kompromis: wolisz, żeby część użytkowników chwilę poczekała, niż żeby dostawca odrzucił zapytania. Wartość limitu dobierasz do swojego planu u dostawcy i do budżetu.

Błąd 429 zamieniamy w odpowiedź 503, bo klient powinien wiedzieć, że problem jest chwilowy i warto spróbować ponownie. Jak ponawiać zapytania i testować taki kod bez wydawania pieniędzy, opisaliśmy w artykule o testowaniu.

Jak odsyłać odpowiedź modelu na bieżąco (streaming)?

Modele generują tekst token po tokenie, a czekanie na całość przez kilka sekund męczy użytkownika. Streaming pozwala pokazywać odpowiedź w miarę jej powstawania:

from fastapi.responses import StreamingResponse


@app.post("/ask/stream")
async def ask_stream(question: Question):
    async def token_generator():
        async with limit:
            stream = await client.chat.completions.create(
                model=MODEL,
                messages=[{"role": "user", "content": question.text}],
                stream=True,
            )
            async for chunk in stream:
                if chunk.choices and chunk.choices[0].delta.content:
                    yield chunk.choices[0].delta.content

    return StreamingResponse(token_generator(), media_type="text/plain")

Co tu się dzieje:

• stream=True sprawia, że klient zwraca strumień fragmentów zamiast jednej odpowiedzi.
• Funkcja z yield jest generatorem asynchronicznym, który oddaje kolejne fragmenty tekstu.
• StreamingResponse wysyła je do przeglądarki od razu, bez czekania na koniec.
• Sprawdzenie chunk.choices chroni przed pustymi fragmentami, które model potrafi wysłać na początku i na końcu.

Przy rzeczywistym frontendzie często używa się formatu Server-Sent Events (text/event-stream), który przeglądarki obsługują wbudowanym EventSource. Wymaga on, by każda porcja miała postać data: ... zakończoną pustą linią. Dla nauki wystarczy zwykły tekst, a format SSE możesz dodać później.

Jakie błędy mentor poprawia najczęściej?

Uwaga mentora: najczęstszy powód, dla którego API z modelem „muli” przy kilku użytkownikach, to zwykłe requests.post wewnątrz async def. Kod działa przy jednym zapytaniu i ujawnia problem dopiero przy dziesięciu. Zamień klienta na asynchronicznego albo przenieś endpoint do zwykłego def.

Druga uwaga: zanim zaczniesz optymalizować serwer, ustaw timeout i limit równoległych wywołań. Bez nich jeden zawieszony model potrafi zablokować wszystkie połączenia.

Co dalej?

Masz endpoint asynchroniczny, streaming, limit równoległych wywołań i obsługę błędów dostawcy. Dalsze kroki to kolejka zadań dla długich operacji, zapamiętywanie rozmów w bazie i ponawianie zapytań z opóźnieniem. W następnych artykułach cyklu porównamy FastAPI z Flaskiem i Django.

Jeśli chcesz uczyć się backendu w Pythonie z mentorem, który ocenia Twój kod, sprawdź kurs Python z certyfikatem.