Build an AI Voice Agent for BPO with VideoSDK

1sequenceDiagram
2    participant User
3    participant AgentSession
4    participant CascadingPipeline
5    participant DeepgramSTT
6    participant OpenAILLM
7    participant ElevenLabsTTS
8    participant VAD
9    participant TurnDetector
10    User->>AgentSession: Speaks
11    AgentSession->>VAD: Detects speech activity
12    VAD->>TurnDetector: Detects turn end
13    TurnDetector->>DeepgramSTT: Sends audio for transcription
14    DeepgramSTT->>OpenAILLM: Sends transcript
15    OpenAILLM->>ElevenLabsTTS: Gets response
16    ElevenLabsTTS->>AgentSession: Plays audio response
17    AgentSession->>User: Responds
18

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 efficient and professional AI Voice Agent designed specifically for Business Process Outsourcing (BPO) environments. Your persona is that of a courteous, knowledgeable, and patient customer service representative. Your primary capabilities include: answering customer queries related to products or services, handling basic troubleshooting, processing simple transactions, escalating complex issues to human agents, and providing information about company policies and procedures. You must always maintain a polite and empathetic tone, ensure customer data privacy, and strictly adhere to provided scripts and compliance guidelines. You are not authorized to make decisions outside predefined protocols, provide personal opinions, or handle sensitive financial or legal matters. Always inform the customer when you are escalating their issue to a human agent. If you are unsure or unable to assist, politely suggest that a human representative will follow up. Never collect or store sensitive personal information beyond what is explicitly permitted by company policy."
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 \
2  -H "Authorization: YOUR_VIDEOSDK_API_KEY" \
3  -H "Content-Type: application/json" \
4  -d '{"region":"sg001"}' \
5  https://api.videosdk.live/v2/rooms
6

1agent_instructions = "You are an efficient and professional AI Voice Agent designed specifically for Business Process Outsourcing (BPO) environments. ..."
2
3class MyVoiceAgent(Agent):
4    def __init__(self):
5        super().__init__(instructions=agent_instructions)
6    async def on_enter(self):
7        await self.session.say("Hello! How can I help?")
8    async def on_exit(self):
9        await self.session.say("Goodbye!")
10

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    agent = MyVoiceAgent()
3    conversation_flow = ConversationFlow(agent)
4    pipeline = CascadingPipeline(...)
5    session = AgentSession(agent=agent, pipeline=pipeline, conversation_flow=conversation_flow)
6    try:
7        await context.connect()
8        await session.start()
9        await asyncio.Event().wait()
10    finally:
11        await session.close()
12        await context.shutdown()
13
14def make_context() -> JobContext:
15    room_options = RoomOptions(
16        name="VideoSDK Cascaded Agent",
17        playground=True
18    )
19    return JobContext(room_options=room_options)
20
21if __name__ == "__main__":
22    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
23    job.start()
24