Function Calling with LLMs: AI Voice Agent Guide

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

1pip install videosdk-agents videosdk-plugins-silero videosdk-plugins-turn_detector videosdk-plugins-deepgram videosdk-plugins-openai videosdk-plugins-elevenlabs
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 AI Voice Agent specializing in 'function calling with LLMs' (Large Language Models). Your primary role is to assist developers and tech enthusiasts in understanding and implementing function calling capabilities using LLMs. \n\nCapabilities:\n1. Explain the concept of function calling with LLMs and its applications.\n2. Provide step-by-step guidance on setting up function calls within LLM frameworks.\n3. Offer examples of code snippets and best practices for efficient function calling.\n4. Answer frequently asked questions related to function calling with LLMs.\n\nConstraints:\n1. You are not a substitute for professional software engineering advice and should encourage users to consult documentation or experts for complex issues.\n2. You must not provide any proprietary or confidential information.\n3. Ensure that all examples and explanations are clear, concise, and suitable for a general audience with basic programming knowledge."
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  https://api.videosdk.live/v1/meetings \
3  -H "Authorization: YOUR_API_KEY" \
4  -H "Content-Type: application/json" \
5  -d '{}'
6

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    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
4        name="VideoSDK Cascaded Agent",
5        playground=True
6    )
7
8    return JobContext(room_options=room_options)
9
10async def start_session(context: JobContext):
11    # Create agent and conversation flow
12    agent = MyVoiceAgent()
13    conversation_flow = ConversationFlow(agent)
14
15    # Create pipeline
16    pipeline = CascadingPipeline(
17        stt=DeepgramSTT(model="nova-2", language="en"),
18        llm=OpenAILLM(model="gpt-4o"),
19        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
20        vad=SileroVAD(threshold=0.35),
21        turn_detector=TurnDetector(threshold=0.8)
22    )
23
24    session = AgentSession(
25        agent=agent,
26        pipeline=pipeline,
27        conversation_flow=conversation_flow
28    )
29
30    try:
31        await context.connect()
32        await session.start()
33        # Keep the session running until manually terminated
34        await asyncio.Event().wait()
35    finally:
36        # Clean up resources when done
37        await session.close()
38        await context.shutdown()
39
40if __name__ == "__main__":
41    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
42    job.start()
43

1python main.py
2