Как вы логируете все вызовы LLM для аудита?

Краткий тезис

Логирование всех вызовов LLM — это обязательная практика для аудита, отладки и соответствия требованиям (compliance). Каждый запрос и ответ сохраняются в структурированном формате (JSON) в специализированное хранилище (ClickHouse), с разными уровнями детализации и политикой ретеншна. Критически важны маскирование персональных данных]] (PII) и обеспечение неизменности журналов для юридической доказательности.

1. Зачем логировать вызовы LLM: пять целей

Аудит — не единственная причина. Логи необходимы для:

Без полного лога невозможно доказать, что система работала корректно (или выявить, что пошло не так).

2. Структура лога: минимальный набор полей в JSON

Каждый лог — это строка JSON. Обязательные поля:

{
  "event_id": "uuid-v4",
  "timestamp": "2025-02-18T10:30:00.123Z",
  "user_id": "hashed_user_abc123",
  "session_id": "session_xyz456",
  "request_id": "req_789",
  "model": "gpt-4o",
  "prompt": "<текст запроса после маскировки>",
  "response": "<текст ответа после маскировки>",
  "prompt_tokens": 245,
  "completion_tokens": 120,
  "total_tokens": 365,
  "latency_ms": 1560,
  "temperature": 0.7,
  "max_tokens": 2000,
  "response_status": "success",
  "pii_masked": true,
  "version": "1.2"
}

Термин «маскировка PII» (Personally Identifiable Information) — замена реальных имён, адресов, номеров телефонов на плейсхолдеры (например, [PII_REDACTED]) до записи в лог.

Дополнительные поля для аудита: hash всего объекта (SHA256) для проверки целостности, signature (приватный ключ сервера) для аутентификации записи.

3. Хранилище: почему ClickHouse и как организовать данные

ClickHouse — колоночная СУБД, оптимизированная для аналитических запросов с высокой скоростью записи. Он идеально подходит для логов, потому что:

• Вставка тысяч строк в секунду (OLAP).
• Быстрая фильтрация по времени, модели, пользователю.
• Нативная поддержка TTL (Time-To-Live) для автоматического удаления старых записей.

Пример схемы таблицы:

CREATE TABLE llm_logs (
    event_id UUID,
    timestamp DateTime64(3),
    user_id String,
    session_id String,
    request_id String,
    model String,
    prompt Mediumtext,
    response Mediumtext,
    prompt_tokens UInt32,
    completion_tokens UInt32,
    latency_ms UInt32,
    response_status String,
    pii_masked Bool,
    event_hash String
) ENGINE = MergeTree()
ORDER BY (timestamp, model)
TTL timestamp + INTERVAL 90 DAY DELETE;

Альтернативы PostgreSQL (если запись <10k/s и нужны транзакции), Elasticsearch (для полнотекстового поиска по промптам), облачные решения вроде S3 + Athena (дёшево, но медленно при частых запросах).

4. Уровни логирования: DEBUG, INFO, ERROR

Типичная трехуровневая модель:

В продакшне обычно используют INFO для 99.9% запросов, включая усечённые данные. DEBUG — только для выбранных пользователей (по session_id) или в отдельной низкошумящей таблице с коротким TTL.

5. Безопасность: маскировка PII и контроль доступа

Перед записью лога необходимо удалить или заменить PII. Процесс:

1. Промпт и ответ проходят через детектор PII (spaCy, Presidio, Microsoft Presidio или собственный regex).
2. Найденные сущности маскируются: { "email": "[EMAIL_REDACTED]", "phone": "[PHONE_REDACTED]" }.
3. Исходный (незамаскированный) лог сохраняется только в системе SIEM с ограниченным доступом и audit trail (журналом доступа к журналу).
4. Доступ к таблицам логов — по ролям (например, auditor, developer) и с обязательным аудитом запросов.

Дополнительная защита: шифрование на уровне хранения (TDE) и при передаче (TLS). Для юридической значимости — WORM (Write Once Read Many) хранилище, которое нельзя изменить постфактум.

6. Политика ретеншна: 30–90 дней

Срок хранения зависит от требований регулятора и бизнеса:

Реализуется TTL на уровне базы или внешний скрипт архивации (cold storage в S3 Glacier). Важно: после удаления логов из горячего хранилища они должны быть восстановимы в течение срока аудита (если требуется).

