diff --git a/CHANGELOG.md b/CHANGELOG.md index 0758360d6..6adf6a994 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `LiveKitTransportLayer`. - Added `enable_prejoin_ui`, `max_participants` and `start_video_off` params to `DailyRoomProperties`. +- Added `session_timeout` to `FastAPIWebsocketTransport` and `WebsocketServerTransport` + for configuring session timeouts (in seconds). Triggers `on_session_timeout` for custom timeout handling. + See [examples/websocket-server/bot.py](https://github.com/pipecat-ai/pipecat/blob/main/examples/websocket-server/bot.py). ### Changed - api_key, aws_access_key_id and region are no longer required parameters for the PollyTTSService (AWSTTSService) +- Added `session_timeout` example in `examples/websocket-server/bot.py` to handle session timeout event. ### Fixed diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index fd792d0df..7c6670205 100644 --- a/examples/websocket-server/bot.py +++ b/examples/websocket-server/bot.py @@ -12,7 +12,7 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import LLMMessagesFrame +from pipecat.frames.frames import BotInterruptionFrame, EndFrame, LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -31,6 +31,56 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") +class SessionTimeoutHandler: + """Handles actions to be performed when a session times out. + Inputs: + - task: Pipeline task (used to queue frames). + - tts: TTS service (used to generate speech output). + """ + + def __init__(self, task, tts): + self.task = task + self.tts = tts + self.background_tasks = set() + + async def handle_timeout(self, client_address): + """ + Handles the timeout event for a session. + """ + try: + logger.info(f"Connection timeout for {client_address}") + + # Queue a BotInterruptionFrame to notify the user + await self.task.queue_frames([BotInterruptionFrame()]) + + # Send the TTS message to inform the user about the timeout + await self.tts.say( + "I'm sorry, we are ending the call now. Please feel free to reach out again if you need assistance." + ) + + # Start the process to gracefully end the call in the background + end_call_task = asyncio.create_task(self._end_call()) + self.background_tasks.add(end_call_task) + end_call_task.add_done_callback(self.background_tasks.discard) + except Exception as e: + logger.error(f"Error during session timeout handling: {e}") + + async def _end_call(self): + """ + Completes the session termination process after the TTS message. + """ + try: + # Wait for a duration to ensure TTS has completed + await asyncio.sleep(15) + + # Queue both BotInterruptionFrame and EndFrame to conclude the session + await self.task.queue_frames([BotInterruptionFrame(), EndFrame()]) + + logger.info("TTS completed and EndFrame pushed successfully.") + except Exception as e: + logger.error(f"Error during call termination: {e}") + + async def main(): transport = WebsocketServerTransport( params=WebsocketServerParams( @@ -40,6 +90,7 @@ async def main(): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, + session_timeout=60 * 3, # 3 minutes ) ) @@ -83,6 +134,14 @@ async def main(): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_session_timeout") + async def on_session_timeout(transport, client): + logger.info(f"Entering in timeout for {client.remote_address}") + + timeout_handler = SessionTimeoutHandler(task, tts) + + await timeout_handler.handle_timeout(client) + runner = PipelineRunner() await runner.run(task) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 02270d5c0..bb3f24b75 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -42,11 +42,13 @@ except ModuleNotFoundError as e: class FastAPIWebsocketParams(TransportParams): add_wav_header: bool = False serializer: FrameSerializer + session_timeout: int | None = None class FastAPIWebsocketCallbacks(BaseModel): on_client_connected: Callable[[WebSocket], Awaitable[None]] on_client_disconnected: Callable[[WebSocket], Awaitable[None]] + on_session_timeout: Callable[[WebSocket], Awaitable[None]] class FastAPIWebsocketInputTransport(BaseInputTransport): @@ -65,6 +67,10 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + if self._params.session_timeout: + self._monitor_websocket_task = self.get_event_loop().create_task( + self._monitor_websocket() + ) await self._callbacks.on_client_connected(self._websocket) self._receive_task = self.get_event_loop().create_task(self._receive_messages()) @@ -88,6 +94,14 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): await self._callbacks.on_client_disconnected(self._websocket) + async def _monitor_websocket(self): + """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.""" + try: + await asyncio.sleep(self._params.session_timeout) + await self._callbacks.on_session_timeout(self._websocket) + except asyncio.CancelledError: + logger.info(f"Monitoring task cancelled for: {self._websocket}") + class FastAPIWebsocketOutputTransport(BaseOutputTransport): def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, **kwargs): @@ -176,6 +190,7 @@ class FastAPIWebsocketTransport(BaseTransport): self._callbacks = FastAPIWebsocketCallbacks( on_client_connected=self._on_client_connected, on_client_disconnected=self._on_client_disconnected, + on_session_timeout=self._on_session_timeout, ) self._input = FastAPIWebsocketInputTransport( @@ -189,6 +204,7 @@ class FastAPIWebsocketTransport(BaseTransport): # these handlers. self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") + self._register_event_handler("on_session_timeout") def input(self) -> FastAPIWebsocketInputTransport: return self._input @@ -201,3 +217,6 @@ class FastAPIWebsocketTransport(BaseTransport): async def _on_client_disconnected(self, websocket): await self._call_event_handler("on_client_disconnected", websocket) + + async def _on_session_timeout(self, websocket): + await self._call_event_handler("on_session_timeout", websocket) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 4bbfeef7e..626fbcbe0 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -40,11 +40,13 @@ except ModuleNotFoundError as e: class WebsocketServerParams(TransportParams): add_wav_header: bool = False serializer: FrameSerializer = ProtobufFrameSerializer() + session_timeout: int | None = None class WebsocketServerCallbacks(BaseModel): on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] + on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] class WebsocketServerInputTransport(BaseInputTransport): @@ -97,6 +99,10 @@ class WebsocketServerInputTransport(BaseInputTransport): # Notify await self._callbacks.on_client_connected(websocket) + # Create a task to monitor the websocket connection + if self._params.session_timeout: + self.get_event_loop().create_task(self._monitor_websocket(websocket)) + # Handle incoming messages async for message in websocket: frame = self._params.serializer.deserialize(message) @@ -117,6 +123,15 @@ class WebsocketServerInputTransport(BaseInputTransport): logger.info(f"Client {websocket.remote_address} disconnected") + async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol): + """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.""" + try: + await asyncio.sleep(self._params.session_timeout) + if not websocket.closed: + await self._callbacks.on_session_timeout(websocket) + except asyncio.CancelledError: + logger.info(f"Monitoring task cancelled for: {websocket.remote_address}") + class WebsocketServerOutputTransport(BaseOutputTransport): def __init__(self, params: WebsocketServerParams, **kwargs): @@ -211,6 +226,7 @@ class WebsocketServerTransport(BaseTransport): self._callbacks = WebsocketServerCallbacks( on_client_connected=self._on_client_connected, on_client_disconnected=self._on_client_disconnected, + on_session_timeout=self._on_session_timeout, ) self._input: WebsocketServerInputTransport | None = None self._output: WebsocketServerOutputTransport | None = None @@ -220,6 +236,7 @@ class WebsocketServerTransport(BaseTransport): # these handlers. self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") + self._register_event_handler("on_session_timeout") def input(self) -> WebsocketServerInputTransport: if not self._input: @@ -246,3 +263,6 @@ class WebsocketServerTransport(BaseTransport): await self._call_event_handler("on_client_disconnected", websocket) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") + + async def _on_session_timeout(self, websocket): + await self._call_event_handler("on_session_timeout", websocket)