Steak fallback in llm - #7274
jatinsm2023 wants to merge 17 commits into
Conversation
That single set_property(...LanguageIdMode, "Continuous") line is what's triggering the "S1 Speech to Text Enhanced Feature - Audio" SKU on your Azure invoice. Continuous LID is one of those premium capabilities. The previous code (Code A) never set this property, so Azure defaulted to AtStart mode — language is detected once at the beginning of an utterance and reused. AtStart LID is not billed as an Enhanced Feature, only Continuous is. So every audio second you process now gets billed twice: once on Standard, once on Enhanced. "Cost for calls that weren't picked up": In _run(), start_continuous_recognition() begins as soon as the SpeechStream is constructed — before the user actually answers. With Continuous LID active, the recognizer is constantly running language-ID inference on whatever audio frames the agent's pipeline pushes in (ringback, hold music, room tone, the agent's own greeting prompt, etc.), and every one of those seconds bills against both meters. Under the previous AtStart behavior, the enhanced meter never fired, so unanswered calls produced negligible cost and you didn't notice. Now they're conspicuous.
Openai type Request Options logs
|
Ubuntu seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account. You have signed the CLA already but the status is still pending? Let us recheck it. |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
There was a problem hiding this comment.
Devin Review found 10 potential issues.
3 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| # stt = self._stt | ||
| # self.update_stt(None) | ||
| # self.update_stt(stt) |
There was a problem hiding this comment.
🔴 Cleared audio returns in later turns
clear_user_turn() leaves the provider STT stream intact. Delayed final transcripts can repopulate the next turn and trigger replies to discarded speech.
Learn more
A user-turn clear must discard both text already accumulated locally and audio still buffered by the active STT provider. The removed reset detached the existing pipeline and created a fresh stream. Without it, the old provider can emit a final transcript after the local fields are cleared. The final-transcript handler then appends that text to the newly empty accumulator in the final transcript path.
Example: The user says “cancel that,” and the application immediately clears the turn. A delayed provider result for “cancel that” then arrives and becomes the first transcript of the next turn, although the application discarded it.
Recommended fix: Restore the STT reset in clear_user_turn() or provide an equivalent provider-level buffer reset that guarantees old results cannot enter the new turn.
| # stt = self._stt | |
| # self.update_stt(None) | |
| # self.update_stt(stt) | |
| stt = self._stt | |
| self.update_stt(None) | |
| self.update_stt(stt) |
Was this helpful? React with 👍 or 👎 to provide feedback.
| drain_task = asyncio.create_task(self._drain_input_while_paused()) | ||
| try: | ||
| await self._resume_event.wait() | ||
| finally: | ||
| drain_task.cancel() | ||
| with contextlib.suppress(asyncio.CancelledError, Exception): | ||
| await drain_task |
There was a problem hiding this comment.
🔴 Paused Soniox streams never finish
When input ends while paused, _drain_input_while_paused consumes the closed channel while _wait_until_resumed waits forever. Stream shutdown then hangs without a later resume.
Learn more
A paused stream runs two waiters: one waits for _resume_event, while the drain task consumes _input_ch. Closing input ends only the drain task. Nothing wakes the resume waiter, so _run() never returns and the base stream's event channel never closes.
Example: Create Soniox with start_paused=True, start a stream, then call end_input() without resuming. The input channel closes, but iteration over transcription events waits forever instead of ending.
Recommended fix: Wait for either resume or drain completion. If the drain completes because _input_ch closed, return from _run() without opening a WebSocket; if resume wins, cancel the drain before accepting post-resume audio.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for i in self._ordered_indices()[1:]: | ||
| if self._status[i].available: | ||
| self._current_llm_index = i | ||
| logger.info( | ||
| f"Auto switch LLM from " | ||
| f"{self._llm_instances[idx].label} " | ||
| f"to {self._llm_instances[i].label}" | ||
| ) | ||
| return |
There was a problem hiding this comment.
🟡 Late failures displace healthy providers
When an older request fails after the pointer moves, _mark_failed excludes the current provider. LLM, STT, and TTS adapters rotate again unnecessarily.
Learn more
The adapter-wide current pointer is shared by concurrent requests. _mark_failed(idx) computes candidates from the pointer's latest value rather than from idx. Its slice omits that current value, so a failure from an older request can replace a healthy provider selected by another request. The same implementation exists in the STT adapter and the TTS adapter.
Example: Request A starts on provider 0. Another failure moves the current pointer to healthy provider 1. When request A later reports provider 0 failed, _mark_failed(0) starts after provider 1 and moves the pointer to provider 2.
Recommended fix: Always mark idx unavailable, but advance the pointer only when idx == _current_*_index or the current provider is unavailable. Apply the same synchronization rule to all three adapters.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if send_task.done() and not recv_task.done(): | ||
| with contextlib.suppress(asyncio.TimeoutError): | ||
| await asyncio.wait_for(asyncio.shield(recv_task), timeout=1.0) |
There was a problem hiding this comment.
🟡 Slow Gnani finals disappear silently
After input ends, SpeechStream waits one second for the final transcript and suppresses timeout. Slower final results are discarded as successful completion.
Learn more
When the sender finishes, the receiver remains responsible for delivering any final server transcript. This branch gives it one second, suppresses expiration, then cancels the receiver and closes the socket in the surrounding finally. The caller therefore cannot distinguish a completed empty transcription from a dropped late result.
Example: Audio input ends at 12:00:00 and Gnani returns its final transcript at 12:00:01.2. The receiver is canceled at 12:00:01, and consumers see a normal end with no transcript.
Recommended fix: Complete the provider's end-of-input protocol and wait for its terminal response. Bound that wait with self._conn_options.timeout, and raise APITimeoutError rather than treating expiration as success.
Was this helpful? React with 👍 or 👎 to provide feedback.
| # speech_config.set_property( | ||
| # speechsdk.PropertyId.SpeechServiceConnection_LanguageIdMode, "Continuous" | ||
| # ) |
There was a problem hiding this comment.
🟡 Azure language switching stops midstream
With multiple languages configured, continuous identification is disabled. Azure cannot update the detected language when speech changes languages later in the stream.
| # speech_config.set_property( | |
| # speechsdk.PropertyId.SpeechServiceConnection_LanguageIdMode, "Continuous" | |
| # ) | |
| speech_config.set_property( | |
| speechsdk.PropertyId.SpeechServiceConnection_LanguageIdMode, "Continuous" | |
| ) |
Was this helpful? React with 👍 or 👎 to provide feedback.
| try: | ||
| async with websockets.connect( | ||
| ws_url, | ||
| additional_headers=headers, | ||
| ping_interval=20, | ||
| ping_timeout=20, | ||
| close_timeout=10, | ||
| ) as ws: |
There was a problem hiding this comment.
🟡 Gnani WebSockets ignore request deadlines
Gnani WebSocket paths ignore APIConnectOptions.timeout. Connection and response waits can exceed caller deadlines, delaying failures and provider fallback.
Learn more
SpeechStream receives connection options but applies only a fixed ten-second timeout to the first message. The WebSocket connect and subsequent receive loop use independent library defaults. Both TTS WebSocket implementations repeat this behavior in batch synthesis and streaming synthesis.
Example: A fallback adapter sets an attempt timeout of five seconds. Gnani connects but never sends a response, so the attempt remains active past five seconds and the fallback provider does not start on schedule.
Recommended fix: Apply conn_options.timeout to connection establishment and the full provider exchange in all three WebSocket paths. Convert expiration to APITimeoutError so base retries and fallback adapters retain their normal behavior.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for offset in range(1, n + 1): | ||
| i = (start + offset) % n | ||
| status = self._status[i] | ||
|
|
||
| if not only_if_available or status.available: | ||
| prev = self._current_llm_index | ||
| self._current_llm_index = i | ||
|
|
||
| logger.info( | ||
| f"Manual switch LLM from " | ||
| f"{self._llm_instances[prev].label} " | ||
| f"to {self._llm_instances[i].label}" | ||
| ) | ||
|
|
||
| return True |
There was a problem hiding this comment.
🟡 No-op provider switches report success
switch_to_next eventually reselects the starting provider. It returns True when no alternative is available, although the pointer never changes.
Learn more
The loop includes offset == n, which wraps i back to start. If the current provider is available, that final iteration always succeeds even though no movement occurred. The STT implementation has the same contract violation in its manual switch.
Example: An adapter has one provider, or provider 0 is current while every other provider is unavailable. Calling switch_to_next() logs a switch from provider 0 to itself and returns True; callers cannot detect that failover was impossible.
Recommended fix: Iterate offsets 1..n-1 only. Return False when no distinct eligible provider exists.
Was this helpful? React with 👍 or 👎 to provide feedback.
| "json_data": { | ||
| "model": model, | ||
| "contents": [content.model_dump(exclude_none=True) for content in contents], | ||
| "config": config.model_dump(exclude_none=True), | ||
| }, |
There was a problem hiding this comment.
🟨 Gemini debug logs expose conversations
DEBUG logging serializes complete Gemini requests. User messages, instructions, and tool arguments can enter retained or exported application logs.
Was this helpful? React with 👍 or 👎 to provide feedback.
| retry_interval=self._retry_interval, | ||
| ), | ||
| ) | ||
| logger.info(f"[{req_id}] STT Response: Received response from {stt.label}: {stt_result}") |
There was a problem hiding this comment.
🟨 STT logs expose user transcripts
INFO logging records complete recognition results and streaming events. User speech transcripts can enter normal production logs without an explicit opt-in.
Was this helpful? React with 👍 or 👎 to provide feedback.
| language: GnaniSTTLanguages | str = "en-IN", | ||
| api_key: str | None = None, | ||
| sample_rate: int = SAMPLE_RATE_16K, | ||
| base_url: str = GNANI_STT_BASE_URL, |
There was a problem hiding this comment.
🟥 Custom Gnani endpoints receive API keys
A caller-controlled base_url receives the Gnani API key. An attacker-controlled endpoint can capture credentials from REST and WebSocket requests.
Was this helpful? React with 👍 or 👎 to provide feedback.
4 participants
No description provided.