Skip to main content
When developing real-time, multimodal AI applications, monitoring two key factors is crucial: performance (latency) and LLM/TTS usage. Performance impacts user experience, while usage can affect operational costs. Pipecat offers built-in metrics for both, which can be enabled with straightforward configuration options.

​
Enabling performance metrics

Set enable_metrics=True in PipelineParams when creating a task:
Example config
task = PipelineTask(
            pipeline,
            params=PipelineParams(
                ...
                enable_metrics=True,
                ...
            ),
        )
Once enabled, Pipecat logs the following metrics:
MetricDescription
TTFBTime To First Byte in seconds
Processing TimeTime taken by the service to respond in seconds
Text AggregationTime from the first LLM token to the first complete sentence (TTS services only)
Sample output
AnthropicLLMService#0 TTFB: 0.8378312587738037
CartesiaTTSService#0 text aggregation time: 0.2134
CartesiaTTSService#0 processing time: 0.0005071163177490234
CartesiaTTSService#0 TTFB: 0.17177796363830566
AnthropicLLMService#0 processing time: 2.4927797317504883

​
Limiting TTFB responses

If you only want the first TTFB measurement for each service, you can optionally pass report_only_initial_ttfb=True in PipelineParams:
Example config
task = PipelineTask(
            pipeline,
            params=PipelineParams(
                ...
                enable_metrics=True,
                report_only_initial_ttfb=True,
                ...
            ),
        )
Note: enable_metrics=True is required for this setting to have an effect.

​
Enabling LLM/TTS Usage Metrics

Set enable_usage_metrics=True in PipelineParams when creating a task:
Example config
task = PipelineTask(
            pipeline,
            params=PipelineParams(
                ...
                enable_usage_metrics=True,
                ...
            ),
        )
Pipecat will log the following as applicable:
MetricDescription
LLM UsageNumber of prompt and completion tokens used
TTS UsageNumber of characters processed
Sample output
CartesiaTTSService#0 usage characters: 65
AnthropicLLMService#0 prompt tokens: 104, completion tokens: 53
Note: Usage metrics are recorded per interaction and do not represent running totals.

​
Capturing Metrics Data

When metrics are enabled, Pipecat will emit a MetricsFrame for each interaction. The MetricsFrame contains:
  • TTFB data
  • Processing time
  • Text aggregation time (TTS)
  • Token usage for LLMs
  • Character usage for TTS
  • Smart Turn prediction data
You can access the metrics data by either adding a custom FrameProcessor to your pipeline or adding an observer to monitor MetricsFrames.

​
Example: Using a Custom FrameProcessor

Create a custom FrameProcessor to handle metrics data. Here’s an example Metrics Processor that can be added to your pipeline after the TTS processor. 
from pipecat.frames.frames import MetricsFrame
from pipecat.metrics.metrics import (
    LLMUsageMetricsData,
    ProcessingMetricsData,
    TextAggregationMetricsData,
    TTFBMetricsData,
    TTSUsageMetricsData,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

class MetricsLogger(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        if isinstance(frame, MetricsFrame):
            for d in frame.data:
                if isinstance(d, TTFBMetricsData):
                    print(f"!!! MetricsFrame: {frame}, ttfb: {d.value}")
                elif isinstance(d, ProcessingMetricsData):
                    print(f"!!! MetricsFrame: {frame}, processing: {d.value}")
                elif isinstance(d, LLMUsageMetricsData):
                    tokens = d.value
                    print(
                        f"!!! MetricsFrame: {frame}, tokens: {tokens.prompt_tokens}, characters: {tokens.completion_tokens}"
                    )
                elif isinstance(d, TextAggregationMetricsData):
                    print(f"!!! MetricsFrame: {frame}, text aggregation: {d.value}")
                elif isinstance(d, TTSUsageMetricsData):
                    print(f"!!! MetricsFrame: {frame}, characters: {d.value}")
        await self.push_frame(frame, direction)