feat(baseten): add Qwen3 STT and TTS support - #2236
feat(baseten): add Qwen3 STT and TTS support#2236rosetta-livekit-bot[bot] wants to merge 1 commit into
Conversation
🦋 Changeset detectedLatest commit: b197a3e The changes in this PR will be included in the next version bump. This PR includes changesets to release 39 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| release(ws: WebSocket, config: Qwen3SessionConfig, discard: boolean): void { | ||
| this.#active.delete(ws); | ||
| if (discard || this.#closing || ws.readyState !== WebSocket.OPEN) { | ||
| shutdownWebSocket(ws, !discard); | ||
| return; | ||
| } | ||
|
|
||
| const stale = this.#warm; | ||
| this.#warm = { ws, config, lastActivity: Date.now() }; | ||
| if (stale && stale.ws !== ws) shutdownWebSocket(stale.ws, true); | ||
| this.#scheduleKeepalive(); | ||
| } |
There was a problem hiding this comment.
🔴 Idle voice connection kept for reuse can crash the whole agent process
The reused connection is parked with no failure handler attached (release at plugins/baseten/src/qwen3_tts.ts:292-303) after the only handler was detached, so a network drop while it sits idle takes the entire agent process down.
Impact: A transient network problem between two spoken turns can kill the agent process instead of just dropping one connection.
Unhandled 'error' event on the parked warm WebSocket
Flow: Qwen3SynthesizeStream.run() creates socketInbox(ws) (plugins/baseten/src/qwen3_tts.ts:435), which is the only place that registers ws.on('error', ...) after connectWebSocket removes its own handlers in cleanup() (plugins/baseten/src/qwen3_tts.ts:887-894). In the finally block the code calls inbox.close() — removing the message/close/error listeners (plugins/baseten/src/qwen3_tts.ts:744-749) — and then this.#backend.release(ws, config ?? {}, discard) (plugins/baseten/src/qwen3_tts.ts:621-624).
On the non-discard path release stores the socket in #warm without attaching any listener. The socket then sits idle for up to KEEPALIVE_INTERVAL (15 s) and beyond. ws emits 'error' for socket-level failures (ECONNRESET, TLS errors, etc.); Node's EventEmitter throws ERR_UNHANDLED_ERROR when 'error' is emitted with no listener, which is an uncaught exception.
Note that shutdownWebSocket deliberately attaches ws.on('error', () => {}) before tearing a socket down (plugins/baseten/src/qwen3_tts.ts:851-852), so the discard path is protected but the park path is not. A narrower version of the same gap exists in plugins/baseten/src/qwen3_stt.ts: after #receiveEvents calls cleanup() (removing ws.off('error', onError)) the run() finally still calls ws.close() with no error listener attached.
Prompt for agents
In plugins/baseten/src/qwen3_tts.ts, sockets returned by connectWebSocket end up with zero 'error' listeners once socketInbox is closed (Qwen3SynthesizeStream.run's finally calls inbox.close() before Qwen3Backend.release). When release parks the socket in #warm, an idle socket-level failure emits 'error' with no listener, which is an uncaught ERR_UNHANDLED_ERROR that crashes the process. Ensure every socket owned by the backend always has at least one 'error' listener for its whole lifetime — for example attach a permanent no-op/logging error handler right after connectWebSocket resolves in acquire() (and make socketInbox's handler additive rather than the only one), instead of relying on shutdownWebSocket to attach one at teardown. The same lifetime gap should be checked for the qwen3_stt.ts socket, whose error listener is removed by #receiveEvents cleanup before run()'s finally closes the socket.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const metadata = { | ||
| vad_params: { | ||
| threshold: this.#opts.vadThreshold, | ||
| min_silence_duration_ms: this.#opts.vadMinSilenceDurationMs, | ||
| speech_pad_ms: this.#opts.vadSpeechPadMs, | ||
| whisper_params: { | ||
| audio_language: this.#opts.audioLanguage ?? 'en', | ||
| show_word_timestamps: this.#opts.showWordTimestamps ?? true, | ||
| ...(this.#opts.languageOptions?.length | ||
| ? { language_options: this.#opts.languageOptions } | ||
| : {}), | ||
| }, | ||
| streaming_whisper_params: { | ||
| streaming_params: { | ||
| encoding: this.#opts.encoding ?? 'pcm_s16le', | ||
| sample_rate: this.#opts.sampleRate ?? 16000, | ||
| enable_partial_transcripts: false, | ||
| audio_language: this.#opts.audioLanguage ?? 'en', | ||
| show_word_timestamps: true, | ||
| enable_partial_transcripts: this.#opts.enablePartialTranscripts ?? true, | ||
| partial_transcript_interval_s: this.#opts.partialTranscriptIntervalS ?? 1, | ||
| final_transcript_max_duration_s: this.#opts.finalTranscriptMaxDurationS ?? 30, | ||
| }, | ||
| streaming_vad_config: { | ||
| threshold: this.#opts.vadThreshold, | ||
| min_silence_duration_ms: this.#opts.vadMinSilenceDurationMs, | ||
| speech_pad_ms: this.#opts.vadSpeechPadMs, | ||
| }, |
There was a problem hiding this comment.
🔴 Speech-to-text settings are sent under field names a previous fix had corrected, so the server may ignore them
The transcription setup message is sent using field names (whisper_params/streaming_params/streaming_vad_config at plugins/baseten/src/stt.ts:273-292) that an earlier fix had explicitly replaced because the server expects different ones, so language, word timings, partial results and silence tuning can all be silently ignored.
Impact: Whisper transcription may fall back to server defaults, losing the configured language, interim results and word-level timings.
Revert of a documented server-compatibility fix
Commit 989f16a ("Add aligned transcript support with word-level timing") changed the handshake from streaming_vad_config/streaming_params/whisper_params to vad_params/streaming_whisper_params and added the comment:
"Note: Baseten server expects 'vad_params' and 'streaming_whisper_params' field names (not 'streaming_vad_config', 'streaming_params', 'whisper_params' as in older versions)"
This PR deletes that comment and restores the older names. If the deployed Whisper runtime still expects the newer names, every option in the handshake (audio_language, show_word_timestamps, language_options, enable_partial_transcripts, partial_transcript_interval_s, final_transcript_max_duration_s, VAD thresholds) is dropped, while the plugin still advertises alignedTranscript: 'word' and interimResults: true to the framework.
If the server genuinely accepts both shapes, this should be verified against a live deployment and the removed comment updated accordingly.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const modelEndpoint = | ||
| opts.modelEndpoint ?? | ||
| (modelId | ||
| ? `wss://model-${modelId}.api.baseten.co/environments/production/websocket` | ||
| : undefined) ?? | ||
| (opts.chainId | ||
| ? `wss://chain-${opts.chainId}.api.baseten.co/environments/production/websocket` | ||
| : undefined) ?? | ||
| (process.env.BASETEN_STT_MODEL_ID | ||
| ? `wss://model-${process.env.BASETEN_STT_MODEL_ID}.api.baseten.co/environments/production/websocket` | ||
| : undefined) ?? | ||
| process.env.BASETEN_MODEL_ENDPOINT; |
There was a problem hiding this comment.
🟡 Custom deployment environment setting is silently ignored for speech-to-text
The connection address is now always built with the hard-coded "production" environment (plugins/baseten/src/stt.ts:71-82), so a user who selects a different deployment environment is connected to the wrong one without any warning.
Impact: Users pointing the plugin at a staging/development deployment silently talk to their production deployment instead.
getWsUrl no longer reaches the environment-aware fallback
Previously modelEndpoint was only set from opts.modelEndpoint/BASETEN_MODEL_ENDPOINT; when only modelId was given, SpeechStream.getWsUrl() (plugins/baseten/src/stt.ts:214-220) built the URL using this.#opts.environment.
Now the constructor always resolves modelEndpoint (throwing otherwise) using templates that hard-code /environments/production/, and additionally sets modelId: undefined. Since getWsUrl() returns this.#opts.modelEndpoint whenever it is set, the environment option (still declared in BasetenSttOptions and defaulted in defaultSTTOptions) is now dead configuration. The same hard-coding exists in plugins/baseten/src/endpoint.ts:6-9 for the Qwen3 path.
Prompt for agents
plugins/baseten/src/stt.ts now always constructs the WebSocket URL with a hard-coded '/environments/production/' segment and clears modelId, which makes SpeechStream.getWsUrl()'s environment-aware fallback unreachable and turns the documented `environment` option into dead configuration. Either thread `opts.environment` (default 'production') into the URL templates used in the constructor, or remove the `environment` option from BasetenSttOptions/defaultSTTOptions and the now-dead getWsUrl fallback so the API no longer advertises a setting it ignores.
Was this helpful? React with 👍 or 👎 to provide feedback.
| protected synthesizeWithStream( | ||
| text: string, | ||
| connOptions: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, | ||
| abortSignal?: AbortSignal, | ||
| ): ChunkedStream { | ||
| return new ChunkedStreamFromStream(text, this, connOptions, abortSignal); | ||
| } |
There was a problem hiding this comment.
🟡 One-shot speech synthesis reports its usage twice
Each one-shot synthesis performed through the streaming helper (synthesizeWithStream at agents/src/tts/tts.ts:156-162) reports its usage once from the inner streaming session and once from the outer one-shot wrapper, so usage numbers are counted twice.
Impact: Character counts, audio duration and token usage for these requests are double-counted in metrics and usage dashboards.
Both SynthesizeStream.monitorMetrics and ChunkedStream.monitorMetrics emit on the same TTS instance
ChunkedStreamFromStream.run() (agents/src/tts/tts.ts:862-882) drives a real SynthesizeStream obtained from this.#tts.stream(...) and calls stream.pushText(...). pushText starts SynthesizeStream.monitorMetrics() (agents/src/tts/tts.ts:530-534), which emits a tts_metrics event (streamed: true) on this.#tts when the segment completes.
The outer ChunkedStream.monitorMetrics() (agents/src/tts/tts.ts:758-802) then emits a second tts_metrics event (streamed: false) on the same TTS instance for the same text and the same audio frames.
The framework treats this pattern as a defect elsewhere: FallbackAdapter in agents/src/stt/fallback_adapter.ts:248-258 and its stream override explicitly skip the base emit precisely to avoid double counting.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const stream = this.makeStream(owner, options); | ||
| const onAbort = () => stream.close(); | ||
| options.abortSignal?.addEventListener('abort', onAbort, { once: true }); | ||
|
|
||
| const texts: string[] = []; | ||
| let language = languageCode(); | ||
| try { | ||
| stream.pushFrame(mergeFrames(buffer)); | ||
| stream.flush(); | ||
| stream.endInput(); | ||
| for await (const event of stream) { | ||
| if (event.type === stt.SpeechEventType.FINAL_TRANSCRIPT && event.alternatives?.length) { | ||
| texts.push(event.alternatives[0].text); | ||
| language = event.alternatives[0].language || language; | ||
| } | ||
| } | ||
| if (stream.error) throw stream.error; |
There was a problem hiding this comment.
🟡 Retried one-shot transcription returns an empty result instead of a transcript
A one-shot transcription request that is retried after a connection failure re-runs with its audio already consumed (recognizeViaStream at plugins/baseten/src/qwen3_stt.ts:189-205), so the caller receives an empty transcript instead of an error or the real text.
Impact: A transient network hiccup turns a spoken utterance into silence with no error reported to the caller.
Input queue is drained by the first attempt while SpeechStream.mainTask retries run()
recognizeViaStream pushes the whole buffer, flushes and calls stream.endInput() once (plugins/baseten/src/qwen3_stt.ts:196-198). The stream is created with options.connOptions ?? DEFAULT_API_CONNECT_OPTIONS (plugins/baseten/src/qwen3_stt.ts:170), i.e. maxRetry: 3.
SpeechStream.mainTask retries run() on retryable APIErrors (agents/src/stt/stt.ts:350-400). On the second attempt #sendAudio finds this.input already closed and drained, so it immediately breaks out of its loop and just sends input_audio_buffer.commit with no audio (plugins/baseten/src/qwen3_stt.ts:322-347). If the server answers with a final/end-of-flush event, run() returns successfully, stream.error is undefined, and recognizeViaStream returns an alternative with text: ''; if the server sends nothing for an empty commit, the call never completes.
A one-shot recognition should either buffer the audio for replay across attempts or run with maxRetry: 0 and let the caller retry.
Prompt for agents
plugins/baseten/src/qwen3_stt.ts: Qwen3Backend.recognizeViaStream pushes the audio buffer into a Qwen3SpeechStream exactly once and closes the input, but the stream is built with DEFAULT_API_CONNECT_OPTIONS (maxRetry: 3) so agents/src/stt/stt.ts SpeechStream.mainTask can re-invoke run(). On a retry the input queue is already drained/closed, so #sendAudio sends only an empty commit and the call either resolves with an empty transcript or hangs. Fix by making the one-shot path own retries itself — e.g. create the stream with maxRetry: 0 (mirroring the TTS ChunkedStreamFromStream helper, which forces maxRetry: 0 on the inner streaming attempt) and let the caller/outer layer retry with a fresh stream, or keep the audio buffered so each attempt can replay it.
Was this helpful? React with 👍 or 👎 to provide feedback.
| export interface RegisterVoiceOptions { | ||
| modelEndpoint: string; | ||
| name: string; | ||
| refAudioPath: string; | ||
| refText?: string; | ||
| apiKey?: string; | ||
| consent?: string; | ||
| timeoutMs?: number; | ||
| } | ||
|
|
||
| export async function registerVoice({ | ||
| modelEndpoint, | ||
| name, | ||
| refAudioPath, | ||
| refText, | ||
| apiKey, | ||
| consent = 'user_consent', | ||
| timeoutMs = DEFAULT_API_CONNECT_OPTIONS.timeoutMs, | ||
| }: RegisterVoiceOptions): Promise<Record<string, unknown>> { |
There was a problem hiding this comment.
🟡 New public helpers and option types ship without documentation comments
Several newly exported items are added with no documentation comments (for example RegisterVoiceOptions/registerVoice at plugins/baseten/src/qwen3_tts.ts:630-648), which CONTRIBUTING.md requires for every new interface, class and method.
Impact: The generated API documentation for the new Baseten Qwen3 surface is incomplete.
Repository rule and affected declarations
CONTRIBUTING.md: "If writing new methods/interfaces/enums/classes, document them. This project uses TypeDoc for automatic API documentation generation, and every new addition has to be properly documented."
Undocumented new exported declarations include RegisterVoiceOptions and registerVoice (plugins/baseten/src/qwen3_tts.ts:630-663), Qwen3TTSOptions and ResolvedQwen3TTSOptions (plugins/baseten/src/qwen3_tts.ts:37-69), ListVoicesOptions (plugins/baseten/src/qwen3_tts.ts:665-669), and Qwen3STTOptions (plugins/baseten/src/qwen3_stt.ts:29-42), all of which are re-exported from plugins/baseten/src/index.ts.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if (!endpoint.startsWith('wss://') && !endpoint.startsWith('ws://')) { | ||
| throw new Error( | ||
| `This model is served over WebSocket only; got ${JSON.stringify(endpoint)}. Endpoints look like wss://model-<id>.api.baseten.co/environments/production/websocket`, | ||
| ); | ||
| } | ||
|
|
||
| const hostname = endpoint.startsWith('ws://') | ||
| ? new URL(endpoint).hostname.replace(/^\[|\]$/g, '') | ||
| : undefined; | ||
| if (hostname !== undefined && !LOOPBACK_HOSTS.has(hostname)) { | ||
| log().warn( | ||
| { endpoint }, | ||
| 'endpoint is plaintext ws://: the Baseten API key and all audio will be sent unencrypted. Use wss:// for any non-local host.', | ||
| ); | ||
| } |
There was a problem hiding this comment.
🟨 Plaintext ws:// endpoints are only warned about, allowing the API key to be sent unencrypted
resolveEndpoint accepts ws:// endpoints pointing at arbitrary non-loopback hosts and merely logs a warning (plugins/baseten/src/endpoint.ts:36-44). The resolved endpoint is then used to open a WebSocket that sends Authorization: Api-Key <key> in the handshake headers (plugins/baseten/src/qwen3_stt.ts:480-482, plugins/baseten/src/qwen3_tts.ts:283-284), so a misconfigured or attacker-influenced endpoint value leaks the Baseten API key and all audio in cleartext.
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
STTandTTSclasses.languageOptions, Qwen3 voice registration/listing, word timestamps, warm TTS socket keepalives, and model-specific defaults.Port of livekit/agents#6700.
Validation
pnpm exec vitest run agents(1,585 passed, 5 skipped)pnpm exec vitest run plugins/baseten(3 credential-gated tests skipped)pnpm buildpnpm lintcue-cliruntime validation was unavailable because this environment does not have a complete LiveKit credential set.Source diff coverage
Authoritative livekit/agents PR #6700 file coverage
livekit-plugins/livekit-plugins-baseten/README.md->plugins/baseten/README.md. Preserved the Qwen3 and Whisper language-options documentation using TypeScript examples and camelCase options.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/__init__.py->plugins/baseten/src/index.ts. Exported voice helpers and model/option types through the JS package entry point.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/_endpoint.py->plugins/baseten/src/endpoint.ts. Preserved endpoint precedence, URL construction, WebSocket validation, and plaintext non-loopback warnings with Node APIs.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/models.py->plugins/baseten/src/types.ts. RepresentedSTTModelsandTTSModelsas the publicSTTModelandTTSModelunions alongside the package's existing option interfaces.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/qwen3_stt.py->plugins/baseten/src/qwen3_stt.ts. Ported the Qwen3-ASR WebSocket protocol, commit behavior, VAD events, language normalization, one-shot recognition, timing offsets, and word timestamps.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/qwen3_tts.py->plugins/baseten/src/qwen3_tts.ts. Ported session configuration, streaming PCM, flushes, timestamps, warm-socket lifecycle, keepalives, interruption cleanup, and voice management.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/stt.py->plugins/baseten/src/stt.ts, with supporting language propagation inagents/src/stt/. Preserved model selection, model-specific defaults, Whisperlanguage_options, runtime updates, and per-call language overrides using JS framework contracts.livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/tts.py->plugins/baseten/src/tts.ts, with the missing streaming-to-chunked helper inagents/src/tts/tts.ts. Preserved model selection, Qwen3 options, active-stream shutdown, timed transcripts, and single-layer retry behavior.Ported from livekit/agents#6700
Original PR description
Summary
Baseten hosts Qwen3-ASR Streaming and Qwen3-TTS alongside the Whisper and Orpheus models this plugin already supports. They speak different wire protocols, so the existing
STT/TTSclasses can't reach them — pointing either at a Qwen3 endpoint connects and then produces nothing.This adds
Qwen3STTandQwen3TTSas separate classes. The existing Orpheus/Whisper paths are untouched, so this is non-breaking.STT/TTSQwen3STT/Qwen3TTSinput_audio_buffer.appendmessage_type/transcripttype: "transcription"/segments[].text{prompt, voice, …}, or WS +__END__sentinelsession.config→input.text→input.donetara)Both accept
model_endpoint,model_id, orchain_idwith the same precedence asSTT(extracted into_endpoint.py).Notes on the design
A few protocol details drove decisions that aren't obvious from the diff:
input.doneis a flush, not a close. The session config stays in effect, soQwen3TTSkeeps one warm socket across turns. Re-dialing per utterance would add a connect plus a config round trip to every agent response.input.doneanswerssession.donewith zero sentences and proves the session is alive.session.closewould keep the GPU busy producing audio nobody hears.SynthesizeStream.push_text()after a flush is dropped by the framework and_main_taskraises on a segment-count mismatch, so a mid-stream flush means "synthesize what's buffered", never "start a new segment"."English"), soQwen3STTmaps the common ones to ISO codes rather than passing a name where a code is expected.Voices
Qwen3-TTS Base ships no built-in speakers — there's no
taraequivalent.voicenames a registered clone, andregister_voice/list_voicesare exported to manage them. Worth knowing: the server stores uploaded voices on the container's local disk, so a runtime-registered voice lives on one replica and is lost on restart. The README documents baking the reference into the deployment instead, or passingref_audio/ref_textto clone inline per session.Also:
language_optionsfor the existing WhisperSTTBundled here because it is the same plugin and came out of the same customer
conversation. Baseten's streaming transcription API has accepted a
language_optionslist since Whisper runtime v0.5.0, which scopes detection tothe languages an agent actually supports. The plugin only ever sent a single
audio_language, forcing a choice between a fixed tag that mistranscribes theother language and
auto, which detects across all 99 and is unreliable on theone- to two-second utterances typical of telephony.
Only added to the handshake when non-empty —
StreamingWhisperInputusesextra="forbid", so sending it unconditionally would break anyone on an olderruntime. Also wired through
update_optionson bothSTTandSpeechStream.Verified against a live Whisper Large V3 Turbo streaming deployment, with a
negative control:
language_options: ["en", "de"]is accepted and transcribesnormally, while a deliberately misspelled field name closes the socket with
1011 — so acceptance confirms the field name rather than showing it was
silently ignored.
Testing
Developed against
livekit-agents1.6.8 with a mock server implementing the Qwen3 protocols, driven through the real framework machinery (AudioEmitter,RecognizeStream, the retry loop) and through a realAgentSessionwith a real-time audio sink. Covered:push_text, socket reuse without config resend, keepalive on a parked socket, barge-in discarding the socket, transient error retried / persistent error propagating, word timestamps rebased across sentence boundariesstart_of_speech→ interim → final →end_of_speech, one-shotrecognize(), partials disabled viainterim_results=FalseAgentSession: TTS audio out with reuse across turns andinterrupt()mid-playout; STT mic audio in surfacing interim + final user turns over consecutive VAD-bounded turnsI'm happy to contribute those as pytest suites if you'd like them in-tree — I left them out to keep the diff focused and avoid adding scripts your CI would try to collect.
Live validation
Since opening this, both adapters have been run end to end against real Baseten
deployments of the same model-registry trusses they target (Qwen3-ASR streaming
on RTX Pro 6000, Qwen3-TTS Base on RTX Pro 6000), using real speech with ground
truth rather than synthetic audio.
STT — a mu-bench en-US utterance, streamed at 100ms frames:
100% word overlap, both through
Qwen3STT.stream()directly and through a realAgentSession(user_input_transcribed, 6 interims + 1 final). Confirms thehandshake, the base64 append frames,
type: "transcription"parsing, thelanguage_code: "English"->enmapping, and clean termination on thecommit-triggered final.
TTS —
voice.listreturns{"voices": [], ...}on the Base checkpoint,confirming it ships no built-in speakers;
voice.addcloning from a 14sreference works; synthesis returns real 24kHz PCM; the second turn reuses the
warm socket (TTFA 1081ms vs a cold first turn).
Round trip — feeding the live TTS output back into the live STT transcribes
at 100% word overlap, so the synthesized audio is genuinely intelligible speech
and not just well-formed bytes.
One operational note worth stating: on a cold replica the first synthesis
exceeded the 60s session timeout and was retried by the framework before
succeeding. That is cold-start behavior rather than an adapter issue, but
production voice agents should keep
min_replica >= 1.ruff checkandruff formatare clean.