7. Аудит и цепочки proof: неизменяемость и верификация

Чтобы лог был принят как доказательство, он должен быть неизменяемым:

• Каждая запись содержит хэш (SHA-256 всех полей) и цифровую подпись (HMAC с секретным ключом).
• Хэш предыдущего блока записей включается в следующий (аналог блокчейна для логов).
• Периодически (каждый час) создаётся snapshot — сжатый файл с агрегированным хэшем, который отправляется в WORM-хранилище.

Пример проверки целостности (Python):

import hashlib, hmac

def verify_log_entry(entry: dict, secret: bytes) -> bool:
    expected_hash = entry.pop('event_hash', None)
    if not expected_hash:
        return False
    data = json.dumps(entry, sort_keys=True, ensure_ascii=False).encode()
    real_hash = hashlib.sha256(data).hexdigest()
    return hmac.compare_digest(expected_hash, real_hash)

8. Инструменты и подходы к сбору

В промышленных системах логирование внедряется через стандартизированный middleware (пример для FastAPI):

from fastapi import Request, Response
import time, uuid, json, hashlib

async def llm_logging_middleware(request: Request, call_next):
    start = time.monotonic()
    body = await request.body()
    response = await call_next(request)
    latency = time.monotonic() - start
    
    log_entry = {
        "event_id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "user_id": request.headers.get("X-User-ID", "anonymous"),
        "model": request.headers.get("X-Model", "unknown"),
        "prompt": mask_pii(body.decode()),
        "response": mask_pii(response.body.decode()),
        "prompt_tokens": count_tokens(body),
        "completion_tokens": count_tokens(response.body),
        "latency_ms": int(latency * 1000),
        "response_status": "success" if response.status_code < 400 else "error"
    }
    log_entry["event_hash"] = hashlib.sha256(
        json.dumps(log_entry, sort_keys=True).encode()
    ).hexdigest()
    
    # Асинхронная отправка в ClickHouse через буфер (Kafka/Redis)
    await log_queue.put(log_entry)
    return response

Стек OpenTelemetry (трейсинг), Kafka (буфер), Logstash (трансформация), ClickHouse (хранение), Grafana (визуализация). Альтернатива — MLflow (для экспериментов) или LangSmith (коммерческое решение для LLM).

9. Стоимость и масштабирование

Логирование всех вызовов может генерировать терабайты в день при высокой нагрузке. Оптимизация:

• Сэмплирование — писать 100% только для ошибок и случайных 1% успешных вызовов (если не требуется полный аудит).
• Партиционирование по дням и моделям — ускорение очистки по TTL.
• Сжатие в ClickHouse (ZSTD) — уменьшение объёма в 3–5 раз.
• Агрегация — вместо хранения каждого вызова хранить минутные агрегаты (средняя latency, p95, количество токенов, количество ошибок), а для полного аудита — только для запросов, помеченных флагом audit_required=true.

10. Пет-проект для закрепления

Задача Разработать систему логирования вызовов LLM с аудитом и дашбордом.

Инструменты Python (FastAPI или Flask), ClickHouse (или SQLite для теста), Docker, Grafana (опционально). Детектор PII — библиотека presidio-analyzer.

Шаги:

1. Напишите middleware, которое перехватывает вызов к LLM (можно заменить на заглушку mock_llm.py).
2. Реализуйте маскировку email, телефона, номера карты.
3. Создайте модель ClickHouse (или таблицу SQLite) по схеме из раздела 3.
4. Сделайте API для записи лога (POST /log) и для чтения логов по времени (GET /logs?from=2025-01-01).
5. Добавьте проверку целостности с помощью HMAC.
6. Настройте TTL (например, 7 дней). Напишите тест, проверяющий, что запись удаляется после TTL.
7. Постройте простой дашборд в Streamlit: количество запросов в час, средняя latency, процент ошибок.

Ожидаемый результат

• Работающий endpoint, который принимает и сохраняет JSON-лог.
• Проверка, что PII заменено на [REDACTED].
• Возможность запросить лог за период и убедиться в его целостности.
• Отчёт с графиками, который помогает выявить аномалии.

Связь с другими вопросами

Навигация

• Предыдущий: 72
• Следующий: 74
• Индекс: 00. Индекс разборов