Observability Hooks

Observability hooks capture component-level latency and token usage without changing the data flow. They are side-effect-only, so you can register many safely, each fires at the end of a component’s turn with a payload of metrics (a dict).

Component Metrics

Register a metrics hook with the @pipeline.metrics.on("<component>") decorator. Each component delivers different payload keys.

Component	Mode	Key Payload Fields
`stt`	Cascade	`stt_latency`
`llm`	Cascade	`llm_ttft`, `llm_duration`, `prompt_tokens`, `completion_tokens`, `total_tokens`
`tts`	Cascade	`ttfb`, `tts_latency`
`eou`	Cascade	`eou_latency`, `eou_wait_ms`
`realtime`	Realtime / Hybrid	`realtime_ttfb`, `realtime_input_tokens`, `realtime_output_tokens`, `realtime_total_tokens`, `realtime_input_text_tokens`, `realtime_output_text_tokens`, `realtime_input_audio_tokens`, `realtime_output_audio_tokens`

Use cases

Use stt, llm, tts, and eou metrics with Cascade pipelines. Use realtime metrics when the pipeline runs in Realtime or Hybrid mode with a realtime model as the LLM.

STT
LLM
TTS
EOU
Realtime

    @pipeline.metrics.on("stt")
    def on_stt_metrics(metrics: dict):
        """Fired when an STT turn completes."""
        print(f"[METRICS] STT Latency: {metrics.get('stt_latency')}ms")

    @pipeline.metrics.on("llm")
    def on_llm_metrics(metrics: dict):
        """Fired when LLM generation completes."""
        print(
            f"[METRICS] LLM TTFT: {metrics.get('llm_ttft')}ms | "
            f"Total Duration: {metrics.get('llm_duration')}ms"
        )
        print(
            "[METRICS] LLM Tokens (P/C/T): "
            f"{metrics.get('prompt_tokens')}/"
            f"{metrics.get('completion_tokens')}/"
            f"{metrics.get('total_tokens')}"
        )

    @pipeline.metrics.on("tts")
    def on_tts_metrics(metrics: dict):
        """Fired when TTS finishes speaking."""
        print(
            f"[METRICS] TTS TTFB: {metrics.get('ttfb')}ms | "
            f"Total Latency: {metrics.get('tts_latency')}ms"
        )

    @pipeline.metrics.on("eou")
    def on_eou_metrics(metrics: dict):
        """Fired when the Turn Detector matches end-of-utterance."""
        print(
            f"[METRICS] EOU Latency: {metrics.get('eou_latency')}ms | "
            f"EOU Wait: {metrics.get('eou_wait_ms')}ms"
        )

    @pipeline.metrics.on("realtime")
    def on_realtime_metrics(metrics: dict):
        """Fired for realtime (speech-to-speech) models."""
        print(
            "[METRICS] Realtime "
            f"TTFB: {metrics.get('realtime_ttfb')}ms | "
            f"Tokens (in/out/total): "
            f"{metrics.get('realtime_input_tokens')}/"
            f"{metrics.get('realtime_output_tokens')}/"
            f"{metrics.get('realtime_total_tokens')} | "
            f"TextTokens (in/out): "
            f"{metrics.get('realtime_input_text_tokens')}/"
            f"{metrics.get('realtime_output_text_tokens')} | "
            f"AudioTokens (in/out): "
            f"{metrics.get('realtime_input_audio_tokens')}/"
            f"{metrics.get('realtime_output_audio_tokens')}"
        )

What’s Next

Observability and Analytics

View the captured metrics, traces, and logs on the Dashboard.

Run Your Agent

Monitor agent and user state changes.

References

Python

Examples

Emit metrics and traces from pipeline hooks.

SDK Reference

Pipeline Hooks

Pipeline Hooks in the Python API reference.

Metrics

Metrics in the Python API reference.

Get started

Modalities

Configure Pipeline

Tools and Capabilities

Turn Detection and Interruptions

Telephony Agents

Context Management

Observability and Monitoring