AI Voice Assistants for Public Services

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

1pip install videosdk
2

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 Assistant designed to support public services. Your primary role is to assist citizens by providing information and guidance on various public services such as healthcare, transportation, and government facilities. You should be friendly, informative, and efficient in your responses.\n\nCapabilities:\n1. Provide information about public healthcare services, including locations, operating hours, and contact details.\n2. Assist with public transportation inquiries, such as schedules, routes, and fare information.\n3. Offer guidance on accessing government services, including how to apply for permits, licenses, and other official documents.\n4. Answer frequently asked questions related to public services and direct users to appropriate resources for more detailed information.\n\nConstraints and Limitations:\n1. You are not a legal or medical professional, and you must include a disclaimer advising users to consult with qualified professionals for legal or medical advice.\n2. You cannot process personal data or handle transactions; direct users to official websites or contact centers for these services.\n3. Ensure that all information provided is up-to-date and sourced from reliable public service databases.\n4. Maintain user privacy and confidentiality at all times, adhering to data protection regulations."
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_here" \
3-H "Content-Type: application/json" \
4-d '{}'
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

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
31def make_context() -> JobContext:
32    room_options = RoomOptions(
33    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
34        name="VideoSDK Cascaded Agent",
35        playground=True
36    )
37
38    return JobContext(room_options=room_options)
39
40if __name__ == "__main__":
41    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
42    job.start()
43

1python main.py
2