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ć 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 z [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.

1# ===========================================
2# Safari Lab: FastAPI + LLM Integration
3# ===========================================
4# Build a REST API that connects to an LLM
5# to serve safari wildlife information.
6
7from fastapi import FastAPI, HTTPException
8from pydantic import BaseModel
9from openai import OpenAI
10
11app = FastAPI(title="Safari AI Guide API")
12client = OpenAI(api_key="your-api-key-here")
13
14# --- Request/Response models ---
15class AnimalQuery(BaseModel):
16    animal: str
17    question: str
18    max_tokens: int = 150
19
20class GuideResponse(BaseModel):
21    animal: str
22    answer: str
23    tokens_used: int
24
25# --- Safari guide endpoint ---
26@app.post("/ask-guide", response_model=GuideResponse)
27async def ask_safari_guide(query: AnimalQuery):
28    """Ask the safari guide AI about any animal."""
29    try:
30        response = client.chat.completions.create(
31            model="gpt-4",
32            messages=[
33                {
34                    "role": "system",
35                    "content": "You are Darwin, an expert safari guide. "
36                               "Give concise, factual answers about wildlife."
37                },
38                {
39                    "role": "user",
40                    "content": f"About the {query.animal}: {query.question}"
41                }
42            ],
43            max_tokens=query.max_tokens,
44            temperature=0.7
45        )
46
47        return GuideResponse(
48            animal=query.animal,
49            answer=response.choices[0].message.content,
50            tokens_used=response.usage.total_tokens
51        )
52    except Exception as e:
53        raise HTTPException(status_code=500, detail=str(e))
54
55# --- Health check endpoint ---
56@app.get("/health")
57async def health_check():
58    return {"status": "active", "service": "Safari AI Guide"}
59
60# --- List available species ---
61@app.get("/species")
62async def list_species():
63    return {
64        "species": [
65            "Lion", "Elephant", "Giraffe", "Zebra",
66            "Cheetah", "Hippo", "Rhino", "Leopard"
67        ]
68    }
69
70# TODO: Add an endpoint for batch animal queries
71# class BatchQuery(BaseModel):
72#     animals: list[str]
73#     question: str
74#
75# @app.post("/ask-guide/batch")
76# async def batch_query(query: BatchQuery):
77#     results = []
78#     for animal in query.animals:
79#         single_query = AnimalQuery(animal=animal, question=query.question)
80#         result = await ask_safari_guide(single_query)
81#         results.append(result)
82#     return {"results": results}
83
84# TODO: Add a streaming endpoint
85# from fastapi.responses import StreamingResponse
86#
87# @app.post("/ask-guide/stream")
88# async def stream_guide(query: AnimalQuery):
89#     async def generate():
90#         stream = client.chat.completions.create(
91#             model="gpt-4",
92#             messages=[{"role": "user", "content": query.question}],
93#             stream=True
94#         )
95#         for chunk in stream:
96#             if chunk.choices[0].delta.content:
97#                 yield chunk.choices[0].delta.content
98#     return StreamingResponse(generate(), media_type="text/plain")
99
100print("=== Safari AI Guide API ===")
101print("Endpoints:")
102print("  POST /ask-guide - Ask about an animal")
103print("  GET  /health    - Health check")
104print("  GET  /species   - List species")
105print("\nRun with: uvicorn main:app --reload")
106