# Online Evaluation

Evaluating agents in production with live traffic, A/B testing, and real-time monitoring.

---

## Online Evaluation Concept

Online evaluation measures agent performance on live traffic with real users and real environments.

```
Online Evaluation Pipeline
├── Traffic Splitting
│   ├── A/B Testing
│   ├── Canary Deployment
│   ├── Shadow Mode
│   └── Interleaving
├── Real-Time Metrics
│   ├── Streaming Aggregation
│   ├── Alerting
│   └── Anomaly Detection
├── User Feedback
│   ├── Explicit (Ratings, Surveys)
│   ├── Implicit (Behavior, Dwell Time)
│   └── Implicit (Corrections, Re-phrasing)
└── Continuous Learning
    ├── Online Evaluation
    ├── Preference Learning
    └── Model Updates
```

---

## Traffic Splitting Strategies

### 1. A/B Testing
```python
class ABTestManager:
    def __init__(self):
        self.experiments: dict[str, ABExperiment] = {}
    
    def create_experiment(self, name: str, variants: list[Variant], 
                          traffic_split: dict[str, float] = None) -> ABExperiment:
        exp = ABExperiment(
            name=name,
            variants=variants,
            traffic_split=traffic_split or {v.name: 1.0/len(variants) for v in variants},
            status="running",
            created_at=datetime.utcnow()
        )
        self.experiments[name] = exp
        return exp
    
    def assign_variant(self, experiment_name: str, user_id: str) -> Variant:
        exp = self.experiments[experiment_name]
        
        # Deterministic assignment via hashing
        hash_val = int(hashlib.md5(f"{experiment_name}:{user_id}".encode()).hexdigest(), 16)
        bucket = hash_val % 10000 / 10000  # 0.0 - 1.0
        
        cumulative = 0.0
        for variant_name, weight in exp.traffic_split.items():
            cumulative += weight
            if bucket < cumulative:
                return exp.get_variant(variant_name)
        
        return exp.variants[-1]  # Fallback
```

### 2. Canary Deployment
```python
class CanaryDeployment:
    def __init__(self):
        self.canaries: dict[str, Canary] = {}
    
    async def start_canary(self, name: str, new_version: str, 
                           canary_percentage: float = 0.05,
                           success_criteria: SuccessCriteria = None) -> Canary:
        canary = Canary(
            name=name,
            new_version=new_version,
            percentage=canary_percentage,
            success_criteria=success_criteria or SuccessCriteria(),
            status="running",
            started_at=datetime.utcnow()
        )
        self.canaries[name] = canary
        
        # Gradual ramp
        asyncio.create_task(self.ramp_up(canary))
        
        return canary
    
    async def ramp_up(self, canary: Canary):
        stages = [0.05, 0.1, 0.25, 0.5, 0.75, 1.0]
        for pct in stages:
            if pct <= canary.percentage:
                continue
            
            canary.percentage = pct
            await asyncio.sleep(3600)  # 1 hour per stage
            
            # Check health
            health = await self.check_health(canary)
            if not health.healthy:
                await self.rollback(canary)
                return
        
        canary.status = "completed"
```

### 3. Shadow Mode
```python
class ShadowMode:
    def __init__(self, primary_agent: Agent, shadow_agent: Agent):
        self.primary = primary_agent
        self.shadow = shadow_agent
    
    async def process(self, request: Request) -> PrimaryResponse:
        # Primary handles request
        primary_response = await self.primary.process(request)
        
        # Shadow processes in background (fire-and-forget)
        asyncio.create_task(self.shadow_process(request))
        
        return primary_response
    
    async def shadow_process(self, request: Request):
        try:
            shadow_response = await self.shadow.process(request)
            
            # Log comparison
            await self.log_comparison(request, primary_response, shadow_response)
            
            # Flag discrepancies
            if self.significant_difference(primary_response, shadow_response):
                await self.alert_discrepancy(request, primary_response, shadow_response)
        except Exception as e:
            logger.error(f"Shadow processing failed: {e}")
```

### 4. Interleaving (Ranking Evaluation)
```python
class InterleavingExperiment:
    def __init__(self):
        self.experiments: dict[str, InterleavingExp] = {}
    
    def create_interleaving(self, name: str, rankers: list[Ranker]) -> InterleavingExp:
        exp = InterleavingExp(
            name=name,
            rankers=rankers,
            method="team_draft"  # or "balanced", "probabilistic"
        )
        self.experiments[name] = exp
        return exp
    
    def interleave_rankings(self, rankings: list[list[Item]]) -> list[Item]:
        """Team-draft interleaving"""
        result = []
        iterators = [iter(r) for r in rankings]
        used = set()
        
        while len(result) < sum(len(r) for r in rankings):
            for it in iterators:
                try:
                    item = next(it)
                    if item.id not in used:
                        result.append(item)
                        used.add(item.id)
                        break
                except StopIteration:
                    pass
        return result
```

