AI Voice Agent WebSocket Disconnect Guide

1python -m venv venv
2source venv/bin/activate  # On Windows use `venv\Scripts\activate`
3

1pip install videosdk
2pip install python-dotenv
3

1VIDEOSDK_API_KEY=your_api_key_here
2

1import asyncio, os
2from videosdk.agents import Agent, AgentSession, CascadingPipeline, JobContext, RoomOptions, WorkerJob, ConversationFlow
3from videosdk.plugins.silero import SileroVAD
4from videosdk.plugins.turn_detector import TurnDetector, pre_download_model
5from videosdk.plugins.deepgram import DeepgramSTT
6from videosdk.plugins.openai import OpenAILLM
7from videosdk.plugins.elevenlabs import ElevenLabsTTS
8from typing import AsyncIterator
9
10# Pre-downloading the Turn Detector model
11pre_download_model()
12
13agent_instructions = "You are an AI Voice Agent specializing in managing WebSocket connections, particularly focusing on handling disconnections. Your persona is that of a technical support assistant for developers integrating AI voice capabilities into their applications. Your primary capabilities include:
14
151. Providing guidance on how to handle WebSocket disconnections effectively.
162. Offering troubleshooting steps for common WebSocket issues.
173. Explaining best practices for maintaining stable WebSocket connections in AI voice applications.
184. Assisting with code examples and implementation strategies related to WebSocket management.
19
20Constraints and limitations:
21- You are not a certified network engineer, and your advice should be considered as guidance rather than professional consultation.
22- Always recommend consulting official WebSocket documentation or a network specialist for complex issues.
23- You cannot execute code or directly interact with WebSocket connections; your role is purely advisory."
24
25class MyVoiceAgent(Agent):
26    def __init__(self):
27        super().__init__(instructions=agent_instructions)
28    async def on_enter(self): await self.session.say("Hello! How can I help?")
29    async def on_exit(self): await self.session.say("Goodbye!")
30
31async def start_session(context: JobContext):
32    # Create agent and conversation flow
33    agent = MyVoiceAgent()
34    conversation_flow = ConversationFlow(agent)
35
36    # Create pipeline
37    pipeline = CascadingPipeline(
38        stt=DeepgramSTT(model="nova-2", language="en"),
39        llm=OpenAILLM(model="gpt-4o"),
40        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
41        vad=SileroVAD(threshold=0.35),
42        turn_detector=TurnDetector(threshold=0.8)
43    )
44
45    session = AgentSession(
46        agent=agent,
47        pipeline=pipeline,
48        conversation_flow=conversation_flow
49    )
50
51    try:
52        await context.connect()
53        await session.start()
54        # Keep the session running until manually terminated
55        await asyncio.Event().wait()
56    finally:
57        # Clean up resources when done
58        await session.close()
59        await context.shutdown()
60
61def make_context() -> JobContext:
62    room_options = RoomOptions(
63    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
64        name="VideoSDK Cascaded Agent",
65        playground=True
66    )
67
68    return JobContext(room_options=room_options)
69
70if __name__ == "__main__":
71    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
72    job.start()
73

1curl -X POST "https://api.videosdk.live/v1/meetings" \
2-H "Authorization: Bearer YOUR_API_KEY" \
3-H "Content-Type: application/json"
4

1class MyVoiceAgent(Agent):
2    def __init__(self):
3        super().__init__(instructions=agent_instructions)
4    async def on_enter(self): await self.session.say("Hello! How can I help?")
5    async def on_exit(self): await self.session.say("Goodbye!")
6

1pipeline = CascadingPipeline(
2    stt=DeepgramSTT(model="nova-2", language="en"),
3    llm=OpenAILLM(model="gpt-4o"),
4    tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
5    vad=SileroVAD(threshold=0.35),
6    turn_detector=TurnDetector(threshold=0.8)
7)
8

1async def start_session(context: JobContext):
2    # Create agent and conversation flow
3    agent = MyVoiceAgent()
4    conversation_flow = ConversationFlow(agent)
5
6    # Create pipeline
7    pipeline = CascadingPipeline(
8        stt=DeepgramSTT(model="nova-2", language="en"),
9        llm=OpenAILLM(model="gpt-4o"),
10        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
11        vad=SileroVAD(threshold=0.35),
12        turn_detector=TurnDetector(threshold=0.8)
13    )
14
15    session = AgentSession(
16        agent=agent,
17        pipeline=pipeline,
18        conversation_flow=conversation_flow
19    )
20
21    try:
22        await context.connect()
23        await session.start()
24        # Keep the session running until manually terminated
25        await asyncio.Event().wait()
26    finally:
27        # Clean up resources when done
28        await session.close()
29        await context.shutdown()
30

1def make_context() -> JobContext:
2    room_options = RoomOptions(
3    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
4        name="VideoSDK Cascaded Agent",
5        playground=True
6    )
7
8    return JobContext(room_options=room_options)
9

1if __name__ == "__main__":
2    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
3    job.start()
4

1python main.py
2