Metrics & tracing

When building real-time AI applications two things matter to monitor: performance (latency), which shapes the user experience, and usage (LLM tokens, TTS characters), which drives cost. pipedog ports Pipecat's built-in metrics for both.

Enabling metricsCopy Anchor Link

Metrics are switched on through PipelineParams when creating the worker:

const worker = new PipelineWorker(pipeline, {
  params: new PipelineParams({
    enableMetrics: true, // TTFB, processing time, ...
    enableUsageMetrics: true, // LLM tokens, TTS characters, ...
    reportOnlyInitialTtfb: true, // only the first TTFB per service
    sendInitialEmptyMetrics: false,
  }),
  observers: [
    // Log only LLM and TTS usage.
    new MetricsLogObserver({
      includeMetrics: new Set([LLMUsageMetricsData, TTSUsageMetricsData]),
    }),
  ],
});

With enableMetrics, services report:

With enableUsageMetrics, services report LLM prompt/completion tokens and TTS characters. Usage metrics are per interaction, not running totals.

• reportOnlyInitialTtfb: true reports only the first TTFB measurement of each service.
• By default the worker sends an initial MetricsFrame with zero values for every service when the pipeline starts; sendInitialEmptyMetrics: false turns that off.

Capturing metrics dataCopy Anchor Link

Every measurement travels as a MetricsFrame (a system frame) whose data is a list of MetricsData objects from @pipedog/core/metrics/metrics:

• TTFBMetricsData, TTFAMetricsData, TTFATMetricsData
• ProcessingMetricsData, TextAggregationMetricsData
• LLMUsageMetricsData (its value is an LLMTokenUsage: promptTokens, completionTokens, totalTokens, cache and reasoning token counts)
• STTUsageMetricsData, TTSUsageMetricsData
• TurnMetricsData

All of them have processor (the name of the processor that produced the metric) and an optional model. Read them with an observer such as MetricsLogObserver (above), or with a custom processor placed after the services:

export class MetricsCollector extends FrameProcessor {
  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);

    if (frame instanceof MetricsFrame) {
      for (const data of frame.data) {
        if (data instanceof TTFBMetricsData) {
          console.log(`${data.processor} TTFB: ${data.value}s`);
        } else if (data instanceof ProcessingMetricsData) {
          console.log(`${data.processor} processing: ${data.value}s`);
        } else if (data instanceof TTFAMetricsData) {
          console.log(
            `${data.processor} TTFA: ${data.ttfa}s (leading silence ${data.leadingSilence}s)`,
          );
        } else if (data instanceof LLMUsageMetricsData) {
          const { promptTokens, completionTokens } = data.value;
          console.log(`${data.processor} prompt: ${promptTokens}, completion: ${completionTokens}`);
        } else if (data instanceof TTSUsageMetricsData) {
          console.log(`${data.processor} characters: ${data.value}`);
        }
      }
    }

    await this.pushFrame(frame, direction);
  }
}

Processors that generate metrics use the helpers on FrameProcessor: startTtfbMetrics() / stopTtfbMetrics(), startProcessingMetrics() / stopProcessingMetrics(), startLlmUsageMetrics(tokens), startTtsUsageMetrics(text) and so on. They only report when metrics are enabled, and canGenerateMetrics() tells the worker which processors to include in the initial empty metrics.

To send TTFB and processing time to Sentry as transactions, see Sentry metrics.

Tracing (OpenTelemetry)Copy Anchor Link

pipedog ports Pipecat's OpenTelemetry tracing: one trace per conversation, a span per turn, and spans for the service calls in each turn. Span names, attributes and their hierarchy are the same as upstream's, so dashboards and LLM-observability tools built for Pipecat traces (Langfuse, Honeycomb, Grafana Tempo, Jaeger, ...) work unchanged.

Like upstream's pipecat-ai[tracing] extra, OpenTelemetry is optional:

• @pipedog/core has @opentelemetry/api as an optional peer dependency. Without it, core runs exactly the same, isTracingAvailable() is false and every tracing helper is a no-op.
• @pipedog/tracing brings the OpenTelemetry SDK, an OTLP/HTTP exporter and setupTracing().

