feat(stt): conversation-aware recognition — keyterms + chat context

.changeset/conversation-aware-stt-keyterms.md

@@ -0,0 +1,7 @@
+---
+'@livekit/agents': patch
+'@livekit/agents-plugin-deepgram': patch
+'@livekit/agents-plugin-assemblyai': patch
+---
+
+Conversation-aware STT recognition (keyterms + chat context), ported from python livekit-agents PR #6039. Adds `keytermsOptions` on `AgentSession` with static `keyterms` and LLM-based `keytermDetection` (confirmation-gated), new `STTCapabilities.keyterms`/`chatContext` flags with `_updateSessionKeyterms()`/`_pushConversationItem()` hooks (forwarded by the fallback and stream adapters), keyterm support for deepgram (v1/v2), assemblyai, and livekit inference STT, and native conversation-context carryover (`agentContextCarryover`) for assemblyai u3-rt-pro.

agents/src/inference/llm.ts

@@ -411,16 +411,22 @@ export class LLMStream extends llm.LLMStream {
       // yields only function tools (sorted by name), so they are skipped here. See AJS-112.
       const tools = this.toolCtx
         ? llm.sortedToolEntries(this.toolCtx).map(([name, func]) => {
+            // zod v3 conversion embeds a `$schema` URI ("draft/2019-09") that some gateway
+            // deployments reject, silently ending the stream with no tool call. Python's
+            // pydantic schemas carry no `$schema` key, so strip it for parity. Destructure
+            // instead of `delete` so a caller-supplied raw JSON schema object isn't mutated.
+            const { $schema: _dropped, ...parameters } = llm.toJsonSchema(
+              func.parameters,
+              true,
+              this.strictToolSchema,
+            ) as unknown as Record<string, unknown>;
             const oaiParams = {
               type: 'function' as const,
               function: {
                 name,
                 description: func.description,
-                parameters: llm.toJsonSchema(
-                  func.parameters,
-                  true,
-                  this.strictToolSchema,
-                ) as unknown as OpenAI.Chat.Completions.ChatCompletionFunctionTool['function']['parameters'],
+                parameters:
+                  parameters as OpenAI.Chat.Completions.ChatCompletionFunctionTool['function']['parameters'],
               } as OpenAI.Chat.Completions.ChatCompletionFunctionTool['function'],
             };
 

agents/src/inference/stt.test.ts

@@ -327,6 +327,50 @@ describe('STT diarization capabilities', () => {
   });
 });
 
+describe('STT session keyterms', () => {
+  it('updateOptions does not bake session keyterms into the user baseline', () => {
+    const stt = makeStt({ model: 'deepgram/nova-3' });
+    const stream = stt.stream();
+
+    stt._updateSessionKeyterms(['Niamh']);
+    // a later user option update must re-apply session terms to live streams...
+    stt.updateOptions({ modelOptions: { endpointing: 500 } as Record<string, unknown> });
+    expect(stream['opts'].modelOptions).toHaveProperty('keyterm', ['Niamh']);
+    // ...but must not pollute the STT's own user baseline with them
+    expect(stt['opts'].modelOptions ?? {}).not.toHaveProperty('keyterm');
+
+    stream.close();
+  });
+
+  it('session keyterm change after updateOptions drops stale terms', () => {
+    const stt = makeStt({ model: 'deepgram/nova-3' });
+    const stream = stt.stream();
+
+    stt._updateSessionKeyterms(['Stale']);
+    stt.updateOptions({ modelOptions: { endpointing: 500 } as Record<string, unknown> });
+
+    // detector replaced the session terms: the old one must disappear downstream
+    stt._updateSessionKeyterms(['Fresh']);
+    expect(stream['opts'].modelOptions).toHaveProperty('keyterm', ['Fresh']);
+
+    stream.close();
+  });
+
+  it('user keyterms from modelOptions are preserved across session updates', () => {
+    const stt = makeStt({ model: 'deepgram/nova-3', modelOptions: { keyterm: ['Acme'] } });
+    const stream = stt.stream();
+
+    stt._updateSessionKeyterms(['Niamh']);
+    expect(stream['opts'].modelOptions).toHaveProperty('keyterm', ['Acme', 'Niamh']);
+
+    stt._updateSessionKeyterms(['Other']);
+    // user term stays; only the session portion is swapped
+    expect(stream['opts'].modelOptions).toHaveProperty('keyterm', ['Acme', 'Other']);
+
+    stream.close();
+  });
+});
+
 describe('STT VAD handling for Speechmatics models', () => {
   class MockVAD extends VAD {
     label = 'mock';

agents/src/inference/stt.ts

@@ -224,6 +224,52 @@ function diarizationEnabled(extraKwargs: Record<string, unknown> | undefined): b
   });
 }
 
+/**
+ * Return the provider's keyterm `extra` entry: user keyterms (from `extraKwargs`)
+ * merged with the framework `sessionKeyterms`.
+ *
+ * `undefined` if the model has no keyterm prompting, so
+ * `keytermsExtraForModel(model) !== undefined` is also the capability check.
+ */
+function keytermsExtraForModel(
+  model: string | undefined,
+  opts?: {
+    extraKwargs?: Record<string, unknown>;
+    sessionKeyterms?: string[];
+  },
+): Record<string, unknown> | undefined {
+  if (typeof model !== 'string' || !model) {
+    return undefined;
+  }
+
+  const extraKwargs = opts?.extraKwargs ?? {};
+  const sessionKeyterms = opts?.sessionKeyterms ?? [];
+
+  if (model.startsWith('speechmatics/')) {
+    // keep existing entries as-is (they may carry sounds_like etc.); append new session terms
+    const rawExisting = extraKwargs.additional_vocab;
+    const existing: Record<string, unknown>[] = Array.isArray(rawExisting) ? [...rawExisting] : [];
+    const seen = new Set(existing.map((v) => v.content));
+    const additions = sessionKeyterms.filter((term) => !seen.has(term));
+    return { additional_vocab: [...existing, ...additions.map((term) => ({ content: term }))] };
+  }
+
+  let key: string | undefined;
+  if (model.startsWith('deepgram/')) {
+    key = 'keyterm';
+  } else if (model.startsWith('assemblyai/')) {
+    key = 'keyterms_prompt';
+  }
+
+  if (key === undefined) {
+    return undefined;
+  }
+  // deepgram's keyterm may be a bare string; wrap it so it isn't splat char-by-char
+  const rawUser = extraKwargs[key] ?? [];
+  const existing = typeof rawUser === 'string' ? [rawUser] : Array.isArray(rawUser) ? rawUser : [];
+  return { [key]: [...new Set([...existing, ...sessionKeyterms])] };
+}
+
 type _STTModels =
   | DeepgramModels
   | DeepgramFluxModels
@@ -336,6 +382,8 @@ export class STT<TModel extends STTModels> extends BaseSTT {
   private streams: Set<SpeechStream<TModel>> = new Set();
   private vad?: VAD;
   private _vadPromise?: Promise<VAD | undefined>;
+  /** framework-managed; merged into modelOptions @internal */
+  _sessionKeyterms: string[] = [];
 
   /**
    * Resolves to the VAD instance for the current model, or `undefined` if the model
@@ -369,6 +417,9 @@ export class STT<TModel extends STTModels> extends BaseSTT {
       interimResults: true,
       alignedTranscript: 'word',
       diarization: diarizationEnabled(modelOptions as Record<string, unknown>),
+      keyterms:
+        keytermsExtraForModel(typeof opts?.model === 'string' ? opts.model : undefined) !==
+        undefined,
     });
 
     const {
@@ -478,19 +529,64 @@ export class STT<TModel extends STTModels> extends BaseSTT {
     if (nextOpts.model !== undefined) {
       this.vad = resolveVADForModel(nextOpts.model, this.vad);
       this._vadPromise = undefined;
+      this.updateCapabilities({
+        keyterms: keytermsExtraForModel(this.opts.model) !== undefined,
+      });
     }
 
     if (nextOpts.modelOptions) {
       this.updateCapabilities({
         diarization: diarizationEnabled(this.opts.modelOptions as Record<string, unknown>),
       });
+      // re-apply the active session keyterms on top of the update sent to live streams,
+      // so a user extra update doesn't drop them. `this.opts.modelOptions` must stay a
+      // pure user baseline (no session terms baked in), or a later session-keyterm
+      // change could never remove previously applied terms.
+      const keytermExtra = keytermsExtraForModel(this.opts.model, {
+        extraKwargs: this.opts.modelOptions as Record<string, unknown>,
+        sessionKeyterms: this._sessionKeyterms,
+      });
+      if (keytermExtra !== undefined) {
+        nextOpts.modelOptions = {
+          ...nextOpts.modelOptions,
+          ...keytermExtra,
+        } as STTOptions<TModel>;
+      }
     }
 
     for (const stream of this.streams) {
       stream.updateOptions(nextOpts);
     }
   }
 
+  override _updateSessionKeyterms(keyterms: string[]): void {
+    if (
+      keyterms.length === this._sessionKeyterms.length &&
+      keyterms.every((t, i) => t === this._sessionKeyterms[i])
+    ) {
+      return;
+    }
+    const keytermExtra = keytermsExtraForModel(this.opts.model, {
+      extraKwargs: this.opts.modelOptions as Record<string, unknown>,
+      sessionKeyterms: keyterms,
+    });
+    if (keytermExtra === undefined) {
+      super._updateSessionKeyterms(keyterms); // warn-and-skip for unsupported models
+      return;
+    }
+
+    this._sessionKeyterms = [...keyterms];
+    // inference applies extra live via session.update; defer to END_OF_SPEECH since the
+    // gateway may reconnect upstream when the keyterms change
+    for (const stream of this.streams) {
+      if (stream._speaking) {
+        stream._pendingExtra = keytermExtra;
+      } else {
+        stream.updateOptions({ modelOptions: keytermExtra as STTOptions<TModel> });
+      }
+    }
+  }
+
   stream(options?: {
     language?: STTLanguages | string;
     connOptions?: APIConnectOptions;
@@ -513,7 +609,14 @@ export class STT<TModel extends STTModels> extends BaseSTT {
       settings: {
         sample_rate: String(this.opts.sampleRate),
         encoding: this.opts.encoding,
-        extra: this.opts.modelOptions,
+        // the framework session keyterms into the user's extra_kwargs keyterm key)
+        extra: {
+          ...this.opts.modelOptions,
+          ...(keytermsExtraForModel(this.opts.model, {
+            extraKwargs: this.opts.modelOptions as Record<string, unknown>,
+            sessionKeyterms: this._sessionKeyterms,
+          }) ?? {}),
+        },
       },
     } as Record<string, unknown>;
 
@@ -562,6 +665,10 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
   private opts: InferenceSTTOptions<TModel>;
   private requestId = shortuuid('stt_request_');
   private speaking = false;
+  // keyterm extra set while the user is speaking; applied at END_OF_SPEECH (latest wins).
+  // inference applies live, but the gateway may reconnect upstream, so defer to a calm moment.
+  /** @internal */
+  _pendingExtra?: Record<string, unknown>;
   private speechDuration = 0;
   private reconnectEvent = new Event();
   private stt: STT<TModel>;
@@ -585,6 +692,11 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
     return 'inference.SpeechStream';
   }
 
+  /** @internal */
+  get _speaking(): boolean {
+    return this.speaking;
+  }
+
   updateOptions(
     opts: Partial<Pick<InferenceSTTOptions<TModel>, 'model' | 'language' | 'modelOptions'>>,
   ): void {
@@ -599,6 +711,10 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
       modelOptions: mergedModelOptions,
     };
 
+    if (opts.modelOptions !== undefined) {
+      this._pendingExtra = undefined;
+    }
+
     // When the WebSocket is live, send a mid-stream session.update so providers
     // that support it (e.g. AssemblyAI, Deepgram Flux) apply changes without
     // reconnecting. Unsupported providers ignore the message.
@@ -617,6 +733,13 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
     }
   }
 
