Build AI Voice Agents with Google Dialogflow

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

1pip install videosdk google-cloud-dialogflow
2

1VIDEOSDK_API_KEY=your_videosdk_api_key
2DIALOGFLOW_API_KEY=your_dialogflow_api_key
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 = "{\n  \"persona\": \"helpful virtual assistant\",\n  \"capabilities\": [\n    \"integrate with Google Dialogflow to understand and process natural language queries\",\n    \"provide information and assistance on a wide range of topics\",\n    \"handle user queries efficiently and escalate to human agents if necessary\",\n    \"support multi-turn conversations and context management\"\n  ],\n  \"constraints\": [\n    \"you are not a human and should not provide personal opinions\",\n    \"you must include a disclaimer that complex queries may require human intervention\",\n    \"ensure user privacy and data protection at all times\"\n  ]\n}"
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

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

1    try:
2        await context.connect()
3        await session.start()
4        # Keep the session running until manually terminated
5        await asyncio.Event().wait()
6    finally:
7        # Clean up resources when done
8        await session.close()
9        await context.shutdown()
10
11def make_context() -> JobContext:
12    room_options = RoomOptions(
13    #  room_id="YOUR_MEETING_ID",  # Set to join a pre-created room; omit to auto-create
14        name="VideoSDK Cascaded Agent",
15        playground=True
16    )
17
18    return JobContext(room_options=room_options)
19
20if __name__ == "__main__":
21    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
22    job.start()
23

1python main.py
2