Monitoring AI na produkcji: metryki, tracing i alerty dla aplikacji LLM

Asystent AI w Twoim produkcie od 3 tygodni zwraca halucynacje na pytania o ceny. Użytkownicy milcząco przestali korzystać z funkcji. Dowiedziałeś się z recenzji w App Store. Bez monitoringu nie wiesz kiedy model zaczął błądzić, jakie pytania są problematyczne ani ile sesji zakończyło się złym wynikiem. Monitoring AI to nie opcja — to warunek działania w produkcji.

Czym różni się monitoring LLM od klasycznego monitoringu

W tradycyjnym software monitorujesz: czy serwer żyje, czy baza odpowiada, ile requestów na sekundę. W aplikacjach LLM te metryki są konieczne, ale niewystarczające. Jakość odpowiedzi jest stochastyczna — ten sam prompt może dawać różne wyniki, a "błąd" to nie wyjątek w kodzie, ale semantycznie niepoprawna odpowiedź. Nie wykryjesz go logiką boolowską.

Pięć kluczowych różnic między monitoringiem LLM a klasycznym:

• Brak deterministycznego "success" — HTTP 200 z halucynacją jest gorszy niż timeout z kodem błędu
• Jakość degraduje się płynnie — nie ma momentu crash, jest powolne, niewidoczne pogorszenie
• Kontekst determinuje poprawność — ta sama odpowiedź może być dobra lub zła zależnie od pytania
• Koszt jest mierzalny w tokenach — każde wywołanie ma konkretną cenę którą można i trzeba śledzić
• Model może się cicho zmienić — OpenAI aktualizuje modele bez ostrzeżenia, zachowanie może dryfować

Trzy warstwy monitoringu AI

Warstwa 1 — Infrastruktura i API

Metryki techniczne, monitorowane tak samo jak w każdej innej aplikacji:

• Latency — p50, p95, p99 end-to-end oraz TTFT (czas do pierwszego tokenu przy streamingu)
• Error rate — błędy API (rate limit, timeout, 5xx) osobno od błędów walidacji danych
• Throughput — zapytania na sekundę i dzienna liczba wywołań per endpoint
• Koszt — tokeny input/output per wywołanie, koszt per użytkownik, trend dzienny
• Rate limit proximity — alert gdy wykorzystujesz 80% limitu minutowego lub dziennego

Warstwa 2 — Jakość odpowiedzi LLM

Metryki specyficzne dla LLM, wymagające ewaluacji — nie można ich zmierzyć kodem:

• Faithfulness — czy odpowiedź bazuje wyłącznie na dostarczonym kontekście (kluczowe dla RAG)
• Answer relevance — czy odpowiedź faktycznie odnosi się do pytania użytkownika
• Hallucination rate — procent odpowiedzi zawierających fakty nieobecne w kontekście
• Completeness — czy wszystkie wymagane elementy odpowiedzi są obecne
• Toxicity / Safety — wykrywanie nieodpowiednich lub niebezpiecznych treści

Warstwa 3 — Wpływ na biznes

Ostateczna miara sukcesu — czy AI faktycznie dostarcza wartość użytkownikom:

• Task completion rate — ile sesji kończy się osiągniętym celem użytkownika
• Thumbs up / down ratio — bezpośredni feedback użytkownika, najtańszy wskaźnik jakości
• Escalation rate — jak często AI przekazuje do człowieka (zbyt wysoko = AI zawodzi)
• Conversion — czy interakcja z AI prowadzi do pożądanej akcji (zakup, rejestracja, kontakt)
• Session abandonment — zbyt wysoki wskaźnik oznacza frustrację i brak odpowiedzi

Tracing — pełny log każdego wywołania LLM

Tracing to zapis pełnego kontekstu wywołania: prompt, odpowiedź, tokeny, latency, model, błędy. Bez tracingu gdy coś pójdzie źle nie masz danych do diagnozy. Każde wywołanie powinno mieć unikalny `trace_id` — pozwala powiązać żądanie użytkownika z wywołaniami LLM, wynikami narzędzi i odpowiedzią końcową w jednym widoku.

Co powinien zawierać log każdego wywołania LLM

Minimalny zestaw pól które musisz logować — każde wywołanie, bez wyjątku:

1. 1.trace_id — unikalny identyfikator sesji lub requestu użytkownika
2. 2.span_id — identyfikator konkretnego kroku (jeden trace może mieć wiele spanów w agencie)
3. 3.timestamp — czas wywołania w UTC z milisekundami
4. 4.model — nazwa modelu wraz z wersją (np. `gpt-4o-2024-08-06`)
5. 5.input_tokens / output_tokens / cached_tokens — z `usage` w odpowiedzi API
6. 6.latency_ms — czas od wysłania żądania do otrzymania pełnej odpowiedzi
7. 7.prompt_hash — hash system message, pozwala wykryć nieautoryzowane zmiany promptu
8. 8.error — kod i treść błędu jeśli wystąpił, `null` przy sukcesie
9. 9.user_id / feature — kontekst biznesowy do filtrowania logów per funkcja

