diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py new file mode 100644 index 000000000..a9bf8e293 --- /dev/null +++ b/examples/foundational/38a-local-smart-turn.py @@ -0,0 +1,108 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.turn.local_smart_turn import LocalSmartTurnAnalyzer +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.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection): + logger.info(f"Starting bot") + + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + end_of_turn_analyzer=LocalSmartTurnAnalyzer(), + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. 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, + 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( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=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": "Please introduce yourself to the user."}) + 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") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py new file mode 100644 index 000000000..c511ae2ba --- /dev/null +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -0,0 +1,51 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import numpy as np +from loguru import logger + +from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState + + +class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): + def __init__(self): + super().__init__() + self._audio_buffer = bytearray() + + logger.debug("Loading Local Smart Turn model...") + + # TODO: implement it + + logger.debug("Loaded Local Smart Turn") + + def analyze_audio(self, buffer: bytes) -> EndOfTurnState: + self._audio_buffer += buffer + + # TODO: we probably don't need this + # Checking if we have at least 6 seconds of audio + # if len(self._audio_buffer) < 16000 * 2 * 6: + # return EndOfTurnState.INCOMPLETE + + audio_int16 = np.frombuffer(self._audio_buffer, dtype=np.int16) + + # Divide by 32768 because we have signed 16-bit data. + audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + + # TODO: implement to use the smart turn + # for now it is always returning as complete only for testing it + prediction = 1 + + state = EndOfTurnState.COMPLETE if prediction == 1 else EndOfTurnState.INCOMPLETE + + if state == EndOfTurnState.COMPLETE: + # clears the buffer completely + self._audio_buffer = bytearray() + else: + # TODO: implement it + pass + + return state diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 153c2d65b..946feb063 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -172,9 +172,16 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, UserStoppedSpeakingFrame): logger.debug("User stopped speaking") await self.push_frame(frame) + + # TODO check, we probably should change here as well. + # if the end of turn is enabled, we should only stop interruption after this point if self.interruptions_allowed: await self._stop_interruption() await self.push_frame(StopInterruptionFrame()) + elif isinstance(frame, UserEndOfTurnFrame): + logger.debug("User end of turn") + await self.push_frame(frame) + # TODO: implement to handle interruptions # # Audio input @@ -220,7 +227,7 @@ class BaseInputTransport(FrameProcessor): ): new_eot_state = await self._end_of_turn_analyze(audio_frame) if new_eot_state != end_of_turn_state: - await self.push_frame(UserEndOfTurnFrame()) + await self._handle_user_interruption(UserEndOfTurnFrame()) return new_eot_state async def _audio_task_handler(self): @@ -239,9 +246,13 @@ class BaseInputTransport(FrameProcessor): # changes from QUIET to SPEAKING and vice versa. if self._params.vad_enabled: vad_state = await self._handle_vad(frame, vad_state) + # TODO: need to check if we need to keep it later + if vad_state == VADState.QUIET: + end_of_turn_state = EndOfTurnState.INCOMPLETE audio_passthrough = self._params.vad_audio_passthrough - if self._params.end_of_turn_analyzer: + # We only need to check for completion if the user is speaking + if self._params.end_of_turn_analyzer and VADState.QUIET != vad_state: end_of_turn_state = await self._handle_end_of_turn(frame, end_of_turn_state) # Push audio downstream if passthrough.