What is the Events Manager
The Events Manager is a class designed to facilitate asynchronous handling of events in the application. It allows for non-blocking actions on events, such as processing transcripts, managing phone calls, and other tasks. The main components of the Events Manager are the
EventsManager class and several
Event subclasses representing various event types.
EventsManager class is responsible for managing the event queue and handling events asynchronously. The class provides methods for publishing events, starting the event loop, handling events, and ending the event loop.
def __init__(self, subscriptions: List[EventType] = ): self.queue = asyncio.Queue() self.subscriptions = set(subscriptions) self.active = False
EventsManager constructor accepts an optional list of
EventType subscriptions. By default, it initializes an empty set of subscriptions, an asynchronous queue, and sets the
active attribute to
def publish_event(self, event: Event): if event.type in self.subscriptions: self.queue.put_nowait(event)
publish_event method takes an
Event object as input and adds it to the queue if its type is in the set of subscribed event types.
Starting the Event Loop
async def start(self): self.active = True while self.active: try: event: Event = await self.queue.get() except asyncio.QueueEmpty: await asyncio.sleep(1) self.handle_event(event)
Current Event Types
The current event types include:
TRANSCRIPT: Indicates a partial transcript for the conversation has been received.
TRANSCRIPT_COMPLETE: Indicates the transcript is complete (ie conversation has ended).
PHONE_CALL_CONNECTED: Indicates a phone call has been connected.
PHONE_CALL_ENDED: Indicates a phone call has ended.
RECORDING: (Vonage Only) Indicates a secure URL containing a recording of the call is available. Requires
The following example demonstrates how the
EventsManager class can be used to consume the
TRANSCRIPT_COMPLETE event and save the transcript to a file using the
import logging from fastapi import FastAPI from vocode.streaming.models.events import Event, EventType, TranscriptCompleteEvent from vocode.streaming.utils import events_manager from call_transcript_utils import add_transcript app = FastAPI(docs_url=None) logging.basicConfig() logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) class CustomEventsManager(events_manager.EventsManager): def __init__(self): super().__init__(subscriptions=[EventType.TRANSCRIPT_COMPLETE]) def handle_event(self, event: Event): if event.type == EventType.TRANSCRIPT_COMPLETE: transcript_complete_event = typing.cast(TranscriptCompleteEvent, event) add_transcript( transcript_complete_event.conversation_id, transcript_complete_event.transcript, ) events_manager_instance = CustomEventsManager() await events_manager_instance.start()
In this example, a custom
EventsManager subclass is created with a subscription to the
TRANSCRIPT_COMPLETE event. The
handle_event method is overridden to save the transcript to a file using the
add_transcript method when the
TRANSCRIPT_COMPLETE event is received.