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`
`knowledge_base`	Cascade / Realtime	`kb_id`, `kb_retrieval_latency`,`kb_retrievals`

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. Use knowledge_base metrics with Managed RAG: it fires after each search, whether from Pipeline(knowledge_base=...) or from KnowledgeBase.retrieve() in a function tool.

@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('stt', async (metrics: Record<string, any>) => {
  // Fired when an STT turn completes.
  console.log(`[METRICS] STT Latency: ${metrics.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('llm', async (metrics: Record<string, any>) => {
  // Fired when LLM generation completes.
  console.log(
    `[METRICS] LLM TTFT: ${metrics.llm_ttft}ms | ` +
    `Total Duration: ${metrics.llm_duration}ms`
  );
  console.log(
    '[METRICS] LLM Tokens (P/C/T): ' +
    `${metrics.prompt_tokens}/` +
    `${metrics.completion_tokens}/` +
    `${metrics.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('tts', async (metrics: Record<string, any>) => {
  // Fired when TTS finishes speaking.
  console.log(
    `[METRICS] TTS TTFB: ${metrics.ttfb}ms | ` +
    `Total Latency: ${metrics.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('eou', async (metrics: Record<string, any>) => {
  // Fired when the Turn Detector matches end-of-utterance.
  console.log(
    `[METRICS] EOU Latency: ${metrics.eou_latency}ms | ` +
    `EOU Wait: ${metrics.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')}"
    )

pipeline.metrics.on('realtime', async (metrics: Record<string, any>) => {
  // Fired for realtime (speech-to-speech) models.
  console.log(
    '[METRICS] Realtime ' +
    `TTFB: ${metrics.realtime_ttfb}ms | ` +
    `Tokens (in/out/total): ` +
    `${metrics.realtime_input_tokens}/` +
    `${metrics.realtime_output_tokens}/` +
    `${metrics.realtime_total_tokens} | ` +
    `TextTokens (in/out): ` +
    `${metrics.realtime_input_text_tokens}/` +
    `${metrics.realtime_output_text_tokens} | ` +
    `AudioTokens (in/out): ` +
    `${metrics.realtime_input_audio_tokens}/` +
    `${metrics.realtime_output_audio_tokens}`
  );
});

@pipeline.metrics.on("knowledge_base")
def on_kb_metrics(metrics: dict):
    """Fired when a knowledge base search completes."""
    print(
        f"[METRICS] KB {metrics.get('kb_id')} | "
        f"Latency: {metrics.get('kb_retrieval_latency')}ms | "
        f"Retrievals: {len(metrics.get('kb_retrievals') or [])}"
    )

The knowledge base hook is Python only. JavaScript SDK support is coming soon.

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
Node JS

Examples

Emit metrics and traces from pipeline hooks.

Overview

Build

Plugins

Deploy