tuner-tts-observer

TTS observability for Tuner. Wraps your TTS provider's synthesis stream to capture agent transcript and latency — Tuner observes synthesis, it does not own or drive it. Removing the context managers leaves your synthesis code completely unaffected.

Supported adapters

Need a provider that isn't listed? Extend BaseTTSAdapter — see Custom providers below.

Installation

pip install tuner-core tuner-tts-observer

WebSocket support (recommended) requires the websockets extra:

pip install "tuner-tts-observer[cartesia]"
# equivalent to: pip install "cartesia[websockets]>=1.0"

Usage

WebSocket (recommended)

WebSocket is the production-standard pattern for real-time voice agents. It supports word-level interruption detection — when the user interrupts the agent, Tuner records only the words actually spoken, not the full intended response.

import os
from cartesia import AsyncCartesia
from tuner_core import TunerConfig, TunerSession
from tuner_tts_observer import CartesiaAdapter

cartesia_client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])

session = TunerSession(config=TunerConfig.from_env(), call_id=call_id)
adapter = session.attach(CartesiaAdapter())

# Open one WebSocket connection per call — reused across turns
async with cartesia_client.tts.websocket_connect() as tts_ws:

    # Per turn:
    ctx = await tts_ws.context(
        model_id="sonic-2",
        voice={"id": "your-voice-id", "mode": "id"},
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
        add_timestamps=True,  # required for word-level interruption detection
    )
    await ctx.push(agent_text)
    await ctx.no_more_inputs()

    async with adapter.track_ws(agent_text) as tracked:
        async for chunk in tracked(ctx.receive()):
            if barge_in_event.is_set():
                break  # interrupted — Tuner records only what was spoken
            if chunk.audio:
                await websocket.send_bytes(chunk.audio)

await session.flush()

SSE (legacy)

SSE is the simpler pattern, available for HTTP-only stacks or existing integrations. Interruption detection is supported but spoken text cannot be cut accurately.

import os
from cartesia import Cartesia
from tuner_core import TunerConfig, TunerSession
from tuner_tts_observer import CartesiaAdapter

cartesia_client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

session = TunerSession(config=TunerConfig.from_env(), call_id=call_id)
adapter = session.attach(CartesiaAdapter())

with adapter.track(agent_text) as tracked:
    for chunk in tracked(cartesia_client.tts.sse(
        model_id="sonic-2",
        transcript=agent_text,
        voice={"id": "your-voice-id", "mode": "id"},
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
    )):
        if chunk.audio:
            await websocket.send_bytes(chunk.audio)

await session.flush()

Custom providers

For providers other than Cartesia, extend BaseTTSAdapter from tuner_tts_observer — its docstring documents the full contract (timestamping, _record_agent_turn() / _record_tts_usage(), mark_interrupted()) with a worked example.

What gets captured automatically

Development

uv sync
make test