diff --git a/src/pipecat/services/heygen/api_interactive_avatar.py b/src/pipecat/services/heygen/api_interactive_avatar.py index c0cbfb1cf..4c50ceaa3 100644 --- a/src/pipecat/services/heygen/api_interactive_avatar.py +++ b/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) diff --git a/src/pipecat/services/heygen/api_liveavatar.py b/src/pipecat/services/heygen/api_liveavatar.py index 14e941852..ca5497f17 100644 --- a/src/pipecat/services/heygen/api_liveavatar.py +++ b/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") diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index cc6451a25..a02d515a7 100644 --- a/src/pipecat/services/heygen/client.py +++ b/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: - message = await self._websocket.recv() - parsed_message = json.loads(message) - await self._handle_ws_server_event(parsed_message) - except ConnectionClosedOK: - break - except Exception as e: - logger.error(f"Error processing WebSocket message: {e}") - break + try: + while self._connected: + try: + message = await self._websocket.recv() + parsed_message = json.loads(message) + await self._handle_ws_server_event(parsed_message) + except ConnectionClosedOK: + logger.debug("HeyGenClient: WebSocket closed normally") + 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): - """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 + async def _write_audio_sleep(self, audio_bytes: int): + """Simulate audio playback timing with appropriate delays. + + 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._next_send_time = time.monotonic() + self._send_interval + self._audio_seconds_sent += chunk_duration + 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. diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py index 11bc06ca2..d7ccc319d 100644 --- a/src/pipecat/services/heygen/video.py +++ b/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 - self._audio_chunk_size = int((HEY_GEN_SAMPLE_RATE * 2) / 25) - await self._client.start(frame, self._audio_chunk_size) + # First chunk: 400ms for faster initial response + self._first_chunk_size = int(HEY_GEN_SAMPLE_RATE * 2 * 0.4) # 19200 bytes + # 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 - streaming management. + HeyGen client. Uses 600ms for the first chunk of each utterance for faster + 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() diff --git a/src/pipecat/transports/heygen/transport.py b/src/pipecat/transports/heygen/transport.py index ac13c2fe9..82a0ae3d0 100644 --- a/src/pipecat/transports/heygen/transport.py +++ b/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 diff --git a/uv.lock b/uv.lock index 8e0cb6e98..49ae58ec4 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.11" resolution-markers = [ "python_full_version >= '3.14'",