fix(deepgram): keep streaming audio while an utterance is in progress

.changeset/deepgram-keep-streaming-until-endpointed.md

@@ -0,0 +1,7 @@
+---
+'@livekit/agents-plugin-deepgram': patch
+---
+
+Keep streaming audio to Deepgram while an utterance is in progress
+
+`AudioEnergyFilter` could stop forwarding frames after Deepgram had reported speech but before it had endpointed the utterance. Deepgram only fires `endpointing` on silence it actually receives, so once starved it never sent `speech_final`, no final transcript arrived, and the turn was never committed. `SpeechStream` now bypasses the energy gate between start of speech and endpoint, and clears that state when a websocket session starts so a reconnect cannot inherit an unfinished utterance.

plugins/deepgram/src/stt.test.ts

@@ -1,13 +1,15 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import { stt as agentStt } from '@livekit/agents';
 import { VAD } from '@livekit/agents-plugin-silero';
 import { stt } from '@livekit/agents-plugins-test';
 import { AudioFrame } from '@livekit/rtc-node';
 import { once } from 'node:events';
 import type { AddressInfo } from 'node:net';
 import { describe, expect, it } from 'vitest';
 import { WebSocketServer } from 'ws';
+import type { SpeechStream } from './stt.js';
 import { STT } from './stt.js';
 
 const hasDeepgramApiKey = Boolean(process.env.DEEPGRAM_API_KEY);
@@ -101,6 +103,119 @@ describe('Deepgram streaming flush', () => {
   });
 });
 
+describe('Deepgram energy gating', () => {
+  // 100ms at 16kHz, exactly one JS repack chunk. makeFrame is far below the energy
+  // filter's RMS threshold, so these only pass the gate while the cooldown lasts.
+  const FRAME_SAMPLES = 1600;
+  // Longer than the filter's one second cooldown, so the gate closes partway through.
+  const QUIET_FRAMES = 15;
+
+  const countAudio = (wire: string[]) => wire.filter((entry) => entry === 'audio').length;
+
+  function collectEvents(stream: SpeechStream): agentStt.SpeechEventType[] {
+    const types: agentStt.SpeechEventType[] = [];
+    void (async () => {
+      try {
+        for await (const event of stream) types.push(event.type);
+      } catch {
+        // the stream throws on close, which every test does in its finally block
+      }
+    })();
+    return types;
+  }
+
+  it('stops sending low-energy audio once the cooldown elapses', async () => {
+    const { wss, baseUrl } = await startWebSocketServer();
+    const wire: string[] = [];
+    wss.on('connection', (ws) => {
+      ws.on('message', (data, isBinary) => {
+        wire.push(isBinary ? 'audio' : (JSON.parse(data.toString()) as { type: string }).type);
+      });
+    });
+
+    const stream = new STT({ apiKey: 'test-key', baseUrl }).stream();
+    try {
+      for (let i = 0; i < QUIET_FRAMES; i++) stream.pushFrame(makeFrame(FRAME_SAMPLES));
+      // Finalize is ordered behind every frame already sent, so it is a barrier.
+      stream.flush();
+      await waitUntil(() => wire.includes('Finalize'));
+
+      expect(countAudio(wire)).toBeLessThan(QUIET_FRAMES);
+    } finally {
+      stream.close();
+      await closeWebSocketServer(wss);
+    }
+  });
+
+  it('keeps sending low-energy audio while an utterance is in progress', async () => {
+    const { wss, baseUrl } = await startWebSocketServer();
+    const wire: string[] = [];
+    wss.on('connection', (ws) => {
+      ws.on('message', (data, isBinary) => {
+        wire.push(isBinary ? 'audio' : (JSON.parse(data.toString()) as { type: string }).type);
+        // Announced on the first frame rather than on connection: the plugin only
+        // attaches its message listener once the socket is open, so anything sent
+        // before it has sent audio is dropped.
+        if (countAudio(wire) === 1) ws.send(JSON.stringify({ type: 'SpeechStarted' }));
+      });
+    });
+
+    const stream = new STT({ apiKey: 'test-key', baseUrl }).stream();
+    const events = collectEvents(stream);
+    try {
+      stream.pushFrame(makeFrame(FRAME_SAMPLES));
+      await waitUntil(() => events.includes(agentStt.SpeechEventType.START_OF_SPEECH));
+
+      for (let i = 0; i < QUIET_FRAMES; i++) stream.pushFrame(makeFrame(FRAME_SAMPLES));
+      stream.flush();
+      await waitUntil(() => wire.includes('Finalize'));
+
+      expect(countAudio(wire)).toBe(QUIET_FRAMES + 1);
+    } finally {
+      stream.close();
+      await closeWebSocketServer(wss);
+    }
+  });
+
+  it('does not carry an unfinished utterance into a reconnected websocket', async () => {
+    const { wss, baseUrl } = await startWebSocketServer();
+    const wires: string[][] = [];
+    wss.on('connection', (ws) => {
+      const wire: string[] = [];
+      wires.push(wire);
+      // Only the first connection reports speech; the reconnect never hears about it.
+      const announcesSpeech = wires.length === 1;
+      ws.on('message', (data, isBinary) => {
+        wire.push(isBinary ? 'audio' : (JSON.parse(data.toString()) as { type: string }).type);
+        if (announcesSpeech && countAudio(wire) === 1) {
+          ws.send(JSON.stringify({ type: 'SpeechStarted' }));
+        }
+      });
+    });
+
+    const stream = new STT({ apiKey: 'test-key', baseUrl }).stream();
+    const events = collectEvents(stream);
+    try {
+      stream.pushFrame(makeFrame(FRAME_SAMPLES));
+      await waitUntil(() => events.includes(agentStt.SpeechEventType.START_OF_SPEECH));
+
+      // Reconnects mid-utterance, without Deepgram ever endpointing it.
+      stream.updateOptions({ keyterm: ['livekit'] });
+      await waitUntil(() => wires.length === 2);
+
+      // Finalize on the new socket proves its send loop is running.
+      stream.pushFrame(makeFrame(FRAME_SAMPLES));
+      stream.flush();
+      await waitUntil(() => wires[1]!.includes('Finalize'));
+
+      expect(stream._speaking).toBe(false);
+    } finally {
+      stream.close();
+      await closeWebSocketServer(wss);
+    }
+  });
+});
+
 if (hasDeepgramApiKey) {
   describe('Deepgram', async () => {
     await stt(new STT(), await VAD.load(), { nonStreaming: false });

plugins/deepgram/src/stt.ts

@@ -375,6 +375,7 @@ export class SpeechStream extends stt.SpeechStream {
 
   async #runWS(ws: WebSocket) {
     this.#resetWS = new Future();
+    this.#speaking = false;
     let closing = false;
 
     const keepalive = setInterval(() => {
@@ -438,7 +439,8 @@ export class SpeechStream extends stt.SpeechStream {
           }
 
           for await (const frame of frames) {
-            if (this.#audioEnergyFilter.pushFrame(frame)) {
+            const energyGateOpen = this.#audioEnergyFilter.pushFrame(frame);
+            if (this.#speaking || energyGateOpen) {
               const frameDuration = frame.samplesPerChannel / frame.sampleRate;
               this.#audioDurationCollector.push(frameDuration);
               ws.send(frame.data.buffer);