fix: add exception handling for FAL smart turn detection

CHANGELOG.md

@@ -9,8 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Updated `SmallWebRTCConnection` to support `ice_servers` with credentials.
-
 - Added `VADUserStartedSpeakingFrame` and `VADUserStoppedSpeakingFrame`,
   indicating when the VAD detected the user to start and stop speaking. These
   events are helpful when using smart turn detection, as the user's stop time
@@ -60,9 +58,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Deprecated
 
-- Function calls with parameters
-  `(function_name, tool_call_id, args, llm, context, result_callback)` are
-  deprectated, use a single `FunctionCallParams` parameter instead.
+- Function calls with parameters `(function_name, tool_call_id, args, llm,
+  context, result_callback)` are deprectated, use a single `FunctionCallParams`
+  parameter instead.
 
 - `TransportParams.camera_*` parameters are now deprecated, use
   `TransportParams.video_*` instead.
@@ -75,10 +73,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- Fixed an issue with HTTP Smart Turn handling, where the service returns a 500
-  error. Previously, this would cause an unhandled exception. Now, a 500 error
-  is treated as an incomplete response.
-
 - Fixed a TTS services issue that could cause assistant output not to be
   aggregated to the context when also using `TTSSpeakFrame`s.
 

examples/deployment/modal-example/requirements.txt

@@ -1,4 +1,5 @@
 python-dotenv==1.0.1
 modal==0.71.3
-pipecat-ai[daily,silero,cartesia,openai]
+pipecat-ai[daily,silero,cartesia,openai]==0.0.52
 fastapi==0.115.6
+aiohttp==3.11.11

examples/foundational/17-detect-user-idle.py

@@ -6,30 +6,16 @@
 
 import argparse
 import os
-from dataclasses import dataclass
 
 from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import (
-    BotSpeakingFrame,
-    BotStartedSpeakingFrame,
-    BotStoppedSpeakingFrame,
-    EndFrame,
-    Frame,
-    LLMMessagesFrame,
-    StartInterruptionFrame,
-    TTSSpeakFrame,
-    UserStartedSpeakingFrame,
-    UserStoppedSpeakingFrame,
-)
-from pipecat.observers.base_observer import BaseObserver
+from pipecat.frames.frames import EndFrame, LLMMessagesFrame, TTSSpeakFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.user_idle_processor import UserIdleProcessor
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
@@ -41,84 +27,6 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
 load_dotenv(override=True)
 
 
-class DebugObserver(BaseObserver):
-    """Observer to log interruptions and bot speaking events to the console.
-
-    Logs all frame instances of:
-    - StartInterruptionFrame
-    - BotStartedSpeakingFrame
-    - BotStoppedSpeakingFrame
-
-    This allows you to see the frame flow from processor to processor through the pipeline for these frames.
-    Log format: [EVENT TYPE]: [source processor] → [destination processor] at [timestamp]s
-    """
-
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        # Convert timestamp to seconds for readability
-        time_sec = timestamp / 1_000_000_000
-
-        # Create direction arrow
-        arrow = "→" if direction == FrameDirection.DOWNSTREAM else "←"
-
-        if isinstance(frame, StartInterruptionFrame):
-            logger.info(f"⚡ INTERRUPTION START: {src} {arrow} {dst} at {time_sec:.2f}s")
-        elif isinstance(frame, BotStartedSpeakingFrame):
-            logger.info(f"🤖 BOT START SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s")
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            logger.info(f"🤖 BOT STOP SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s")
-
-
-@dataclass
-class StartConversationFrame(Frame):
-    """Frame to initiate a conversation.
-
-    This frame is used to signal the start of a conversation in the pipeline.
-    It can be used to trigger specific actions or responses from the system.
-    """
-
-    pass
-
-
-class ConversationStarterProcessor(FrameProcessor):
-    def __init__(self, message: str = "Hi! I'm a default message!"):
-        super().__init__()
-        self.message = message
-        self._user_stopped_speaking_count = 0
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
-        """Say a default message when the user starts speaking.
-
-        This processor listens for the UserStartedSpeakingFrame and sends a default message
-        when the user starts speaking for the first time.
-
-        Args:
-            frame: The frame to process
-            direction: Direction of the frame flow
-        """
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, (StartConversationFrame, UserStoppedSpeakingFrame)):
-            self._user_stopped_speaking_count += 1
-            logger.info(
-                f"++ {frame.name} User stopped speaking, count: {self._user_stopped_speaking_count}"
-            )
-            if self._user_stopped_speaking_count == 1:
-                # First time user started speaking, send the message
-                await self.push_frame(TTSSpeakFrame(self.message))
-            else:
-                await self.push_frame(frame)
-        else:
-            # Pass through other frames
-            await self.push_frame(frame)
-
-
 async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
     logger.info(f"Starting bot")
 
