Używamy cookies, żeby zwiększyć Twoje doświadczenia na stronie

Logowanie/Rejestracja->

Production Deployment - wdrożenie AI

Kod z poprzedniej lekcji działa u Ciebie w terminalu i nikt poza Tobą nie ma do niego dostępu. Wdrożenie to zamiana notatnika terenowego na stację radiową: liczy się nie tylko to, że nadaje, ale też ile kosztuje i czy widać, gdy zamilknie.

Endpoint /chat - najprostsze wystawienie modelu

FastAPI opisuje endpoint dekoratorem

@app.post

, a kształt danych wejściowych i wyjściowych - klasą Pydantic

BaseModel

. Dzięki temu nie sprawdzasz ręcznie, czy klient przysłał pole

message

: biblioteka sama odrzuci błędne zapytanie.

1from fastapi import FastAPI
2from pydantic import BaseModel
3from openai import AsyncOpenAI
4
5app = FastAPI()
6client = AsyncOpenAI()
7
8class ChatRequest(BaseModel):
9    message: str
10
11class ChatResponse(BaseModel):
12    response: str
13
14@app.post("/chat", response_model=ChatResponse)
15async def chat(request: ChatRequest):
16    completion = await client.chat.completions.create(
17        model="gpt-5",
18        messages=[{"role": "user", "content": request.message}]
19    )
20    return ChatResponse(response=completion.choices[0].message.content)

ChatRequest

opisuje, co przychodzi,

ChatResponse

- co wychodzi, a

response_model

pilnuje, żeby odpowiedź miała ten kształt. Klasa

AsyncOpenAI

pozwala użyć

await

, więc serwer w czasie oczekiwania obsługuje innych klientów. Instrukcja

return

oddaje całą odpowiedź naraz, więc użytkownik kilkanaście sekund patrzy w pustkę.

Streaming: StreamingResponse i generator asynchroniczny

Zwykłego

return

nie da się namówić na streaming. To nie kwestia ustawienia, tylko definicji:

return

kończy funkcję i oddaje wartość, więc wartość musi już być kompletna. Nie pomogą pliki tymczasowe - zapis następuje i tak po otrzymaniu całego tekstu. Da się to jednak zrobić: potrzebujesz funkcji oddającej tekst kawałkami i klasy

StreamingResponse

, która opakuje ją w odpowiedź HTTP.

1from fastapi.responses import StreamingResponse
2
3@app.post("/chat/stream")
4async def chat_stream(request: ChatRequest):
5    async def generate():
6        stream = await client.chat.completions.create(
7            model="gpt-5",
8            messages=[{"role": "user", "content": request.message}],
9            stream=True
10        )
11        async for chunk in stream:
12            content = chunk.choices[0].delta.content
13            if content:
14                yield f"data: {content}\n\n"
15        yield "data: [DONE]\n\n"
16
17    return StreamingResponse(generate(), media_type="text/event-stream")

Cała różnica siedzi w słowie

yield

. Funkcja

generate

nie zwraca wyniku i nie kończy się - oddaje fragment i zawiesza się, aż ktoś poprosi o następny. Taka funkcja to generator asynchroniczny, a

StreamingResponse

wysyła jej fragmenty klientowi na bieżąco.

Dwie ciche awarie warto zapamiętać, @name, bo żadna nie zgłasza błędu. Bez warunku

if content:

w środku strumienia wyląduje tekst

None

, bo skrajne fragmenty mają puste

delta.content

. A bez końcowego

yield

[DONE]

klient kręci wskaźnik ładowania mimo skończonej pracy serwera.

WebSocket - kiedy jeden kierunek to za mało

Streaming rozwiązuje połowę problemu: serwer mówi dłużej, ale wciąż tylko w odpowiedzi na pytanie. HTTP działa w schemacie zapytanie-odpowiedź i serwer nie ma jak odezwać się pierwszy - a przy chatbocie chcesz wysyłać powiadomienia albo wynik zadania, o które nikt akurat nie pyta.

WebSocket utrzymuje jedno otwarte połączenie, w którym obie strony wysyłają wiadomości w dowolnym momencie - to właśnie dwukierunkowa komunikacja w czasie rzeczywistym. I to jedyny prawdziwy powód, żeby po niego sięgnąć. Nie jest "szybszy niż HTTP": jedzie po tym samym TCP. Nie jest bezpieczniejszy -

ws://

idzie otwartym tekstem, a

wss://

używa tego samego TLS co

https

. I na pewno nie jest tak, że nie wymaga serwera: wymaga go bardziej, bo ten trzyma połączenie otwarte przez całą rozmowę.

1from fastapi import WebSocket, WebSocketDisconnect
2
3active_connections: dict[str, WebSocket] = {}
4
5@app.websocket("/ws/{client_id}")
6async def websocket_endpoint(websocket: WebSocket, client_id: str):
7    await websocket.accept()
8    active_connections[client_id] = websocket
9    try:
10        while True:
11            question = await websocket.receive_text()
12            answer = await ask_model(question)
13            await websocket.send_text(answer)
14    except WebSocketDisconnect:
15        del active_connections[client_id]

Funkcja

ask_model

to wywołanie modelu z pierwszego przykładu, a pętla

while True

