Add AEC warmup functionality to AgentSession and AgentActivity (#1091)

Author: toubatbrian

.changeset/itchy-rocks-cheer.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Add AEC warmup functionality to AgentSession and AgentActivity

agents/src/voice/agent_activity.ts

@@ -7,7 +7,7 @@ import type { Span } from '@opentelemetry/api';
 import { ROOT_CONTEXT, context as otelContext, trace } from '@opentelemetry/api';
 import { Heap } from 'heap-js';
 import { AsyncLocalStorage } from 'node:async_hooks';
-import { ReadableStream } from 'node:stream/web';
+import { ReadableStream, TransformStream } from 'node:stream/web';
 import { type ChatContext, ChatMessage } from '../llm/chat_context.js';
 import {
   type ChatItem,
@@ -485,8 +485,24 @@ export class AgentActivity implements RecognitionHooks {
     void this.audioStream.close();
     this.audioStream = new MultiInputStream<AudioFrame>();
 
+    // Filter is applied on this.audioStream.stream (downstream of MultiInputStream) rather
+    // than on the source audioStream via pipeThrough. pipeThrough locks its source stream, so
+    // if it were applied directly on audioStream, that lock would survive MultiInputStream.close()
+    // and make audioStream permanently locked for subsequent attachAudioInput calls (e.g. handoff).
+    const aecWarmupAudioFilter = new TransformStream<AudioFrame, AudioFrame>({
+      transform: (frame, controller) => {
+        const shouldDiscardForAecWarmup =
+          this.agentSession.agentState === 'speaking' && this.agentSession._aecWarmupRemaining > 0;
+        if (!shouldDiscardForAecWarmup) {
+          controller.enqueue(frame);
+        }
+      },
+    });
+
     this.audioStreamId = this.audioStream.addInputStream(audioStream);
-    const [realtimeAudioStream, recognitionAudioStream] = this.audioStream.stream.tee();
+    const [realtimeAudioStream, recognitionAudioStream] = this.audioStream.stream
+      .pipeThrough(aecWarmupAudioFilter)
+      .tee();
 
     if (this.realtimeSession) {
       this.realtimeSession.setInputAudioStream(realtimeAudioStream);
@@ -755,6 +771,11 @@ export class AgentActivity implements RecognitionHooks {
   }
 
   private interruptByAudioActivity(): void {
+    if (this.agentSession._aecWarmupRemaining > 0) {
+      // Disable interruption from audio activity while AEC warmup is active.
+      return;
+    }
+
     if (this.llm instanceof RealtimeModel && this.llm.capabilities.turnDetection) {
       // skip speech handle interruption if server side turn detection is enabled
       return;

agents/src/voice/agent_session.ts

@@ -77,6 +77,7 @@ export interface VoiceOptions {
   maxToolSteps: number;
   preemptiveGeneration: boolean;
   userAwayTimeout?: number | null;
+  aecWarmupDuration: number | null;
   useTtsAlignedTranscript: boolean;
 }
 
@@ -90,6 +91,7 @@ const defaultVoiceOptions: VoiceOptions = {
   maxToolSteps: 3,
   preemptiveGeneration: false,
   userAwayTimeout: 15.0,
+  aecWarmupDuration: 3000,
   useTtsAlignedTranscript: true,
 } as const;
 
@@ -158,6 +160,8 @@ export class AgentSession<
   private closingTask: Promise<void> | null = null;
   private userAwayTimer: NodeJS.Timeout | null = null;
 
+  private _aecWarmupTimer: NodeJS.Timeout | null = null;
+
   // Connection options for STT, LLM, and TTS
   private _connOptions: ResolvedSessionConnectOptions;
 
@@ -169,6 +173,9 @@ export class AgentSession<
   private userSpeakingSpan?: Span;
   private agentSpeakingSpan?: Span;
 
+  /** @internal */
+  _aecWarmupRemaining = 0;
+
   /** @internal */
   _recorderIO?: RecorderIO;
 
@@ -241,6 +248,7 @@ export class AgentSession<
     // This is the "global" chat context, it holds the entire conversation history
     this._chatCtx = ChatContext.empty();
     this.options = { ...defaultVoiceOptions, ...voiceOptions };
+    this._aecWarmupRemaining = this.options.aecWarmupDuration ?? 0;
 
     this._onUserInputTranscribed = this._onUserInputTranscribed.bind(this);
     this.on(AgentSessionEventTypes.UserInputTranscribed, this._onUserInputTranscribed);
@@ -845,6 +853,14 @@ export class AgentSession<
       this.agentSpeakingSpan = undefined;
     }
 
+    if (state === 'speaking' && this._aecWarmupRemaining > 0 && this._aecWarmupTimer === null) {
+      this._aecWarmupTimer = setTimeout(() => this._onAecWarmupExpired(), this._aecWarmupRemaining);
+      this.logger.debug(
+        { warmupDurationMs: this._aecWarmupRemaining },
+        'aec warmup active, disabling interruptions',
+      );
+    }
+
     const oldState = this._agentState;
     this._agentState = state;
 
@@ -938,6 +954,19 @@ export class AgentSession<
     }
   }
 
+  /** @internal */
+  _onAecWarmupExpired(): void {
+    if (this._aecWarmupRemaining > 0) {
+      this.logger.debug('aec warmup expired, re-enabling interruptions');
+    }
+
+    this._aecWarmupRemaining = 0;
+    if (this._aecWarmupTimer !== null) {
+      clearTimeout(this._aecWarmupTimer);
+      this._aecWarmupTimer = null;
+    }
+  }
+
   private _onUserInputTranscribed(ev: UserInputTranscribedEvent): void {
     if (this.userState === 'away' && ev.isFinal) {
       this.logger.debug('User returned from away state due to speech input');
@@ -969,6 +998,7 @@ export class AgentSession<
     }
 
     this._cancelUserAwayTimer();
+    this._onAecWarmupExpired();
     this.off(AgentSessionEventTypes.UserInputTranscribed, this._onUserInputTranscribed);
 
     if (this.activity) {

examples/src/basic_agent.ts

@@ -71,6 +71,7 @@ export default defineAgent({
         // allow the LLM to generate a response while waiting for the end of turn
         preemptiveGeneration: true,
         useTtsAlignedTranscript: true,
+        aecWarmupDuration: 3000,
       },
       connOptions: {
         // Example of overriding the default connection options for the LLM/TTS/STT