Разработка системы мониторинга производительности AI-воркфорса

Разработка AI-системы мониторинга производительности AI-агентов

Мониторинг производительности AI-агентов — систематическое отслеживание качества, скорости и надёжности работы AI-воркеров в production. Отличается от мониторинга обычных сервисов: метрики качества (точность, relevance) важнее метрик latency.

Метрики производительности AI-агента

Три группы метрик:

Технические метрики: latency (p50/p95/p99), throughput (tasks/hour), error rate, availability, cost per task (токены × цена).

Качественные метрики: task completion rate (доля успешно выполненных задач), accuracy (правильность результата), hallucination rate, human override rate (как часто человек исправляет агента).

Бизнес-метрики: ROI (сэкономленные часы × ставка), customer satisfaction (если агент взаимодействует с клиентами), SLA compliance.

Система сбора метрик

from dataclasses import dataclass, field
from datetime import datetime
import uuid

@dataclass
class AgentTaskMetrics:
    task_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    agent_id: str = ""
    task_type: str = ""
    started_at: datetime = field(default_factory=datetime.utcnow)
    completed_at: datetime | None = None

    # Технические
    latency_ms: float | None = None
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0
    retries: int = 0

    # Качественные (заполняются post-hoc или авто-eval)
    task_completed: bool | None = None
    quality_score: float | None = None      # 0-1, авто-eval или human
    human_override: bool = False
    error_type: str | None = None

class AgentMonitor:
    def __init__(self, metrics_backend: MetricsBackend):
        self.backend = metrics_backend

    def track_task(self, agent_id: str, task_type: str):
        """Context manager для трекинга задачи."""
        return AgentTaskTracker(agent_id, task_type, self.backend)

class AgentTaskTracker:
    def __enter__(self) -> AgentTaskMetrics:
        self.metrics = AgentTaskMetrics(agent_id=self.agent_id, task_type=self.task_type)
        return self.metrics

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.metrics.completed_at = datetime.utcnow()
        self.metrics.latency_ms = (
            self.metrics.completed_at - self.metrics.started_at
        ).total_seconds() * 1000
        if exc_type:
            self.metrics.error_type = exc_type.__name__
        self.backend.record(self.metrics)

Автоматическая оценка качества

Для большинства агентских задач человеческая проверка каждого результата невозможна. Авто-eval:

def auto_evaluate_task(task: AgentTask, result: AgentResult) -> float:
    """Оценка качества результата через LLM-судью."""
    eval_prompt = f"""Оцени качество выполнения задачи агентом.

Задача: {task.description}
Ожидаемый результат: {task.expected_outcome}
Фактический результат: {result.output}

Оцени от 0 до 1, где:
1.0 — задача выполнена полностью и корректно
0.5 — частичное выполнение или незначительные ошибки
0.0 — задача не выполнена или критические ошибки

Ответь только числом."""

    score = float(eval_llm.generate(eval_prompt, max_tokens=10).strip())
    return min(max(score, 0.0), 1.0)

Дашборд мониторинга агентов

Ключевые панели: SLA compliance (% задач в рамках SLA), качество по типам задач, стоимость в динамике (рост цены = рост токенов или рост ошибок с retry), human override rate (тренд: если растёт — агент деградирует), error taxonomy.

Алерты деградации

class DegradationDetector:
    def check(self, metrics: AgentMetricsSummary) -> list[Alert]:
        alerts = []

        if metrics.error_rate_7d > metrics.error_rate_30d * 1.5:
            alerts.append(Alert(
                severity="warning",
                message=f"Error rate grew by {metrics.error_rate_7d/metrics.error_rate_30d:.1f}x over 7 days"
            ))

        if metrics.avg_quality_score_7d < metrics.avg_quality_score_30d - 0.1:
            alerts.append(Alert(
                severity="warning",
                message=f"Quality score dropped from {metrics.avg_quality_score_30d:.2f} to {metrics.avg_quality_score_7d:.2f}"
            ))

        if metrics.human_override_rate_7d > 0.15:  # > 15% задач переделываются
            alerts.append(Alert(
                severity="critical",
                message=f"Human override rate too high: {metrics.human_override_rate_7d:.1%}"
            ))

        return alerts