Agent Observability Logger

Medium
Agents

Implement observability logging for production agents:

  1. log_trace(operation, inputs, outputs, latency_ms, tokens): Record execution
    • Auto-generate trace_id and timestamp
  2. log_error(operation, error, context): Record errors
  3. get_metrics(): Return aggregated stats
    • avg_latency, throughput, error_rate
  4. get_trace(trace_id): Retrieve specific trace
  5. export_logs(format): Export as JSON or structured text

Trace Structure: {'trace_id': ..., 'timestamp': ..., 'operation': ..., 'inputs': ..., 'outputs': ..., 'latency_ms': ..., 'tokens': ...}

Metrics: Compute averages and rates from accumulated traces.

Examples

Example 1:
Input: obs = AgentObservability('agent1'); obs.log_trace('search', 'query', 'results', 100, 50); obs.metrics['total_calls']
Output: 1
Explanation: One trace logged, counter incremented

Starter Code

class AgentObservability:
    """
    Comprehensive observability for agent execution.
    """
    
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.traces = []
        self.metrics = {
            'total_calls': 0,
            'total_tokens': 0,
            'total_latency': 0,
            'errors': 0
        }
    
    def log_trace(self, operation, inputs, outputs, latency_ms, tokens):
        """Log detailed execution trace"""
        # Your implementation here
        pass
    
    def log_error(self, operation, error, context):
        """Log error with context"""
        # Your implementation here
        pass
    
    def get_metrics(self):
        """Get aggregated metrics"""
        # Your implementation here
        pass
    
    def get_trace(self, trace_id):
        """Retrieve specific trace by ID"""
        # Your implementation here
        pass
    
    def export_logs(self, format='json'):
        """Export logs in specified format"""
        # Your implementation here
        pass
Lines: 1Characters: 0
Ready
The AI Interview - Master AI/ML Interviews