wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audio(turn): remove FalSmartTurnAnalyzer and LocalSmartTurnAnalyzer

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2026-03-30 14:04:29 -07:00
parent 136e6a58be
commit dd1bea2a5f
3 changed files with 0 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
examples/foundational/38-smart-turn-fal.py
View File

@@ -1,143 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import aiohttp
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.turn.smart_turn.fal_smart_turn import FalSmartTurnAnalyzer
from pipecat.audio.vad.silero import SileroVADAnalyzer
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_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
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 BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies
load_dotenv(override=True)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAILLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
),
)
context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(
user_turn_strategies=UserTurnStrategies(
stop=[
TurnAnalyzerUserTurnStopStrategy(
turn_analyzer=FalSmartTurnAnalyzer(
api_key=os.getenv("FAL_SMART_TURN_API_KEY"),
aiohttp_session=aiohttp.ClientSession(),
)
)
]
),
vad_analyzer=SileroVADAnalyzer(),
),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
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(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

64
src/pipecat/audio/turn/smart_turn/fal_smart_turn.py
View File

@@ -1,64 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Fal.ai smart turn analyzer implementation.
This module provides a smart turn analyzer that uses Fal.ai's hosted smart-turn model
for end-of-turn detection in conversations.
Note: To learn more about the smart-turn model, visit:
- https://fal.ai/models/fal-ai/smart-turn/playground
- https://github.com/pipecat-ai/smart-turn
"""
import warnings
from typing import Optional
import aiohttp
from pipecat.audio.turn.smart_turn.http_smart_turn import HttpSmartTurnAnalyzer
class FalSmartTurnAnalyzer(HttpSmartTurnAnalyzer):
"""Smart turn analyzer using Fal.ai's hosted smart-turn model.
Extends HttpSmartTurnAnalyzer to provide integration with Fal.ai's
smart turn detection API endpoint with proper authentication.
.. deprecated:: 0.98.0
FalSmartTurnAnalyzer is deprecated and will be removed in a future version.
Use LocalSmartTurnAnalyzerV3 instead.
"""
def __init__(
self,
*,
aiohttp_session: aiohttp.ClientSession,
url: str = "https://fal.run/fal-ai/smart-turn/raw",
api_key: Optional[str] = None,
**kwargs,
):
"""Initialize the Fal.ai smart turn analyzer.
Args:
aiohttp_session: HTTP client session for making API requests.
url: Fal.ai API endpoint URL for smart turn detection.
api_key: API key for authenticating with Fal.ai service.
**kwargs: Additional arguments passed to parent HttpSmartTurnAnalyzer.
"""
headers = {}
if api_key:
headers = {"Authorization": f"Key {api_key}"}
super().__init__(url=url, aiohttp_session=aiohttp_session, headers=headers, **kwargs)
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"FalSmartTurnAnalyzer is deprecated and will be removed in a future version. "
"Use LocalSmartTurnAnalyzerV3 instead.",
DeprecationWarning,
stacklevel=2,
)

107
src/pipecat/audio/turn/smart_turn/local_smart_turn.py
View File

@@ -1,107 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Local PyTorch smart turn analyzer for on-device ML inference.
This module provides a smart turn analyzer that uses PyTorch models for
local end-of-turn detection without requiring network connectivity.
"""
import warnings
from typing import Any, Dict
import numpy as np
from loguru import logger
from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn
try:
import torch
from transformers import AutoFeatureExtractor, Wav2Vec2BertForSequenceClassification
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 LocalSmartTurnAnalyzer(BaseSmartTurn):
"""Local smart turn analyzer using PyTorch models.
Provides end-of-turn detection using locally-stored PyTorch models,
enabling offline operation without network dependencies. Uses
Wav2Vec2-BERT architecture for audio sequence classification.
.. deprecated:: 0.0.98
LocalSmartTurnAnalyzer is deprecated and will be removed in a future version.
Use LocalSmartTurnAnalyzerV3 instead.
"""
def __init__(self, *, smart_turn_model_path: str, **kwargs):
"""Initialize the local PyTorch smart turn analyzer.
Args:
smart_turn_model_path: Path to directory containing the PyTorch model
and feature extractor files. If empty, uses default HuggingFace model.
**kwargs: Additional arguments passed to BaseSmartTurn.
"""
super().__init__(**kwargs)
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"LocalSmartTurnAnalyzer is deprecated and will be removed in a future version. "
"Use LocalSmartTurnAnalyzerV3 instead.",
DeprecationWarning,
stacklevel=2,
)
if not smart_turn_model_path:
# Define the path to the pretrained model on Hugging Face
smart_turn_model_path = "pipecat-ai/smart-turn"
logger.debug("Loading Local Smart Turn model...")
# Load the pretrained model for sequence classification
self._turn_model = Wav2Vec2BertForSequenceClassification.from_pretrained(
smart_turn_model_path
)
# Load the corresponding feature extractor for preprocessing audio
self._turn_processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path)
# Set device to GPU if available, else CPU
self._device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Move model to selected device and set it to evaluation mode
self._turn_model = self._turn_model.to(self._device)
self._turn_model.eval()
logger.debug("Loaded Local Smart Turn")
def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
"""Predict end-of-turn using local PyTorch model."""
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",
)
# Move input tensors to the same device as the model
inputs = {k: v.to(self._device) for k, v in inputs.items()}
# Disable gradient calculation for inference
with torch.no_grad():
outputs = self._turn_model(**inputs)
logits = outputs.logits
probabilities = torch.nn.functional.softmax(logits, 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,
}