From a1622136a2e965b25ba71ab46d8514baae16e033 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 16 Jun 2026 14:34:54 +0100 Subject: [PATCH 1/2] feat(eot): add backchannel support --- .changeset/eot-backchannel-opportunity.md | 12 ++ agents/package.json | 2 +- agents/src/inference/eot/base.ts | 20 ++- agents/src/inference/eot/detector.test.ts | 120 ++++++++++++++++++ agents/src/inference/eot/detector.ts | 34 ++++- agents/src/inference/eot/languages.ts | 110 +++++++++++++--- agents/src/inference/eot/transports.ts | 8 +- .../interruption_failover.test.ts | 1 + agents/src/voice/agent_activity.ts | 6 + agents/src/voice/audio_recognition.ts | 29 +++++ .../audio_recognition_backchannel.test.ts | 1 + .../audio_recognition_endpointing.test.ts | 1 + .../src/voice/audio_recognition_eou.test.ts | 1 + .../voice/audio_recognition_handoff.test.ts | 1 + .../audio_recognition_interruption.test.ts | 1 + .../audio_recognition_push_audio.test.ts | 1 + .../src/voice/audio_recognition_span.test.ts | 2 + .../audio_recognition_turn_detection.test.ts | 118 ++++++++++++++++- .../voice/audio_recognition_vad_reset.test.ts | 1 + agents/src/voice/events.ts | 50 ++++++++ pnpm-lock.yaml | 12 +- 21 files changed, 498 insertions(+), 33 deletions(-) create mode 100644 .changeset/eot-backchannel-opportunity.md diff --git a/.changeset/eot-backchannel-opportunity.md b/.changeset/eot-backchannel-opportunity.md new file mode 100644 index 000000000..24d49c0db --- /dev/null +++ b/.changeset/eot-backchannel-opportunity.md @@ -0,0 +1,12 @@ +--- +"@livekit/agents": patch +--- + +feat(eot): emit agent backchannel opportunity events (AGT-2520) + +The multimodal EOT model now returns a backchannel probability alongside the end-of-turn probability. The turn detector compares it to a server-provided threshold and, when it clears, surfaces an internal backchannel *opportunity* (a window where the agent could say a short "mm-hmm" while the user still holds the floor) to `AgentActivity`. + +- `inference.TurnDetector` gains a `backchannelThreshold` option (and `updateOptions({ backchannelThreshold })`); `ThresholdOptions.lookupBackchannel()` resolves server-provided defaults layered with user overrides, mirroring the existing EOT threshold resolution. +- Backchannel thresholds are server-driven and cloud-only — disabled when the gateway sends none, after a cloud→local fallback (the mini model produces no backchannel probability), and for any non-positive threshold. +- Internal only: `AgentActivity.onAgentBackchannelOpportunity` is a no-op with a TODO; the event is not surfaced as a public `AgentSession` event (absent from the `AgentEvent` union, `AgentSessionEventTypes`, and package exports), treated the same way as the internal EOT prediction plumbing. +- Requires `@livekit/protocol` >= 1.46.8 (adds `EotPrediction.backchannelProbability` and `SessionCreated.defaultBackchannelThresholds` / `defaultBackchannelThreshold`). diff --git a/agents/package.json b/agents/package.json index e39309524..3a028e195 100644 --- a/agents/package.json +++ b/agents/package.json @@ -54,7 +54,7 @@ "@ffmpeg-installer/ffmpeg": "^1.1.0", "@livekit/local-inference": "^0.2.5", "@livekit/mutex": "^1.1.1", - "@livekit/protocol": "^1.46.5", + "@livekit/protocol": "^1.46.8", "@livekit/throws-transformer": "0.1.8", "@livekit/typed-emitter": "^3.0.0", "@opentelemetry/api": "^1.9.0", diff --git a/agents/src/inference/eot/base.ts b/agents/src/inference/eot/base.ts index 97cf01cee..af4261ac5 100644 --- a/agents/src/inference/eot/base.ts +++ b/agents/src/inference/eot/base.ts @@ -45,6 +45,9 @@ export interface TurnDetectionEvent { detectionDelay?: number; /** Server-side model inference time (ms). */ inferenceDuration?: number; + /** How appropriate it is for the agent to backchannel at this pause. + * `undefined` when the detector does not produce one (e.g. the local mini model). */ + backchannelProbability?: number; } /** @@ -125,6 +128,12 @@ export abstract class BaseStreamingTurnDetector extends (EventEmitter as new () return this._opts.thresholds.lookup(language); } + /** Threshold above which a pause is a backchannel opportunity, or `undefined` + * when backchannel is disabled (server sent none, or the local mini model). */ + async backchannelThreshold(language: LanguageCode | undefined): Promise { + return this._opts.thresholds.lookupBackchannel(language); + } + async supportsLanguage(language: LanguageCode | undefined): Promise { return this._opts.thresholds.supports(language); } @@ -219,6 +228,10 @@ export class BaseStreamingTurnDetectorStream { return this._opts.thresholds.lookup(language); } + async backchannelThreshold(language: LanguageCode | undefined): Promise { + return this._opts.thresholds.lookupBackchannel(language); + } + async supportsLanguage(language: LanguageCode | undefined): Promise { return this._opts.thresholds.supports(language); } @@ -367,7 +380,11 @@ export class BaseStreamingTurnDetectorStream { _resolvePrediction( requestId: string, probability: number, - opts: { inferenceDuration?: number; detectionDelay?: number } = {}, + opts: { + inferenceDuration?: number; + detectionDelay?: number; + backchannelProbability?: number; + } = {}, ): void { // Drop predictions that land after teardown — an in-flight transport // predict can resolve after `aclose` closed the channels. @@ -387,6 +404,7 @@ export class BaseStreamingTurnDetectorStream { lastSpeakingTimeMs: Date.now(), detectionDelay: opts.detectionDelay, inferenceDuration: opts.inferenceDuration, + backchannelProbability: opts.backchannelProbability, }); } } diff --git a/agents/src/inference/eot/detector.test.ts b/agents/src/inference/eot/detector.test.ts index 609b492dd..9c94aafa0 100644 --- a/agents/src/inference/eot/detector.test.ts +++ b/agents/src/inference/eot/detector.test.ts @@ -41,6 +41,10 @@ import { LocalTransport } from './transports.js'; const SERVER_THRESHOLDS: Record = { en: 0.56, ja: 0.37, fr: 0.575 }; const SERVER_DEFAULT_THRESHOLD = 0.5; +// Backchannel defaults a gateway returns alongside the EOT defaults. +const SERVER_BACKCHANNEL_THRESHOLDS: Record = { en: 0.62, ja: 0.7 }; +const SERVER_BACKCHANNEL_DEFAULT = 0.6; + async function waitFor(predicate: () => boolean, ticks = 50): Promise { for (let i = 0; i < ticks; i++) { if (predicate()) return; @@ -429,6 +433,107 @@ describe('ResolveThresholds', () => { }); }); +describe('BackchannelThresholds', () => { + // Server-provided backchannel defaults, disabled on the mini model and after + // fallback (see ResolveBackchannelThresholds for override layering). + function cloud(): ThresholdOptions { + const opts = new ThresholdOptions('turn-detector-v1'); + opts._updateDefaults( + { ...SERVER_THRESHOLDS }, + SERVER_DEFAULT_THRESHOLD, + { ...SERVER_BACKCHANNEL_THRESHOLDS }, + SERVER_BACKCHANNEL_DEFAULT, + ); + return opts; + } + + it('lookup per-language and default', () => { + const opts = cloud(); + expect(opts.lookupBackchannel('en')).toBeCloseTo(SERVER_BACKCHANNEL_THRESHOLDS.en!); + // absent language → catch-all backchannel default + expect(opts.lookupBackchannel('de')).toBeCloseTo(SERVER_BACKCHANNEL_DEFAULT); + // undefined language defaults to "en" + expect(opts.lookupBackchannel(undefined)).toBeCloseTo(SERVER_BACKCHANNEL_THRESHOLDS.en!); + }); + + it('disabled when server omits backchannel', () => { + const opts = new ThresholdOptions('turn-detector-v1'); + opts._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD); + expect(opts.lookupBackchannel('en')).toBeUndefined(); + }); + + it('disabled for the local mini model', () => { + const opts = new ThresholdOptions('turn-detector-v1-mini'); + expect(opts.lookupBackchannel('en')).toBeUndefined(); + }); + + it('non-positive threshold treated as disabled', () => { + const opts = new ThresholdOptions('turn-detector-v1'); + opts._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD, { en: 0.0 }, 0.6); + // en explicitly 0 → disabled for en, but the positive default still applies elsewhere + expect(opts.lookupBackchannel('en')).toBeUndefined(); + expect(opts.lookupBackchannel('de')).toBeCloseTo(0.6); + }); + + it('cleared on local fallback', () => { + const opts = cloud(); + opts._toLocalFallback(); + expect(opts.lookupBackchannel('en')).toBeUndefined(); + }); +}); + +describe('ResolveBackchannelThresholds', () => { + // User backchannel-threshold overrides layered against the server defaults, + // mirroring the EOT override resolution in ResolveThresholds. + function cloud(overrides?: number | Record): ThresholdOptions { + const opts = new ThresholdOptions('turn-detector-v1', undefined, overrides); + opts._updateDefaults( + { ...SERVER_THRESHOLDS }, + SERVER_DEFAULT_THRESHOLD, + { ...SERVER_BACKCHANNEL_THRESHOLDS }, + SERVER_BACKCHANNEL_DEFAULT, + ); + return opts; + } + + it('no override adopts server backchannel', () => { + const opts = cloud(); + expect(opts.lookupBackchannel('en')).toBeCloseTo(SERVER_BACKCHANNEL_THRESHOLDS.en!); + }); + + it('scalar override applies to every language', () => { + const opts = cloud(0.8); + expect(opts.lookupBackchannel('en')).toBeCloseTo(0.8); + expect(opts.lookupBackchannel('ja')).toBeCloseTo(0.8); + }); + + it('dict override layers on server map', () => { + const opts = cloud({ en: 0.5 }); + expect(opts.lookupBackchannel('en')).toBeCloseTo(0.5); + // unmapped languages keep the server values + server default + expect(opts.lookupBackchannel('ja')).toBeCloseTo(SERVER_BACKCHANNEL_THRESHOLDS.ja!); + expect(opts.lookupBackchannel('de')).toBeCloseTo(SERVER_BACKCHANNEL_DEFAULT); + }); + + it('dict keys normalized', () => { + const opts = cloud({ English: 0.5 }); + expect(opts.lookupBackchannel('en')).toBeCloseTo(0.5); + }); + + it('scalar override enables before server defaults', () => { + // an explicit scalar override resolves up front, even though the server + // backchannel defaults haven't arrived yet + const opts = new ThresholdOptions('turn-detector-v1', undefined, 0.8); + expect(opts.lookupBackchannel('en')).toBeCloseTo(0.8); + }); + + it('updateBackchannelOverrides re-resolves', () => { + const opts = cloud(); + opts.updateBackchannelOverrides(0.45); + expect(opts.lookupBackchannel('ja')).toBeCloseTo(0.45); + }); +}); + describe('ServerDefaults', () => { it('cloud thresholds pending before session created', async () => { const transport = new ScriptedTransport({ runBehavior: 'idle' }); @@ -492,6 +597,21 @@ describe('OverrideWarning', () => { } }); + it('warns on construction with backchannel override', () => { + const warnSpy = vi.spyOn(log(), 'warn'); + try { + withEnv({ LIVEKIT_REMOTE_EOT_URL: undefined }, () => { + new TurnDetector({ backchannelThreshold: 0.7 }); + }); + const warned = warnSpy.mock.calls.some((c) => + JSON.stringify(c).includes('non-default backchannel threshold'), + ); + expect(warned).toBe(true); + } finally { + warnSpy.mockRestore(); + } + }); + it('no warning without override', () => { const warnSpy = vi.spyOn(log(), 'warn'); try { diff --git a/agents/src/inference/eot/detector.ts b/agents/src/inference/eot/detector.ts index f01cf7d96..327e96758 100644 --- a/agents/src/inference/eot/detector.ts +++ b/agents/src/inference/eot/detector.ts @@ -33,6 +33,12 @@ export interface TurnDetectorOptions { */ version?: TurnDetectorVersion; unlikelyThreshold?: number | Record; + /** + * Backchannel threshold(s): above this, a pause is a backchannel opportunity. + * Server-driven and cloud-only by default; this is an override seam. A scalar + * applies to every language; a map is layered over the server defaults. + */ + backchannelThreshold?: number | Record; baseUrl?: string; apiKey?: string; apiSecret?: string; @@ -103,7 +109,11 @@ export class TurnDetector extends BaseStreamingTurnDetector { const detectorOpts: BaseStreamingTurnDetectorOptions = { sampleRate: opts.sampleRate ?? DEFAULT_SAMPLE_RATE, - thresholds: new ThresholdOptions(resolvedModel, opts.unlikelyThreshold), + thresholds: new ThresholdOptions( + resolvedModel, + opts.unlikelyThreshold, + opts.backchannelThreshold, + ), }; super(detectorOpts); this._model = resolvedModel; @@ -146,13 +156,31 @@ export class TurnDetector extends BaseStreamingTurnDetector { 'defaults and overriding them may be suboptimal', ); } + const bcOverrides = this._opts.thresholds.backchannelOverrides; + if (bcOverrides !== undefined) { + log().warn( + { backchannelThreshold: bcOverrides }, + 'a non-default backchannel threshold was provided; the server provides calibrated ' + + 'defaults and overriding them may be suboptimal', + ); + } } /** Replace the user threshold override at runtime. The shared * `ThresholdOptions` re-resolves against the current (server or shipped) * defaults, so an active stream picks it up immediately. */ - updateOptions(opts: { unlikelyThreshold?: number | Record } = {}): void { - this._opts.thresholds.updateOverrides(opts.unlikelyThreshold); + updateOptions( + opts: { + unlikelyThreshold?: number | Record; + backchannelThreshold?: number | Record; + } = {}, + ): void { + if (opts.unlikelyThreshold !== undefined) { + this._opts.thresholds.updateOverrides(opts.unlikelyThreshold); + } + if (opts.backchannelThreshold !== undefined) { + this._opts.thresholds.updateBackchannelOverrides(opts.backchannelThreshold); + } this._warnThresholdOverride(); } diff --git a/agents/src/inference/eot/languages.ts b/agents/src/inference/eot/languages.ts index c4dfaa689..4457a6edb 100644 --- a/agents/src/inference/eot/languages.ts +++ b/agents/src/inference/eot/languages.ts @@ -107,18 +107,31 @@ function normalizeOverrides(overrides: ThresholdOverride): ThresholdOverride { export class ThresholdOptions { private _model: TurnDetectorModel; private _overrides: ThresholdOverride; + private _bcOverrides: ThresholdOverride; // server/shipped defaults private _serverThresholds: Record | undefined; private _serverDefault: number | undefined; - // materialized values + // backchannel server defaults: cloud-only (the local mini model produces no + // backchannel probability), arrive via `SessionCreated`. + private _serverBcThresholds: Record | undefined; + private _serverBcDefault: number | undefined; + + // materialized values (server defaults layered with user overrides) private _thresholds: Record = {}; private _default: number | undefined = undefined; + private _bcThresholds: Record = {}; + private _bcDefault: number | undefined = undefined; - constructor(model: TurnDetectorModel, overrides: ThresholdOverride = undefined) { + constructor( + model: TurnDetectorModel, + overrides: ThresholdOverride = undefined, + backchannelOverrides: ThresholdOverride = undefined, + ) { this._model = model; this._overrides = normalizeOverrides(overrides); + this._bcOverrides = normalizeOverrides(backchannelOverrides); if (model === 'turn-detector-v1-mini') { this._serverThresholds = { ...LOCAL_LANGUAGES }; this._serverDefault = LOCAL_LANGUAGES.en; @@ -134,6 +147,10 @@ export class ThresholdOptions { return this._overrides; } + get backchannelOverrides(): ThresholdOverride { + return this._bcOverrides; + } + get thresholds(): Readonly> { return this._thresholds; } @@ -149,6 +166,20 @@ export class ThresholdOptions { return key in this._thresholds ? this._thresholds[key] : this._default; } + /** + * Backchannel threshold for a language, or `undefined` when backchannel is + * disabled — no server defaults / overrides resolved, or the resolved value is + * non-positive (an explicit "off"). Backchannel is server-driven and cloud-only. + */ + lookupBackchannel(language: LanguageCode | string | undefined): number | undefined { + if (Object.keys(this._bcThresholds).length === 0 && this._bcDefault === undefined) { + return undefined; + } + const key = language ? normalizeLanguage(language) : 'en'; + const threshold = key in this._bcThresholds ? this._bcThresholds[key] : this._bcDefault; + return threshold !== undefined && threshold > 0 ? threshold : undefined; + } + supports(language: LanguageCode | string | undefined): boolean { // A cloud detector reports every language as supported until its server // defaults arrive, so the first turn (before `SessionCreated`) isn't @@ -162,12 +193,22 @@ export class ThresholdOptions { this._resolve(); } + updateBackchannelOverrides(overrides: ThresholdOverride): void { + this._bcOverrides = normalizeOverrides(overrides); + this._resolve(); + } + /** * @internal Adopt the calibrated defaults a `turn-detector` gateway sends in * `SessionCreated`. Raises (non-retryable) when the server produced no usable * thresholds — the caller degrades the session to the local model. */ - _updateDefaults(serverThresholds: Record, serverDefault: number): void { + _updateDefaults( + serverThresholds: Record, + serverDefault: number, + backchannelThresholds?: Record, + backchannelDefault = 0, + ): void { if (!serverThresholds || Object.keys(serverThresholds).length === 0 || serverDefault <= 0) { throw new APIError('turn detector session created without usable default thresholds', { retryable: false, @@ -179,6 +220,19 @@ export class ThresholdOptions { } this._serverThresholds = norm; this._serverDefault = round4(serverDefault); + + // backchannel defaults are optional; an absent/empty map keeps backchannel disabled + if (backchannelThresholds && Object.keys(backchannelThresholds).length > 0) { + const bcNorm: Record = {}; + for (const [lang, value] of Object.entries(backchannelThresholds)) { + bcNorm[normalizeLanguage(lang)] = round4(value); + } + this._serverBcThresholds = bcNorm; + } else { + this._serverBcThresholds = undefined; + } + this._serverBcDefault = backchannelDefault > 0 ? round4(backchannelDefault) : undefined; + this._resolve(); } @@ -208,6 +262,9 @@ export class ThresholdOptions { this._model = 'turn-detector-v1-mini'; this._serverThresholds = { ...LOCAL_LANGUAGES }; this._serverDefault = LOCAL_LANGUAGES.en; + // the mini model produces no backchannel probability + this._serverBcThresholds = undefined; + this._serverBcDefault = undefined; this._resolve(); if (rescaled !== undefined) { @@ -217,30 +274,43 @@ export class ThresholdOptions { } private _resolve(): void { - const scalarOverride = typeof this._overrides === 'number'; - if (this._serverThresholds === undefined || this._serverDefault === undefined) { + [this._thresholds, this._default] = ThresholdOptions._resolveLayer( + this._serverThresholds, + this._serverDefault, + this._overrides, + ); + [this._bcThresholds, this._bcDefault] = ThresholdOptions._resolveLayer( + this._serverBcThresholds, + this._serverBcDefault, + this._bcOverrides, + ); + } + + /** + * Layer a user override onto the server defaults. A scalar override replaces + * the whole map (every language resolves through it); a dict override is + * merged over the server map. Before server defaults arrive, only a scalar + * override resolves up front. + */ + private static _resolveLayer( + serverThresholds: Record | undefined, + serverDefault: number | undefined, + overrides: ThresholdOverride, + ): [Record, number | undefined] { + const scalarOverride = typeof overrides === 'number'; + if (serverThresholds === undefined || serverDefault === undefined) { // cloud defaults not received yet; only a scalar override resolves up front - this._thresholds = {}; - this._default = scalarOverride ? (this._overrides as number) : undefined; - return; + return [{}, scalarOverride ? (overrides as number) : undefined]; } - if (this._overrides === undefined) { - this._thresholds = { ...this._serverThresholds }; - this._default = this._serverDefault; - return; + if (overrides === undefined) { + return [{ ...serverThresholds }, serverDefault]; } if (scalarOverride) { - this._thresholds = {}; - this._default = this._overrides as number; - return; + return [{}, overrides as number]; } - this._thresholds = { - ...this._serverThresholds, - ...(this._overrides as Record), - }; - this._default = this._serverDefault; + return [{ ...serverThresholds, ...(overrides as Record) }, serverDefault]; } } diff --git a/agents/src/inference/eot/transports.ts b/agents/src/inference/eot/transports.ts index a08bbc1c5..d1ccdc0c0 100644 --- a/agents/src/inference/eot/transports.ts +++ b/agents/src/inference/eot/transports.ts @@ -397,6 +397,7 @@ export class CloudTransport implements StreamingTurnDetectionTransport { stream._resolvePrediction(msg.requestId ?? '', prediction.probability, { detectionDelay: detectionDelayMs, inferenceDuration: inferenceDurationMs, + backchannelProbability: prediction.backchannelProbability, }); const detector = this._detectorRef.deref(); if (detector !== undefined) { @@ -423,7 +424,12 @@ export class CloudTransport implements StreamingTurnDetectionTransport { // response (no usable thresholds) throws a non-retryable `APIError` that // propagates out of the recv task → `run()` → the stream's cloud→local // fallback. - stream.thresholdsOptions._updateDefaults(created.defaultThresholds, created.defaultThreshold); + stream.thresholdsOptions._updateDefaults( + created.defaultThresholds, + created.defaultThreshold, + created.defaultBackchannelThresholds, + created.defaultBackchannelThreshold, + ); this._logger.debug( { model: stream.thresholdsOptions.model, diff --git a/agents/src/inference/interruption/interruption_failover.test.ts b/agents/src/inference/interruption/interruption_failover.test.ts index b1cd77bed..d70b6538a 100644 --- a/agents/src/inference/interruption/interruption_failover.test.ts +++ b/agents/src/inference/interruption/interruption_failover.test.ts @@ -347,6 +347,7 @@ function createHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), }; diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index f3b88351c..e6bba7f67 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -98,6 +98,7 @@ import type { AgentStateChangedEvent, EotPredictionEvent, UserTurnExceededEvent, + _AgentBackchannelOpportunityEvent, } from './events.js'; import { AgentSessionEventTypes, @@ -1490,6 +1491,11 @@ export class AgentActivity implements RecognitionHooks { this.agentSession.emit(AgentSessionEventTypes.EotPrediction, ev); } + onAgentBackchannelOpportunity(_ev: _AgentBackchannelOpportunityEvent): void { + // TODO: consume the backchannel opportunity internally (e.g. trigger a + // backchannel phrase). Kept internal for now — not surfaced as a public event. + } + onPreemptiveGeneration(info: PreemptiveGenerationInfo): void { const preemptiveOpts = this.agentSession.sessionOptions.turnHandling.preemptiveGeneration; if ( diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index 2284f35bd..65c0f1306 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -45,6 +45,8 @@ import type { TurnDetectionMode } from './agent_session.js'; import { type EotPredictionEvent, type UserTurnExceededEvent, + type _AgentBackchannelOpportunityEvent, + _createAgentBackchannelOpportunityEvent, createEotPredictionEvent, createUserTurnExceededEvent, } from './events.js'; @@ -99,6 +101,7 @@ export interface RecognitionHooks { onFinalTranscript: (ev: SpeechEvent, speaking: boolean | undefined) => void; onEndOfTurn: (info: EndOfTurnInfo) => Promise; onEotPrediction: (ev: EotPredictionEvent) => void; + onAgentBackchannelOpportunity: (ev: _AgentBackchannelOpportunityEvent) => void; onPreemptiveGeneration: (info: PreemptiveGenerationInfo) => void; onUserTurnExceeded: (ev: UserTurnExceededEvent) => void; @@ -1462,6 +1465,7 @@ export class AudioRecognition { // below. let endOfTurnProbability: number | undefined; let unlikelyThreshold: number | undefined; + let backchannelThreshold: number | undefined; // True when the held future was already resolved when this // bounce started — i.e. the prediction was served from the // request the silence tick warmed, not awaited fresh. @@ -1500,6 +1504,9 @@ export class AudioRecognition { predictionEvent = winner.ev; endOfTurnProbability = predictionEvent.endOfTurnProbability; unlikelyThreshold = await turnDetector.unlikelyThreshold(this.lastLanguage); + backchannelThreshold = await turnDetector.backchannelThreshold( + this.lastLanguage, + ); } else { this.logger.warn( { timeoutMs: endpointingDelay }, @@ -1588,6 +1595,28 @@ export class AudioRecognition { ); } + // Surface the backchannel opportunity whenever it clears its + // threshold, regardless of end-of-turn; AgentActivity decides + // whether to acknowledge mid-turn or let it lead the reply. + const backchannelProbability = prediction?.backchannelProbability; + if ( + backchannelProbability !== undefined && + backchannelThreshold !== undefined && + endOfTurnProbability !== undefined && + unlikelyThreshold !== undefined && + backchannelProbability >= backchannelThreshold + ) { + this.hooks.onAgentBackchannelOpportunity( + _createAgentBackchannelOpportunityEvent({ + probability: backchannelProbability, + threshold: backchannelThreshold, + endOfTurnProbability, + endOfTurnThreshold: unlikelyThreshold, + language: this.lastLanguage, + }), + ); + } + if (prediction?.detectionDelay !== undefined) { span.setAttribute(traceTypes.ATTR_EOU_DETECTION_DELAY, prediction.detectionDelay); } diff --git a/agents/src/voice/audio_recognition_backchannel.test.ts b/agents/src/voice/audio_recognition_backchannel.test.ts index fbdf4e457..009bd82f0 100644 --- a/agents/src/voice/audio_recognition_backchannel.test.ts +++ b/agents/src/voice/audio_recognition_backchannel.test.ts @@ -20,6 +20,7 @@ function createHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), }; diff --git a/agents/src/voice/audio_recognition_endpointing.test.ts b/agents/src/voice/audio_recognition_endpointing.test.ts index 58002ca92..ec0d09533 100644 --- a/agents/src/voice/audio_recognition_endpointing.test.ts +++ b/agents/src/voice/audio_recognition_endpointing.test.ts @@ -16,6 +16,7 @@ function createHooks(): RecognitionHooks { onInterimTranscript: () => {}, onFinalTranscript: () => {}, onPreemptiveGeneration: () => {}, + onAgentBackchannelOpportunity: () => {}, retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: async () => true, }; diff --git a/agents/src/voice/audio_recognition_eou.test.ts b/agents/src/voice/audio_recognition_eou.test.ts index da9c5e490..78c8c2ed1 100644 --- a/agents/src/voice/audio_recognition_eou.test.ts +++ b/agents/src/voice/audio_recognition_eou.test.ts @@ -56,6 +56,7 @@ function createHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), onUserTurnExceeded: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), diff --git a/agents/src/voice/audio_recognition_handoff.test.ts b/agents/src/voice/audio_recognition_handoff.test.ts index 8d3b072b4..372f9c188 100644 --- a/agents/src/voice/audio_recognition_handoff.test.ts +++ b/agents/src/voice/audio_recognition_handoff.test.ts @@ -19,6 +19,7 @@ function createHooks() { onFinalTranscript: vi.fn(), onEndOfTurn: vi.fn(async () => true), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), onUserTurnExceeded: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), }; diff --git a/agents/src/voice/audio_recognition_interruption.test.ts b/agents/src/voice/audio_recognition_interruption.test.ts index 3011f661a..de3c78406 100644 --- a/agents/src/voice/audio_recognition_interruption.test.ts +++ b/agents/src/voice/audio_recognition_interruption.test.ts @@ -16,6 +16,7 @@ function createHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), }; diff --git a/agents/src/voice/audio_recognition_push_audio.test.ts b/agents/src/voice/audio_recognition_push_audio.test.ts index e2f234333..ddfa18a07 100644 --- a/agents/src/voice/audio_recognition_push_audio.test.ts +++ b/agents/src/voice/audio_recognition_push_audio.test.ts @@ -20,6 +20,7 @@ function createHooks(): RecognitionHooks { onFinalTranscript: vi.fn(), onEndOfTurn: vi.fn(async () => true), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), }; } diff --git a/agents/src/voice/audio_recognition_span.test.ts b/agents/src/voice/audio_recognition_span.test.ts index 5ce592042..09dd4a889 100644 --- a/agents/src/voice/audio_recognition_span.test.ts +++ b/agents/src/voice/audio_recognition_span.test.ts @@ -111,6 +111,7 @@ describe('AudioRecognition user_turn span parity', () => { onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), onEotPrediction: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), }; @@ -193,6 +194,7 @@ describe('AudioRecognition user_turn span parity', () => { onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), onEotPrediction: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), }; diff --git a/agents/src/voice/audio_recognition_turn_detection.test.ts b/agents/src/voice/audio_recognition_turn_detection.test.ts index ba2bf8e81..82092f9e3 100644 --- a/agents/src/voice/audio_recognition_turn_detection.test.ts +++ b/agents/src/voice/audio_recognition_turn_detection.test.ts @@ -83,6 +83,7 @@ function makeHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onEotPrediction: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), onPreemptiveGeneration: vi.fn(), onUserTurnExceeded: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), @@ -122,6 +123,9 @@ function makeAudioStream(): BaseStreamingTurnDetectorStream { const stream = Object.create(BaseStreamingTurnDetectorStream.prototype); stream.supportsLanguage = vi.fn(async () => true); stream.unlikelyThreshold = vi.fn(async () => 0.5); + // backchannel disabled by default (server sent no thresholds); the + // backchannel-emit tests override this with a positive threshold. + stream.backchannelThreshold = vi.fn(async () => undefined); stream.predict = vi.fn(() => new Future()); stream.cancelInference = vi.fn(); stream.flush = vi.fn(); @@ -137,7 +141,11 @@ function makeAudioDetector(stream: BaseStreamingTurnDetectorStream): BaseStreami /** A resolved prediction future, as if the transport already answered. */ function resolvedPrediction( probability: number, - opts: { inferenceDuration?: number; detectionDelay?: number } = {}, + opts: { + inferenceDuration?: number; + detectionDelay?: number; + backchannelProbability?: number; + } = {}, ): { fut: Future; event: TurnDetectionEvent } { const event: TurnDetectionEvent = { type: 'eot_prediction', @@ -145,6 +153,7 @@ function resolvedPrediction( lastSpeakingTimeMs: 0, inferenceDuration: opts.inferenceDuration, detectionDelay: opts.detectionDelay, + backchannelProbability: opts.backchannelProbability, }; const fut = new Future(); fut.resolve(event); @@ -386,6 +395,113 @@ describe('TestEotPredictionDedup', () => { }); }); +describe('TestBackchannelOpportunityEmit', () => { + // `onAgentBackchannelOpportunity` fires whenever the backchannel probability + // clears its threshold, regardless of end-of-turn; the event carries the + // end-of-turn probability and threshold so AgentActivity can gauge how close + // the pause is to a reply. + async function drive(internals: RecognitionInternals): Promise { + internals.runEOUDetection(ChatContext.empty(), 'vad'); + await flush(); + await flush(); + await internals.bounceEOUTask?.cancelAndWait().catch(() => {}); + } + + it('emits with eot context when the turn continues', async () => { + const { internals, hooks } = makeRecognition(); + const stream = makeAudioStream(); + stream.backchannelThreshold = vi.fn(async () => 0.5); + internals.turnDetectorStream = stream; + internals.turnDetector = makeAudioDetector(stream); + // eot 0.2 < unlikely 0.5 → turn continues; backchannel 0.8 >= 0.5 → emit + internals.turnDetectorPredictionFut = resolvedPrediction(0.2, { + backchannelProbability: 0.8, + }).fut; + + await drive(internals); + + expect(hooks.onAgentBackchannelOpportunity).toHaveBeenCalledTimes(1); + const ev = (hooks.onAgentBackchannelOpportunity as ReturnType).mock.calls[0]![0]; + expect(ev.probability).toBeCloseTo(0.8); + expect(ev.threshold).toBeCloseTo(0.5); + expect(ev.endOfTurnProbability).toBeCloseTo(0.2); + expect(ev.endOfTurnThreshold).toBeCloseTo(0.5); + }); + + it('emits with eot context when the turn ends', async () => { + // The turn-continuing gate was dropped: a backchannel above threshold still + // fires at end-of-turn, carrying the EOT context (probability past the + // threshold) so AgentActivity can let it lead the reply. + const { internals, hooks } = makeRecognition(); + const stream = makeAudioStream(); + stream.backchannelThreshold = vi.fn(async () => 0.5); + internals.turnDetectorStream = stream; + internals.turnDetector = makeAudioDetector(stream); + // eot 0.9 >= unlikely 0.5 → turn ends; backchannel 0.8 >= 0.5 → still emits + internals.turnDetectorPredictionFut = resolvedPrediction(0.9, { + backchannelProbability: 0.8, + }).fut; + + await drive(internals); + + expect(hooks.onAgentBackchannelOpportunity).toHaveBeenCalledTimes(1); + const ev = (hooks.onAgentBackchannelOpportunity as ReturnType).mock.calls[0]![0]; + expect(ev.endOfTurnProbability).toBeCloseTo(0.9); + expect(ev.endOfTurnThreshold).toBeCloseTo(0.5); + }); + + it('does not emit below threshold', async () => { + const { internals, hooks } = makeRecognition(); + const stream = makeAudioStream(); + stream.backchannelThreshold = vi.fn(async () => 0.7); + internals.turnDetectorStream = stream; + internals.turnDetector = makeAudioDetector(stream); + // backchannel 0.4 < 0.7 → no emit (turn continues at eot 0.2) + internals.turnDetectorPredictionFut = resolvedPrediction(0.2, { + backchannelProbability: 0.4, + }).fut; + + await drive(internals); + + expect(hooks.onAgentBackchannelOpportunity).not.toHaveBeenCalled(); + }); + + it('does not emit when backchannel is disabled', async () => { + const { internals, hooks } = makeRecognition(); + const stream = makeAudioStream(); + // default fake threshold is undefined (server sent no backchannel defaults) + internals.turnDetectorStream = stream; + internals.turnDetector = makeAudioDetector(stream); + internals.turnDetectorPredictionFut = resolvedPrediction(0.2, { + backchannelProbability: 0.9, + }).fut; + + await drive(internals); + + expect(hooks.onAgentBackchannelOpportunity).not.toHaveBeenCalled(); + }); + + it('does not emit for a text-based detector', async () => { + // A text detector produces no streaming prediction event, so there is no + // backchannel probability to act on. + const { internals, hooks } = makeRecognition(); + const textDetector: _TurnDetector = { + model: 'fake', + provider: 'fake', + supportsLanguage: vi.fn(async () => true), + unlikelyThreshold: vi.fn(async () => 0.5), + predictEndOfTurn: vi.fn(async () => 0.2), + }; + internals.turnDetector = textDetector; + internals.turnDetectorStream = undefined; + internals.audioTranscript = 'hello there'; + + await drive(internals); + + expect(hooks.onAgentBackchannelOpportunity).not.toHaveBeenCalled(); + }); +}); + describe('TestPredictionFutureLifecycle', () => { it('silence tick starts a request once', async () => { const { internals } = makeRecognition(); diff --git a/agents/src/voice/audio_recognition_vad_reset.test.ts b/agents/src/voice/audio_recognition_vad_reset.test.ts index be548b876..8ec404fb0 100644 --- a/agents/src/voice/audio_recognition_vad_reset.test.ts +++ b/agents/src/voice/audio_recognition_vad_reset.test.ts @@ -33,6 +33,7 @@ function makeHooks(): RecognitionHooks { onInterimTranscript: vi.fn(), onFinalTranscript: vi.fn(), onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), onUserTurnExceeded: vi.fn(), retrieveChatCtx: () => ChatContext.empty(), onEndOfTurn: vi.fn(async () => true), diff --git a/agents/src/voice/events.ts b/agents/src/voice/events.ts index c1396effb..b79b72930 100644 --- a/agents/src/voice/events.ts +++ b/agents/src/voice/events.ts @@ -286,6 +286,56 @@ export const createEotPredictionEvent = ({ createdAt, }); +/** + * Internal: a window in which the agent could backchannel (a short acknowledgment + * such as "mm-hmm"), as predicted by the turn detector. Passed to `AgentActivity` + * only — not surfaced as a public `AgentSession` event (absent from `AgentEvent`, + * `AgentSessionEventTypes`, and the package exports). + * + * `AgentActivity` owns the decision of what to do with it. The end-of-turn margin + * (`endOfTurnThreshold - endOfTurnProbability`) gives a progressive risk axis: a + * large positive margin means the user is clearly still going, so riskier + * backchannels (yeah/okay/right) are safe; a small margin (or a negative one, + * where `endOfTurnProbability >= endOfTurnThreshold` and a reply is imminent) + * calls for safe, less ambiguous ones (hmm/uh-huh) that won't collide with the reply. + * + * @internal + */ +export type _AgentBackchannelOpportunityEvent = { + type: 'agent_backchannel_opportunity'; + probability: number; + threshold: number; + endOfTurnProbability: number; + endOfTurnThreshold: number; + language?: string; + createdAt: number; +}; + +/** @internal */ +export const _createAgentBackchannelOpportunityEvent = ({ + probability, + threshold, + endOfTurnProbability, + endOfTurnThreshold, + language, + createdAt = Date.now(), +}: { + probability: number; + threshold: number; + endOfTurnProbability: number; + endOfTurnThreshold: number; + language?: string; + createdAt?: number; +}): _AgentBackchannelOpportunityEvent => ({ + type: 'agent_backchannel_opportunity', + probability, + threshold, + endOfTurnProbability, + endOfTurnThreshold, + language, + createdAt, +}); + export type UserTurnExceededEvent = { type: 'user_turn_exceeded'; /** Transcript from the current uncommitted user turn only. */ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7e7780bc7..a396f9a37 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -119,8 +119,8 @@ importers: specifier: ^1.1.1 version: 1.1.1 '@livekit/protocol': - specifier: ^1.46.5 - version: 1.46.6 + specifier: ^1.46.8 + version: 1.46.8 '@livekit/throws-transformer': specifier: 0.1.8 version: 0.1.8(typescript@5.9.3) @@ -2193,8 +2193,8 @@ packages: cpu: [x64] os: [win32] - '@livekit/protocol@1.46.6': - resolution: {integrity: sha512-upzlHP1vi/kZ/QqALZTFskQ0ifqc2f15RKucHYOsIHJsaXvEYanG75mAb7o+Yomfs4XhQ4BaRsdY+TFHXpaqrg==} + '@livekit/protocol@1.46.8': + resolution: {integrity: sha512-mOjcCVLy4Q7qEaEE7gGLi5wXan0K3VTvSpto5Y0ftek2hauALxBW0+cyxNRoakT7dbWFfH+gqc2XQM0P4M1Q/g==} '@livekit/rtc-ffi-bindings-darwin-arm64@0.12.60': resolution: {integrity: sha512-YHXqybkYfaTc3txJXXWoVogiSP3yKJdkaZlIlZ6IDMGnN9elUoHDYU+ZSn/rbdGu0pp4HUOzffXkbkItN735Bw==} @@ -5894,7 +5894,7 @@ snapshots: '@livekit/noise-cancellation-win32-x64@0.1.9': optional: true - '@livekit/protocol@1.46.6': + '@livekit/protocol@1.46.8': dependencies: '@bufbuild/protobuf': 1.10.1 @@ -8084,7 +8084,7 @@ snapshots: livekit-server-sdk@2.14.1: dependencies: '@bufbuild/protobuf': 1.10.1 - '@livekit/protocol': 1.46.6 + '@livekit/protocol': 1.46.8 camelcase-keys: 9.1.3 jose: 5.2.4 From 28f9bb413923cb9b21d48065fda173a45d1e8165 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Tue, 16 Jun 2026 15:23:44 +0100 Subject: [PATCH 2/2] address comment: fix backchannel emission duplication due to incorrect indentation --- agents/src/voice/audio_recognition.ts | 41 +++++++++---------- .../audio_recognition_turn_detection.test.ts | 26 ++++++++++++ 2 files changed, 46 insertions(+), 21 deletions(-) diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index 65c0f1306..dc89a1b7e 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -1593,28 +1593,27 @@ export class AudioRecognition { delayMs, }), ); - } - // Surface the backchannel opportunity whenever it clears its - // threshold, regardless of end-of-turn; AgentActivity decides - // whether to acknowledge mid-turn or let it lead the reply. - const backchannelProbability = prediction?.backchannelProbability; - if ( - backchannelProbability !== undefined && - backchannelThreshold !== undefined && - endOfTurnProbability !== undefined && - unlikelyThreshold !== undefined && - backchannelProbability >= backchannelThreshold - ) { - this.hooks.onAgentBackchannelOpportunity( - _createAgentBackchannelOpportunityEvent({ - probability: backchannelProbability, - threshold: backchannelThreshold, - endOfTurnProbability, - endOfTurnThreshold: unlikelyThreshold, - language: this.lastLanguage, - }), - ); + // Surface the backchannel opportunity whenever it clears its + // threshold, regardless of end-of-turn; AgentActivity decides + // whether to acknowledge mid-turn or let it lead the reply. + // Shares the eot-emit dedupe so it fires once per request. + const backchannelProbability = prediction?.backchannelProbability; + if ( + backchannelProbability !== undefined && + backchannelThreshold !== undefined && + backchannelProbability >= backchannelThreshold + ) { + this.hooks.onAgentBackchannelOpportunity( + _createAgentBackchannelOpportunityEvent({ + probability: backchannelProbability, + threshold: backchannelThreshold, + endOfTurnProbability, + endOfTurnThreshold: unlikelyThreshold, + language: this.lastLanguage, + }), + ); + } } if (prediction?.detectionDelay !== undefined) { diff --git a/agents/src/voice/audio_recognition_turn_detection.test.ts b/agents/src/voice/audio_recognition_turn_detection.test.ts index 82092f9e3..61d7c9c1f 100644 --- a/agents/src/voice/audio_recognition_turn_detection.test.ts +++ b/agents/src/voice/audio_recognition_turn_detection.test.ts @@ -428,6 +428,32 @@ describe('TestBackchannelOpportunityEmit', () => { expect(ev.endOfTurnThreshold).toBeCloseTo(0.5); }); + it('emits once across vad then stt triggers (shares the eot dedupe)', async () => { + const { internals, hooks } = makeRecognition(); + const stream = makeAudioStream(); + stream.backchannelThreshold = vi.fn(async () => 0.5); + internals.turnDetectorStream = stream; + internals.turnDetector = makeAudioDetector(stream); + // both triggers read the same cached prediction by reference + internals.turnDetectorPredictionFut = resolvedPrediction(0.2, { + backchannelProbability: 0.8, + }).fut; + + internals.runEOUDetection(ChatContext.empty(), 'vad'); + await flush(); + await flush(); + expect(hooks.onAgentBackchannelOpportunity).toHaveBeenCalledTimes(1); + + // stt trigger runs a fresh bounce against the same resolved future; the + // dedupe that suppresses the second eot emit must suppress this too. + internals.runEOUDetection(ChatContext.empty(), 'stt'); + await flush(); + await flush(); + + expect(hooks.onAgentBackchannelOpportunity).toHaveBeenCalledTimes(1); + await internals.bounceEOUTask?.cancelAndWait().catch(() => {}); + }); + it('emits with eot context when the turn ends', async () => { // The turn-continuing gate was dropped: a backchannel above threshold still // fires at end-of-turn, carrying the EOT context (probability past the