obsługuje kolejne wiadomości bez rozłączania. Słownik

active_connections

trzyma połączenia po identyfikatorze klienta - dzięki temu wyślesz komuś wiadomość nawet wtedy, gdy o nic nie pytał. Pominięcie usunięcia wpisu przy
WebSocketDisconnect
to wyciek pamięci: serwer działa tygodniami, a słownik puchnie od martwych wpisów.

Cache i limity - żeby rachunek nie rósł w nieskończoność

Każde wywołanie modelu to osobna opłata za tokeny: gdy sto osób zapyta o to samo, zapłacisz sto razy za identyczną odpowiedź. Caching zmniejsza koszty API i przyspiesza odpowiedzi, bo zapamiętany wynik wraca w milisekundach, bez ruszania sieci.

1from cachetools import TTLCache
2from slowapi import Limiter
3from slowapi.util import get_remote_address
4
5limiter = Limiter(key_func=get_remote_address)
6response_cache = TTLCache(maxsize=1000, ttl=3600)
7
8@app.post("/chat/cached")
9@limiter.limit("10/minute")
10async def chat_cached(request: ChatRequest):
11    key = request.message.strip().lower()
12    if key in response_cache:
13        return {"response": response_cache[key], "cached": True}
14
15    response_cache[key] = await ask_model(request.message)
16    return {"response": response_cache[key], "cached": False}

TTLCache

sam usuwa wpisy po godzinie (

ttl=3600

), więc odpowiedzi nie zestarzeją się na dobre, a kluczem jest treść pytania - dwa identyczne pytania trafiają w ten sam wpis.

Warto jednak wiedzieć, że cache nie poprawia jakości odpowiedzi - zwraca ten sam tekst, który padł wcześniej, więc słaba odpowiedź wróci równie słaba, tylko szybciej. Nie jest też wymagany przez żadne prawo, to Twoja decyzja o kosztach. Dekorator

@limiter.limit("10/minute")

z biblioteki

slowapi

ogranicza zapytania z jednego adresu, żeby jeden klient nie wyczerpał budżetu w kwadrans.

Zanim wystawisz to na świat

Przed wdrożeniem LLM do produkcji najważniejsze są trzy rzeczy: testowanie, monitoring i zabezpieczenie kluczy API. Klucz musi przyjść ze zmiennej środowiskowej, nigdy z kodu - wpisany na stałe trafia do repozytorium, a stamtąd na cudzy rachunek. Monitoring zaczyna się od endpointu

/health

, po którym hosting poznaje, że proces jeszcze żyje.

1import logging
2import os
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger("safari-ai")
6
7client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
8
9@app.get("/health")
10async def health_check():
11    logger.info("health ok")
12    return {"status": "healthy"}

Odczyt

os.environ["OPENAI_API_KEY"]

ma dodatkową zaletę: gdy zmiennej brakuje, aplikacja przerywa start, zamiast przyjmować ruch i dopiero przy pierwszym pytaniu zwrócić błąd. Wywołanie modelu warto opakować w

try

, żeby awaria dostawcy wróciła jako

HTTPException

, a

logger.exception

zapisał ślad do logów.

Trzy pokusy, którym trzeba się oprzeć. Najnowszy model bez testowania wygląda niewinnie, bo nazwa modelu to zwykły napis - tyle że nowy model inaczej formatuje odpowiedzi i cicho psuje parsowanie po Twojej stronie. Wyłączanie logowania dla szybkości oszczędza ułamek milisekundy, a odbiera jedyne źródło wiedzy o tym, co stało się w nocy. API bez limitów kończy się rachunkiem: klucz jest Twój, więc cudze zapytania też opłacasz Ty.

Podsumowanie

Streaming w FastAPI implementujesz przez
```
StreamingResponse
```
z generatorem asynchronicznym, czyli funkcją używającą
```
yield
```
.
Zwykły
```
return
```
streamingu nie daje, bo oddaje kompletną wartość i kończy funkcję. Pliki tymczasowe też nie pomagają, a zadanie jest wykonalne.
WebSocket wybierasz dlatego, że pozwala na dwukierunkową komunikację w czasie rzeczywistym. Nie jest szybszy niż HTTP, nie jest bezpieczniejszy i na pewno wymaga serwera trzymającego otwarte połączenie.
Endpoint
```
/chat
```
to
```
@app.post
```
i modele Pydantic:
```
ChatRequest
```
na wejściu,
```
ChatResponse
```
na wyjściu przez
```
response_model
```
.
Przed wdrożeniem LLM do produkcji najważniejsze są testowanie, monitoring i zabezpieczenie kluczy API w zmiennych środowiskowych. Najnowszy model bez testowania, wyłączone logowanie i API bez limitów to trzy najprostsze sposoby na zepsucie produkcji.
Caching zmniejsza koszty API i przyspiesza odpowiedzi. Nie poprawia ich jakości i nie jest wymagany przez prawo.

W następnej lekcji poznasz wzorzec ReAct, w którym agent na przemian myśli i działa, powtarzając cykl Thought - Action - Observation. Na razie zapamiętaj: wdrożenie to nie ostatni commit, tylko pierwszy dzień, gdy ktoś inny płaci Twoim kluczem za Twoje błędy.

Przejdź do CodeWorlds→