From b25ad21941351f7569c24d07f8ea44a2c6dd061a Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 12 Dec 2025 08:51:35 -0300 Subject: [PATCH] Refactoring HeyGenVideoService and HeyGenTransport to work with both APIs. --- src/pipecat/services/heygen/client.py | 88 +++++++++++++++++----- src/pipecat/services/heygen/video.py | 20 +++-- src/pipecat/transports/heygen/transport.py | 17 +++-- 3 files changed, 94 insertions(+), 31 deletions(-) diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index ba517090c..cd62b1af0 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -16,7 +16,8 @@ import base64 import json import time import uuid -from typing import Awaitable, Callable, Optional +from enum import Enum +from typing import Awaitable, Callable, Optional, Union import aiohttp from loguru import logger @@ -28,7 +29,12 @@ from pipecat.frames.frames import ( StartFrame, ) from pipecat.processors.frame_processor import FrameProcessorSetup -from pipecat.services.heygen.api import HeyGenApi, HeyGenSession, NewSessionRequest +from pipecat.services.heygen.api_interactive_avatar import HeyGenApi, NewSessionRequest +from pipecat.services.heygen.api_liveavatar import ( + LiveAvatarApi, + LiveAvatarNewSessionRequest, +) +from pipecat.services.heygen.base_api import StandardSessionResponse from pipecat.transports.base_transport import TransportParams from pipecat.utils.asyncio.task_manager import BaseTaskManager @@ -45,6 +51,13 @@ except ModuleNotFoundError as e: HEY_GEN_SAMPLE_RATE = 24000 +class ServiceType(Enum): + """Enum for HeyGen service types.""" + + INTERACTIVE_AVATAR = "INTERACTIVE_AVATAR" + LIVE_AVATAR = "LIVE_AVATAR" + + class HeyGenCallbacks(BaseModel): """Callback handlers for HeyGen events. @@ -78,10 +91,8 @@ class HeyGenClient: api_key: str, session: aiohttp.ClientSession, params: TransportParams, - session_request: NewSessionRequest = NewSessionRequest( - avatarName="Shawn_Therapist_public", - version="v2", - ), + session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None, + service_type: Optional[ServiceType] = None, callbacks: HeyGenCallbacks, connect_as_user: bool = False, ) -> None: @@ -91,12 +102,52 @@ class HeyGenClient: 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) + session_request: Configuration for the HeyGen session (optional) + service_type: Type of service to use callbacks: Callback handlers for HeyGen events connect_as_user: Whether to connect using the user token or not (default: False) """ - self._api = HeyGenApi(api_key, session=session) - self._heyGen_session: Optional[HeyGenSession] = None + # Set default service type for backwards compatibility + self._service_type = ( + service_type if service_type is not None else ServiceType.INTERACTIVE_AVATAR + ) + + # Validate session_request matches service_type if both are provided + if session_request is not None and service_type is not None: + if service_type == ServiceType.LIVE_AVATAR and not isinstance( + session_request, LiveAvatarNewSessionRequest + ): + logger.warning( + f"Service type is LIVE_AVATAR but session_request is not SessionTokenRequest. Ignoring session_request." + ) + session_request = None + elif service_type == ServiceType.INTERACTIVE_AVATAR and not isinstance( + session_request, NewSessionRequest + ): + logger.warning( + f"Service type is INTERACTIVE_AVATAR but session_request is not NewSessionRequest. Ignoring session_request." + ) + session_request = None + + # Create default session_request based on service_type if not provided + if session_request is None: + if self._service_type == ServiceType.INTERACTIVE_AVATAR: + session_request = NewSessionRequest( + avatar_id="Shawn_Therapist_public", + version="v2", + ) + else: # LIVE_AVATAR + session_request = LiveAvatarNewSessionRequest( + avatar_id="1c690fe7-23e0-49f9-bfba-14344450285b" + ) + + # Initialize API based on service type + if self._service_type == ServiceType.INTERACTIVE_AVATAR: + self._api = HeyGenApi(api_key, session=session) + else: + self._api = LiveAvatarApi(api_key, session=session) + + self._heyGen_session: Optional[StandardSessionResponse] = None self._websocket = None self._task_manager: Optional[BaseTaskManager] = None self._params = params @@ -130,14 +181,12 @@ class HeyGenClient: 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.debug(f"HeyGen realtime_endpoint: {self._heyGen_session.ws_url}") + logger.debug(f"HeyGen livekit URL: {self._heyGen_session.livekit_url}") + logger.debug(f"HeyGen livekit token: {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}" + f"Full Link: https://meet.livekit.io/custom?liveKitUrl={self._heyGen_session.livekit_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: @@ -222,7 +271,7 @@ class HeyGenClient: return logger.debug(f"HeyGenClient ws connecting") self._websocket = await websocket_connect( - uri=self._heyGen_session.realtime_endpoint, + uri=self._heyGen_session.ws_url, ) self._connected = True self._receive_task = self._task_manager.create_task( @@ -509,7 +558,9 @@ class HeyGenClient: async def _livekit_connect(self): """Connect to LiveKit room.""" try: - logger.debug(f"HeyGenClient livekit connecting to room URL: {self._heyGen_session.url}") + logger.debug( + f"HeyGenClient livekit connecting to room URL: {self._heyGen_session.livekit_url}" + ) self._livekit_room = rtc.Room() @self._livekit_room.on("participant_connected") @@ -574,7 +625,8 @@ class HeyGenClient: if not self._connect_as_user else self._heyGen_session.access_token ) - await self._livekit_room.connect(self._heyGen_session.url, access_token) + + await self._livekit_room.connect(self._heyGen_session.livekit_url, 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( diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py index b2df15119..63df0fd9d 100644 --- a/src/pipecat/services/heygen/video.py +++ b/src/pipecat/services/heygen/video.py @@ -12,7 +12,7 @@ audio/video streaming capabilities through the HeyGen API. """ import asyncio -from typing import Optional +from typing import Optional, Union import aiohttp from loguru import logger @@ -37,8 +37,14 @@ from pipecat.frames.frames import ( ) 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.services.heygen.api_interactive_avatar import NewSessionRequest +from pipecat.services.heygen.api_liveavatar import LiveAvatarNewSessionRequest +from pipecat.services.heygen.client import ( + HEY_GEN_SAMPLE_RATE, + HeyGenCallbacks, + HeyGenClient, + ServiceType, +) from pipecat.transports.base_transport import TransportParams # Using the same values that we do in the BaseOutputTransport @@ -72,7 +78,8 @@ class HeyGenVideoService(AIService): *, api_key: str, session: aiohttp.ClientSession, - session_request: NewSessionRequest = NewSessionRequest(avatar_id="Shawn_Therapist_public"), + session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None, + service_type: Optional[ServiceType] = None, **kwargs, ) -> None: """Initialize the HeyGen video service. @@ -80,7 +87,8 @@ class HeyGenVideoService(AIService): 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) + session_request: Configuration for the HeyGen session + service_type: Service type for the avatar session **kwargs: Additional arguments passed to parent AIService """ super().__init__(**kwargs) @@ -91,6 +99,7 @@ class HeyGenVideoService(AIService): self._resampler = create_stream_resampler() self._is_interrupting = False self._session_request = session_request + self._service_type = service_type self._other_participant_has_joined = False self._event_id = None self._audio_chunk_size = 0 @@ -117,6 +126,7 @@ class HeyGenVideoService(AIService): audio_out_sample_rate=HEY_GEN_SAMPLE_RATE, ), session_request=self._session_request, + service_type=self._service_type, callbacks=HeyGenCallbacks( on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected, diff --git a/src/pipecat/transports/heygen/transport.py b/src/pipecat/transports/heygen/transport.py index 24f47a8aa..2d6b543fb 100644 --- a/src/pipecat/transports/heygen/transport.py +++ b/src/pipecat/transports/heygen/transport.py @@ -16,7 +16,7 @@ The module consists of three main components: - HeyGenTransport: Main transport implementation that coordinates input/output transports """ -from typing import Any, Optional +from typing import Any, Optional, Union import aiohttp from loguru import logger @@ -36,8 +36,9 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup -from pipecat.services.heygen.api import NewSessionRequest -from pipecat.services.heygen.client import HeyGenCallbacks, HeyGenClient +from pipecat.services.heygen.api_interactive_avatar import NewSessionRequest +from pipecat.services.heygen.api_liveavatar import LiveAvatarNewSessionRequest +from pipecat.services.heygen.client import HeyGenCallbacks, HeyGenClient, ServiceType from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams @@ -297,10 +298,8 @@ class HeyGenTransport(BaseTransport): params: HeyGenParams = HeyGenParams(), input_name: Optional[str] = None, output_name: Optional[str] = None, - session_request: NewSessionRequest = NewSessionRequest( - avatar_id="Shawn_Therapist_public", - version="v2", - ), + session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None, + service_type: Optional[ServiceType] = None, ): """Initialize the HeyGen transport. @@ -313,7 +312,8 @@ class HeyGenTransport(BaseTransport): params: HeyGen-specific configuration parameters (default: HeyGenParams()) input_name: Optional custom name for the input transport output_name: Optional custom name for the output transport - session_request: Configuration for the HeyGen session (default: uses Shawn_Therapist_public avatar) + session_request: Configuration for the HeyGen session + service_type: Service type for the avatar session Note: The transport will automatically join the same virtual room as the HeyGen Avatar @@ -326,6 +326,7 @@ class HeyGenTransport(BaseTransport): session=session, params=params, session_request=session_request, + service_type=service_type, callbacks=HeyGenCallbacks( on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected,