Мониторинг и логирование: Интеграция с системами observability

Введение: Почему ваш лог-файл больше не работает?

Добро пожаловать в Модуль 6. Если вы дошли до этого момента, значит, вы уже умеете создавать мощные приложения на базе Gemini 3. Но давайте будем честными: заставить код работать на локальной машине — это лишь 20% успеха. Остальные 80% — это понимание того, как ваше приложение ведет себя в дикой природе, когда им пользуются тысячи людей.

В традиционной разработке мы привыкли следить за CPU, RAM и статусами HTTP (200 OK или 500 Error). В мире LLM (Large Language Models) эти метрики всё ещё важны, но они отходят на второй план. Если ваш сервер работает стабильно, но Gemini выдает галлюцинации или отвечает грубостью пользователям — ваш продукт сломан, даже если мониторинг показывает «зеленый свет».

LLMOps (Large Language Model Operations) вводит новые измерения наблюдаемости (observability):

• Стоимость: Каждый запрос стоит денег. Как отследить, какая фича сжигает бюджет?
• Качество: Как узнать, что ответы модели стали хуже после обновления промпта?
• Латентность (Latency): Не просто время ответа, а Time to First Token (TTFT) — время до появления первого символа на экране.

В этом уроке мы разберем, как построить систему мониторинга для Gemini 3, которая не просто собирает логи, а дает реальные инсайты.

Три столпа наблюдаемости в контексте GenAI

Чтобы «видеть» сквозь черный ящик нейросети, нам нужно интегрировать три классических компонента observability, но с нюансами для AI.

1. Логи (Logs)

В веб-разработке мы логируем ошибки. В LLMOps мы логируем промпты и комплиты (responses). Это ваше «золото» для отладки. Если пользователь жалуется на странный ответ, вы должны иметь возможность найти точный текст запроса и ответа модели.

Важно: Gemini 3 API возвращает богатые метаданные (usage metadata, safety ratings), которые обязательно нужно сохранять вместе с текстом.

2. Метрики (Metrics)

Это агрегированные данные. Нам нужны:

• Token Usage: Количество входных и выходных токенов (для контроля бюджета).
• Request Duration: Общее время выполнения.
• Error Rate: Процент ошибок API (например, Rate Limits 429 или Content Policy Violations).

3. Трейсинг (Tracing)

Самый сложный, но необходимый элемент. Если вы используете агентов (Agents) или цепочки вызовов (Chains), один запрос пользователя может породить 5-10 обращений к Gemini. Трейсинг позволяет визуализировать эту цепочку: Пользователь -> Агент -> Поиск в базе знаний -> Gemini (суммирование) -> Ответ.

import time
import logging
import google.generativeai as genai
from dataclasses import dataclass

# Настройка базового логгера
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("gemini_monitor")

@dataclass
class LLMMetrics:
    request_id: str
    latency_ms: float
    input_tokens: int
    output_tokens: int
    total_cost_estimated: float
    model_name: str

def calculate_cost(input_tok, output_tok, model="gemini-1.5-pro"):
    # Примерные цены (условные, для демонстрации)
    # В реальном проекте лучше хранить прайс-лист в конфиге
    pricing = {
        "gemini-1.5-pro": {"in": 0.0000035, "out": 0.0000105},
        "gemini-1.5-flash": {"in": 0.00000035, "out": 0.00000105}
    }
    rates = pricing.get(model, pricing["gemini-1.5-pro"])
    return (input_tok * rates["in"]) + (output_tok * rates["out"])

def log_gemini_interaction(model_name, prompt, response, start_time, request_id="req_123"):
    duration = (time.time() - start_time) * 1000
    
    # Извлечение метрик использования из объекта ответа Gemini
    # Обратите внимание: структура может зависеть от версии SDK
    usage = response.usage_metadata
    in_tokens = usage.prompt_token_count
    out_tokens = usage.candidates_token_count
    
    cost = calculate_cost(in_tokens, out_tokens, model_name)
    
    metrics = LLMMetrics(
        request_id=request_id,
        latency_ms=round(duration, 2),
        input_tokens=in_tokens,
        output_tokens=out_tokens,
        total_cost_estimated=round(cost, 6),
        model_name=model_name
    )
    
    # Структурированный лог (JSON-like) для легкого парсинга в ELK/Datadog
    logger.info(f"LLM_METRIC: {metrics.__dict__}")
    
    # Логирование контента (с осторожностью в продакшене!)
    logger.debug(f"PROMPT: {prompt[:50]}... | RESPONSE: {response.text[:50]}...")

# Пример использования
# genai.configure(api_key="...") 
# model = genai.GenerativeModel('gemini-1.5-flash')
# start = time.time()
# response = model.generate_content("Explain quantum physics like I'm 5")
# log_gemini_interaction('gemini-1.5-flash', "Explain quantum physics...", response, start)

Продвинутый уровень: Интеграция с OpenTelemetry

Ручное логирование через `print` или стандартный `logging` (как в примере выше) отлично подходит для MVP. Но когда у вас микросервисная архитектура, вам нужен стандарт. Таким стандартом в индустрии стал OpenTelemetry (OTel).

OTel позволяет собирать трейсы и метрики независимо от того, куда вы их отправляете (Jaeger, Prometheus, Datadog, Honeycomb или Google Cloud Trace).

Почему это критично для RAG и Агентов?

Представьте, что у вас есть RAG-система (Retrieval Augmented Generation). Пользователь задает вопрос. Система:

1. Идет в векторную базу данных (200мс).
2. Формирует контекст.
3. Отправляет запрос в Gemini (2000мс).
4. Получает ответ.

Если пользователь ждет 5 секунд, где проблема? В базе данных или в Gemini? Без распределенного трейсинга (Distributed Tracing) вы будете гадать. OTel позволяет видеть «водопад» вызовов.

