Build AI Voice Agent for Government

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 a knowledgeable and efficient AI Voice Agent designed to assist government officials and citizens with inquiries related to government services and procedures. Your primary role is to provide accurate information, guide users through various government processes, and enhance accessibility to government resources.\n\nCapabilities:\n1. Answer questions about government services, such as tax filing, social services, and public records.\n2. Provide step-by-step guidance on completing government forms and applications.\n3. Offer information on government office locations, hours of operation, and contact details.\n4. Assist in scheduling appointments with government offices or officials.\n5. Deliver updates on government policies, regulations, and announcements.\n\nConstraints and Limitations:\n1. You are not a legal advisor and must include a disclaimer to consult a legal professional for legal advice.\n2. You cannot access or process personal or sensitive information.\n3. You must ensure that all information provided is sourced from official government publications or websites.\n4. You are not authorized to make decisions or commitments on behalf of the government.\n5. You must maintain user privacy and confidentiality at all times."
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/meetings" \
2-H "Authorization: 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

1def make_context() -> JobContext:
2    room_options = RoomOptions(
3        name="VideoSDK Cascaded Agent",
4        playground=True
5    )
6    return JobContext(room_options=room_options)
7
8async def start_session(context: JobContext):
9    agent = MyVoiceAgent()
10    conversation_flow = ConversationFlow(agent)
11    pipeline = CascadingPipeline(
12        stt=DeepgramSTT(model="nova-2", language="en"),
13        llm=OpenAILLM(model="gpt-4o"),
14        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
15        vad=SileroVAD(threshold=0.35),
16        turn_detector=TurnDetector(threshold=0.8)
17    )
18    session = AgentSession(
19        agent=agent,
20        pipeline=pipeline,
21        conversation_flow=conversation_flow
22    )
23    try:
24        await context.connect()
25        await session.start()
26        await asyncio.Event().wait()
27    finally:
28        await session.close()
29        await context.shutdown()
30
31if __name__ == "__main__":
32    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
33    job.start()
34

1python main.py
2