Observability & Monitoring

Overview

You can't fix what you can't see. Observability is the foundation of production systems.

---

Structured Logging

Agent Execution Logs

class StructuredAgentLogging:
 """Log every agent action in structured format"""

 def log_execution_step(self, step_data):
 """Log structured step data"""

 log_entry = {
 'timestamp': time.time(),
 'request_id': step_data.request_id,
 'agent_id': step_data.agent_id,
 'step_number': step_data.step_number,
 'action': step_data.action,
 'input': step_data.input,
 'output': step_data.output,
 'latency_ms': step_data.latency_ms,
 'tokens_used': step_data.tokens,
 'cost': step_data.cost,
 'status': 'success'| 'error'| 'timeout',
 'metadata': {
 'model': step_data.model,
 'temperature': step_data.temperature
 }
 }

 # Log to structured storage
 self.structured_logger.info(log_entry)

 def query_logs(self, request_id):
 """Retrieve all logs for a request"""

 logs = self.structured_logger.filter(
 request_id=request_id
).order_by('timestamp')

 return logs

---

Trace Collection

End-to-End Request Tracing

class RequestTracing:
 """Track request through entire system"""

 def __init__(self):
 self.tracer = Tracer() # e.g., Jaeger

 def trace_request(self, request):
 """Create trace for full request lifecycle"""

 with self.tracer.start_span("handle_request") as span:
 span.set_tag("request_id", request.id)
 span.set_tag("user_id", request.user_id)

 # Validation
 with self.tracer.start_span("validate"):
 self.validate(request)

 # Routing
 with self.tracer.start_span("route"):
 agent = self.router.select(request)

 # Execution
 with self.tracer.start_span("execute"):
 result = agent.execute(request)

 # Monitoring
 with self.tracer.start_span("monitor"):
 self.monitor(result)

 return result

 def view_trace(self, trace_id):
 """Visualize full request trace"""

 # Show Gantt chart of operations
 trace = self.tracer.get_trace(trace_id)
 self.display_trace_chart(trace)

---

Metrics & Dashboards

Key Production Metrics

class ProductionMetrics:
 """Track production health"""

 def __init__(self):
 self.metrics = {
 # Throughput
 'requests_per_second': Counter(),
 'requests_total': Counter(),

 # Latency
 'latency_p50': Histogram(),
 'latency_p95': Histogram(),
 'latency_p99': Histogram(),

 # Quality
 'accuracy': Gauge(),
 'error_rate': Gauge(),

 # Cost
 'cost_per_request': Histogram(),
 'monthly_cost': Gauge(),

 # Safety
 'policy_violations': Counter(),
 'safety_issues': Counter()
 }

 def record_request(self, request, result):
 """Record metrics after request"""

 self.metrics['requests_per_second'].inc()
 self.metrics['latency_p95'].observe(result.latency)

 if result.success:
 self.metrics['accuracy'].set(
 self.calculate_accuracy(result)
)
 else:
 self.metrics['error_rate'].inc()

 self.metrics['cost_per_request'].observe(result.cost)

-

Alerting Strategies

Smart Alerting

class AlertingStrategy:
 """Only alert on real issues"""

 def __init__(self):
 self.alert_rules = [
 {
 'name': 'high_error_rate',
 'metric': 'error_rate',
 'threshold': 0.05,
 'duration': '5m',
 'severity': 'critical'
 },
 {
 'name': 'high_latency_p99',
 'metric': 'latency_p99',
 'threshold': 5000, # 5 seconds
 'duration': '10m',
 'severity': 'warning'
 },
 {
 'name': 'cost_spike',
 'metric': 'cost_per_request',
 'threshold': 1.0, # $1/request
 'duration': '1m',
 'severity': 'warning'
 },
 {
 'name': 'safety_violation',
 'metric': 'policy_violations',
 'threshold': 1,
 'duration': '0m', # Immediate
 'severity': 'critical'
 }
]

 def check_alerts(self):
 """Evaluate alert rules"""

 for rule in self.alert_rules:
 current_value = self.get_metric(
 rule['metric'],
 duration=rule['duration']
)

 if current_value > rule['threshold']:
 self.send_alert(
 title=rule['name'],
 severity=rule['severity'],
 metric=rule['metric'],
 value=current_value,
 threshold=rule['threshold']
)

---

Post-Incident Analysis

Incident Review Process

class PostIncidentAnalysis:
 """Learn from incidents"""

 def analyze_incident(self, incident_id):
 """Comprehensive incident analysis"""

 incident = self.incidents.get(incident_id)

 analysis = {
 'timeline': self.build_timeline(incident),
 'root_cause': self.identify_root_cause(incident),
 'contributing_factors': self.find_contributing_factors(incident),
 'impact': self.quantify_impact(incident),
 'recovery_actions': self.identify_recovery_actions(incident),
 'preventive_measures': self.suggest_preventions(incident)
 }

 # Generate report
 report = self.generate_report(analysis)

 # Share findings
 self.notify_team(report)
 self.add_to_knowledge_base(report)

 def build_timeline(self, incident):
 """Extract timeline from logs"""

 logs = self.structured_logger.filter(
 request_ids=incident.affected_requests
).order_by('timestamp')

 timeline = []
 for log in logs:
 timeline.append({
 'time': log.timestamp,
 'action': log.action,
 'status': log.status,
 'error': log.error
 })

 return timeline

-

3 Warnings

Warning 1: Too Much Logging

# WRONG
# Log everything at high verbosity
for step in agent.steps:
 logger.debug(f"Step: {step}")
 logger.debug(f"Input: {step.input}")
 logger.debug(f"Output: {step.output}")
 logger.debug(f"Tokens: {step.tokens}")
 # Logs grow to GBs/day

# RIGHT
# Log only important events
logger.info({
 'event': 'step_complete',
 'step_number': step.number,
 'status': step.status,
 'latency': step.latency
})
# Logs stay manageable

Warning 2: No Aggregation

# WRONG
# Store every log event individually
for event in events:
 store.save(event) # Millions of rows

# Hard to analyze later

# RIGHT
# Aggregate metrics
metrics.latency_histogram.observe(latency)
metrics.error_counter.inc()
metrics.cost_gauge.set(cost)

# Queryable metrics, not raw logs

Warning 3: Missing Context

# WRONG
logger.error(f"Request failed")
# No context! Which request? Why failed?

# RIGHT
logger.error(
 "Request failed",
 extra={
 'request_id': request.id,
 'user_id': request.user_id,
 'error': error,
 'stage': 'execution',
 'agent': agent.id
 }
)
# Full context for debugging

-

Last Updated: August 9, 2026