delightful-airline-96880
02/25/2025, 2:06 PM@self._llm.on("metrics_collected")
def _on_llm_metrics(llm_metrics: metrics.LLMMetrics) -> None:
speech_data = SpeechDataContextVar.get(None)
if speech_data is None:
sequence_id = None
else:
sequence_id = speech_data.sequence_id
self.emit(
"metrics_collected",
metrics.PipelineLLMMetrics(
**llm_metrics.__dict__,
sequence_id=sequence_id,
),
)
gray-monkey-83525
02/25/2025, 2:06 PMon
event for LLM metrics in your CustomVoiceAssistant
class, you can override the existing event handler by using the @self._llm.on("metrics_collected")
decorator. This allows you to customize the behavior when LLM metrics are collected. Here's how you can implement it:
class CustomVoiceAssistant(agents.voice_assistant.VoiceAssistant):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._llm.on("metrics_collected", self._on_llm_metrics)
def _on_llm_metrics(self, llm_metrics: metrics.LLMMetrics) -> None:
speech_data = SpeechDataContextVar.get(None)
sequence_id = speech_data.sequence_id if speech_data else None
self.emit(
"metrics_collected",
metrics.PipelineLLMMetrics(
**llm_metrics.__dict__,
sequence_id=sequence_id,
),
)
This code snippet sets up a custom event handler for the metrics_collected
event, which processes LLM metrics and emits them with additional context. Would you like more details on this implementation?
Sources:
- LiveKit Voice Assistant Documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.delightful-airline-96880
02/25/2025, 2:11 PMgray-monkey-83525
02/25/2025, 2:11 PMdelightful-airline-96880
02/25/2025, 2:13 PMself._llm.on("metrics_collected", self._on_llm_metrics)
def _on_llm_metrics(self, llm_metrics: metrics.LLMMetrics) -> None:
speech_data = SpeechDataContextVar.get(None)
sequence_id = speech_data.sequence_id if speech_data else None
self.emit(
"metrics_collected",
metrics.PipelineLLMMetrics(
**llm_metrics.__dict__,
sequence_id=sequence_id,
),
)
But this handler from parent class is still executed. I need to exec on metrics collected only mine code
@self._llm.on("metrics_collected")
def _on_llm_metrics(llm_metrics: metrics.LLMMetrics) -> None:
speech_data = SpeechDataContextVar.get(None)
if speech_data is None:
sequence_id = None
else:
sequence_id = speech_data.sequence_id
self.emit(
"metrics_collected",
metrics.PipelineLLMMetrics(
**llm_metrics.__dict__,
sequence_id=sequence_id,
),
)
gray-monkey-83525
02/25/2025, 2:13 PM