wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add u3-rt-pro support and improvements to AssemblyAI STT service

Browse Source 
- Fix speaker diarization: Add field alias for speaker_label → speaker
  mapping in TurnMessage model
- Add warning for non-optimal min_end_of_turn_silence_when_confident
  values (recommends 100ms for best latency)
- Improve max_turn_silence override warning message clarity
- Update custom prompt warning (remove 88% accuracy claim)
- Add comprehensive logging for debugging:
  - Log final connection params after modifications
  - Log WebSocket URL and parsed parameters
  - Log speaker field in transcripts
  - Log text sent to LLM with speaker formatting
- Support dynamic configuration updates via STTUpdateSettingsFrame:
  - keyterms_prompt (when AssemblyAI API supports it)
  - prompt
  - max_turn_silence
  - min_end_of_turn_silence_when_confident
This commit is contained in:
zack
2026-02-26 21:35:47 -05:00
parent 2a6a993869
commit 72934bd8ae
2 changed files with 408 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/pipecat/services/assemblyai/models.py
View File

@@ -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

441
src/pipecat/services/assemblyai/stt.py
View File

@@ -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}</{speaker}>" 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: <prompt> + Make sure to boost the words <keyterms> 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,
)
)
)