From 998f9650095b99e907d47280b0c5449c5c22dd78 Mon Sep 17 00:00:00 2001 From: abhinav-t41 Date: Wed, 15 Jul 2026 01:24:47 +0530 Subject: [PATCH 1/4] Meter STT usage on audio streamed to the provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit STT providers bill on audio streamed to them (silence included), but we only measured VAD-detected turn audio, systematically underestimating real usage. Add SttUsageMeterMixin that counts bytes at the run_stt seam — the exact point audio leaves for the provider — on both instrumented STT services, and emit a usage.stt trace event at session teardown with streamed_seconds (billing truth), speech_seconds (VAD turn audio, kept for analytics), and provider_reported_seconds (Deepgram's own duration from ListenV1Metadata, for calibration). Also documents the broader cross-provider cost-comparison design in docs/cost-metering-design.md. Co-Authored-By: Claude Fable 5 --- docs/cost-metering-design.md | 309 ++++++++++++++++++ .../app/services/pipecat_streaming_runtime.py | 57 +++- server/app/services/stt_evaluation_service.py | 4 + server/tests/test_stt_usage_meter.py | 64 ++++ 4 files changed, 430 insertions(+), 4 deletions(-) create mode 100644 docs/cost-metering-design.md create mode 100644 server/tests/test_stt_usage_meter.py diff --git a/docs/cost-metering-design.md b/docs/cost-metering-design.md new file mode 100644 index 0000000..ec6cfa8 --- /dev/null +++ b/docs/cost-metering-design.md @@ -0,0 +1,309 @@ +# Cross-Provider Cost Comparison: Design & Metering Plan + +**Status:** Design proposal (not yet implemented) +**Date:** 2026-07-15 +**Scope:** STT + TTS cost metering and cross-provider cost comparison for voice sessions. + +--- + +## 1. Intent + +VoiceLab runs each voice session on one configured STT provider and one TTS +provider (currently Deepgram or ElevenLabs). At the end of a session, the user +should see a **cost comparison**: what this exact session cost on the +configured providers, and what it *would have cost* on each alternative +provider/model. + +## 2. Guiding principle + +> **Cost is not a thing you measure — usage is. Cost = usage × rate.** + +If usage is captured once, in provider-neutral billing units, then comparing +N providers is pure arithmetic against a rate table. The billing units are: + +| Service | Billing unit | Comparable across providers? | +|---------|-------------|------------------------------| +| STT | Seconds of **audio streamed** to the provider (silence included) | Yes — same audio would be streamed to any provider | +| TTS | **Characters** of input text | Yes — exactly, the text is identical regardless of provider | +| LLM | Input/output **tokens** | Approximately (tokenizers differ) — out of scope for now | + +A direct corollary: **do not run shadow sessions on other providers to learn +cost.** Real parallel calls are the right tool for comparing *quality and +latency* (which the existing STT-evaluation feature does), but for cost they +are pure waste — you would pay every provider to learn a number that is +computable from published rates. Extrapolation from measured usage is the +correct method for cost. + +--- + +## 3. Audit of the current implementation + +Findings from a full sweep of the codebase (2026-07-15): + +### 3.1 Two contradictory pricing tables + +- `server/app/services/pricing.py` — LLM + STT + TTS rates. **Dead code**: + imported nowhere in `app/` (only its own test). Consumes `usage.llm` / + `usage.stt` / `usage.tts` trace events that are no longer emitted. +- `server/app/services/stt_evaluation_pricing.py` — STT-only, the live one. + Deepgram, ElevenLabs, Sarvam rates per audio-minute. +- The two tables **disagree** (e.g. Deepgram STT $0.0043/min vs $0.0048/min) + and use different model naming. + +### 3.2 Only STT usage is measured, and it measures the wrong thing + +- `AudioBufferProcessor(enable_turn_audio=True)` + (`pipecat_streaming_runtime.py:524`) captures per-user-turn audio. +- `compute_all_model_costs(duration_sec)` + (`stt_evaluation_pricing.py:71`) multiplies that single turn duration by + every provider's rate. +- **Problem:** streaming STT providers bill on audio *streamed* to them — + silence included — not on VAD-detected speech turns. Turn-only counting + systematically underestimates real cost. + +### 3.3 TTS and LLM are not metered at all + +- `PipelineParams(enable_usage_metrics=False)` + (`pipecat_streaming_runtime.py:565`) disables Pipecat's usage metrics. +- No TTS character counting exists in the active path; no LLM token counting. +- The TTS half of the cost-comparison intent is currently 100% missing. + +### 3.4 Results are stored on disk, ranked on the frontend + +- Costs live in `recordings//metrics.jsonl` + (`stt_evaluation_store.py`), not Postgres — not transactional, not + queryable; losing the recordings dir loses cost history even though the + runs still exist in the DB. +- `GET /api/audio-evaluations/agent/{agent_id}` re-reads JSONL files per + request. +- The highest/lowest ranking is recomputed client-side on every render + (`client/src/components/audio/AudioView.tsx:159-166`). + +### 3.5 Other gaps + +- **Pricing alias gap:** a live agent using Deepgram `nova-3` misses the + pricing keys `nova-3-monolingual` / `nova-3-multilingual` and silently + computes **$0** (`stt_evaluation_pricing.py:64-66`). +- **Sarvam mismatch:** Sarvam is priced and shadow-evaluated but is not a + selectable live provider (`SUPPORTED_STT_PROVIDERS` excludes it, + `schemas/agent.py:10`). +- Shadow STT calls to ElevenLabs + Sarvam run on every turn regardless of + the agent's provider — justified for quality comparison only, and already + gated by `enable_stt_evaluation`; keep it that way. + +--- + +## 4. Target architecture (four layers) + +### Layer 1 — Meter usage in neutral units, per session, into `trace_events` + +Emit immutable usage facts alongside existing transcript events. **No prices +anywhere in these events** — prices change, facts don't. + +- `usage.stt` → `{ provider, model, streamed_seconds, speech_seconds }` +- `usage.tts` → `{ processor, model, characters }` (one per synthesized + utterance; sum per run at aggregation time) +- `usage.llm` → `{ model, input_tokens, output_tokens }` (future) + +### Layer 2 — One pricing catalog, server-side + +A single source of truth (delete the dead `pricing.py`). Entries keyed by +`(service, provider, model)` with: + +- `unit` — per audio-minute / per 1M chars / per 1M tokens +- `rate` +- `effective_from` — so rate history is explicit +- **billing adjusters** — per-provider minimum billable duration, + second-rounding, character-block rounding. These make the comparison + honest; naive `duration × rate` is only an approximation. + +Start as a Python module; keep the door open to a `pricing_rates` DB table so +rates can change without deploys. Fix the `nova-3` alias gap. + +### Layer 3 — Compute + snapshot the comparison server-side, at session end + +When the run closes: aggregate the run's usage events → apply the catalog → +produce one cost object: + +- actual cost for the configured providers, +- hypothetical cost per alternative provider/model, +- **the rate used embedded in each line** (snapshot — when a provider changes + prices next quarter, old runs must still show what was true at run time). + +Store it in `runs.summary` (existing, unused JSONB column) or a small +`run_costs` table — **not** JSONL files on disk. The server also computes +ranking and deltas; the frontend only renders. + +### Layer 4 — Presentation + +End-of-session summary in the test-call panel and the Runs view: a table of +provider/model → estimated cost, the actually-used provider highlighted, +delta vs. actual ("ElevenLabs would have been +$0.0021, +38%"). Label +alternatives explicitly as *estimates*. + +--- + +## 5. STT metering: what we can do, do, and should do + +**The question:** how many seconds does the provider bill for a session? +Providers bill on **audio sent to them**, silence included. + +### Three possible measuring points + +| Option | How | Verdict | +|--------|-----|---------| +| 1. Speech turns | Capture audio only while VAD says the user is speaking | What we do today. Right for analytics, **wrong for billing** — most of a session is silence/agent speech that still streams to the provider | +| 2. Connection wall-time | Websocket open duration × sample rate | Crude approximation; ignores mutes, reconnect gaps | +| 3. **Bytes actually sent** | Count bytes crossing the wire to the provider; seconds = bytes ÷ (sample_rate × 2) for 16-bit mono PCM | **Correct** — this is exactly the billed quantity | + +### Does Pipecat provide a tool for this? + +**No.** Verified against the installed Pipecat 1.3.0: there is no +`STTUsageMetricsData` and no STT usage metering of any kind (STT services +emit only TTFB/processing metrics). We must count it ourselves. + +### The one right seam + +Pipecat's `STTService.process_audio_frame` +(`pipecat/services/stt_service.py:349`) applies all "will this audio actually +be sent" guards — reconnect buffering, mute-drop, empty-frame skip — and only +then calls `run_stt(frame.audio)`, which pushes the bytes over the provider +websocket. So **`run_stt` receives exactly the bytes the provider bills +for.** + +We already subclass both STT services in +`pipecat_streaming_runtime.py` (`InstrumentedDeepgramSTTService:143` and the +ElevenLabs equivalent), so: + +```python +class InstrumentedDeepgramSTTService(ProviderRequestTraceMixin, DeepgramSTTService): + def __init__(self, ...): + ... + self._streamed_bytes = 0 + + async def run_stt(self, audio: bytes): + self._streamed_bytes += len(audio) + async for frame in super().run_stt(audio): + yield frame + + @property + def streamed_seconds(self) -> float: + # 16-bit mono PCM: 2 bytes per sample + return self._streamed_bytes / (self.sample_rate * 2) +``` + +(`self.sample_rate` is set once the `StartFrame` arrives; input is mono per +the websocket serializer at `pipecat_streaming_runtime.py:71`.) + +At session teardown (after `runner.run(task)` returns), emit one event: + +```python +await record_trace("usage.stt", { + "provider": ..., + "model": ..., + "streamed_seconds": stt.streamed_seconds, +}) +``` + +### Keep both numbers + +Do **not** delete the `AudioBufferProcessor` turn capture. Keep: + +- `streamed_seconds` — the billing truth (feeds cost), +- `speech_seconds` — the analytics number (feeds quality/latency work). + +They answer different questions. + +### Free calibration signal + +`InstrumentedDeepgramSTTService._on_message` already intercepts Deepgram's +`ListenV1Metadata` for `request_id` +(`pipecat_streaming_runtime.py:158-160`). That metadata also carries a +`duration` field — the seconds **Deepgram itself** says it processed. Log it +next to our byte-counted `streamed_seconds`: if they agree within rounding, +the meter is provably accurate, which is what makes extrapolated costs for +*other* providers trustworthy. No extra HTTP calls needed. (A fuller +usage-API reconciliation was attempted in commit `07f2dd9` and reverted; +revisit once metering is solid.) + +--- + +## 6. TTS metering: Pipecat already does this — it's switched off + +Pipecat has a built-in TTS usage metric. Every TTS service calls +`start_tts_usage_metrics(text)` after synthesizing, emitting a `MetricsFrame` +carrying `TTSUsageMetricsData(value=)`. Verified in the +installed package for both providers we use: + +- Deepgram: `pipecat/services/deepgram/tts.py:497` +- ElevenLabs: `pipecat/services/elevenlabs/tts.py:1026` + +It is gated behind the flag we currently disable: + +```python +# pipecat_streaming_runtime.py:565 +enable_usage_metrics=False, # ← flip to True +``` + +Our `MetricsSink` (`server/app/services/pipeline_metrics.py`) already sits at +the pipeline tail and iterates `MetricsFrame` items for TTFB, so TTS metering +is a small addition: + +```python +from pipecat.metrics.metrics import TTFBMetricsData, TTSUsageMetricsData + +# in MetricsSink._handle_metric: +elif isinstance(item, TTSUsageMetricsData): + await self._record_trace( + "usage.tts", + {"processor": item.processor, "model": item.model, "characters": int(item.value)}, + ) +``` + +Since TTS providers bill on input characters and the text is identical +regardless of provider, the cross-provider TTS comparison is essentially +exact (modulo per-provider rounding/minimum rules from the pricing catalog). + +**Caveat:** `enable_usage_metrics=True` is pipeline-wide. If the pipecat-adk +LLM bridge emits `LLMUsageMetricsData` (token counts), those frames will also +reach `MetricsSink` — handle or ignore them explicitly rather than letting +them fall through silently. If the bridge doesn't emit them, LLM tokens can +later be read from ADK/Gemini response `usage_metadata`. + +--- + +## 7. Implementation checklist (in order) + +1. **Pricing catalog** — consolidate to one table with units, effective + dates, and billing adjusters; delete dead `pricing.py`; fix the `nova-3` + alias gap (currently silently $0). +2. **STT metering** — ✅ implemented (2026-07-15). `SttUsageMeterMixin` counts + bytes in `run_stt` on both instrumented STT subclasses; a `usage.stt` + trace event (streamed + speech + provider-reported seconds) is emitted at + session teardown; Deepgram metadata `duration` is accumulated for + calibration. Tests in `server/tests/test_stt_usage_meter.py`. +3. **TTS metering** — flip `enable_usage_metrics=True`; add + `TTSUsageMetricsData` branch to `MetricsSink` emitting `usage.tts`; + explicitly handle/ignore `LLMUsageMetricsData`. +4. **Aggregation** — at run close, aggregate `usage.*` events → compute + actual + hypothetical costs with rate snapshots → store in + `runs.summary` (or `run_costs` table). Stop treating `metrics.jsonl` as + the source of truth for cost. +5. **Presentation** — server-computed ranking/deltas; render in the + test-call end screen and Runs view; label alternatives as estimates. + +## 8. Key file reference + +| Concern | Path | +|---------|------| +| Live pricing (STT-only) | `server/app/services/stt_evaluation_pricing.py` | +| Dead pricing (to delete) | `server/app/services/pricing.py` | +| Pipeline + instrumented services | `server/app/services/pipecat_streaming_runtime.py` | +| Metrics sink (pipeline tail) | `server/app/services/pipeline_metrics.py` | +| STT shadow evaluation | `server/app/services/stt_evaluation_service.py` | +| JSONL metrics store (to retire for cost) | `server/app/services/stt_evaluation_store.py` | +| Evaluation API | `server/app/api/routes/audio_evaluations.py` | +| Agent provider config schema | `server/app/schemas/agent.py` | +| Cost comparison UI | `client/src/components/audio/AudioView.tsx` | +| Pipecat STT base (the `run_stt` seam) | `.venv/.../pipecat/services/stt_service.py:349` | +| Pipecat usage metric types | `.venv/.../pipecat/metrics/metrics.py` | diff --git a/server/app/services/pipecat_streaming_runtime.py b/server/app/services/pipecat_streaming_runtime.py index 5812685..c6ff552 100644 --- a/server/app/services/pipecat_streaming_runtime.py +++ b/server/app/services/pipecat_streaming_runtime.py @@ -1,6 +1,6 @@ import json import logging -from collections.abc import Awaitable, Callable +from collections.abc import AsyncGenerator, Awaitable, Callable from typing import Any from uuid import uuid4 @@ -140,7 +140,37 @@ async def _record_provider_request_from_mapping( return -class InstrumentedDeepgramSTTService(ProviderRequestTraceMixin, DeepgramSTTService): +class SttUsageMeterMixin: + """Meter raw audio bytes actually sent to the STT provider. + + `STTService.process_audio_frame` applies the mute/reconnect/empty-frame + guards and only then calls `run_stt`, so every byte counted here is audio + the provider bills for (silence included). VAD turn audio undercounts. + """ + + _streamed_audio_bytes: int = 0 + _provider_reported_audio_seconds: float = 0.0 + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]: + self._streamed_audio_bytes += len(audio) + async for frame in super().run_stt(audio): + yield frame + + @property + def streamed_audio_seconds(self) -> float: + if not self.sample_rate: + return 0.0 + # 16-bit mono PCM: 2 bytes per sample. + return self._streamed_audio_bytes / (self.sample_rate * 2) + + @property + def provider_reported_audio_seconds(self) -> float | None: + return self._provider_reported_audio_seconds or None + + +class InstrumentedDeepgramSTTService( + SttUsageMeterMixin, ProviderRequestTraceMixin, DeepgramSTTService +): def __init__( self, *, record_trace: TraceRecorder, provider_model: str, run_tag: str, **kwargs: Any ) -> None: @@ -158,6 +188,11 @@ def __init__( async def _on_message(self, message: Any) -> None: if isinstance(message, ListenV1Metadata): await self._record_provider_request(message.request_id) + # Deepgram reports processed audio duration per connection on + # close; accumulate across reconnects for meter calibration. + duration = getattr(message, "duration", None) + if duration: + self._provider_reported_audio_seconds += float(duration) elif isinstance(message, ListenV1Results): metadata = getattr(message, "metadata", None) await self._record_provider_request(getattr(metadata, "request_id", None)) @@ -193,7 +228,9 @@ async def _connect_websocket(self): await self._record_provider_request_from_headers(response_headers, "dg-request-id") -class InstrumentedElevenLabsSTTService(ProviderRequestTraceMixin, ElevenLabsRealtimeSTTService): +class InstrumentedElevenLabsSTTService( + SttUsageMeterMixin, ProviderRequestTraceMixin, ElevenLabsRealtimeSTTService +): def __init__( self, *, record_trace: TraceRecorder, provider_model: str, run_tag: str, **kwargs: Any ) -> None: @@ -609,5 +646,17 @@ async def on_session_timeout(_transport, _websocket): try: await runner.run(task) finally: - await stt_evaluation.finalize() + try: + await record_trace( + "usage.stt", + { + "provider": config.stt_provider, + "model": config.stt_model, + "streamed_seconds": round(stt.streamed_audio_seconds, 3), + "speech_seconds": stt_evaluation.session_duration_sec, + "provider_reported_seconds": stt.provider_reported_audio_seconds, + }, + ) + finally: + await stt_evaluation.finalize() logger.info("pipecat streaming test-call ended run_id=%s", run_id) diff --git a/server/app/services/stt_evaluation_service.py b/server/app/services/stt_evaluation_service.py index c603417..ce0ea2c 100644 --- a/server/app/services/stt_evaluation_service.py +++ b/server/app/services/stt_evaluation_service.py @@ -64,6 +64,10 @@ def __init__( self._background_tasks: set[asyncio.Task[None]] = set() self._provider_results_by_name: dict[str, list[ProviderEvaluationResult]] = {} + @property + def session_duration_sec(self) -> float: + return round(self._session_duration_sec, 3) + async def handle_user_turn_audio( self, audio: bytes, sample_rate: int, num_channels: int ) -> None: diff --git a/server/tests/test_stt_usage_meter.py b/server/tests/test_stt_usage_meter.py new file mode 100644 index 0000000..62b2eb5 --- /dev/null +++ b/server/tests/test_stt_usage_meter.py @@ -0,0 +1,64 @@ +from collections.abc import AsyncGenerator + +import pytest +from pipecat.frames.frames import Frame + +from app.services.pipecat_streaming_runtime import SttUsageMeterMixin + + +class _FakeSTTService: + def __init__(self, sample_rate: int) -> None: + self._sample_rate = sample_rate + self.received: list[bytes] = [] + + @property + def sample_rate(self) -> int: + return self._sample_rate + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]: + self.received.append(audio) + yield None + + +class _MeteredService(SttUsageMeterMixin, _FakeSTTService): + pass + + +async def _drain(gen: AsyncGenerator) -> None: + async for _ in gen: + pass + + +@pytest.mark.asyncio +async def test_streamed_seconds_counts_all_forwarded_audio() -> None: + service = _MeteredService(sample_rate=16000) + # 16-bit mono at 16 kHz: 32000 bytes per second. + await _drain(service.run_stt(b"\x00" * 32000)) + await _drain(service.run_stt(b"\x00" * 16000)) + assert service.streamed_audio_seconds == pytest.approx(1.5) + assert service.received == [b"\x00" * 32000, b"\x00" * 16000] + + +@pytest.mark.asyncio +async def test_streamed_seconds_zero_without_audio_or_sample_rate() -> None: + service = _MeteredService(sample_rate=16000) + assert service.streamed_audio_seconds == 0.0 + + unstarted = _MeteredService(sample_rate=0) + await _drain(unstarted.run_stt(b"\x00" * 3200)) + assert unstarted.streamed_audio_seconds == 0.0 + + +@pytest.mark.asyncio +async def test_meter_state_is_per_instance() -> None: + first = _MeteredService(sample_rate=16000) + second = _MeteredService(sample_rate=16000) + await _drain(first.run_stt(b"\x00" * 32000)) + assert second.streamed_audio_seconds == 0.0 + + +def test_provider_reported_seconds_defaults_to_none() -> None: + service = _MeteredService(sample_rate=16000) + assert service.provider_reported_audio_seconds is None + service._provider_reported_audio_seconds += 2.5 + assert service.provider_reported_audio_seconds == pytest.approx(2.5) From 4435049f916eda910d21a9f7cfaab83f97cc470d Mon Sep 17 00:00:00 2001 From: Himanshu singh Date: Wed, 15 Jul 2026 13:50:19 +0530 Subject: [PATCH 2/4] Use streamed STT seconds for dashboard pricing --- client/src/lib/types.ts | 2 ++ server/app/api/routes/audio_evaluations.py | 34 +++++++++++++++++-- server/app/repositories/run_repository.py | 1 + .../app/services/pipecat_streaming_runtime.py | 25 ++++++++------ server/app/services/pricing.py | 20 ++++++++--- server/app/services/stt_evaluation_pricing.py | 17 +++++++--- server/tests/test_pricing.py | 22 ++++++++---- server/tests/test_stt_evaluation_pricing.py | 6 ++-- 8 files changed, 97 insertions(+), 30 deletions(-) diff --git a/client/src/lib/types.ts b/client/src/lib/types.ts index 311dd22..51fe7f1 100644 --- a/client/src/lib/types.ts +++ b/client/src/lib/types.ts @@ -73,6 +73,8 @@ export interface AudioEvaluationRecord { created_at: string; turn_count: number; session_stt_duration_sec: number; + streamed_seconds: number; + stt_cost_usd: number | null; session_model_costs_usd: Record>; provider_session_metrics: Record; file_paths: string[]; diff --git a/server/app/api/routes/audio_evaluations.py b/server/app/api/routes/audio_evaluations.py index ff53ed8..5858d6f 100644 --- a/server/app/api/routes/audio_evaluations.py +++ b/server/app/api/routes/audio_evaluations.py @@ -9,6 +9,7 @@ from app.core.config import get_settings from app.core.db import get_db_session from app.repositories.run_repository import RunRepository +from app.services.stt_evaluation_pricing import compute_all_model_costs from app.services.stt_evaluation_store import resolve_recordings_root router = APIRouter(prefix="/audio-evaluations", tags=["audio-evaluations"]) @@ -31,6 +32,8 @@ class AudioEvaluationRead(BaseModel): created_at: str turn_count: int session_stt_duration_sec: float + streamed_seconds: float + stt_cost_usd: float | None = None session_model_costs_usd: dict[str, dict[str, float]] provider_session_metrics: dict[str, AudioProviderMetricsRead] file_paths: list[str] @@ -51,6 +54,7 @@ async def list_audio_evaluations(agent_id: str, session: SessionDep) -> list[Aud payload = _load_metrics_summary(metrics_path) if payload is None: continue + usage = _extract_usage_stt(run.trace_events) records.append( AudioEvaluationRead( session_id=run.adk_session_id, @@ -58,8 +62,13 @@ async def list_audio_evaluations(agent_id: str, session: SessionDep) -> list[Aud adk_session_id=run.adk_session_id, created_at=run.created_at.isoformat(), turn_count=payload["turn_count"], - session_stt_duration_sec=payload["session_stt_duration_sec"], - session_model_costs_usd=payload["session_model_costs_usd"], + session_stt_duration_sec=usage["streamed_seconds"] + or payload["session_stt_duration_sec"], + streamed_seconds=usage["streamed_seconds"], + stt_cost_usd=usage["cost_usd"], + session_model_costs_usd=compute_all_model_costs( + usage["streamed_seconds"] or payload["session_stt_duration_sec"] + ), provider_session_metrics=payload["provider_session_metrics"], file_paths=payload["file_paths"], evaluate_mode=payload["evaluate_mode"], @@ -96,3 +105,24 @@ def _load_metrics_summary(metrics_path: Path) -> dict[str, Any] | None: "provider_session_metrics": session_summary.get("provider_session_metrics") or {}, "evaluate_mode": bool(session_summary.get("evaluate_mode", False)), } + + +def _extract_usage_stt(trace_events: list[Any]) -> dict[str, float | None]: + for event in reversed(trace_events): + if getattr(event, "event_type", None) != "usage.stt": + continue + payload = getattr(event, "payload", {}) or {} + return { + "streamed_seconds": float(payload.get("streamed_seconds") or 0.0), + "cost_usd": _to_float(payload.get("cost_usd")), + } + return {"streamed_seconds": 0.0, "cost_usd": None} + + +def _to_float(value: Any) -> float | None: + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None diff --git a/server/app/repositories/run_repository.py b/server/app/repositories/run_repository.py index 3c37a13..84740c2 100644 --- a/server/app/repositories/run_repository.py +++ b/server/app/repositories/run_repository.py @@ -39,6 +39,7 @@ async def list_by_agent(self, agent_id: str) -> list[RunRecord]: result = await self.session.execute( select(RunRecord) .where(RunRecord.agent_id == agent_id) + .options(selectinload(RunRecord.trace_events)) .order_by(RunRecord.created_at.desc()) ) return list(result.scalars()) diff --git a/server/app/services/pipecat_streaming_runtime.py b/server/app/services/pipecat_streaming_runtime.py index c6ff552..36a7013 100644 --- a/server/app/services/pipecat_streaming_runtime.py +++ b/server/app/services/pipecat_streaming_runtime.py @@ -4,6 +4,7 @@ from typing import Any from uuid import uuid4 +from deepgram.listen.v1.types import ListenV1Metadata, ListenV1Results from fastapi import WebSocket from pipecat.frames.frames import ( BotStartedSpeakingFrame, @@ -20,14 +21,14 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer -from pipecat.services.settings import LLMSettings from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.tts import DeepgramTTSService from pipecat.services.elevenlabs.stt import CommitStrategy, ElevenLabsRealtimeSTTService from pipecat.services.elevenlabs.tts import ElevenLabsTTSService +from pipecat.services.settings import LLMSettings from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams, FastAPIWebsocketTransport from pipecat_adk import AdkLLMService, SessionParams, VqlTTSMixin from pipecat_adk.frames import ( @@ -35,15 +36,15 @@ VqlLLMFullResponseStartFrame, VqlLLMTextFrame, ) -from deepgram.listen.v1.types import ListenV1Metadata, ListenV1Results from app.core.config import get_settings -from app.schemas.agent import AgentConfig, DEFAULT_TTS_MODEL_BY_PROVIDER +from app.schemas.agent import DEFAULT_TTS_MODEL_BY_PROVIDER, AgentConfig from app.services.adk_session_service import create_adk_session_service, ensure_adk_session from app.services.pipecat_adk_runtime import PipecatAdkRuntime from app.services.pipeline_metrics import ( MetricsSink, ) +from app.services.pricing import compute_cost from app.services.stt_evaluation_service import SttEvaluationSession logger = logging.getLogger("uvicorn.error") @@ -647,15 +648,17 @@ async def on_session_timeout(_transport, _websocket): await runner.run(task) finally: try: + stt_usage_payload = { + "provider": config.stt_provider, + "model": config.stt_model, + "streamed_seconds": round(stt.streamed_audio_seconds, 3), + "speech_seconds": stt_evaluation.session_duration_sec, + "provider_reported_seconds": stt.provider_reported_audio_seconds, + } + stt_usage_payload["cost_usd"] = float(compute_cost("usage.stt", stt_usage_payload)) await record_trace( "usage.stt", - { - "provider": config.stt_provider, - "model": config.stt_model, - "streamed_seconds": round(stt.streamed_audio_seconds, 3), - "speech_seconds": stt_evaluation.session_duration_sec, - "provider_reported_seconds": stt.provider_reported_audio_seconds, - }, + stt_usage_payload, ) finally: await stt_evaluation.finalize() diff --git a/server/app/services/pricing.py b/server/app/services/pricing.py index 2e20ac7..ae6c702 100644 --- a/server/app/services/pricing.py +++ b/server/app/services/pricing.py @@ -23,9 +23,14 @@ # STT rates: USD per audio-minute submitted. STT_RATES: dict[str, Decimal] = { - "deepgram:nova-3": Decimal("0.0043"), + "deepgram:nova-3": Decimal("0.0077"), + "deepgram:nova-3-streaming": Decimal("0.0077"), + "deepgram:nova-3-multilingual": Decimal("0.0043"), "deepgram:nova-2": Decimal("0.0043"), - "deepgram:base": Decimal("0.0125"), + "elevenlabs:scribe-v2": Decimal("0.00367"), + "elevenlabs:scribe-v2-realtime": Decimal("0.0065"), + "sarvam:saarika": Decimal("0.005263"), + "sarvam:saarika-diarization": Decimal("0.007895"), } # TTS rates: USD per 1M characters synthesized. @@ -55,7 +60,7 @@ def _stt_cost(payload: dict[str, Any]) -> Decimal: rate = STT_RATES.get(key) if rate is None: return Decimal("0") - seconds = Decimal(str(payload.get("audio_seconds") or 0)) + seconds = _stt_seconds(payload) return (seconds / Decimal("60")) * rate @@ -107,7 +112,7 @@ def session_totals(events: list[Any]) -> dict[str, Any]: llm_latencies_ms.append(float(inline_latency)) elif event.event_type == "usage.stt": payload = event.payload or {} - local_stt_seconds += Decimal(str(payload.get("audio_seconds") or 0)) + local_stt_seconds += _stt_seconds(payload) local_stt_cost += _stt_cost(payload) elif event.event_type == "usage.tts": payload = event.payload or {} @@ -141,6 +146,7 @@ def session_totals(events: list[Any]) -> dict[str, Any]: }, "stt": { "audio_seconds": _q(stt_seconds, "0.001"), + "streamed_seconds": _q(stt_seconds, "0.001"), "cost_usd": _q(stt_cost), "source": "runtime", }, @@ -160,5 +166,11 @@ def _avg(values: list[float]) -> float: return round(sum(values) / len(values), 1) +def _stt_seconds(payload: dict[str, Any]) -> Decimal: + return Decimal( + str(payload.get("streamed_seconds") or payload.get("audio_seconds") or 0) + ) + + def _q(value: Decimal, quant: str = "0.000001") -> float: return float(value.quantize(Decimal(quant))) diff --git a/server/app/services/stt_evaluation_pricing.py b/server/app/services/stt_evaluation_pricing.py index c10606d..a5eb126 100644 --- a/server/app/services/stt_evaluation_pricing.py +++ b/server/app/services/stt_evaluation_pricing.py @@ -4,27 +4,31 @@ RATE_PER_MINUTE_USD: dict[str, dict[str, Decimal]] = { "deepgram": { - "nova-3-monolingual": Decimal("0.0048"), - "nova-3-multilingual": Decimal("0.0058"), + "nova-3": Decimal("0.0043"), + "nova-3-streaming": Decimal("0.0077"), + "nova-3-multilingual": Decimal("0.0043"), }, "elevenlabs": { "scribe-v2": Decimal("0.00367"), "scribe-v2-realtime": Decimal("0.0065"), }, "sarvam": { - "saarika": Decimal("0.0060"), + "saarika": Decimal("0.005263"), + "saarika-diarization": Decimal("0.007895"), }, } PRODUCTION_MODEL_BY_PROVIDER = { - "deepgram": "nova-3-monolingual", + "deepgram": "nova-3", "elevenlabs": "scribe_v2", "sarvam": "saarika:v2", } PRICING_MODEL_ALIASES = { "deepgram": { - "nova-3-monolingual": "nova-3-monolingual", + "nova-3": "nova-3", + "nova-3-monolingual": "nova-3", + "nova-3-streaming": "nova-3-streaming", "nova-3-multilingual": "nova-3-multilingual", }, "elevenlabs": { @@ -39,6 +43,9 @@ "saarika:v2": "saarika", "saarika:v2.5": "saarika", "saarika:flash": "saarika", + "saarika:diarization": "saarika-diarization", + "saarika-v2-diarization": "saarika-diarization", + "saarika-diarization": "saarika-diarization", "saarika-v1": "saarika", "saarika-v2": "saarika", "saarika": "saarika", diff --git a/server/tests/test_pricing.py b/server/tests/test_pricing.py index 87c592b..34aaa87 100644 --- a/server/tests/test_pricing.py +++ b/server/tests/test_pricing.py @@ -22,12 +22,20 @@ def test_llm_cost_unknown_model_is_zero() -> None: def test_stt_cost_deepgram_nova_3() -> None: - # $0.0043 per audio minute. + # $0.0077 per audio minute for streaming. + cost = compute_cost( + "usage.stt", + {"streamed_seconds": 60, "provider": "deepgram", "model": "nova-3"}, + ) + assert float(cost) == 0.0077 + + +def test_stt_cost_falls_back_to_audio_seconds_for_legacy_payloads() -> None: cost = compute_cost( "usage.stt", {"audio_seconds": 60, "provider": "deepgram", "model": "nova-3"}, ) - assert float(cost) == 0.0043 + assert float(cost) == 0.0077 def test_tts_cost_deepgram_aura_2() -> None: @@ -58,7 +66,7 @@ def test_session_totals_sums_across_events() -> None: ), SimpleNamespace( event_type="usage.stt", - payload={"audio_seconds": 30, "provider": "deepgram", "model": "nova-3"}, + payload={"streamed_seconds": 30, "provider": "deepgram", "model": "nova-3"}, ), SimpleNamespace( event_type="usage.tts", @@ -70,13 +78,14 @@ def test_session_totals_sums_across_events() -> None: assert totals["llm"]["total_tokens"] == 2500 # 2000*0.30/1e6 + 500*2.50/1e6 = 0.0006 + 0.00125 = 0.00185 assert totals["llm"]["cost_usd"] == 0.00185 - # 30/60 * 0.0043 = 0.00215 + # 30/60 * 0.0077 = 0.00385 assert totals["stt"]["audio_seconds"] == 30.0 - assert totals["stt"]["cost_usd"] == 0.00215 + assert totals["stt"]["streamed_seconds"] == 30.0 + assert totals["stt"]["cost_usd"] == 0.00385 # 500/1e6 * 30.0 = 0.015 assert totals["tts"]["characters"] == 500 assert totals["tts"]["cost_usd"] == 0.015 - assert totals["total_cost_usd"] == 0.019 + assert totals["total_cost_usd"] == 0.0207 assert totals["stt"]["source"] == "runtime" assert totals["tts"]["source"] == "runtime" @@ -125,6 +134,7 @@ def test_session_totals_ignores_legacy_provider_usage_reconciliation_events() -> totals = session_totals(events) assert totals["stt"]["audio_seconds"] == 30.0 + assert totals["stt"]["streamed_seconds"] == 30.0 assert totals["stt"]["cost_usd"] == 0.00215 assert totals["stt"]["source"] == "runtime" assert totals["tts"]["characters"] == 500 diff --git a/server/tests/test_stt_evaluation_pricing.py b/server/tests/test_stt_evaluation_pricing.py index bf97d28..eeec2e4 100644 --- a/server/tests/test_stt_evaluation_pricing.py +++ b/server/tests/test_stt_evaluation_pricing.py @@ -18,9 +18,11 @@ def test_compute_duration_seconds_for_stereo_pcm16() -> None: def test_compute_all_model_costs_reuses_same_duration() -> None: costs = compute_all_model_costs(60.0) - assert costs["deepgram"]["nova-3-monolingual"] == 0.0048 + assert costs["deepgram"]["nova-3"] == 0.0043 + assert costs["deepgram"]["nova-3-streaming"] == 0.0077 assert costs["elevenlabs"]["scribe-v2"] == 0.00367 - assert costs["sarvam"]["saarika"] == 0.006 + assert costs["sarvam"]["saarika"] == 0.005263 + assert costs["sarvam"]["saarika-diarization"] == 0.007895 def test_compute_model_cost_returns_zero_for_unknown_model() -> None: From 8dc393ffcc36ef5dc6c5660eadfb3a1f026bcd2b Mon Sep 17 00:00:00 2001 From: abhinav-t41 Date: Thu, 16 Jul 2026 10:43:11 +0530 Subject: [PATCH 3/4] Meter TTS usage on characters sent to the provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TTS providers bill on characters sent over the wire, not characters the user hears: Pipecat pushes each sentence to the provider as the LLM streams it, so on interruption Deepgram has already received (and bills for) text whose audio is discarded by Clear. Pipecat's built-in TTSUsageMetricsData metric cannot measure this — the Deepgram websocket service never emits it, and the base-class fallback drops accumulated text on interruption, which is exactly where billing diverges from the transcript. Add TtsUsageMeterMixin counting characters at the run_tts seam — the exact prepared text sent to the provider — on both instrumented TTS services, and emit a usage.tts trace event (provider, model, voice, sent_characters) at session teardown next to usage.stt. Also rewrite docs/cost-metering-design.md §6 with the verified billing semantics: sent = billed for Deepgram; ElevenLabs deducts credits on successful generation so sent_characters is a tight upper bound there; per-provider differences belong in the pricing catalog, not the meter. Co-Authored-By: Claude Fable 5 --- docs/cost-metering-design.md | 162 +++++++++++++----- .../app/services/pipecat_streaming_runtime.py | 43 ++++- server/tests/test_tts_usage_meter.py | 56 ++++++ 3 files changed, 216 insertions(+), 45 deletions(-) create mode 100644 server/tests/test_tts_usage_meter.py diff --git a/docs/cost-metering-design.md b/docs/cost-metering-design.md index ec6cfa8..3e98452 100644 --- a/docs/cost-metering-design.md +++ b/docs/cost-metering-design.md @@ -24,7 +24,7 @@ N providers is pure arithmetic against a rate table. The billing units are: | Service | Billing unit | Comparable across providers? | |---------|-------------|------------------------------| | STT | Seconds of **audio streamed** to the provider (silence included) | Yes — same audio would be streamed to any provider | -| TTS | **Characters** of input text | Yes — exactly, the text is identical regardless of provider | +| TTS | Characters **sent** to the provider (including text cleared on interruption, see §6) | Yes — the pipeline sends the same text to any provider; interruption timing, not the provider, decides what gets sent | | LLM | Input/output **tokens** | Approximately (tokenizers differ) — out of scope for now | A direct corollary: **do not run shadow sessions on other providers to learn @@ -101,8 +101,9 @@ Emit immutable usage facts alongside existing transcript events. **No prices anywhere in these events** — prices change, facts don't. - `usage.stt` → `{ provider, model, streamed_seconds, speech_seconds }` -- `usage.tts` → `{ processor, model, characters }` (one per synthesized - utterance; sum per run at aggregation time) +- `usage.tts` → `{ provider, model, voice, sent_characters }` + (one per session, emitted at teardown like `usage.stt`; see §6. + `heard_characters` is a future analytics addition) - `usage.llm` → `{ model, input_tokens, output_tokens }` (future) ### Layer 2 — One pricing catalog, server-side @@ -228,47 +229,120 @@ revisit once metering is solid.) --- -## 6. TTS metering: Pipecat already does this — it's switched off - -Pipecat has a built-in TTS usage metric. Every TTS service calls -`start_tts_usage_metrics(text)` after synthesizing, emitting a `MetricsFrame` -carrying `TTSUsageMetricsData(value=)`. Verified in the -installed package for both providers we use: - -- Deepgram: `pipecat/services/deepgram/tts.py:497` -- ElevenLabs: `pipecat/services/elevenlabs/tts.py:1026` - -It is gated behind the flag we currently disable: +## 6. TTS metering: bill = characters *sent*, and Pipecat's built-in metric can't measure that + +*(Rewritten 2026-07-16 — the earlier version of this section was wrong on two +counts, marked below.)* + +### 6.1 How Deepgram actually bills + +- Aura is billed per input character: Aura-2 $0.030/1k chars pay-as-you-go + ($0.027 Growth), Aura-1 $0.015/1k. +- We use the **websocket** TTS API. Deepgram counts characters **sent to the + websocket** — its own throughput limit is documented as "measured by the + number of characters sent to the websocket." Nothing in the `Clear` docs + promises a refund for buffered text cleared before synthesis; assume + **sent = billed** until reconciled against the console (§6.4). +- **Interruption effect:** Pipecat pushes each sentence to the socket as soon + as the LLM streams it — usually several sentences ahead of audio playback. + On user interruption, Pipecat sends `Clear` + (`pipecat/services/deepgram/tts.py:274`); audio stops, but every character + already submitted via `Speak` messages was sent and is billed. So + **Deepgram's bill > characters the user actually heard**, and > any count + derived from the spoken transcript. Same lesson as STT: the billing unit is + what crosses the wire (characters sent), not what the user experiences + (characters heard). + +### 6.2 Why `enable_usage_metrics=True` is NOT sufficient (correction) + +The earlier draft said flipping the flag suffices. Verified against the +installed package, it does not: + +- Our `AdkDeepgramTTSService` extends the **websocket** `DeepgramTTSService`, + whose `run_tts` never calls `start_tts_usage_metrics`. The call the earlier + draft cited (`deepgram/tts.py:497`) is in `DeepgramHttpTTSService`, which we + don't use. In our default SENTENCE aggregation mode, Deepgram WS TTS would + report **zero** usage. +- The base-class fallback (`_streamed_text` accumulated and emitted at + `LLMFullResponseEndFrame`, `tts_service.py:723-726`) only operates in TOKEN + aggregation mode — and even there, `_handle_interruption` + (`tts_service.py:902-910`) wipes `_streamed_text` **without emitting**, so + interrupted turns are never counted. Interrupted turns are exactly where + billing diverges from the transcript, so this is the worst possible gap. +- ElevenLabs WS `run_tts` *does* call `start_tts_usage_metrics(text)` right + after sending (`elevenlabs/tts.py:1026`), i.e. the built-in metric is + per-provider inconsistent: roughly right for ElevenLabs, absent for + Deepgram. + +### 6.3 The fix: meter at the send seam (mirror of `SttUsageMeterMixin`) + +Count characters ourselves at the exact point text leaves for the provider — +`run_tts(text, context_id)`, which receives the final prepared/transformed +text (post `normalize_tts_text`) and immediately sends it (`Speak` for +Deepgram, `_send_text` for ElevenLabs): ```python -# pipecat_streaming_runtime.py:565 -enable_usage_metrics=False, # ← flip to True -``` - -Our `MetricsSink` (`server/app/services/pipeline_metrics.py`) already sits at -the pipeline tail and iterates `MetricsFrame` items for TTFB, so TTS metering -is a small addition: - -```python -from pipecat.metrics.metrics import TTFBMetricsData, TTSUsageMetricsData - -# in MetricsSink._handle_metric: -elif isinstance(item, TTSUsageMetricsData): - await self._record_trace( - "usage.tts", - {"processor": item.processor, "model": item.model, "characters": int(item.value)}, - ) +class TtsUsageMeterMixin: + """Count characters at the run_tts seam — the exact text sent to the provider.""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._sent_characters = 0 + + async def run_tts(self, text: str, context_id: str): + self._sent_characters += len(text) + async for frame in super().run_tts(text, context_id): + yield frame ``` -Since TTS providers bill on input characters and the text is identical -regardless of provider, the cross-provider TTS comparison is essentially -exact (modulo per-provider rounding/minimum rules from the pricing catalog). - -**Caveat:** `enable_usage_metrics=True` is pipeline-wide. If the pipecat-adk -LLM bridge emits `LLMUsageMetricsData` (token counts), those frames will also -reach `MetricsSink` — handle or ignore them explicitly rather than letting -them fall through silently. If the bridge doesn't emit them, LLM tokens can -later be read from ADK/Gemini response `usage_metadata`. +Apply to both `AdkDeepgramTTSService` and `AdkElevenLabsTTSService`. At +session teardown (next to `usage.stt`), emit: + +- `usage.tts` → `{ provider, model, voice, sent_characters }` + - `sent_characters` — billing truth. + - `heard_characters` (future, analytics) — characters actually played to + the user (derivable from the `agent.text` transcript events). The delta + quantifies interruption waste — worth surfacing in the UI as its own + insight ("23% of TTS spend was interrupted audio"). + +### 6.4 Calibration + +- **Deepgram:** the Management/Usage API reports per-request character counts; + the console usage page shows the same. Run one deliberately-interrupted + session, compare our `sent_characters` against Deepgram's reported count — + this both validates the meter and settles the cleared-text question + empirically (same role `ListenV1Metadata.duration` plays for STT). +- **ElevenLabs:** the subscription/user API exposes a cumulative character + counter; snapshot before/after a session. + +### 6.5 Cross-provider comparison remains valid + +`sent_characters` is **pipeline-determined** (LLM sentence pacing + +interruption timing), not provider-determined — the same conversation with +the same interruptions pushes the same text to whichever TTS service sits in +the pipeline, and both Deepgram and ElevenLabs bill on characters received. +So `sent_characters × rate` extrapolates fairly. Catalog detail: ElevenLabs +bills in credits with per-model multipliers (e.g. Flash v2.5 ≈ 0.5 +credits/char vs Multilingual v2 at 1 credit/char) — encode as model-specific +rates, and label its effective $/char from a subscription tier explicitly, +since ElevenLabs pricing is plan-based rather than pure pay-as-you-go. + +**Per-provider interruption semantics differ — in the catalog, not the +meter.** ElevenLabs' stated policy is that credits are deducted on +*successful audio generation* (failed requests aren't charged), and on +interruption Pipecat sends `close_context` (`elevenlabs/tts.py:843-845`), so +text buffered but never synthesized *may* not be billed — unlike Deepgram, +where we must assume sent = billed. Neither vendor documents the +interrupted-buffer case precisely, and since ElevenLabs synthesizes eagerly +per chunk, sent ≈ generated in practice. Treat `sent_characters` as exact for +Deepgram and as a tight upper bound for ElevenLabs; per-provider calibration +(§6.4) is what turns these assumptions into measured facts. The meter itself +never changes per provider — one `TtsUsageMeterMixin` on the shared `run_tts` +seam covers every Pipecat TTS service, current and future. + +**Caveat (unchanged):** if we also flip `enable_usage_metrics=True` for other +reasons, `LLMUsageMetricsData` frames may reach `MetricsSink` — handle or +ignore them explicitly. LLM tokens can otherwise be read from ADK/Gemini +response `usage_metadata`. --- @@ -282,9 +356,11 @@ later be read from ADK/Gemini response `usage_metadata`. trace event (streamed + speech + provider-reported seconds) is emitted at session teardown; Deepgram metadata `duration` is accumulated for calibration. Tests in `server/tests/test_stt_usage_meter.py`. -3. **TTS metering** — flip `enable_usage_metrics=True`; add - `TTSUsageMetricsData` branch to `MetricsSink` emitting `usage.tts`; - explicitly handle/ignore `LLMUsageMetricsData`. +3. **TTS metering** — `TtsUsageMeterMixin` counting `len(text)` in `run_tts` + on both instrumented TTS subclasses (do NOT rely on + `enable_usage_metrics` — see §6.2); emit `usage.tts` with + `sent_characters` (+ optional `heard_characters`) at session teardown; + calibrate one interrupted session against Deepgram console usage. 4. **Aggregation** — at run close, aggregate `usage.*` events → compute actual + hypothetical costs with rate snapshots → store in `runs.summary` (or `run_costs` table). Stop treating `metrics.jsonl` as diff --git a/server/app/services/pipecat_streaming_runtime.py b/server/app/services/pipecat_streaming_runtime.py index c6ff552..f775447 100644 --- a/server/app/services/pipecat_streaming_runtime.py +++ b/server/app/services/pipecat_streaming_runtime.py @@ -199,7 +199,35 @@ async def _on_message(self, message: Any) -> None: await super()._on_message(message) -class AdkDeepgramTTSService(ProviderRequestTraceMixin, VqlTTSMixin, DeepgramTTSService): +class TtsUsageMeterMixin: + """Meter characters actually sent to the TTS provider. + + `run_tts` receives the final prepared text (post text-transforms) and + immediately sends it to the provider (`Speak` message for Deepgram, + context text for ElevenLabs). Providers bill on characters sent — + including text later cleared/closed by an interruption — so this counts + the billed quantity. Transcript-derived counts undercount whenever the + user interrupts, and Pipecat's built-in `TTSUsageMetricsData` metric is + never emitted by the Deepgram websocket service and is dropped on + interruption, so we count at the send seam ourselves (same approach as + `SttUsageMeterMixin`). + """ + + _sent_characters: int = 0 + + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]: + self._sent_characters += len(text) + async for frame in super().run_tts(text, context_id): + yield frame + + @property + def sent_characters(self) -> int: + return self._sent_characters + + +class AdkDeepgramTTSService( + TtsUsageMeterMixin, ProviderRequestTraceMixin, VqlTTSMixin, DeepgramTTSService +): def __init__( self, *, @@ -261,7 +289,9 @@ async def _process_response(self, data: dict): await super()._process_response(data) -class AdkElevenLabsTTSService(ProviderRequestTraceMixin, VqlTTSMixin, ElevenLabsTTSService): +class AdkElevenLabsTTSService( + TtsUsageMeterMixin, ProviderRequestTraceMixin, VqlTTSMixin, ElevenLabsTTSService +): def __init__( self, *, @@ -657,6 +687,15 @@ async def on_session_timeout(_transport, _websocket): "provider_reported_seconds": stt.provider_reported_audio_seconds, }, ) + await record_trace( + "usage.tts", + { + "provider": config.tts_provider, + "model": metrics_tts_model, + "voice": config.tts_voice, + "sent_characters": tts.sent_characters, + }, + ) finally: await stt_evaluation.finalize() logger.info("pipecat streaming test-call ended run_id=%s", run_id) diff --git a/server/tests/test_tts_usage_meter.py b/server/tests/test_tts_usage_meter.py new file mode 100644 index 0000000..479d668 --- /dev/null +++ b/server/tests/test_tts_usage_meter.py @@ -0,0 +1,56 @@ +from collections.abc import AsyncGenerator + +import pytest +from pipecat.frames.frames import Frame + +from app.services.pipecat_streaming_runtime import TtsUsageMeterMixin + + +class _FakeTTSService: + def __init__(self) -> None: + self.received: list[str] = [] + + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]: + self.received.append(text) + yield None + + +class _MeteredService(TtsUsageMeterMixin, _FakeTTSService): + pass + + +async def _drain(gen: AsyncGenerator) -> None: + async for _ in gen: + pass + + +@pytest.mark.asyncio +async def test_sent_characters_counts_all_text_sent_to_provider() -> None: + service = _MeteredService() + await _drain(service.run_tts("Hello there.", "ctx-1")) + await _drain(service.run_tts("How can I help?", "ctx-1")) + assert service.sent_characters == len("Hello there.") + len("How can I help?") + assert service.received == ["Hello there.", "How can I help?"] + + +@pytest.mark.asyncio +async def test_sent_characters_counted_even_when_generator_not_drained() -> None: + # Characters are billed once sent; an interruption that abandons the + # generator must not un-count text that already left for the provider. + service = _MeteredService() + gen = service.run_tts("This sentence gets interrupted mid-playback.", "ctx-1") + await gen.__anext__() + await gen.aclose() + assert service.sent_characters == len("This sentence gets interrupted mid-playback.") + + +@pytest.mark.asyncio +async def test_meter_state_is_per_instance() -> None: + first = _MeteredService() + second = _MeteredService() + await _drain(first.run_tts("only counted on first", "ctx-1")) + assert second.sent_characters == 0 + + +def test_sent_characters_defaults_to_zero() -> None: + assert _MeteredService().sent_characters == 0 From 2c83c8b96115217d72a6bef70c879e13f85ed653 Mon Sep 17 00:00:00 2001 From: shubhamthink41 Date: Thu, 16 Jul 2026 14:34:19 +0530 Subject: [PATCH 4/4] Add TTS metrics and cost comparison to AudioView --- client/src/components/audio/AudioView.tsx | 141 ++++++++++++------ client/src/lib/types.ts | 2 + server/app/api/routes/audio_evaluations.py | 12 +- .../app/services/pipecat_streaming_runtime.py | 2 +- server/app/services/stt_evaluation_service.py | 32 ++-- server/app/services/tts_evaluation_pricing.py | 86 +++++++++++ server/tests/test_tts_evaluation_pricing.py | 60 ++++++++ 7 files changed, 276 insertions(+), 59 deletions(-) create mode 100644 server/app/services/tts_evaluation_pricing.py create mode 100644 server/tests/test_tts_evaluation_pricing.py diff --git a/client/src/components/audio/AudioView.tsx b/client/src/components/audio/AudioView.tsx index e3db43f..63c35df 100644 --- a/client/src/components/audio/AudioView.tsx +++ b/client/src/components/audio/AudioView.tsx @@ -117,6 +117,7 @@ export function AudioView({ agent, records }: AudioViewProps) { +
@@ -151,59 +152,111 @@ export function AudioView({ agent, records }: AudioViewProps) {
-
-
-

Cost comparision across models

-

Computed from this audio session’s total user-turn duration. No extra provider calls required.

-
- {(() => { - const allEntries = Object.entries(selectedRecord.session_model_costs_usd).flatMap(([provider, models]) => - Object.entries(models).map(([model, cost]) => ({ provider, model, cost })), - ); - const rankedAll = [...allEntries].sort((a, b) => a.cost - b.cost); - const globalLowest = rankedAll[0] ?? null; - const globalHighest = rankedAll[rankedAll.length - 1] ?? null; - const hasSpread = rankedAll.length > 1; + + + {selectedRecord.session_tts_sent_characters != null || + (selectedRecord.session_tts_model_costs_usd && + Object.keys(selectedRecord.session_tts_model_costs_usd).length > 0) ? ( + <> +
+ + {selectedRecord.session_tts_sent_characters != null ? ( +
+ +
+ ) : null} + {selectedRecord.session_tts_model_costs_usd && + Object.keys(selectedRecord.session_tts_model_costs_usd).length > 0 ? ( + + ) : null} + + ) : null} + + ) : null} + +
+ ); +} + +function CostComparisonGrid({ + title, + subtitle, + costs, + providerSuffix, +}: { + title: string; + subtitle: string; + costs: Record>; + providerSuffix: string; +}) { + const allEntries = Object.entries(costs).flatMap(([provider, models]) => + Object.entries(models).map(([model, cost]) => ({ provider, model, cost })), + ); + const rankedAll = [...allEntries].sort((a, b) => a.cost - b.cost); + const globalLowest = rankedAll[0] ?? null; + const globalHighest = rankedAll[rankedAll.length - 1] ?? null; + const hasSpread = rankedAll.length > 1; + return ( +
+
+

{title}

+

{subtitle}

+
+
+ {Object.entries(costs).map(([provider, models]) => ( +
+
+ {titleCase(provider)} {providerSuffix} +
+
+ {Object.entries(models).map(([model, cost]) => { + const isHighest = hasSpread && globalHighest?.provider === provider && globalHighest?.model === model; + const isLowest = hasSpread && globalLowest?.provider === provider && globalLowest?.model === model; return ( -
- {Object.entries(selectedRecord.session_model_costs_usd).map(([provider, models]) => ( -
-
{titleCase(provider)} STT
-
- {Object.entries(models).map(([model, cost]) => { - const isHighest = hasSpread && globalHighest?.provider === provider && globalHighest?.model === model; - const isLowest = hasSpread && globalLowest?.provider === provider && globalLowest?.model === model; - return ( -
-
-
{model}
-
- {isHighest ? ( - Highest - ) : null} - {isLowest ? ( - Lowest - ) : null} -
-
- ~${cost.toFixed(6)} -
- ); - })} -
+
+
+
{model}
+
+ {isHighest ? ( + Highest + ) : null} + {isLowest ? ( + Lowest + ) : null}
- ))} +
+ ~${cost.toFixed(6)}
); - })()} + })}
- - ) : null} - +
+ ))} +
); } +function SectionHeader({ label }: { label: string }) { + return ( +

{label}

+ ); +} + function MetricCard({ label, value, sub }: { label: string; value: string; sub: string }) { return (
diff --git a/client/src/lib/types.ts b/client/src/lib/types.ts index 311dd22..e1965e6 100644 --- a/client/src/lib/types.ts +++ b/client/src/lib/types.ts @@ -77,4 +77,6 @@ export interface AudioEvaluationRecord { provider_session_metrics: Record; file_paths: string[]; evaluate_mode: boolean; + session_tts_sent_characters?: number | null; + session_tts_model_costs_usd?: Record>; } diff --git a/server/app/api/routes/audio_evaluations.py b/server/app/api/routes/audio_evaluations.py index ff53ed8..36402fd 100644 --- a/server/app/api/routes/audio_evaluations.py +++ b/server/app/api/routes/audio_evaluations.py @@ -3,7 +3,7 @@ from typing import Annotated, Any from fastapi import APIRouter, Depends -from pydantic import BaseModel +from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import get_settings @@ -35,6 +35,8 @@ class AudioEvaluationRead(BaseModel): provider_session_metrics: dict[str, AudioProviderMetricsRead] file_paths: list[str] evaluate_mode: bool + session_tts_sent_characters: int | None = None + session_tts_model_costs_usd: dict[str, dict[str, float]] = Field(default_factory=dict) @router.get("/agent/{agent_id}", response_model=list[AudioEvaluationRead]) @@ -63,6 +65,8 @@ async def list_audio_evaluations(agent_id: str, session: SessionDep) -> list[Aud provider_session_metrics=payload["provider_session_metrics"], file_paths=payload["file_paths"], evaluate_mode=payload["evaluate_mode"], + session_tts_sent_characters=payload["session_tts_sent_characters"], + session_tts_model_costs_usd=payload["session_tts_model_costs_usd"], ) ) return records @@ -95,4 +99,10 @@ def _load_metrics_summary(metrics_path: Path) -> dict[str, Any] | None: "session_model_costs_usd": session_summary.get("session_model_costs_usd") or {}, "provider_session_metrics": session_summary.get("provider_session_metrics") or {}, "evaluate_mode": bool(session_summary.get("evaluate_mode", False)), + "session_tts_sent_characters": ( + int(session_summary["session_tts_sent_characters"]) + if session_summary.get("session_tts_sent_characters") is not None + else None + ), + "session_tts_model_costs_usd": session_summary.get("session_tts_model_costs_usd") or {}, } diff --git a/server/app/services/pipecat_streaming_runtime.py b/server/app/services/pipecat_streaming_runtime.py index f775447..f3d4221 100644 --- a/server/app/services/pipecat_streaming_runtime.py +++ b/server/app/services/pipecat_streaming_runtime.py @@ -697,5 +697,5 @@ async def on_session_timeout(_transport, _websocket): }, ) finally: - await stt_evaluation.finalize() + await stt_evaluation.finalize(tts_sent_characters=tts.sent_characters) logger.info("pipecat streaming test-call ended run_id=%s", run_id) diff --git a/server/app/services/stt_evaluation_service.py b/server/app/services/stt_evaluation_service.py index ce0ea2c..a563a6a 100644 --- a/server/app/services/stt_evaluation_service.py +++ b/server/app/services/stt_evaluation_service.py @@ -21,6 +21,9 @@ pricing_model_name, ) from app.services.stt_evaluation_store import SavedTurnAudio, SttEvaluationStore +from app.services.tts_evaluation_pricing import ( + compute_all_model_costs as compute_all_tts_model_costs, +) logger = logging.getLogger("uvicorn.error") TraceRecorder = Callable[[str, dict[str, object]], Awaitable[None]] @@ -156,22 +159,25 @@ async def handle_user_turn_audio( self._background_tasks.add(task) task.add_done_callback(self._background_tasks.discard) - async def finalize(self) -> None: + async def finalize(self, *, tts_sent_characters: int | None = None) -> None: if self._background_tasks: await asyncio.gather(*self._background_tasks, return_exceptions=True) - await self._store.append_metrics( - self._session_id, - { - "type": "session.summary", - "session_id": self._session_id, - "run_id": self._run_id, - "session_stt_duration_sec": round(self._session_duration_sec, 3), - "session_model_costs_usd": compute_session_model_costs(self._session_duration_sec), - "provider_session_metrics": self._provider_session_metrics(), - "evaluate_mode": self._evaluate_mode, - }, - ) + summary: dict[str, Any] = { + "type": "session.summary", + "session_id": self._session_id, + "run_id": self._run_id, + "session_stt_duration_sec": round(self._session_duration_sec, 3), + "session_model_costs_usd": compute_session_model_costs(self._session_duration_sec), + "provider_session_metrics": self._provider_session_metrics(), + "evaluate_mode": self._evaluate_mode, + } + if tts_sent_characters is not None: + summary["session_tts_sent_characters"] = int(tts_sent_characters) + summary["session_tts_model_costs_usd"] = compute_all_tts_model_costs( + int(tts_sent_characters) + ) + await self._store.append_metrics(self._session_id, summary) await self._record_trace( "evaluation.stt.session_summary", { diff --git a/server/app/services/tts_evaluation_pricing.py b/server/app/services/tts_evaluation_pricing.py new file mode 100644 index 0000000..7b725a0 --- /dev/null +++ b/server/app/services/tts_evaluation_pricing.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from decimal import Decimal + +# USD per 1,000,000 characters sent to the TTS provider. +# +# Rates are pay-as-you-go list prices. Enterprise / committed-spend rates +# can be materially lower. Sources: +# - Deepgram TTS pricing: https://deepgram.com/pricing +# - ElevenLabs pricing: https://elevenlabs.io/pricing +# +# TTS providers price per *model*, not per *voice*. All Aura-2 voices +# (thalia, asteria, luna, orion, ...) share the aura-2 rate. All ElevenLabs +# voices used with eleven_multilingual_v2 share the same rate; the same +# voice on eleven_flash_v2_5 uses the flash rate instead. +# +# ElevenLabs bills in "credits per character" (Flash/Turbo = 0.5×, +# Multilingual v2 / v1 / Monolingual v1 = 1×, v3 alpha = 1× for now). +# The dollar-per-M-char figures below convert those credit ratios against +# the $165/M anchor used in the legacy pricing.py module. +RATE_PER_MILLION_CHARS_USD: dict[str, dict[str, Decimal]] = { + "deepgram": { + "aura-2": Decimal("30.00"), + "aura": Decimal("15.00"), + }, + "elevenlabs": { + # UI has no ElevenLabs model dropdown; the app drives eleven_turbo_v2_5 + # as the default (see server/app/schemas/agent.py). Add more entries here + # only when the UI actually exposes them. + "eleven_turbo_v2_5": Decimal("82.50"), + }, +} + +# Deepgram Aura model IDs are voice-suffixed at runtime (e.g. `aura-2-thalia-en`, +# `aura-2-asteria-en`). We resolve any string starting with a known prefix to the +# parent model rate, so new voices don't need explicit entries. +DEEPGRAM_MODEL_PREFIXES: tuple[str, ...] = ("aura-2", "aura") + +# ElevenLabs model IDs are stable exact strings; users occasionally pass the +# dash-hyphenated variant. Map those to the underscore form used above. +ELEVENLABS_MODEL_ALIASES: dict[str, str] = { + "v3": "eleven_v3", + "eleven-v3": "eleven_v3", + "multilingual-v2": "eleven_multilingual_v2", + "eleven-multilingual-v2": "eleven_multilingual_v2", + "multilingual-v1": "eleven_multilingual_v1", + "monolingual-v1": "eleven_monolingual_v1", + "flash-v2-5": "eleven_flash_v2_5", + "flash-v2": "eleven_flash_v2", + "turbo-v2-5": "eleven_turbo_v2_5", + "turbo-v2": "eleven_turbo_v2", +} + + +def pricing_model_name(*, provider: str, model: str) -> str: + if provider == "deepgram": + for prefix in DEEPGRAM_MODEL_PREFIXES: + if model == prefix or model.startswith(f"{prefix}-") or model.startswith(f"{prefix}_"): + return prefix + return model + if provider == "elevenlabs": + if model in RATE_PER_MILLION_CHARS_USD["elevenlabs"]: + return model + return ELEVENLABS_MODEL_ALIASES.get(model, model) + return model + + +def compute_model_cost(sent_characters: int, *, provider: str, model: str) -> float: + pricing_model = pricing_model_name(provider=provider, model=model) + rate = RATE_PER_MILLION_CHARS_USD.get(provider, {}).get(pricing_model) + if rate is None: + return 0.0 + chars = Decimal(str(max(sent_characters, 0))) / Decimal("1000000") + return _q(chars * rate) + + +def compute_all_model_costs(sent_characters: int) -> dict[str, dict[str, float]]: + chars = Decimal(str(max(sent_characters, 0))) / Decimal("1000000") + return { + provider: {model: _q(chars * rate) for model, rate in models.items()} + for provider, models in RATE_PER_MILLION_CHARS_USD.items() + } + + +def _q(value: Decimal) -> float: + return float(value.quantize(Decimal("0.000001"))) diff --git a/server/tests/test_tts_evaluation_pricing.py b/server/tests/test_tts_evaluation_pricing.py new file mode 100644 index 0000000..a169240 --- /dev/null +++ b/server/tests/test_tts_evaluation_pricing.py @@ -0,0 +1,60 @@ +from app.services.tts_evaluation_pricing import ( + compute_all_model_costs, + compute_model_cost, + pricing_model_name, +) + + +def test_deepgram_voice_suffixed_models_resolve_to_aura_2() -> None: + for voice_model in ( + "aura-2", + "aura-2-thalia-en", + "aura-2-asteria-en", + "aura-2-luna-en", + "aura-2-orion-en", + ): + assert pricing_model_name(provider="deepgram", model=voice_model) == "aura-2" + + +def test_deepgram_legacy_aura_voice_suffixes_resolve_to_aura() -> None: + for voice_model in ("aura", "aura-luna-en", "aura-orion-en"): + assert pricing_model_name(provider="deepgram", model=voice_model) == "aura" + + +def test_elevenlabs_dash_variants_alias_to_underscore_model_ids() -> None: + assert ( + pricing_model_name(provider="elevenlabs", model="turbo-v2-5") + == "eleven_turbo_v2_5" + ) + + +def test_elevenlabs_exact_model_ids_pass_through() -> None: + assert ( + pricing_model_name(provider="elevenlabs", model="eleven_turbo_v2_5") + == "eleven_turbo_v2_5" + ) + + +def test_compute_model_cost_uses_resolved_rate() -> None: + # 1M chars on aura-2 should be $30 + assert compute_model_cost(1_000_000, provider="deepgram", model="aura-2-thalia-en") == 30.0 + # 1M chars on eleven_turbo_v2_5 should be $82.50 + assert ( + compute_model_cost(1_000_000, provider="elevenlabs", model="turbo-v2-5") + == 82.5 + ) + + +def test_compute_model_cost_returns_zero_for_unknown_model() -> None: + assert compute_model_cost(1_000_000, provider="deepgram", model="unknown") == 0.0 + assert compute_model_cost(1_000_000, provider="unknown", model="whatever") == 0.0 + + +def test_compute_all_model_costs_covers_full_model_list() -> None: + costs = compute_all_model_costs(1_000_000) + assert costs["deepgram"] == {"aura-2": 30.0, "aura": 15.0} + assert costs["elevenlabs"] == {"eleven_turbo_v2_5": 82.5} + + +def test_compute_costs_clamp_negative_characters_to_zero() -> None: + assert compute_model_cost(-100, provider="deepgram", model="aura-2") == 0.0