Use TurnAnalyzerUserTurnStopStrategy as default stop strategy

Change the default user turn stop strategy from
TranscriptionUserTurnStopStrategy to TurnAnalyzerUserTurnStopStrategy
with LocalSmartTurnAnalyzerV3. Also reduce AUDIO_INPUT_TIMEOUT_SECS
from 1.0 to 0.5 and remove its debug log.

changelog/3672.changed.md

@@ -0,0 +1 @@
+- Changed `FunctionCallCancelFrame` to broadcast in both directions for consistency with other function call frames.

changelog/3672.fixed.md

@@ -0,0 +1 @@
+- Fixed `RTVIObserver` sending duplicate client messages for frames that are broadcast in both directions (e.g. `UserStartedSpeakingFrame`, `FunctionCallResultFrame`).

src/pipecat/processors/frameworks/rtvi.py

@@ -1023,7 +1023,7 @@ class RTVIObserverParams:
             sets the default level for unlisted functions::
 
                 function_call_report_level={
-                    "*": RTVIFunctionCallReportLevel.DISABLED,  # Default: no events
+                    "*": RTVIFunctionCallReportLevel.NONE,  # Default: events with no metadata
                     "get_weather": RTVIFunctionCallReportLevel.FULL,  # Expose everything
                 }
 
@@ -1199,6 +1199,12 @@ class RTVIObserver(BaseObserver):
         frame = data.frame
         direction = data.direction
 
+        # Only process downstream frames. Some frames are broadcast in both
+        # directions (e.g. UserStartedSpeakingFrame, FunctionCallResultFrame),
+        # and we only want to send one RTVI message per event.
+        if direction != FrameDirection.DOWNSTREAM:
+            return
+
         # If we have already seen this frame, let's skip it.
         if frame.id in self._frames_seen:
             return

src/pipecat/services/llm_service.py

@@ -745,8 +745,9 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
                     await self.cancel_task(task)
                     cancelled_tasks.add(task)
 
-                frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id)
-                await self.push_frame(frame)
+                await self.broadcast_frame(
+                    FunctionCallCancelFrame, function_name=name, tool_call_id=tool_call_id
+                )
 
                 logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
 

src/pipecat/services/sarvam/tts.py

@@ -25,6 +25,15 @@ Indian languages and two model variants:
       kavya, amit, dev, ishita, shreya, ratan, varun, manan, sumit, roopa, kabir,
       aayan, shubh, ashutosh, advait, amelia, sophia
 
+- **bulbul:v3**: Advanced TTS model with temperature control
+    - Does NOT support: pitch, loudness
+    - Supports: pace (0.5-2.0), temperature (0.01-1.0)
+    - Default sample rate: 24000 Hz
+    - Preprocessing is always enabled
+    - Speakers: aditya (default), ritu, priya, neha, rahul, pooja, rohan, simran,
+      kavya, amit, dev, ishita, shreya, ratan, varun, manan, sumit, roopa, kabir,
+      aayan, shubh, ashutosh, advait, amelia, sophia
+
 See https://docs.sarvam.ai/api-reference-docs/text-to-speech/stream for full API details.
 """
 
@@ -83,6 +92,7 @@ class SarvamTTSModel(str, Enum):
 
     BULBUL_V2 = "bulbul:v2"
     BULBUL_V3_BETA = "bulbul:v3-beta"
+    BULBUL_V3 = "bulbul:v3"
 
 
 class SarvamTTSSpeakerV2(str, Enum):
@@ -180,6 +190,16 @@ TTS_MODEL_CONFIGS: Dict[str, TTSModelConfig] = {
         preprocessing_always_enabled=True,
         speakers=tuple(s.value for s in SarvamTTSSpeakerV3),
     ),
+    "bulbul:v3": TTSModelConfig(
+        supports_pitch=False,
+        supports_loudness=False,
+        supports_temperature=True,
+        default_sample_rate=24000,
+        default_speaker="shubh",
+        pace_range=(0.5, 2.0),
+        preprocessing_always_enabled=True,
+        speakers=tuple(s.value for s in SarvamTTSSpeakerV3),
+    ),
 }
 
 

src/pipecat/transports/base_input.py

@@ -48,7 +48,7 @@ from pipecat.metrics.metrics import MetricsData
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.transports.base_transport import TransportParams
 
-AUDIO_INPUT_TIMEOUT_SECS = 1.0
+AUDIO_INPUT_TIMEOUT_SECS = 0.5
 
 
 class BaseInputTransport(FrameProcessor):
@@ -449,8 +449,6 @@ class BaseInputTransport(FrameProcessor):
                 if not audio_received:
                     continue
 
-                logger.debug(f"{self}: audio not received for more than {AUDIO_INPUT_TIMEOUT_SECS}")
-
                 ###################################################################
                 # DEPRECATED.
                 if self._user_speaking:

src/pipecat/turns/user_turn_strategies.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from typing import List, Optional
 
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
 from pipecat.turns.user_start import (
     BaseUserTurnStartStrategy,
     ExternalUserTurnStartStrategy,
@@ -18,7 +19,7 @@ from pipecat.turns.user_start import (
 from pipecat.turns.user_stop import (
     BaseUserTurnStopStrategy,
     ExternalUserTurnStopStrategy,
-    TranscriptionUserTurnStopStrategy,
+    TurnAnalyzerUserTurnStopStrategy,
 )
 
 
@@ -29,7 +30,7 @@ class UserTurnStrategies:
     If no strategies are specified, the following defaults are used:
 
         start: [VADUserTurnStartStrategy, TranscriptionUserTurnStartStrategy]
-         stop: [TranscriptionUserTurnStopStrategy]
+         stop: [TurnAnalyzerUserTurnStopStrategy(LocalSmartTurnAnalyzerV3)]
 
     Attributes:
         start: A list of user turn start strategies used to detect when
@@ -46,7 +47,7 @@ class UserTurnStrategies:
         if not self.start:
             self.start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
         if not self.stop:
-            self.stop = [TranscriptionUserTurnStopStrategy()]
+            self.stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
 
 
 @dataclass