Настройка логирования запросов и ответов AI-модели

Настройка логирования запросов и ответов AI-модели

Логирование всех запросов/ответов AI-модели — основа для отладки, аудита, анализа качества и compliance. Для LLM это нетривиально: большие объёмы данных, PII в промптах, high cardinality значений.

Структурированное логирование

import structlog
import json
from datetime import datetime

logger = structlog.get_logger()

def log_llm_request(
    request_id: str,
    user_id: str,
    model: str,
    messages: list[dict],
    response: str,
    usage: dict,
    latency_ms: float,
    error: str | None = None
):
    # PII-фильтрация перед логированием
    safe_messages = pii_filter(messages)

    logger.info(
        "llm_request",
        request_id=request_id,
        user_id=hash_user_id(user_id),  # псевдонимизация
        model=model,
        prompt_tokens=usage.get("prompt_tokens"),
        completion_tokens=usage.get("completion_tokens"),
        total_tokens=usage.get("total_tokens"),
        cost_usd=calculate_cost(model, usage),
        latency_ms=latency_ms,
        has_error=error is not None,
        error_type=type(error).__name__ if error else None,
        # Полные сообщения только если включено детальное логирование
        messages=safe_messages if DETAILED_LOGGING else None,
        response_preview=response[:200] if response else None,  # первые 200 символов
    )

Хранение полных логов (для аудита)

Для compliance нужны полные запросы/ответы. Отдельно от стандартных логов — большой объём:

class LLMRequestStore:
    def __init__(self, s3_bucket: str, retention_days: int = 90):
        self.s3 = boto3.client("s3")
        self.bucket = s3_bucket
        self.retention_days = retention_days

    def store(self, request: LLMRequest) -> str:
        # Путь: year/month/day/hour/request_id.json.gz
        key = f"llm-logs/{datetime.utcnow().strftime('%Y/%m/%d/%H')}/{request.id}.json.gz"

        data = gzip.compress(json.dumps(request.to_dict()).encode())
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=data,
            ContentEncoding="gzip",
        )
        return key

Защита персональных данных в логах

import re

class PIIFilter:
    PATTERNS = [
        (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD_NUMBER]'),
        (r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]'),
        (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
        (r'\+?[\d\s\-\(\)]{10,15}', '[PHONE]'),
    ]

    def filter(self, text: str) -> str:
        for pattern, replacement in self.PATTERNS:
            text = re.sub(pattern, replacement, text)
        return text

Retention и архивирование

Горячие логи (< 7 дней): Elasticsearch или ClickHouse для быстрого поиска. Тёплые (7–30 дней): S3 Standard. Холодные (30–365 дней): S3 Glacier Instant Retrieval. Удаление: по истечении retention_days через S3 Lifecycle rules.