tuner-stt-observer

STT observability for Tuner. Hooks into your existing STT provider's native event system to capture user transcript, turn timing, and STT latency — your WebSocket/streaming loop is completely untouched. This is a pure observer: it never intercepts audio or makes decisions that affect the call (e.g. barge-in/VAD) — that belongs in your own application.

Supported adapters

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

Installation

pip install tuner-core tuner-stt-observer

Each provider adapter lives behind an optional extra so you only pull in the SDKs you actually use:

pip install "tuner-stt-observer[deepgram]"       # DeepgramAdapter
pip install "tuner-stt-observer[speechmatics]"   # SpeechmaticsAdapter

Usage — Deepgram

DeepgramAdapter registers on Open, Transcript, and UtteranceEnd. speech_final transcript fragments are buffered internally and flushed as a single joined utterance when UtteranceEnd fires, so you get one complete user turn per utterance instead of one per fragment.

from tuner_core import TunerConfig, TunerSession
from tuner_stt_observer import DeepgramAdapter

dg_connection = deepgram_client.listen.asyncwebsocket.v("1")

session = TunerSession(config=TunerConfig.from_env(), call_id=call_id)

adapter = DeepgramAdapter(connection=dg_connection)
adapter.on_utterance = lambda text: transcript_queue.put_nowait(text)
session.attach(adapter)

await dg_connection.start(options)   # requires utterance_end_ms + vad_events=True

# In the audio receive loop, send to Deepgram as you normally would —
# the adapter only listens on the connection's events, it never touches audio:
async for chunk in websocket.iter_bytes():
    await dg_connection.send(chunk)

await session.flush()  # at call end

LiveOptions must set utterance_end_ms (e.g. "1000") and vad_events=True — without them UtteranceEnd never fires and the fragment buffer never flushes.

Need barge-in detection? That's call-affecting logic, not observation — build it in your own application (e.g. a local VAD in your audio loop, or Deepgram's own SpeechStarted event), not in this adapter.

Usage — Speechmatics

SpeechmaticsAdapter registers on ADD_TRANSCRIPT and END_OF_UTTERANCE. Speechmatics-rt stacks event handlers rather than replacing them, so this coexists safely with any handlers you register yourself (e.g. for barge-in via ADD_PARTIAL_TRANSCRIPT).

from tuner_stt_observer import SpeechmaticsAdapter

adapter = SpeechmaticsAdapter(client=stt_client)
session.attach(adapter)

# Your own handlers still work — Speechmatics-rt stacks, not replaces:
stt_client.on(SMEvent.ADD_PARTIAL_TRANSCRIPT)(my_barge_in_handler)

Speechmatics reports times already relative to session start, so no timing anchor is needed (unlike Deepgram's stream-relative start field).

Custom providers

For providers other than Deepgram/Speechmatics, extend BaseSTTAdapter from tuner_stt_observer — its docstring documents the full contract (_record_user_turn(), _record_stt_usage(), timestamp conversion) with a worked example.

What gets captured automatically

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy