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
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
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
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)
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
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
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
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
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
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
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
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
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)
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)
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
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
- Agent Benchmarks — Benchmarks for online eval
- Evaluation Frameworks — Frameworks
- Metrics — Detailed metrics
- Offline Eval — Offline evaluation
- Red Teaming — Red teaming
- Observability — Monitoring
- Reliability — Reliability
- Security — Security
- Reliability — Reliability
- Deployment Architecture — Deployment for online eval
- Research Directions — 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.