Improve HeyGen LiveAvatar plugin reliability and performance (#4312)

* Improve HeyGen LiveAvatar plugin reliability and performance

- Add WebSocket ready gate: wait for session.state_updated connected
  event before sending commands (prevents silently dropped messages)
- Add keep-alive mechanism: send session.keep_alive every 2.5 min to
  prevent 5-minute inactivity timeout
- Optimize audio chunking: 600ms first chunk for faster initial
  response, 1s subsequent chunks for efficient streaming
- Fix audio buffer flush: send remaining buffered audio on utterance
  end instead of discarding it
- Fix WS state cleanup: properly reset connected/ready state when
  WebSocket drops unexpectedly
- Add livekit_config passthrough in LiveAvatar session token creation
- Replace stray print() with logger.debug()

* Fix HeyGenOutputTransport.start() signature and use 400ms first chunk

- Update transport.py to match new client.start() signature (no
  audio_chunk_size param)
- Change first chunk size from 600ms to 400ms per feedback

* Fix transport audio resampling and client.start() error propagation

- Add audio resampling in HeyGenOutputTransport.write_audio_frame() to
  ensure audio is always 24kHz before sending to HeyGen (was sending
  at pipeline sample rate, causing garbled audio)
- Raise exception on WS ready timeout instead of silently returning,
  preventing transport from appearing ready when WS connection failed

* Fix session readiness gate to work with LITE mode

LITE mode does not send session.state_updated WS events. Instead,
use a dual-signal _session_ready event that fires on either:
- WS session.state_updated connected (FULL mode)
- LiveKit participant connected (LITE mode)

Also reorder start() to connect both WS and LiveKit before waiting,
since the WS events may depend on LiveKit being connected.

Verified with live sandbox session - all tests pass.

* Simplify session readiness to use only WS ready gate

Remove _session_ready dual-signal and use only _ws_ready, which fires
on the session.state_updated connected WS event. Increase timeout to
30s. LiveKit is connected before waiting so the WS event can arrive.

* Reduce WS ready gate timeout back to 10s

* Remove WS ready gate (session.state_updated not reliably received)

The session.state_updated connected event is not reliably received
via the websockets library. Remove the gate for now and assume the
session is ready after WS + LiveKit connect. Keep-alive, chunking,
buffer flush, state cleanup, and other improvements remain.

src/pipecat/services/heygen/api_interactive_avatar.py

@@ -225,7 +225,7 @@ class HeyGenApi(BaseAvatarApi):
             "activity_idle_timeout": request_data.activity_idle_timeout,
         }
         session_info = await self._request("/streaming.new", params)
-        print("heygen session info", session_info)
+        logger.debug(f"HeyGen session info: {session_info}")
 
         heygen_session = HeyGenSession.model_validate(session_info)
 

src/pipecat/services/heygen/api_liveavatar.py

@@ -276,6 +276,13 @@ class LiveAvatarApi(BaseAvatarApi):
             # Fall back to VP8 encoding if video_settings is not provided
             params["video_settings"] = {"encoding": VideoEncoding.VP8.value}
 
+        if request_data.livekit_config is not None:
+            params["livekit_config"] = {
+                "livekit_url": request_data.livekit_config.livekit_url,
+                "livekit_room": request_data.livekit_config.livekit_room,
+                "livekit_client_token": request_data.livekit_config.livekit_client_token,
+            }
+
         logger.debug(f"Creating LiveAvatar session token with params: {params}")
         response = await self._request("POST", "/sessions/token", params)
         logger.debug(f"LiveAvatar session token created")

src/pipecat/services/heygen/client.py

@@ -156,6 +156,7 @@ class HeyGenClient:
         self._in_sample_rate = 0
         self._out_sample_rate = 0
         self._connected = False
+        self._keep_alive_task = None
         self._session_request = session_request
         self._callbacks = callbacks
         self._event_queue: asyncio.Queue | None = None