bun add @pipedog/tracing @opentelemetry/api

SetupCopy Anchor Link

Call setupTracing() once at startup, then turn tracing on per worker with enableTracing:

// Once, at startup (upstream: setup_tracing("my-voice-bot", exporter=...)).
setupTracing({
  serviceName: "my-voice-bot",
  exporter: new OTLPTraceExporter({
    url: "http://localhost:4318/v1/traces", // any OTLP/HTTP collector
  }),
  consoleExport: false, // true also prints every span (debugging)
});

// Then per conversation:
const worker = new PipelineWorker(pipeline, {
  enableTracing: true,
  conversationId: "customer-123", // optional; a UUID otherwise
  additionalSpanAttributes: { "app.tenant": "acme" }, // added to the conversation span
});

setupTracing({ serviceName, exporter, consoleExport }) matches upstream's setup_tracing(service_name, exporter, console_export). It sets the resource attributes service.name, service.instance.id ($HOSTNAME) and deployment.environment ($ENVIRONMENT, default development), adds a batch span processor for each exporter, and registers the global tracer provider (which, as in Python, can only be set once per process). It returns false if setup failed.

@pipedog/tracing re-exports OTLPTraceExporter (from @opentelemetry/exporter-trace-otlp-http, which sends OTLP/HTTP JSON with fetch or node:http), ConsoleSpanExporter, InMemorySpanExporter and the span processors. Any OpenTelemetry SpanExporter works. Pointing it at a hosted backend is a matter of URL and headers:

// Langfuse (OTLP endpoint, Basic auth with your public/secret key pair)
const langfuse = new OTLPTraceExporter({
  url: "https://cloud.langfuse.com/api/public/otel/v1/traces",
  headers: { Authorization: `Basic ${btoa("pk-lf-...:sk-lf-...")}` },
});

// Honeycomb
const honeycomb = new OTLPTraceExporter({
  url: "https://api.honeycomb.io/v1/traces",
  headers: { "x-honeycomb-team": "YOUR_API_KEY" },
});

Names: pipedog by default, pipecat for compatibilityCopy Anchor Link

Span names and attribute keys are upstream's (see below). Two names that say which library made the trace differ, so a pipedog bot can be told apart from a Pipecat one:

Always set your own serviceName in production: it should name your app, not the library.

If anything filters on Pipecat's tracer names (a dashboard, an alert, a query on otel.scope.name, or a Python and a TypeScript bot reporting to the same place), switch them back with setTracerName("pipecat"), or tracerName: "pipecat" in setupTracing. Do it at startup, before a pipeline is created, because the turn tracer is fetched when its observer is built. Span names (conversation, turn, stt, llm, tts) and attribute keys (gen_ai.*, metrics.ttfb, ...) don't depend on this: they are the same either way.

// Tracers are called "pipedog" and "pipedog.turn" by default. For dashboards, alerts or queries
// that filter on Pipecat's names ("pipecat" and "pipecat.turn"), switch them back at startup,
// before any pipeline is created:
setTracerName("pipecat");

// or in setupTracing, which does the same:
setupTracing({
  serviceName: "my-voice-bot",
  tracerName: "pipecat",
  exporter: new OTLPTraceExporter({ url: "http://localhost:4318/v1/traces" }),
});

Spans and attributesCopy Anchor Link

conversation                    conversation.id, conversation.type = "voice", + additionalSpanAttributes
└── turn                        turn.number, turn.type, conversation.id, turn.duration_seconds,
    │                           turn.was_interrupted, turn.user_bot_latency_seconds
    ├── stt                     gen_ai.provider.name, gen_ai.request.model, gen_ai.operation.name,
    │                           transcript, is_final, language, vad_enabled, metrics.ttfb,
    │                           metrics.audio_seconds, settings.*
    ├── llm                     gen_ai.provider.name, gen_ai.request.model, gen_ai.operation.name = "chat",
    │                           stream, input, output, tools, tool_count, tool_choice,
    │                           gen_ai.system_instructions, gen_ai.usage.input_tokens,
    │                           gen_ai.usage.output_tokens, metrics.ttfb, gen_ai.request.*, param.*
    └── tts                     gen_ai.provider.name, gen_ai.request.model, gen_ai.output.type = "speech",
                                voice_id, text, metrics.character_count, metrics.ttfb, settings.*