@@ -151,10 +59,15 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
     context_aggregator = llm.create_context_aggregator(context)
 
     async def handle_user_idle(user_idle: UserIdleProcessor, retry_count: int) -> bool:
-        logger.info(f"User idle, timeout : {user_idle._timeout} retry count: {retry_count}")
         if retry_count == 1:
-            # First attempt: Trigger the conversation starter
-            await user_idle.push_frame(StartConversationFrame())
+            # First attempt: Add a gentle prompt to the conversation
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "The user has been quiet. Politely and briefly ask if they're still there.",
+                }
+            )
+            await user_idle.push_frame(LLMMessagesFrame(messages))
             return True
         elif retry_count == 2:
             # Second attempt: More direct prompt
@@ -171,19 +84,16 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
             await user_idle.push_frame(
                 TTSSpeakFrame("It seems like you're busy right now. Have a nice day!")
             )
-            await user_idle.push_frame(EndFrame(), FrameDirection.UPSTREAM)
+            await task.queue_frame(EndFrame())
             return False
 
-    user_idle = UserIdleProcessor(callback=handle_user_idle, timeout=4.0)
-
-    conversation_starter = ConversationStarterProcessor(message="This is a default message.")
+    user_idle = UserIdleProcessor(callback=handle_user_idle, timeout=5.0)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             stt,
             user_idle,  # Idle user check-in
-            conversation_starter,
             context_aggregator.user(),
             llm,  # LLM
             tts,  # TTS
@@ -198,15 +108,15 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
             allow_interruptions=True,
             enable_metrics=True,
             report_only_initial_ttfb=True,
-            observers=[DebugObserver()],
         ),
     )
 
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
-        # Start the user idle timer
-        await task.queue_frames([BotSpeakingFrame()])
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
 
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):

examples/foundational/run.py

@@ -20,7 +20,7 @@ from fastapi.responses import RedirectResponse
 from loguru import logger
 from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
 
-from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
 
 # Load environment variables
 load_dotenv(override=True)
@@ -30,11 +30,7 @@ app = FastAPI()
 # Store connections by pc_id
 pcs_map: Dict[str, SmallWebRTCConnection] = {}
 
-ice_servers = [
-    IceServer(
-        urls="stun:stun.l.google.com:19302",
-    )
-]
+ice_servers = ["stun:stun.l.google.com:19302"]
 
 # Mount the frontend at /
 app.mount("/client", SmallWebRTCPrebuiltUI)

examples/p2p-webrtc/daily-interop-bridge/server.py

@@ -18,7 +18,7 @@ from fastapi.responses import RedirectResponse
 from loguru import logger
 from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
 
-from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
 
 # Load environment variables
 load_dotenv(override=True)
@@ -28,11 +28,7 @@ app = FastAPI()
 # Store connections by pc_id
 pcs_map: Dict[str, SmallWebRTCConnection] = {}
 
-ice_servers = [
-    IceServer(
-        urls="stun:stun.l.google.com:19302",
-    )
-]
+ice_servers = ["stun:stun.l.google.com:19302"]
 
 # Mount the frontend at /
 app.mount("/prebuilt", SmallWebRTCPrebuiltUI)

examples/p2p-webrtc/video-transform/server/server.py