@@ -170,7 +171,6 @@ class HeyGenClient:
         # would be sending it to quickly. Instead, we want to block to emulate
         # an audio device, this is what the send interval is. It will be
         # computed on StartFrame.
-        self._send_interval = 0
         self._next_send_time = 0
         self._audio_seconds_sent = 0.0
         self._transport_ready = False
@@ -232,7 +232,7 @@ class HeyGenClient:
         except Exception as e:
             logger.error(f"Exception during cleanup: {e}")
 
-    async def start(self, frame: StartFrame, audio_chunk_size: int) -> None:
+    async def start(self, frame: StartFrame) -> None:
         """Start the client and establish all necessary connections.
 
         Initializes WebSocket and LiveKit connections using the provided configuration.
@@ -240,7 +240,6 @@ class HeyGenClient:
 
         Args:
             frame: Initial configuration frame containing audio parameters
-            audio_chunk_size: Audio chunk size for output processing
         """
         if self._websocket:
             logger.debug("heygen client already started")
@@ -249,10 +248,11 @@ class HeyGenClient:
         logger.debug(f"HeyGenClient starting")
         self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
         self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
-        self._send_interval = (audio_chunk_size / self._out_sample_rate) / 2
-        logger.debug(f"HeyGenClient send_interval: {self._send_interval}")
         await self._ws_connect()
         await self._livekit_connect()
+        self._keep_alive_task = self._task_manager.create_task(
+            self._keep_alive_handler(), name="HeyGenClient_KeepAlive"
+        )
         self._call_event_callback(self._callbacks.on_connected)
 
     async def stop(self) -> None:
@@ -261,6 +261,9 @@ class HeyGenClient:
         Disconnects from WebSocket and LiveKit endpoints, and performs cleanup.
         """
         logger.debug(f"HeyGenVideoService stopping")
+        if self._keep_alive_task:
+            await self._task_manager.cancel_task(self._keep_alive_task)
+            self._keep_alive_task = None
         await self._ws_disconnect()
         await self._livekit_disconnect()
         await self.cleanup()
@@ -286,21 +289,29 @@ class HeyGenClient:
 
     async def _ws_receive_task_handler(self):
         """Handle incoming WebSocket messages."""
-        while self._connected:
+        try:
-            try:
+            while self._connected:
-                message = await self._websocket.recv()
+                try:
-                parsed_message = json.loads(message)
+                    message = await self._websocket.recv()
-                await self._handle_ws_server_event(parsed_message)
+                    parsed_message = json.loads(message)
-            except ConnectionClosedOK:
+                    await self._handle_ws_server_event(parsed_message)
-                break
+                except ConnectionClosedOK:
-            except Exception as e:
+                    logger.debug("HeyGenClient: WebSocket closed normally")
-                logger.error(f"Error processing WebSocket message: {e}")
+                    break
-                break
+                except Exception as e:
+                    logger.error(f"Error processing WebSocket message: {e}")
+                    break
+        finally:
+            self._connected = False
+            logger.debug("HeyGenClient: WS receive handler exited, state cleaned up")
 
     async def _handle_ws_server_event(self, event: dict) -> None:
         """Handle an event from HeyGen websocket."""
         event_type = event.get("type")
-        if event_type == "agent.state":
+        if event_type == "session.state_updated":
+            state = event.get("state")
+            logger.debug(f"HeyGenClient ws session state updated: {state}")
+        elif event_type == "agent.state":
             logger.debug(f"HeyGenClient ws received agent status: {event}")
         else:
             logger.trace(f"HeyGenClient ws received unknown event: {event_type}")
@@ -328,6 +339,22 @@ class HeyGenClient:
             logger.error(f"Error sending message to HeyGen websocket: {e}")
             raise e
 
+    async def _keep_alive_handler(self):
+        """Periodically send keep-alive to prevent session timeout (5 min inactivity limit)."""
+        while self._connected:
+            await asyncio.sleep(150)  # 2.5 minutes
+            if self._connected:
+                try:
+                    await self._ws_send(
+                        {
+                            "type": "session.keep_alive",
+                            "event_id": str(uuid.uuid4()),
+                        }
+                    )
+                    logger.debug("HeyGenClient: Sent keep-alive")
+                except Exception as e:
+                    logger.warning(f"HeyGenClient: Keep-alive failed: {e}")
+
     async def interrupt(self, event_id: str) -> None:
         """Interrupt the avatar's current action.
 
