Observability & Monitoring: Seeing What Happens

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