---

## Real-Time Metrics

### Streaming Aggregation
```python
class StreamingMetrics:
    def __init__(self, window_seconds: int = 60):
        self.window = window_seconds
        self.buckets: dict[int, MetricBucket] = defaultdict(MetricBucket)
    
    def record(self, metric: str, value: float, labels: dict = None):
        bucket_key = int(time.time() // self.window) * self.window
        self.buckets[bucket_key].add(metric, value, labels)
    
    def get_window(self, metric: str, labels: dict = None, 
                   window_seconds: int = None) -> MetricSummary:
        window = window_seconds or self.window
        cutoff = time.time() - window
        
        values = []
        for bucket_key, bucket in self.buckets.items():
            if bucket_key >= cutoff:
                values.extend(bucket.get(metric, labels))
        
        if not values:
            return MetricSummary(count=0)
        
        return MetricSummary(
            count=len(values),
            mean=statistics.mean(values),
            std=statistics.stdev(values) if len(values) > 1 else 0,
            min=min(values),
            max=max(values),
            p50=np.percentile(values, 50),
            p95=np.percentile(values, 95),
            p99=np.percentile(values, 99)
        )
```

### Alerting
```python
class AlertManager:
    def __init__(self):
        self.rules: list[AlertRule] = []
        self.active_alerts: dict[str, Alert] = {}
    
    def add_rule(self, rule: AlertRule):
        self.rules.append(rule)
    
    async def check_alerts(self, metrics: dict[str, MetricSummary]):
        for rule in self.rules:
            value = metrics.get(rule.metric)
            if not value:
                continue
            
            triggered = rule.condition(value)
            
            if triggered and rule.name not in self.active_alerts:
                alert = Alert(
                    rule=rule.name,
                    value=value.value,
                    threshold=rule.threshold,
                    severity=rule.severity,
                    message=rule.message.format(value=value.value),
                    started_at=datetime.utcnow()
                )
                self.active_alerts[rule.name] = alert
                await self.send_alert(alert)
            
            elif not triggered and rule.name in self.active_alerts:
                alert = self.active_alerts.pop(rule.name)
                alert.resolved_at = datetime.utcnow()
                await self.send_resolution(alert)
```

### Anomaly Detection
```python
class AnomalyDetector:
    def __init__(self, sensitivity: float = 3.0):
        self.sensitivity = sensitivity  # Z-score threshold
        self.history: dict[str, list[float]] = defaultdict(list)
        self.max_history = 10000
    
    def record(self, metric: str, value: float):
        self.history[metric].append(value)
        if len(self.history[metric]) > self.max_history:
            self.history[metric].pop(0)
    
    def detect(self, metric: str, value: float) -> AnomalyResult:
        history = self.history.get(metric, [])
        if len(history) < 30:
            return AnomalyResult(anomaly=False)
        
        mean = statistics.mean(history)
        stdev = statistics.stdev(history) if len(history) > 1 else 0
        
        if stdev == 0:
            return AnomalyResult(anomaly=False)
        
        z_score = abs(value - mean) / stdev
        anomaly = z_score > self.sensitivity
        
        return AnomalyResult(
            anomaly=anomaly,
            z_score=z_score,
            value=value,
            mean=mean,
            stdev=stdev
        )
```

---

## User Feedback Collection

### Explicit Feedback
```python
class ExplicitFeedbackCollector:
    def __init__(self):
        self.feedback_store = FeedbackStore()
    
    async def collect_explicit(self, interaction_id: str, feedback: ExplicitFeedback):
        feedback_record = FeedbackRecord(
            interaction_id=interaction_id,
            feedback_type=feedback.type,  # rating, thumbs, text, correction
            value=feedback.value,
            user_id=feedback.user_id,
            timestamp=datetime.utcnow()
        )
        await self.feedback_store.save(feedback_record)
        
        # Trigger relearning if negative
        if feedback.type == "rating" and feedback.value <= 2:
            await self.queue_for_review(interaction_id)
    
    def get_feedback_stats(self, window: TimeWindow) -> FeedbackStats:
        feedback = self.feedback_store.query(window)
        
        return FeedbackStats(
            total=len(feedback),
            avg_rating=statistics.mean(f.value for f in feedback if f.type == "rating"),
            thumbs_up=sum(1 for f in feedback if f.type == "thumbs" and f.value == "up"),
            thumbs_down=sum(1 for f in feedback if f.type == "thumbs" and f.value == "down"),
            corrections=sum(1 for f in feedback if f.type == "correction"),
            nps=self.calculate_nps(feedback)
        )
```