@@ -394,7 +421,7 @@ class HeyGenClient:
         """Send audio data to the agent speak.
 
         Args:
-            audio: Audio data in base64 encoded format
+            audio: Audio data as raw bytes (will be base64 encoded)
             event_id: Unique identifier for the event
         """
         audio_base64 = base64.b64encode(audio).decode("utf-8")
@@ -406,30 +433,36 @@ class HeyGenClient:
             }
         )
         # Simulate audio playback with a sleep.
-        await self._write_audio_sleep()
+        await self._write_audio_sleep(len(audio))
 
     def _reset_audio_timing(self):
         """Reset audio timing control variables."""
         self._audio_seconds_sent = 0.0
         self._next_send_time = 0
 
-    async def _write_audio_sleep(self):
+    async def _write_audio_sleep(self, audio_bytes: int):
-        """Simulate audio playback timing with appropriate delays."""
+        """Simulate audio playback timing with appropriate delays.
-        # Only sleep after we've sent the first second of audio
+
-        # This appears to reduce the latency to receive the answer from HeyGen
+        Args:
+            audio_bytes: Number of raw audio bytes sent (24kHz, 16-bit mono).
+        """
+        # Compute actual audio duration from bytes (24kHz, 16-bit mono = 2 bytes/sample)
+        chunk_duration = audio_bytes / (HEY_GEN_SAMPLE_RATE * 2)
+
+        # Skip sleeping for the first 3 seconds of audio to reduce initial latency
         if self._audio_seconds_sent < 3.0:
-            self._audio_seconds_sent += self._send_interval
+            self._audio_seconds_sent += chunk_duration
-            self._next_send_time = time.monotonic() + self._send_interval
+            self._next_send_time = time.monotonic() + chunk_duration
             return
 
-        # After first second, use normal timing
+        # After first 3 seconds, pace sends to match real-time playback
         current_time = time.monotonic()
         sleep_duration = max(0, self._next_send_time - current_time)
         if sleep_duration > 0:
             await asyncio.sleep(sleep_duration)
-            self._next_send_time += self._send_interval
+            self._next_send_time += chunk_duration
         else:
-            self._next_send_time = time.monotonic() + self._send_interval
+            self._next_send_time = time.monotonic() + chunk_duration
 
     async def agent_speak_end(self, event_id: str) -> None:
         """Send signaling that the agent has finished speaking.

src/pipecat/services/heygen/video.py

@@ -211,9 +211,11 @@ class HeyGenVideoService(AIService):
             frame: The start frame containing initialization parameters.
         """
         await super().start(frame)
-        # 40 ms of audio, match the default behavior from the output transport
+        # First chunk: 400ms for faster initial response
-        self._audio_chunk_size = int((HEY_GEN_SAMPLE_RATE * 2) / 25)
+        self._first_chunk_size = int(HEY_GEN_SAMPLE_RATE * 2 * 0.4)  # 19200 bytes
-        await self._client.start(frame, self._audio_chunk_size)
+        # Subsequent chunks: 1000ms for efficient streaming
+        self._chunk_size = int(HEY_GEN_SAMPLE_RATE * 2 * 1.0)  # 48000 bytes
+        await self._client.start(frame)
         await self._create_send_task()
 
     async def stop(self, frame: EndFrame):
@@ -339,12 +341,14 @@ class HeyGenVideoService(AIService):
         """Handle sending audio frames to the HeyGen client.
 
         Continuously processes audio frames from the queue and sends them to the
-        HeyGen client. Handles timeouts and silence detection for proper audio
+        HeyGen client. Uses 600ms for the first chunk of each utterance for faster
-        streaming management.
+        initial response, then 1000ms chunks for efficient streaming. Handles
+        timeouts and silence detection for proper audio streaming management.
         """
         sample_rate = self._client.out_sample_rate
         audio_buffer = bytearray()
         self._event_id = None
