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

Розробка AI-системи моніторингу продуктивності AI-агентів

Моніторинг продуктивності AI-агентів - систематичне відстеження якості, швидкості та надійності роботи AI-воркерів у production. Відрізняється від моніторингу звичайних сервісів: метрики якості (точність, relevance) важливіше за метрик latency.

Метрики продуктивності AI-агента

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

Технічні метрики: latency (p50/p95/p99), черезput (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