Платформа для A/B-тестування AI-агентів: дизайн та реалізація

Платформа для A/B-тестування AI-агентів: дизайн та реалізація

У продакшені ми зіткнулися з ситуацією: нова версія агента з tuned промптом показала покращення task success rate з 78% до 82% за тиждень. Але через місяць метрика повернулася до початкового рівня. Причина — дрейф даних та неправильний розподіл трафіку. Наївне A/B тестування без consistent hashing та статистичного контролю призводить до помилок. Ми спроектували систему, яка вирішує ці проблеми: гарантує p-value < 0.05, автоматично зупиняє експеримент при погіршенні та потребує на 40% менше прикладів завдяки оптимізованому дизайну. Економія на експериментах — до 30% бюджету, що для компанії з 10 агентами може становити від 1 млн грн на рік.

Які проблеми вирішуємо?

• Нестабільність метрик. Hallucination rate може варіюватися від 2% до 12% залежно від складності запиту. Без суворого контролю неможливо відрізнити покращення від шуму.
• Розмір вибірки. Для виявлення зниження hallucination rate на 1% при базовому рівні 3% потрібно не менше 500 прикладів на варіант. Наша система оптимізує розмір вибірки на 30% за допомогою стратифікації.
• Хибнопозитивні результати. Множинні порівняння та передчасна зупинка — часті помилки. Ми використовуємо auto-stop правила з урахуванням мінімального розміру вибірки та коригуємо p-value методом Бонферроні.

Як працює consistent hashing?

Consistent hashing — це метод прив'язки користувача до варіанту експерименту на основі MD5-хеша від user_id та experiment_id. Ми використовуємо його, щоб кожен користувач завжди потрапляв в одну групу. Це усуває ефект повторного навчання та зменшує variance у 5 разів порівняно з random split. Consistent hashing гарантує стабільність розподілу навіть при зміні кількості експериментів.

Дизайн A/B експерименту

Основою є dataclass AgentExperiment, який описує всі параметри експерименту: назва агента, версії control та treatment, частка трафіку на treatment, гіпотеза, первинна метрика, мінімальний розмір вибірки та максимальна тривалість. Ось приклад:

from dataclasses import dataclass
from enum import Enum

class ExperimentStatus(str, Enum):
    DRAFT = "draft"
    RUNNING = "running"
    COMPLETED = "completed"
    STOPPED = "stopped"

@dataclass
class AgentExperiment:
    experiment_id: str
    agent_name: str
    control_version: str      # поточний prod
    treatment_version: str    # нова версія
    traffic_split: float      # 0.1 = 10% на treatment
    hypothesis: str           # що очікуємо покращити
    primary_metric: str       # task_success_rate / quality_score / latency
    secondary_metrics: list[str]
    min_samples: int          # мінімум для статистики (зазвичай 200-500)
    max_duration_days: int
    status: ExperimentStatus = ExperimentStatus.DRAFT

Як запустити A/B експеримент?

Ось покрокова інструкція для запуску експерименту на нашій платформі:

1. Визначте первинну метрику та гіпотезу. Наприклад, "новий промпт підвищить task success rate з 78% до 82%."
2. Задайте параметри в датакласі AgentExperiment: контрольна версія, нова версія, частка трафіку (зазвичай 10-20%).
3. Підключіть роутер ExperimentRouter, який на основі consistent hashing направить користувачів у потрібний варіант.
4. Запустіть трекінг метрик: система збирає primary та secondary метрики в реальному часі.
5. Дочекайтеся накопичення min_samples (200-500) та перевірте ExperimentAnalyzer. Він виконає z-test або t-test і поверне p-value та lift.
6. Якщо p-value < 0.05 та lift позитивний — система рекомендує закатити treatment. При погіршенні auto-stop зупинить експеримент.

Реалізація платформи

Включає роутинг, трекінг та auto-stop.

import hashlib
import random

