Event System Guide - Vision Agents

This guide shows you how to use the event system in your Vision Agents applications. We’ll build a practical example that responds to user interactions using events.

Prerequisites

Basic understanding of Vision Agents (see Voice Agents)
Python async/await knowledge

Example: Greeting New Participants

Let’s enhance the simple agent example to automatically greet users when they join a call using the event system.

Starting with the Basic Agent

First, let’s look at the basic agent setup:

import asyncio
import logging
from uuid import uuid4

from dotenv import load_dotenv
from vision_agents.core.edge.types import User
from vision_agents.plugins import elevenlabs, deepgram, openai, getstream
from vision_agents.core import agents, cli

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
load_dotenv()

async def start_agent():
    call_id = str(uuid4())
    
    # Create the agent with basic components
    agent = agents.Agent(
        edge=getstream.Edge(),
        agent_user=User(name="My AI Assistant", id="agent"),
        instructions="You're a helpful voice AI assistant. Keep responses short and conversational.",
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(),
    )
    
    await agent.create_user()
    
    # Create and join the call
    call = agent.edge.client.video.call("default", call_id)
    await agent.edge.open_demo(call)
    
    with await agent.join(call):
        await agent.finish()

Adding Event Handling

Now let’s add event handling to greet participants when they join:

import asyncio
import logging
from uuid import uuid4

from dotenv import load_dotenv
from vision_agents.core.edge.types import User
from vision_agents.plugins import elevenlabs, deepgram, openai, getstream
from vision_agents.core import agents, cli
from vision_agents.core.events import CallSessionParticipantJoinedEvent

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
load_dotenv()

async def start_agent():
    call_id = str(uuid4())
    
    # Create the agent with basic components
    agent = agents.Agent(
        edge=getstream.Edge(),
        agent_user=User(name="My AI Assistant", id="agent"),
        instructions="You're a helpful voice AI assistant. Keep responses short and conversational.",
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(),
    )
    
    await agent.create_user()
    
    # Subscribe to participant joined events
    @agent.events.subscribe
    async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
        # Skip if the participant joining is the agent itself
        if event.participant.user.id == "agent":
            return
        
        # Log the event
        logger.info(f"New participant joined: {event.participant.user.name}")
        
        # Greet the new participant
        await agent.simple_response(f"Hello {event.participant.user.name}! Welcome to the call.")
    
    # Create and join the call
    call = agent.edge.client.video.call("default", call_id)
    await agent.edge.open_demo(call)
    
    with await agent.join(call):
        await agent.finish()

if __name__ == "__main__":
    asyncio.run(cli.start_dispatcher(start_agent))

Understanding the Event Handler

Let’s break down the event handler:

@agent.events.subscribe
async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
    # Skip if the participant joining is the agent itself
    if event.participant.user.id == "agent":
        return
    
    # Log the event
    logger.info(f"New participant joined: {event.participant.user.name}")
    
    # Greet the new participant
    await agent.simple_response(f"Hello {event.participant.user.name}! Welcome to the call.")

Key Points:

@agent.events.subscribe decorator registers the function as an event handler
Type hint CallSessionParticipantJoinedEvent tells the system which events to handle
The function is async to support non-blocking operations
We check if the participant is the agent itself to avoid self-greeting
agent.simple_response() is used to make the agent speak

Adding More Event Handlers

Let’s add handlers for when participants leave and for other call events:

# Subscribe to participant left events
@agent.events.subscribe
async def handle_participant_left(event: CallSessionParticipantLeftEvent):
    if event.participant.user.id == "agent":
        return
    
    logger.info(f"Participant left: {event.participant.user.name}")
    await agent.simple_response(f"Goodbye {event.participant.user.name}!")

# Subscribe to call session started events
@agent.events.subscribe
async def handle_session_started(event: CallSessionStartedEvent):
    logger.info("Call session started")
    await agent.simple_response("The call has started. I'm here to help!")

# Subscribe to call session ended events
@agent.events.subscribe
async def handle_session_ended(event: CallSessionEndedEvent):
    logger.info("Call session ended")
    # No response needed as the call is ending

Handling AI Component Events

You can also listen to events from AI components:

# Listen to speech-to-text events
@agent.events.subscribe
async def handle_transcript(event: STTTranscriptEvent):
    logger.info(f"User said: {event.text}")
    
    # You can add custom logic here, like:
    # - Logging conversations
    # - Triggering specific responses
    # - Analyzing sentiment

# Listen to LLM response events
@agent.events.subscribe
async def handle_llm_response(event: LLMResponseEvent):
    logger.info(f"Agent responding: {event.text}")

# Listen to TTS events
@agent.events.subscribe
async def handle_tts_audio(event: TTSAudioEvent):
    logger.info("Agent is speaking...")

Advanced Event Handling

For more sophisticated event handling, you can:

1. Handle Multiple Event Types

from typing import Union
from vision_agents.core.events import CallSessionParticipantJoinedEvent, CallSessionParticipantLeftEvent

@agent.events.subscribe
async def handle_participant_changes(event: Union[CallSessionParticipantJoinedEvent, CallSessionParticipantLeftEvent]):
    if isinstance(event, CallSessionParticipantJoinedEvent):
        logger.info(f"Participant joined: {event.participant.user.name}")
    else:
        logger.info(f"Participant left: {event.participant.user.name}")