### Implicit Feedback
```python
class ImplicitFeedbackCollector:
    def __init__(self):
        self.events = EventStore()
    
    async def record_implicit(self, event: ImplicitEvent):
        await self.events.append(event)
    
    def infer_satisfaction(self, session_id: str) -> float:
        """Infer satisfaction from behavioral signals"""
        events = self.events.get_session(session_id)
        
        signals = {
            "completion": 1.0 if any(e.type == "task_complete" for e in events) else 0.0,
            "rephrase_count": max(0, 1.0 - 0.1 * sum(1 for e in events if e.type == "rephrase")),
            "dwell_time": min(1.0, sum(e.duration for e in events) / 300),  # 5 min = 1.0
            "scroll_depth": max(e.scroll_depth for e in events if e.type == "scroll"),
            "clicks": min(1.0, sum(1 for e in events if e.type == "click") / 10),
            "copy_events": min(1.0, sum(1 for e in events if e.type == "copy") / 5),
            "correction": 0.0 if any(e.type == "correction" for e in events) else 1.0
        }
        
        # Weighted combination
        weights = {"completion": 0.3, "rephrase": 0.15, "dwell": 0.15, 
                   "scroll": 0.1, "clicks": 0.1, "copy": 0.1, "correction": 0.2}
        
        return sum(signals[k] * weights[k] for k in signals)
```

### Feedback Loop
```python
class FeedbackLoop:
    def __init__(self, agent: Agent, feedback_store: FeedbackStore):
        self.agent = agent
        self.store = feedback_store
    
    async def process_feedback(self):
        # Get negative feedback
        negative = await self.store.get_negative_feedback(hours=24)
        
        # Cluster by issue
        clusters = self.cluster_feedback(negative)
        
        # Generate fixes
        for cluster in clusters:
            fix = await self.generate_fix(cluster)
            if fix.confidence > 0.8:
                await self.apply_fix(fix)
        
        # Retrain preference model
        await self.update_preference_model()
```

---

## Online A/B Testing Framework

### Experiment Platform
```python
class ExperimentPlatform:
    def __init__(self):
        self.experiments: dict[str, Experiment] = {}
        self.assignment_service = AssignmentService()
        self.metrics_collector = MetricsCollector()
    
    def create_experiment(self, config: ExperimentConfig) -> Experiment:
        exp = Experiment(
            id=str(uuid4()),
            name=config.name,
            hypothesis=config.hypothesis,
            variants=config.variants,
            metrics=config.metrics,
            traffic_allocation=config.traffic_allocation,
            start_date=config.start_date,
            end_date=config.end_date,
            status="draft"
        )
        self.experiments[exp.id] = exp
        return exp
    
    def start_experiment(self, experiment_id: str):
        exp = self.experiments[experiment_id]
        exp.status = "running"
        exp.started_at = datetime.utcnow()
    
    def get_variant(self, experiment_id: str, user_id: str) -> Variant:
        exp = self.experiments[experiment_id]
        return self.assignment_service.assign(exp, user_id)
    
    def record_metric(self, experiment_id: str, user_id: str, 
                      metric_name: str, value: float):
        variant = self.get_variant(experiment_id, user_id)
        self.metrics_collector.record(
            experiment_id=experiment_id,
            variant=variant.name,
            metric=metric_name,
            value=value
        )
```

### Statistical Analysis
```python
class ExperimentAnalyzer:
    @staticmethod
    def analyze_experiment(experiment: Experiment) -> ExperimentResult:
        variant_data = experiment.get_variant_data()
        
        results = {}
        for variant_name, data in variant_data.items():
            results[variant_name] = VariantResult(
                sample_size=len(data),
                conversions=sum(1 for d in data if d.converted),
                conversion_rate=sum(1 for d in data if d.converted) / len(data),
                avg_revenue=statistics.mean(d.revenue for d in data),
                avg_engagement=statistics.mean(d.engagement for d in data)
            )
        
        # Statistical significance
        significance = StatisticalAnalyzer.significance_test(
            variant_data[experiment.control].converted,
            variant_data[experiment.treatment].converted
        )
        
        return ExperimentResult(
            variant_results=results,
            significance=significance,
            recommendation="launch" if significance.significant and 
                          results[experiment.treatment].conversion_rate > 
                          results[experiment.control].conversion_rate else "hold"
        )
```

