diff --git a/dot-env.template b/dot-env.template index f0b5bdc0f..18ddbee0a 100644 --- a/dot-env.template +++ b/dot-env.template @@ -92,4 +92,8 @@ ASSEMBLYAI_API_KEY=... OPENROUTER_API_KEY=... # Piper -PIPER_BASE_URL=... \ No newline at end of file +PIPER_BASE_URL=... + +# Smart turn +LOCAL_SMART_TURN_MODEL_PATH= +REMOTE_SMART_TURN_URL= \ No newline at end of file diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py new file mode 100644 index 000000000..03d530b90 --- /dev/null +++ b/examples/foundational/38-smart-turn.py @@ -0,0 +1,111 @@ +# +# 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.smart_turn import SmartTurnAnalyzer +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +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") + + remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL") + + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + vad_audio_passthrough=True, + end_of_turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url), + ), + ) + + 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")) + + 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/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py new file mode 100644 index 000000000..7baedf10e --- /dev/null +++ b/examples/foundational/38a-local-smart-turn.py @@ -0,0 +1,129 @@ +# +# 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.base_smart_turn import SmartTurnParams +from pipecat.audio.turn.local_smart_turn import LocalCoreMLSmartTurnAnalyzer +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +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") + + # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH + # to the path where the smart-turn repo is cloned. + # + # Example setup: + # + # # Git LFS (Large File Storage) + # brew install git-lfs + # # Hugging Face uses LFS to store large model files, including .mlpackage + # git lfs install + # # Clone the repo with the smart_turn_classifier.mlpackage + # git clone https://huggingface.co/pipecat-ai/smart-turn + # + # Then set the env variable: + # export LOCAL_SMART_TURN_MODEL_PATH=./smart-turn + # or add it to your .env file + smart_turn_model_path = os.getenv("LOCAL_SMART_TURN_MODEL_PATH") + + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + vad_audio_passthrough=True, + end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer( + smart_turn_model_path=smart_turn_model_path, params=SmartTurnParams() + ), + ), + ) + + 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")) + + 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/pyproject.toml b/pyproject.toml index cb0cd3520..e10e6daf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -79,6 +79,8 @@ qwen = [] rime = [ "websockets~=13.1" ] riva = [ "nvidia-riva-client~=2.19.0" ] sentry = [ "sentry-sdk~=2.23.1" ] +local-smart-turn = [ "coremltools>=8.0", "transformers", "torch==2.5.0", "torchaudio==2.5.0" ] +remote-smart-turn = [] silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.10"] soundfile = [ "soundfile~=0.13.0" ] diff --git a/src/pipecat/audio/turn/__init__.py b/src/pipecat/audio/turn/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py new file mode 100644 index 000000000..0716d4e7c --- /dev/null +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -0,0 +1,182 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import time +from abc import ABC, abstractmethod +from enum import Enum +from typing import Dict, Optional + +import numpy as np +from loguru import logger +from pydantic import BaseModel + + +# Enum for end-of-turn detection states +class EndOfTurnState(Enum): + COMPLETE = 1 + INCOMPLETE = 2 + + +# Default timing parameters +STOP_SECS = 3 +PRE_SPEECH_MS = 0 +MAX_DURATION_SECONDS = 8 # Max allowed segment duration +USE_ONLY_LAST_VAD_SEGMENT = True + + +class SmartTurnParams(BaseModel): + stop_secs: float = STOP_SECS + pre_speech_ms: float = PRE_SPEECH_MS + max_duration_secs: float = MAX_DURATION_SECONDS + # not exposing this for now yet until the model can handle it. + # use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT + + +class BaseSmartTurn(ABC): + def __init__( + self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams() + ): + self._init_sample_rate = sample_rate + self._params = params + # Configuration + self._sample_rate = 0 + self._stop_ms = self._params.stop_secs * 1000 # silence threshold in ms + # Inference state + self._audio_buffer = [] + self._speech_triggered = False + self._silence_ms = 0 + self._speech_start_time = None + + @property + def sample_rate(self) -> int: + return self._sample_rate + + def set_sample_rate(self, sample_rate: int): + self._sample_rate = sample_rate + + @property + def speech_triggered(self) -> bool: + return self._speech_triggered + + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: + # Convert raw audio to float32 format and append to the buffer + audio_int16 = np.frombuffer(buffer, dtype=np.int16) + audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + self._audio_buffer.append((time.time(), audio_float32)) + + state = EndOfTurnState.INCOMPLETE + + if is_speech: + # Reset silence tracking on speech + self._silence_ms = 0 + self._speech_triggered = True + if self._speech_start_time is None: + self._speech_start_time = time.time() + logger.debug(f"Speech started at {self._speech_start_time}") + else: + if self._speech_triggered: + chunk_duration_ms = len(audio_int16) / (self._sample_rate / 1000) + self._silence_ms += chunk_duration_ms + # If silence exceeds threshold, mark end of turn + if self._silence_ms >= self._stop_ms: + logger.debug( + f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}" + ) + state = EndOfTurnState.COMPLETE + self._clear(state) + else: + # Trim buffer to prevent unbounded growth before speech + max_buffer_time = ( + (self._params.pre_speech_ms / 1000) + + self._params.stop_secs + + self._params.max_duration_secs + ) + while ( + self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time + ): + self._audio_buffer.pop(0) + + return state + + def analyze_end_of_turn(self) -> EndOfTurnState: + logger.debug("Analyzing End of Turn...") + state = self._process_speech_segment(self._audio_buffer) + if state == EndOfTurnState.COMPLETE or USE_ONLY_LAST_VAD_SEGMENT: + self._clear(state) + logger.debug(f"End of Turn result: {state}") + return state + + def _clear(self, turn_state: EndOfTurnState): + # Reset internal state for next turn + logger.debug("Clearing audio buffer...") + # If the state is still incomplete, keep the _speech_triggered as True + self._speech_triggered = turn_state == EndOfTurnState.INCOMPLETE + self._audio_buffer = [] + self._speech_start_time = None + self._silence_ms = 0 + + def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: + state = EndOfTurnState.INCOMPLETE + + if not audio_buffer: + return state + + # Extract recent audio segment for prediction + start_time = self._speech_start_time - (self._params.pre_speech_ms / 1000) + start_index = 0 + for i, (t, _) in enumerate(audio_buffer): + if t >= start_time: + start_index = i + break + + end_index = len(audio_buffer) - 1 + + # Extract the audio segment + segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] + segment_audio = np.concatenate(segment_audio_chunks) + + logger.debug(f"Segment audio chunks after start index: {len(segment_audio)}") + + # Limit maximum duration + max_samples = int(self._params.max_duration_secs * self.sample_rate) + if len(segment_audio) > max_samples: + # slices the array to keep the last max_samples samples, discarding the earlier part. + segment_audio = segment_audio[-max_samples:] + + logger.debug(f"Segment audio chunks after limiting duration: {len(segment_audio)}") + + if len(segment_audio) > 0: + start_time = time.perf_counter() + result = self._predict_endpoint(segment_audio) + state = ( + EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE + ) + end_time = time.perf_counter() + + logger.debug("--------") + logger.debug(f"Prediction: {'Complete' if result['prediction'] == 1 else 'Incomplete'}") + logger.debug(f"Probability of complete: {result['probability']:.4f}") + logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds") + else: + logger.debug(f"params: {self._params}, stop_ms: {self._stop_ms}") + logger.debug("Captured empty audio segment, skipping prediction.") + + return state + + @abstractmethod + def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, any]: + """ + Abstract method to predict if a turn has ended based on audio. + + Args: + buffer: Float32 numpy array of audio samples at 16kHz. + + Returns: + Dictionary with: + - prediction: 1 if turn is complete, else 0 + - probability: Confidence of the prediction + """ + pass 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..665e4b64d --- /dev/null +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -0,0 +1,65 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os +from typing import Dict + +import numpy as np +import torch +from loguru import logger + +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn + +try: + import coremltools as ct + from transformers import AutoFeatureExtractor +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use the LocalSmartTurnAnalyzer, you need to `pip install pipecat-ai[local-smart-turn]`." + ) + raise Exception(f"Missing module: {e}") + + +class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn): + def __init__(self, smart_turn_model_path: str, **kwargs): + super().__init__(**kwargs) + + if not smart_turn_model_path: + logger.error("smart_turn_model_path is not set.") + raise Exception("smart_turn_model_path must be provided.") + + core_ml_model_path = f"{smart_turn_model_path}/coreml/smart_turn_classifier.mlpackage" + + logger.debug("Loading Local Smart Turn model...") + # Only load the processor, not the torch model + self._turn_processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path) + self._turn_model = ct.models.MLModel(core_ml_model_path) + logger.debug("Loaded Local Smart Turn") + + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]: + inputs = self._turn_processor( + audio_array, + sampling_rate=16000, + padding="max_length", + truncation=True, + max_length=800, # Maximum length as specified in training + return_attention_mask=True, + return_tensors="pt", + ) + + output = self._turn_model.predict(dict(inputs)) + logits = output["logits"] # Core ML returns numpy array + logits_tensor = torch.tensor(logits) + probabilities = torch.nn.functional.softmax(logits_tensor, dim=1) + completion_prob = probabilities[0, 1].item() # Probability of class 1 (Complete) + prediction = 1 if completion_prob > 0.5 else 0 + + return { + "prediction": prediction, + "probability": completion_prob, + } diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py new file mode 100644 index 000000000..5378e4b5b --- /dev/null +++ b/src/pipecat/audio/turn/smart_turn.py @@ -0,0 +1,75 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import io +import os +from typing import Dict + +import numpy as np +import requests +from loguru import logger + +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn + + +class SmartTurnAnalyzer(BaseSmartTurn): + def __init__(self, url: str, **kwargs): + super().__init__(**kwargs) + self.remote_smart_turn_url = url + + if not self.remote_smart_turn_url: + logger.error("remote_smart_turn_url is not set.") + raise Exception("remote_smart_turn_url must be provided.") + + # Use a session to reuse connections (keep-alive) + self.session = requests.Session() + self.session.headers.update({"Connection": "keep-alive"}) + + def _serialize_array(self, audio_array: np.ndarray) -> bytes: + logger.trace("Serializing NumPy array to bytes...") + buffer = io.BytesIO() + np.save(buffer, audio_array) + serialized_bytes = buffer.getvalue() + logger.trace(f"Serialized size: {len(serialized_bytes)} bytes") + return serialized_bytes + + def _send_raw_request(self, data_bytes: bytes): + headers = {"Content-Type": "application/octet-stream"} + logger.trace( + f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." + ) + try: + response = self.session.post( + self.remote_smart_turn_url, + data=data_bytes, + headers=headers, + timeout=60, + ) + + logger.trace("\n--- Response ---") + logger.trace(f"Status Code: {response.status_code}") + + if response.ok: + try: + logger.trace("Response JSON:") + logger.trace(response.json()) + return response.json() + except requests.exceptions.JSONDecodeError: + logger.trace("Response Content (non-JSON):") + logger.trace(response.text) + else: + logger.trace("Response Content (Error):") + logger.trace(response.text) + response.raise_for_status() + + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send raw request to Daily Smart Turn: {e}") + raise Exception("Failed to send raw request to Daily Smart Turn.") + + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]: + serialized_array = self._serialize_array(audio_array) + return self._send_raw_request(serialized_array) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 26f386576..3c7cf868d 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -10,6 +10,7 @@ from typing import Optional from loguru import logger +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, EndOfTurnState from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState from pipecat.frames.frames import ( BotInterruptionFrame, @@ -64,12 +65,19 @@ class BaseInputTransport(FrameProcessor): def vad_analyzer(self) -> Optional[VADAnalyzer]: return self._params.vad_analyzer + @property + def end_of_turn_analyzer(self) -> Optional[BaseSmartTurn]: + return self._params.end_of_turn_analyzer + async def start(self, frame: StartFrame): self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate # Configure VAD analyzer. if self._params.vad_enabled and self._params.vad_analyzer: self._params.vad_analyzer.set_sample_rate(self._sample_rate) + # Configure End of turn analyzer. + if self._params.end_of_turn_analyzer: + self._params.end_of_turn_analyzer.set_sample_rate(self._sample_rate) # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) @@ -187,10 +195,18 @@ class BaseInputTransport(FrameProcessor): and new_vad_state != VADState.STOPPING ): frame = None - if new_vad_state == VADState.SPEAKING: - frame = UserStartedSpeakingFrame() - elif new_vad_state == VADState.QUIET: - frame = UserStoppedSpeakingFrame() + # If the turn analyser is enabled, this will prevent: + # - Creating the UserStoppedSpeakingFrame + # - Creating the UserStartedSpeakingFrame multiple times + can_create_user_frames = ( + self._params.end_of_turn_analyzer is None + or not self._params.end_of_turn_analyzer.speech_triggered + ) + if can_create_user_frames: + if new_vad_state == VADState.SPEAKING: + frame = UserStartedSpeakingFrame() + elif new_vad_state == VADState.QUIET: + frame = UserStoppedSpeakingFrame() if frame: await self._handle_user_interruption(frame) @@ -198,6 +214,29 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state + async def _handle_end_of_turn(self): + if self.end_of_turn_analyzer: + state = await self.get_event_loop().run_in_executor( + self._executor, self.end_of_turn_analyzer.analyze_end_of_turn + ) + await self._handle_end_of_turn_complete(state) + + async def _handle_end_of_turn_complete(self, state: EndOfTurnState): + if state == EndOfTurnState.COMPLETE: + await self._handle_user_interruption(UserStoppedSpeakingFrame()) + + async def _run_turn_analyzer( + self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState + ): + is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING + # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE + end_of_turn_state = self._params.end_of_turn_analyzer.append_audio(frame.audio, is_speech) + if end_of_turn_state == EndOfTurnState.COMPLETE: + await self._handle_end_of_turn_complete(end_of_turn_state) + # Otherwise we are going to trigger to check if the turn is completed based on the VAD + elif vad_state == VADState.QUIET and vad_state != previous_vad_state: + await self._handle_end_of_turn() + async def _audio_task_handler(self): vad_state: VADState = VADState.QUIET while True: @@ -211,10 +250,14 @@ class BaseInputTransport(FrameProcessor): # Check VAD and push event if necessary. We just care about # changes from QUIET to SPEAKING and vice versa. + previous_vad_state = vad_state if self._params.vad_enabled: vad_state = await self._handle_vad(frame, vad_state) audio_passthrough = self._params.vad_audio_passthrough + if self._params.end_of_turn_analyzer: + await self._run_turn_analyzer(frame, vad_state, previous_vad_state) + # Push audio downstream if passthrough. if audio_passthrough: await self.push_frame(frame) diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 411456071..79c876fc4 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -11,6 +11,7 @@ from pydantic import BaseModel, ConfigDict from pipecat.audio.filters.base_audio_filter import BaseAudioFilter from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.processors.frame_processor import FrameProcessor from pipecat.utils.base_object import BaseObject @@ -41,6 +42,7 @@ class TransportParams(BaseModel): vad_enabled: bool = False vad_audio_passthrough: bool = False vad_analyzer: Optional[VADAnalyzer] = None + end_of_turn_analyzer: Optional[BaseSmartTurn] = None class BaseTransport(BaseObject):