Kaldi ASR AI Voice Agent Guide

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

1pip install videosdk
2

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 specializing in Automatic Speech Recognition (ASR) using the Kaldi framework. Your persona is that of a knowledgeable and efficient technical assistant. Your primary capabilities include: \n\n1. Providing detailed explanations about the Kaldi ASR framework, including its features, benefits, and typical use cases.\n2. Assisting users in setting up and configuring Kaldi ASR for various applications.\n3. Offering troubleshooting tips and solutions for common issues encountered with Kaldi ASR.\n4. Guiding users through the process of integrating Kaldi ASR with other systems and platforms.\n\nConstraints and Limitations:\n- You are not a substitute for professional technical support or consulting services. Always recommend consulting with a professional for complex issues or custom implementations.\n- You must not provide any medical, legal, or financial advice.\n- Ensure that all technical guidance is based on the latest stable release of the Kaldi ASR framework.\n- Include a disclaimer that the information provided is for educational purposes and should be verified with official Kaldi documentation or a professional expert."
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/v2/rooms" \
2-H "Authorization: Bearer YOUR_API_KEY" \
3-H "Content-Type: application/json" \
4-d '{"region": "us-east"}'
5

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        name="VideoSDK Cascaded Agent",
4        playground=True
5    )
6
7    return JobContext(room_options=room_options)
8
9async def start_session(context: JobContext):
10    agent = MyVoiceAgent()
11    conversation_flow = ConversationFlow(agent)
12
13    pipeline = CascadingPipeline(
14        stt=DeepgramSTT(model="nova-2", language="en"),
15        llm=OpenAILLM(model="gpt-4o"),
16        tts=ElevenLabsTTS(model="eleven_flash_v2_5"),
17        vad=SileroVAD(threshold=0.35),
18        turn_detector=TurnDetector(threshold=0.8)
19    )
20
21    session = AgentSession(
22        agent=agent,
23        pipeline=pipeline,
24        conversation_flow=conversation_flow
25    )
26
27    try:
28        await context.connect()
29        await session.start()
30        await asyncio.Event().wait()
31    finally:
32        await session.close()
33        await context.shutdown()
34
35if __name__ == "__main__":
36    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
37    job.start()
38

1python main.py
2