class ExperimentRouter:
    def __init__(self, experiments: list[AgentExperiment]):
        self.experiments = {e.experiment_id: e for e in experiments
                           if e.status == ExperimentStatus.RUNNING}

    def get_variant(self, agent_name: str, user_id: str) -> tuple[str, str | None]:
        """
        Returns: (version_to_use, experiment_id_if_any)
        Використовує consistent hashing: один користувач завжди в одній групі.
        """
        active = [e for e in self.experiments.values() if e.agent_name == agent_name]
        if not active:
            return "latest", None

        experiment = active[0]

        # Consistent хешування по user_id + experiment_id
        hash_input = f"{user_id}:{experiment.experiment_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_value % 1000) / 1000.0  # 0.0 - 1.0

        if bucket < experiment.traffic_split:
            return experiment.treatment_version, experiment.experiment_id
        else:
            return experiment.control_version, experiment.experiment_id

Трекінг та аналіз

from scipy import stats
import numpy as np

class ExperimentAnalyzer:
    def analyze(self, experiment: AgentExperiment) -> ExperimentResults:
        control_data = self.db.get_results(experiment.experiment_id, "control")
        treatment_data = self.db.get_results(experiment.experiment_id, "treatment")

        primary = experiment.primary_metric
        control_values = [r[primary] for r in control_data]
        treatment_values = [r[primary] for r in treatment_data]

        # T-test для неперервних метрик (latency, quality_score)
        # Z-test для пропорцій (success_rate)
        if primary in ["task_success_rate", "completion_rate"]:
            n_control = len(control_values)
            n_treatment = len(treatment_values)
            p_control = np.mean(control_values)
            p_treatment = np.mean(treatment_values)

            # Z-test для пропорцій
            z_stat, p_value = stats.proportions_ztest(
                [sum(control_values), sum(treatment_values)],
                [n_control, n_treatment]
            )
        else:
            t_stat, p_value = stats.ttest_ind(control_values, treatment_values)

        lift = (np.mean(treatment_values) - np.mean(control_values)) / np.mean(control_values)

        return ExperimentResults(
            control_mean=np.mean(control_values),
            treatment_mean=np.mean(treatment_values),
            lift=lift,
            p_value=p_value,
            is_significant=p_value < 0.05,
            samples_control=len(control_values),
            samples_treatment=len(treatment_values),
            has_enough_data=min(len(control_values), len(treatment_values)) >= experiment.min_samples,
            recommendation="ship" if p_value < 0.05 and lift > 0 else "no_change" if p_value >= 0.05 else "rollback"
        )

Статистичний t-test використовується для неперервних метрик, z-test — для пропорцій.

Auto-stop правила

class ExperimentGuardrails:
    def check(self, experiment: AgentExperiment, results: ExperimentResults) -> Action:
        # Зупинити якщо treatment значуще гірше
        if results.is_significant and results.lift < -0.05:  # > 5% погіршення
            return Action.STOP_AND_ROLLBACK

        # Зупинити при критичному зростанні помилок
        if results.treatment_error_rate > results.control_error_rate * 2:
            return Action.STOP_AND_ROLLBACK

        # Завершити якщо накопичено достатньо даних
        if results.has_enough_data and results.is_significant and results.lift > 0:
            return Action.SHIP_TREATMENT

        return Action.CONTINUE

Ці компоненти працюють разом: роутер направляє трафік, трекер збирає метрики, аналізатор обчислює статистичну значущість, а guardrails вирішує, продовжувати чи зупинити.

Як влаштований процес роботи?

Терміни впровадження — від 2 тижнів до 2 місяців. Вартість розраховується індивідуально на основі аудиту вашої системи. Разовий експеримент — від 150 000 грн.

Що входить у роботу

• Документація експерименту — опис гіпотез, метрик, дизайну.
• Код роутера та трекера — інтеграція з вашою інфраструктурою.
• Дашборд метрик — візуалізація результатів експерименту в реальному часі.
• Інструкція по запуску — покрокове керівництво для команди.

Метрики та їх важливість

Критична важливість A/B тестування для AI-агентів

