From f6164e3bdeafeef26058e8a430232b16b496fec8 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 24 Sep 2025 14:33:56 -0300 Subject: [PATCH] HeyGen implementation for Pipecat - HeyGenTransport. --- CHANGELOG.md | 5 + examples/foundational/43-heygen-transport.py | 113 ++++++ src/pipecat/services/heygen/api.py | 2 + src/pipecat/services/heygen/client.py | 12 +- src/pipecat/services/heygen/video.py | 1 + src/pipecat/transports/heygen/__init__.py | 0 src/pipecat/transports/heygen/transport.py | 381 +++++++++++++++++++ 7 files changed, 510 insertions(+), 4 deletions(-) create mode 100644 examples/foundational/43-heygen-transport.py create mode 100644 src/pipecat/transports/heygen/__init__.py create mode 100644 src/pipecat/transports/heygen/transport.py diff --git a/CHANGELOG.md b/CHANGELOG.md index fec450bcb..494fb338a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `HeyGenTransport`. 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/). When used, the + Pipecat bot joins the same virtual room as the HeyGen Avatar and the user. + - Added support to `TwilioFrameSerializer` for `region` and `edge` settings. - Added support for using universal `LLMContext` with: diff --git a/examples/foundational/43-heygen-transport.py b/examples/foundational/43-heygen-transport.py new file mode 100644 index 000000000..78aab9622 --- /dev/null +++ b/examples/foundational/43-heygen-transport.py @@ -0,0 +1,113 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContext, + LLMContextAggregatorPair, +) +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.transports.heygen.transport import HeyGenParams, HeyGenTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + transport = HeyGenTransport( + api_key=os.getenv("HEYGEN_API_KEY"), + session=session, + params=HeyGenParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), + ) + + 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")) + + 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 = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + 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([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/services/heygen/api.py b/src/pipecat/services/heygen/api.py index 210a79938..6c9a4cf0e 100644 --- a/src/pipecat/services/heygen/api.py +++ b/src/pipecat/services/heygen/api.py @@ -108,12 +108,14 @@ class HeyGenSession(BaseModel): Parameters: session_id (str): Unique identifier for the streaming session. access_token (str): Token for accessing the session securely. + livekit_agent_token (str): Token for HeyGen’s audio agents(Pipecat). realtime_endpoint (str): Real-time communication endpoint URL. url (str): Direct URL for the session. """ session_id: str access_token: str + livekit_agent_token: str realtime_endpoint: str url: str diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index 07da6feb1..1b464ce9c 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -393,7 +393,9 @@ class HeyGenClient: 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}") + logger.debug( + f"capture_participant_audio: {participant_id}, sample_rate: {self._in_sample_rate}" + ) self._audio_frame_callback = callback if self._audio_task is not None: logger.warning( @@ -407,7 +409,9 @@ class HeyGenClient: 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) + audio_stream = rtc.AudioStream( + track=track_pub.track, sample_rate=self._in_sample_rate + ) self._audio_task = self._task_manager.create_task( self._process_audio_frames(audio_stream), name="HeyGenClient_Receive_Audio" ) @@ -536,7 +540,7 @@ class HeyGenClient: and self._audio_task is None ): logger.debug(f"Creating audio stream processor for track: {publication.sid}") - audio_stream = rtc.AudioStream(track) + audio_stream = rtc.AudioStream(track=track, sample_rate=self._in_sample_rate) self._audio_task = self._task_manager.create_task( self._process_audio_frames(audio_stream), name="HeyGenClient_Receive_Audio" ) @@ -559,7 +563,7 @@ class HeyGenClient: ) await self._livekit_room.connect( - self._heyGen_session.url, self._heyGen_session.access_token + self._heyGen_session.url, self._heyGen_session.livekit_agent_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}") diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py index 1fe00daee..bf7fbeecd 100644 --- a/src/pipecat/services/heygen/video.py +++ b/src/pipecat/services/heygen/video.py @@ -110,6 +110,7 @@ class HeyGenVideoService(AIService): api_key=self._api_key, session=self._session, params=TransportParams( + audio_in_sample_rate=48000, audio_in_enabled=True, video_in_enabled=True, audio_out_enabled=True, diff --git a/src/pipecat/transports/heygen/__init__.py b/src/pipecat/transports/heygen/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/transports/heygen/transport.py b/src/pipecat/transports/heygen/transport.py new file mode 100644 index 000000000..24f47a8aa --- /dev/null +++ b/src/pipecat/transports/heygen/transport.py @@ -0,0 +1,381 @@ +# +# 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. + +The module consists of three main components: +- HeyGenInputTransport: Handles incoming audio and events from HeyGen conversations +- HeyGenOutputTransport: Manages outgoing audio and events to HeyGen conversations +- HeyGenTransport: Main transport implementation that coordinates input/output transports +""" + +from typing import Any, Optional + +import aiohttp +from loguru import logger + +from pipecat.frames.frames import ( + AudioRawFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + InterruptionFrame, + OutputAudioRawFrame, + StartFrame, + UserStartedSpeakingFrame, + 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.transports.base_input import BaseInputTransport +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.transports.base_transport import BaseTransport, TransportParams + + +class HeyGenInputTransport(BaseInputTransport): + """Input transport for receiving audio and events from HeyGen conversations. + + Handles incoming audio streams from participants and manages audio capture + from the Daily room connected to the HeyGen conversation. + """ + + def __init__( + self, + client: HeyGenClient, + params: TransportParams, + **kwargs, + ): + """Initialize the HeyGen input transport. + + Args: + client: The HeyGen transport client instance. + params: Transport configuration parameters. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(params, **kwargs) + self._client = client + self._params = params + # Whether we have seen a StartFrame already. + self._initialized = False + + async def setup(self, setup: FrameProcessorSetup): + """Setup the input transport. + + Args: + setup: The frame processor setup configuration. + """ + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + """Cleanup input transport resources.""" + await super().cleanup() + await self._client.cleanup() + + async def start(self, frame: StartFrame): + """Start the input transport. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + + if self._initialized: + return + + self._initialized = True + + await self.set_transport_ready(frame) + + async def stop(self, frame: EndFrame): + """Stop the input transport. + + Args: + frame: The end frame signaling transport shutdown. + """ + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + """Cancel the input transport. + + Args: + frame: The cancel frame signaling immediate cancellation. + """ + await super().cancel(frame) + await self._client.stop() + + async def start_capturing_audio(self, participant_id: str): + """Start capturing audio from a participant. + + Args: + participant_id: The participant to capture audio from. + """ + if self._params.audio_in_enabled: + logger.info(f"HeyGenTransport start capturing audio for participant {participant_id}") + await self._client.capture_participant_audio( + participant_id, self._on_participant_audio_data + ) + + async def _on_participant_audio_data(self, audio_frame: AudioRawFrame): + """Handle received participant audio data.""" + frame = InputAudioRawFrame( + audio=audio_frame.audio, + sample_rate=audio_frame.sample_rate, + num_channels=audio_frame.num_channels, + ) + await self.push_audio_frame(frame) + + +class HeyGenOutputTransport(BaseOutputTransport): + """Output transport for sending audio and events to HeyGen conversations. + + Handles outgoing audio streams to participants and manages the custom + audio track expected by the HeyGen platform. + """ + + def __init__( + self, + client: HeyGenClient, + params: TransportParams, + **kwargs, + ): + """Initialize the HeyGen output transport. + + Args: + client: The HeyGen transport client instance. + params: Transport configuration parameters. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(params, **kwargs) + self._client = client + self._params = params + + # Whether we have seen a StartFrame already. + self._initialized = False + self._event_id = None + + async def setup(self, setup: FrameProcessorSetup): + """Setup the output transport. + + Args: + setup: The frame processor setup configuration. + """ + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + """Cleanup output transport resources.""" + await super().cleanup() + await self._client.cleanup() + + async def start(self, frame: StartFrame): + """Start the output transport. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.start(frame, self.audio_chunk_size) + await self.set_transport_ready(frame) + self._client.transport_ready() + + async def stop(self, frame: EndFrame): + """Stop the output transport. + + Args: + frame: The end frame signaling transport shutdown. + """ + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + """Cancel the output transport. + + Args: + frame: The cancel frame signaling immediate cancellation. + """ + await super().cancel(frame) + await self._client.stop() + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame to the next processor in the pipeline. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + # The BotStartedSpeakingFrame and BotStoppedSpeakingFrame are created inside BaseOutputTransport + # This is a workaround, so we can more reliably be aware when the bot has started or stopped speaking + if direction == FrameDirection.DOWNSTREAM: + if isinstance(frame, BotStartedSpeakingFrame): + if self._event_id is not None: + logger.warning("self._event_id is already defined!") + self._event_id = str(frame.id) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._client.agent_speak_end(self._event_id) + self._event_id = None + await super().push_frame(frame, direction) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames and handle interruptions. + + Handles various types of frames including interruption events and user speaking states. + Updates the HeyGen client state based on the received frames. + + Args: + frame: The frame to process + direction: The direction of frame flow in the pipeline + + Note: + Special handling is implemented for: + - InterruptionFrame: Triggers interruption of current speech + - UserStartedSpeakingFrame: Initiates agent listening mode + - UserStoppedSpeakingFrame: Stops agent listening mode + """ + await super().process_frame(frame, direction) + if isinstance(frame, InterruptionFrame): + await self._client.interrupt(self._event_id) + await self.push_frame(frame, direction) + if isinstance(frame, UserStartedSpeakingFrame): + await self._client.start_agent_listening() + await self.push_frame(frame, direction) + elif isinstance(frame, UserStoppedSpeakingFrame): + await self._client.stop_agent_listening() + await self.push_frame(frame, direction) + + async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: + """Write an audio frame to the HeyGen transport. + + Args: + frame: The audio frame to write. + """ + await self._client.agent_speak(bytes(frame.audio), self._event_id) + return True + + +class HeyGenParams(TransportParams): + """Configuration parameters for the HeyGen transport. + + Parameters: + audio_in_enabled: Whether to enable audio input from participants. + audio_out_enabled: Whether to enable audio output to participants. + """ + + audio_in_enabled: bool = True + audio_out_enabled: bool = True + + +class HeyGenTransport(BaseTransport): + """Transport implementation for HeyGen video calls. + + When used, the Pipecat bot joins the same virtual room as the HeyGen Avatar and the user. + This is achieved by using `HeyGenTransport`, which initiates the conversation via + `HeyGenApi` and obtains a room URL that all participants connect to. + """ + + def __init__( + self, + session: aiohttp.ClientSession, + api_key: str, + params: HeyGenParams = HeyGenParams(), + input_name: Optional[str] = None, + output_name: Optional[str] = None, + session_request: NewSessionRequest = NewSessionRequest( + avatar_id="Shawn_Therapist_public", + version="v2", + ), + ): + """Initialize the HeyGen transport. + + Sets up a new HeyGen transport instance with the specified configuration for + handling video calls between the Pipecat bot and HeyGen Avatar. + + Args: + session: aiohttp session for making async HTTP requests + api_key: HeyGen API key for authentication + 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) + + Note: + The transport will automatically join the same virtual room as the HeyGen Avatar + and user through the HeyGenClient, which handles session initialization via HeyGenApi. + """ + super().__init__(input_name=input_name, output_name=output_name) + self._params = params + self._client = HeyGenClient( + api_key=api_key, + session=session, + params=params, + session_request=session_request, + callbacks=HeyGenCallbacks( + on_participant_connected=self._on_participant_connected, + on_participant_disconnected=self._on_participant_disconnected, + ), + ) + self._input: Optional[HeyGenInputTransport] = None + self._output: Optional[HeyGenOutputTransport] = None + self._HeyGen_participant_id = None + + # Register supported handlers. The user will only be able to register + # these handlers. + self._register_event_handler("on_client_connected") + self._register_event_handler("on_client_disconnected") + + async def _on_participant_disconnected(self, participant_id: str): + logger.debug(f"HeyGen participant {participant_id} disconnected") + if participant_id != "heygen": + await self._on_client_disconnected(participant_id) + + async def _on_participant_connected(self, participant_id: str): + logger.debug(f"HeyGen participant {participant_id} connected") + if participant_id != "heygen": + await self._on_client_connected(participant_id) + if self._input: + await self._input.start_capturing_audio(participant_id) + + def input(self) -> FrameProcessor: + """Get the input transport for receiving media and events. + + Returns: + The HeyGen input transport instance. + """ + if not self._input: + self._input = HeyGenInputTransport(client=self._client, params=self._params) + return self._input + + def output(self) -> FrameProcessor: + """Get the output transport for sending media and events. + + Returns: + The HeyGen output transport instance. + """ + if not self._output: + self._output = HeyGenOutputTransport(client=self._client, params=self._params) + return self._output + + async def _on_client_connected(self, participant: Any): + """Handle client connected events.""" + await self._call_event_handler("on_client_connected", participant) + + async def _on_client_disconnected(self, participant: Any): + """Handle client disconnected events.""" + await self._call_event_handler("on_client_disconnected", participant)