Workflow 003: Continuous Monitoring
Overview
| Property | Value |
|---|---|
| Workflow ID | 003 |
| Title | Continuous Monitoring |
| Complexity | Medium |
| Duration | Ongoing |
Purpose
Monitor agent quality over time in production, detect drift, and trigger re-optimization when performance degrades.
Architecture
┌─────────────────────────────────────────────────────────────────────┐
│ CONTINUOUS MONITORING │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ PRODUCTION │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ User │───►│ Agent │───►│ Response │ │ │
│ │ │ Request │ │ (Deployed) │ │ │ │ │
│ │ └─────────────┘ └──────┬──────┘ └─────────────┘ │ │
│ │ │ │ │
│ │ │ Sample (10-20%) │ │
│ │ ▼ │ │
│ │ ┌─────────────┐ │ │
│ │ │ Monitoring │ │ │
│ │ │ Queue │ │ │
│ │ └──────┬──────┘ │ │
│ └───────────────────────────┼──────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ MONITORING LOOP │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Run Eval │───►│ Calculate │───►│ Check │ │ │
│ │ │ (Async) │ │ Metrics │ │ Thresholds │ │ │
│ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │
│ │ │ │ │
│ │ ┌─────────────────────┼───────────┐ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ [HEALTHY] [WARNING] [ALERT] │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ [LOG] [NOTIFY] [RE-OPT]│ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘Prerequisites
- Completed Workflow 001: Self-Evolving Loop
- Agent deployed in production
- Monitoring infrastructure (logging, alerts)
Alert Thresholds
┌─────────────────────────────────────────────────────────────────────┐
│ ALERT THRESHOLDS │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Metric │ Healthy │ Warning │ Alert │
│ ──────────────────────────┼─────────┼─────────┼──────────────────│
│ Average Score │ > 0.85 │ 0.75-85 │ < 0.75 │
│ Pass Rate │ > 90% │ 75-90% │ < 75% │
│ Chemical Name Accuracy │ > 95% │ 85-95% │ < 85% │
│ Score Variance │ < 0.1 │ 0.1-0.2 │ > 0.2 │
│ Consecutive Failures │ 0 │ 1-2 │ >= 3 │
│ │
│ Actions: │
│ ├─► HEALTHY: Log metrics, continue monitoring │
│ ├─► WARNING: Notify team, increase sampling rate │
│ └─► ALERT: Trigger re-optimization workflow │
│ │
└─────────────────────────────────────────────────────────────────────┘Implementation
Step 1: Define Monitoring Configuration
python
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
from enum import Enum
class AlertLevel(Enum):
HEALTHY = "healthy"
WARNING = "warning"
ALERT = "alert"
@dataclass
class MonitoringConfig:
"""Configuration for continuous monitoring."""
eval_id: str
sample_rate: float = 0.1 # 10% of requests
window_size: int = 100 # Number of samples to aggregate
# Thresholds
score_healthy: float = 0.85
score_warning: float = 0.75
pass_rate_healthy: float = 0.90
pass_rate_warning: float = 0.75
max_consecutive_failures: int = 3
max_score_variance: float = 0.2
# Intervals
check_interval: timedelta = field(default_factory=lambda: timedelta(minutes=5))
alert_cooldown: timedelta = field(default_factory=lambda: timedelta(hours=1))
@dataclass
class MetricWindow:
"""Sliding window of metrics."""
scores: list[float] = field(default_factory=list)
passed: list[bool] = field(default_factory=list)
timestamps: list[datetime] = field(default_factory=list)
consecutive_failures: int = 0
last_alert: Optional[datetime] = None
def add(self, score: float, passed: bool):
"""Add a new metric to the window."""
self.scores.append(score)
self.passed.append(passed)
self.timestamps.append(datetime.utcnow())
if passed:
self.consecutive_failures = 0
else:
self.consecutive_failures += 1
def trim(self, window_size: int):
"""Keep only the most recent samples."""
if len(self.scores) > window_size:
self.scores = self.scores[-window_size:]
self.passed = self.passed[-window_size:]
self.timestamps = self.timestamps[-window_size:]
@property
def average_score(self) -> float:
return sum(self.scores) / len(self.scores) if self.scores else 0.0
@property
def pass_rate(self) -> float:
return sum(self.passed) / len(self.passed) if self.passed else 0.0
@property
def score_variance(self) -> float:
if len(self.scores) < 2:
return 0.0
mean = self.average_score
return sum((s - mean) ** 2 for s in self.scores) / len(self.scores)Step 2: Create Monitoring Service
python
import asyncio
import random
from openai import OpenAI
client = OpenAI()
class MonitoringService:
"""Service for continuous agent monitoring."""
def __init__(self, config: MonitoringConfig):
self.config = config
self.metrics = MetricWindow()
self.is_running = False
def should_sample(self) -> bool:
"""Determine if this request should be sampled."""
return random.random() < self.config.sample_rate
async def evaluate_sample(self, section: str, summary: str) -> tuple[float, bool]:
"""Run evaluation on a sampled request."""
eval_run = run_eval(
eval_id=self.config.eval_id,
section=section,
summary=summary
)
run_output = poll_eval_run(
eval_id=self.config.eval_id,
run_id=eval_run.id
)
grader_scores = parse_eval_run_output(run_output)
avg_score = calculate_grader_score(grader_scores)
passed = is_lenient_pass(grader_scores, avg_score)
return avg_score, passed
def record_metric(self, score: float, passed: bool):
"""Record a new metric."""
self.metrics.add(score, passed)
self.metrics.trim(self.config.window_size)
def check_health(self) -> AlertLevel:
"""Check current health status."""
if len(self.metrics.scores) < 10:
return AlertLevel.HEALTHY # Not enough data
# Check consecutive failures
if self.metrics.consecutive_failures >= self.config.max_consecutive_failures:
return AlertLevel.ALERT
# Check score variance
if self.metrics.score_variance > self.config.max_score_variance:
return AlertLevel.WARNING
# Check average score
avg = self.metrics.average_score
if avg < self.config.score_warning:
return AlertLevel.ALERT
elif avg < self.config.score_healthy:
return AlertLevel.WARNING
# Check pass rate
rate = self.metrics.pass_rate
if rate < self.config.pass_rate_warning:
return AlertLevel.ALERT
elif rate < self.config.pass_rate_healthy:
return AlertLevel.WARNING
return AlertLevel.HEALTHY
def get_status_report(self) -> dict:
"""Generate status report."""
return {
"timestamp": datetime.utcnow().isoformat(),
"samples": len(self.metrics.scores),
"average_score": self.metrics.average_score,
"pass_rate": self.metrics.pass_rate,
"score_variance": self.metrics.score_variance,
"consecutive_failures": self.metrics.consecutive_failures,
"health": self.check_health().value,
}Step 3: Create Alert Handler
python
from abc import ABC, abstractmethod
class AlertHandler(ABC):
"""Base class for alert handlers."""
@abstractmethod
async def handle(self, level: AlertLevel, report: dict):
pass
class LoggingAlertHandler(AlertHandler):
"""Log alerts to console/file."""
async def handle(self, level: AlertLevel, report: dict):
if level == AlertLevel.HEALTHY:
print(f"[HEALTHY] Score: {report['average_score']:.3f}, Pass Rate: {report['pass_rate']:.1%}")
elif level == AlertLevel.WARNING:
print(f"[WARNING] Score: {report['average_score']:.3f}, Pass Rate: {report['pass_rate']:.1%}")
print(f" Variance: {report['score_variance']:.3f}")
else:
print(f"[ALERT] Score: {report['average_score']:.3f}, Pass Rate: {report['pass_rate']:.1%}")
print(f" Consecutive Failures: {report['consecutive_failures']}")
class ReoptimizationHandler(AlertHandler):
"""Trigger re-optimization on alert."""
def __init__(self, optimization_callback):
self.callback = optimization_callback
async def handle(self, level: AlertLevel, report: dict):
if level == AlertLevel.ALERT:
print("[ALERT] Triggering re-optimization...")
await self.callback(report)
class WebhookAlertHandler(AlertHandler):
"""Send alerts to webhook (Slack, PagerDuty, etc.)."""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
async def handle(self, level: AlertLevel, report: dict):
import aiohttp
if level in [AlertLevel.WARNING, AlertLevel.ALERT]:
payload = {
"level": level.value,
"message": f"Agent quality {level.value}: score={report['average_score']:.3f}",
"report": report
}
async with aiohttp.ClientSession() as session:
await session.post(self.webhook_url, json=payload)Step 4: Integrate with Production
python
class ProductionMonitor:
"""Integration layer for production monitoring."""
def __init__(self, config: MonitoringConfig, handlers: list[AlertHandler]):
self.service = MonitoringService(config)
self.handlers = handlers
self.config = config
async def process_request(self, section: str, summary: str):
"""Process a production request with optional monitoring."""
if self.service.should_sample():
# Run async evaluation
asyncio.create_task(self._evaluate_and_record(section, summary))
async def _evaluate_and_record(self, section: str, summary: str):
"""Evaluate sample and record metrics."""
try:
score, passed = await self.service.evaluate_sample(section, summary)
self.service.record_metric(score, passed)
# Check health and handle alerts
level = self.service.check_health()
report = self.service.get_status_report()
for handler in self.handlers:
await handler.handle(level, report)
except Exception as e:
print(f"Monitoring error: {e}")
async def run_periodic_check(self):
"""Run periodic health checks."""
while True:
await asyncio.sleep(self.config.check_interval.total_seconds())
level = self.service.check_health()
report = self.service.get_status_report()
print(f"\n[Periodic Check] {report['timestamp']}")
for handler in self.handlers:
await handler.handle(level, report)Step 5: Usage Example
python
async def main():
"""Example monitoring setup."""
# Configuration
config = MonitoringConfig(
eval_id="eval_...",
sample_rate=0.1, # 10% sampling
window_size=100,
score_healthy=0.85,
score_warning=0.75,
)
# Alert handlers
handlers = [
LoggingAlertHandler(),
# WebhookAlertHandler("https://hooks.slack.com/..."),
# ReoptimizationHandler(trigger_reoptimization),
]
# Create monitor
monitor = ProductionMonitor(config, handlers)
# Start periodic checks
asyncio.create_task(monitor.run_periodic_check())
# Simulate production requests
sections = [...] # Your production data
agent = ... # Your deployed agent
for section in sections:
# Generate summary
result = await Runner.run(agent, section)
summary = result.final_output
# Process with monitoring
await monitor.process_request(section, summary)
await asyncio.sleep(0.1) # Simulate request rate
if __name__ == "__main__":
asyncio.run(main())Monitoring Dashboard Metrics
┌─────────────────────────────────────────────────────────────────────┐
│ DASHBOARD METRICS │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Real-time Metrics: │
│ ├─► Average Score (rolling window) │
│ ├─► Pass Rate (percentage) │
│ ├─► Score Variance (stability indicator) │
│ ├─► Consecutive Failures (trend indicator) │
│ └─► Requests per minute (throughput) │
│ │
│ Historical Metrics: │
│ ├─► Daily average score trend │
│ ├─► Weekly pass rate comparison │
│ ├─► Alert frequency by type │
│ └─► Re-optimization trigger count │
│ │
│ Breakdown by Grader: │
│ ├─► chemical_name_grader: accuracy over time │
│ ├─► word_length_deviation_grader: compliance rate │
│ ├─► cosine_similarity: average similarity │
│ └─► llm_as_judge: subjective quality trend │
│ │
└─────────────────────────────────────────────────────────────────────┘Verification Checklist
- [ ] Monitoring configuration defined
- [ ] Sample rate appropriate for volume
- [ ] Thresholds calibrated for use case
- [ ] Alert handlers configured
- [ ] Periodic checks running
- [ ] Dashboard/logging accessible
- [ ] Re-optimization trigger tested