Для Python существует отличная библиотека `opentelemetry-instrumentation`, и хотя официального авто-инструментатора для Gemini SDK может не быть в моменте, мы можем легко создать «спан» (Span) вручную вокруг вызова API.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
import google.generativeai as genai

# 1. Настройка OpenTelemetry (в продакшене это делается глобально)
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Экспортер выводит трейсы в консоль (в проде - в коллектор)
trace.get_tracer_provider().add_span_processor(
    SimpleSpanProcessor(ConsoleSpanExporter())
)

async def traced_gemini_call(prompt: str, model_name: str = "gemini-1.5-flash"):
    # Создаем Span (отрезок времени выполнения)
    with tracer.start_as_current_span("gemini_generation") as span:
        # Добавляем атрибуты к спану - это теги для поиска
        span.set_attribute("llm.system", "google")
        span.set_attribute("llm.model", model_name)
        span.set_attribute("llm.request.type", "chat")
        
        try:
            model = genai.GenerativeModel(model_name)
            
            # Можно добавить событие внутри спана
            span.add_event("sending_request_to_api")
            
            response = await model.generate_content_async(prompt)
            
            # Фиксируем метрики использования прямо в трейс
            usage = response.usage_metadata
            span.set_attribute("llm.usage.prompt_tokens", usage.prompt_token_count)
            span.set_attribute("llm.usage.completion_tokens", usage.candidates_token_count)
            
            # Если ответ заблокирован safety-фильтрами, это важно отметить
            if response.prompt_feedback:
                 span.set_attribute("llm.safety_rating", str(response.prompt_feedback))
            
            return response.text
            
        except Exception as e:
            # Записываем ошибку в трейс
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            raise e

# Теперь этот вызов будет виден в любой системе observability

Безопасность и Приватность данных (PII Redaction)

Это, пожалуй, самый деликатный аспект мониторинга LLM. Вы хотите логировать промпты, чтобы отлаживать качество. Но что, если пользователь введет в чат: «Проверь мой договор, паспортные данные: 1234 567890»?

Если вы сохраните этот промпт в логах как есть, вы нарушите GDPR, 152-ФЗ и другие законы о защите данных. Ваши логи станут вектором утечки данных.

Стратегии защиты (Sanitization):

1. PII Scrubbing (Очистка PII): Использование инструментов (например, Microsoft Presidio или простых RegEx) для поиска и замены чувствительных данных перед отправкой в логи.
   Пример: «Меня зовут Иван» -> «Меня зовут ».
2. Хэширование: Если вам нужно отслеживать уникальных пользователей, никогда не логируйте email или user_id в открытом виде. Используйте необратимый хэш (SHA-256).
3. Разделение уровней логирования:
   - `INFO`: Только метаданные (токены, латентность, статус).
   - `DEBUG`: Полный текст (включается только временно и с ограниченным доступом разработчиков).

Совет эксперта: Никогда не отправляйте сырые тексты промптов в облачные системы мониторинга (SaaS), если у вас нет подписанного DPA (Data Processing Agreement) с ними. Для чувствительных данных лучше использовать self-hosted решения (например, ELK Stack внутри вашего контура).

Метрики UX: Time to First Token (TTFT)

Для чат-ботов общее время генерации не так важно, как TTFT. Пользователь готов ждать 10 секунд полного ответа, если первые слова начнут появляться через 0.5 секунды. Если же экран пуст 3 секунды, пользователь считает, что приложение зависло.

При работе с Gemini 3 в режиме стриминга (`stream=True`), измерение TTFT выглядит так:

• Засекаем время T0 перед отправкой запроса.
• Ждем первый `chunk` данных от итератора.
• Как только `chunk` получен — фиксируем T1.
• TTFT = T1 - T0.

Если ваш TTFT стабильно выше 1-1.5 секунд, вам нужно либо оптимизировать промпт (сделать его короче), либо переключиться на более быструю модель (например, с Pro на Flash), либо проверить сетевые задержки.

import time
import json
import functools
from unittest.mock import Mock

# Имитация исключения для типизации (в реальном коде импорт из google.api_core)
class GoogleAPIError(Exception): pass

def monitor_gemini(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = None
        error = None
        
        try:
            result = func(*args, **kwargs)
            return result
        except Exception as e:
            error = str(e)
            raise e
        finally:
            duration = round((time.time() - start_time) * 1000, 2)
            
            log_entry = {
                "event": "gemini_call",
                "function": func.__name__,
                "duration_ms": duration,
                "status": "error" if error else "success"
            }
            
            if error:
                log_entry["error_message"] = error
            elif result and hasattr(result, 'usage_metadata'):
                # Пытаемся достать данные, если это объект Gemini
                usage = result.usage_metadata
                log_entry["input_tokens"] = usage.prompt_token_count
                log_entry["output_tokens"] = usage.candidates_token_count
            
            print(json.dumps(log_entry))

    return wrapper

# --- ТЕСТ ДЛЯ ПРОВЕРКИ ---

# Создаем мок-объект, имитирующий ответ Gemini
mock_response = Mock()
mock_response.text = "Hello AI"
mock_response.usage_metadata.prompt_token_count = 10
mock_response.usage_metadata.candidates_token_count = 5

@monitor_gemini
def generate_text_mock(prompt):
    time.sleep(0.1) # Имитация задержки
    return mock_response

print("--- Тест успешного вызова ---")
generate_text_mock("Test prompt")

@monitor_gemini
def generate_error_mock():
    time.sleep(0.05)
    raise ValueError("API key invalid")

print("\n--- Тест вызова с ошибкой ---")
try:
    generate_error_mock()
except:
    pass