Events manager

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

The 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.

Initialization

def __init__(self, subscriptions: List[EventType] = []):
    self.queue = asyncio.Queue()
    self.subscriptions = set(subscriptions)
    self.active = False

The 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 False.

Publishing Events

def publish_event(self, event: Event):
    if event.type in self.subscriptions:
        self.queue.put_nowait(event)

The 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 recording=true in VonageConfig.

Example Usage

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 add_transcript method:

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.

Guides

Python

Events manager

What is the Events Manager

EventsManager Class

Initialization

Publishing Events

Starting the Event Loop

Current Event Types

Example Usage

​What is the Events Manager

​EventsManager Class

​Initialization

​Publishing Events

​Starting the Event Loop

​Current Event Types

​Example Usage

What is the Events Manager

EventsManager Class

Initialization

Publishing Events

Starting the Event Loop

Current Event Types

Example Usage