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. Requiresrecording=true
inVonageConfig
.
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.