2. Add Event Filtering

@agent.events.subscribe
async def handle_transcript(event: STTTranscriptEvent):
    # Only respond to final transcripts with high confidence
    if event.is_final and event.confidence > 0.8:
        logger.info(f"High confidence transcript: {event.text}")
        
        # Add custom logic based on transcript content
        if "help" in event.text.lower():
            await agent.simple_response("I'm here to help! What do you need?")

3. Error Handling

@agent.events.subscribe
async def handle_errors(event: PluginErrorEvent):
    logger.error(f"Plugin error: {event.error_message}")
    
    # Notify users of errors
    if event.is_fatal:
        await agent.simple_response("I'm experiencing technical difficulties. Please try again.")

Complete Example

Here’s the complete enhanced agent with multiple event handlers:

import asyncio
import logging
from uuid import uuid4
from typing import Union

from dotenv import load_dotenv
from vision_agents.core.edge.types import User
from vision_agents.plugins import elevenlabs, deepgram, openai, getstream
from vision_agents.core import agents, cli
from vision_agents.core.events import (
    CallSessionParticipantJoinedEvent,
    CallSessionParticipantLeftEvent,
    CallSessionStartedEvent,
    CallSessionEndedEvent,
    STTTranscriptEvent,
    LLMResponseEvent,
    PluginErrorEvent
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
load_dotenv()

async def start_agent():
    call_id = str(uuid4())
    
    # Create the agent
    agent = agents.Agent(
        edge=getstream.Edge(),
        agent_user=User(name="My AI Assistant", id="agent"),
        instructions="You're a helpful voice AI assistant. Keep responses short and conversational.",
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(),
    )
    
    await agent.create_user()
    
    # Event handlers
    @agent.events.subscribe
    async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
        if event.participant.user.id == "agent":
            return
        logger.info(f"New participant: {event.participant.user.name}")
        await agent.simple_response(f"Hello {event.participant.user.name}! Welcome!")
    
    @agent.events.subscribe
    async def handle_participant_left(event: CallSessionParticipantLeftEvent):
        if event.participant.user.id == "agent":
            return
        logger.info(f"Participant left: {event.participant.user.name}")
        await agent.simple_response(f"Goodbye {event.participant.user.name}!")
    
    @agent.events.subscribe
    async def handle_session_started(event: CallSessionStartedEvent):
        logger.info("Call session started")
        await agent.simple_response("The call has started. I'm here to help!")
    
    @agent.events.subscribe
    async def handle_transcript(event: STTTranscriptEvent):
        if event.is_final and event.confidence > 0.8:
            logger.info(f"User said: {event.text}")
    
    @agent.events.subscribe
    async def handle_llm_response(event: LLMResponseEvent):
        logger.info(f"Agent responding: {event.text}")
    
    @agent.events.subscribe
    async def handle_errors(event: PluginErrorEvent):
        logger.error(f"Plugin error: {event.error_message}")
        if event.is_fatal:
            await agent.simple_response("I'm experiencing technical difficulties.")
    
    # Create and join the call
    call = agent.edge.client.video.call("default", call_id)
    await agent.edge.open_demo(call)
    
    with await agent.join(call):
        await agent.finish()

if __name__ == "__main__":
    asyncio.run(cli.start_dispatcher(start_agent))

Best Practices

1. Always Check Event Source

@agent.events.subscribe
async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
    # Always check if the event is from the agent itself
    if event.participant.user.id == "agent":
        return
    # ... handle the event

2. Use Appropriate Logging

@agent.events.subscribe
async def handle_transcript(event: STTTranscriptEvent):
    # Log important events for debugging
    logger.info(f"Transcript: {event.text} (confidence: {event.confidence})")

3. Handle Errors Gracefully

@agent.events.subscribe
async def handle_errors(event: PluginErrorEvent):
    # Log errors and provide user feedback
    logger.error(f"Error in {event.plugin_name}: {event.error_message}")
    if event.is_fatal:
        await agent.simple_response("I'm having technical issues. Please try again.")

4. Keep Handlers Focused

# Good: Single responsibility
@agent.events.subscribe
async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
    await agent.simple_response(f"Hello {event.participant.user.name}!")

# Avoid: Multiple responsibilities
@agent.events.subscribe
async def handle_everything(event):
    # Don't handle all events in one function
    pass

Next Steps

The event system provides a powerful way to create responsive, interactive AI agents. By understanding how to listen to and respond to events, you can build sophisticated applications that react intelligently to user interactions and system changes.

How-to Guides

​Prerequisites

​Example: Greeting New Participants

​Starting with the Basic Agent

​Adding Event Handling

​Understanding the Event Handler

​Adding More Event Handlers

​Handling AI Component Events

​Advanced Event Handling

​1. Handle Multiple Event Types

​2. Add Event Filtering

​3. Error Handling

​Complete Example

​Best Practices

​1. Always Check Event Source

​2. Use Appropriate Logging

​3. Handle Errors Gracefully

​4. Keep Handlers Focused

​Next Steps