---

## Sequential Testing

### Sequential Probability Ratio Test (SPRT)
```python
class SPRT:
    def __init__(self, alpha: float = 0.05, beta: float = 0.2, 
                 p0: float = 0.5, p1: float = 0.55):
        self.alpha = alpha  # Type I error
        self.beta = beta    # Type II error
        self.p0 = p0        # Null hypothesis probability
        self.p1 = p1        # Alternative hypothesis probability
        
        self.A = (1 - beta) / alpha
        self.B = beta / (1 - alpha)
    
    def update(self, success: bool) -> SPRTResult:
        self.n += 1
        if success:
            self.successes += 1
        
        # Likelihood ratio
        lr = (self.p1 ** self.successes * (1 - self.p1) ** (self.n - self.successes)) / \
             (self.p0 ** self.successes * (1 - self.p0) ** (self.n - self.successes))
        
        if lr >= self.A:
            return SPRTResult(decision="accept_h1", likelihood_ratio=lr)
        elif lr <= self.B:
            return SPRTResult(decision="accept_h0", likelihood_ratio=lr)
        else:
            return SPRTResult(decision="continue", likelihood_ratio=lr)
```

---

## Continuous Learning

### Preference Learning (RLHF)
```python
class PreferenceLearner:
    def __init__(self, model: Model, reward_model: RewardModel):
        self.model = model
        self.reward_model = reward_model
        self.optimizer = torch.optim.Adam(model.parameters(), lr=1e-5)
    
    async def learn_from_preferences(self, preferences: list[PreferencePair]):
        # DPO (Direct Preference Optimization)
        for pair in preferences:
            # chosen > rejected
            chosen_logprob = self.model.logprob(pair.chosen)
            rejected_logprob = self.model.logprob(pair.rejected)
            
            # DPO loss
            loss = -F.logsigmoid(self.beta * (chosen_logprob - rejected_logprob))
            loss.backward()
        
        self.optimizer.step()
        self.optimizer.zero_grad()
```

### Online Model Update
```python
class OnlineModelUpdater:
    def __init__(self, model: Model, update_frequency: int = 1000):
        self.model = model
        self.update_freq = update_frequency
        self.batch_buffer = []
        self.step_count = 0
    
    async def add_experience(self, experience: Experience):
        self.batch_buffer.append(experience)
        self.step_count += 1
        
        if self.step_count % self.update_freq == 0:
            await self.update_model()
    
    async def update_model(self):
        if len(self.batch_buffer) < 32:
            return
        
        # Mini-batch update
        batch = self.batch_buffer[-32:]
        self.batch_buffer = self.batch_buffer[:-32]
        
        # Gradient step
        loss = self.compute_loss(batch)
        loss.backward()
        self.optimizer.step()
        self.optimizer.zero_grad()
```

---

## Related Notes

- [[Evaluation/Agent-Benchmarks.md]] — Benchmarks for online eval
- [[Evaluation/Evaluation-Frameworks.md]] — Frameworks
- [[Evaluation/Metrics.md]] — Detailed metrics
- [[Evaluation/Offline-Eval.md]] — Offline evaluation
- [[Evaluation/Red-Teaming.md]] — Red teaming
- [[Production/Observability.md]] — Monitoring
- [[Production/Reliability.md]] — Reliability
- [[Production/Security.md]] — Security
- [[Production/Reliability.md]] — Reliability
- [[Production/Deployment-Architecture.md]] — Deployment for online eval
- [[Frontiers/Research-Directions.md]] — Online learning research

---

*Sources: "Trustworthy Online Controlled Experiments" (Kohavi et al.), "A/B Testing" (Kohavi & Longbotham), "Sequential Analysis" (Wald), "Online Learning" (Shalev-Shwartz), "Bandit Algorithms" (Lattimore & Szepesvári), "RLHF" (Ouyang et al.), "DPO" (Rafailov et al.), "Constitutional AI" (Bai et al.), Netflix Experimentation, Airbnb Experimentation, Google Experimentation, Microsoft Experimentation Platform, Spotify Experimentation, Eppo, Statsig, GrowthBook, Optimizely, VWO, LaunchDarkly, Split.io.*