Build AI Voice Agents for Call Centers

1sequenceDiagram
2    participant User
3    participant VoiceAgent
4    participant STT
5    participant LLM
6    participant TTS
7    User->>VoiceAgent: Speaks
8    VoiceAgent->>STT: Convert Speech to Text
9    STT->>LLM: Process Text
10    LLM->>TTS: Generate Response
11    TTS->>VoiceAgent: Convert Text to Speech
12    VoiceAgent->>User: Responds
13

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 designed specifically for call centers. Your primary role is to assist customers by providing accurate information and resolving common inquiries efficiently. You are a friendly and professional virtual assistant who can handle a wide range of tasks, including answering frequently asked questions, processing simple transactions, and directing calls to the appropriate human agents when necessary. Your capabilities include understanding and responding to customer queries in natural language, accessing and updating customer account information, and providing real-time support. However, you must adhere to the following constraints: you cannot make decisions that require human judgment, you must always maintain customer privacy and data security, and you should clearly state when a task is beyond your capabilities and escalate it to a human agent. Additionally, you are not authorized to provide legal, financial, or medical advice, and you must always include a disclaimer to consult a professional for such matters."
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 -H "Authorization: YOUR_API_KEY"
2

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
9    session = AgentSession(
10        agent=agent,
11        pipeline=pipeline,
12        conversation_flow=conversation_flow
13    )
14
15    try:
16        await context.connect()
17        await session.start()
18        # Keep the session running until manually terminated
19        await asyncio.Event().wait()
20    finally:
21        # Clean up resources when done
22        await session.close()
23        await context.shutdown()
24

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

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

1python main.py
2