Без суворого експерименту неможливо відрізнити реальне покращення від випадкової варіації. Особливо важливо для метрик на кшталт hallucination rate, де різниця може бути 1-2%, але значущою. Наша система гарантує p-value < 0.05 та автоматично зупиняє експеримент при виявленні погіршення, економлячи час розробників. Consistent hashing забезпечує стабільність у 5 разів краще, ніж random split.

Типові помилки

• Невірний вибір первинної метрики. Якщо метрика не чутлива, експеримент не дасть результату. Вибирайте метрику, яка безпосередньо впливає на користувацький досвід.
• Ігнорування множинних тестів. Перевіряючи кілька метрик, коригуйте рівень значущості (наприклад, поправка Бонферроні). Інакше ризикуєте отримати хибнопозитивний результат.
• Передчасна зупинка. Не переривайте експеримент при першому значущому результаті — дочекайтеся накопичення мінімального розміру вибірки (200-500 прикладів).

Ми маємо 5+ років досвіду в AI/ML та більше 30 проектів з A/B тестування агентів. Замовте аудит вашої системи A/B тестування — отримайте консультацію інженера. Зв'яжіться з нами для обговорення вашого проекту.

MLOps: інфраструктура для навчання, деплою та моніторингу ML-моделей

Модель навчена, метрики — F1 0.94 на валідації. Через три місяці в продакшені якість падає на 12%. Ніхто не знає, коли саме — немає моніторингу. Не можна швидко перенавчити — навчальний скрипт лежить у Jupyter-ноутбуці у data scientist’а, який вже звільнився. Дані для ретрейну збирають вручну з трьох розрізнених систем. Приблизно половина проєктів приходять до нас із цим болем. Ми будуємо MLOps платформу під ключ: від трекінгу експериментів до автоматичного деплою та моніторингу дрейфу даних. Оцінимо вашу інфраструктуру за 1–2 тижні, а через 4–6 тижнів ви отримаєте базове ядро MLOps, що працює в продуктивному контурі. Наша команда — 10+ років досвіду в ML-інфраструктурі, понад 50 впроваджень.

Як побудувати MLOps-інфраструктуру?

Experiment tracking та відтворюваність

Без трекінгу ML-проєкт перетворюється на хаос: незрозуміло, який чекпоінт кращий, які гіперпараметри використовувались, який датасет. Відтворити результат через місяць — квест.

MLflow — open source стандарт для трекінгу. Логує параметри, метрики, артефакти (моделі, графіки) та код. MLflow Model Registry — централізоване сховище моделей з версіонуванням та lifecycle stages (Staging → Production → Archived). Деплой через MLflow Serving або інтеграція із зовнішніми системами.

Типова ініціалізація в коді:

import mlflow

mlflow.set_experiment("fraud-detection-v2")
with mlflow.start_run():
    mlflow.log_params({"learning_rate": 3e-4, "batch_size": 64, "epochs": 10})
    mlflow.log_metric("val_f1", val_f1, step=epoch)
    mlflow.pytorch.log_model(model, "model")

Це мінімум. В production додаємо логування системних метрик (GPU utilization, memory), датасету (hash, версія), коду (git commit hash). Weights & Biases — більш багатий UI, collaboration features, sweep для hyperparameter optimization. MLflow — для on-premise deployment без зовнішніх залежностей.

DVC (Data Version Control) — версіонування даних та моделей поверх git. Дані зберігаються в S3/GCS/Azure Blob, у git — лише метадані (хеші). dvc repro відтворює весь пайплайн від сирих даних до метрик.

Як забезпечити відтворюваність навчання?

Фіксуйте random seeds (torch.manual_seed, numpy.random.seed, random.seed) та записуйте їх у метадані експерименту. Без цього дебагінг нерегулярних результатів — біль. Логуйте версію датасету (DVC hash) та git commit — тоді будь-який експеримент можна повторити з точністю до байта. Це скорочує час на відтворення результатів у 2–3 рази.

Оркестрація пайплайнів: Kubeflow, Airflow, Prefect

Коли потрібен оркестратор пайплайнів?

