Moduł 1 · Wprowadzenie i architektura API z AI
Projektowanie API dla aplikacji AI
Gdy tworzysz API dla aplikacji z integracją sztucznej inteligencji, stoisz przed zupełnie innymi wyzwaniami niż w tradycyjnych aplikacjach webowych. Asynchroniczne przetwarzanie, zarządzanie kosztami tokenów i nieprzewidywalność odpowiedzi AI to tylko początek listy rzeczy, o których musisz pomyśleć już na etapie projektowania.
Specyfika API z integracją AI
Główne różnice, które musisz wziąć pod uwagę:
- Czas odpowiedzi — zapytania mogą trwać od kilku sekund do kilku minut
- Koszty operacji — każde wywołanie kosztuje pieniądze (tokeny)
- Nieprzewidywalność — AI może zwrócić niespodziewane lub niewłaściwe treści
- Limity zewnętrznych serwisów — rate limiting ze strony providerów AI
- Duże rozmiary danych — długie teksty wejściowe i wyjściowe
Kluczowe aspekty architektury
1. Obsługa długotrwałych operacji
Zapytania do modeli AI nie są błyskawiczne jak typowe zapytania do bazy danych. Twoje API musi być przygotowane na:
- Implementację wzorca async/await — wszystkie wywołania AI powinny być asynchroniczne
- Streaming odpowiedzi dla długich tekstów — użytkownik widzi fragmenty tekstu w miarę generowania
- WebSocket connections dla komunikacji w czasie rzeczywistym
- Queue system dla zadań batch — gdy użytkownik wysyła wiele dokumentów do analizy
Praktyczny przykład: Gdy użytkownik prosi o napisanie długiego artykułu, zamiast czekać 2 minuty na kompletną odpowiedź, możesz zwracać fragmenty tekstu za pomocą Server-Sent Events.
2. Zarządzanie kosztami i limitami
Każde wywołanie AI kosztuje pieniądze. Dlatego musisz:
- Monitorować usage tokenów — zarówno wejściowych (prompt) jak i wyjściowych (completion)
- Implementować rate limiting per użytkownik — np. 100 zapytań dziennie dla darmowych kont
- Cache'ować popularne odpowiedzi — jeśli ktoś pyta "Co to jest JavaScript?" po raz setny
- Optymalizować prompty dla efektywności kosztowej — krótsze prompty = mniejsze koszty
Przykład implementacji limitów:
class AIUsageManager:
def check_user_limits(self, user_id: str) -> bool:
daily_usage = self.get_daily_token_usage(user_id)
user_tier = self.get_user_tier(user_id)
limits = {
"free": 10000,
"premium": 100000,
"enterprise": 1000000
}
return daily_usage < limits[user_tier]3. Bezpieczeństwo i walidacja
AI może generować nieprzewidywalne lub niewłaściwe odpowiedzi:
- Sanityzacja danych wejściowych — sprawdzanie, czy użytkownik nie próbuje "zhackować" prompta
- Content filtering na output — blokowanie treści nieodpowiednich, szkodliwych
- Audit logs wszystkich interakcji — pełna historia zapytań i odpowiedzi
- Szyfrowanie kluczy API — bezpieczne przechowywanie tokenów dostępu
Uwaga bezpieczeństwa: Użytkownicy mogą próbować "prompt injection" — wstrzykiwania instrukcji do Twojego prompta. Zawsze waliduj i czysć dane wejściowe.
Wzorce projektowe dla API z AI
Repository Pattern
Oddziel logikę dostępu do AI od logiki biznesowej:
class OpenAIRepository:
async def generate_text(self, prompt: str, max_tokens: int) -> str:
# Logika komunikacji z OpenAI
pass
async def analyze_sentiment(self, text: str) -> dict:
# Analiza sentymentu
pass
async def summarize_document(self, content: str) -> str:
# Streszczanie dokumentów
passZalety tego podejścia:
- Łatwe testowanie (możesz mockować Repository)
- Możliwość zmiany providera AI bez zmian w logice biznesowej
- Centralne zarządzanie konfiguracją AI
Circuit Breaker
Ochrona przed awariami zewnętrznych serwisów AI:
- Automatic fallback do cache'owanych odpowiedzi
- Graceful degradation funkcjonalności — gdy AI nie działa, aplikacja dalej działa w ograniczonym zakresie
- Health checks dla endpoints AI — regularne sprawdzanie dostępności
class AICircuitBreaker:
def __init__(self, failure_threshold=5, timeout=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.last_failure_time = None
self.timeout = timeout
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
async def call_ai_service(self, request):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
else:
return self.get_fallback_response()
try:
response = await self.ai_service.call(request)
self.reset_failure_count()
return response
except Exception:
self.record_failure()
raiseStrategy Pattern
Wymiana różnych providerów AI w zależności od potrzeb:
- OpenAI GPT-4 dla kreatywnego pisania
- Claude dla analizy długich dokumentów
- Modele lokalne dla danych wrażliwych (RODO, tajemnica lekarska)
class AIStrategy:
def generate_response(self, prompt: str) -> str:
raise NotImplementedError
class OpenAIStrategy(AIStrategy):
def generate_response(self, prompt: str) -> str:
# Implementacja dla OpenAI
pass
class ClaudeStrategy(AIStrategy):
def generate_response(self, prompt: str) -> str:
# Implementacja dla Claude
pass
class LocalModelStrategy(AIStrategy):
def generate_response(self, prompt: str) -> str:
# Implementacja dla modelu lokalnego
passPraktyczne wskazówki implementacyjne
Monitoring i metryki
Śledź kluczowe wskaźniki:
- Średni czas odpowiedzi AI
- Koszty tokenów per użytkownik/endpoint
- Rate limiting hits — jak często użytkownicy przekraczają limity
- Error rates — awarie providerów AI
- Cache hit ratio — ile zapytań obsłużył cache
Testowanie API z AI
- Mockuj odpowiedzi AI w testach jednostkowych
- Używaj deterministycznych promptów w testach integracyjnych
- Testuj edge cases — długie teksty, znaki specjalne, różne języki
- Symuluj awarie zewnętrznych serwisów
Podsumowanie
Projektowanie API z integracją AI wymaga myślenia o skalowalności, kosztach i niezawodności już od pierwszego dnia. Kluczem do sukcesu jest implementacja odpowiednich wzorców projektowych, systemów cache'owania i fallbacków, które zapewnią Ci stabilne działanie nawet gdy zewnętrzne serwisy AI mają problemy.
Następny krok: Przećwicz implementację prostego API z integracją OpenAI, uwzględniając rate limiting i obsługę błędów. Zacznij od jednego endpointu i stopniowo dodawaj kolejne funkcjonalności.