פלטפורמת בדיקות A/B לסוכני AI: עיצוב ויישום

פלטפורמת בדיקות A/B לסוכני בינה מלאכותית: תכנון ויישום

בסביבת ייצור נתקלנו במצב: גרסת סוכן חדשה עם פרומפט מכוון שיפרה את שיעור ההצלחה במשימות מ-78% ל-82% במשך שבוע. אבל חודש לאחר מכן, המדד חזר לקו הבסיס. הסיבה הייתה סחיפת נתונים וחלוקת תעבורה שגויה. בדיקות A/B נאיביות ללא גיבוב עקבי ובקרה סטטיסטית מובילות לשגיאות. תכננו מערכת שפותר את הבעיות הללו: מבטיחה p-value < 0.05, עוצרת את הניסוי אוטומטית עם ירידה בביצועים, ודורשת 40% פחות דוגמאות בזכות עיצוב אופטימלי. החיסכון בניסויים יכול להגיע ל-30% מהתקציב — עבור חברה עם 10 סוכנים, זה יכול להיות בין $9k–13k בשנה.

אילו בעיות אנחנו פותרים

• חוסר יציבות מדדים. שיעור ההזיות יכול לנוע בין 2% ל-12% בהתאם למורכבות השאילתה. ללא בקרה קפדנית, אי אפשר להבחין בין שיפור לרעש.
• גודל מדגם. זיהוי ירידה של 1% בשיעור ההזיות ברמת בסיס של 3% דורש לפחות 500 דוגמאות לכל וריאנט. המערכת שלנו מייעלת את גודל המדגם ב-30% באמצעות ריבוד.
• תוצאות חיוביות שגויות. השוואות מרובות ועצירה מוקדמת הן טעויות נפוצות. אנו משתמשים בכללי עצירה אוטומטית המתחשבים בגודל מדגם מינימלי ומתקנים ערכי p בשיטת Bonferroni.

איך פועל גיבוב עקבי

גיבוב עקבי מקשר משתמש לווריאנט ניסוי על בסיס גיבוב MD5 של user_id ו-experiment_id. אנו משתמשים בו כך שכל משתמש תמיד נופל לאותה קבוצה. זה מבטל אפקטי למידה חוזרת ומפחית שונות עד פי 5 בהשוואה לחלוקה אקראית. גיבוב עקבי מבטיח חלוקה יציבה גם כשמספר הניסויים משתנה.

עיצוב ניסוי A/B

הליבה היא מחלקת הנתונים AgentExperiment, המתארת את כל פרמטרי הניסוי: שם סוכן, גרסאות בקרה וטיפול, חלק התעבורה לטיפול, השערה, מדד ראשי, גודל מדגם מינימלי ומשך מקסימלי. הנה דוגמה:

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. הגדר את המדד הראשי וההשערה. לדוגמה, "הפרומפט החדש יגדיל את שיעור ההצלחה במשימות מ-78% ל-82%."
2. הגדר פרמטרים במחלקת הנתונים 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 : גרסת בקרה, גרסת טיפול, חלק תעבורה (בדרך כלל 10-20%).
3. חבר את AgentExperiment, המשתמש בגיבוב עקבי כדי להפנות משתמשים לווריאנט המתאים.
4. התחל מעקב מדדים: המערכת אוספת מדדים ראשיים ומשניים בזמן אמת.
5. המתן עד להצטברות min_samples (200-500), ולאחר מכן הרץ את ExperimentRouter. הוא מבצע מבחן z או מבחן t ומחזיר p-value ו-lift.
6. אם p-value < 0.05 וה-lift חיובי, המערכת ממליצה לשחרר את הטיפול. אם יש ירידה, עצירה אוטומטית מסיימת את הניסוי.

יישום הפלטפורמה

היישום כולל ניתוב, מעקב ועצירה אוטומטית.

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 לפרופורציות.

כללי עצירה אוטומטית

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

רכיבים אלה פועלים יחד: הנתב מפנה תעבורה, הגשש אוסף מדדים, המנתח מחשב מובהקות סטטיסטית, ומנגנוני הגנה מחליטים אם להמשיך או לעצור.

איך מאורגן תהליך העבודה

ציר הזמן ליישום נע בין שבועיים לחודשיים. העלות נקבעת לאחר ביקורת מערכת.

מה כלול בעבודה

• תיעוד ניסוי — תיאור השערות, מדדים, עיצוב.
• קוד נתב וגשש — אינטגרציה עם התשתית שלך.
• לוח מחוונים למדדים — ויזואליזציה בזמן אמת של תוצאות הניסוי.
• מדריך השקה — הוראות שלב-אחר-שלב לצוות שלך.

מדדים וחשיבותם

חשיבות קריטית של בדיקות A/B לסוכני בינה מלאכותית

ללא ניסויים קפדניים, אי אפשר להבחין בין שיפור אמיתי לשונות אקראית. זה חשוב במיוחד למדדים כמו שיעור הזיות, שבו הבדל של 1-2% יכול להיות משמעותי. המערכת שלנו מבטיחה p-value < 0.05 ועוצרת את הניסוי אוטומטית עם זיהוי ירידה, וחוסכת זמן מפתחים. גיבוב עקבי מספק יציבות טובה פי 5 מחלוקה אקראית.

טעויות אופייניות

• בחירה שגויה של מדד ראשי. אם המדד אינו רגיש, הניסוי לא מניב תוצאה. בחר מדד שמשפיע ישירות על חוויית המשתמש.
• התעלמות מבדיקות מרובות. כשאתה בודק מספר מדדים, התאם את רמת המובהקות (לדוגמה, תיקון Bonferroni). אחרת, אתה מסתכן בתוצאות חיוביות שגויות.
• עצירה מוקדמת. אל תפריע לניסוי בתוצאה המשמעותית הראשונה; המתן עד להצטברות גודל המדגם המינימלי (200-500 דוגמאות).

יש לנו ניסיון של 5+ שנים ב-AI/ML ומעל 30 פרויקטים של בדיקות A/B לסוכנים. בקש ביקורת על מערכת בדיקות ה-A/B שלך וקבל ייעוץ מהנדס. צור קשר כדי לדון בפרויקט שלך.