From 3aa89fb13a0985f1898a362d4c299f9ee31fa0b7 Mon Sep 17 00:00:00 2001 From: Vaibhav-Lodha Date: Thu, 21 Nov 2024 12:20:51 +0530 Subject: [PATCH 1/9] adding session_timeout param --- src/pipecat/transports/network/websocket_server.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index c0c8595e9..23d349d74 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -41,6 +41,7 @@ except ModuleNotFoundError as e: class WebsocketServerParams(TransportParams): add_wav_header: bool = False serializer: FrameSerializer = ProtobufFrameSerializer() + session_timeout: int | None = None class WebsocketServerCallbacks(BaseModel): @@ -218,6 +219,9 @@ class WebsocketServerTransport(BaseTransport): self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") + if self._params.session_timeout: + self._register_event_handler("on_session_timeout") + def input(self) -> WebsocketServerInputTransport: if not self._input: self._input = WebsocketServerInputTransport( From e11f128110562874fc42bbf58b23f79c1258ae73 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 21 Nov 2024 12:34:32 +0530 Subject: [PATCH 2/9] adding on_session_timeout --- src/pipecat/transports/network/websocket_server.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 23d349d74..49034f208 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -234,9 +234,22 @@ class WebsocketServerTransport(BaseTransport): self._output = WebsocketServerOutputTransport(self._params, name=self._output_name) return self._output + async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol): + """ + Wait for 60 seconds, if the websocket is still open, trigger timeout event. + """ + try: + await asyncio.sleep(self._params.session_timeout) + if not websocket.closed: + await self._call_event_handler("on_session_timeout", websocket) + except asyncio.CancelledError: + logger.info(f"Monitoring task cancelled for: {websocket.remote_address}") + async def _on_client_connected(self, websocket): if self._output: await self._output.set_client_connection(websocket) + if self._params.session_timeout: + self._loop.create_task(self._monitor_websocket(websocket)) await self._call_event_handler("on_client_connected", websocket) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") @@ -247,3 +260,4 @@ class WebsocketServerTransport(BaseTransport): await self._call_event_handler("on_client_disconnected", websocket) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") + From da254c51439c27f48c575cc71db5545d4f1f80f4 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 21 Nov 2024 12:36:51 +0530 Subject: [PATCH 3/9] correcting _monitor_websocket --- src/pipecat/transports/network/websocket_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 49034f208..2145aeee3 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -236,7 +236,7 @@ class WebsocketServerTransport(BaseTransport): async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol): """ - Wait for 60 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) From 7dfa88666916e51bd033e8e2b4271764a2291fc6 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 21 Nov 2024 14:45:24 +0530 Subject: [PATCH 4/9] moving logic to WebsocketServerInputTransport --- .../transports/network/websocket_server.py | 37 +++++++++++-------- 1 file changed, 21 insertions(+), 16 deletions(-) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 2145aeee3..a40d00485 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -47,6 +47,7 @@ class WebsocketServerParams(TransportParams): 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): @@ -99,6 +100,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) @@ -125,6 +130,17 @@ 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): @@ -209,6 +225,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 @@ -218,9 +235,7 @@ class WebsocketServerTransport(BaseTransport): # these handlers. self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") - - if self._params.session_timeout: - self._register_event_handler("on_session_timeout") + self._register_event_handler("on_session_timeout") def input(self) -> WebsocketServerInputTransport: if not self._input: @@ -234,22 +249,9 @@ class WebsocketServerTransport(BaseTransport): self._output = WebsocketServerOutputTransport(self._params, name=self._output_name) return self._output - 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._call_event_handler("on_session_timeout", websocket) - except asyncio.CancelledError: - logger.info(f"Monitoring task cancelled for: {websocket.remote_address}") - async def _on_client_connected(self, websocket): if self._output: await self._output.set_client_connection(websocket) - if self._params.session_timeout: - self._loop.create_task(self._monitor_websocket(websocket)) await self._call_event_handler("on_client_connected", websocket) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") @@ -261,3 +263,6 @@ class WebsocketServerTransport(BaseTransport): 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) + From 6e8e7fa19a7817fa08fbac795dd3c0175eb4240e Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 21 Nov 2024 14:56:42 +0530 Subject: [PATCH 5/9] adding session_timeout in fastapi --- .../transports/network/fastapi_websocket.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index c0363c1a9..1c939f952 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -44,11 +44,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): @@ -67,6 +69,7 @@ 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()) await self._callbacks.on_client_connected(self._websocket) self._receive_task = self.get_event_loop().create_task(self._receive_messages()) @@ -88,6 +91,16 @@ 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): @@ -163,6 +176,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( @@ -176,6 +190,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 @@ -188,3 +203,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) From af90d65b3be34eeb44b274ebb2d5fd2940ad79c0 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Wed, 8 Jan 2025 10:43:10 +0530 Subject: [PATCH 6/9] 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) - From 2da5620d10f6893f9fd88c99ee8793f8efb2dcd3 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Wed, 8 Jan 2025 10:50:09 +0530 Subject: [PATCH 7/9] adding changelog --- CHANGELOG.md | 5 +++++ examples/websocket-server/bot.py | 5 ++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0758360d6..e4ee4e5f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,15 @@ 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` param in `FastAPIWebsocketTransport` and `WebsocketServerTransport` + to set the session timeout in seconds, this triggers callback method `on_session_timeout` + where you can handle the timeout event as per business logic. Refer to `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 483920184..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, BotInterruptionFrame, EndFrame +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 @@ -32,8 +32,7 @@ logger.add(sys.stderr, level="DEBUG") class SessionTimeoutHandler: - """ - Handles actions to be performed when a session times out. + """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). From 768487ffb358a538f3497e254c769e92e31f2d2c Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Wed, 8 Jan 2025 10:53:32 +0530 Subject: [PATCH 8/9] final changelog --- CHANGELOG.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4ee4e5f6..523f43ebc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,9 @@ 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` param in `FastAPIWebsocketTransport` and `WebsocketServerTransport` - to set the session timeout in seconds, this triggers callback method `on_session_timeout` - where you can handle the timeout event as per business logic. Refer to `examples/websocket-server/bot.py`. - +- 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 From 3b446234a71205b2ee65319b55d0784f6068bc7a Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Wed, 8 Jan 2025 10:54:27 +0530 Subject: [PATCH 9/9] fix hyperlink --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 523f43ebc..6adf6a994 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,7 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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]. + See [examples/websocket-server/bot.py](https://github.com/pipecat-ai/pipecat/blob/main/examples/websocket-server/bot.py). ### Changed