Build an AI Voice Agent with Open Source STT

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 providing information and guidance on open source speech-to-text (STT) engines. Your persona is that of a knowledgeable and friendly tech assistant. Your primary capabilities include:
14
151. Explaining the features and benefits of various open source STT engines.
162. Providing guidance on how to integrate these engines into different applications.
173. Offering troubleshooting tips for common issues encountered with open source STT engines.
184. Recommending resources and documentation for further learning.
19
20Constraints and limitations:
21
221. You are not a software developer, so you cannot provide custom code solutions.
232. You must always encourage users to verify information with official documentation or consult with a professional for complex integration tasks.
243. You cannot endorse any specific product or service; your role is to provide unbiased information."
25
26class MyVoiceAgent(Agent):
27    def __init__(self):
28        super().__init__(instructions=agent_instructions)
29    async def on_enter(self): await self.session.say("Hello! How can I help?")
30    async def on_exit(self): await self.session.say("Goodbye!")
31
32async def start_session(context: JobContext):
33    # Create agent and conversation flow
34    agent = MyVoiceAgent()
35    conversation_flow = ConversationFlow(agent)
36
37    # Create pipeline
38    pipeline = CascadingPipeline(
39        stt=DeepgramSTT(model="nova-2", language="en"),
40        llm=OpenAILLM(model="gpt-4o"),
41        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
42        vad=[Silero Voice Activity Detection](https://docs.videosdk.live/ai_agents/plugins/silero-vad)(threshold=0.35),
43        turn_detector=[Turn detector for AI voice Agents](https://docs.videosdk.live/ai_agents/plugins/turn-detector)(threshold=0.8)
44    )
45
46    session = [AI voice Agent Sessions](https://docs.videosdk.live/ai_agents/core-components/agent-session)(
47        agent=agent,
48        pipeline=pipeline,
49        conversation_flow=conversation_flow
50    )
51
52    try:
53        await context.connect()
54        await session.start()
55        # Keep the session running until manually terminated
56        await asyncio.Event().wait()
57    finally:
58        # Clean up resources when done
59        await session.close()
60        await context.shutdown()
61
62def make_context() -> JobContext:
63    room_options = RoomOptions(
64    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
65        name="VideoSDK Cascaded Agent",
66        playground=True
67    )
68
69    return JobContext(room_options=room_options)
70
71if __name__ == "__main__":
72    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
73    job.start()
74

1curl -X POST "https://api.videosdk.live/v1/rooms" \
2-H "Authorization: Bearer YOUR_VIDEOSDK_API_KEY" \
3-H "Content-Type: application/json" \
4-d '{"name": "My Meeting Room"}'
5

1python main.py
2