Pipecat: speech to text

Transcribe user audio in a Pipecat pipeline with a Sprag transcription model.

SpragSTTService transcribes the user's audio in a Pipecat pipeline. It holds one WebSocket to wss://api.sprag.ai/v1/audio/transcriptions across turns, streams audio as it is captured, and emits each turn's transcript once the turn closes: an InterimTranscriptionFrame per delta, then a TranscriptionFrame.

Setup

from pipecat_sprag.stt import SpragSTTService

stt = SpragSTTService(
    settings=SpragSTTService.Settings(model="rhythm"),
)

Add it to the pipeline in the transcribe position, between the transport input and the user context aggregator:

pipeline = Pipeline([
    transport.input(),
    stt,
    user_aggregator,
    llm,
    tts,
    transport.output(),
    assistant_aggregator,
])

Argument	Required	Type	Notes
`settings.model`	No	string	A transcription model id from Models. Defaults to `rhythm`.
`turn_detection`	No	dict or `False`	Who closes a turn; see Turn detection. Defaults to `False`.
`api_key`	No	string	Your Sprag API key. Defaults to `SPRAG_API_KEY`.
`base_url`	No	string	Defaults to `wss://api.sprag.ai/v1/audio/transcriptions`.

Turn detection

A turn is transcribed once it closes, so its transcript arrives after the user stops speaking rather than word by word. turn_detection sets who closes the turn.

If turn_detection is False, a VAD in your pipeline, such as Pipecat's SileroVADAnalyzer, closes the turn, and SpragSTTService commits the audio when the pipeline reports that the user stopped speaking.

If turn_detection is a dict, the server closes the turn when the audio goes quiet, and the pipeline needs no VAD of its own. The dict takes the same fields as the realtime API's turn detection.

stt = SpragSTTService(
    turn_detection={"type": "server_vad", "silence_duration_ms": 800},
)

Word timings and speaker labels

SpragSTTService surfaces transcript text only. For word timings or speaker labels, post the utterance to POST /v1/audio/transcriptions with response_format=verbose_json and, for speakers, speaker_labels=true. See timestamps and diarization.

Limits

More information about session limits can be found in the realtime transcription docs.

Pipecat's OpenAI services

Pipecat's two OpenAI transcription services run against Sprag with base_url and api_key passed to the constructor.

OpenAISTTService posts each turn to POST /v1/audio/transcriptions once the pipeline's VAD closes it. You get one final transcript per turn, after a delay proportional to the turn's length.

import os

from pipecat.services.openai.stt import OpenAISTTService

stt = OpenAISTTService(
    api_key=os.environ["SPRAG_API_KEY"],
    base_url="https://api.sprag.ai/v1",
    settings=OpenAISTTService.Settings(model="rhythm"),
)

OpenAIRealtimeSTTService is the class SpragSTTService extends. With base_url="wss://api.sprag.ai/v1/audio/transcriptions" it behaves the same way. It appends ?intent=transcription to the URL itself, which the route accepts.

import os

from pipecat.services.openai.stt import OpenAIRealtimeSTTService

stt = OpenAIRealtimeSTTService(
    api_key=os.environ["SPRAG_API_KEY"],
    base_url="wss://api.sprag.ai/v1/audio/transcriptions",
    settings=OpenAIRealtimeSTTService.Settings(model="rhythm"),
)