diff --git a/CHANGELOG.md b/CHANGELOG.md index 8201001c3..3868effb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `HeyGenVideoService`. This is an integration for HeyGen Interactive Avatar. + A video service that handles audio streaming and requests HeyGen to generate + avatar video responses. (see https://www.heygen.com/) + - Added Async.ai TTS integration (https://async.ai/) - `AsyncAITTSService` – WebSocket-based streaming TTS with interruption support - `AsyncAIHttpTTSService` – HTTP-based streaming TTS service diff --git a/dot-env.template b/dot-env.template index 4f96b8d73..696c2ca0a 100644 --- a/dot-env.template +++ b/dot-env.template @@ -128,3 +128,6 @@ SAMBANOVA_API_KEY=... # Sentry SENTRY_DSN=... + +# Heygen +HEYGEN_API_KEY=... \ No newline at end of file diff --git a/examples/foundational/43a-heygen-video-service.py b/examples/foundational/43a-heygen-video-service.py new file mode 100644 index 000000000..7dfa5dd28 --- /dev/null +++ b/examples/foundational/43a-heygen-video-service.py @@ -0,0 +1,123 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.heygen.video import HeyGenVideoService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, + video_out_width=1280, + video_out_height=720, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, + video_out_width=1280, + video_out_height=720, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): + logger.info(f"Starting bot") + async with aiohttp.ClientSession() as session: + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="00967b2f-88a6-4a31-8153-110a92134b9f", + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + heyGen = HeyGenVideoService(api_key=os.getenv("HEYGEN_API_KEY"), session=session) + + messages = [ + { + "role": "system", + "content": "You are a helpful assistant. Your output will be converted to audio so don't include special characters in your answers. Be succinct and respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + heyGen, # Avatar + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by saying 'Hello' and then a short greeting.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=handle_sigint) + + await runner.run(task) + + +if __name__ == "__main__": + from pipecat.examples.run import main + + main(run_example, transport_params=transport_params) diff --git a/pyproject.toml b/pyproject.toml index 8c57bc8d0..9c9014762 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,6 +63,7 @@ google = [ "google-cloud-speech~=2.32.0", "google-cloud-texttospeech~=2.26.0", " grok = [] groq = [ "groq~=0.23.0" ] gstreamer = [ "pygobject~=3.50.0" ] +heygen = [ "livekit>=0.22.0", "websockets>=13.1,<15.0" ] inworld = [] krisp = [ "pipecat-ai-krisp~=0.4.0" ] koala = [ "pvkoala~=2.0.3" ] diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e75a56fd9..498c6e485 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1196,6 +1196,16 @@ class StopFrame(ControlFrame): pass +@dataclass +class OutputTransportReadyFrame(ControlFrame): + """Frame indicating that the output transport is ready. + + Indicates that the output transport is ready and able to receive frames. + """ + + pass + + @dataclass class HeartbeatFrame(ControlFrame): """Frame used by pipeline task to monitor pipeline health. diff --git a/src/pipecat/services/heygen/__init__.py b/src/pipecat/services/heygen/__init__.py new file mode 100644 index 000000000..d23112945 --- /dev/null +++ b/src/pipecat/services/heygen/__init__.py @@ -0,0 +1,5 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# diff --git a/src/pipecat/services/heygen/api.py b/src/pipecat/services/heygen/api.py new file mode 100644 index 000000000..210a79938 --- /dev/null +++ b/src/pipecat/services/heygen/api.py @@ -0,0 +1,281 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""HeyGen API. + +API to communicate with HeyGen Streaming API. +""" + +from enum import Enum +from typing import Any, Dict, Literal, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel, Field + + +class AvatarQuality(str, Enum): + """Enum representing different avatar quality levels.""" + + low = "low" + medium = "medium" + high = "high" + + +class VideoEncoding(str, Enum): + """Enum representing the video encoding.""" + + H264 = "H264" + VP8 = "VP8" + + +class VoiceEmotion(str, Enum): + """Enum representing different voice emotion types.""" + + EXCITED = "excited" + SERIOUS = "serious" + FRIENDLY = "friendly" + SOOTHING = "soothing" + BROADCASTER = "broadcaster" + + +class ElevenLabsSettings(BaseModel): + """Settings for ElevenLabs voice configuration. + + Parameters: + stability (Optional[float]): Stability of the voice synthesis. + similarity_boost (Optional[float]): Adjustment for similarity in voice performance. + model_id (Optional[str]): Identifier for the ElevenLabs model to use. + style (Optional[int]): Style metric to apply for the voice. + use_speaker_boost (Optional[bool]): Flag to enable speaker boost. + """ + + stability: Optional[float] = None + similarity_boost: Optional[float] = None + model_id: Optional[str] = None + style: Optional[int] = None + use_speaker_boost: Optional[bool] = None + + +class VoiceSettings(BaseModel): + """Voice configuration settings. + + Parameters: + voice_id (Optional[str]): ID of the voice to be used. + rate (Optional[float]): Speaking rate for the voice. + emotion (Optional[VoiceEmotion]): Emotion tone for the voice. + elevenlabs_settings (Optional[ElevenLabsSettings]): Details for ElevenLabs configuration. + """ + + voice_id: Optional[str] = Field(None, alias="voiceId") + rate: Optional[float] = None + emotion: Optional[VoiceEmotion] = None + elevenlabs_settings: Optional[ElevenLabsSettings] = Field(None, alias="elevenlabsSettings") + + +class NewSessionRequest(BaseModel): + """Requesting model for creating a new HeyGen session. + + Parameters: + quality (Optional[AvatarQuality]): Desired quality of the avatar. + avatar_id (Optional[str]): Unique identifier for the avatar. + voice (Optional[VoiceSettings]): Voice configurations for the session. + video_encoding (Optional[VideoEncoding]): Desired encoding for the video stream. + knowledge_id (Optional[str]): Identifier for the knowledge base (if applicable). + knowledge_base (Optional[str]): Details of any external knowledge base. + version (Literal["v2"]): API version to use. + disable_idle_timeout (Optional[bool]): Flag to disable automatic idle timeout. + activity_idle_timeout (Optional[int]): Timeout in seconds for activity-based idle detection. + """ + + quality: Optional[AvatarQuality] = None + avatar_id: Optional[str] = None + voice: Optional[VoiceSettings] = None + video_encoding: Optional[VideoEncoding] = None + knowledge_id: Optional[str] = None + knowledge_base: Optional[str] = None + version: Literal["v2"] = "v2" + disable_idle_timeout: Optional[bool] = None + activity_idle_timeout: Optional[int] = None + + +class HeyGenSession(BaseModel): + """Response model for a HeyGen session. + + Parameters: + session_id (str): Unique identifier for the streaming session. + access_token (str): Token for accessing the session securely. + realtime_endpoint (str): Real-time communication endpoint URL. + url (str): Direct URL for the session. + """ + + session_id: str + access_token: str + realtime_endpoint: str + url: str + + +class HeygenApiError(Exception): + """Custom exception for HeyGen API errors.""" + + def __init__(self, message: str, status: int, response_text: str) -> None: + """Initialize the HeyGen API error. + + Args: + message: Error message + status: HTTP status code + response_text: Raw response text from the API + """ + super().__init__(message) + self.status = status + self.response_text = response_text + + +class HeyGenApi: + """HeyGen Streaming API client.""" + + BASE_URL = "https://api.heygen.com/v1" + + def __init__(self, api_key: str, session: aiohttp.ClientSession) -> None: + """Initialize the HeyGen API. + + Args: + api_key: HeyGen API key + session: Optional aiohttp client session + """ + self.api_key = api_key + self.session = session + + async def _request(self, path: str, params: Dict[str, Any], expect_data: bool = True) -> Any: + """Make a POST request to the HeyGen API. + + Args: + path: API endpoint path. + params: JSON-serializable parameters. + expect_data: Whether to expect and extract 'data' field from response (default: True). + + Returns: + Parsed JSON response data. + + Raises: + HeygenApiError: If the API response is not successful or data is missing when expected. + aiohttp.ClientError: For network-related errors. + """ + url = f"{self.BASE_URL}{path}" + headers = { + "x-api-key": self.api_key, + "Content-Type": "application/json", + } + + logger.debug(f"HeyGen API request: {url}") + + try: + async with self.session.post(url, json=params, headers=headers) as response: + if not response.ok: + response_text = await response.text() + logger.error(f"HeyGen API error: {response_text}") + raise HeygenApiError( + f"API request failed with status {response.status}", + response.status, + response_text, + ) + if expect_data: + json_data = await response.json() + data = json_data.get("data") + return data + return await response.text() + except aiohttp.ClientError as e: + logger.error(f"Network error while calling HeyGen API: {str(e)}") + raise + + async def new_session(self, request_data: NewSessionRequest) -> HeyGenSession: + """Create a new streaming session. + + https://docs.heygen.com/reference/new-session + + Args: + request_data: Session configuration parameters. + + Returns: + Session information, including ID and access token. + """ + params = { + "quality": request_data.quality, + "avatar_id": request_data.avatar_id, + "voice": { + "voice_id": request_data.voice.voiceId if request_data.voice else None, + "rate": request_data.voice.rate if request_data.voice else None, + "emotion": request_data.voice.emotion if request_data.voice else None, + "elevenlabs_settings": ( + request_data.voice.elevenlabsSettings if request_data.voice else None + ), + }, + "knowledge_id": request_data.knowledge_id, + "knowledge_base": request_data.knowledge_base, + "version": request_data.version, + "video_encoding": request_data.video_encoding, + "disable_idle_timeout": request_data.disable_idle_timeout, + "activity_idle_timeout": request_data.activity_idle_timeout, + } + session_info = await self._request("/streaming.new", params) + print("heygen session info", session_info) + + return HeyGenSession.model_validate(session_info) + + async def start_session(self, session_id: str) -> Any: + """Start the streaming session. + + https://docs.heygen.com/reference/start-session + + Args: + session_id: ID of the session to start. + + Returns: + Response data from the start session API call. + + Raises: + ValueError: If session ID is not set. + """ + if not session_id: + raise ValueError("Session ID is not set. Call new_session first.") + + params = { + "session_id": session_id, + } + return await self._request("/streaming.start", params) + + async def close_session(self, session_id: str) -> Any: + """Terminate an active the streaming session. + + https://docs.heygen.com/reference/close-session + + Args: + session_id: ID of the session to stop. + + Returns: + Response data from the stop session API call. + + Raises: + ValueError: If session ID is not set. + """ + if not session_id: + raise ValueError("Session ID is not set. Call new_session first.") + + params = { + "session_id": session_id, + } + return await self._request("/streaming.stop", params, expect_data=False) + + async def create_token(self) -> str: + """Create a streaming token. + + https://docs.heygen.com/reference/streaming-token + + Returns: + str: The generated access token for the streaming session + """ + token_info = await self._request("/streaming.create_token", {}) + return token_info["token"] diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py new file mode 100644 index 000000000..9bd66d6a8 --- /dev/null +++ b/src/pipecat/services/heygen/client.py @@ -0,0 +1,624 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""HeyGen implementation for Pipecat. + +This module provides integration with the HeyGen platform for creating conversational +AI applications with avatars. It manages conversation sessions and provides real-time +audio/video streaming capabilities through the HeyGen API. +""" + +import asyncio +import base64 +import json +import time +import uuid +from typing import Awaitable, Callable, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel + +from pipecat.frames.frames import ( + AudioRawFrame, + ImageRawFrame, + StartFrame, +) +from pipecat.processors.frame_processor import FrameProcessorSetup +from pipecat.services.heygen.api import HeyGenApi, HeyGenSession, NewSessionRequest +from pipecat.transports.base_transport import TransportParams +from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue + +try: + from livekit import rtc + from livekit.rtc._proto.video_frame_pb2 import VideoBufferType + from websockets.asyncio.client import connect as websocket_connect + from websockets.exceptions import ConnectionClosedOK +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use HeyGen, you need to `pip install pipecat-ai[heygen]`.") + raise Exception(f"Missing module: {e}") + +HEY_GEN_SAMPLE_RATE = 24000 + + +class HeyGenCallbacks(BaseModel): + """Callback handlers for HeyGen events. + + Parameters: + on_participant_connected: Called when a participant connects + on_participant_disconnected: Called when a participant disconnects + """ + + on_participant_connected: Callable[[str], Awaitable[None]] + on_participant_disconnected: Callable[[str], Awaitable[None]] + + +class HeyGenClient: + """A client for interacting with HeyGen's Interactive Avatar Realtime API. + + This client manages both WebSocket and LiveKit connections for real-time avatar streaming, + handling bi-directional audio/video communication and avatar control. It implements the API defined in + https://docs.heygen.com/docs/interactive-avatar-realtime-api + + The client manages the following connections: + 1. WebSocket connection for avatar control and audio streaming + 2. LiveKit connection for receiving avatar video and audio + + Attributes: + HEY_GEN_SAMPLE_RATE (int): The required sample rate for HeyGen's audio processing (24000 Hz) + """ + + def __init__( + self, + *, + api_key: str, + session: aiohttp.ClientSession, + params: TransportParams, + session_request: NewSessionRequest = NewSessionRequest( + avatarName="Shawn_Therapist_public", + version="v2", + ), + callbacks: HeyGenCallbacks, + ) -> None: + """Initialize the HeyGen client. + + Args: + api_key: HeyGen API key for authentication + session: HTTP client session for API requests + params: Transport configuration parameters + session_request: Configuration for the HeyGen session (default: uses Shawn_Therapist_public avatar) + callbacks: Callback handlers for HeyGen events + """ + self._api = HeyGenApi(api_key, session=session) + self._heyGen_session: Optional[HeyGenSession] = None + self._websocket = None + self._task_manager: Optional[BaseTaskManager] = None + self._params = params + self._in_sample_rate = 0 + self._out_sample_rate = 0 + self._connected = False + self._session_request = session_request + self._callbacks = callbacks + self._event_queue: Optional[WatchdogQueue] = None + self._event_task = None + # Currently supporting to capture the audio and video from a single participant + self._video_task = None + self._audio_task = None + self._video_frame_callback = None + self._audio_frame_callback = None + # write_audio_frame() is called quickly, as soon as we get audio + # (e.g. from the TTS), and since this is just a network connection we + # 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 + + async def _initialize(self): + self._heyGen_session = await self._api.new_session(self._session_request) + logger.debug(f"HeyGen sessionId: {self._heyGen_session.session_id}") + logger.debug(f"HeyGen realtime_endpoint: {self._heyGen_session.realtime_endpoint}") + logger.debug(f"HeyGen livekit URL: {self._heyGen_session.url}") + logger.debug(f"HeyGen livekit toke: {self._heyGen_session.access_token}") + logger.info( + f"Full Link: https://meet.livekit.io/custom?liveKitUrl={self._heyGen_session.url}&token={self._heyGen_session.access_token}" + ) + + await self._api.start_session(self._heyGen_session.session_id) + logger.info("HeyGen session started") + + async def setup(self, setup: FrameProcessorSetup) -> None: + """Setup the client and initialize the conversation. + + Establishes a new session with HeyGen's API if one doesn't exist. + + Args: + setup: The frame processor setup configuration. + """ + if self._heyGen_session is not None: + logger.debug("heygen_session already initialized") + return + self._task_manager = setup.task_manager + try: + await self._initialize() + + self._event_queue = WatchdogQueue(self._task_manager) + self._event_task = self._task_manager.create_task( + self._callback_task_handler(self._event_queue), + f"{self}::event_callback_task", + ) + except Exception as e: + logger.error(f"Failed to setup HeyGenClient: {e}") + await self.cleanup() + + async def cleanup(self) -> None: + """Cleanup client resources. + + Closes the active HeyGen session and resets internal state. + """ + try: + if self._heyGen_session is not None: + await self._api.close_session(self._heyGen_session.session_id) + self._heyGen_session = None + self._connected = False + + if self._event_task and self._task_manager: + self._event_queue.cancel() + await self._task_manager.cancel_task(self._event_task) + self._event_task = None + except Exception as e: + logger.exception(f"Exception during cleanup: {e}") + + async def start(self, frame: StartFrame, audio_chunk_size: int) -> None: + """Start the client and establish all necessary connections. + + Initializes WebSocket and LiveKit connections using the provided configuration. + Sets up audio processing with the specified sample rates. + + 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") + return + + 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() + + async def stop(self) -> None: + """Stop the client and terminate all connections. + + Disconnects from WebSocket and LiveKit endpoints, and performs cleanup. + """ + logger.debug(f"HeyGenVideoService stopping") + await self._ws_disconnect() + await self._livekit_disconnect() + await self.cleanup() + + # websocket connection methods + async def _ws_connect(self): + """Connect to HeyGen websocket endpoint.""" + try: + if self._websocket: + logger.debug(f"HeyGenClient ws already connected!") + return + logger.debug(f"HeyGenClient ws connecting") + self._websocket = await websocket_connect( + uri=self._heyGen_session.realtime_endpoint, + ) + self._connected = True + self._receive_task = self._task_manager.create_task( + self._ws_receive_task_handler(), name="HeyGenClient_Websocket" + ) + except Exception as e: + logger.error(f"{self} initialization error: {e}") + self._websocket = None + + async def _ws_receive_task_handler(self): + """Handle incoming WebSocket messages.""" + while self._connected: + try: + message = await asyncio.wait_for(self._websocket.recv(), timeout=1.0) + parsed_message = json.loads(message) + await self._handle_ws_server_event(parsed_message) + except asyncio.TimeoutError: + self._task_manager.task_reset_watchdog() + except ConnectionClosedOK: + break + except Exception as e: + logger.error(f"Error processing WebSocket message: {e}") + break + + 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": + logger.debug(f"HeyGenClient ws received agent status: {event}") + else: + logger.error(f"HeyGenClient ws received unknown event: {event_type}") + + async def _ws_disconnect(self) -> None: + """Disconnect from HeyGen websocket endpoint.""" + try: + self._connected = False + if self._websocket: + await self._websocket.close() + except Exception as e: + logger.error(f"{self} disconnect error: {e}") + finally: + self._websocket = None + + async def _ws_send(self, message: dict) -> None: + """Send a message to HeyGen websocket.""" + if not self._connected: + logger.debug(f"{self} websocket is not connected anymore.") + return + try: + if self._websocket: + await self._websocket.send(json.dumps(message)) + except Exception as e: + logger.error(f"Error sending message to HeyGen websocket: {e}") + raise e + + async def interrupt(self, event_id: str) -> None: + """Interrupt the avatar's current action. + + Stops the current animation/speech and returns the avatar to idle state. + Useful for handling user interruptions during avatar speech. + """ + logger.debug("HeyGenClient interrupt") + self._reset_audio_timing() + await self._ws_send( + { + "type": "agent.interrupt", + "event_id": event_id, + } + ) + + async def start_agent_listening(self) -> None: + """Start the avatar's listening animation. + + Triggers visual cues indicating the avatar is listening to user input. + """ + logger.debug("HeyGenClient start_agent_listening") + await self._ws_send( + { + "type": "agent.start_listening", + "event_id": str(uuid.uuid4()), + } + ) + + async def stop_agent_listening(self) -> None: + """Stop the avatar's listening animation. + + Returns the avatar to idle state from listening state. + """ + await self._ws_send( + { + "type": "agent.stop_listening", + "event_id": str(uuid.uuid4()), + } + ) + + def transport_ready(self) -> None: + """Indicates that the output transport is ready and able to receive frames.""" + self._transport_ready = True + + @property + def out_sample_rate(self) -> int: + """Get the output sample rate. + + Returns: + The output sample rate in Hz. + """ + return self._out_sample_rate + + @property + def in_sample_rate(self) -> int: + """Get the input sample rate. + + Returns: + The input sample rate in Hz. + """ + return self._in_sample_rate + + async def agent_speak(self, audio: bytes, event_id: str) -> None: + """Send audio data to the agent speak. + + Args: + audio: Audio data in base64 encoded format + event_id: Unique identifier for the event + """ + audio_base64 = base64.b64encode(audio).decode("utf-8") + await self._ws_send( + { + "type": "agent.speak", + "audio": audio_base64, + "event_id": event_id, + } + ) + # Simulate audio playback with a sleep. + await self._write_audio_sleep() + + 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 + if self._audio_seconds_sent < 1.0: + self._audio_seconds_sent += self._send_interval + self._next_send_time = time.monotonic() + self._send_interval + return + + # After first second, use normal timing + 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 + else: + self._next_send_time = time.monotonic() + self._send_interval + + async def agent_speak_end(self, event_id: str) -> None: + """Send signaling that the agent has finished speaking. + + Args: + event_id: Unique identifier for the event + """ + self._reset_audio_timing() + await self._ws_send( + { + "type": "agent.speak_end", + "event_id": event_id, + } + ) + + async def capture_participant_audio(self, participant_id: str, callback) -> None: + """Capture audio frames from the HeyGen avatar. + + Args: + participant_id: Identifier of the participant to capture audio from + callback: Async function to handle received audio frames + """ + logger.debug(f"capture_participant_audio: {participant_id}") + self._audio_frame_callback = callback + if self._audio_task is not None: + logger.warning( + "Trying to capture more than one audio stream. It is currently not supported." + ) + return + + # Check if we already have audio tracks and participant is connected + if self._livekit_room and participant_id in self._livekit_room.remote_participants: + participant = self._livekit_room.remote_participants[participant_id] + for track_pub in participant.track_publications.values(): + if track_pub.kind == rtc.TrackKind.KIND_AUDIO and track_pub.track is not None: + logger.debug(f"Starting audio capture for existing track: {track_pub.sid}") + audio_stream = rtc.AudioStream(track_pub.track) + self._audio_task = self._task_manager.create_task( + self._process_audio_frames(audio_stream), name="HeyGenClient_Receive_Audio" + ) + break + + async def capture_participant_video(self, participant_id: str, callback) -> None: + """Capture video frames from the HeyGen avatar. + + Args: + participant_id: Identifier of the participant to capture video from + callback: Async function to handle received video frames + """ + logger.debug(f"capture_participant_video: {participant_id}") + self._video_frame_callback = callback + if self._video_task is not None: + logger.warning( + "Trying to capture more than one audio stream. It is currently not supported." + ) + return + + # Check if we already have video tracks and participant is connected + if self._livekit_room and participant_id in self._livekit_room.remote_participants: + participant = self._livekit_room.remote_participants[participant_id] + for track_pub in participant.track_publications.values(): + if track_pub.kind == rtc.TrackKind.KIND_VIDEO and track_pub.track is not None: + logger.debug(f"Starting video capture for existing track: {track_pub.sid}") + video_stream = rtc.VideoStream(track_pub.track) + self._video_task = self._task_manager.create_task( + self._process_video_frames(video_stream), name="HeyGenClient_Receive_Video" + ) + break + + # Livekit integration to receive audio and video + async def _process_audio_frames(self, stream: rtc.AudioStream): + """Process audio frames from LiveKit stream.""" + try: + logger.debug("Starting audio frame processing...") + async for frame_event in stream: + try: + audio_frame = frame_event.frame + # Convert audio to raw bytes + audio_data = bytes(audio_frame.data) + + audio_frame = AudioRawFrame( + audio=audio_data, + sample_rate=audio_frame.sample_rate, + num_channels=1, # HeyGen uses mono audio + ) + if self._transport_ready and self._audio_frame_callback: + await self._audio_frame_callback(audio_frame) + + except Exception as e: + logger.error(f"Error processing audio frame: {e}") + except Exception as e: + logger.error(f"Error processing audio frames: {e}") + finally: + logger.debug(f"Audio frame processing ended.") + + async def _process_video_frames(self, stream: rtc.VideoStream): + """Process video frames from LiveKit stream.""" + try: + logger.debug("Starting video frame processing...") + async for frame_event in stream: + try: + video_frame = frame_event.frame + + # Convert to RGB24 if not already + if video_frame.type != VideoBufferType.RGB24: + video_frame = video_frame.convert(VideoBufferType.RGB24) + + # Create frame with original dimensions + image_frame = ImageRawFrame( + image=bytes(video_frame.data), + size=(video_frame.width, video_frame.height), + format="RGB", + ) + image_frame.pts = frame_event.timestamp_us // 1000 # Convert to milliseconds + + if self._transport_ready and self._video_frame_callback: + await self._video_frame_callback(image_frame) + except Exception as e: + logger.error(f"Error processing individual video frame: {e}") + except Exception as e: + logger.error(f"Error processing video frames: {e}") + finally: + logger.debug(f"Video frame processing ended.") + + async def _livekit_connect(self): + """Connect to LiveKit room.""" + try: + logger.debug(f"HeyGenClient livekit connecting to room URL: {self._heyGen_session.url}") + self._livekit_room = rtc.Room() + + @self._livekit_room.on("participant_connected") + def on_participant_connected(participant: rtc.RemoteParticipant): + logger.debug( + f"Participant connected - SID: {participant.sid}, Identity: {participant.identity}" + ) + for track_pub in participant.track_publications.values(): + logger.debug( + f"Available track - SID: {track_pub.sid}, Kind: {track_pub.kind}, Name: {track_pub.name}" + ) + self._call_event_callback( + self._callbacks.on_participant_connected, participant.identity + ) + + @self._livekit_room.on("track_subscribed") + def on_track_subscribed( + track: rtc.Track, + publication: rtc.RemoteTrackPublication, + participant: rtc.RemoteParticipant, + ): + if ( + track.kind == rtc.TrackKind.KIND_VIDEO + and self._video_frame_callback is not None + and self._video_task is None + ): + logger.debug(f"Creating video stream processor for track: {publication.sid}") + video_stream = rtc.VideoStream(track) + self._video_task = self._task_manager.create_task( + self._process_video_frames(video_stream), name="HeyGenClient_Receive_Video" + ) + elif ( + track.kind == rtc.TrackKind.KIND_AUDIO + and self._audio_frame_callback is not None + and self._audio_task is None + ): + logger.debug(f"Creating audio stream processor for track: {publication.sid}") + audio_stream = rtc.AudioStream(track) + self._audio_task = self._task_manager.create_task( + self._process_audio_frames(audio_stream), name="HeyGenClient_Receive_Audio" + ) + + @self._livekit_room.on("track_unsubscribed") + def on_track_unsubscribed( + track: rtc.Track, + publication: rtc.RemoteTrackPublication, + participant: rtc.RemoteParticipant, + ): + logger.debug(f"Track unsubscribed - SID: {publication.sid}, Kind: {track.kind}") + + @self._livekit_room.on("participant_disconnected") + def on_participant_disconnected(participant: rtc.RemoteParticipant): + logger.debug( + f"Participant disconnected - SID: {participant.sid}, Identity: {participant.identity}" + ) + self._call_event_callback( + self._callbacks.on_participant_disconnected, participant.identity + ) + + await self._livekit_room.connect( + self._heyGen_session.url, self._heyGen_session.access_token + ) + logger.debug(f"Successfully connected to LiveKit room: {self._livekit_room.name}") + logger.debug(f"Local participant SID: {self._livekit_room.local_participant.sid}") + logger.debug( + f"Number of remote participants: {len(self._livekit_room.remote_participants)}" + ) + + # Log existing participants and their tracks + for participant in self._livekit_room.remote_participants.values(): + logger.debug( + f"Existing participant - SID: {participant.sid}, Identity: {participant.identity}" + ) + self._call_event_callback( + self._callbacks.on_participant_connected, participant.identity + ) + for track_pub in participant.track_publications.values(): + logger.debug( + f"Existing track - SID: {track_pub.sid}, Kind: {track_pub.kind}, Name: {track_pub.name}" + ) + + except Exception as e: + logger.error(f"LiveKit initialization error: {e}") + self._livekit_room = None + + async def _livekit_disconnect(self): + """Disconnect from LiveKit room.""" + try: + logger.debug("Starting LiveKit disconnect...") + if self._video_task: + await self._task_manager.cancel_task(self._video_task) + self._video_task = None + + if self._audio_task: + await self._task_manager.cancel_task(self._audio_task) + self._audio_task = None + + if self._livekit_room: + logger.debug("Disconnecting from LiveKit room") + await self._livekit_room.disconnect() + self._livekit_room = None + logger.debug("Successfully disconnected from LiveKit room") + except Exception as e: + logger.error(f"LiveKit disconnect error: {e}") + + # + # Queue callback handling + # + + def _call_event_callback(self, callback, *args): + """Queue an event callback for async execution.""" + self._event_queue.put_nowait((callback, *args)) + + async def _callback_task_handler(self, queue: asyncio.Queue): + """Handle queued callbacks from the specified queue.""" + while True: + (callback, *args) = await queue.get() + await callback(*args) + queue.task_done() diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py new file mode 100644 index 000000000..011151fdf --- /dev/null +++ b/src/pipecat/services/heygen/video.py @@ -0,0 +1,328 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""HeyGen implementation for Pipecat. + +This module provides integration with the HeyGen platform for creating conversational +AI applications with avatars. It manages conversation sessions and provides real-time +audio/video streaming capabilities through the HeyGen API. +""" + +import asyncio +from typing import Optional + +import aiohttp +from loguru import logger + +from pipecat.audio.utils import create_stream_resampler +from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, + EndFrame, + Frame, + ImageRawFrame, + OutputAudioRawFrame, + OutputImageRawFrame, + OutputTransportReadyFrame, + StartFrame, + TTSAudioRawFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup +from pipecat.services.ai_service import AIService +from pipecat.services.heygen.api import NewSessionRequest +from pipecat.services.heygen.client import HEY_GEN_SAMPLE_RATE, HeyGenCallbacks, HeyGenClient +from pipecat.transports.base_transport import TransportParams +from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue + +# Using the same values that we do in the BaseOutputTransport +AVATAR_VAD_STOP_SECS = 0.35 + + +class HeyGenVideoService(AIService): + """A service that integrates HeyGen's interactive avatar capabilities into the pipeline. + + This service manages the lifecycle of a HeyGen avatar session by handling bidirectional + audio/video streaming, avatar animations, and user interactions. It processes various frame types + to coordinate the avatar's behavior and maintains synchronization between audio and video streams. + + The service supports: + + - Real-time avatar animation based on audio input + - Voice activity detection for natural interactions + - Interrupt handling for more natural conversations + - Audio resampling for optimal quality + - Automatic session management + + Args: + api_key (str): HeyGen API key for authentication + session (aiohttp.ClientSession): HTTP client session for API requests + session_request (NewSessionRequest, optional): Configuration for the HeyGen session. + Defaults to using the "Shawn_Therapist_public" avatar with "v2" version. + """ + + def __init__( + self, + *, + api_key: str, + session: aiohttp.ClientSession, + session_request: NewSessionRequest = NewSessionRequest(avatar_id="Shawn_Therapist_public"), + **kwargs, + ) -> None: + """Initialize the HeyGen video service. + + Args: + api_key: HeyGen API key for authentication + session: HTTP client session for API requests + session_request: Configuration for the HeyGen session (default: uses Shawn_Therapist_public avatar) + **kwargs: Additional arguments passed to parent AIService + """ + super().__init__(**kwargs) + self._api_key = api_key + self._session = session + self._client: Optional[HeyGenClient] = None + self._send_task: Optional[asyncio.Task] = None + self._resampler = create_stream_resampler() + self._is_interrupting = False + self._session_request = session_request + self._other_participant_has_joined = False + self._event_id = None + self._audio_chunk_size = 0 + + async def setup(self, setup: FrameProcessorSetup): + """Set up the HeyGen video service with necessary configuration. + + Initializes the HeyGen client, establishes connections, and prepares the service + for audio/video processing. This includes setting up audio/video streams, + configuring callbacks, and initializing the resampler. + + Args: + setup: Configuration parameters for the frame processor. + """ + await super().setup(setup) + self._client = HeyGenClient( + api_key=self._api_key, + session=self._session, + params=TransportParams( + audio_in_enabled=True, + video_in_enabled=True, + audio_out_enabled=True, + audio_out_sample_rate=HEY_GEN_SAMPLE_RATE, + ), + session_request=self._session_request, + callbacks=HeyGenCallbacks( + on_participant_connected=self._on_participant_connected, + on_participant_disconnected=self._on_participant_disconnected, + ), + ) + await self._client.setup(setup) + + async def cleanup(self): + """Clean up the service and release resources. + + Terminates the HeyGen client session and cleans up associated resources. + """ + await super().cleanup() + await self._client.cleanup() + self._client = None + + async def _on_participant_connected(self, participant_id: str): + """Handle participant connected events.""" + logger.info(f"Participant connected {participant_id}") + if not self._other_participant_has_joined: + self._other_participant_has_joined = True + await self._client.capture_participant_video( + participant_id, self._on_participant_video_frame + ) + await self._client.capture_participant_audio( + participant_id, self._on_participant_audio_data + ) + + async def _on_participant_disconnected(self, participant_id: str): + """Handle participant disconnected events.""" + logger.info(f"Participant disconnected {participant_id}") + + async def _on_participant_video_frame(self, video_frame: ImageRawFrame): + """Handle incoming video frames from participants.""" + frame = OutputImageRawFrame( + image=video_frame.image, + size=video_frame.size, + format=video_frame.format, + ) + await self.push_frame(frame) + + async def _on_participant_audio_data(self, audio_frame: AudioRawFrame): + """Handle incoming audio data from participants.""" + frame = OutputAudioRawFrame( + audio=audio_frame.audio, + sample_rate=audio_frame.sample_rate, + num_channels=audio_frame.num_channels, + ) + await self.push_frame(frame) + + async def start(self, frame: StartFrame): + """Start the HeyGen video service and initialize the avatar session. + + Creates necessary tasks for audio/video processing and establishes + the connection with the HeyGen service. + + Args: + 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) + await self._create_send_task() + + async def stop(self, frame: EndFrame): + """Stop the HeyGen video service gracefully. + + Performs cleanup by ending the conversation and canceling ongoing tasks + in a controlled manner. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._end_conversation() + await self._cancel_send_task() + + async def cancel(self, frame: CancelFrame): + """Cancel the HeyGen video service. + + Performs an immediate termination of the service, cleaning up resources + without waiting for ongoing operations to complete. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + await self._end_conversation() + await self._cancel_send_task() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames and coordinate avatar behavior. + + Handles different types of frames to manage avatar interactions: + - UserStartedSpeakingFrame: Activates avatar's listening animation + - UserStoppedSpeakingFrame: Deactivates avatar's listening state + - TTSAudioRawFrame: Processes audio for avatar speech + - Other frames: Forwards them through the pipeline + + Args: + frame: The frame to be processed. + direction: The direction of frame processing (input/output). + """ + await super().process_frame(frame, direction) + + if isinstance(frame, UserStartedSpeakingFrame): + await self._handle_user_started_speaking() + await self.push_frame(frame, direction) + elif isinstance(frame, UserStoppedSpeakingFrame): + await self._client.stop_agent_listening() + await self.push_frame(frame, direction) + elif isinstance(frame, OutputTransportReadyFrame): + self._client.transport_ready() + await self.push_frame(frame, direction) + elif isinstance(frame, TTSAudioRawFrame): + await self._handle_audio_frame(frame) + else: + await self.push_frame(frame, direction) + + async def _handle_user_started_speaking(self): + """Handle the event when a user starts speaking. + + Manages the interruption flow by: + 1. Setting the interruption flag + 2. Signaling the client to interrupt current speech + 3. Canceling ongoing audio sending tasks + 4. Creating a new send task + 5. Activating the avatar's listening animation + """ + self._is_interrupting = True + await self._client.interrupt(self._event_id) + await self._cancel_send_task() + self._is_interrupting = False + await self._create_send_task() + await self._client.start_agent_listening() + + async def _end_conversation(self): + """End the current conversation and reset state. + + Stops the HeyGen client and cleans up conversation-specific resources. + """ + self._other_participant_has_joined = False + await self._client.stop() + + async def _create_send_task(self): + """Create the audio sending task if it doesn't exist. + + Initializes a new WatchdogQueue and creates a task for handling audio sending. + """ + if not self._send_task: + self._queue = WatchdogQueue(self.task_manager) + self._send_task = self.create_task(self._send_task_handler()) + + async def _cancel_send_task(self): + """Cancel the audio sending task if it exists. + + Cancels and cleans up the audio sending task and associated queue. + """ + if self._send_task: + self._queue.cancel() + await self.cancel_task(self._send_task) + self._send_task = None + + async def _handle_audio_frame(self, frame: OutputAudioRawFrame): + """Queue an audio frame for processing. + + Places the audio frame in the processing queue for synchronized + delivery to the HeyGen service. + + Args: + frame: The audio frame to process. + """ + await self._queue.put(frame) + + async def _send_task_handler(self): + """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. + """ + sample_rate = self._client.out_sample_rate + audio_buffer = bytearray() + self._event_id = None + + while True: + try: + frame = await asyncio.wait_for(self._queue.get(), timeout=AVATAR_VAD_STOP_SECS) + if self._is_interrupting: + break + if isinstance(frame, TTSAudioRawFrame): + # starting the new inference + if self._event_id is None: + self._event_id = str(frame.id) + + 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 :] + + await self._client.agent_speak(bytes(chunk), self._event_id) + self._queue.task_done() + except asyncio.TimeoutError: + # Bot has stopped speaking + if self._event_id is not None: + await self._client.agent_speak_end(self._event_id) + self._event_id = None + audio_buffer.clear() diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 4d1120cc7..4c60f7490 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -34,6 +34,7 @@ from pipecat.frames.frames import ( OutputDTMFFrame, OutputDTMFUrgentFrame, OutputImageRawFrame, + OutputTransportReadyFrame, SpriteFrame, StartFrame, StartInterruptionFrame, @@ -174,6 +175,9 @@ class BaseOutputTransport(FrameProcessor): ) await self._media_senders[destination].start(frame) + # Sending a frame indicating that the output transport is ready and able to receive frames. + await self.push_frame(OutputTransportReadyFrame(), FrameDirection.UPSTREAM) + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): """Send a transport message.