Text-to-Speech - Tuner Documentation

This guide uses tuner-tts-observer. It assumes you have a TunerSession set up — see the Quickstart for the session lifecycle and credentials.

tuner-tts-observer wraps your TTS provider’s synthesis stream to capture the agent side of the conversation — transcript, TTS time-to-first-byte, and end-to-end latency. Tuner observes synthesis, it does not own or drive it: removing the context managers leaves your synthesis code completely unaffected.

Supported adapters

Provider	Adapter	Status
Cartesia	`CartesiaAdapter`	✅ Supported
ElevenLabs	—	SOON
OpenAI TTS	—	SOON

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

Install

pip install tuner-core tuner-tts-observer

WebSocket support (recommended) requires the cartesia extra:

pip install "tuner-tts-observer[cartesia]"

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. Attach the adapter once per call, then wrap each turn’s receive loop with track_ws():

from tuner_tts_observer import CartesiaAdapter

tts = session.attach(CartesiaAdapter())

# around your existing Cartesia WebSocket receive loop, per turn:
async with tts.track_ws(agent_text) as tracked:
    async for chunk in tracked(ctx.receive()):    # ctx: your Cartesia context
        if barge_in_event.is_set():               # your barge-in signal
            tts.mark_interrupted()
            break
        ...                                        # your playback / send code
    else:
        await tts.wait_for_interruption(barge_in_event)

Word-accurate interruption capture requires add_timestamps=True on your Cartesia context:

ctx = await tts_ws.context(
    ...,                    # your model, voice, and output format
    add_timestamps=True,    # required for word-accurate interruption capture
)

Without it, interruptions are still recorded, but the spoken text cannot be cut at the word the user interrupted. TTS providers stream faster than real-time, so your loop can finish seconds before the audio finishes playing out loud. mark_interrupted() on the break path handles a mid-stream interruption; wait_for_interruption() on the else path keeps watching during that remaining gap and calls mark_interrupted() for you if one happens there. Skip the else branch and any interruption after the loop ends naturally is silently lost.

What your app owns

This package only records interruptions — it never detects or acts on them. You need: (1) a detection signal (VAD, provider event, client-side signal), (2) a shared flag or event, (3) the actual stop — breaking your loop, and telling the client to stop already-buffered audio if detection was server-side:

# server, on server-side detection
barge_in_event.set()
await websocket.send_text(json.dumps({"type": "stop_playback"}))

// client
if (msg.type === 'stop_playback') stopPlaybackImmediately()

Both mark_interrupted() and wait_for_interruption() are optional — skip them and turns just record in full, nothing breaks.

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 — the full intended response is recorded. Wrap your synthesis stream with track():

from tuner_tts_observer import CartesiaAdapter

tts = session.attach(CartesiaAdapter())

# around your existing SSE synthesis loop, per turn:
with tts.track(agent_text) as tracked:
    for chunk in tracked(your_sse_stream):   # e.g. cartesia_client.tts.sse(...)
        ...                                   # your playback / send code

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

Signal	SSE	WebSocket
Agent transcript text	✓ full text	✓ spoken words only on interruption
Turn start timestamp	✓	✓
Turn duration	✓	✓
TTS TTFB	✓	✓
E2e latency	✓	✓
LLM latency	✓*	✓*
`interrupted: true` on barge-in	✓	✓
Word-accurate spoken text cut	✗	✓

^* Populated from the session — requires a LangChain/LangGraph handler (tuner-langchain) attached to the same session. With no LLM handler attached, this field is absent.

Troubleshooting

Interrupted turns record the full intended response

On WebSocket, confirm add_timestamps=True is set on your Cartesia context — without word timestamps the spoken text cannot be cut.
Confirm your loop has the else: await tts.wait_for_interruption(barge_in_event) branch — without it, interruptions that happen after the loop ends naturally are never recorded.
On SSE this is expected behaviour — word-accurate cutting requires the WebSocket pattern.

Agent turns are missing from the call

Every synthesis turn must go through track_ws() / track() — turns synthesized outside the wrapper are invisible to Tuner.
Confirm the adapter was attached to the session before the first turn: tts = session.attach(CartesiaAdapter()).

Latency numbers look wrong or missing

E2e latency and LLM latency are assembled from other observers on the same session rather than measured by this adapter alone. If either is missing, confirm the relevant observer — tuner-stt-observer for e2e, tuner-langchain for LLM latency — is attached to the same session.

What’s Next?

Quickstart

The session lifecycle, credentials, and every observer wired together.

Speech-to-Text

Capture the user side — transcript and turn timing.

LangChain / LangGraph

Node transitions, tool calls, and LLM latency in your call timeline.

Create Call API

The underlying API every observer feeds into at flush time.

​Supported adapters

​Install

​WebSocket (recommended)

​What your app owns

​SSE (legacy)

​Custom providers

​What gets captured

​Troubleshooting

​What’s Next?

Quickstart

Speech-to-Text

LangChain / LangGraph

Create Call API

Supported adapters

Install

WebSocket (recommended)

What your app owns

SSE (legacy)

Custom providers

What gets captured

Troubleshooting

What’s Next?