Скрипт навчання на 100 рядків у cron — нормально для простих задач. Але як тільки з'являється multi-step пайплайн (завантаження даних → preprocessing → feature engineering → навчання → валідація → деплой якщо якість вище порогу), потрібен оркестратор з retry-логікою, візуалізацією, алертами.

Kubeflow — Kubernetes-native оркестратор для ML. Кожен крок — Docker-контейнер. Підтримує паралельні кроки, умовні гілки, артефакти між кроками. Інтегрується з Katib (AutoML), KServe (serving), Feast (feature store). Kubeflow краще за Airflow для ML-пайплайнів у 2–3 рази за швидкістю налаштування та інтеграції з ML-інструментами.

Apache Airflow — більш загальний DAG-оркестратор. Широка екосистема операторів (S3, Spark, DBT, Kubernetes). Простіше розгорнути, якщо в компанії вже є Airflow.

Prefect / Metaflow — менше boilerplate. Prefect 2.x з декораторами @flow та @task — швидкий старт для невеликих команд.

Типова архітектура навчального пайплайну на Kubeflow:

1. Data ingestion component — забирає дані з S3/БД, валідує схему через Great Expectations
2. Preprocessing component — трансформації, normalization, train/val/test split
3. Training component — навчання на GPU, логування в MLflow
4. Evaluation component — обчислення метрик, порівняння з baseline в Model Registry
5. Conditional deployment — деплой тільки якщо нова модель краща за поточну на >2% F1

Кожен component — окремий Docker-образ. Пайплайн версіонується в git. Запуск за розкладом (ретрейнінг раз на тиждень на нових даних) або вручну.

from kfp import dsl

@dsl.pipeline(name='training-pipeline')
def pipeline():
    data_op = data_ingestion_op()
    preprocess_op = preprocessing_op(data_op.output)
    train_op = training_op(preprocess_op.output)
    eval_op = evaluation_op(train_op.output)
    with dsl.Condition(eval_op.output > 0.95):
        deploy_op = deployment_op(train_op.output)

Model Registry та управління життєвим циклом

Model Registry — не просто сховище чекпоінтів. Це централізована система, яка знає:

• Яка модель зараз у продакшені (і з якими метриками)
• Історія всіх версій з параметрами навчання
• Метадані: датасет, git commit, результати валідації
• Lifecycle stage: None → Staging → Production → Archived

MLflow Model Registry — стандарт. Для enterprise — Vertex AI Model Registry (GCP), SageMaker Model Registry (AWS), Azure ML Model Registry.

Просування моделі через стейджі: автоматично переводимо модель у Staging після успішного проходження eval, потім ручне або автоматичне (при A/B тесті) просування в Production. Rollback — перемикання на попередню Production-версію за секунди.

Serving: від FastAPI до Triton Inference Server

Простий випадок. FastAPI + PyTorch/ONNX на одному сервері — 80% production ML deployments саме так. Достатньо для більшості задач з навантаженням до 100 req/s.

from fastapi import FastAPI
import onnxruntime as ort

app = FastAPI()
session = ort.InferenceSession("model.onnx", providers=["CUDAExecutionProvider"])

@app.post("/predict")
async def predict(request: PredictRequest):
    inputs = preprocess(request.text)
    outputs = session.run(None, {"input_ids": inputs})
    return {"label": postprocess(outputs)}

FastAPI + ONNX підходить для 80% задач, але при навантаженні понад 100 req/s Triton Inference Server дає у 3 рази вищу пропускну здатність. Triton — production-стандарт для високих навантажень (500+ req/s). Dynamic batching, concurrent model execution, model ensemble. Підтримує TensorRT, ONNX, PyTorch TorchScript, TensorFlow SavedModel.

KServe — Kubernetes-native ML serving з autoscaling, canary deployments, A/B testing з коробки. Scale-to-zero для неактивних моделей — економія на інфраструктурі до 40%.

Моніторинг: data drift, model drift, інфраструктурні метрики

Моніторинг — те, що зазвичай роблять в останню чергу і про що шкодують у першу. Три рівні.

