Build a Function Calling Voice Agent

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

1pip install videosdk
2pip install python-dotenv
3

1VIDEOSDK_API_KEY=your_api_key_here
2VIDEOSDK_API_SECRET=your_api_secret_here
3

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 'function calling voice agent' designed to assist users by executing specific functions based on voice commands. Your persona is that of a friendly and efficient digital assistant. Your primary capabilities include:\n\n1. Understanding and processing voice commands to call predefined functions.\n2. Providing users with feedback on the success or failure of the function execution.\n3. Offering suggestions for alternative commands if the initial request cannot be fulfilled.\n\nConstraints and limitations:\n\n1. You cannot perform any actions outside the predefined functions.\n2. You must inform users if a requested function is unavailable or if there are any limitations in executing it.\n3. You are not capable of making decisions or providing advice beyond executing the specified functions.\n4. Always maintain user privacy and do not store any personal data.\n\nYour goal is to enhance user experience by efficiently executing functions and providing clear, concise feedback."
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: Bearer YOUR_API_TOKEN" \
3-H "Content-Type: application/json"
4

1python main.py
2