Engineering Production AI Voice Agents: Low-Latency STT, Tool Calling & TTS Avatars

import { WebSocket } from "ws";
import { GoogleGenAI } from "@google/genai";
import { createClient, LiveTranscriptionEvents } from "@deepgram/sdk";

export function handleVoiceSession(clientWs: WebSocket) {
  const deepgram = createClient(process.env.DEEPGRAM_API_KEY!);
  const dgConnection = deepgram.listen.live({
    model: "nova-2",
    language: "en-US",
    smart_format: true,
    interim_results: false,
  });

  const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY });

  // 1. Pipe client mic stream directly into Deepgram STT
  clientWs.on("message", (audioData: Buffer) => {
    if (dgConnection.getReadyState() === 1) {
      dgConnection.send(audioData);
    }
  });

  // 2. On finalized speech transcript, trigger streaming AI inference
  dgConnection.on(LiveTranscriptionEvents.Transcript, async (data) => {
    const transcript = data.channel.alternatives[0]?.transcript;
    if (!transcript || transcript.trim().length === 0) return;

    // Send barge-in signal to interrupt any currently playing audio
    clientWs.send(JSON.stringify({ type: "INTERRUPT_PLAYBACK" }));

    // Stream response from Gemini 2.5 Flash
    const responseStream = await ai.models.generateContentStream({
      model: "gemini-2.5-flash",
      contents: [{ role: "user", parts: [{ text: transcript }] }],
      config: {
        systemInstruction: "You are a concise, helpful voice assistant. Keep answers brief and conversational.",
      },
    });

    let sentenceBuffer = "";
    for await (const chunk of responseStream) {
      const text = chunk.text || "";
      sentenceBuffer += text;

      // As soon as a sentence boundary is reached, trigger TTS chunk
      if (/[.?!]\s$/.test(sentenceBuffer)) {
        await synthesizeAndStreamAudio(sentenceBuffer, clientWs);
        sentenceBuffer = "";
      }
    }
    
    if (sentenceBuffer.trim()) {
      await synthesizeAndStreamAudio(sentenceBuffer, clientWs);
    }
  });
}