+        is_first_chunk = True
 
         while True:
             try:
@@ -355,20 +359,30 @@ class HeyGenVideoService(AIService):
                     # starting the new inference
                     if self._event_id is None:
                         self._event_id = str(frame.id)
+                        is_first_chunk = True
 
                     audio = await self._resampler.resample(
                         frame.audio, frame.sample_rate, sample_rate
                     )
                     audio_buffer.extend(audio)
-                    while len(audio_buffer) >= self._audio_chunk_size:
-                        chunk = audio_buffer[: self._audio_chunk_size]
-                        audio_buffer = audio_buffer[self._audio_chunk_size :]
 
+                    current_chunk_size = (
+                        self._first_chunk_size if is_first_chunk else self._chunk_size
+                    )
+                    while len(audio_buffer) >= current_chunk_size:
+                        chunk = audio_buffer[:current_chunk_size]
+                        audio_buffer = audio_buffer[current_chunk_size:]
                         await self._client.agent_speak(bytes(chunk), self._event_id)
+                        if is_first_chunk:
+                            is_first_chunk = False
+                            current_chunk_size = self._chunk_size
                 self._queue.task_done()
             except TimeoutError:
                 # Bot has stopped speaking
                 if self._event_id is not None:
+                    # Flush any remaining buffered audio
+                    if len(audio_buffer) > 0:
+                        await self._client.agent_speak(bytes(audio_buffer), self._event_id)
+                        audio_buffer.clear()
                     await self._client.agent_speak_end(self._event_id)
                     self._event_id = None
-                    audio_buffer.clear()

src/pipecat/transports/heygen/transport.py

@@ -21,6 +21,7 @@ from typing import Any
 import aiohttp
 from loguru import logger
 
+from pipecat.audio.utils import create_stream_resampler
 from pipecat.frames.frames import (
     AudioRawFrame,
     BotConnectedFrame,
@@ -40,7 +41,12 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.services.heygen.api_interactive_avatar import NewSessionRequest
 from pipecat.services.heygen.api_liveavatar import LiveAvatarNewSessionRequest
-from pipecat.services.heygen.client import HeyGenCallbacks, HeyGenClient, ServiceType
+from pipecat.services.heygen.client import (
+    HEY_GEN_SAMPLE_RATE,
+    HeyGenCallbacks,
+    HeyGenClient,
+    ServiceType,
+)
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -164,6 +170,7 @@ class HeyGenOutputTransport(BaseOutputTransport):
         super().__init__(params, **kwargs)
         self._client = client
         self._params = params
+        self._resampler = create_stream_resampler()
 
         # Whether we have seen a StartFrame already.
         self._initialized = False
@@ -195,7 +202,7 @@ class HeyGenOutputTransport(BaseOutputTransport):
             return
 
         self._initialized = True
-        await self._client.start(frame, self.audio_chunk_size)
+        await self._client.start(frame)
         await self.set_transport_ready(frame)
         self._client.transport_ready()
 
@@ -266,10 +273,15 @@ class HeyGenOutputTransport(BaseOutputTransport):
     async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the HeyGen transport.
 
+        Resamples audio to 24kHz if needed before sending.
+
         Args:
             frame: The audio frame to write.
         """
-        await self._client.agent_speak(bytes(frame.audio), self._event_id)
+        audio = frame.audio
+        if frame.sample_rate != HEY_GEN_SAMPLE_RATE:
+            audio = await self._resampler.resample(audio, frame.sample_rate, HEY_GEN_SAMPLE_RATE)
+        await self._client.agent_speak(bytes(audio), self._event_id)
         return True
 
 

uv.lock

@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 2
 requires-python = ">=3.11"
 resolution-markers = [
     "python_full_version >= '3.14'",