From 3f04b5bfcdd85883035edef325deef6cb1d7f334 Mon Sep 17 00:00:00 2001 From: Campbell Mercer <37949826+CampbellMBXJ@users.noreply.github.com> Date: Fri, 7 Aug 2026 12:43:28 -0700 Subject: [PATCH] fix(deepgram): keep streaming audio while an utterance is in progress AudioEnergyFilter stops forwarding frames once its cooldown elapses with no audio above the RMS threshold. Deepgram only fires endpointing on silence it actually receives, so when the gate closed after Deepgram had reported speech but before it had endpointed the utterance, speech_final never arrived. The pending final was only flushed when the speaker started talking again, and until then audio_recognition had no transcript to commit the turn with. Bypass the gate between start of speech and endpoint, and clear that state when a websocket session starts so a reconnect cannot inherit an unfinished utterance from the previous socket. The sarvam plugin already resets its equivalent flag per session. pushFrame is still evaluated on every frame so the filter's cooldown stays accurate once the utterance ends. --- ...eepgram-keep-streaming-until-endpointed.md | 7 ++ plugins/deepgram/src/stt.test.ts | 115 ++++++++++++++++++ plugins/deepgram/src/stt.ts | 4 +- 3 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 .changeset/deepgram-keep-streaming-until-endpointed.md diff --git a/.changeset/deepgram-keep-streaming-until-endpointed.md b/.changeset/deepgram-keep-streaming-until-endpointed.md new file mode 100644 index 000000000..369bad80f --- /dev/null +++ b/.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. diff --git a/plugins/deepgram/src/stt.test.ts b/plugins/deepgram/src/stt.test.ts index 5932ce4cf..ad446459a 100644 --- a/plugins/deepgram/src/stt.test.ts +++ b/plugins/deepgram/src/stt.test.ts @@ -1,6 +1,7 @@ // 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'; @@ -8,6 +9,7 @@ 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 }); diff --git a/plugins/deepgram/src/stt.ts b/plugins/deepgram/src/stt.ts index eabac17ac..8aa84fec4 100644 --- a/plugins/deepgram/src/stt.ts +++ b/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);