• The conversation span starts with the pipeline (StartFrame) and ends when the worker finishes; its id is conversationId, or a random UUID.
• A turn span covers one user turn and the bot's answer, as tracked by TurnTrackingObserver.
• Service spans are children of the current turn (or of the conversation, between turns). They come from the tracedStt / tracedLlm / tracedTts decorators (and tracedGeminiLive, tracedOpenaiRealtime for speech-to-speech services, whose spans are named after the operation: llm_setup, llm_request, llm_response, llm_tool_call, llm_tool_result) in @pipedog/core/utils/tracing/service_decorators. Provider packages apply them as upstream's services do, so a service shows up in traces once its package is ported.

The metrics described above are recorded on these spans (metrics.ttfb, token usage, character counts), not as separate OpenTelemetry metrics: like Pipecat, pipedog has no OpenTelemetry metrics exporter. Read MetricsFrames with an observer if you need them as time series.

Keeping content out of tracesCopy Anchor Link

Like Pipecat's, the spans carry what people said and what the model answered: transcript, text, input, output, gen_ai.system_instructions, tool definitions. With sensitive data (health, payments, anything personal) pass redactContent and the spans keep timings, ids and metadata only:

import { setupTracing } from "@pipedog/tracing";

setupTracing({ serviceName: "my-voice-bot", exporter, redactContent: true });

• String attributes are dropped unless their key is allowed. The default list (DEFAULT_REDACTION_ALLOWED_ATTRIBUTES) holds the non-content keys pipedog writes: the model, provider and operation (gen_ai.*), voice_id, language, settings.model / voice / language, conversation.id, conversation.type, turn.type, tool names. Numbers and booleans (metrics.*, token counts, turn.* durations) are kept on any key.
• Events keep their name and only exception.type; a recorded exception keeps its type but not its message or stack; a status keeps its code but not its message.
• redactContent: { extraAllowedAttributes: ["app.tenant"] } adds your own keys (for example from additionalSpanAttributes), allowedAttributes replaces the list, and isAllowed(key, value) decides on its own.

Redaction happens as each attribute is written, on the SDK's own span, so nothing reaches a span processor or exporter. If you set up OpenTelemetry yourself, call registerWithCore({ redactContent: true }) instead; redactTracer, redactTracerProvider and redactSpan apply it to your own spans. Not in upstream.

Cloudflare WorkersCopy Anchor Link

@pipedog/core and @pipedog/tracing run on Workers (see the compatibility table). The OTLP exporter posts spans with fetch, and spans are batched, so flush them before the request ends:

export default {
  async fetch(_request: Request, _env: unknown, ctx: { waitUntil(p: Promise<unknown>): void }) {
    const worker = new PipelineWorker(new Pipeline([]), { enableTracing: true });
    const runner = new WorkerRunner({ handleSigint: false });
    await runner.addWorkers(worker);
    await runner.run();
    // Spans are batched: export what is left before the request ends.
    ctx.waitUntil(forceFlushTracing());
    return new Response("ok");
  },
};

forceFlushTracing() and shutdownTracing() are pipedog additions (a Python process flushes on exit).

Without @pipedog/tracingCopy Anchor Link

If you already set up the OpenTelemetry SDK yourself (for example with @opentelemetry/sdk-node), you only need @opentelemetry/api installed: core imports it on its own, and a worker with enableTracing: true traces whenever it is found, wherever and whenever the worker was built:

// Without @pipedog/tracing: set up the OpenTelemetry SDK your way (it registers
// the global tracer provider), then enable tracing on the worker as usual. Core
// imports @opentelemetry/api on its own.
const traced = new PipelineWorker(pipeline, { enableTracing: true });

// Optional: check whether OpenTelemetry was found...
console.log(await loadOpenTelemetry(), isTracingAvailable());
// ...or hand in a specific copy of the API yourself.
registerOpenTelemetry({ api });