HeyGen implementation for Pipecat - HeyGenVideoService

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

dot-env.template

@@ -128,3 +128,6 @@ SAMBANOVA_API_KEY=...
 
 # Sentry
 SENTRY_DSN=...
+
+# Heygen
+HEYGEN_API_KEY=...

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)

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" ]

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.

src/pipecat/services/heygen/__init__.py

@@ -0,0 +1,5 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#

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"]

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()

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()

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.