Real-Time AI Voice Agents Using VideoSDK and Deepgram

1├── agent/                 # Voice agent implementation (center of the diagram)
2│   ├── agent.py           # Main agent class for meeting handling
3│   └── audio_stream_track.py  # Custom audio track implementation (bottom of diagram)
4├── client/                # React frontend for user interaction (left side of diagram)
5├── intelligence/          # LLM integration for response generation (middle-bottom of diagram)
6├── stt/                   # Speech-to-text implementation using Deepgram (top-right of diagram)
7├── tts/                   # Text-to-speech implementation using Deepgram (bottom-right of diagram)
8└── main.py                # Main application entry point tying everything together
9

1function App() {
2  const [meetingId, setMeetingId] = React.useState<string | null>(null);
3  const [isLoading, setIsLoading] = React.useState(false);
4  const [useVideo, setUseVideo] = React.useState(false);
5
6  const createMeeting = async () => {
7    try {
8      setIsLoading(true);
9      const response = await fetch("https://api.videosdk.live/v2/rooms", {
10        method: "POST",
11        headers: {
12          Authorization: import.meta.env.VITE_APP_AUTH_TOKEN,
13          "Content-Type": "application/json",
14        },
15      });
16
17      const { roomId } = await response.json();
18      setMeetingId(roomId);
19    } catch (error) {
20      toast.error("Failed to create meeting. Please try again.");
21      console.error("Error creating meeting:", error);
22    } finally {
23      setIsLoading(false);
24    }
25  };
26
27  // Rest of the component implementation...
28}
29

1class DeepgramSTT(STT):
2    def __init__(
3        self,
4        loop: AbstractEventLoop,
5        api_key,
6        language,
7        intelligence: Intelligence
8    ) -> None:
9        # Initialize Deepgram client
10        self.deepgram_client = DeepgramClient(
11            api_key=api_key,
12            config=DeepgramClientOptions(options={"keepalive": True}),
13        )
14        # Configure STT parameters
15        self.model = "nova-2"
16        self.language = language
17        self.intelligence = intelligence
18        # Other initialization...
19
20    def start(self, peer_id: str, peer_name: str, stream: Stream):
21        # Set up Deepgram connection for a participant
22        deepgram_options = LiveOptions(
23            model=self.model,
24            language=self.language,
25            smart_format=True,
26            encoding="linear16",
27            channels=2,
28            sample_rate=48000,
29            interim_results=True,
30            vad_events=True,
31            # Other options...
32        )
33        # Register event handlers and start connection
34        # ...
35

1class OpenAIIntelligence(Intelligence):
2    def __init__(self, api_key: str, tts: TTS, base_url: Optional[str]="https://api.openai.com/v1", model: Optional[str] = None, system_prompt: Optional[str] = None):
3        self.client = OpenAI(
4            base_url=base_url,
5            api_key=api_key,
6            max_retries=3,
7        ).chat
8        
9        self.tts = tts
10        self.system_prompt = "You are AI Interviewer and you are interviewing a candidate for a software engineering position."
11        self.chat_history = []
12        self.model = model or "gpt-3.5-turbo"
13        self.system_prompt = system_prompt or "You are AI Interviewer and you are interviewing a candidate for a software engineering position."
14
15    def generate(self, text: str, sender_name: str):
16        # Build conversation history
17        messages = self.build_messages(text, sender_name=sender_name)
18        
19        # Generate LLM completion
20        response = self.client.completions.create(
21            model=self.model,
22            messages=messages,
23            max_tokens=100,
24            temperature=0.5,
25            stream=False
26        )
27        
28        response_text = response.choices[0].message.content
29        # Send response to TTS
30        self.tts.generate(text=response_text)
31        
32        # Publish message to the meeting
33        if self.pubsub is not None:
34            self.pubsub(message=f"[Interviewer]: {response_text}")
35        
36        # Add response to history
37        self.add_response(response_text)
38

1class DeepgramTTS(TTS):
2    def __init__(self, api_key: str, output_track: MediaStreamTrack):
3        base_url = f"wss://api.deepgram.com/v1/speak?encoding=linear16&sample_rate=24000&&model=aura-stella-en"
4        self.output_track = output_track
5        self.api_key = api_key
6        
7        # Initialize WebSocket connection
8        self._socket = connect(
9            base_url, additional_headers={"Authorization": f"Token {self.api_key}"}
10        )
11        
12        # Set up receiver thread to handle incoming audio data
13        # ...
14
15    def generate(self, text: Union[str, Iterator[str]]):
16        if self._socket is None:
17            print("WebSocket is not connected.")
18            return
19
20        if isinstance(text, str):
21            print(f"Sending: {text}")
22            self._socket.send(json.dumps({"type": "Speak", "text": text}))
23        elif isinstance(text, Iterator):
24            for t in text:
25                print(f"Sending: {t}")
26                self._socket.send(json.dumps({"type": "Speak", "text": t}))
27

1class CustomAudioStreamTrack(AudioStreamTrack):
2    def __init__(
3        self, loop, handle_interruption: Optional[bool] = True,
4    ):
5        super().__init__()
6        self.loop = loop
7        # Initialize audio properties
8        self.sample_rate = 24000
9        self.channels = 1
10        self.sample_width = 2
11        self.time_base_fraction = Fraction(1, self.sample_rate)
12        self.samples = int(AUDIO_PTIME * self.sample_rate)
13        self.chunk_size = int(self.samples * self.channels * self.sample_width)
14        
15        # Set up frame buffer and processing thread
16        self.frame_buffer = []
17        self.audio_data_buffer = bytearray()
18        self._process_audio_task_queue = asyncio.Queue()
19        self._process_audio_thread = threading.Thread(target=self.process_incoming_audio)
20        self._process_audio_thread.daemon = True
21        self._process_audio_thread.start()
22
23    def add_new_bytes(self, bytes: Iterator[bytes]):
24        self._process_audio_task_queue.put_nowait(bytes)
25
26    async def recv(self) -> AudioFrame:
27        # Implement audio frame delivery for VideoSDK
28        # ...
29

1async def run():
2    try:
3        print("Loading Interviewer...")
4        # Create audio track
5        audio_track = CustomAudioStreamTrack(
6            loop=loop,
7            handle_interruption=True,
8        )
9
10        # Set up TTS client
11        tts_client = DeepgramTTS(
12            api_key=stt_api_key,
13            output_track=audio_track,
14        )
15        
16        # Set up intelligence client
17        intelligence_client = OpenAIIntelligence( 
18            api_key = llm_api_key, 
19            model="gpt-4o", 
20            tts=tts_client,
21            system_prompt=(
22                "You are an AI Copilot in a meeting. Keep responses short and engaging. "
23                "Start by introducing yourself: 'Hi, I'm your AI Copilot!'. "
24                "If asked, explain how you work briefly: 'I process speech using Deepgram STT and reply using OpenAI LLM.' "
25                "Keep the conversation interactive and avoid long responses unless explicitly requested."
26                )
27        )
28
29        # Set up STT client
30        stt_client = DeepgramSTT(
31            loop=loop,
32            api_key=stt_api_key,
33            language=language,
34            intelligence=intelligence_client
35        )
36
37        # Create and join the meeting
38        interviewer = AIInterviewer(loop=loop, audio_track=audio_track, stt=stt_client, intelligence=intelligence_client)
39        await interviewer.join(meeting_id=room_id, token=auth_token)
40
41    except Exception as e:
42        traceback.print_exc()
43        print("error while joining", e)
44