Proof of concept je jednoduchý. Produkčný AI systém je iná liga. Tu sú skúsenosti z reálnych nasadení.

Monitoring a Observability

Logujte všetko

import logging
from datetime import datetime

def log_llm_call(prompt, response, model, latency):
    logging.info({
        "timestamp": datetime.utcnow().isoformat(),
        "model": model,
        "prompt_tokens": count_tokens(prompt),
        "response_tokens": count_tokens(response),
        "latency_ms": latency,
        "prompt_hash": hash(prompt),  # Pre debugging bez PII
    })

Metriky

# Kľúčové metriky
metrics = {
    "latency_p50": [],
    "latency_p99": [],
    "error_rate": [],
    "token_usage": [],
    "cost_per_request": [],
    "user_satisfaction": [],  # Thumbs up/down
}

Alerting

def check_anomalies():
    # Latencia
    if avg_latency > 5000:  # ms
        alert("High latency detected")

    # Error rate
    if error_rate > 0.05:  # 5%
        alert("High error rate")

    # Cost spike
    if daily_cost > budget_limit:
        alert("Cost budget exceeded")

Caching

Response cache

import hashlib
import redis

redis_client = redis.Redis()

def cached_llm_call(prompt: str, model: str) -> str:
    cache_key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()

    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return cached.decode()

    # Call LLM
    response = llm.generate(prompt, model=model)

    # Cache for 1 hour
    redis_client.setex(cache_key, 3600, response)

    return response

Semantic cache

from langchain.cache import RedisSemanticCache

# Cache podobné queries
set_llm_cache(RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(),
    score_threshold=0.95
))

Rate Limiting

from ratelimit import limits, sleep_and_retry

# OpenAI rate limits
@sleep_and_retry
@limits(calls=60, period=60)  # 60 calls per minute
def rate_limited_call(prompt):
    return client.chat.completions.create(...)

# User-level limits
def check_user_quota(user_id: str) -> bool:
    daily_usage = get_daily_usage(user_id)
    return daily_usage < USER_DAILY_LIMIT

Fallback Strategy

async def resilient_llm_call(prompt: str) -> str:
    providers = [
        ("openai", "gpt-4o"),
        ("anthropic", "claude-3-5-sonnet"),
        ("mistral", "mistral-large"),
    ]

    for provider, model in providers:
        try:
            return await call_provider(provider, model, prompt)
        except (RateLimitError, ServiceUnavailable) as e:
            logging.warning(f"{provider} failed: {e}")
            continue

    raise AllProvidersFailedError()

Cost Control

def estimate_cost(prompt: str, model: str) -> float:
    input_tokens = count_tokens(prompt)

    costs = {
        "gpt-4o": {"input": 5.0, "output": 15.0},  # per 1M
        "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
    }

    # Estimate output (avg 2x input)
    estimated_output = input_tokens * 2

    cost = (input_tokens * costs[model]["input"] +
            estimated_output * costs[model]["output"]) / 1_000_000

    return cost

def should_proceed(user_id: str, estimated_cost: float) -> bool:
    user_balance = get_user_balance(user_id)
    return user_balance >= estimated_cost

Testing

Unit testy s mockami

def test_summarization():
    with patch('openai.ChatCompletion.create') as mock:
        mock.return_value = Mock(
            choices=[Mock(message=Mock(content="Summary"))]
        )

        result = summarize("Long text...")
        assert "Summary" in result

Eval dataset

eval_cases = [
    {
        "input": "Preklad: Hello world",
        "expected_contains": ["Ahoj", "svet"],
        "expected_not_contains": ["error"],
    }
]

def run_evals():
    results = []
    for case in eval_cases:
        response = llm.generate(case["input"])
        passed = (
            all(kw in response for kw in case["expected_contains"]) and
            all(kw not in response for kw in case["expected_not_contains"])
        )
        results.append({"case": case, "passed": passed})

    return results

Checklist pre produkciu

  • Monitoring a alerting nastavený
  • Caching implementovaný
  • Rate limiting pre API aj users
  • Fallback providers nakonfigurované
  • Cost tracking a limity
  • PII filtering na vstupe aj výstupe
  • Eval dataset a pravidelné testovanie
  • Incident response plán

Záver

AI v produkcii vyžaduje rovnakú starostlivosť ako každý iný kritický systém - monitoring, testing, fallbacks a cost control. Investícia do infraštruktúry sa vráti.