@@ -18,7 +18,7 @@ from fastapi.responses import RedirectResponse
 from loguru import logger
 from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
 
-from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
 
 # Load environment variables
 load_dotenv(override=True)
@@ -28,11 +28,7 @@ app = FastAPI()
 # Store connections by pc_id
 pcs_map: Dict[str, SmallWebRTCConnection] = {}
 
-ice_servers = [
-    IceServer(
-        urls="stun:stun.l.google.com:19302",
-    )
-]
+ice_servers = ["stun:stun.l.google.com:19302"]
 
 # Mount the frontend at /
 app.mount("/prebuilt", SmallWebRTCPrebuiltUI)

examples/p2p-webrtc/voice-agent/README.md

@@ -46,46 +46,6 @@ http://localhost:7860
 
 ---
 
-## WebRTC ICE Servers Configuration
-
-When implementing WebRTC in your project, **STUN** (Session Traversal Utilities for NAT) and **TURN** (Traversal Using Relays around NAT) 
-servers are usually needed in cases where users are behind routers or firewalls.
-
-In local networks (e.g., testing within the same home or office network), you usually don’t need to configure STUN or TURN servers. 
-In such cases, WebRTC can often directly establish peer-to-peer connections without needing to traverse NAT or firewalls.
-
-### What are STUN and TURN Servers?
-
-- **STUN Server**: Helps clients discover their public IP address and port when they're behind a NAT (Network Address Translation) device (like a router). 
-This allows WebRTC to attempt direct peer-to-peer communication by providing the public-facing IP and port.
-  
-- **TURN Server**: Used as a fallback when direct peer-to-peer communication isn't possible due to strict NATs or firewalls blocking connections. 
-The TURN server relays media traffic between peers.
-
-### Why are ICE Servers Important?
-
-**ICE (Interactive Connectivity Establishment)** is a framework used by WebRTC to handle network traversal and NAT issues. 
-The `iceServers` configuration provides a list of **STUN** and **TURN** servers that WebRTC uses to find the best way to connect two peers. 
-
-### Example Configuration for ICE Servers
-
-Here’s how you can configure a basic `iceServers` object in WebRTC for testing purposes, using Google's public STUN server:
-
-```javascript
-const config = {
-  iceServers: [
-    {
-      urls: ["stun:stun.l.google.com:19302"], // Google's public STUN server
-    }
-  ],
-};
-```
-
-> For testing purposes, you can either use public **STUN** servers (like Google's) or set up your own **TURN** server. 
-If you're running your own TURN server, make sure to include your server URL, username, and credential in the configuration.
-
----
-
 ### 💡 Notes
 - Ensure all dependencies are installed before running the server.
 - Check the `.env` file for missing configurations.

examples/p2p-webrtc/voice-agent/index.html

@@ -24,47 +24,27 @@
         let connected = false
         let peerConnection = null
 
-        const waitForIceGatheringComplete = async (pc, timeoutMs = 2000) => {
+        /*const waitForIceGatheringComplete = async (pc) => {
             if (pc.iceGatheringState === 'complete') return;
-            console.log("Waiting for ICE gathering to complete. Current state:", pc.iceGatheringState);
             return new Promise((resolve) => {
-                let timeoutId;
                 const checkState = () => {
-                    console.log("icegatheringstatechange:", pc.iceGatheringState);
                     if (pc.iceGatheringState === 'complete') {
-                        cleanup();
+                        pc.removeEventListener('icegatheringstatechange', checkState);
                         resolve();
                     }
                 };
-                const onTimeout = () => {
-                    console.warn(`ICE gathering timed out after ${timeoutMs} ms.`);
-                    cleanup();
-                    resolve();
-                };
-                const cleanup = () => {
-                    pc.removeEventListener('icegatheringstatechange', checkState);
-                    clearTimeout(timeoutId);
-                };
                 pc.addEventListener('icegatheringstatechange', checkState);
-                timeoutId = setTimeout(onTimeout, timeoutMs);
-                // Checking the state again to avoid any eventual race condition
-                checkState();
             });
-        };
-
+        }*/
 
         const createSmallWebRTCConnection = async (audioTrack) => {
-            const config = {
-              iceServers: [],
-            };
-            const pc = new RTCPeerConnection(config)
-            addPeerConnectionEventListeners(pc)
+            const pc = new RTCPeerConnection()
             pc.ontrack = e => audioEl.srcObject = e.streams[0]
             // SmallWebRTCTransport expects to receive both transceivers
             pc.addTransceiver(audioTrack, { direction: 'sendrecv' })
             pc.addTransceiver('video', { direction: 'sendrecv' })
             await pc.setLocalDescription(await pc.createOffer())
-            await waitForIceGatheringComplete(pc)
+            //await waitForIceGatheringComplete(pc)
             const offer = pc.localDescription
             const response = await fetch('/api/offer', {
                 body: JSON.stringify({ sdp: offer.sdp, type: offer.type}),
@@ -77,37 +57,16 @@
         }
 
         const connect = async () => {
-            _onConnecting()
             const audioStream = await navigator.mediaDevices.getUserMedia({audio: true})
             peerConnection= await createSmallWebRTCConnection(audioStream.getAudioTracks()[0])
-        }
-
-        const addPeerConnectionEventListeners = (pc) => {
-            pc.oniceconnectionstatechange = () => {
-                console.log("oniceconnectionstatechange", pc?.iceConnectionState)
-            }
-            pc.onconnectionstatechange = () => {
-                console.log("onconnectionstatechange", pc?.connectionState)
-                let connectionState = pc?.connectionState
+            peerConnection.onconnectionstatechange = () => {
+                let connectionState = peerConnection?.connectionState
                 if (connectionState === 'connected') {
                     _onConnected()
                 } else if (connectionState === 'disconnected') {
                     _onDisconnected()
                 }
             }
-            pc.onicecandidate = (event) => {
-                if (event.candidate) {
-                    console.log("New ICE candidate:", event.candidate);
-                } else {
-                    console.log("All ICE candidates have been sent.");
-                }
-            };
-        }
-
-        const _onConnecting = () => {
-            statusEl.textContent = "Connecting"
-            buttonEl.textContent = "Disconnect"
-            connected = true
         }
 
         const _onConnected = () => {

examples/p2p-webrtc/voice-agent/server.py

@@ -17,7 +17,7 @@ from fastapi import BackgroundTasks, FastAPI
 from fastapi.responses import FileResponse
 from loguru import logger
 
-from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
 
 # Load environment variables
 load_dotenv(override=True)
@@ -28,13 +28,6 @@ app = FastAPI()
 pcs_map: Dict[str, SmallWebRTCConnection] = {}
 
 
-ice_servers = [
-    IceServer(
-        urls="stun:stun.l.google.com:19302",
-    )
-]
-
-
 @app.post("/api/offer")
 async def offer(request: dict, background_tasks: BackgroundTasks):
     pc_id = request.get("pc_id")
@@ -44,7 +37,7 @@ async def offer(request: dict, background_tasks: BackgroundTasks):
         logger.info(f"Reusing existing connection for pc_id: {pc_id}")
         await pipecat_connection.renegotiate(sdp=request["sdp"], type=request["type"])
     else:
-        pipecat_connection = SmallWebRTCConnection(ice_servers)
+        pipecat_connection = SmallWebRTCConnection()
         await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
 
         @pipecat_connection.event_handler("closed")

src/pipecat/audio/turn/smart_turn/base_smart_turn.py

@@ -176,6 +176,8 @@ class BaseSmartTurn(BaseTurnAnalyzer):
                     f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}"
                 )
                 state = EndOfTurnState.COMPLETE
+            except Exception as e:
+                logger.error(f"Error during prediction: {e}")
 
         else:
             logger.trace(f"params: {self._params}, stop_ms: {self._stop_ms}")

src/pipecat/audio/turn/smart_turn/http_smart_turn.py

@@ -40,7 +40,7 @@ class HttpSmartTurnAnalyzer(BaseSmartTurn):
     async def _send_raw_request(self, data_bytes: bytes) -> Dict[str, Any]:
         headers = {"Content-Type": "application/octet-stream"}
         headers.update(self._headers)
-
+        logger.trace(f"Sending {len(data_bytes)} bytes as raw body to {self._url}...")
         try:
             timeout = aiohttp.ClientTimeout(total=self._params.stop_secs)
 
@@ -50,30 +50,23 @@ class HttpSmartTurnAnalyzer(BaseSmartTurn):
                 logger.trace("\n--- Response ---")
                 logger.trace(f"Status Code: {response.status}")
 
-                # Check if successful
-                if response.status != 200:
+                if response.status == 200:
+                    try:
+                        json_data = await response.json()
+                        logger.trace("Response JSON:")
+                        logger.trace(json_data)
+                        return json_data
+                    except aiohttp.ContentTypeError:
+                        # Non-JSON response
+                        text = await response.text()
+                        logger.trace("Response Content (non-JSON):")
+                        logger.trace(text)
+                        raise Exception(f"Non-JSON response: {text}")
+                else:
                     error_text = await response.text()
                     logger.trace("Response Content (Error):")
                     logger.trace(error_text)
-
-                    if response.status == 500:
-                        logger.warning(f"Smart turn service returned 500 error: {error_text}")
-                        raise Exception(f"Server returned HTTP 500: {error_text}")
-                    else:
-                        response.raise_for_status()
-
-                # Process successful response
-                try:
-                    json_data = await response.json()
-                    logger.trace("Response JSON:")
-                    logger.trace(json_data)
-                    return json_data
-                except aiohttp.ContentTypeError:
-                    # Non-JSON response
-                    text = await response.text()
-                    logger.trace("Response Content (non-JSON):")
-                    logger.trace(text)
-                    raise Exception(f"Non-JSON response: {text}")
+                    response.raise_for_status()
 
         except asyncio.TimeoutError:
             logger.error(f"Request timed out after {self._params.stop_secs} seconds")
@@ -83,14 +76,5 @@ class HttpSmartTurnAnalyzer(BaseSmartTurn):
             raise Exception("Failed to send raw request to Daily Smart Turn.")
 
     async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
-        try:
-            serialized_array = self._serialize_array(audio_array)
-            return await self._send_raw_request(serialized_array)
-        except Exception as e:
-            logger.error(f"Smart turn prediction failed: {str(e)}")
-            # Return an incomplete prediction when a failure occurs
-            return {
-                "prediction": 0,
-                "probability": 0.0,
-                "metrics": {"inference_time": 0.0, "total_time": 0.0},
-            }
+        serialized_array = self._serialize_array(audio_array)
+        return await self._send_raw_request(serialized_array)

src/pipecat/transports/network/webrtc_connection.py

@@ -7,7 +7,7 @@
 import asyncio
 import json
 import time
-from typing import Any, List, Literal, Optional, Union
+from typing import Any, Literal, Optional, Union
 
 from av.frame import Frame
 from loguru import logger
@@ -87,21 +87,13 @@ class SmallWebRTCTrack:
         return getattr(self._track, name)
 
 
-# Alias so we don't need to expose RTCIceServer
-IceServer = RTCIceServer
-
-
 class SmallWebRTCConnection(BaseObject):
-    def __init__(self, ice_servers: Optional[Union[List[str], List[IceServer]]] = None):
+    def __init__(self, ice_servers=None):
         super().__init__()
-        if not ice_servers:
-            self.ice_servers: List[IceServer] = []
-        elif all(isinstance(s, IceServer) for s in ice_servers):
-            self.ice_servers = ice_servers
-        elif all(isinstance(s, str) for s in ice_servers):
-            self.ice_servers = [IceServer(urls=s) for s in ice_servers]
+        if ice_servers:
+            self.ice_servers = [RTCIceServer(urls=server) for server in ice_servers]
         else:
-            raise TypeError("ice_servers must be either List[str] or List[RTCIceServer]")
+            self.ice_servers = []
         self._connect_invoked = False
         self._track_map = {}
         self._track_getters = {