feat(qwen): add Qwen STT, TTS and LLM plugin for Alibaba Cloud Model Studio - #7224
feat(qwen): add Qwen STT, TTS and LLM plugin for Alibaba Cloud Model Studio#7224satviksinha wants to merge 10 commits into
Conversation
|
|
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…ed utterances, bill usage on teardown Review follow-ups on livekit#7224. RecognizeStream reruns _run on a retryable error but keeps no input replay buffer, so a retry opened a fresh socket with no audio, finished cleanly and reported an empty transcript instead of the failure. The stream now tracks whether any frame has left the channel and clears the retry flag once it has, handing the provider's own error to the FallbackAdapter. A failed utterance left true and emitted no END_OF_SPEECH, which strands the user turn under turn_detection="stt". The stream now promotes the last interim text to a final and closes the speech pair. Cancellation skipped report_usage(), so audio streamed after the last final was never billed and a stream with no final reported nothing.
…ed utterances, bill usage on teardown Review follow-ups on livekit#7224. `RecognizeStream` reruns `_run` on a retryable error but keeps no input replay buffer, so a retry opened a fresh socket with no audio, finished cleanly and reported an empty transcript instead of the failure. The stream now tracks whether any frame has left the channel and clears the retry flag once it has, handing the provider's own error to the FallbackAdapter rather than wrapping it. A failed utterance left `speaking` true and emitted no END_OF_SPEECH, which strands the user turn under turn_detection="stt". The stream now promotes the last interim text to a final and closes the speech pair. Cancellation skipped `report_usage()`, so audio streamed after the last final was never billed and a stream with no final reported nothing at all.
032a7e9 to
705e824
Compare
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 4 new potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| except Exception as e: | ||
| raise APIConnectionError() from e |
There was a problem hiding this comment.
🟨 Connection errors retain credential-bearing causes
Non-handshake failures preserve e as __cause__. Request exceptions can retain the bearer header, exposing the API key through tracebacks.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Not changing this one. Two things I checked:
RequestInfo is the only aiohttp object that carries request headers, and it lives on ClientResponseError. That is the branch directly above, and it already re-raises from None, which is the fix from #6739 / #7031 and what tests/test_ws_handshake_credential_redaction.py asserts. What reaches the generic branch is ClientConnectorError, ClientOSError, DNS failures and InvalidURL. Those carry a host, a port or the URL, and the key is in a header, not the URL.
Dropping the cause here would also cost real diagnostics. APIConnectionError.__str__ deliberately walks __cause__ to name the root failure because its default message says nothing, and the other plugins keep the cause on this branch for that reason (57 raise APIConnectionError() from e sites across the STT and TTS modules). If there is an aiohttp exception on this path that does carry headers, point me at it and I will switch.
| session.ws_connect( | ||
| f"{base_url}?model={model}", | ||
| headers={"Authorization": f"Bearer {api_key}"}, | ||
| ), |
There was a problem hiding this comment.
🟨 Plaintext endpoints expose credentials and content
A ws:// base_url sends the bearer key and customer content without encryption. Production misconfiguration exposes credentials, speech, and synthesized text.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Fixed in e15f490. resolve_realtime_url now logs a warning for a plaintext ws:// URL whose host is not loopback, following the Baseten plugin's approach. It stays allowed, since local proxies and the test suite legitimately use ws://127.0.0.1. Three tests: remote plaintext is warned about, loopback is silent, and wss:// plus both region defaults are silent.
…failure, warn on plaintext ws:// Second round of review follow-ups on livekit#7224. A barge-in cancels the TTS stream before send() reaches session.finish, and Model Studio books that as a failed request. The cancellation path now sends the finish and closes without waiting for the reply: the voice pipeline awaits the stream's aclose() before it clears the playout buffer, so a bounded wait here would keep the agent talking over the user. STT usage is now reported from the common finally, so a provider failure bills the audio that was sent just like a clean finish or a cancellation does. A plaintext ws:// base_url to a non-loopback host is logged as a warning; the key travels in an Authorization header and the audio is unencrypted. Loopback stays silent for local proxies and tests.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 5 new potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| return APIStatusError( | ||
| message=message or "Model Studio returned an error event", | ||
| status_code=400 if error_type == "invalid_request_error" else -1, | ||
| request_id=event.get("event_id"), | ||
| body=error, |
There was a problem hiding this comment.
🟨 Provider payloads leak through exceptions
Error events place the full provider payload in APIStatusError.body. Downstream exception telemetry can record customer content without a PII marker.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Not changing this one. The body here is the provider's error object, {code, type, message} from the error event, not audio, a transcript or anything the user said, and the same message goes into APIStatusError.message exactly as in every other plugin (69 STT/TTS call sites in this repo pass a non-None body= with the provider payload). The place the framework handles exception text reaching telemetry is telemetry/pii.py, which redacts exception details centrally when redaction is enabled, so a per-plugin marker on the exception would not add coverage. If a maintainer would rather the plugins stop passing provider bodies at all, that is a repo-wide change and I am happy to follow it here.
…d text, single one-shot metric, bounded barge-in close Third round of review follow-ups on livekit#7224. The consumed-audio flag was set after a successful send, so a frame that left the channel but stalled in the chunker, or whose send failed, still allowed a retry that could not replay it. It is now set the moment a frame is read from the channel. A failed utterance promoted `text + stash` to a final. `stash` is the tail the model may still revise, so only the confirmed `text` is committed now; a failure with nothing confirmed closes the turn without a final. recognize() emitted two STTMetrics for one request: the base class's batch metric plus the inner stream's streamed usage. Manual-commit streams no longer report usage. close_with_finish() now runs under a 0.5 s budget shared by the finish write and the close handshake, always leaving the close a small floor. aiohttp's own close handshake would otherwise wait its default 10 s on a stalled peer, and this sits on the barge-in path.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 4 new potential issues.
3 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| if not self._event_ch.closed: | ||
| report_usage() |
There was a problem hiding this comment.
🟡 Interrupted STT usage metrics disappear
Cancellation queues RECOGNITION_USAGE, then RecognizeStream.aclose immediately cancels its metrics monitor. Interrupted streams can omit billed audio from metrics_collected.
Learn more
report_usage() writes an event to _event_ch; a separate framework task converts that event into STTMetrics. During aclose(), the Qwen _run task queues this final event while unwinding from cancellation. The framework then cancels _metrics_task instead of draining it, so scheduling determines whether the queued usage reaches telemetry. Forwarding the event to a caller does not guarantee that the independent metrics branch processed it.
Example: A live stream sends 0.2 seconds of audio and is interrupted before a final transcript. Qwen queues usage for 0.2 seconds, but immediate task cancellation can leave the metrics listener with no metrics_collected event.
Recommended fix: Ensure cancellation-time usage is emitted synchronously to the STT metrics interface, or change stream shutdown to drain the metrics tee after closing the event channel. Avoid producing both a direct metric and a usage event on paths where the monitor remains active.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Not changing this one in the plugin. You are right that it is a race: RecognizeStream.aclose() cancels _metrics_task instead of draining the tee, so a usage event queued during cancellation may or may not be turned into STTMetrics. I noted the same when adding the cancellation-time report, and kept it because it can only add a metric that would otherwise be lost and reported_samples rules out double counting.
The suggested alternative, emitting STTMetrics directly from the plugin on that path, means one plugin reconstructing the framework's metric shape (streamed, acquire_time, connection_reused, metadata) and diverging from every other STT plugin, with a double-count risk whenever the monitor does win the race. The fix that actually closes the gap is in RecognizeStream.aclose(): drain the metrics branch after closing the event channel rather than cancelling it. That is a framework change and benefits all streaming STTs, so I will raise it there rather than paper over it here.
…lose, query-safe realtime URL Fourth round of review follow-ups on livekit#7224. recognize() handed the inner SpeechStream the caller's retry budget while STT.recognize() applies the same budget around it, so a connect failure made up to 16 attempts on the defaults. The inner stream now runs with max_retry=0; the base class owns the retry and is the only place the whole buffer can be replayed. close_gracefully() bounded the finish handshake but not the aiohttp close that followed it, whose default lets a stalled peer add up to 10 s. Both teardown paths now share one bounded close with a small floor. connect() concatenated "?model=" onto base_url, which corrupted a gateway URL that already carried a query and never encoded the model id. The URL is built with yarl's update_query.
There was a problem hiding this comment.
Devin Review found 1 new potential issue.
3 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| stream = SpeechStream( | ||
| stt=self, | ||
| opts=opts, | ||
| conn_options=replace(conn_options, max_retry=0), | ||
| manual_commit=True, |
There was a problem hiding this comment.
🔴 Retries emit unrecoverable STT errors
When a recognize() attempt fails, SpeechStream emits an unrecoverable error because max_retry is zero. The outer retry can still succeed, while listeners count the transient failure toward session shutdown.
Learn more
RecognizeStream._main_task emits recoverable=False whenever its own max_retry is zero. The same exception then reaches STT.recognize, which owns the actual retry and emits another error with the correct recoverability. Agent sessions count each unrecoverable STT event and close after the configured tolerance, even when a later outer attempt returns a transcript.
Example: With max_retry=3, the first connection failure emits an unrecoverable inner error, then a recoverable outer error. If the second connection succeeds, recognition returns normally, but the session has already recorded an unrecoverable failure.
Recommended fix: Keep one retry owner without letting the helper stream emit independent STT errors. Add a way to suppress or delegate RecognizeStream error emission for this one-shot wrapper, so only STT.recognize() reports attempt recoverability.
Was this helpful? React with 👍 or 👎 to provide feedback.
2 participants
Summary
Adds
livekit-plugins-qwen: realtime STT and TTS for Qwen speech models on Alibaba Cloud Model Studio, plus an LLM helper for its OpenAI-compatible chat endpoint.Model Studio's OpenAI-compatible API only covers chat completions. There is no
/v1/audio/transcriptionsand no/v1/audio/speech, soopenai.STTandopenai.TTShave nothing to point at. Its realtime WebSocket borrows OpenAI's event names but not the schema. Session config is flat rather than nested, interim text arrives asconversation.item.input_audio_transcription.textinstead of.delta, and input must be 16 kHz where the OpenAI plugin hardcodes 24 kHz. Hence a separate plugin.STT (
qwen3-asr-flash-realtime)input_audio_buffer.appendevents.textplusstash, the tail the model may still revise.END_OF_SPEECHafter each final. Model Studio has no such event, and LiveKit's turn detection needs the pair closed.RECOGNITION_USAGEon every final. Model Studio bills ASR per second of audio and sends no usage of its own. It reports per final rather than once at the end, because a live stream ends throughaclose(), which cancels the task before any epilogue could run.recognize()over the same socket with an explicit commit, so theFallbackAdapterprobe works.TTS (
qwen3-tts-flash-realtime)server_commitmode, so the model starts speaking before the turn's text is complete.response.audio.deltainto one emitter segment per stream.session.finishinstead of clipping the tail.Shared transport
session.finishhandshake. A server that holds the socket open without answering raisesAPITimeoutError.APIConnectOptions.timeoutonly covers the connect, so nothing else catches this.aclose(). Model Studio counts a socket dropped without it as a failed request.invalid_request_errorto a non-retryableAPIStatusError. Retrying a request that cannot succeed only delays failover.APIConnectionErroron a bare close, which would otherwise read as a clean end of stream.from None, so theRequestInfoholding the API key cannot leak into a log.LLM and regions
qwen.LLMsubclassesopenai.LLMwith the region base URL andDASHSCOPE_API_KEY. It pinsenable_thinkingoff by default, because LiveKit's stream never readsdelta.reasoning_contentand a thinking turn plays as dead air.region="intl"selects Singapore and is the default;region="cn"selects Beijing.base_urloverrides both for workspace-dedicated domains. Model Studio keys are region-bound and the README says so.Also adds the workspace source and the
livekit-agents[qwen]extra.Relation to #4489
This supersedes #4489, which stalled on the rename to
qwenasked for in review. Thanks @rocky-terracotta for the groundwork. This one uses that name and adds tests, usage reporting, the finish handshake, and arecognize()that works.One choice worth flagging. The tests drive the plugin against an in-process aiohttp WebSocket server speaking Model Studio's documented events, and I marked the modules
unitso the CI unit gate runs them. They touch loopback only, take no credentials and make no outbound calls. The repo's otherTCPSiteprovider tests are markedplugin(name)though, so say the word and I will switch.Testing
uv run pytest tests/test_plugin_qwen_realtime.py tests/test_plugin_qwen_stt.py tests/test_plugin_qwen_tts.py tests/test_plugin_qwen_llm.py --unit, 81 passed.make check, clean across 671 source files.make unit-tests, 2560 passed and 5 skipped. The 1 failure and 9 errors there reproduce with the qwen tests excluded, so they do not come from this branch.tests/test_room.pyerrors withFileNotFoundError: 'livekit-server', andtest_google_credentials.py::TestSTTCredentials::test_clear_error_when_project_unresolvablefails because this host has Google Cloud SDK credentials, so the project resolves and the expected error never fires.recognize()ontests/change-sophie.wavreturned "The people who are crazy enough to think they can change the world are the ones who do." with languageen.synthesize()on a mixed English and Chinese sentence returned about 2.0 s of 24 kHz PCM.qwen.LLManswered onqwen-plus.