From af90d65b3be34eeb44b274ebb2d5fd2940ad79c0 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Wed, 8 Jan 2025 10:43:10 +0530 Subject: [PATCH] adding session timeout example in websocket-server example --- examples/websocket-server/bot.py | 62 ++++++++++++++++++- .../transports/network/fastapi_websocket.py | 9 +-- .../transports/network/websocket_server.py | 7 +-- 3 files changed, 68 insertions(+), 10 deletions(-) diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index fd792d0df..483920184 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 LLMMessagesFrame, BotInterruptionFrame, EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -31,6 +31,57 @@ 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 +91,7 @@ async def main(): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, + session_timeout=60 * 3, # 3 minutes ) ) @@ -83,6 +135,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 638d93018..bb3f24b75 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -67,7 +67,10 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) - self._monitor_websocket_task = self.get_event_loop().create_task(self._monitor_websocket()) + 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()) @@ -92,9 +95,7 @@ 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. - """ + """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) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 37271744f..626fbcbe0 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -124,9 +124,7 @@ 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. - """ + """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: @@ -228,7 +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 + on_session_timeout=self._on_session_timeout, ) self._input: WebsocketServerInputTransport | None = None self._output: WebsocketServerOutputTransport | None = None @@ -268,4 +266,3 @@ class WebsocketServerTransport(BaseTransport): async def _on_session_timeout(self, websocket): await self._call_event_handler("on_session_timeout", websocket) -