import timeimport hashlibimport uuidimport loggingfrom dataclasses import dataclass, asdictfrom typing import Optionalfrom openai import OpenAIlogger = logging.getLogger(__name__)client = OpenAI()COST_PER_1M = {    "gpt-4o-mini": (0.15, 0.60),    "gpt-4o": (2.50, 10.00),    "o3-mini": (1.10, 4.40)}@dataclassclass LLMTrace:    trace_id: str    model: str    input_tokens: int    output_tokens: int    cached_tokens: int    latency_ms: float    prompt_hash: str    cost_usd: float    error: Optional[str] = None    user_id: Optional[str] = None    feature: Optional[str] = Nonedef traced_call(messages: list, model: str = "gpt-4o-mini", trace_id: str = None, user_id: str = None, feature: str = None) -> tuple[str, LLMTrace]:    trace_id = trace_id or str(uuid.uuid4())    prompt_hash = hashlib.sha256(str(messages[0]).encode()).hexdigest()[:12]    t0 = time.perf_counter()    error = None    content = ""    input_t = output_t = cached_t = 0    try:        resp = client.chat.completions.create(model=model, messages=messages)        content = resp.choices[0].message.content        input_t = resp.usage.prompt_tokens        output_t = resp.usage.completion_tokens        details = getattr(resp.usage, "prompt_tokens_details", None)        cached_t = details.cached_tokens if details else 0    except Exception as e:        error = str(e)    latency_ms = (time.perf_counter() - t0) * 1000    in_c, out_c = COST_PER_1M.get(model, (2.50, 10.00))    cost = (input_t * in_c + output_t * out_c) / 1_000_000    trace = LLMTrace(trace_id=trace_id, model=model, input_tokens=input_t, output_tokens=output_t, cached_tokens=cached_t, latency_ms=round(latency_ms, 1), prompt_hash=prompt_hash, cost_usd=round(cost, 6), error=error, user_id=user_id, feature=feature)    logger.info("llm_trace %s", asdict(trace))    if error:        raise RuntimeError(error)    return content, trace

Output loggera to JSON gotowy do Datadog, Grafana Loki lub ElasticSearch — wystarczy dodać `JsonFormatter` do handlera. `prompt_hash` pozwala automatycznie wykryć nieautoryzowaną zmianę system message: jeśli hash zmienia się między deploymentami, wiesz dokładnie kiedy i który deployment to zrobił.

Ewaluacja jakości — LLM-as-judge

LLM-as-judge to wzorzec gdzie inny model ocenia jakość odpowiedzi Twojego systemu. Korelacja z oceną człowieka wynosi 0.85+ dla modeli klasy GPT-4o. Kluczowa zasada: oceniaj offline na próbce logów, nie w real-time dla każdego użytkownika — to redukuje koszt ewaluacji o 95% przy zachowaniu pełnej diagnozy.

Metryki RAG i progi alertów

import instructorfrom openai import OpenAIfrom pydantic import BaseModel, Fieldfrom typing import Optionalic = instructor.from_openai(OpenAI())class EvalResult(BaseModel):    faithfulness: int = Field(ge=1, le=5, description="1=pełna halucynacja, 5=w pełni oparta na kontekście")    relevance: int = Field(ge=1, le=5, description="1=zupełnie nie na temat, 5=idealnie odpowiada na pytanie")    completeness: int = Field(ge=1, le=5, description="1=brakuje kluczowych elementów, 5=wyczerpująca odpowiedź")    hallucination_detected: bool = Field(description="True jeśli odpowiedź zawiera fakty których nie ma w kontekście")    issues: Optional[list[str]] = Field(default=None, description="Lista konkretnych problemów jeśli wykryte, null jeśli brak")def evaluate_rag_response(question: str, context: str, answer: str) -> EvalResult:    user_content = "PYTANIE: " + question + "KONTEKST (jedyne dozwolone źródło faktów): " + context + "ODPOWIEDŹ AI DO OCENY: " + answer    return ic.chat.completions.create(        model="gpt-4o-mini",        response_model=EvalResult,        messages=[            {"role": "system", "content": "Jesteś ewaluatorem systemu RAG. Oceń odpowiedź AI bazując wyłącznie na dostarczonym kontekście — nie na własnej wiedzy o świecie."},            {"role": "user", "content": user_content}        ]    )

Koszt ewaluacji offline: dla 10 000 wywołań/dzień → losowa próbka 5% = 500 ewaluacji × ~$0.002/eval = $1/dzień za pełny quality monitoring. Uruchamiaj jako nocny job, wyniki umieszczaj na dashboardzie z trendami tygodniowymi.

Alerty produkcyjne — kiedy i jak reagować

Alerty twarde — reaguj w ciągu 15 minut

Błędy wymagające natychmiastowej interwencji (on-call, PagerDuty):

• Error rate > 5% przez 5 minut — problem z API lub kodem aplikacji
• Latency p95 > 30s — model przeciążony lub prompt zbyt długi, sprawdź rozmiar kontekstu
• Rate limit hit — przekroczono dzienny lub minutowy limit API, może zablokować wszystkich
• Cost spike > 300% normy dziennej — pętla nieskończona agenta lub nieoczekiwany atak
• Wszystkie odpowiedzi identyczne przez 10+ minut — stuck state lub bug z cachingiem

