Implement observability logging for production agents:
log_trace(operation, inputs, outputs, latency_ms, tokens): Record execution- Auto-generate trace_id and timestamp
log_error(operation, error, context): Record errorsget_metrics(): Return aggregated stats- avg_latency, throughput, error_rate
get_trace(trace_id): Retrieve specific traceexport_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:
1Explanation: 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
passPython3
ReadyLines: 1Characters: 0
Ready