From a698c4064b04e08ccafc076bfb80fe2ad3a2a424 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Sep 2025 09:17:11 -0700 Subject: [PATCH 1/4] BaseObject: allow synchronous event handlers --- CHANGELOG.md | 9 +++++ src/pipecat/utils/base_object.py | 63 +++++++++++++++++++++++--------- 2 files changed, 54 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 74f5aa18c..f0ba80455 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- It is now possible to register synchronous event handlers. By default, all + event handlers are executed in a separate task. However, in some cases we want + to guarantee order of execution, for example, executing something before + disconnecting a transport. + + ```python + self._register_event_handler("on_event_name", sync=True) + ``` + - Added support for global location in `GoogleVertexLLMService`. The service now supports both regional locations (e.g., "us-east4") and the "global" location for Vertex AI endpoints. When using "global" location, the service will use diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py index bf34515a6..b40911fec 100644 --- a/src/pipecat/utils/base_object.py +++ b/src/pipecat/utils/base_object.py @@ -14,13 +14,33 @@ and async cleanup for all Pipecat components. import asyncio import inspect from abc import ABC -from typing import Optional +from dataclasses import dataclass +from typing import Any, Dict, List, Optional from loguru import logger from pipecat.utils.utils import obj_count, obj_id +@dataclass +class EventHandler: + """Data class to store event handlers information. + + This data class stores the event name, a list of handlers to run for this + event, and whether these handlers will be executed in a task. + + Attributes: + name (str): The name of the event handler. + handlers (List[Any]): A list of functions to be called when this event is triggered. + is_sync (bool): Indicates whether the functions are executed in a task. + + """ + + name: str + handlers: List[Any] + is_sync: bool + + class BaseObject(ABC): """Abstract base class providing common functionality for Pipecat objects. @@ -41,7 +61,7 @@ class BaseObject(ABC): self._name = name or f"{self.__class__.__name__}#{obj_count(self)}" # Registered event handlers. - self._event_handlers: dict = {} + self._event_handlers: Dict[str, EventHandler] = {} # Set of tasks being executed. When a task finishes running it gets # automatically removed from the set. When we cleanup we wait for all @@ -103,18 +123,21 @@ class BaseObject(ABC): Can be sync or async. """ if event_name in self._event_handlers: - self._event_handlers[event_name].append(handler) + self._event_handlers[event_name].handlers.append(handler) else: logger.warning(f"Event handler {event_name} not registered") - def _register_event_handler(self, event_name: str): + def _register_event_handler(self, event_name: str, sync: bool = False): """Register an event handler type. Args: event_name: The name of the event type to register. + sync: Whether this event handler will be executed in a task. """ if event_name not in self._event_handlers: - self._event_handlers[event_name] = [] + self._event_handlers[event_name] = EventHandler( + name=event_name, handlers=[], is_sync=sync + ) else: logger.warning(f"Event handler {event_name} not registered") @@ -126,36 +149,40 @@ class BaseObject(ABC): *args: Positional arguments to pass to event handlers. **kwargs: Keyword arguments to pass to event handlers. """ - # If we haven't registered an event handler, we don't need to do - # anything. - if not self._event_handlers.get(event_name): + if event_name not in self._event_handlers: return - # Create the task. - task = asyncio.create_task(self._run_task(event_name, *args, **kwargs)) + event_handler = self._event_handlers[event_name] - # Add it to our list of event tasks. - self._event_tasks.add((event_name, task)) + if event_handler.is_sync: + # Just run the handler. + await self._run_handler(event_handler, *args, **kwargs) + else: + # Create the task. + task = asyncio.create_task(self._run_handler(event_handler, *args, **kwargs)) - # Remove the task from the event tasks list when the task completes. - task.add_done_callback(self._event_task_finished) + # Add it to our list of event tasks. + self._event_tasks.add((event_name, task)) - async def _run_task(self, event_name: str, *args, **kwargs): + # Remove the task from the event tasks list when the task completes. + task.add_done_callback(self._event_task_finished) + + async def _run_handler(self, event_handler: EventHandler, *args, **kwargs): """Execute all handlers for an event. Args: - event_name: The name of the event being handled. + event_handler: The event handler to run. *args: Positional arguments to pass to handlers. **kwargs: Keyword arguments to pass to handlers. """ try: - for handler in self._event_handlers[event_name]: + for handler in event_handler.handlers: if inspect.iscoroutinefunction(handler): await handler(self, *args, **kwargs) else: handler(self, *args, **kwargs) except Exception as e: - logger.exception(f"Exception in event handler {event_name}: {e}") + logger.exception(f"Exception in event handler {event_handler.name}: {e}") def _event_task_finished(self, task: asyncio.Task): """Clean up completed event handler tasks. From 492da16cc9deedb72d11342758538f8bdb843fe2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Sep 2025 09:18:58 -0700 Subject: [PATCH 2/4] DailyTransport: added synchronous on_before_disconnect event --- CHANGELOG.md | 2 ++ src/pipecat/transports/daily/transport.py | 16 +++++++++++++++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0ba80455..8964c174c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `on_before_disconnect` synchronous event to `DailyTransport`. + - It is now possible to register synchronous event handlers. By default, all event handlers are executed in a separate task. However, in some cases we want to guarantee order of execution, for example, executing something before diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 1568a12bd..fa6950f81 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -215,6 +215,7 @@ class DailyCallbacks(BaseModel): on_active_speaker_changed: Called when the active speaker of the call has changed. on_joined: Called when bot successfully joined a room. on_left: Called when bot left a room. + on_before_leave: Called when bot is about to leave the room. on_error: Called when an error occurs. on_app_message: Called when receiving an app message. on_call_state_updated: Called when call state changes. @@ -244,6 +245,7 @@ class DailyCallbacks(BaseModel): on_active_speaker_changed: Callable[[Mapping[str, Any]], Awaitable[None]] on_joined: Callable[[Mapping[str, Any]], Awaitable[None]] on_left: Callable[[], Awaitable[None]] + on_before_leave: Callable[[], Awaitable[None]] on_error: Callable[[str], Awaitable[None]] on_app_message: Callable[[Any, str], Awaitable[None]] on_call_state_updated: Callable[[str], Awaitable[None]] @@ -720,6 +722,9 @@ class DailyTransportClient(EventHandler): logger.info(f"Leaving {self._room_url}") + # Call callback before leaving. + await self._callbacks.on_before_leave() + if self._params.transcription_enabled: await self.stop_transcription() @@ -1880,6 +1885,7 @@ class DailyTransport(BaseTransport): on_active_speaker_changed=self._on_active_speaker_changed, on_joined=self._on_joined, on_left=self._on_left, + on_before_leave=self._on_before_leave, on_error=self._on_error, on_app_message=self._on_app_message, on_call_state_updated=self._on_call_state_updated, @@ -1943,6 +1949,10 @@ class DailyTransport(BaseTransport): self._register_event_handler("on_recording_started") self._register_event_handler("on_recording_stopped") self._register_event_handler("on_recording_error") + self._register_event_handler("on_before_disconnect", sync=True) + # Deprecated + self._register_event_handler("on_joined") + self._register_event_handler("on_left") # # BaseTransport @@ -2194,6 +2204,10 @@ class DailyTransport(BaseTransport): """Handle room left events.""" await self._call_event_handler("on_left") + async def _on_before_leave(self): + """Handle before leave room events.""" + await self._call_event_handler("on_before_disconnect") + async def _on_error(self, error): """Handle error events and push error frames.""" await self._call_event_handler("on_error", error) @@ -2333,7 +2347,7 @@ class DailyTransport(BaseTransport): """Handle participant updated events.""" await self._call_event_handler("on_participant_updated", participant) - async def _on_transcription_message(self, message: Dict[str, Any]) -> None: + async def _on_transcription_message(self, message: Mapping[str, Any]) -> None: """Handle transcription message events.""" await self._call_event_handler("on_transcription_message", message) From 3a6c9786e838f4a29c58dac38c69685ff7efae50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Sep 2025 09:33:26 -0700 Subject: [PATCH 3/4] LiveKitTransport: added synchronous before_disconnect event --- CHANGELOG.md | 3 ++- src/pipecat/transports/livekit/transport.py | 8 ++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8964c174c..3573bec5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added `on_before_disconnect` synchronous event to `DailyTransport`. +- Added `on_before_disconnect` synchronous event to `DailyTransport` and + `LiveKitTransport`. - It is now possible to register synchronous event handlers. By default, all event handlers are executed in a separate task. However, in some cases we want diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 3e7ccfdf7..227a9f4f9 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -114,6 +114,7 @@ class LiveKitCallbacks(BaseModel): on_connected: Callable[[], Awaitable[None]] on_disconnected: Callable[[], Awaitable[None]] + on_before_disconnect: Callable[[], Awaitable[None]] on_participant_connected: Callable[[str], Awaitable[None]] on_participant_disconnected: Callable[[str], Awaitable[None]] on_audio_track_subscribed: Callable[[str], Awaitable[None]] @@ -282,6 +283,7 @@ class LiveKitTransportClient: return logger.info(f"Disconnecting from {self._room_name}") + await self._callbacks.on_before_disconnect() await self.room.disconnect() self._connected = False logger.info(f"Disconnected from {self._room_name}") @@ -918,6 +920,7 @@ class LiveKitTransport(BaseTransport): callbacks = LiveKitCallbacks( on_connected=self._on_connected, on_disconnected=self._on_disconnected, + on_before_disconnect=self._on_before_disconnect, on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected, on_audio_track_subscribed=self._on_audio_track_subscribed, @@ -947,6 +950,7 @@ class LiveKitTransport(BaseTransport): self._register_event_handler("on_first_participant_joined") self._register_event_handler("on_participant_left") self._register_event_handler("on_call_state_updated") + self._register_event_handler("on_before_disconnect", sync=True) def input(self) -> LiveKitInputTransport: """Get the input transport for receiving media and events. @@ -1041,6 +1045,10 @@ class LiveKitTransport(BaseTransport): """Handle room disconnected events.""" await self._call_event_handler("on_disconnected") + async def _on_before_disconnect(self): + """Handle before disconnection room events.""" + await self._call_event_handler("on_before_disconnect") + async def _on_participant_connected(self, participant_id: str): """Handle participant connected events.""" await self._call_event_handler("on_participant_connected", participant_id) From 077b949bb2929d6ac0297871f163ab14f28e725c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Sep 2025 10:47:40 -0700 Subject: [PATCH 4/4] BaseObject: run each handler for the same event in a separate task --- CHANGELOG.md | 2 ++ src/pipecat/utils/base_object.py | 41 ++++++++++++++++++-------------- 2 files changed, 25 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3573bec5b..5102eda37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue where multiple handlers for an event would not run in parallel. + - Fixed `DailyTransport.sip_call_transfer()` to automatically use the session ID from the `on_dialin_connected` event, when not explicitly provided. Now supports cold transfers (from incoming dial-in calls) by automatically diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py index b40911fec..a1c75f733 100644 --- a/src/pipecat/utils/base_object.py +++ b/src/pipecat/utils/base_object.py @@ -154,35 +154,40 @@ class BaseObject(ABC): event_handler = self._event_handlers[event_name] - if event_handler.is_sync: - # Just run the handler. - await self._run_handler(event_handler, *args, **kwargs) - else: - # Create the task. - task = asyncio.create_task(self._run_handler(event_handler, *args, **kwargs)) + for handler in event_handler.handlers: + if event_handler.is_sync: + # Just run the handler. + await self._run_handler(event_handler.name, handler, *args, **kwargs) + else: + # Create the task. Note that this is a task per each function + # handler. Users can register to an event handler multiple + # times. + task = asyncio.create_task( + self._run_handler(event_handler.name, handler, *args, **kwargs) + ) - # Add it to our list of event tasks. - self._event_tasks.add((event_name, task)) + # Add it to our list of event tasks. + self._event_tasks.add((event_name, task)) - # Remove the task from the event tasks list when the task completes. - task.add_done_callback(self._event_task_finished) + # Remove the task from the event tasks list when the task completes. + task.add_done_callback(self._event_task_finished) - async def _run_handler(self, event_handler: EventHandler, *args, **kwargs): + async def _run_handler(self, event_name: str, handler, *args, **kwargs): """Execute all handlers for an event. Args: - event_handler: The event handler to run. + event_name: The event name for this handler. + handler: The handler function to run. *args: Positional arguments to pass to handlers. **kwargs: Keyword arguments to pass to handlers. """ try: - for handler in event_handler.handlers: - if inspect.iscoroutinefunction(handler): - await handler(self, *args, **kwargs) - else: - handler(self, *args, **kwargs) + if inspect.iscoroutinefunction(handler): + await handler(self, *args, **kwargs) + else: + handler(self, *args, **kwargs) except Exception as e: - logger.exception(f"Exception in event handler {event_handler.name}: {e}") + logger.exception(f"Exception in event handler {event_name}: {e}") def _event_task_finished(self, task: asyncio.Task): """Clean up completed event handler tasks.