Build a Voice Agent for Accents & Dialects

1python -m venv myenv
2source myenv/bin/activate  # On Windows use `myenv\\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 a 'voice agent for handling different accents and dialects', designed to assist users with diverse linguistic backgrounds. Your primary role is to facilitate seamless communication by accurately understanding and responding to queries in various accents and dialects.\n\n**Persona:**\n- You are a friendly and patient communication assistant, dedicated to bridging language gaps and ensuring users feel understood and supported.\n\n**Capabilities:**\n- Accurately recognize and interpret a wide range of accents and dialects.\n- Provide clear and concise responses to user queries.\n- Adapt to different speech patterns and linguistic nuances.\n- Offer suggestions for improving communication clarity if needed.\n\n**Constraints and Limitations:**\n- You are not a language expert and should not provide language learning advice.\n- Always include a disclaimer that users should consult a language professional for detailed linguistic assistance.\n- You must respect user privacy and confidentiality at all times.\n- Avoid making assumptions about a user's background based on their accent or dialect."
14
15class MyVoiceAgent(Agent):
16    def __init__(self):
17        super().__init__(instructions=agent_instructions)
18    async def on_enter(self): await self.session.say("Hello! How can I help?")
19    async def on_exit(self): await self.session.say("Goodbye!")
20
21async def start_session(context: JobContext):
22    # Create agent and conversation flow
23    agent = MyVoiceAgent()
24    conversation_flow = ConversationFlow(agent)
25
26    # Create pipeline
27    pipeline = CascadingPipeline(
28        stt=DeepgramSTT(model="nova-2", language="en"),
29        llm=OpenAILLM(model="gpt-4o"),
30        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
31        vad=SileroVAD(threshold=0.35),
32        turn_detector=TurnDetector(threshold=0.8)
33    )
34
35    session = AgentSession(
36        agent=agent,
37        pipeline=pipeline,
38        conversation_flow=conversation_flow
39    )
40
41    try:
42        await context.connect()
43        await session.start()
44        # Keep the session running until manually terminated
45        await asyncio.Event().wait()
46    finally:
47        # Clean up resources when done
48        await session.close()
49        await context.shutdown()
50
51def make_context() -> JobContext:
52    room_options = RoomOptions(
53    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
54        name="VideoSDK Cascaded Agent",
55        playground=True
56    )
57
58    return JobContext(room_options=room_options)
59
60if __name__ == "__main__":
61    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
62    job.start()
63

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

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

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
10if __name__ == "__main__":
11    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
12    job.start()
13

1python main.py
2