Refactoring HeyGenVideoService and HeyGenTransport to work with both APIs.

This commit is contained in:
Filipi Fuchter
2025-12-12 08:51:35 -03:00
parent debcea3baa
commit b25ad21941
3 changed files with 94 additions and 31 deletions

View File

@@ -16,7 +16,8 @@ import base64
import json import json
import time import time
import uuid import uuid
from typing import Awaitable, Callable, Optional from enum import Enum
from typing import Awaitable, Callable, Optional, Union
import aiohttp import aiohttp
from loguru import logger from loguru import logger
@@ -28,7 +29,12 @@ from pipecat.frames.frames import (
StartFrame, StartFrame,
) )
from pipecat.processors.frame_processor import FrameProcessorSetup 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.transports.base_transport import TransportParams
from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.asyncio.task_manager import BaseTaskManager
@@ -45,6 +51,13 @@ except ModuleNotFoundError as e:
HEY_GEN_SAMPLE_RATE = 24000 HEY_GEN_SAMPLE_RATE = 24000
class ServiceType(Enum):
"""Enum for HeyGen service types."""
INTERACTIVE_AVATAR = "INTERACTIVE_AVATAR"
LIVE_AVATAR = "LIVE_AVATAR"
class HeyGenCallbacks(BaseModel): class HeyGenCallbacks(BaseModel):
"""Callback handlers for HeyGen events. """Callback handlers for HeyGen events.
@@ -78,10 +91,8 @@ class HeyGenClient:
api_key: str, api_key: str,
session: aiohttp.ClientSession, session: aiohttp.ClientSession,
params: TransportParams, params: TransportParams,
session_request: NewSessionRequest = NewSessionRequest( session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None,
avatarName="Shawn_Therapist_public", service_type: Optional[ServiceType] = None,
version="v2",
),
callbacks: HeyGenCallbacks, callbacks: HeyGenCallbacks,
connect_as_user: bool = False, connect_as_user: bool = False,
) -> None: ) -> None:
@@ -91,12 +102,52 @@ class HeyGenClient:
api_key: HeyGen API key for authentication api_key: HeyGen API key for authentication
session: HTTP client session for API requests session: HTTP client session for API requests
params: Transport configuration parameters 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 callbacks: Callback handlers for HeyGen events
connect_as_user: Whether to connect using the user token or not (default: False) connect_as_user: Whether to connect using the user token or not (default: False)
""" """
self._api = HeyGenApi(api_key, session=session) # Set default service type for backwards compatibility
self._heyGen_session: Optional[HeyGenSession] = None 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._websocket = None
self._task_manager: Optional[BaseTaskManager] = None self._task_manager: Optional[BaseTaskManager] = None
self._params = params self._params = params
@@ -130,14 +181,12 @@ class HeyGenClient:
async def _initialize(self): async def _initialize(self):
self._heyGen_session = await self._api.new_session(self._session_request) 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 sessionId: {self._heyGen_session.session_id}")
logger.debug(f"HeyGen realtime_endpoint: {self._heyGen_session.realtime_endpoint}") logger.debug(f"HeyGen realtime_endpoint: {self._heyGen_session.ws_url}")
logger.debug(f"HeyGen livekit URL: {self._heyGen_session.url}") logger.debug(f"HeyGen livekit URL: {self._heyGen_session.livekit_url}")
logger.debug(f"HeyGen livekit toke: {self._heyGen_session.access_token}") logger.debug(f"HeyGen livekit token: {self._heyGen_session.access_token}")
logger.info( 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") logger.info("HeyGen session started")
async def setup(self, setup: FrameProcessorSetup) -> None: async def setup(self, setup: FrameProcessorSetup) -> None:
@@ -222,7 +271,7 @@ class HeyGenClient:
return return
logger.debug(f"HeyGenClient ws connecting") logger.debug(f"HeyGenClient ws connecting")
self._websocket = await websocket_connect( self._websocket = await websocket_connect(
uri=self._heyGen_session.realtime_endpoint, uri=self._heyGen_session.ws_url,
) )
self._connected = True self._connected = True
self._receive_task = self._task_manager.create_task( self._receive_task = self._task_manager.create_task(
@@ -509,7 +558,9 @@ class HeyGenClient:
async def _livekit_connect(self): async def _livekit_connect(self):
"""Connect to LiveKit room.""" """Connect to LiveKit room."""
try: 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 = rtc.Room()
@self._livekit_room.on("participant_connected") @self._livekit_room.on("participant_connected")
@@ -574,7 +625,8 @@ class HeyGenClient:
if not self._connect_as_user if not self._connect_as_user
else self._heyGen_session.access_token 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"Successfully connected to LiveKit room: {self._livekit_room.name}")
logger.debug(f"Local participant SID: {self._livekit_room.local_participant.sid}") logger.debug(f"Local participant SID: {self._livekit_room.local_participant.sid}")
logger.debug( logger.debug(

View File

@@ -12,7 +12,7 @@ audio/video streaming capabilities through the HeyGen API.
""" """
import asyncio import asyncio
from typing import Optional from typing import Optional, Union
import aiohttp import aiohttp
from loguru import logger from loguru import logger
@@ -37,8 +37,14 @@ from pipecat.frames.frames import (
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
from pipecat.services.ai_service import AIService from pipecat.services.ai_service import AIService
from pipecat.services.heygen.api import NewSessionRequest from pipecat.services.heygen.api_interactive_avatar import NewSessionRequest
from pipecat.services.heygen.client import HEY_GEN_SAMPLE_RATE, HeyGenCallbacks, HeyGenClient 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 from pipecat.transports.base_transport import TransportParams
# Using the same values that we do in the BaseOutputTransport # Using the same values that we do in the BaseOutputTransport
@@ -72,7 +78,8 @@ class HeyGenVideoService(AIService):
*, *,
api_key: str, api_key: str,
session: aiohttp.ClientSession, 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, **kwargs,
) -> None: ) -> None:
"""Initialize the HeyGen video service. """Initialize the HeyGen video service.
@@ -80,7 +87,8 @@ class HeyGenVideoService(AIService):
Args: Args:
api_key: HeyGen API key for authentication api_key: HeyGen API key for authentication
session: HTTP client session for API requests 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 **kwargs: Additional arguments passed to parent AIService
""" """
super().__init__(**kwargs) super().__init__(**kwargs)
@@ -91,6 +99,7 @@ class HeyGenVideoService(AIService):
self._resampler = create_stream_resampler() self._resampler = create_stream_resampler()
self._is_interrupting = False self._is_interrupting = False
self._session_request = session_request self._session_request = session_request
self._service_type = service_type
self._other_participant_has_joined = False self._other_participant_has_joined = False
self._event_id = None self._event_id = None
self._audio_chunk_size = 0 self._audio_chunk_size = 0
@@ -117,6 +126,7 @@ class HeyGenVideoService(AIService):
audio_out_sample_rate=HEY_GEN_SAMPLE_RATE, audio_out_sample_rate=HEY_GEN_SAMPLE_RATE,
), ),
session_request=self._session_request, session_request=self._session_request,
service_type=self._service_type,
callbacks=HeyGenCallbacks( callbacks=HeyGenCallbacks(
on_participant_connected=self._on_participant_connected, on_participant_connected=self._on_participant_connected,
on_participant_disconnected=self._on_participant_disconnected, on_participant_disconnected=self._on_participant_disconnected,

View File

@@ -16,7 +16,7 @@ The module consists of three main components:
- HeyGenTransport: Main transport implementation that coordinates input/output transports - HeyGenTransport: Main transport implementation that coordinates input/output transports
""" """
from typing import Any, Optional from typing import Any, Optional, Union
import aiohttp import aiohttp
from loguru import logger from loguru import logger
@@ -36,8 +36,9 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame, UserStoppedSpeakingFrame,
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.services.heygen.api import NewSessionRequest from pipecat.services.heygen.api_interactive_avatar import NewSessionRequest
from pipecat.services.heygen.client import HeyGenCallbacks, HeyGenClient 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_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -297,10 +298,8 @@ class HeyGenTransport(BaseTransport):
params: HeyGenParams = HeyGenParams(), params: HeyGenParams = HeyGenParams(),
input_name: Optional[str] = None, input_name: Optional[str] = None,
output_name: Optional[str] = None, output_name: Optional[str] = None,
session_request: NewSessionRequest = NewSessionRequest( session_request: Optional[Union[LiveAvatarNewSessionRequest, NewSessionRequest]] = None,
avatar_id="Shawn_Therapist_public", service_type: Optional[ServiceType] = None,
version="v2",
),
): ):
"""Initialize the HeyGen transport. """Initialize the HeyGen transport.
@@ -313,7 +312,8 @@ class HeyGenTransport(BaseTransport):
params: HeyGen-specific configuration parameters (default: HeyGenParams()) params: HeyGen-specific configuration parameters (default: HeyGenParams())
input_name: Optional custom name for the input transport input_name: Optional custom name for the input transport
output_name: Optional custom name for the output 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: Note:
The transport will automatically join the same virtual room as the HeyGen Avatar The transport will automatically join the same virtual room as the HeyGen Avatar
@@ -326,6 +326,7 @@ class HeyGenTransport(BaseTransport):
session=session, session=session,
params=params, params=params,
session_request=session_request, session_request=session_request,
service_type=service_type,
callbacks=HeyGenCallbacks( callbacks=HeyGenCallbacks(
on_participant_connected=self._on_participant_connected, on_participant_connected=self._on_participant_connected,
on_participant_disconnected=self._on_participant_disconnected, on_participant_disconnected=self._on_participant_disconnected,