kind-iron-94532
08/02/2025, 3:37 PMtall-airline-16756
08/02/2025, 3:37 PMtall-airline-16756
08/02/2025, 3:38 PMfrom livekit.agents import metrics
2. Set Up Logging:
import logging
logger = logging.getLogger("agent-metrics")
logger.setLevel(<http://logging.INFO|logging.INFO>)
3. Subscribe to Metrics Events:
Use the metrics_collected event to capture metrics for each component:
@agent.on("metrics_collected")
def _on_metrics_collected(ev: metrics.MetricsCollectedEvent):
metrics.log_metrics(ev.metrics, logger=logger)
4. Log Specific Metrics:
For detailed logging of each component:
@agent.on("metrics_collected")
def _on_metrics_collected(ev: metrics.MetricsCollectedEvent):
if isinstance(ev.metrics, metrics.STTMetrics):
<http://logger.info|logger.info>(f"STT Metrics: Audio Duration: {ev.metrics.audio_duration}, Streamed: {ev.metrics.streamed}, Error: {ev.metrics.error}")
elif isinstance(ev.metrics, metrics.TTSMetrics):
<http://logger.info|logger.info>(f"TTS Metrics: Audio Duration: {ev.metrics.audio_duration}, TTFB: {ev.metrics.ttfb}")
elif isinstance(ev.metrics, metrics.LLMMetrics):
<http://logger.info|logger.info>(f"LLM Metrics: TTFT: {ev.metrics.ttft}, Tokens per Second: {ev.metrics.tokens_per_second}, Total Tokens: {ev.metrics.total_tokens}")
5. Aggregate Metrics:
Use the UsageCollector class to aggregate metrics across a session:
usage_collector = metrics.UsageCollector()
@agent.on("metrics_collected")
def _on_metrics_collected(ev: metrics.MetricsCollectedEvent):
usage_collector.collect(ev.metrics)
async def log_usage():
summary = usage_collector.get_summary()
<http://logger.info|logger.info>(f"Usage Summary: {summary}")
For more advanced use cases, such as sending metrics to a server, you can modify the event handler to send data to an HTTP endpoint. Let me know if you need further assistance!kind-iron-94532
08/02/2025, 3:39 PMtall-airline-16756
08/02/2025, 3:39 PM