diff --git a/src/pipecat/services/assemblyai/models.py b/src/pipecat/services/assemblyai/models.py index ca58cb848..fb883ac99 100644 --- a/src/pipecat/services/assemblyai/models.py +++ b/src/pipecat/services/assemblyai/models.py @@ -12,7 +12,7 @@ transcription WebSocket messages and connection configuration. from typing import List, Literal, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field class Word(BaseModel): @@ -68,8 +68,16 @@ class TurnMessage(BaseMessage): transcript: The transcribed text for this turn. end_of_turn_confidence: Confidence score for end-of-turn detection. words: List of individual words with timing and confidence data. + language_code: Detected language code (e.g., "es", "fr"). Only present with + complete utterances or when end_of_turn is True. + language_confidence: Confidence score (0-1) for language detection. Only present + with complete utterances or when end_of_turn is True. + speaker: Speaker label (e.g., "A", "B"). Only present when speaker_labels is + enabled and end_of_turn is True. Maps to 'speaker_label' in JSON response. """ + model_config = ConfigDict(populate_by_name=True) + type: Literal["Turn"] = "Turn" turn_order: int turn_is_formatted: bool @@ -77,6 +85,21 @@ class TurnMessage(BaseMessage): transcript: str end_of_turn_confidence: float words: List[Word] + language_code: Optional[str] = None + language_confidence: Optional[float] = None + speaker: Optional[str] = Field(default=None, alias="speaker_label") + + +class SpeechStartedMessage(BaseMessage): + """Message sent when speech is first detected in the audio stream. + + Parameters: + type: Always "SpeechStarted" for this message type. + timestamp: Audio timestamp in milliseconds when speech was detected. + """ + + type: Literal["SpeechStarted"] = "SpeechStarted" + timestamp: int class TerminationMessage(BaseMessage): @@ -94,7 +117,7 @@ class TerminationMessage(BaseMessage): # Union type for all possible message types -AnyMessage = BeginMessage | TurnMessage | TerminationMessage +AnyMessage = BeginMessage | TurnMessage | SpeechStartedMessage | TerminationMessage class AssemblyAIConnectionParams(BaseModel): @@ -109,7 +132,15 @@ class AssemblyAIConnectionParams(BaseModel): min_end_of_turn_silence_when_confident: Minimum silence duration when confident about end-of-turn. max_turn_silence: Maximum silence duration before forcing end-of-turn. keyterms_prompt: List of key terms to guide transcription. Will be JSON serialized before sending. - speech_model: Select between English and multilingual models. Defaults to "universal-streaming-english". + prompt: Optional text prompt to guide the transcription. Only used when speech_model is "u3-rt-pro". + speech_model: Select between English, multilingual, and u3-rt-pro models. Defaults to "u3-rt-pro". + language_detection: Enable automatic language detection. Only applicable to + universal-streaming-multilingual. When enabled, Turn messages include + language_code and language_confidence fields. Defaults to None (not sent). + format_turns: Whether to format transcript turns. Defaults to True. + speaker_labels: Enable speaker diarization. When enabled, final transcripts + (end_of_turn=True) include a speaker field identifying the speaker + (e.g., "Speaker A", "Speaker B"). Defaults to None (not sent). """ sample_rate: int = 16000 @@ -120,6 +151,10 @@ class AssemblyAIConnectionParams(BaseModel): min_end_of_turn_silence_when_confident: Optional[int] = None max_turn_silence: Optional[int] = None keyterms_prompt: Optional[List[str]] = None - speech_model: Literal["universal-streaming-english", "universal-streaming-multilingual"] = ( - "universal-streaming-english" + prompt: Optional[str] = None + speech_model: Literal["universal-streaming-english", "universal-streaming-multilingual", "u3-rt-pro"] = ( + "u3-rt-pro" ) + language_detection: Optional[bool] = None + format_turns: bool = True + speaker_labels: Optional[bool] = None diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index a89f5fe52..edd5ac7b7 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -13,7 +13,7 @@ WebSocket API for streaming audio transcription. import asyncio import json from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, Dict, Optional +from typing import Any, AsyncGenerator, Dict, Mapping, Optional from urllib.parse import urlencode from loguru import logger @@ -41,6 +41,7 @@ from .models import ( AssemblyAIConnectionParams, BaseMessage, BeginMessage, + SpeechStartedMessage, TerminationMessage, TurnMessage, ) @@ -54,6 +55,26 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +def map_language_from_assemblyai(language_code: str) -> Language: + """Map AssemblyAI language codes to Pipecat Language enum. + + AssemblyAI returns simple language codes like "es", "fr", etc. + This function maps them to the corresponding Language enum values. + + Args: + language_code: AssemblyAI language code (e.g., "es", "fr", "de") + + Returns: + Corresponding Language enum value, defaulting to Language.EN if not found. + """ + try: + # Try to match the language code directly + return Language(language_code.lower()) + except ValueError: + logger.warning(f"Unknown language code from AssemblyAI: {language_code}, defaulting to English") + return Language.EN + + @dataclass class AssemblyAISTTSettings(STTSettings): """Settings for the AssemblyAI STT service. @@ -87,6 +108,8 @@ class AssemblyAISTTService(WebsocketSTTService): api_endpoint_base_url: str = "wss://streaming.assemblyai.com/v3/ws", connection_params: AssemblyAIConnectionParams = AssemblyAIConnectionParams(), vad_force_turn_endpoint: bool = True, + should_interrupt: bool = True, + speaker_format: Optional[str] = None, ttfs_p99_latency: Optional[float] = ASSEMBLYAI_TTFS_P99, **kwargs, ): @@ -97,18 +120,64 @@ class AssemblyAISTTService(WebsocketSTTService): language: Language code for transcription. Defaults to English (Language.EN). api_endpoint_base_url: WebSocket endpoint URL. Defaults to AssemblyAI's streaming endpoint. connection_params: Connection configuration parameters. Defaults to AssemblyAIConnectionParams(). - vad_force_turn_endpoint: Whether to force turn endpoint on VAD stop. When True, - disables AssemblyAI's model-based turn detection and relies on external VAD - to trigger turn endpoints. Automatically sets end_of_turn_confidence_threshold=1.0 - and max_turn_silence=2000 unless explicitly overridden. Defaults to True. + vad_force_turn_endpoint: Controls turn detection mode. + When True (Pipecat mode, default): Forces AssemblyAI to return finals ASAP + so Pipecat's turn detection (e.g., Smart Turn) decides when the user is done. + - min_end_of_turn_silence_when_confident defaults to 100ms (user can override) + - max_turn_silence is ALWAYS set equal to min_end_of_turn_silence_when_confident + - VAD stop sends ForceEndpoint as ceiling + - No UserStarted/StoppedSpeakingFrame emitted from STT + When False (STT mode, u3-rt-pro only): AssemblyAI's model controls turn endings. + - Uses AssemblyAI API defaults for all parameters (unless user explicitly sets them) + - Respects all user-provided connection_params as-is + - Emits UserStarted/StoppedSpeakingFrame from STT + - No ForceEndpoint on VAD stop + should_interrupt: Whether to interrupt the bot when the user starts speaking + in STT mode (vad_force_turn_endpoint=False). Only applies to STT mode. + Defaults to True. + speaker_format: Optional format string for speaker labels when diarization is enabled. + Use {speaker} for speaker label and {text} for transcript text. + Example: "<{speaker}>{text}" or "{speaker}: {text}" + If None, transcript text is not modified. Defaults to None. ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark **kwargs: Additional arguments passed to parent STTService class. """ - # When vad_force_turn_endpoint is enabled, configure connection params for manual - # turn detection mode (disable model-based turn detection) + # STT turn detection (vad_force_turn_endpoint=False) requires the + # SpeechStarted event for reliable barge-in. Only u3-rt-pro supports + # this. Other models must use Pipecat turn detection. + is_u3_pro = connection_params.speech_model == "u3-rt-pro" + if not vad_force_turn_endpoint and not is_u3_pro: + raise ValueError( + f"STT turn detection (vad_force_turn_endpoint=False) requires " + f"u3-rt-pro for SpeechStarted support. Either set " + f"vad_force_turn_endpoint=True for {connection_params.speech_model}, " + f"or use speech_model='u3-rt-pro'." + ) + + # Validate that prompt and keyterms_prompt are not both set + if connection_params.prompt is not None and connection_params.keyterms_prompt is not None: + raise ValueError( + "The prompt and keyterms_prompt parameters cannot be used in the same request. " + "Please choose either one or the other based on your use case. When you use " + "keyterms_prompt, your boosted words are appended to the default prompt automatically. " + "Or to boost within prompt: + Make sure to boost the words in the audio. " + "For more info go to: https://www.assemblyai.com/docs/streaming/universal-3-pro" + ) + + # Warn if user sets a custom prompt (recommend testing without one first) + if connection_params.prompt is not None: + logger.warning( + "Custom prompt detected. We recommend testing with no prompt first, as this " + "will use our optimized default prompt for voice agents. Bad prompts may lead " + "to bad results. If you'd like to create your own prompt, check out our " + "prompting guide at: https://www.assemblyai.com/docs/streaming/prompting" + ) + + # When vad_force_turn_endpoint is enabled, configure connection params + # for Pipecat turn detection mode (fast finals for smart turn analyzer) if vad_force_turn_endpoint: - connection_params = self._configure_manual_turn_mode(connection_params) + connection_params = self._configure_pipecat_turn_mode(connection_params, is_u3_pro) super().__init__( sample_rate=connection_params.sample_rate, @@ -124,6 +193,8 @@ class AssemblyAISTTService(WebsocketSTTService): self._api_key = api_key self._api_endpoint_base_url = api_endpoint_base_url self._vad_force_turn_endpoint = vad_force_turn_endpoint + self._should_interrupt = should_interrupt + self._speaker_format = speaker_format self._termination_event = asyncio.Event() self._received_termination = False @@ -135,45 +206,77 @@ class AssemblyAISTTService(WebsocketSTTService): self._chunk_size_ms = 50 self._chunk_size_bytes = 0 - def _configure_manual_turn_mode( - self, connection_params: AssemblyAIConnectionParams - ) -> AssemblyAIConnectionParams: - """Configure connection params for manual turn detection mode. + self._user_speaking = False + self._vad_speaking = False - When vad_force_turn_endpoint is enabled, we want to disable AssemblyAI's - model-based turn detection and rely on external VAD. This requires: - - end_of_turn_confidence_threshold=1.0 (disable semantic turn detection) - - max_turn_silence=2000 (high value since VAD handles turn endings) + # Log final connection params after any modifications + logger.info(f"{self} Final connection params being sent to AssemblyAI:") + logger.info(f" min_end_of_turn_silence_when_confident: {self._settings.connection_params.min_end_of_turn_silence_when_confident}") + logger.info(f" max_turn_silence: {self._settings.connection_params.max_turn_silence}") + + # Warn if min_end_of_turn_silence_when_confident is not 100ms + if self._settings.connection_params.min_end_of_turn_silence_when_confident != 100: + logger.warning( + f"For best latency, set min_end_of_turn_silence_when_confident to 100ms. " + f"Current value: {self._settings.connection_params.min_end_of_turn_silence_when_confident}ms" + ) + + def _configure_pipecat_turn_mode( + self, connection_params: AssemblyAIConnectionParams, is_u3_pro: bool + ) -> AssemblyAIConnectionParams: + """Configure connection params for Pipecat turn detection mode. + + When vad_force_turn_endpoint is enabled, force AssemblyAI to return + finals as fast as possible so Pipecat's smart turn analyzer can decide + when the user is done speaking. VAD stop is the absolute ceiling. + + u3-rt-pro: + - min_end_of_turn_silence_when_confident defaults to 100ms (user can override) + - max_turn_silence is ALWAYS set equal to min_end_of_turn_silence_when_confident + to avoid double turn detection (AssemblyAI + Pipecat both analyzing) + - If user sets max_turn_silence, it's ignored with a warning + - end_of_turn_confidence_threshold: not set (API default) + + universal-streaming-*: + - end_of_turn_confidence_threshold=0.0 (disable semantic turn detection) + - min_end_of_turn_silence_when_confident=160 + - max_turn_silence: not set (API default) Args: connection_params: The user-provided connection parameters. + is_u3_pro: Whether using u3-rt-pro model. Returns: - Updated connection parameters configured for manual turn mode. + Updated connection parameters configured for Pipecat turn mode. """ updates = {} - # Check end_of_turn_confidence_threshold - if connection_params.end_of_turn_confidence_threshold is None: - updates["end_of_turn_confidence_threshold"] = 1.0 - elif connection_params.end_of_turn_confidence_threshold != 1.0: - logger.warning( - f"vad_force_turn_endpoint is enabled but end_of_turn_confidence_threshold " - f"is set to {connection_params.end_of_turn_confidence_threshold}. " - f"For manual turn detection mode, this should be 1.0 to disable " - f"model-based turn detection. The current value will be used." - ) + if is_u3_pro: + # u3-rt-pro: Synchronize max_turn_silence with min_end_of_turn_silence_when_confident + min_silence = connection_params.min_end_of_turn_silence_when_confident + if min_silence is None: + min_silence = 100 - # Check max_turn_silence - if connection_params.max_turn_silence is None: - updates["max_turn_silence"] = 2000 - elif connection_params.max_turn_silence < 1000: - logger.warning( - f"vad_force_turn_endpoint is enabled but max_turn_silence is set to " - f"{connection_params.max_turn_silence}ms. With manual turn detection, " - f"a higher value (e.g., 2000ms) is recommended to avoid premature " - f"turn endings. The current value will be used." - ) + # Warn if user set max_turn_silence (will be overridden) + if connection_params.max_turn_silence is not None: + logger.warning( + f"Your max_turn_silence value ({connection_params.max_turn_silence}ms) will be " + f"OVERRIDDEN in Pipecat mode (vad_force_turn_endpoint=True). It will be set to " + f"{min_silence}ms (matching min_end_of_turn_silence_when_confident) and SENT to " + f"AssemblyAI to avoid double turn detection. To use your max_turn_silence as-is, " + f"switch to STT mode (vad_force_turn_endpoint=False)." + ) + + updates = { + "min_end_of_turn_silence_when_confident": min_silence, + "max_turn_silence": min_silence, + } + else: + # universal-streaming: Different configuration (works differently) + updates = { + "end_of_turn_confidence_threshold": 0.0, + "min_end_of_turn_silence_when_confident": 160, + } # Apply updates if any if updates: @@ -190,9 +293,14 @@ class AssemblyAISTTService(WebsocketSTTService): return True async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: - """Apply a settings delta. + """Apply a settings delta and send UpdateConfiguration if connected. - Settings are stored but not applied to the active connection. + Stores settings changes and sends UpdateConfiguration message to AssemblyAI + without reconnecting. Supports updating: + - keyterms_prompt: List of terms to boost (can be empty array to clear) + - prompt: Custom prompt text (u3-rt-pro only) + - max_turn_silence: Maximum silence before forcing turn end + - min_end_of_turn_silence_when_confident: Silence before EOT check Args: delta: A :class:`STTSettings` (or ``AssemblyAISTTSettings``) delta. @@ -205,18 +313,63 @@ class AssemblyAISTTService(WebsocketSTTService): if not changed: return changed - # TODO: someday we could reconnect here to apply updated settings. - # Code might look something like the below: - # # Re-apply manual turn mode config if vad_force_turn_endpoint is active - # # and connection_params were updated. - # if self._vad_force_turn_endpoint and "connection_params" in changed: - # self._settings.connection_params = self._configure_manual_turn_mode( - # self._settings.connection_params - # ) - # await self._disconnect() - # await self._connect() + # If websocket is connected, send UpdateConfiguration for supported params + if self._websocket and self._websocket.state is State.OPEN and "connection_params" in changed: + # Build UpdateConfiguration message + update_config = {"type": "UpdateConfiguration"} + conn_params = self._settings.connection_params - self._warn_unhandled_updated_settings(changed) + # Get the old connection_params to see what changed + old_conn_params = changed.get("connection_params") + + # Check each potentially changed parameter + if hasattr(conn_params, "keyterms_prompt"): + if old_conn_params is None or conn_params.keyterms_prompt != old_conn_params.keyterms_prompt: + if conn_params.keyterms_prompt is not None: + update_config["keyterms_prompt"] = conn_params.keyterms_prompt + logger.info(f"Updating keyterms_prompt to: {conn_params.keyterms_prompt}") + + if hasattr(conn_params, "prompt"): + if old_conn_params is None or conn_params.prompt != old_conn_params.prompt: + if conn_params.prompt is not None: + if conn_params.speech_model != "u3-rt-pro": + logger.warning( + f"prompt parameter is only supported with u3-rt-pro model, " + f"current model is {conn_params.speech_model}" + ) + else: + update_config["prompt"] = conn_params.prompt + logger.info(f"Updating prompt") + + if hasattr(conn_params, "max_turn_silence"): + if old_conn_params is None or conn_params.max_turn_silence != old_conn_params.max_turn_silence: + if conn_params.max_turn_silence is not None: + update_config["max_turn_silence"] = conn_params.max_turn_silence + logger.info(f"Updating max_turn_silence to: {conn_params.max_turn_silence}ms") + + if hasattr(conn_params, "min_end_of_turn_silence_when_confident"): + if old_conn_params is None or conn_params.min_end_of_turn_silence_when_confident != old_conn_params.min_end_of_turn_silence_when_confident: + if conn_params.min_end_of_turn_silence_when_confident is not None: + update_config["min_end_of_turn_silence_when_confident"] = conn_params.min_end_of_turn_silence_when_confident + logger.info(f"Updating min_end_of_turn_silence_when_confident to: {conn_params.min_end_of_turn_silence_when_confident}ms") + + # Send update if we have parameters to update + if len(update_config) > 1: # More than just "type" + try: + await self._websocket.send(json.dumps(update_config)) + logger.info(f"Sent UpdateConfiguration: {update_config}") + except Exception as e: + logger.error(f"Failed to send UpdateConfiguration: {e}") + elif "connection_params" in changed: + logger.warning( + "Connection params changed but WebSocket not connected. " + "Settings will be applied on next connection." + ) + + # Warn about other settings that can't be changed dynamically + other_changes = {k: v for k, v in changed.items() if k not in ["connection_params"]} + if other_changes: + self._warn_unhandled_updated_settings(other_changes) return changed @@ -305,7 +458,13 @@ class AssemblyAISTTService(WebsocketSTTService): if params: query_string = urlencode(params) - return f"{self._api_endpoint_base_url}?{query_string}" + full_url = f"{self._api_endpoint_base_url}?{query_string}" + logger.info(f"{self} WebSocket URL being sent to AssemblyAI:") + logger.info(f" {full_url}") + logger.info(f" Parsed params:") + for k, v in params.items(): + logger.info(f" {k}: {v}") + return full_url return self._api_endpoint_base_url async def _connect(self): @@ -421,6 +580,9 @@ class AssemblyAISTTService(WebsocketSTTService): async for message in self._get_websocket(): try: data = json.loads(message) + # Log raw JSON for Turn messages to debug speaker_label + if data.get("type") == "Turn": + logger.debug(f"{self} RAW JSON from AssemblyAI: {json.dumps(data, indent=2)}") await self._handle_message(data) except json.JSONDecodeError: logger.warning(f"Received non-JSON message: {message}") @@ -433,6 +595,8 @@ class AssemblyAISTTService(WebsocketSTTService): return BeginMessage.model_validate(message) elif msg_type == "Turn": return TurnMessage.model_validate(message) + elif msg_type == "SpeechStarted": + return SpeechStartedMessage.model_validate(message) elif msg_type == "Termination": return TerminationMessage.model_validate(message) else: @@ -449,11 +613,37 @@ class AssemblyAISTTService(WebsocketSTTService): ) elif isinstance(parsed_message, TurnMessage): await self._handle_transcription(parsed_message) + elif isinstance(parsed_message, SpeechStartedMessage): + await self._handle_speech_started(parsed_message) elif isinstance(parsed_message, TerminationMessage): await self._handle_termination(parsed_message) except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) + async def _handle_speech_started(self, message: SpeechStartedMessage): + """Handle SpeechStarted event — fast barge-in for Mode 2. + + Broadcasts UserStartedSpeakingFrame to signal the start of user + speech, then pushes an interruption to cancel any bot audio. + SpeechStarted fires before any transcript arrives, so the turn + is cleanly started before any transcription frames are pushed. + + Only applies to Mode 2 (STT turn detection). In Mode 1, VAD + + smart turn analyzer handle interruptions via the aggregator. + """ + logger.debug(f"{self} SpeechStarted received (vad_force_turn_endpoint={self._vad_force_turn_endpoint})") + if self._vad_force_turn_endpoint: + logger.debug(f"{self} SpeechStarted ignored in Pipecat mode") + return # Mode 1: handled by aggregator + + logger.debug(f"{self} Processing SpeechStarted in STT mode") + await self.start_processing_metrics() + await self.broadcast_frame(UserStartedSpeakingFrame) + if self._should_interrupt: + await self.push_interruption_task_frame_and_wait() + self._user_speaking = True + logger.debug(f"{self} _user_speaking set to True") + async def _handle_termination(self, message: TerminationMessage): """Handle termination message.""" self._received_termination = True @@ -466,30 +656,135 @@ class AssemblyAISTTService(WebsocketSTTService): await self.push_frame(EndFrame()) async def _handle_transcription(self, message: TurnMessage): - """Handle transcription results.""" + """Handle transcription results with two-mode turn detection. + + Mode 1 (vad_force_turn_endpoint=True, Pipecat turn detection): + - No UserStarted/StoppedSpeakingFrame from STT + - end_of_turn → TranscriptionFrame (finalized set by base class + if this is a ForceEndpoint response) + - else → InterimTranscriptionFrame + + Mode 2 (vad_force_turn_endpoint=False, STT turn detection): + - UserStartedSpeakingFrame on first transcript + - end_of_turn → TranscriptionFrame + UserStoppedSpeakingFrame + - else → InterimTranscriptionFrame + """ + # Log transcript details + logger.info(f"{self} ===== TRANSCRIPT RECEIVED =====") + logger.info(f" Text: \"{message.transcript}\"") + logger.info(f" end_of_turn: {message.end_of_turn}") + logger.info(f" turn_is_formatted: {message.turn_is_formatted}") + logger.info(f" turn_order: {message.turn_order}") + if message.end_of_turn_confidence is not None: + logger.info(f" end_of_turn_confidence: {message.end_of_turn_confidence}") + logger.info(f" speaker: {message.speaker}") + logger.info(f"===============================") + if not message.transcript: return - if message.end_of_turn and ( - not self._settings.connection_params.formatted_finals or message.turn_is_formatted - ): - await self.push_frame( - TranscriptionFrame( - message.transcript, - self._user_id, - time_now_iso8601(), - self._settings.language, - message, + + # Use detected language if available with sufficient confidence + language = Language.EN + if message.language_code and message.language_confidence: + if message.language_confidence >= 0.7: + language = map_language_from_assemblyai(message.language_code) + else: + logger.warning( + f"Low language detection confidence ({message.language_confidence:.2f}) " + f"for language '{message.language_code}', falling back to English" ) - ) - await self._trace_transcription(message.transcript, True, self._settings.language) - await self.stop_processing_metrics() + + # Handle speaker diarization + speaker_id = self._user_id + transcript_text = message.transcript + + if message.speaker: + speaker_id = message.speaker + # Format transcript with speaker labels if format string provided + if self._speaker_format: + transcript_text = self._speaker_format.format( + speaker=message.speaker, + text=message.transcript + ) + logger.info(f"{self} 🤖 TEXT SENT TO LLM (with speaker format): \"{transcript_text}\"") + else: + logger.info(f"{self} 🤖 TEXT SENT TO LLM (speaker {message.speaker}): \"{transcript_text}\"") else: - await self.push_frame( - InterimTranscriptionFrame( - message.transcript, - self._user_id, - time_now_iso8601(), - self._settings.language, - message, + logger.info(f"{self} 🤖 TEXT SENT TO LLM: \"{transcript_text}\"") + + # Determine if this is a final turn from AssemblyAI + is_final_turn = message.end_of_turn and ( + not self._settings.connection_params.format_turns or message.turn_is_formatted + ) + + if self._vad_force_turn_endpoint: + # --- Mode 1: Pipecat turn detection --- + # No UserStarted/StoppedSpeakingFrame — VAD + smart turn analyzer handle this + if is_final_turn: + finalize_confirmed = bool(message.turn_is_formatted) + if finalize_confirmed: + self.confirm_finalize() + logger.debug(f"{self} Final transcript: \"{transcript_text}\"") + await self.push_frame( + TranscriptionFrame( + transcript_text, + speaker_id, + time_now_iso8601(), + language, + message, + ) + ) + await self._trace_transcription(transcript_text, True, language) + await self.stop_processing_metrics() + else: + logger.debug(f"{self} Interim transcript: \"{transcript_text}\"") + await self.push_frame( + InterimTranscriptionFrame( + transcript_text, + speaker_id, + time_now_iso8601(), + language, + message, + ) + ) + else: + # --- Mode 2: STT turn detection --- + # SpeechStarted handles UserStartedSpeakingFrame + interruption. + # If SpeechStarted hasn't fired yet (shouldn't happen, but guard), + # broadcast here as fallback. + logger.debug(f"{self} Transcript received in STT mode (_user_speaking={self._user_speaking})") + if not self._user_speaking: + logger.warning(f"{self} Transcript arrived before SpeechStarted, broadcasting fallback UserStartedSpeakingFrame") + await self.broadcast_frame(UserStartedSpeakingFrame) + self._user_speaking = True + + if is_final_turn: + if message.turn_is_formatted: + self.confirm_finalize() + await self.push_frame( + TranscriptionFrame( + transcript_text, + speaker_id, + time_now_iso8601(), + language, + message, + finalized=True, + ) + ) + await self._trace_transcription(transcript_text, True, language) + await self.stop_processing_metrics() + # AAI is authoritative — emit UserStoppedSpeakingFrame immediately. + # broadcast_frame pushes downstream (same queue as TranscriptionFrame + # above, so ordering is preserved) and upstream. + await self.broadcast_frame(UserStoppedSpeakingFrame) + self._user_speaking = False + else: + await self.push_frame( + InterimTranscriptionFrame( + transcript_text, + speaker_id, + time_now_iso8601(), + language, + message, + ) ) - )