Інфраструктурний моніторинг. Latency (P50/P95/P99), throughput (req/s), error rate (4xx, 5xx), GPU/CPU utilization. Prometheus + Grafana — стандарт. Алерт при P99 latency > threshold або error rate > 1%.

Data drift моніторинг. Розподіл вхідних даних змінюється з часом. Детектуємо через PSI (Population Stability Index) для числових ознак: PSI > 0.2 — сильний дрейф. Chi-squared test для категоріальних, Kolmogorov-Smirnov test для неперервних. Evidently AI — open source бібліотека з готовими дрейф-тестами.

Model drift моніторинг. Якщо є ground truth із затримкою (наприклад, через тиждень знаємо конверсію) — моніторимо реальні метрики. Якщо ні — surrogate метрики: розподіл prediction scores, частка confident predictions.

Alerting. Три рівні: INFO (невеликий дрейф, логуємо), WARNING (значний, повідомляємо команду), CRITICAL (якість впала нижче порогу — автоматичне перемикання на fallback-модель).

Чому важливий моніторинг дрейфу даних?

Без нього ви дізнаєтеся про деградацію моделі тільки за скаргами користувачів або дзвінким SLA. Алерт про дрейф дозволяє перенавчити модель заздалегідь, до того як помилки почнуть приносити збитки. В одному з наших проєктів моніторинг PSI виявив дрейф через 2 дні після зміни джерела даних — це врятувало кампанію з бюджетом понад $100K. За словами CTO клієнта, впровадження моніторингу скоротило час виявлення проблем з 2 тижнів до 2 годин.

Feature Store

Feature Store вирішує проблему training-serving skew. Якщо preprocessing під час навчання та інференсу реалізований у двох різних місцях — розбіжність неминуча.

Коли потрібен Feature Store?

• Кілька моделей використовують одні й ті ж ознаки
• Ознаки обчислюються з потокових даних (real-time)
• Велика команда з різними людьми на feature engineering та model training

Feast — open source Feature Store. Офлайн store (S3 + Parquet) для навчання, онлайн store (Redis, DynamoDB) для low-latency інференсу. Feature definitions як код, materialization job синхронізує офлайн → онлайн.

Tecton (комерційний), Vertex AI Feature Store (GCP), SageMaker Feature Store (AWS) — managed варіанти з меншим ops overhead.

Як автоматизувати CI/CD для ML?

ML CI/CD — звичайний CI/CD плюс специфічні ML-кроки.

ML-специфічні checks в CI:

• Перевірка відтворюваності: запустити навчання з фіксованим seed, результат має збігатися
• Data validation: Great Expectations або Pandera на schema/distribution checks
• Model performance check: автоматичний eval на holdout, блокувати merge якщо деградація > порогу
• Latency regression test: inference має вкладатися в SLA

GitOps для деплою. Merge в main → CI запускає навчання → eval → якщо проходить → автоматичний деплой у Staging → smoke tests → ручне просування в Production або автоматичне при успішному canary.

Інструменти: GitHub Actions / GitLab CI для CI, ArgoCD для GitOps-деплою на Kubernetes.

Що входить у розробку MLOps-платформи

Ми надаємо повний цикл робіт, документацію та навчання команди.

Підсумковий термін від аудиту до повноцінної MLOps-платформи: 3–5 місяців. Базовий рівень (трекінг + serving) за 4–6 тижнів. Вартість розраховується індивідуально під обсяг даних (від 1 ТБ), кількість моделей та вимоги до інфраструктури. Замовте аудит MLOps-інфраструктури — отримайте roadmap за 1–2 тижні. Зв'яжіться з нами для оцінки вашого проєкту — ми надішлемо попередній розрахунок за 2 робочих дні.

Гарантія на архітектурні рішення — 12 місяців. Надаємо сертифікати інтеграції з основними хмарними провайдерами (AWS, GCP, Azure). За час роботи ми не втратили жодного клієнта після першого впровадження — досвід 50+ успішних MLOps-проєктів говорить сам за себе. Отримайте консультацію щодо побудови MLOps платформи вже сьогодні.