+  private onEndOfSpeech(): void {
+    if (this._pendingExtra !== undefined) {
+      this.updateOptions({ modelOptions: this._pendingExtra as STTOptions<TModel> });
+      this._pendingExtra = undefined;
+    }
+  }
+
   protected async run(): Promise<void> {
     while (true) {
       const vad = await this.stt.vadPromise;
@@ -935,6 +1058,7 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
         if (this.speaking) {
           this.speaking = false;
           this.queue.put({ type: SpeechEventType.END_OF_SPEECH });
+          this.onEndOfSpeech();
         }
       } else {
         this.queue.put({

agents/src/stt/fallback_adapter.ts

@@ -7,6 +7,7 @@ import type { STTMetrics } from '../metrics/base.js';
 import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js';
 import { Task, cancelAndWait } from '../utils.js';
 import type { VAD } from '../vad.js';
+import type { ConversationItemAddedEvent } from '../voice/events.js';
 import { StreamAdapter } from './stream_adapter.js';
 import {
   STT,
@@ -139,6 +140,8 @@ export class FallbackAdapter extends STT {
       interimResults: wrapped.every((s) => s.capabilities.interimResults),
       diarization: wrapped.every((s) => !!s.capabilities.diarization),
       alignedTranscript,
+      keyterms: wrapped.some((s) => !!s.capabilities.keyterms),
+      chatContext: wrapped.some((s) => !!s.capabilities.chatContext),
     });
 
     this.sttInstances = wrapped;
@@ -186,6 +189,20 @@ export class FallbackAdapter extends STT {
     return this._status;
   }
 
+  override _updateSessionKeyterms(keyterms: string[]): void {
+    // forward to every underlying STT; unsupported ones warn-and-skip internally
+    for (const sttInstance of this.sttInstances) {
+      sttInstance._updateSessionKeyterms(keyterms);
+    }
+  }
+
+  override _pushConversationItem(ev: ConversationItemAddedEvent): void {
+    // forward to every underlying STT; unsupported ones warn-and-skip internally
+    for (const sttInstance of this.sttInstances) {
+      sttInstance._pushConversationItem(ev);
+    }
+  }
+
   private setupEventForwarding(): void {
     // We intentionally do NOT forward child 'error' events. The adapter's job
     // is to mask transient child failures via fallback — surfacing them to

agents/src/stt/stream_adapter.ts

@@ -8,6 +8,7 @@ import type { APIConnectOptions } from '../types.js';
 import { isStreamClosedError } from '../utils.js';
 import type { VAD, VADStream } from '../vad.js';
 import { VADEventType } from '../vad.js';
+import type { ConversationItemAddedEvent } from '../voice/events.js';
 import type { SpeechEvent } from './stt.js';
 import { STT, SpeechEventType, SpeechStream } from './stt.js';
 
@@ -17,7 +18,12 @@ export class StreamAdapter extends STT {
   label: string;
 
   constructor(stt: STT, vad: VAD) {
-    super({ streaming: true, interimResults: false });
+    super({
+      streaming: true,
+      interimResults: false,
+      keyterms: stt.capabilities.keyterms,
+      chatContext: stt.capabilities.chatContext,
+    });
     this.#stt = stt;
     this.#vad = vad;
     this.label = `stt.StreamAdapter<${this.#stt.label}>`;
@@ -35,6 +41,14 @@ export class StreamAdapter extends STT {
     return this.#stt.recognize(frame, abortSignal);
   }
 
+  override _updateSessionKeyterms(keyterms: string[]): void {
+    this.#stt._updateSessionKeyterms(keyterms);
+  }
+
+  override _pushConversationItem(ev: ConversationItemAddedEvent): void {
+    this.#stt._pushConversationItem(ev);
+  }
+
   stream(options?: { connOptions?: APIConnectOptions }): StreamAdapterWrapper {
     return new StreamAdapterWrapper(this.#stt, this.#vad, options?.connOptions);
   }