diff --git a/.changeset/conversation-aware-stt-keyterms.md b/.changeset/conversation-aware-stt-keyterms.md new file mode 100644 index 000000000..6c26e4be5 --- /dev/null +++ b/.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. diff --git a/agents/src/inference/llm.ts b/agents/src/inference/llm.ts index 578614dd5..67d4ec01e 100644 --- a/agents/src/inference/llm.ts +++ b/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; 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'], }; diff --git a/agents/src/inference/stt.test.ts b/agents/src/inference/stt.test.ts index 6e119cab5..bed5eb490 100644 --- a/agents/src/inference/stt.test.ts +++ b/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 }); + 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 }); + + // 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'; diff --git a/agents/src/inference/stt.ts b/agents/src/inference/stt.ts index 9ef50142e..3cfc8e852 100644 --- a/agents/src/inference/stt.ts +++ b/agents/src/inference/stt.ts @@ -224,6 +224,52 @@ function diarizationEnabled(extraKwargs: Record | 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; + sessionKeyterms?: string[]; + }, +): Record | 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[] = 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 extends BaseSTT { private streams: Set> = new Set(); private vad?: VAD; private _vadPromise?: Promise; + /** 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 extends BaseSTT { interimResults: true, alignedTranscript: 'word', diarization: diarizationEnabled(modelOptions as Record), + keyterms: + keytermsExtraForModel(typeof opts?.model === 'string' ? opts.model : undefined) !== + undefined, }); const { @@ -478,12 +529,29 @@ export class STT 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), }); + // 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, + sessionKeyterms: this._sessionKeyterms, + }); + if (keytermExtra !== undefined) { + nextOpts.modelOptions = { + ...nextOpts.modelOptions, + ...keytermExtra, + } as STTOptions; + } } for (const stream of this.streams) { @@ -491,6 +559,34 @@ export class STT extends BaseSTT { } } + 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, + 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 }); + } + } + } + stream(options?: { language?: STTLanguages | string; connOptions?: APIConnectOptions; @@ -513,7 +609,14 @@ export class STT 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, + sessionKeyterms: this._sessionKeyterms, + }) ?? {}), + }, }, } as Record; @@ -562,6 +665,10 @@ export class SpeechStream extends BaseSpeechStream { private opts: InferenceSTTOptions; 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; private speechDuration = 0; private reconnectEvent = new Event(); private stt: STT; @@ -585,6 +692,11 @@ export class SpeechStream extends BaseSpeechStream { return 'inference.SpeechStream'; } + /** @internal */ + get _speaking(): boolean { + return this.speaking; + } + updateOptions( opts: Partial, 'model' | 'language' | 'modelOptions'>>, ): void { @@ -599,6 +711,10 @@ export class SpeechStream 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 extends BaseSpeechStream { } } + private onEndOfSpeech(): void { + if (this._pendingExtra !== undefined) { + this.updateOptions({ modelOptions: this._pendingExtra as STTOptions }); + this._pendingExtra = undefined; + } + } + protected async run(): Promise { while (true) { const vad = await this.stt.vadPromise; @@ -935,6 +1058,7 @@ export class SpeechStream extends BaseSpeechStream { if (this.speaking) { this.speaking = false; this.queue.put({ type: SpeechEventType.END_OF_SPEECH }); + this.onEndOfSpeech(); } } else { this.queue.put({ diff --git a/agents/src/stt/fallback_adapter.ts b/agents/src/stt/fallback_adapter.ts index 9ec3fc93f..002a22470 100644 --- a/agents/src/stt/fallback_adapter.ts +++ b/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 diff --git a/agents/src/stt/stream_adapter.ts b/agents/src/stt/stream_adapter.ts index d8651ea5b..648efc76b 100644 --- a/agents/src/stt/stream_adapter.ts +++ b/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); } diff --git a/agents/src/stt/stt.ts b/agents/src/stt/stt.ts index 5bd922e32..923517964 100644 --- a/agents/src/stt/stt.ts +++ b/agents/src/stt/stt.ts @@ -14,6 +14,7 @@ import { DeferredReadableStream } from '../stream/deferred_stream.js'; import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS, intervalForRetry } from '../types.js'; import type { AudioBuffer } from '../utils.js'; import { AsyncIterableQueue, delay, startSoon, toError } from '../utils.js'; +import type { ConversationItemAddedEvent } from '../voice/events.js'; import type { TimedString } from '../voice/index.js'; /** Indicates start/middle/end of speech */ @@ -136,6 +137,10 @@ export interface STTCapabilities { alignedTranscript?: 'word' | 'chunk' | false; /** Whether this STT supports speaker diarization. */ diarization?: boolean; + /** Whether the STT supports keyterm prompting */ + keyterms?: boolean; + /** Whether the STT can natively consume conversation context (see STT._pushConversationItem) */ + chatContext?: boolean; } export interface STTError { @@ -161,6 +166,8 @@ export type STTCallbacks = { export abstract class STT extends (EventEmitter as new () => TypedEmitter) { abstract label: string; #capabilities: STTCapabilities; + #keytermsUnsupportedWarned = false; + #chatContextUnsupportedWarned = false; constructor(capabilities: STTCapabilities) { super(); @@ -226,6 +233,47 @@ export abstract class STT extends (EventEmitter as new () => TypedEmitter; + /** + * Set the framework-managed keyterms (session config + auto-detection). + * + * Internal hook called by the framework, kept separate from the user's own keyterms + * (constructor / `updateOptions`). Plugins that support keyterms override this to + * store the session set and apply it merged with the user keyterms. + * + * @internal + */ + _updateSessionKeyterms(_keyterms: string[]): void { + if (!this.#capabilities.keyterms) { + if (!this.#keytermsUnsupportedWarned) { + this.#keytermsUnsupportedWarned = true; + log() + .child({ stt: this.label }) + .warn('keyterms are not supported by this STT, ignoring keyterms update'); + } + return; + } + } + + /** + * Feed a new conversation turn to the STT to bias recognition (context carryover). + * + * Plugins with native context support set `STTCapabilities.chatContext` and override + * this to forward the item to their provider's carryover field. + * + * @internal + */ + _pushConversationItem(_ev: ConversationItemAddedEvent): void { + if (!this.#capabilities.chatContext) { + if (!this.#chatContextUnsupportedWarned) { + this.#chatContextUnsupportedWarned = true; + log() + .child({ stt: this.label }) + .warn('chat context is not supported by this STT, ignoring chat context update'); + } + return; + } + } + /** * Returns a {@link SpeechStream} that can be used to push audio frames and receive * transcriptions diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 656646681..9fd906783 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -102,6 +102,7 @@ import { import type { AgentState, AgentStateChangedEvent, + ConversationItemAddedEvent, EotPredictionEvent, UserTurnExceededEvent, _AgentBackchannelOpportunityEvent, @@ -585,6 +586,25 @@ export class AgentActivity implements RecognitionHooks { this._resolvedTurnDetection.on('metrics_collected', this.onMetricsCollected); } + // keyterm detection runs its own LLM, surface its usage + this.agentSession._keytermDetector.on('metrics_collected', this.onMetricsCollected); + + if (this.stt instanceof STT) { + // bind the session's keyterm detector to this activity's STT (detection uses its + // own LLM, configured via keytermsOptions, not the agent's) + this.agentSession._keytermDetector.start(this.agentSession, this.stt); + + // forward conversation turns to STTs that consume context natively; gated by the + // STT's own capability (toggled via the STT's args). stateless and activity-scoped, + // so it lives here rather than in the detector. + if (this.stt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + } + // Bundled-default VAD is treated as absent when the RealtimeModel does // its own server-side turn detection — the realtime session is already // canonical and an extra audio pipeline would just pay the native model @@ -1164,6 +1184,14 @@ export class AgentActivity implements RecognitionHooks { ); }; + // (session.on("conversation_item_added", stt._push_conversation_item) — JS needs a stable + // bound reference so the listener can be removed in _closeSessionResources) + private pushConversationItemToStt = (ev: ConversationItemAddedEvent): void => { + if (this.stt instanceof STT) { + this.stt._pushConversationItem(ev); + } + }; + private onError(ev: RealtimeModelError | STTError | TTSError | LLMError): void { if (ev.type === 'realtime_model_error') { const errorEvent = createErrorEvent(ev, this.llm); @@ -4383,6 +4411,10 @@ export class AgentActivity implements RecognitionHooks { if (this.stt instanceof STT) { this.stt.off('metrics_collected', this.onMetricsCollected); this.stt.off('error', this.onModelError); + this.agentSession.off( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); } if (this.tts instanceof TTS) { @@ -4398,6 +4430,10 @@ export class AgentActivity implements RecognitionHooks { this._resolvedTurnDetection.off('metrics_collected', this.onMetricsCollected); } + // pause/aclose; keyterm state is kept on the session and survives handoffs) + this.agentSession._keytermDetector.off('metrics_collected', this.onMetricsCollected); + await this.agentSession._keytermDetector.aclose(); + this.detachAudioInput(); this.realtimeSpans?.clear(); await this.realtimeSession?.close(); diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index f7a8dbb75..e470730a1 100644 --- a/agents/src/voice/agent_session.ts +++ b/agents/src/voice/agent_session.ts @@ -92,6 +92,11 @@ import { createUserStateChangedEvent, } from './events.js'; import { AgentInput, AgentOutput } from './io.js'; +import { + KeytermDetector, + type KeytermsOptions, + resolveKeytermsOptions, +} from './keyterm_detection.js'; import { RecorderIO } from './recorder_io/index.js'; import { RoomSessionTransport, SessionHost } from './remote_session.js'; import { RoomIO, type RoomInputOptions, type RoomOutputOptions } from './room_io/index.js'; @@ -304,6 +309,13 @@ export type AgentSessionOptions = { */ turnHandling?: Partial; + /** + * Keyterm biasing for the STT. Holds static `keyterms` plus `keytermDetection` + * (LLM extraction). Applies to STTs that accept a term list; on others it warns + * and is ignored. + */ + keytermsOptions?: KeytermsOptions; + useTtsAlignedTranscript?: boolean; /** @@ -335,6 +347,12 @@ export type AgentSessionUpdateOptions = { * - `TurnDetectionMode`: set the turn detection strategy to the provided value. */ turnDetection?: TurnDetectionMode | null; + + /** + * Replace the user-defined keyterms applied to the STT. Auto-detected keyterms + * are left untouched. + */ + keyterms?: string[]; }; type ActivityTransitionOptions = { @@ -408,6 +426,9 @@ export class AgentSession< /** @internal */ _usageCollector: ModelUsageCollector = new ModelUsageCollector(); + /** @internal */ + readonly _keytermDetector: KeytermDetector; + /** @internal */ _roomIO?: RoomIO; @@ -551,6 +572,13 @@ export class AgentSession< // This is the "global" chat context, it holds the entire conversation history this._chatCtx = ChatContext.empty(); this.sessionOptions = resolvedSessionOptions; + + const keytermsOptions = resolveKeytermsOptions(this.sessionOptions.keytermsOptions); + this._keytermDetector = new KeytermDetector({ + staticKeyterms: keytermsOptions.keyterms, + options: keytermsOptions.keytermDetection, + }); + this.options = legacyVoiceOptions; this._aecWarmupRemaining = this.sessionOptions.aecWarmupDuration ?? 0; @@ -595,6 +623,11 @@ export class AgentSession< return this._chatCtx; } + /** The effective keyterms (user-defined + auto-detected) currently applied to the STT. */ + get keyterms(): string[] { + return this._keytermDetector.keyterms; + } + /** Connection options for STT, LLM, and TTS. */ get connOptions(): ResolvedSessionConnectOptions { return this._connOptions; @@ -993,6 +1026,10 @@ export class AgentSession< } updateOptions(options: AgentSessionUpdateOptions): void { + if (options.keyterms !== undefined) { + this._keytermDetector.setStaticKeyterms(options.keyterms); + } + const endpointing = options.turnHandling?.endpointing; const turnDetection = options.turnHandling?.turnDetection !== undefined diff --git a/agents/src/voice/index.ts b/agents/src/voice/index.ts index 6d18e5aa5..8d787a3db 100644 --- a/agents/src/voice/index.ts +++ b/agents/src/voice/index.ts @@ -36,6 +36,11 @@ export { TcpSessionTransport, } from './remote_session.js'; export * from './events.js'; +export { + KeytermDetector, + type KeytermDetectionOptions, + type KeytermsOptions, +} from './keyterm_detection.js'; export { AudioOutput, type AudioOutputCapabilities, diff --git a/agents/src/voice/keyterm_detection.test.ts b/agents/src/voice/keyterm_detection.test.ts new file mode 100644 index 000000000..8802c7232 --- /dev/null +++ b/agents/src/voice/keyterm_detection.test.ts @@ -0,0 +1,679 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { EventEmitter } from 'node:events'; +import { beforeAll, describe, expect, it } from 'vitest'; +import { ChatContext, ChatMessage, FunctionCall } from '../llm/chat_context.js'; +import { LLM, type LLMStream } from '../llm/llm.js'; +import { initializeLogger } from '../log.js'; +import type { LLMMetrics } from '../metrics/base.js'; +import type { SpeechEvent, SpeechStream } from '../stt/stt.js'; +import { STT } from '../stt/stt.js'; +import type { AudioBuffer } from '../utils.js'; +import { Future } from '../utils.js'; +import { createConversationItemAddedEvent } from './events.js'; +import type { ConversationItemAddedEvent } from './events.js'; +import { + type KeytermDetectionOptions, + KeytermDetector, + PENDING_TTL, + detectKeyterms, + formatInput, + parseToolCall, + resolveDetection, +} from './keyterm_detection.js'; + +type DetectionResult = [string[], string[], string[]]; + +const detector = (opts?: { + staticKeyterms?: string[]; + options?: KeytermDetectionOptions; +}): KeytermDetector => new KeytermDetector(opts); + +/** Detected terms with their confirmed flag (confirmed first, then pending). */ +const entries = (d: KeytermDetector): [string, boolean][] => [ + ...d._detectedTerms.map((t): [string, boolean] => [t, true]), + ...[...d._pendingTerms.keys()].map((t): [string, boolean] => [t, false]), +]; + +const ctx = (text: string = 'hello'): ChatContext => { + const c = ChatContext.empty(); + c.addMessage({ role: 'user', content: text }); + return c; +}; + +/** STT that records every _updateSessionKeyterms() / _pushConversationItem() call. */ +class RecordingSTT extends STT { + label = 'recording.STT'; + pushed: string[][] = []; + chatItems: ChatMessage[] = []; + + constructor(opts?: { supportsKeyterms?: boolean; supportsChatContext?: boolean }) { + super({ + streaming: true, + interimResults: false, + keyterms: opts?.supportsKeyterms ?? true, + chatContext: opts?.supportsChatContext ?? false, + }); + } + + protected _recognize(_: AudioBuffer): Promise { + throw new Error('not implemented'); + } + + stream(): SpeechStream { + throw new Error('not implemented'); + } + + override _updateSessionKeyterms(keyterms: string[]): void { + this.pushed.push([...keyterms]); + } + + override _pushConversationItem(ev: ConversationItemAddedEvent): void { + if (ev.item instanceof ChatMessage) { + this.chatItems.push(ev.item); + } + } +} + +class FakeStream { + private args: string; + + constructor(pending: string[], confirm: string[], remove: string[]) { + this.args = JSON.stringify({ pending, confirm, remove }); + } + + async collect() { + const call = new FunctionCall({ callId: '1', name: 'record_keyterms', args: this.args }); + return { text: '', toolCalls: [call], usage: undefined, extra: {} }; + } + + close(): void {} +} + +/** + * Fake LLM: returns a `record_keyterms` call per `chat()`, one result tuple per call. + * + * Subclasses LLM so the detector's `instanceof LLM` gate passes; the last result + * repeats once the sequence is exhausted. + */ +class RecordingLLM extends LLM { + private results: DetectionResult[]; + calls = 0; + lastChatCtx: ChatContext | undefined; + + constructor(...results: DetectionResult[]) { + super(); + this.results = results.length > 0 ? results : [[[], [], []]]; + } + + label(): string { + return 'recording-llm'; + } + + chat({ chatCtx }: { chatCtx: ChatContext }): LLMStream { + const result = this.results[Math.min(this.calls, this.results.length - 1)]!; + this.calls += 1; + this.lastChatCtx = chatCtx; + return new FakeStream(...result) as unknown as LLMStream; + } +} + +class BlockingStream extends FakeStream { + constructor( + private gate: Future, + result: DetectionResult = [[], [], []], + ) { + super(...result); + } + + override async collect() { + await this.gate.await; + return super.collect(); + } + + // mirror the real LLMStream: close() ends the output queue, unblocking collect() + override close(): void { + if (!this.gate.done) { + this.gate.resolve(); + } + } +} + +/** Fake LLM whose response blocks until `gate` is resolved (for single-flight tests). */ +class BlockingLLM extends LLM { + gate = new Future(); + calls = 0; + + constructor(private result: DetectionResult = [[], [], []]) { + super(); + } + + label(): string { + return 'blocking-llm'; + } + + chat(_: { chatCtx: ChatContext }): LLMStream { + this.calls += 1; + return new BlockingStream(this.gate, this.result) as unknown as LLMStream; + } +} + +class FakeSession extends EventEmitter { + history = ChatContext.empty(); + + addUser(text: string): void { + const msg = this.history.addMessage({ role: 'user', content: text }); + this.emit('conversation_item_added', createConversationItemAddedEvent(msg)); + } + + addAssistant(text: string): void { + const msg = this.history.addMessage({ role: 'assistant', content: text }); + this.emit('conversation_item_added', createConversationItemAddedEvent(msg)); + } +} + +const drain = async (d: KeytermDetector): Promise => { + await Promise.resolve(); + if (d._detectTask !== undefined) { + // a failed pass is logged + re-raised on the task + await d._detectTask.result.catch(() => {}); + } +}; + +beforeAll(() => { + initializeLogger({ pretty: false }); +}); + +// -- keyterm state machine (driven through one detection pass each) -- + +describe('keyterm state machine', () => { + it('only confirmed terms are applied', async () => { + const d = detector({ + staticKeyterms: ['Acme'], + options: { llm: new RecordingLLM([['Niamh'], ['Foo'], []]) }, + }); + await d.runOnce(ctx()); + // pending terms are tracked but not applied (entries: confirmed then pending) + expect(entries(d)).toEqual([ + ['Foo', true], + ['Niamh', false], + ]); + expect(d.keyterms).toEqual(['Acme', 'Foo']); + }); + + it('pending then confirmed', async () => { + const d = detector({ + options: { + llm: new RecordingLLM([['Kubernetes'], [], []], [[], ['Kubernetes'], []]), + }, + }); + await d.runOnce(ctx()); + expect(d.keyterms).toEqual([]); + await d.runOnce(ctx()); + expect(entries(d)).toEqual([['Kubernetes', true]]); + expect(d.keyterms).toEqual(['Kubernetes']); + }); + + it('static terms shown to llm as applied', async () => { + const fake = new RecordingLLM(); + const d = detector({ staticKeyterms: ['Acme Corp'], options: { llm: fake } }); + await d.runOnce(ctx()); + // user terms must appear in the applied list, or the LLM keeps re-proposing them + expect(fake.lastChatCtx).toBeDefined(); + const items = fake.lastChatCtx!.items; + const lastItem = items[items.length - 1]!; + const userMsg = (lastItem instanceof ChatMessage ? lastItem.textContent : undefined) ?? ''; + const appliedSection = userMsg.split('## Applied keyterms')[1]!.split('\n')[1]!; + expect(appliedSection).toContain('Acme Corp'); + }); + + it('user precedence and dedup', async () => { + const d = detector({ + staticKeyterms: ['Acme', 'Acme', 'LiveKit'], + options: { llm: new RecordingLLM([[], ['LiveKit', 'Foo'], []]) }, + }); + expect(d.staticKeyterms).toEqual(['Acme', 'LiveKit']); + await d.runOnce(ctx()); // an auto term equal to a user term is dropped + expect(entries(d).map(([t]) => t)).toEqual(['Foo']); + expect(d.keyterms).toEqual(['Acme', 'LiveKit', 'Foo']); + }); + + it('confirmed cannot revert to pending', async () => { + const d = detector({ + options: { llm: new RecordingLLM([[], ['Niamh'], []], [['Niamh'], [], []]) }, + }); + await d.runOnce(ctx()); + expect(d.keyterms).toEqual(['Niamh']); + await d.runOnce(ctx()); // a stray `pending` must not reset a confirmed term + expect(entries(d)).toEqual([['Niamh', true]]); + }); + + it('correction removes and replaces', async () => { + const d = detector({ + options: { + llm: new RecordingLLM([['Jon'], [], []], [['John'], [], ['Jon']], [[], ['John'], []]), + }, + }); + await d.runOnce(ctx()); + expect(entries(d)).toEqual([['Jon', false]]); + await d.runOnce(ctx()); // misheard spelling removed, corrected one added as pending + expect(entries(d)).toEqual([['John', false]]); + await d.runOnce(ctx()); + expect(d.keyterms).toEqual(['John']); + }); + + it('remove applies to confirmed terms', async () => { + const d = detector({ + options: { llm: new RecordingLLM([[], ['Jon'], []], [[], ['John'], ['Jon']]) }, + }); + await d.runOnce(ctx()); + expect(d.keyterms).toEqual(['Jon']); + await d.runOnce(ctx()); // a user correction can remove an already-applied term + expect(d.keyterms).toEqual(['John']); + }); + + it('remove unknown is noop', async () => { + const d = detector({ + options: { llm: new RecordingLLM([[], ['Foo'], []], [[], [], ['does-not-exist']]) }, + }); + await d.runOnce(ctx()); + await d.runOnce(ctx()); + expect(d.keyterms).toEqual(['Foo']); + }); + + it('cap evicts oldest confirmed', async () => { + const d = detector({ + options: { maxKeyterms: 3, llm: new RecordingLLM([[], ['a', 'b', 'c', 'd', 'e'], []]) }, + }); + await d.runOnce(ctx()); + expect(entries(d).map(([t]) => t)).toEqual(['c', 'd', 'e']); + }); + + it('pending evicted when not confirmed', async () => { + // pass 1 adds "Tmp" pending; later passes never confirm it, so it ages out + const d = detector({ + options: { llm: new RecordingLLM([['Tmp'], [], []], [[], ['Other'], []]) }, + }); + await d.runOnce(ctx()); + for (let i = 0; i < PENDING_TTL - 1; i++) { + await d.runOnce(ctx()); + } + expect(entries(d).map(([t]) => t)).toContain('Tmp'); + await d.runOnce(ctx()); // TTL exceeded + expect(entries(d).map(([t]) => t)).not.toContain('Tmp'); + }); + + it('confirmed not evicted by staleness', async () => { + const d = detector({ + options: { llm: new RecordingLLM([[], ['Keep'], []], [['x'], [], []]) }, + }); + await d.runOnce(ctx()); + for (let i = 0; i < PENDING_TTL + 2; i++) { + await d.runOnce(ctx()); // pending churn ages out, but the confirmed term stays + } + expect(d.keyterms).toEqual(['Keep']); + }); + + it('failed pass keeps state', async () => { + class BoomLLM extends LLM { + label(): string { + return 'boom-llm'; + } + + chat(_: { chatCtx: ChatContext }): LLMStream { + throw new Error('boom'); + } + } + + const d = detector({ options: { llm: new BoomLLM() } }); + // a failed pass is logged and re-raised on the (fire-and-forget) task; state is untouched + await expect(d.runOnce(ctx())).rejects.toThrow('boom'); + expect(d.keyterms).toEqual([]); + }); +}); + +// -- STT binding -- + +describe('STT binding', () => { + it('push only on applied change', async () => { + const stt = new RecordingSTT(); + const session = new FakeSession(); + const d = detector({ + staticKeyterms: ['Acme'], + options: { + enabled: true, + llm: new RecordingLLM([['Foo'], [], []], [[], ['Foo'], []]), + }, + }); + d.start(session, stt); + expect(stt.pushed).toEqual([['Acme']]); // start pushes the current set + + session.addUser('u1'); + await drain(d); // pending Foo: tracked, no applied change -> no push + expect(stt.pushed).toEqual([['Acme']]); + + session.addUser('u2'); + await drain(d); // confirm Foo: push + expect(stt.pushed[stt.pushed.length - 1]).toEqual(['Acme', 'Foo']); + await d.aclose(); + }); + + it('detection llm metrics forwarded', async () => { + // the detector re-emits its detection LLM's usage, so it reaches the session metrics pipeline + const received: LLMMetrics[] = []; + const fake = new RecordingLLM(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.on('metrics_collected', (m) => received.push(m)); + d.start(new FakeSession(), new RecordingSTT()); + + const metrics: LLMMetrics = { + type: 'llm_metrics', + label: fake.label(), + requestId: 'r1', + timestamp: 0, + durationMs: 0, + ttftMs: 0, + cancelled: false, + completionTokens: 1, + promptTokens: 2, + promptCachedTokens: 0, + totalTokens: 3, + tokensPerSecond: 0, + }; + fake.emit('metrics_collected', metrics); + expect(received).toEqual([metrics]); + + await d.aclose(); // detaches from the detection LLM so a later emit is dropped + fake.emit('metrics_collected', metrics); + expect(received).toEqual([metrics]); + }); + + it('start same stt does not repush', async () => { + const stt = new RecordingSTT(); + const session = new FakeSession(); + const d = detector({ + staticKeyterms: ['Acme'], + options: { enabled: true, llm: new RecordingLLM() }, + }); + d.start(session, stt); + expect(stt.pushed).toEqual([['Acme']]); + await d.aclose(); + // re-binding the same instance on the next activity must not re-push (some STTs reconnect) + d.start(session, stt); + expect(stt.pushed).toEqual([['Acme']]); + await d.aclose(); + }); + + it('static terms pushed without detection', async () => { + const stt = new RecordingSTT(); + const session = new FakeSession(); + const d = detector({ staticKeyterms: ['Acme'], options: { enabled: false } }); + d.start(session, stt); // detection off must still bind the STT and push + expect(stt.pushed).toEqual([['Acme']]); + d.setStaticKeyterms(['New']); + expect(stt.pushed[stt.pushed.length - 1]).toEqual(['New']); + await d.aclose(); + }); + + it('start pushes empty list to keyterm-capable stt', async () => { + const stt = new RecordingSTT(); + const session = new FakeSession(); + // simulate an earlier binding that left session keyterms on the instance + stt._updateSessionKeyterms(['Stale']); + const d = detector({ options: { enabled: false } }); + d.start(session, stt); // empty set still pushed, so stale terms are cleared + expect(stt.pushed).toEqual([['Stale'], []]); + await d.aclose(); + }); + + it('start without terms does not warn unsupported stt', async () => { + const stt = new RecordingSTT({ supportsKeyterms: false }); + const session = new FakeSession(); + const d = detector({ options: { enabled: false } }); + d.start(session, stt); // nothing to apply + no capability -> no push, no warning + expect(stt.pushed).toEqual([]); + await d.aclose(); + }); + + it('set static keyterms pushes', async () => { + const stt = new RecordingSTT(); + const session = new FakeSession(); + const d = detector({ options: { enabled: true, llm: new RecordingLLM() } }); + d.start(session, stt); + d.setStaticKeyterms(['New']); + expect(stt.pushed[stt.pushed.length - 1]).toEqual(['New']); + await d.aclose(); + }); + + it('unsupported stt warn and skip', () => { + const stt = new RecordingSTT({ supportsKeyterms: false }); + // exercise the base method (warn-and-skip), not the recorder override + STT.prototype._updateSessionKeyterms.call(stt, ['a', 'b']); + expect(stt.pushed).toEqual([]); + }); +}); + +// -- chat context sink (native carryover) -- +// forwarding (subscribe + push every turn) lives in AgentActivity; here we only cover the +// STT sink contract: a supporting STT receives the pushed turns, an unsupported one warns. + +describe('chat context sink', () => { + it('push conversation item forwards to supporting stt', () => { + const stt = new RecordingSTT({ supportsChatContext: true }); + const user = createConversationItemAddedEvent( + new ChatMessage({ role: 'user', content: ['hi'] }), + ); + const agent = createConversationItemAddedEvent( + new ChatMessage({ role: 'assistant', content: ['Welcome'] }), + ); + stt._pushConversationItem(user); // both user and agent turns are forwarded + stt._pushConversationItem(agent); + expect(stt.chatItems.map((m) => m.textContent)).toEqual(['hi', 'Welcome']); + }); + + it('unsupported stt chat ctx warn and skip', () => { + const stt = new RecordingSTT({ supportsChatContext: false }); + const ev = createConversationItemAddedEvent( + new ChatMessage({ role: 'assistant', content: ['hi'] }), + ); + // exercise the base method (warn-and-skip), not the recorder override + STT.prototype._pushConversationItem.call(stt, ev); + expect(stt.chatItems).toEqual([]); + }); +}); + +// -- triggering -- + +describe('triggering', () => { + it('triggers every n user turns', async () => { + const session = new FakeSession(); + const fake = new RecordingLLM([[], ['Acme'], []]); + const d = detector({ options: { enabled: true, turnInterval: 2, llm: fake } }); + d.start(session, new RecordingSTT()); + + session.addUser('first'); // below interval + await drain(d); + expect(fake.calls).toBe(0); + + session.addAssistant('ack'); // assistant turns don't advance the counter + await drain(d); + expect(fake.calls).toBe(0); + + session.addUser('second'); // triggers + await drain(d); + expect(fake.calls).toBe(1); + + await d.aclose(); + }); + + it('ignores assistant messages for counting', async () => { + const session = new FakeSession(); + const fake = new RecordingLLM(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, new RecordingSTT()); + + session.addAssistant('hello'); + await drain(d); + expect(fake.calls).toBe(0); + + session.addUser('hi'); + await drain(d); + expect(fake.calls).toBe(1); + + await d.aclose(); + }); + + it('empty user turn does not trigger', async () => { + const session = new FakeSession(); + const fake = new RecordingLLM(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, new RecordingSTT()); + + session.addUser(''); + await drain(d); + expect(fake.calls).toBe(0); + + await d.aclose(); + }); + + it('single flight skips overlapping pass', async () => { + const session = new FakeSession(); + const fake = new BlockingLLM(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, new RecordingSTT()); + + session.addUser('first'); + await Promise.resolve(); + expect(fake.calls).toBe(1); + + session.addUser('second'); // a pass is still in flight -> skipped, not queued + await Promise.resolve(); + expect(fake.calls).toBe(1); + + fake.gate.resolve(); + await drain(d); + expect(fake.calls).toBe(1); + await d.aclose(); + }); + + it('aclose cancels an in-flight pass without applying its result', async () => { + const session = new FakeSession(); + // if the pass were allowed to finish, it would confirm "Acme" + const fake = new BlockingLLM([[], ['Acme'], []]); + const stt = new RecordingSTT(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, stt); + expect(stt.pushed).toEqual([[]]); // bind-time push of the (empty) current set + + session.addUser('hi'); + await Promise.resolve(); + expect(fake.calls).toBe(1); + + // the gate is never resolved by the test: aclose must unblock the pass itself + // (via the abort signal) instead of waiting out the detection timeout + await d.aclose(); + + expect(d.keyterms).toEqual([]); // cancelled pass must not touch state + expect(stt.pushed).toEqual([[]]); // ...nor push to the STT + }); + + it('aclose unsubscribes', async () => { + const session = new FakeSession(); + const fake = new RecordingLLM(); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, new RecordingSTT()); + await d.aclose(); + + session.addUser('hi'); + await Promise.resolve(); + expect(fake.calls).toBe(0); + }); + + it('disabled detection does not trigger', async () => { + const session = new FakeSession(); + const fake = new RecordingLLM([[], ['Acme'], []]); + const d = detector({ options: { enabled: false, llm: fake } }); + d.start(session, new RecordingSTT()); + + session.addUser('the Acme Grand'); + await drain(d); + expect(fake.calls).toBe(0); + expect(d.keyterms).toEqual([]); + }); + + it('unsupported stt skips detection', async () => { + // no point running LLM detection passes when the STT can't consume the keyterms + const session = new FakeSession(); + const fake = new RecordingLLM([[], ['Acme'], []]); + const d = detector({ options: { enabled: true, llm: fake } }); + d.start(session, new RecordingSTT({ supportsKeyterms: false })); + + session.addUser('the Acme Grand'); + await drain(d); + expect(fake.calls).toBe(0); + }); +}); + +// -- module helpers -- + +describe('module helpers', () => { + it('detect keyterms parses result', async () => { + const llm = new RecordingLLM([[], ['Niamh'], ['Jon']]); + const [pending, confirm, remove] = await detectKeyterms(llm, ctx("It's Niamh"), { + currentKeyterms: [], + }); + expect(pending).toEqual([]); + expect(confirm).toEqual(['Niamh']); + expect(remove).toEqual(['Jon']); + // no transcript -> no LLM call, empty result + expect(await detectKeyterms(llm, ChatContext.empty())).toEqual([[], [], []]); + }); + + it('parse tool call', () => { + const call = new FunctionCall({ + callId: '1', + name: 'record_keyterms', + args: JSON.stringify({ + pending: ['John', ' ', 5], // blanks and non-strings are dropped + confirm: ['Foo'], + remove: ['Jon'], + }), + }); + const [pending, confirm, remove] = parseToolCall([call]); + expect(pending).toEqual(['John']); + expect(confirm).toEqual(['Foo']); + expect(remove).toEqual(['Jon']); + }); + + it('parse tool call missing', () => { + expect(parseToolCall([])).toEqual([[], [], []]); + const bad = new FunctionCall({ callId: '1', name: 'record_keyterms', args: 'not json' }); + expect(parseToolCall([bad])).toEqual([[], [], []]); + }); + + it('format input splits applied and candidate', () => { + const text = formatInput(ctx('hi'), [ + ['Term1', true], + ['Term2', false], + ]); + expect(text).toBeDefined(); + expect(text).toContain('Applied keyterms'); + expect(text).toContain('Term1'); + expect(text).toContain('Candidate keyterms'); + expect(text).toContain('Term2'); + expect(text).toContain('record_keyterms'); // trailing instruction + // no transcript yet -> nothing to send + expect(formatInput(ChatContext.empty(), [])).toBeUndefined(); + }); + + it('resolve detection', () => { + expect(resolveDetection(undefined).enabled).toBe(false); + expect(resolveDetection({ enabled: false }).enabled).toBe(false); + + const resolved = resolveDetection({ enabled: true }); + expect(resolved.enabled).toBe(true); + expect(resolved.turnInterval).toBe(1); + expect(resolved.maxKeyterms).toBeUndefined(); + }); +}); diff --git a/agents/src/voice/keyterm_detection.ts b/agents/src/voice/keyterm_detection.ts new file mode 100644 index 000000000..317a3ef66 --- /dev/null +++ b/agents/src/voice/keyterm_detection.ts @@ -0,0 +1,638 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter'; +import { EventEmitter } from 'node:events'; +import { z } from 'zod'; +import { LLM as InferenceLLM } from '../inference/llm.js'; +import { ChatContext, ChatMessage, type FunctionCall } from '../llm/chat_context.js'; +import { LLM } from '../llm/llm.js'; +import { tool } from '../llm/tool_context.js'; +import { log } from '../log.js'; +import type { LLMMetrics } from '../metrics/base.js'; +import type { STT } from '../stt/stt.js'; +import { Task, delay } from '../utils.js'; +import { AgentSessionEventTypes, type ConversationItemAddedEvent } from './events.js'; + +/** + * Keyterm biasing for STTs that accept a term list. + * + * Can be passed as a plain object: + * + * ```ts + * new AgentSession({ + * keytermsOptions: { + * keyterms: ['LiveKit', 'Acme Corp'], + * keytermDetection: { enabled: true, turnInterval: 1 }, + * }, + * }); + * ``` + */ +export interface KeytermsOptions { + /** Static keyterms applied wherever the STT accepts a term list; never touched by detection. */ + keyterms?: string[]; + /** LLM-based keyterm extraction, for STTs that accept a term list. */ + keytermDetection?: KeytermDetectionOptions; +} + +/** + * Configuration for automatic keyterm detection. + * + * Lives under the `keytermDetection` key of {@link KeytermsOptions}. Absent or + * `{ enabled: false }` keeps detection off. + */ +export interface KeytermDetectionOptions { + /** Whether to run the background detector. Defaults to `false`. */ + enabled?: boolean; + /** + * LLM used for extraction. An `LLM` instance, or a model string (e.g. + * `"google/gemini-3.5-flash"`) resolved via the inference gateway. Defaults to a + * built-in detection model; the agent's own LLM is not used. + */ + llm?: LLM | string; + /** Run a pass once per N user turns. Defaults to `1`. */ + turnInterval?: number; + /** Cap on the confirmed (applied) detected keyterms if provided. Defaults to `undefined`. */ + maxKeyterms?: number; + /** Override the built-in extraction prompt. */ + instructions?: string; + /** + * Milliseconds a single detection pass may run before it is dropped (no keyterm change). + * Defaults to `10_000`. Raise it if a slow detection `llm` needs longer. + */ + timeout?: number; +} + +// bound a single pass so a stuck LLM call can't hold the single-flight guard forever and +// stall detection for the rest of the call; a timed-out pass simply makes no change +const DETECTION_TIMEOUT = 10_000; + +const KEYTERM_DETECTION_DEFAULTS = { + enabled: false, + llm: undefined, + turnInterval: 1, + maxKeyterms: undefined, + instructions: undefined, + timeout: DETECTION_TIMEOUT, +} as const; + +/** A fully-defaulted keyterm-detection config. @internal */ +export interface ResolvedKeytermDetectionOptions { + enabled: boolean; + llm?: LLM | string; + turnInterval: number; + maxKeyterms?: number; + instructions?: string; + timeout: number; +} + +/** A fully-defaulted keyterms config. @internal */ +export interface ResolvedKeytermsOptions { + keyterms: string[]; + keytermDetection: ResolvedKeytermDetectionOptions; +} + +/** A pending term not confirmed within this many passes is dropped. @internal */ +export const PENDING_TTL = 3; +const MAX_TRANSCRIPT_MESSAGES = 12; + +// default model for keyterm extraction when `keytermDetection.llm` is not set +const DEFAULT_DETECTION_MODEL = 'google/gemma-4-31b-it'; + +// set LK_KEYTERMS_DEBUG=1 to log the input/output of every detection pass +const lkKeytermsDebug = parseInt(process.env.LK_KEYTERMS_DEBUG ?? '0', 10) !== 0; + +/** Return a fully-defaulted keyterm-detection config (`enabled` defaults to false). @internal */ +export function resolveDetection( + config?: KeytermDetectionOptions, +): ResolvedKeytermDetectionOptions { + const merged = { ...KEYTERM_DETECTION_DEFAULTS, ...(config ?? {}) }; + return { + enabled: merged.enabled ?? false, + llm: merged.llm, + turnInterval: merged.turnInterval ?? 1, + maxKeyterms: merged.maxKeyterms, + instructions: merged.instructions, + timeout: merged.timeout ?? DETECTION_TIMEOUT, + }; +} + +/** Return a fully-defaulted keyterms config. @internal */ +export function resolveKeytermsOptions(config?: KeytermsOptions): ResolvedKeytermsOptions { + const cfg = config ?? {}; + return { + keyterms: [...(cfg.keyterms ?? [])], + keytermDetection: resolveDetection(cfg.keytermDetection), + }; +} + +/** + * Resolve the configured detection `llm`: an `LLM` instance is used directly; a + * model string (or the default model when unset) is created via the inference gateway. + */ +function resolveDetectionLLM(configured?: LLM | string): LLM | undefined { + if (configured instanceof LLM) { + return configured; + } + const model = typeof configured === 'string' ? configured : DEFAULT_DETECTION_MODEL; + try { + return InferenceLLM.fromModelString(model); + } catch { + // never let detection setup break the session + log().warn(`keyterm detection: could not create detection LLM ${model}; skipping`); + return undefined; + } +} + +const DEFAULT_KEYTERM_INSTRUCTIONS = `\ +You maintain STT keyterms that bias a recognizer toward the correct spelling of distinctive \ +words (names, places, companies, products, technical terms). Each turn, adjust them with one \ +\`record_keyterms\` call. + +A WRONG spelling biases the recognizer for the rest of the call with no recovery, so precision \ +beats coverage: apply only a spelling you can CORROBORATE, and when unsure change nothing. + +USER lines are raw STT — often wrong, and the same error recurs, so repetition is NOT proof a \ +spelling is right. ASSISTANT lines are the agent's own writing: trust the agent's confident use \ +of its OWN names (brands, staff, locations) and confirm those promptly — but an assistant merely \ +echoing the user's sounds, or hedging about a spelling, does NOT corroborate. + +CONFIRM a pending term only when corroborated by one of: + 1. a letter-by-letter spell-out the assistant then accepts WITHOUT reservation — confirm \ +exactly those letters, appending nothing; + 2. the assistant's own confident use of that exact distinctive spelling; + 3. an explicit user correction ("no, not X — it's Y"). +Recurrence alone never confirms. + +HEDGE RULE: if after a spell-out or name read-back the assistant signals the letters may be off \ +("for now", "with that caveat", "may have that slightly off", "did I catch that?", "to be \ +confirmed", "I don't want to guess", "double-check"), the spelling is unreliable — keep the term \ +PENDING and never confirm it, EVEN IF the user replies "yes". Only a cleanly accepted spell-out \ +confirms. + +Never apply: a user-line word that sounds like a known term (it's that term misheard); a \ +distinctive name glued to an ordinary word ("Blue Haven Hotel" — keep the bare name pending); an \ +odd phrase only the user says and the assistant never adopts; a fragment left by an interruption; \ +ordinary words or fillers. + +Report only CHANGES; never re-list an applied term. + - \`pending\`: a distinctive term seen but not yet corroborated; + - \`confirm\`: a pending term that just met the bar above; + - \`remove\`: only a spelling the user just corrected away. Applied terms are otherwise sticky. +If nothing meets the bar this turn, change nothing.`; + +// only used to elicit a structured tool call; never executed +const recordKeyterms = tool({ + name: 'record_keyterms', + description: 'Update the STT keyterms based on the latest transcript.', + parameters: z.object({ + pending: z + .array(z.string()) + .describe('Distinctive terms seen but not yet trusted — tracked, not applied.'), + confirm: z + .array(z.string()) + .describe('Pending terms the transcript has now corroborated — applied.'), + remove: z + .array(z.string()) + .describe('Only a spelling the user corrected away; applied terms are otherwise sticky.'), + }), + execute: async () => {}, +}); + +/** + * The minimal session surface the detector needs. `AgentSession` satisfies this + * structurally; tests can supply a lightweight fake. + * + */ +export interface KeytermDetectorSession { + readonly history: ChatContext; + on( + event: AgentSessionEventTypes.ConversationItemAdded, + listener: (ev: ConversationItemAddedEvent) => void, + ): unknown; + off( + event: AgentSessionEventTypes.ConversationItemAdded, + listener: (ev: ConversationItemAddedEvent) => void, + ): unknown; +} + +export type KeytermDetectorCallbacks = { + ['metrics_collected']: (metrics: LLMMetrics) => void; +}; + +/** + * Maintains the STT keyterm set and, when enabled, auto-detects keyterms during a call. + * + * Owned by the {@link AgentSession} so keyterm state survives agent handoffs. Each agent + * activity binds it to that activity's STT via {@link KeytermDetector.start} and releases it via + * {@link KeytermDetector.aclose}. When detection is on, an LLM extracts distinctive spellings from + * the conversation; only confirmed terms are pushed to the STT, while pending terms are tracked + * (and fed back to the detector) without biasing recognition. + */ +export class KeytermDetector extends (EventEmitter as new () => TypedEmitter) { + private detection: ResolvedKeytermDetectionOptions; + private maxKeyterms?: number; + private turnInterval: number; + private instructions: string; + private detectionTimeout: number; + + private staticTerms: string[]; + /** confirmed terms, oldest first (for eviction) @internal */ + _detectedTerms: string[] = []; + /** term -> pass it was added (for TTL) @internal */ + _pendingTerms: Map = new Map(); + private tick = 0; // detection-pass counter + + // bound per agent activity (see start/aclose) + private stt?: STT; + private llm?: LLM; + private session?: KeytermDetectorSession; + private turnCount = 0; + /** @internal */ + _detectTask?: Task; + + #logger = log(); + + constructor(opts?: { staticKeyterms?: string[]; options?: KeytermDetectionOptions }) { + super(); + const options = resolveDetection(opts?.options); + this.detection = options; + this.maxKeyterms = options.maxKeyterms; + this.turnInterval = Math.max(1, options.turnInterval); + this.instructions = options.instructions ?? DEFAULT_KEYTERM_INSTRUCTIONS; + this.detectionTimeout = options.timeout; + + this.staticTerms = [...new Set(opts?.staticKeyterms ?? [])]; + this.llm = options.llm instanceof LLM ? options.llm : undefined; + } + + /** The effective list applied to the STT: static terms + confirmed detected terms. */ + get keyterms(): string[] { + return [...new Set([...this.staticTerms, ...this._detectedTerms])]; + } + + get staticKeyterms(): string[] { + return [...this.staticTerms]; + } + + setStaticKeyterms(terms: string[]): void { + this.staticTerms = [...new Set(terms)]; + if (this.stt !== undefined) { + this.stt._updateSessionKeyterms(this.keyterms); + } + } + + /** Bind this activity's STT (always) and start detection (if enabled). */ + start(session: KeytermDetectorSession, stt: STT): void { + // static keyterms must reach the recognizer even with detection disabled + if (stt !== this.stt) { + this.stt = stt; + // push even an empty list to a keyterm-capable STT: a reused instance may + // still hold session keyterms from a previous detector binding + if (this.keyterms.length > 0 || stt.capabilities.keyterms) { + this.stt._updateSessionKeyterms(this.keyterms); + } + } + + if (!this.detection.enabled) { + return; + } + + // don't waste LLM detection passes when no STT can consume the keyterms + if (!stt.capabilities.keyterms) { + this.#logger + .child({ stt: stt.label }) + .warn( + 'keyterm detection is enabled but the STT does not support keyterms; skipping detection', + ); + return; + } + + const detectLLM = resolveDetectionLLM(this.detection.llm); + if (detectLLM === undefined) { + this.#logger.warn('keyterm detection is enabled but no detection LLM is available; skipping'); + return; + } + + this.llm = detectLLM; + detectLLM.on('metrics_collected', this.forwardMetrics); + this.session = session; + this.turnCount = 0; + session.on(AgentSessionEventTypes.ConversationItemAdded, this.onConversationItemAdded); + } + + /** Stop detection for the current activity; keyterm state is kept. */ + async aclose(): Promise { + if (this.llm !== undefined) { + this.llm.off('metrics_collected', this.forwardMetrics); + } + if (this.session !== undefined) { + this.session.off(AgentSessionEventTypes.ConversationItemAdded, this.onConversationItemAdded); + this.session = undefined; + } + if (this._detectTask !== undefined) { + await this._detectTask.cancelAndWait(); + this._detectTask = undefined; + } + } + + private forwardMetrics = (ev: LLMMetrics): void => { + this.emit('metrics_collected', ev); + }; + + private onConversationItemAdded = (ev: ConversationItemAddedEvent): void => { + const session = this.session; + if (session === undefined) { + return; + } + + const item = ev.item; + // keyterm detection triggers on non-empty user turns + if (!(item instanceof ChatMessage) || item.role !== 'user' || !item.textContent) { + return; + } + + this.turnCount += 1; + if (this.turnCount % this.turnInterval !== 0) { + return; + } + + // single-flight: skip while a pass is still running + if (this._detectTask !== undefined && !this._detectTask.done) { + return; + } + + // snapshot the transcript now so the pass isn't affected by later turns + const snapshot = KeytermDetector.snapshot(session); + this._detectTask = Task.from(async (controller) => { + try { + await this.runOnce(snapshot, controller.signal); + } catch (error) { + this.#logger.child({ error }).error('keyterm detection pass failed'); + throw error; + } + }); + // fire-and-forget: failures are logged above; keep the task awaitable without unhandled rejections + this._detectTask.result.catch(() => {}); + }; + + private static snapshot(session: KeytermDetectorSession): ChatContext { + return session.history.copy({ + excludeConfigUpdate: true, + excludeFunctionCall: true, + excludeHandoff: true, + excludeEmptyMessage: true, + }); + } + + /** @internal exposed for tests */ + async runOnce(chatCtx: ChatContext, abortSignal?: AbortSignal): Promise { + if (!(this.llm instanceof LLM)) { + return; + } + + // show static terms as applied too, or the LLM keeps re-proposing them + const current: [string, boolean][] = [ + ...this.staticTerms.map((t): [string, boolean] => [t, true]), + ...this._detectedTerms.map((t): [string, boolean] => [t, true]), + ...[...this._pendingTerms.keys()].map((t): [string, boolean] => [t, false]), + ]; + const [pending, confirm, remove] = await detectKeyterms(this.llm, chatCtx, { + currentKeyterms: current, + instructions: this.instructions, + timeout: this.detectionTimeout, + abortSignal, + }); + + // cancelled mid-flight (e.g. activity shutdown): don't touch keyterm state + if (abortSignal?.aborted) { + return; + } + + const before = this.keyterms; + this.tick += 1; + + // update the keyterm state + for (const term of remove) { + this._pendingTerms.delete(term); + const idx = this._detectedTerms.indexOf(term); + if (idx !== -1) { + this._detectedTerms.splice(idx, 1); + } + } + + for (const term of pending) { + // track a new candidate; ignore static terms and ones already known + if (term && !this.staticTerms.includes(term)) { + if (!this._detectedTerms.includes(term) && !this._pendingTerms.has(term)) { + this._pendingTerms.set(term, this.tick); + } + } + } + + for (const term of confirm) { + if (term && !this.staticTerms.includes(term)) { + this._pendingTerms.delete(term); // promote out of pending + if (!this._detectedTerms.includes(term)) { + this._detectedTerms.push(term); + } + } + } + + // drop pending terms that were never confirmed in time + for (const [term, addedAt] of [...this._pendingTerms.entries()]) { + if (this.tick - addedAt >= PENDING_TTL) { + this._pendingTerms.delete(term); + } + } + + // evict oldest confirmed terms if over the cap + if (this.maxKeyterms !== undefined) { + while (this._detectedTerms.length > this.maxKeyterms) { + this._detectedTerms.shift(); + } + } + + // update the STT if the keyterms changed + const newKeyterms = this.keyterms; + if ( + this.stt !== undefined && + (newKeyterms.length !== before.length || newKeyterms.some((t, i) => t !== before[i])) + ) { + this.stt._updateSessionKeyterms(newKeyterms); + const beforeSet = new Set(before); + const newSet = new Set(newKeyterms); + this.#logger + .child({ + added: newKeyterms.filter((t) => !beforeSet.has(t)), + removed: before.filter((t) => !newSet.has(t)), + }) + .debug('keyterms changed'); + } + } +} + +/** + * Run one extraction pass via a forced function call. + * + * Returns `[pending, confirm, remove]`. + * + * @internal exposed for tests + */ +export async function detectKeyterms( + llm: LLM, + chatCtx: ChatContext, + opts?: { + instructions?: string; + currentKeyterms?: [string, boolean][]; + timeout?: number; + abortSignal?: AbortSignal; + }, +): Promise<[string[], string[], string[]]> { + const current = opts?.currentKeyterms ?? []; + const timeout = opts?.timeout ?? DETECTION_TIMEOUT; + const abortSignal = opts?.abortSignal; + if (abortSignal?.aborted) { + return [[], [], []]; + } + const userMsg = formatInput(chatCtx, current); + if (userMsg === undefined) { + // no transcript yet — nothing to detect + return [[], [], []]; + } + const reqCtx = ChatContext.empty(); + reqCtx.addMessage({ + role: 'system', + content: opts?.instructions ?? DEFAULT_KEYTERM_INSTRUCTIONS, + }); + reqCtx.addMessage({ role: 'user', content: userMsg }); + + const stream = llm.chat({ chatCtx: reqCtx, toolCtx: [recordKeyterms], toolChoice: 'required' }); + const timedOut = Symbol('keyterm-detection-timeout'); + const timeoutController = new AbortController(); + const timeoutPromise: Promise = delay(timeout, { + signal: timeoutController.signal, + }).then( + () => timedOut, + () => timedOut, // aborted: the collect() already won the race + ); + // cancellation parity with Python: closing the stream ends its output queue, + // which unblocks collect() immediately instead of waiting out the LLM request + const onAbort = () => stream.close(); + abortSignal?.addEventListener('abort', onAbort, { once: true }); + let raced: Awaited> | typeof timedOut; + try { + raced = await Promise.race([stream.collect(), timeoutPromise]); + } finally { + timeoutController.abort(); + abortSignal?.removeEventListener('abort', onAbort); + } + if (abortSignal?.aborted) { + return [[], [], []]; + } + if (raced === timedOut) { + stream.close(); + log().warn(`keyterm detection: pass timed out after ${timeout}ms; skipping`); + return [[], [], []]; + } + const result = parseToolCall(raced.toolCalls); + + if (lkKeytermsDebug) { + debugDump(userMsg, result); + } + return result; +} + +/** + * Render the detector's user message: recent transcript + current keyterms. + * + * Returns `undefined` when the transcript holds no user/assistant text yet. + * + * @internal exposed for tests + */ +export function formatInput( + chatCtx: ChatContext, + currentKeyterms: [string, boolean][], +): string | undefined { + // walk newest-first and stop once we have enough, then restore chronological order + const turns: string[] = []; + for (let i = chatCtx.items.length - 1; i >= 0; i--) { + const item = chatCtx.items[i]!; + if (!(item instanceof ChatMessage) || (item.role !== 'user' && item.role !== 'assistant')) { + continue; + } + const text = item.textContent; + if (text) { + // keep the message's line structure but drop blank lines, so the blank line + // between turns is the only blank line and reliably marks a turn boundary + const body = text + .split('\n') + .filter((line) => line.trim()) + .join('\n'); + turns.push(`${item.role.toUpperCase()}: ${body}`); + if (turns.length >= MAX_TRANSCRIPT_MESSAGES) { + break; + } + } + } + if (turns.length === 0) { + return undefined; + } + turns.reverse(); + + const applied = currentKeyterms.filter(([, ok]) => ok).map(([term]) => term); + const candidates = currentKeyterms.filter(([, ok]) => !ok).map(([term]) => term); + // always show both lists (even empty) so the model has explicit state to diff against + const sections = [ + '## Transcript (USER = raw STT, may be wrong; ASSISTANT = correct spelling)\n' + + turns.join('\n\n'), // blank line between turns + '## Applied keyterms (biasing the recognizer now)\n' + (applied.join(', ') || '(none)'), + '## Candidate keyterms (seen, not yet applied)\n' + (candidates.join(', ') || '(none)'), + 'Update the keyterms from the latest turns, then call `record_keyterms` once.', + ]; + return sections.join('\n\n'); // blank line + ## heading between sections +} + +/** + * Parse the `record_keyterms` tool call into `[pending, confirm, remove]`. + * + * @internal exposed for tests + */ +export function parseToolCall(toolCalls: FunctionCall[]): [string[], string[], string[]] { + const fnc = toolCalls.find((c) => c.name === 'record_keyterms'); + if (fnc === undefined) { + return [[], [], []]; + } + let data: Record; + try { + data = JSON.parse(fnc.args); + } catch { + return [[], [], []]; + } + + const terms = (key: string): string[] => { + const value = data[key]; + if (!Array.isArray(value)) { + return []; + } + return value.filter((t): t is string => typeof t === 'string' && t.trim().length > 0); + }; + + return [terms('pending'), terms('confirm'), terms('remove')]; +} + +/** Log the input/output of one detection pass (gated by `LK_KEYTERMS_DEBUG`). */ +function debugDump(userMsg: string, result: [string[], string[], string[]]): void { + const [pending, confirm, remove] = result; + log().debug( + [ + '──────── keyterm detection ────────', + userMsg, + '──── output ────', + `pending: ${JSON.stringify(pending)}`, + `confirm: ${JSON.stringify(confirm)}`, + `remove: ${JSON.stringify(remove)}`, + '───────────────────────────────────', + ].join('\n'), + ); +} diff --git a/examples/package.json b/examples/package.json index 1f2081386..2dfff00dc 100644 --- a/examples/package.json +++ b/examples/package.json @@ -25,6 +25,7 @@ "dependencies": { "@livekit/agents": "workspace:*", "@livekit/agents-plugin-anam": "workspace:*", + "@livekit/agents-plugin-assemblyai": "workspace:*", "@livekit/agents-plugin-baseten": "workspace:*", "@livekit/agents-plugin-bey": "workspace:*", "@livekit/agents-plugin-cartesia": "workspace:*", diff --git a/examples/src/basic_agent.ts b/examples/src/basic_agent.ts index 3947b911c..fc9b371de 100644 --- a/examples/src/basic_agent.ts +++ b/examples/src/basic_agent.ts @@ -88,6 +88,14 @@ export default defineAgent({ enabled: true, }, }, + // automatically detect keyterms and apply them to the STT per user turn + keytermsOptions: { + keyterms: ['LiveKit'], + keytermDetection: { + enabled: true, + turnInterval: 1, // increase to reduce LLM API calls + }, + }, useTtsAlignedTranscript: true, aecWarmupDuration: 3000, connOptions: { @@ -102,6 +110,9 @@ export default defineAgent({ // Log metrics as they are emitted session.on(AgentSessionEventTypes.MetricsCollected, (ev) => { + if (ev.metrics.type === 'stt_metrics') { + return; + } logMetrics(ev.metrics); }); diff --git a/plugins/assemblyai/src/stt.ts b/plugins/assemblyai/src/stt.ts index 7db090bf4..421116ef3 100644 --- a/plugins/assemblyai/src/stt.ts +++ b/plugins/assemblyai/src/stt.ts @@ -6,6 +6,8 @@ import { type APIConnectOptions, type AudioBuffer, AudioByteStream, + ChatMessage, + type ConversationItemAddedEvent, Future, Task, createTimedString, @@ -118,6 +120,13 @@ export interface STTOptions { * or `max_accuracy`. Explicit turn-silence values still take precedence over mode defaults. */ mode?: 'min_latency' | 'balanced' | 'max_accuracy'; + /** + * When the model supports it, let an `AgentSession` push each assistant reply into + * `agentContext` so it is carried into the model's conversation context. Defaults to false; + * set true to enable. Prior user turns are carried automatically by the model regardless of + * this flag. Ignored on models without context support. + */ + agentContextCarryover?: boolean; baseUrl: string; } @@ -133,6 +142,9 @@ const defaultSTTOptions: STTOptions = { export class STT extends stt.STT { #opts: STTOptions; #streams = new Set>(); + // set (user + session)) + #userKeyterms: string[]; + #sessionKeyterms: string[] = []; label = 'assemblyai.STT'; get model(): string { @@ -144,10 +156,20 @@ export class STT extends stt.STT { } constructor(opts: Partial = {}) { + // u3-rt-pro family — "u3-pro" is normalized below — and is opt-in via the user) + const rawModel = opts.speechModel ?? defaultSTTOptions.speechModel; + const supportsCarryover = isU3ProModel(rawModel) || rawModel === 'u3-pro'; + if (opts.agentContextCarryover && !supportsCarryover) { + log().warn( + `agentContextCarryover is enabled but model '${rawModel}' does not support it; ignoring`, + ); + } super({ streaming: true, interimResults: true, alignedTranscript: 'word', + keyterms: true, + chatContext: (opts.agentContextCarryover ?? false) && supportsCarryover, }); if (opts.speechModel === 'u3-pro') { @@ -189,6 +211,7 @@ export class STT extends stt.STT { apiKey, minTurnSilence, }; + this.#userKeyterms = [...(this.#opts.keytermsPrompt ?? [])]; } // eslint-disable-next-line @typescript-eslint/no-unused-vars @@ -197,17 +220,51 @@ export class STT extends stt.STT { } updateOptions(opts: Partial) { - this.#opts = { ...this.#opts, ...opts }; + // session keyterms so a user update doesn't drop them) + const nextOpts = { ...opts }; + if (nextOpts.keytermsPrompt !== undefined) { + this.#userKeyterms = [...nextOpts.keytermsPrompt]; + nextOpts.keytermsPrompt = [...new Set([...this.#userKeyterms, ...this.#sessionKeyterms])]; + } + this.#opts = { ...this.#opts, ...nextOpts }; + for (const ref of this.#streams) { + const stream = ref.deref(); + if (stream) { + stream.updateOptions(nextOpts); + } else { + this.#streams.delete(ref); + } + } + } + + override _updateSessionKeyterms(keyterms: string[]): void { + if ( + keyterms.length === this.#sessionKeyterms.length && + keyterms.every((t, i) => t === this.#sessionKeyterms[i]) + ) { + return; + } + this.#sessionKeyterms = [...keyterms]; + const merged = [...new Set([...this.#userKeyterms, ...keyterms])]; + this.#opts.keytermsPrompt = merged; + // applied live via the stream's UpdateConfiguration (no reconnect) for (const ref of this.#streams) { const stream = ref.deref(); if (stream) { - stream.updateOptions(opts); + stream.updateOptions({ keytermsPrompt: merged }); } else { this.#streams.delete(ref); } } } + override _pushConversationItem(ev: ConversationItemAddedEvent): void { + const chatItem = ev.item; + if (chatItem instanceof ChatMessage && chatItem.role === 'assistant' && chatItem.textContent) { + this.updateOptions({ agentContext: chatItem.textContent }); + } + } + stream(options?: { connOptions?: APIConnectOptions }): SpeechStream { const stream = new SpeechStream(this, this.#opts, options?.connOptions); this.#streams.add(new WeakRef(stream)); @@ -339,7 +396,7 @@ export class SpeechStream extends stt.SpeechStream { min_turn_silence: minSilence, max_turn_silence: maxSilence, keyterms_prompt: - this.#opts.keytermsPrompt !== undefined + this.#opts.keytermsPrompt !== undefined && this.#opts.keytermsPrompt.length > 0 ? JSON.stringify(this.#opts.keytermsPrompt) : undefined, language_detection: languageDetection, diff --git a/plugins/deepgram/src/stt.ts b/plugins/deepgram/src/stt.ts index ede2ca215..df15f16be 100644 --- a/plugins/deepgram/src/stt.ts +++ b/plugins/deepgram/src/stt.ts @@ -83,6 +83,10 @@ const defaultSTTOptions: STTOptions = { export class STT extends stt.STT { #opts: STTOptions; #logger = log(); + // session keyterm propagation) + #streams = new Set>(); + #userKeyterm: string[]; + #sessionKeyterms: string[] = []; label = 'deepgram.STT'; private abortController = new AbortController(); @@ -99,6 +103,7 @@ export class STT extends stt.STT { streaming: true, interimResults: opts.interimResults ?? defaultSTTOptions.interimResults, alignedTranscript: 'word', + keyterms: true, }); if (opts.apiKey === undefined && defaultSTTOptions.apiKey === undefined) { throw new Error( @@ -117,6 +122,8 @@ export class STT extends stt.STT { } else { this.#opts.model = this.#validateModel(this.#opts.model, this.#opts.language); } + + this.#userKeyterm = [...this.#opts.keyterm]; } async _recognize(buffer: AudioBuffer, abortSignal?: AbortSignal): Promise { @@ -170,14 +177,45 @@ export class STT extends stt.STT { ? this.#validateModel(opts.model ?? this.#opts.model, language) : this.#opts.model; + const nextOpts = { ...opts }; + if (nextOpts.keyterm !== undefined) { + this.#userKeyterm = [...nextOpts.keyterm]; + nextOpts.keyterm = [...new Set([...this.#userKeyterm, ...this.#sessionKeyterms])]; + } + this.#opts = { ...this.#opts, - ...opts, + ...nextOpts, language, model, }; } + override _updateSessionKeyterms(keyterms: string[]): void { + if ( + keyterms.length === this.#sessionKeyterms.length && + keyterms.every((t, i) => t === this.#sessionKeyterms[i]) + ) { + return; + } + this.#sessionKeyterms = [...keyterms]; + const merged = [...new Set([...this.#userKeyterm, ...keyterms])]; + this.#opts.keyterm = merged; + for (const ref of this.#streams) { + const stream = ref.deref(); + if (!stream) { + this.#streams.delete(ref); + continue; + } + if (stream._speaking) { + // defer the reconnect to the end of the utterance so we don't cut it off + stream._pendingKeyterm = merged; + } else { + stream.updateOptions({ keyterm: merged }); + } + } + } + #validateModel(model: STTModels | string, language?: string) { if ( language && @@ -211,7 +249,9 @@ export class STT extends stt.STT { 'language detection is not supported in streaming mode, please disable it and specify a language', ); } - return new SpeechStream(this, this.#opts, options?.connOptions); + const stream = new SpeechStream(this, this.#opts, options?.connOptions); + this.#streams.add(new WeakRef(stream)); + return stream; } async close() { @@ -226,6 +266,9 @@ export class SpeechStream extends stt.SpeechStream { #speaking = false; #resetWS = new Future(); #requestId = ''; + // keyterms set while the user is speaking; applied at END_OF_SPEECH (latest wins) + /** @internal */ + _pendingKeyterm: string[] | null = null; #audioDurationCollector: PeriodicCollector; label = 'deepgram.SpeechStream'; @@ -310,11 +353,26 @@ export class SpeechStream extends stt.SpeechStream { this.closed = true; } + /** @internal */ + get _speaking(): boolean { + return this.#speaking; + } + updateOptions(opts: Partial) { this.#opts = { ...this.#opts, ...opts }; + if (opts.keyterm !== undefined) { + this._pendingKeyterm = null; + } this.#resetWS.resolve(); } + #onEndOfSpeech() { + if (this._pendingKeyterm !== null) { + this.updateOptions({ keyterm: this._pendingKeyterm }); + this._pendingKeyterm = null; + } + } + async #runWS(ws: WebSocket) { this.#resetWS = new Future(); let closing = false; @@ -465,6 +523,7 @@ export class SpeechStream extends stt.SpeechStream { if (isEndpoint && this.#speaking) { this.#speaking = false; putMessage({ type: stt.SpeechEventType.END_OF_SPEECH }); + this.#onEndOfSpeech(); } break; @@ -476,6 +535,7 @@ export class SpeechStream extends stt.SpeechStream { if (this.#speaking) { this.#speaking = false; putMessage({ type: stt.SpeechEventType.END_OF_SPEECH }); + this.#onEndOfSpeech(); } break; } diff --git a/plugins/deepgram/src/stt_v2.ts b/plugins/deepgram/src/stt_v2.ts index 2558bc787..28605ec26 100644 --- a/plugins/deepgram/src/stt_v2.ts +++ b/plugins/deepgram/src/stt_v2.ts @@ -98,6 +98,10 @@ export class STTv2 extends stt.STT { #opts: STTv2Options; #apiKey: string; #logger = log(); + // session keyterm propagation) + #streams = new Set>(); + #userKeyterms: string[]; + #sessionKeyterms: string[] = []; /** * Create a new Deepgram STTv2 instance. @@ -120,6 +124,7 @@ export class STTv2 extends stt.STT { streaming: true, interimResults: true, alignedTranscript: 'word', + keyterms: true, }); this.#opts = { @@ -127,6 +132,7 @@ export class STTv2 extends stt.STT { ...opts, language: opts.language ? normalizeLanguage(opts.language) : defaultSTTv2Options.language, }; + this.#userKeyterms = [...this.#opts.keyterms]; const apiKey = opts.apiKey || process.env.DEEPGRAM_API_KEY; if (!apiKey) { @@ -176,7 +182,9 @@ export class STTv2 extends stt.STT { */ stream(options?: { connOptions?: APIConnectOptions }): stt.SpeechStream { const streamOpts = { ...this.#opts, apiKey: this.#apiKey }; - return new SpeechStreamv2(this, streamOpts, options?.connOptions); + const stream = new SpeechStreamv2(this, streamOpts, options?.connOptions); + this.#streams.add(new WeakRef(stream)); + return stream; } /** @@ -185,9 +193,14 @@ export class STTv2 extends stt.STT { * @param opts - Partial options to update */ updateOptions(opts: Partial) { + const nextOpts = { ...opts }; + if (nextOpts.keyterms !== undefined) { + this.#userKeyterms = [...nextOpts.keyterms]; + nextOpts.keyterms = [...new Set([...this.#userKeyterms, ...this.#sessionKeyterms])]; + } this.#opts = { ...this.#opts, - ...opts, + ...nextOpts, language: opts.language !== undefined ? normalizeLanguage(opts.language) : this.#opts.language, }; @@ -205,6 +218,31 @@ export class STTv2 extends stt.STT { } this.#logger.debug('Updated STTv2 options'); } + + override _updateSessionKeyterms(keyterms: string[]): void { + if ( + keyterms.length === this.#sessionKeyterms.length && + keyterms.every((t, i) => t === this.#sessionKeyterms[i]) + ) { + return; + } + this.#sessionKeyterms = [...keyterms]; + const merged = [...new Set([...this.#userKeyterms, ...keyterms])]; + this.#opts.keyterms = merged; + for (const ref of this.#streams) { + const stream = ref.deref(); + if (!stream) { + this.#streams.delete(ref); + continue; + } + if (stream._speaking) { + // defer the reconnect to the end of the utterance so we don't cut it off + stream._pendingKeyterm = merged; + } else { + stream.updateOptions({ keyterms: merged }); + } + } + } } // --- Stream Implementation --- @@ -222,6 +260,10 @@ class SpeechStreamv2 extends stt.SpeechStream { // Parity: _reconnect_event - using existing Event class from @livekit/agents #reconnectEvent = new Event(); + // keyterms set while the user is speaking; applied at END_OF_SPEECH (latest wins) + /** @internal */ + _pendingKeyterm: string[] | null = null; + constructor( sttInstance: STTv2, opts: STTv2Options & { apiKey: string }, @@ -236,6 +278,11 @@ class SpeechStreamv2 extends stt.SpeechStream { ); } + /** @internal */ + get _speaking(): boolean { + return this.#speaking; + } + updateOptions(opts: Partial) { this.#logger.debug('Stream received option update', opts); this.#opts = { @@ -245,11 +292,21 @@ class SpeechStreamv2 extends stt.SpeechStream { opts.language !== undefined ? normalizeLanguage(opts.language) : this.#opts.language, }; if (opts.tags) this.#opts.tags = validateTags(opts.tags); + if (opts.keyterms !== undefined) { + this._pendingKeyterm = null; + } // Trigger reconnection loop this.#reconnectEvent.set(); } + #onEndOfSpeech() { + if (this._pendingKeyterm !== null) { + this.updateOptions({ keyterms: this._pendingKeyterm }); + this._pendingKeyterm = null; + } + } + protected async run() { // Outer Loop: Handles reconnections (Configuration updates) while (!this.closed) { @@ -444,6 +501,7 @@ class SpeechStreamv2 extends stt.SpeechStream { type: stt.SpeechEventType.END_OF_SPEECH, requestId: this.#requestId, }); + this.#onEndOfSpeech(); } } else if (data.type === 'Error') { this.#logger.warn('deepgram sent an error', { data }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 92ec3169b..272adf4ba 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -248,6 +248,9 @@ importers: '@livekit/agents-plugin-anam': specifier: workspace:* version: link:../plugins/anam + '@livekit/agents-plugin-assemblyai': + specifier: workspace:* + version: link:../plugins/assemblyai '@livekit/agents-plugin-baseten': specifier: workspace:* version: link:../plugins/baseten diff --git a/turbo.json b/turbo.json index b3a1c44c5..e3d0866ae 100644 --- a/turbo.json +++ b/turbo.json @@ -91,6 +91,8 @@ "SIP_PHONE_NUMBER", "LK_OPENAI_DEBUG", "LK_GOOGLE_DEBUG", + "LK_KEYTERMS_DEBUG", + "KEYTERM_STT", "LK_DRAIN_PLAYOUT_TIMEOUT_MS", "LIVEKIT_EVALS_VERBOSE", "OVHCLOUD_API_KEY",