08: Observability, Evaluation, and AgentOps
Treating AI Agents as Production Systems
Pillar: #8 of 12 - Supporting Priority
Timeline: 6 Weeks
Dependencies: None
Expected ROI: Real-time agent health visibility, 50% cost reduction via optimization, data-driven routing decisions
Executive Summary
AI agents are production systems that need the same observability as any critical service. This pillar implements comprehensive telemetry (traces, metrics, logs), evaluation harnesses for quality measurement, and AgentOps dashboards that provide real-time visibility into agent performance, cost, and behavior.
Key Insight: You can't improve what you can't measure.
Problem Statement
Current State: Blind Agents
# File: adapters/claude_adapter.py
# Current: Fire and forget
async def invoke(self, context: Context) -> Response:
response = await self.client.messages.create(
model=self.model,
messages=context.messages,
)
return response # No telemetry, no cost tracking, no quality measurement
What We Don't Know
Current Visibility Gaps:
1. Cost Attribution
- Which workflows are expensive?
- Which personas use the most tokens?
- Where is the waste?
2. Quality Measurement
- Are outputs improving or degrading?
- Which model versions perform better?
- What's the failure rate by query type?
3. Performance Analysis
- Where are the bottlenecks?
- What's the latency distribution?
- How do retries affect throughput?
4. Behavioral Patterns
- What patterns lead to failures?
- How do personas interact?
- What triggers escalations?
The Cost of Blindness
Without observability:
- Can't identify expensive patterns
- Can't prove quality improvements
- Can't debug production issues
- Can't optimize routing
- Can't justify investments
Result:
- Flying blind
- Reactive firefighting
- No continuous improvement
Solution Architecture
┌─────────────────────────────────────────────────┐
│ AGENTOPS PLATFORM │
├─────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐│
│ │ Traces │ │ Metrics │ │ Logs ││
│ │ (OpenTel) │ │(Prometheus) │ │ (Loki) ││
│ └──────┬──────┘ └──────┬──────┘ └────┬─────┘│
│ │ │ │ │
│ └────────┬───────┴──────────────┘ │
│ ▼ │
│ ┌────────────────┐ │
│ │ Data Lake │ │
│ │ (Analysis) │ │
│ └───────┬────────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────┐ ┌─────────┐ ┌────────┐ │
│ │Evals │ │Dashboard│ │Alerts │ │
│ │Harness│ │(Grafana)│ │(PagerD)│ │
│ └──────┘ └─────────┘ └────────┘ │
│ │
└─────────────────────────────────────────────────┘
Core Components
1. Agent Telemetry Collector
from opentelemetry import trace
from opentelemetry.metrics import get_meter
class AgentTelemetryCollector:
"""
Collect comprehensive telemetry from agent operations.
"""
def __init__(self):
self.tracer = trace.get_tracer("maestro-agents")
self.meter = get_meter("maestro-agents")
# Metrics
self.token_counter = self.meter.create_counter(
"agent.tokens.total",
description="Total tokens used",
)
self.latency_histogram = self.meter.create_histogram(
"agent.latency.seconds",
description="Agent invocation latency",
)
self.cost_counter = self.meter.create_counter(
"agent.cost.dollars",
description="LLM cost in dollars",
)
self.success_counter = self.meter.create_counter(
"agent.invocations.total",
description="Total invocations by status",
)
@contextmanager
def trace_invocation(
self,
persona: str,
model: str,
workflow_id: str
) -> Span:
with self.tracer.start_as_current_span(
"agent.invoke",
attributes={
"agent.persona": persona,
"agent.model": model,
"workflow.id": workflow_id,
}
) as span:
start = time.monotonic()
try:
yield span
status = "success"
except Exception as e:
status = "error"
span.record_exception(e)
raise
finally:
duration = time.monotonic() - start
self.latency_histogram.record(
duration,
{"persona": persona, "model": model, "status": status}
)
self.success_counter.add(
1,
{"persona": persona, "model": model, "status": status}
)
def record_tokens(
self,
persona: str,
model: str,
input_tokens: int,
output_tokens: int
):
self.token_counter.add(
input_tokens,
{"persona": persona, "model": model, "direction": "input"}
)
self.token_counter.add(
output_tokens,
{"persona": persona, "model": model, "direction": "output"}
)
# Calculate cost
cost = self._calculate_cost(model, input_tokens, output_tokens)
self.cost_counter.add(
cost,
{"persona": persona, "model": model}
)
def _calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
PRICING = {
"claude-3-opus": {"input": 0.015, "output": 0.075},
"claude-3-sonnet": {"input": 0.003, "output": 0.015},
"claude-3-haiku": {"input": 0.00025, "output": 0.00125},
}
rates = PRICING.get(model, PRICING["claude-3-sonnet"])
return (
input_tokens * rates["input"] / 1000 +
output_tokens * rates["output"] / 1000
)
2. Evaluation Harness
class EvaluationHarness:
"""
Continuous evaluation of agent quality.
Golden datasets + synthetic challenges.
"""
def __init__(self):
self.golden_tasks = self._load_golden_tasks()
self.evaluators = {
"correctness": CorrectnessEvaluator(),
"helpfulness": HelpfulnessEvaluator(),
"safety": SafetyEvaluator(),
"coherence": CoherenceEvaluator(),
}
async def run_evaluation(
self,
agent: Agent,
dataset: str = "golden"
) -> EvaluationReport:
tasks = self._get_tasks(dataset)
results = []
for task in tasks:
# Run agent
response = await agent.execute(task.input)
# Evaluate response
scores = {}
for name, evaluator in self.evaluators.items():
scores[name] = await evaluator.evaluate(
task=task,
response=response,
)
results.append(TaskResult(
task_id=task.id,
scores=scores,
response=response,
expected=task.expected_output,
))
return EvaluationReport(
agent=agent.name,
dataset=dataset,
results=results,
summary=self._summarize(results),
)
async def run_regression_check(
self,
agent: Agent,
baseline: EvaluationReport
) -> RegressionResult:
"""Check for quality regressions against baseline."""
current = await self.run_evaluation(agent, "golden")
regressions = []
for metric, baseline_score in baseline.summary.items():
current_score = current.summary[metric]
delta = current_score - baseline_score
if delta < -0.05: # 5% regression threshold
regressions.append(Regression(
metric=metric,
baseline=baseline_score,
current=current_score,
delta=delta,
))
return RegressionResult(
passed=len(regressions) == 0,
regressions=regressions,
current=current,
)
class CorrectnessEvaluator:
"""Evaluate factual correctness using LLM-as-judge."""
async def evaluate(self, task: Task, response: Response) -> float:
# Use a separate judge model
judgment = await self.judge_model.evaluate(
prompt=f"""
Task: {task.input}
Expected: {task.expected_output}
Actual: {response.content}
Rate the correctness from 0.0 to 1.0.
Consider: accuracy, completeness, absence of errors.
"""
)
return judgment.score
3. Cost Analytics Engine
class CostAnalyticsEngine:
"""
Analyze and optimize LLM costs.
"""
async def generate_cost_report(
self,
time_range: TimeRange
) -> CostReport:
# Query metrics
data = await self.metrics.query(
metric="agent.cost.dollars",
time_range=time_range,
group_by=["persona", "model", "workflow_type"],
)
return CostReport(
total_cost=sum(d.value for d in data),
by_persona=self._group_by_persona(data),
by_model=self._group_by_model(data),
by_workflow=self._group_by_workflow(data),
top_expensive_workflows=self._find_expensive(data),
optimization_recommendations=self._generate_recommendations(data),
)
def _generate_recommendations(self, data: List[Metric]) -> List[Recommendation]:
recommendations = []
# Find expensive simple queries
for persona, metrics in self._group_by_persona(data).items():
if metrics.avg_tokens < 500 and metrics.model == "opus":
recommendations.append(Recommendation(
type="model_downgrade",
persona=persona,
current_model="opus",
suggested_model="haiku",
estimated_savings=metrics.cost * 0.95,
reason="Simple queries using expensive model",
))
# Find high retry rates
retry_data = await self.metrics.query("agent.retries.total")
for persona, retry_rate in self._calculate_retry_rates(retry_data).items():
if retry_rate > 0.1:
recommendations.append(Recommendation(
type="retry_reduction",
persona=persona,
current_rate=retry_rate,
target_rate=0.05,
estimated_savings=self._estimate_retry_savings(persona),
reason=f"High retry rate ({retry_rate:.1%})",
))
return recommendations
class CostDashboard:
"""Real-time cost visibility."""
def get_dashboard_data(self) -> DashboardData:
return DashboardData(
current_month_cost=self._get_month_cost(),
cost_trend=self._get_cost_trend(),
cost_by_persona=self._get_cost_by_persona(),
cost_by_workflow=self._get_cost_by_workflow(),
projected_month_end=self._project_cost(),
budget_utilization=self._get_budget_utilization(),
anomalies=self._detect_cost_anomalies(),
)
4. Quality Dashboard
class QualityDashboard:
"""
Real-time quality visibility.
"""
def __init__(self):
self.eval_harness = EvaluationHarness()
self.metrics = MetricsClient()
async def get_dashboard_data(self) -> QualityDashboardData:
return QualityDashboardData(
# Real-time metrics
success_rate=await self._get_success_rate(),
error_rate_by_type=await self._get_error_breakdown(),
latency_percentiles=await self._get_latency_percentiles(),
# Quality scores
eval_scores=await self._get_latest_eval_scores(),
score_trends=await self._get_score_trends(),
# Comparisons
model_comparison=await self._compare_models(),
persona_comparison=await self._compare_personas(),
# Alerts
active_alerts=await self._get_active_alerts(),
recent_regressions=await self._get_recent_regressions(),
)
async def _compare_models(self) -> ModelComparison:
"""Compare quality across models."""
models = ["opus", "sonnet", "haiku"]
results = {}
for model in models:
metrics = await self.metrics.query(
metric="agent.success.rate",
filters={"model": model},
time_range="24h",
)
results[model] = {
"success_rate": metrics.avg,
"latency_p50": await self._get_latency(model, "p50"),
"latency_p99": await self._get_latency(model, "p99"),
"cost_per_success": await self._get_cost_per_success(model),
}
return ModelComparison(results)
5. Alerting System
class AgentAlertingSystem:
"""
Proactive alerting for agent issues.
"""
ALERT_RULES = [
AlertRule(
name="high_error_rate",
query="rate(agent.errors[5m]) > 0.05",
severity="critical",
summary="Agent error rate above 5%",
),
AlertRule(
name="high_latency",
query="histogram_quantile(0.99, agent.latency) > 30",
severity="warning",
summary="Agent p99 latency above 30 seconds",
),
AlertRule(
name="cost_spike",
query="increase(agent.cost.dollars[1h]) > 100",
severity="warning",
summary="Hourly cost spike above $100",
),
AlertRule(
name="quality_regression",
query="agent.eval.score < 0.8",
severity="critical",
summary="Agent quality score below threshold",
),
AlertRule(
name="token_waste",
query="rate(agent.tokens[1h]) / rate(agent.success[1h]) > 10000",
severity="info",
summary="High tokens per successful request",
),
]
async def evaluate_alerts(self) -> List[Alert]:
alerts = []
for rule in self.ALERT_RULES:
result = await self.prometheus.query(rule.query)
if result.value:
alerts.append(Alert(
rule=rule,
value=result.value,
timestamp=now(),
))
return alerts
async def send_alert(self, alert: Alert):
if alert.rule.severity == "critical":
await self.pagerduty.trigger(alert)
await self.slack.send(alert)
await self.metrics.record_alert(alert)
Integration Points
Files to Modify
Grafana Dashboard Configuration
{
"dashboard": {
"title": "AgentOps Overview",
"panels": [
{
"title": "Success Rate by Persona",
"type": "timeseries",
"targets": [{
"expr": "rate(agent_invocations_total{status='success'}[5m]) / rate(agent_invocations_total[5m])"
}]
},
{
"title": "Cost by Model",
"type": "piechart",
"targets": [{
"expr": "sum(agent_cost_dollars_total) by (model)"
}]
},
{
"title": "Latency P99",
"type": "gauge",
"targets": [{
"expr": "histogram_quantile(0.99, agent_latency_seconds_bucket)"
}]
},
{
"title": "Token Usage",
"type": "timeseries",
"targets": [{
"expr": "rate(agent_tokens_total[5m])"
}]
}
]
}
}
Implementation Plan
Week 1: Telemetry Foundation
- [ ] Add OpenTelemetry to adapters
- [ ] Implement token counting
- [ ] Add cost calculation
- [ ] Set up trace collection
Week 2: Metrics Pipeline
- [ ] Configure Prometheus exporters
- [ ] Set up Grafana dashboards
- [ ] Create initial alert rules
- [ ] Implement log aggregation
Week 3: Evaluation Harness
- [ ] Build golden task dataset
- [ ] Implement evaluators
- [ ] Create regression testing
- [ ] Add to CI pipeline
Week 4: Cost Analytics
- [ ] Build cost attribution
- [ ] Create cost dashboard
- [ ] Implement recommendations engine
- [ ] Add budget tracking
Week 5: Quality Dashboard
- [ ] Build quality metrics
- [ ] Create comparison views
- [ ] Add trend analysis
- [ ] Implement anomaly detection
Week 6: Production Rollout
- [ ] Enable for all agents
- [ ] Tune alert thresholds
- [ ] Training for operations team
- [ ] Documentation
Expected Results
Visibility Improvements
Cost Optimization
Quality Improvements
Risks & Mitigations
Success Criteria
- #05 Persona Routing: Metrics inform routing decisions
- #06 Autonomous CI/CD: Evals gate deployments
- #03 Agent Orchestration: Emergence metrics
Status: Ready for Implementation
Key Insight: You can't improve what you can't measure
Owner: Platform Engineering Team