Alerty miękkie — reaguj w ciągu kilku godzin

Sygnały degradacji jakości wymagające analizy — Slack wystarczy, nie PagerDuty:

• Faithfulness < 3.5 przez 2+ godziny — problem z retrieval lub zepsuty schemat promptu
• Hallucination rate > 10% — model "wymyśla" zamiast przyznać brak wiedzy w kontekście
• Thumbs down ratio > 15% — sprawdź które konkretne pytania dostają negatywny feedback
• Session abandonment > 40% — użytkownicy porzucają rozmowę bez otrzymania odpowiedzi
• Escalation rate < 2% lub > 25% — AI albo zbyt rzadko, albo zbyt często przekazuje do człowieka

import statisticsfrom collections import dequefrom datetime import datetime, timedeltafrom dataclasses import dataclass, fieldfrom typing import Callable@dataclassclass Alert:    name: str    severity: str    message: str    triggered_at: datetime = field(default_factory=datetime.utcnow)class MetricBuffer:    def __init__(self, window_minutes: int = 10):        self.window = timedelta(minutes=window_minutes)        self._data: deque = deque()    def add(self, value: float):        now = datetime.utcnow()        self._data.append((now, value))        while self._data and self._data[0][0] < now - self.window:            self._data.popleft()    def mean(self) -> float:        return statistics.mean(v for _, v in self._data) if self._data else 0.0    def p95(self) -> float:        if not self._data:            return 0.0        vals = sorted(v for _, v in self._data)        return vals[int(len(vals) * 0.95)]    def count(self) -> int:        return len(self._data)latency = MetricBuffer(window_minutes=5)error_rate = MetricBuffer(window_minutes=5)faithfulness = MetricBuffer(window_minutes=60)def check_alerts(notify: Callable[[Alert], None]):    if error_rate.count() > 10 and error_rate.mean() > 0.05:        notify(Alert("high_error_rate", "critical", f"Error rate {error_rate.mean():.1%} > 5% in last 5 min"))    if latency.count() > 5 and latency.p95() > 30_000:        notify(Alert("high_latency", "critical", f"p95 latency {latency.p95():.0f}ms exceeds 30s"))    if faithfulness.count() > 20 and faithfulness.mean() < 3.5:        notify(Alert("low_faithfulness", "warning", f"Avg faithfulness {faithfulness.mean():.2f} < 3.5 threshold"))

Alerty przez webhook do Slack lub PagerDuty — nie e-mail. Dla alertów twardych: rotacja on-call z 15-minutowym SLA odpowiedzi. Dla alertów miękkich: kanał `#ai-quality` z dziennym digest i wykresami trendów tygodniowych.

Testy A/B promptów w produkcji

Przed wdrożeniem nowej wersji system message przetestuj ją na realnym ruchu. Pięć zasad:

1. 1.Ruch 50/50 przydzielany losowo per `user_id` — nie per sesja, jeden użytkownik widzi jeden wariant przez cały czas trwania testu
2. 2.Minimum 7 dni i 500 próbek per wariant — krótsze testy są statystycznie nieistotne i wprowadzają w błąd
3. 3.Mierz primary metric (np. faithfulness lub thumbs up) plus zestaw secondary (latency, koszt, escalation rate)
4. 4.Feature flag — przełączaj wariant bez redeploy, rollback w 30 sekund na poziomie konfiguracji
5. 5.Nie testuj wielu zmian jednocześnie — nie możesz izolować przyczyny różnic między wariantami

Lista kontrolna — przed każdym wdrożeniem produkcyjnym

10 elementów monitoringu AI które musisz mieć zanim wypuścisz system na użytkowników:

1. 1.Structured logging każdego wywołania LLM — trace_id, model, tokeny, latency, koszt
2. 2.Dashboard z p95 latency, error rate, kosztem dziennym i trendami
3. 3.Alert na rate limit i cost spike powyżej 300% normy dziennej
4. 4.Ewaluacja offline na próbce logów — minimum raz dziennie, wyniki na dashboardzie
5. 5.LLM-judge dla krytycznych ścieżek (odpowiedzi dotyczące cen, prawa, bezpieczeństwa)
6. 6.Thumbs up/down w UI — najtańszy i najcenniejszy sygnał jakości jaki możesz zebrać
7. 7.Prompt versioning — każda zmiana system message ma numer wersji i datę wdrożenia w logach
8. 8.A/B test framework — choćby prosty feature flag przed każdą zmianą promptu
9. 9.Retention policy — ustal ile logów przechowujesz pełnych, ile tylko jako metadane
10. 10.Runbook — kto i jak reaguje gdy każdy typ alertu się uruchomi, z checklistą kroków

---

Buduję systemy monitoringu AI dla firm wdrażających LLM na produkcji — od structured logging i tracingu po pełne pipeline'y ewaluacji z alertami i dashboardami. Napisz do mnie — zaczynam od audytu obecnego stanu obserwowalności Twoich systemów AI.