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

Merge pull request #4228 from pipecat-ai/mb/remove-turn-deprecations

Browse Source
This commit is contained in:
Mark Backman
2026-04-02 14:32:21 -04:00
committed by GitHub
parent c28b7e8f26 3be0ea05ef
commit f93d13efff
33 changed files with 23 additions and 2608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/pipecat/audio/interruptions/__init__.py
View File

58
src/pipecat/audio/interruptions/base_interruption_strategy.py
View File

@@ -1,58 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base interruption strategy for determining when users can interrupt bot speech."""
from abc import ABC, abstractmethod
class BaseInterruptionStrategy(ABC):
"""Base class for interruption strategies.
This is a base class for interruption strategies. Interruption strategies
decide when the user can interrupt the bot while the bot is speaking. For
example, there could be strategies based on audio volume or strategies based
on the number of words the user spoke.
"""
async def append_audio(self, audio: bytes, sample_rate: int):
"""Append audio data to the strategy for analysis.
Not all strategies handle audio. Default implementation does nothing.
Args:
audio: Raw audio bytes to append.
sample_rate: Sample rate of the audio data in Hz.
"""
pass
async def append_text(self, text: str):
"""Append text data to the strategy for analysis.
Not all strategies handle text. Default implementation does nothing.
Args:
text: Text string to append for analysis.
"""
pass
@abstractmethod
async def should_interrupt(self) -> bool:
"""Determine if the user should interrupt the bot.
This is called when the user stops speaking and it's time to decide
whether the user should interrupt the bot. The decision will be based on
the aggregated audio and/or text.
Returns:
True if the user should interrupt the bot, False otherwise.
"""
pass
@abstractmethod
async def reset(self):
"""Reset the current accumulated text and/or audio."""
pass

75
src/pipecat/audio/interruptions/min_words_interruption_strategy.py
View File

@@ -1,75 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Minimum words interruption strategy for word count-based interruptions."""
from loguru import logger
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
class MinWordsInterruptionStrategy(BaseInterruptionStrategy):
"""Interruption strategy based on minimum number of words spoken.
This is an interruption strategy based on a minimum number of words said
by the user. That is, the strategy will be true if the user has said at
least that amount of words.
.. deprecated:: 0.0.99
This class is deprecated, use
`pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s
new `user_turn_strategies` parameter instead.
"""
def __init__(self, *, min_words: int):
"""Initialize the minimum words interruption strategy.
Args:
min_words: Minimum number of words required to trigger an interruption.
"""
super().__init__()
self._min_words = min_words
self._text = ""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"'pipecat.audio.interruptions' is deprecated. "
"Use `pipecat.turns.user_start.MinWordsUserTurnStartStrategy` with `PipelineTask`'s "
"new `user_turn_strategies` parameter instead.",
DeprecationWarning,
)
async def append_text(self, text: str):
"""Append text for word count analysis.
Args:
text: Text string to append to the accumulated text.
Note: Not all strategies need to handle text.
"""
self._text += text
async def should_interrupt(self) -> bool:
"""Check if the minimum word count has been reached.
Returns:
True if the user has spoken at least the minimum number of words.
"""
word_count = len(self._text.split())
interrupt = word_count >= self._min_words
logger.debug(
f"should_interrupt={interrupt} num_spoken_words={word_count} min_words={self._min_words}"
)
return interrupt
async def reset(self):
"""Reset the accumulated text for the next analysis cycle."""
self._text = ""

212
src/pipecat/frames/frames.py
View File

@@ -29,7 +29,6 @@ from typing import (
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.audio.turn.base_turn_analyzer import BaseTurnParams
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.metrics.metrics import MetricsData
@@ -462,137 +461,6 @@ class LLMContextAssistantTimestampFrame(DataFrame):
timestamp: str
@dataclass
class TranscriptionMessage:
"""A message in a conversation transcript.
A message in a conversation transcript containing the role and content.
Messages are in standard format with roles normalized to user/assistant.
Parameters:
role: The role of the message sender (user or assistant).
content: The message content/text.
user_id: Optional identifier for the user.
timestamp: Optional timestamp when the message was created.
.. deprecated:: 0.0.99
`TranscriptionMessage` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
role: Literal["user", "assistant"]
content: str
user_id: Optional[str] = None
timestamp: Optional[str] = None
def __post_init__(self):
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"TranscriptionMessage is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class ThoughtTranscriptionMessage:
"""An LLM thought message in a conversation transcript.
.. deprecated:: 0.0.99
`ThoughtTranscriptionMessage` is deprecated and will be removed in a future version.
Use `LLMAssistantAggregator`'s new events instead.
"""
role: Literal["assistant"] = field(default="assistant", init=False)
content: str
timestamp: Optional[str] = None
def __post_init__(self):
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"ThoughtTranscriptionMessage is deprecated and will be removed in a future version. "
"Use `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class TranscriptionUpdateFrame(DataFrame):
"""Frame containing new messages added to conversation transcript.
A frame containing new messages added to the conversation transcript.
This frame is emitted when new messages are added to the conversation history,
containing only the newly added messages rather than the full transcript.
Messages have normalized roles (user/assistant) regardless of the LLM service used.
Messages are always in the OpenAI standard message format, which supports both:
Examples:
Simple format::
[
{
"role": "user",
"content": "Hi, how are you?"
},
{
"role": "assistant",
"content": "Great! And you?"
}
]
Content list format::
[
{
"role": "user",
"content": [{"type": "text", "text": "Hi, how are you?"}]
},
{
"role": "assistant",
"content": [{"type": "text", "text": "Great! And you?"}]
}
]
OpenAI supports both formats. Anthropic and Google messages are converted to the
content list format.
Parameters:
messages: List of new transcript messages that were added.
.. deprecated:: 0.0.99
`TranscriptionUpdateFrame` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
messages: List[TranscriptionMessage | ThoughtTranscriptionMessage]
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"TranscriptionUpdateFrame is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
stacklevel=2,
)
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, messages: {len(self.messages)})"
@dataclass
class LLMContextFrame(Frame):
"""Frame containing a universal LLM context.
@@ -878,30 +746,18 @@ class StartFrame(SystemFrame):
Parameters:
audio_in_sample_rate: Input audio sample rate in Hz.
audio_out_sample_rate: Output audio sample rate in Hz.
allow_interruptions: Whether to allow user interruptions.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.
enable_metrics: Whether to enable performance metrics collection.
enable_tracing: Whether to enable OpenTelemetry tracing.
enable_usage_metrics: Whether to enable usage metrics collection.
interruption_strategies: List of interruption handling strategies.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead.
report_only_initial_ttfb: Whether to report only initial time-to-first-byte.
tracing_context: Pipeline-scoped tracing context for span hierarchy.
"""
audio_in_sample_rate: int = 16000
audio_out_sample_rate: int = 24000
allow_interruptions: bool = False
enable_metrics: bool = False
enable_tracing: bool = False
enable_usage_metrics: bool = False
interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list)
report_only_initial_ttfb: bool = False
tracing_context: Optional["TracingContext"] = None
@@ -1010,16 +866,9 @@ class UserStartedSpeakingFrame(SystemFrame):
Emitted when the user turn starts, which usually means that some
transcriptions are already available.
Parameters:
emulated: Whether this event was emulated rather than detected by VAD.
.. deprecated:: 0.0.99
This field is deprecated and will be removed in a future version.
"""
emulated: bool = False
pass
@dataclass
@@ -1028,16 +877,9 @@ class UserStoppedSpeakingFrame(SystemFrame):
Emitted when the user turn ends. This usually coincides with the start of
the bot turn.
Parameters:
emulated: Whether this event was emulated rather than detected by VAD.
.. deprecated:: 0.0.99
This field is deprecated and will be removed in a future version.
"""
emulated: bool = False
pass
@dataclass
@@ -1072,56 +914,6 @@ class UserSpeakingFrame(SystemFrame):
pass
@dataclass
class EmulateUserStartedSpeakingFrame(SystemFrame):
"""Frame to emulate user started speaking behavior.
Emitted by internal processors upstream to emulate VAD behavior when a
user starts speaking.
.. deprecated:: 0.0.99
This frame is deprecated and will be removed in a future version.
"""
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"EmulateUserStartedSpeakingFrame is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class EmulateUserStoppedSpeakingFrame(SystemFrame):
"""Frame to emulate user stopped speaking behavior.
Emitted by internal processors upstream to emulate VAD behavior when a
user stops speaking.
.. deprecated:: 0.0.99
This frame is deprecated and will be removed in a future version.
"""
def __post_init__(self):
super().__post_init__()
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"EmulateUserStoppedSpeakingFrame is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
@dataclass
class VADUserStartedSpeakingFrame(SystemFrame):
"""Frame emitted when VAD definitively detects user started speaking.

15
src/pipecat/pipeline/task.py
View File

@@ -20,7 +20,6 @@ from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tupl
from loguru import logger
from pydantic import BaseModel, ConfigDict, Field
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.clocks.base_clock import BaseClock
from pipecat.clocks.system_clock import SystemClock
from pipecat.frames.frames import (
@@ -111,11 +110,6 @@ class PipelineParams(BaseModel):
constructor arguments instead.
Parameters:
allow_interruptions: Whether to allow pipeline interruptions.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead.
audio_in_sample_rate: Input audio sample rate in Hz.
audio_out_sample_rate: Output audio sample rate in Hz.
enable_heartbeats: Whether to enable heartbeat monitoring.
@@ -124,11 +118,6 @@ class PipelineParams(BaseModel):
heartbeats_period_secs: Period between heartbeats in seconds.
heartbeats_monitor_secs: Timeout (in seconds) before warning about
missed heartbeats. Defaults to 10 seconds.
interruption_strategies: [deprecated] Strategies for bot interruption behavior.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead.
report_only_initial_ttfb: Whether to report only initial time to first byte.
send_initial_empty_metrics: Whether to send initial empty metrics.
start_metadata: Additional metadata for pipeline start.
@@ -136,7 +125,6 @@ class PipelineParams(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
allow_interruptions: bool = True
audio_in_sample_rate: int = 16000
audio_out_sample_rate: int = 24000
enable_heartbeats: bool = False
@@ -144,7 +132,6 @@ class PipelineParams(BaseModel):
enable_usage_metrics: bool = False
heartbeats_period_secs: float = HEARTBEAT_SECS
heartbeats_monitor_secs: float = HEARTBEAT_MONITOR_SECS
interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
report_only_initial_ttfb: bool = False
send_initial_empty_metrics: bool = True
start_metadata: Dict[str, Any] = Field(default_factory=dict)
@@ -778,14 +765,12 @@ class PipelineTask(BasePipelineTask):
self._maybe_start_idle_task()
start_frame = StartFrame(
allow_interruptions=self._params.allow_interruptions,
audio_in_sample_rate=self._params.audio_in_sample_rate,
audio_out_sample_rate=self._params.audio_out_sample_rate,
enable_metrics=self._params.enable_metrics,
enable_tracing=self._enable_tracing,
enable_usage_metrics=self._params.enable_usage_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
interruption_strategies=self._params.interruption_strategies,
tracing_context=self._tracing_context,
)
start_frame.metadata = self._create_start_metadata()

2
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -731,7 +731,7 @@ class LLMUserAggregator(LLMContextAggregator):
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
if params.enable_interruptions and self._allow_interruptions:
if params.enable_interruptions:
await self.broadcast_interruption()
await self._call_event_handler("on_user_turn_started", strategy)

64
src/pipecat/processors/aggregators/user_response.py
View File

@@ -1,64 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""User response aggregation for text frames.
This module provides an aggregator that collects user responses and outputs
them as TextFrame objects, useful for capturing and processing user input
in conversational pipelines.
"""
from pipecat.frames.frames import TextFrame
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMUserAggregator
class UserResponseAggregator(LLMUserAggregator):
"""Aggregates user responses into TextFrame objects.
This aggregator extends LLMUserAggregator to specifically handle
user input by collecting text responses and outputting them as TextFrame
objects when the aggregation is complete.
"""
def __init__(self, **kwargs):
"""Initialize the user response aggregator.
.. deprecated:: 0.0.92
`UserResponseAggregator` is deprecated and will be removed in a future version.
Args:
**kwargs: Additional arguments passed to parent LLMUserAggregator.
"""
super().__init__(context=LLMContext(), **kwargs)
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`UserResponseAggregator` is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
async def push_aggregation(self):
"""Push the aggregated user response as a TextFrame.
Creates a TextFrame from the current aggregation if it contains content,
resets the aggregation state, and pushes the frame downstream.
"""
if len(self._aggregation) > 0:
frame = TextFrame(self._aggregation.strip())
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
await self.push_frame(frame)
# Reset our accumulator state.
await self.reset()

243
src/pipecat/processors/filters/stt_mute_filter.py
View File

@@ -1,243 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Speech-to-text (STT) muting control module.
This module provides functionality to control STT muting based on different strategies,
such as during function calls, bot speech, or custom conditions. It helps manage when
the STT service should be active or inactive during a conversation.
"""
from dataclasses import dataclass
from enum import Enum
from typing import Awaitable, Callable, Optional
from loguru import logger
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
InterruptionFrame,
StartFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class STTMuteStrategy(Enum):
"""Strategies determining when STT should be muted.
Each strategy defines different conditions under which speech-to-text
processing should be temporarily disabled to prevent unwanted audio
processing during specific conversation states.
Parameters:
FIRST_SPEECH: Mute STT until the first bot speech is detected.
MUTE_UNTIL_FIRST_BOT_COMPLETE: Mute STT until the first bot completes speaking,
regardless of whether it is the first speech.
FUNCTION_CALL: Mute STT during function calls to prevent interruptions.
ALWAYS: Always mute STT when the bot is speaking.
CUSTOM: Use a custom callback to determine muting logic dynamically.
.. deprecated:: 0.0.99
`STTMuteStrategy` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
"""
FIRST_SPEECH = "first_speech"
MUTE_UNTIL_FIRST_BOT_COMPLETE = "mute_until_first_bot_complete"
FUNCTION_CALL = "function_call"
ALWAYS = "always"
CUSTOM = "custom"
@dataclass
class STTMuteConfig:
"""Configuration for STT muting behavior.
Defines which muting strategies to apply and provides optional custom
callback for advanced muting logic. Multiple strategies can be combined
to create sophisticated muting behavior.
Parameters:
strategies: Set of muting strategies to apply simultaneously.
should_mute_callback: Optional callback for custom muting logic.
Only required when using STTMuteStrategy.CUSTOM. Called with
the STTMuteFilter instance to determine muting state.
Note:
MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together
as they handle the first bot speech differently.
.. deprecated:: 0.0.99
`STTMuteConfig` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
"""
strategies: set[STTMuteStrategy]
should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None
def __post_init__(self):
"""Validate configuration after initialization.
Raises:
ValueError: If incompatible strategies are used together.
"""
if (
STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE in self.strategies
and STTMuteStrategy.FIRST_SPEECH in self.strategies
):
raise ValueError(
"MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together"
)
class STTMuteFilter(FrameProcessor):
"""A processor that handles STT muting and interruption control.
This processor combines STT muting and interruption control as a coordinated
feature. When STT is muted, interruptions are automatically disabled by
suppressing VAD-related frames. This prevents unwanted speech detection
during bot speech, function calls, or other specified conditions.
.. deprecated:: 0.0.99
`STTMuteFilter` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
"""
def __init__(self, *, config: STTMuteConfig, **kwargs):
"""Initialize the STT mute filter.
Args:
config: Configuration specifying muting strategies and behavior.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._config = config
self._first_speech_handled = False
self._bot_is_speaking = False
self._function_call_in_progress = set()
self._is_muted = False
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`STTMuteFilter` is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s new `user_mute_strategies` instead.",
DeprecationWarning,
)
async def _handle_mute_state(self, should_mute: bool):
"""Handle STT muting and interruption control state changes."""
if should_mute != self._is_muted:
logger.debug(f"STTMuteFilter {'muting' if should_mute else 'unmuting'}")
self._is_muted = should_mute
# Note: We don't send STTMuteFrame to the STT service itself.
# The filter blocks frames locally, but the STT service continues
# processing audio to keep streaming connections alive (e.g., Google STT).
async def _should_mute(self) -> bool:
"""Determine if STT should be muted based on current state and strategies."""
for strategy in self._config.strategies:
match strategy:
case STTMuteStrategy.FUNCTION_CALL:
if self._function_call_in_progress:
return True
case STTMuteStrategy.ALWAYS:
if self._bot_is_speaking:
return True
case STTMuteStrategy.FIRST_SPEECH:
if self._bot_is_speaking and not self._first_speech_handled:
self._first_speech_handled = True
return True
case STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE:
if not self._first_speech_handled:
return True
case STTMuteStrategy.CUSTOM:
if self._bot_is_speaking and self._config.should_mute_callback:
should_mute = await self._config.should_mute_callback(self)
if should_mute:
return True
return False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and manage muting state.
Monitors conversation state through frame types and applies muting
strategies accordingly. Suppresses VAD-related frames when muted
while allowing other frames to pass through.
Args:
frame: The incoming frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
# Determine if we need to change mute state based on frame type
should_mute = None
# Process frames to determine mute state
if isinstance(frame, StartFrame):
should_mute = await self._should_mute()
elif isinstance(frame, FunctionCallsStartedFrame):
for f in frame.function_calls:
self._function_call_in_progress.add(f.tool_call_id)
should_mute = await self._should_mute()
elif isinstance(frame, (FunctionCallCancelFrame, FunctionCallResultFrame)):
self._function_call_in_progress.remove(frame.tool_call_id)
should_mute = await self._should_mute()
elif isinstance(frame, BotStartedSpeakingFrame):
self._bot_is_speaking = True
should_mute = await self._should_mute()
elif isinstance(frame, BotStoppedSpeakingFrame):
self._bot_is_speaking = False
if not self._first_speech_handled:
self._first_speech_handled = True
should_mute = await self._should_mute()
# Then push the original frame
if isinstance(
frame,
(
InterruptionFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
TranscriptionFrame,
),
):
# Only pass VAD-related frames when not muted
if not self._is_muted:
await self.push_frame(frame, direction)
else:
logger.trace(f"{frame.__class__.__name__} suppressed - STT currently muted")
else:
# Pass all other frames through
await self.push_frame(frame, direction)
# Finally handle mute state change if needed
if should_mute is not None and should_mute != self._is_muted:
await self._handle_mute_state(should_mute)

43
src/pipecat/processors/frame_processor.py
View File

@@ -23,14 +23,12 @@ from typing import (
Coroutine,
List,
Optional,
Sequence,
Tuple,
Type,
)
from loguru import logger
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.clocks.base_clock import BaseClock
from pipecat.frames.frames import (
CancelFrame,
@@ -193,9 +191,6 @@ class FrameProcessor(BaseObject):
self._enable_metrics = False
self._enable_usage_metrics = False
self._report_only_initial_ttfb = False
# Other properties (deprecated)
self._allow_interruptions = False
self._interruption_strategies: List[BaseInterruptionStrategy] = []
# Indicates whether we have received the StartFrame.
self.__started = False
@@ -307,29 +302,6 @@ class FrameProcessor(BaseObject):
"""
return self._prev
@property
def interruptions_allowed(self):
"""Check if interruptions are allowed for this processor.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.
Returns:
True if interruptions are allowed.
"""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`FrameProcessor.interruptions_allowed` is deprecated. "
"Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.",
DeprecationWarning,
stacklevel=2,
)
return self._allow_interruptions
@property
def metrics_enabled(self):
"""Check if metrics collection is enabled.
@@ -357,19 +329,6 @@ class FrameProcessor(BaseObject):
"""
return self._report_only_initial_ttfb
@property
def interruption_strategies(self) -> Sequence[BaseInterruptionStrategy]:
"""Get the interruption strategies for this processor.
.. deprecated:: 0.0.99
This function is deprecated, use the new user and bot turn start
strategies insted.
Returns:
Sequence of interruption strategies.
"""
return self._interruption_strategies
@property
def task_manager(self) -> BaseTaskManager:
"""Get the task manager for this processor.
@@ -819,10 +778,8 @@ class FrameProcessor(BaseObject):
frame: The start frame containing initialization parameters.
"""
self.__started = True
self._allow_interruptions = frame.allow_interruptions
self._enable_metrics = frame.enable_metrics
self._enable_usage_metrics = frame.enable_usage_metrics
self._interruption_strategies = frame.interruption_strategies
self._report_only_initial_ttfb = frame.report_only_initial_ttfb
self.__create_process_task()

370
src/pipecat/processors/transcript_processor.py
View File

@@ -1,370 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Transcript processing utilities for conversation recording and analysis.
This module provides processors that convert speech and text frames into structured
transcript messages with timestamps, enabling conversation history tracking and analysis.
"""
from typing import List, Optional
from loguru import logger
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
Frame,
InterruptionFrame,
LLMThoughtEndFrame,
LLMThoughtStartFrame,
LLMThoughtTextFrame,
ThoughtTranscriptionMessage,
TranscriptionFrame,
TranscriptionMessage,
TranscriptionUpdateFrame,
TTSTextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
from pipecat.utils.time import time_now_iso8601
class BaseTranscriptProcessor(FrameProcessor):
"""Base class for processing conversation transcripts.
Provides common functionality for handling transcript messages and updates.
"""
def __init__(self, **kwargs):
"""Initialize processor with empty message store.
Args:
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._processed_messages: List[TranscriptionMessage] = []
self._register_event_handler("on_transcript_update")
async def _emit_update(self, messages: List[TranscriptionMessage]):
"""Emit transcript updates for new messages.
Args:
messages: New messages to emit in update.
"""
if messages:
self._processed_messages.extend(messages)
update_frame = TranscriptionUpdateFrame(messages=messages)
await self._call_event_handler("on_transcript_update", update_frame)
await self.push_frame(update_frame)
class UserTranscriptProcessor(BaseTranscriptProcessor):
"""Processes user transcription frames into timestamped conversation messages."""
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process TranscriptionFrames into user conversation messages.
Args:
frame: Input frame to process.
direction: Frame processing direction.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
message = TranscriptionMessage(
role="user", user_id=frame.user_id, content=frame.text, timestamp=frame.timestamp
)
await self._emit_update([message])
await self.push_frame(frame, direction)
class AssistantTranscriptProcessor(BaseTranscriptProcessor):
"""Processes assistant TTS text frames and LLM thought frames into timestamped messages.
This processor aggregates both TTS text frames and LLM thought frames into
complete utterances and thoughts, emitting them as transcript messages.
An assistant utterance is completed when:
- The bot stops speaking (BotStoppedSpeakingFrame)
- The bot is interrupted (InterruptionFrame)
- The pipeline ends (EndFrame, CancelFrame)
A thought is completed when:
- The thought ends (LLMThoughtEndFrame)
- The bot is interrupted (InterruptionFrame)
- The pipeline ends (EndFrame, CancelFrame)
"""
def __init__(self, *, process_thoughts: bool = False, **kwargs):
"""Initialize processor with aggregation state.
Args:
process_thoughts: Whether to process LLM thought frames. Defaults to False.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._process_thoughts = process_thoughts
self._current_assistant_text_parts: List[TextPartForConcatenation] = []
self._assistant_text_start_time: Optional[str] = None
self._current_thought_parts: List[TextPartForConcatenation] = []
self._thought_start_time: Optional[str] = None
self._thought_active = False
async def _emit_aggregated_assistant_text(self):
"""Aggregates and emits text fragments as a transcript message.
This method aggregates text fragments that may arrive in multiple
TTSTextFrame instances and emits them as a single TranscriptionMessage.
"""
if self._current_assistant_text_parts and self._assistant_text_start_time:
content = concatenate_aggregated_text(self._current_assistant_text_parts)
if content:
logger.trace(f"Emitting aggregated assistant message: {content}")
message = TranscriptionMessage(
role="assistant",
content=content,
timestamp=self._assistant_text_start_time,
)
await self._emit_update([message])
else:
logger.trace("No content to emit after stripping whitespace")
# Reset aggregation state
self._current_assistant_text_parts = []
self._assistant_text_start_time = None
async def _emit_aggregated_thought(self):
"""Aggregates and emits thought text fragments as a thought transcript message.
This method aggregates thought fragments that may arrive in multiple
LLMThoughtTextFrame instances and emits them as a single ThoughtTranscriptionMessage.
"""
if self._current_thought_parts and self._thought_start_time:
content = concatenate_aggregated_text(self._current_thought_parts)
if content:
logger.trace(f"Emitting aggregated thought message: {content}")
message = ThoughtTranscriptionMessage(
content=content,
timestamp=self._thought_start_time,
)
await self._emit_update([message])
else:
logger.trace("No thought content to emit after stripping whitespace")
# Reset aggregation state
self._current_thought_parts = []
self._thought_start_time = None
self._thought_active = False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames into assistant conversation messages and thought messages.
Handles different frame types:
- TTSTextFrame: Aggregates text for current utterance
- LLMThoughtStartFrame: Begins aggregating a new thought
- LLMThoughtTextFrame: Aggregates text for current thought
- LLMThoughtEndFrame: Completes current thought
- BotStoppedSpeakingFrame: Completes current utterance
- InterruptionFrame: Completes current utterance and thought due to interruption
- EndFrame: Completes current utterance and thought at pipeline end
- CancelFrame: Completes current utterance and thought due to cancellation
Args:
frame: Input frame to process.
direction: Frame processing direction.
"""
await super().process_frame(frame, direction)
if isinstance(frame, (InterruptionFrame, CancelFrame)):
# Push frame first otherwise our emitted transcription update frame
# might get cleaned up.
await self.push_frame(frame, direction)
# Emit accumulated text and thought with interruptions
await self._emit_aggregated_assistant_text()
if self._process_thoughts and self._thought_active:
await self._emit_aggregated_thought()
elif isinstance(frame, LLMThoughtStartFrame):
# Start a new thought
if self._process_thoughts:
self._thought_active = True
self._thought_start_time = time_now_iso8601()
self._current_thought_parts = []
# Push frame.
await self.push_frame(frame, direction)
elif isinstance(frame, LLMThoughtTextFrame):
# Aggregate thought text if we have an active thought
if self._process_thoughts and self._thought_active:
self._current_thought_parts.append(
TextPartForConcatenation(
frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
)
)
# Push frame.
await self.push_frame(frame, direction)
elif isinstance(frame, LLMThoughtEndFrame):
# Emit accumulated thought when thought ends
if self._process_thoughts and self._thought_active:
await self._emit_aggregated_thought()
# Push frame.
await self.push_frame(frame, direction)
elif isinstance(frame, TTSTextFrame):
# Start timestamp on first text part
if not self._assistant_text_start_time:
self._assistant_text_start_time = time_now_iso8601()
self._current_assistant_text_parts.append(
TextPartForConcatenation(
frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
)
)
# Push frame.
await self.push_frame(frame, direction)
elif isinstance(frame, (BotStoppedSpeakingFrame, EndFrame)):
# Emit accumulated text when bot finishes speaking or pipeline ends.
await self._emit_aggregated_assistant_text()
# Emit accumulated thought at pipeline end if still active
if isinstance(frame, EndFrame) and self._process_thoughts and self._thought_active:
await self._emit_aggregated_thought()
# Push frame.
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
class TranscriptProcessor:
"""Factory for creating and managing transcript processors.
Provides unified access to user and assistant transcript processors
with shared event handling. The assistant processor handles both TTS text
and LLM thought frames.
Example::
transcript = TranscriptProcessor()
pipeline = Pipeline(
[
transport.input(),
stt,
transcript.user(), # User transcripts
context_aggregator.user(),
llm,
tts,
transport.output(),
transcript.assistant(), # Assistant transcripts (including thoughts)
context_aggregator.assistant(),
]
)
@transcript.event_handler("on_transcript_update")
async def handle_update(processor, frame):
print(f"New messages: {frame.messages}")
.. deprecated:: 0.0.99
`TranscriptProcessor` is deprecated and will be removed in a future version.
Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
"""
def __init__(self, *, process_thoughts: bool = False):
"""Initialize factory.
Args:
process_thoughts: Whether the assistant processor should handle LLM thought
frames. Defaults to False.
"""
self._process_thoughts = process_thoughts
self._user_processor = None
self._assistant_processor = None
self._event_handlers = {}
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`TranscriptProcessor` is deprecated and will be removed in a future version. "
"Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
DeprecationWarning,
)
def user(self, **kwargs) -> UserTranscriptProcessor:
"""Get the user transcript processor.
Args:
**kwargs: Arguments specific to UserTranscriptProcessor.
Returns:
The user transcript processor instance.
"""
if self._user_processor is None:
self._user_processor = UserTranscriptProcessor(**kwargs)
# Apply any registered event handlers
for event_name, handler in self._event_handlers.items():
@self._user_processor.event_handler(event_name)
async def user_handler(processor, frame):
return await handler(processor, frame)
return self._user_processor
def assistant(self, **kwargs) -> AssistantTranscriptProcessor:
"""Get the assistant transcript processor.
Args:
**kwargs: Arguments specific to AssistantTranscriptProcessor.
Returns:
The assistant transcript processor instance.
"""
if self._assistant_processor is None:
self._assistant_processor = AssistantTranscriptProcessor(
process_thoughts=self._process_thoughts, **kwargs
)
# Apply any registered event handlers
for event_name, handler in self._event_handlers.items():
@self._assistant_processor.event_handler(event_name)
async def assistant_handler(processor, frame):
return await handler(processor, frame)
return self._assistant_processor
def event_handler(self, event_name: str):
"""Register event handler for both processors.
Args:
event_name: Name of event to handle.
Returns:
Decorator function that registers handler with both processors.
"""
def decorator(handler):
self._event_handlers[event_name] = handler
# Apply handler to existing processors if they exist
if self._user_processor:
@self._user_processor.event_handler(event_name)
async def user_handler(processor, frame):
return await handler(processor, frame)
if self._assistant_processor:
@self._assistant_processor.event_handler(event_name)
async def assistant_handler(processor, frame):
return await handler(processor, frame)
return handler
return decorator

14
src/pipecat/services/aws/nova_sonic/llm.py
View File

@@ -1281,18 +1281,8 @@ class AWSNovaSonicLLMService(LLMService):
# HACK: Check if this transcription was triggered by our own
# assistant response trigger. If so, we need to wrap it with
# UserStarted/StoppedSpeakingFrames; otherwise the user aggregator
# would fire an EmulatedUserStartedSpeakingFrame, which would
# trigger an interruption, which would prevent us from writing the
# assistant response to context.
#
# Sending an EmulateUserStartedSpeakingFrame ourselves doesn't
# work: it just causes the interruption we're trying to avoid.
#
# Setting enable_emulated_vad_interruptions also doesn't work: at
# the time the user aggregator receives the TranscriptionFrame, it
# doesn't yet know the assistant has started responding, so it
# doesn't know that emulating the user starting to speak would
# cause an interruption.
# would trigger an interruption, which would prevent us from
# writing the assistant response to context.
should_wrap_in_user_started_stopped_speaking_frames = (
self._waiting_for_trigger_transcription
and self._user_text_buffer.strip().lower() == "ready"

2
src/pipecat/services/google/stt.py
View File

@@ -1004,7 +1004,7 @@ class GoogleSTTService(STTService):
except Aborted as e:
# Handle stream abort due to inactivity (409 error).
# This occurs when no audio is sent to the stream for 10+ seconds,
# which can happen when InputAudioRawFrames are blocked (e.g., by STTMuteFilter).
# which can happen when InputAudioRawFrames are blocked.
# Google's STT service automatically closes the stream in this case.
# We log at DEBUG level (not ERROR) since this is recoverable, then re-raise
# to trigger automatic reconnection in _stream_audio.

31
src/pipecat/transports/base_input.py
View File

@@ -25,8 +25,6 @@ from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
EndFrame,
FilterUpdateSettingsFrame,
Frame,
@@ -313,12 +311,6 @@ class BaseInputTransport(FrameProcessor):
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._deprecated_handle_bot_stopped_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, EmulateUserStartedSpeakingFrame):
logger.debug("Emulating user started speaking")
await self._deprecated_handle_user_interruption(VADState.SPEAKING, emulated=True)
elif isinstance(frame, EmulateUserStoppedSpeakingFrame):
logger.debug("Emulating user stopped speaking")
await self._deprecated_handle_user_interruption(VADState.QUIET, emulated=True)
# All other system frames
elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
@@ -500,36 +492,21 @@ class BaseInputTransport(FrameProcessor):
"""Update bot speaking state when bot stops speaking."""
self._bot_speaking = False
async def _deprecated_handle_user_interruption(
self, vad_state: VADState, emulated: bool = False
):
async def _deprecated_handle_user_interruption(self, vad_state: VADState):
"""Handle user interruption events based on speaking state."""
if vad_state == VADState.SPEAKING:
logger.debug("User started speaking")
self._user_speaking = True
await self.broadcast_frame(UserStartedSpeakingFrame, emulated=emulated)
# Only push InterruptionFrame if:
# 1. No interruption config is set, OR
# 2. Interruption config is set but bot is not speaking
should_push_immediate_interruption = (
not self.interruption_strategies or not self._bot_speaking
)
await self.broadcast_frame(UserStartedSpeakingFrame)
# Make sure we notify about interruptions quickly out-of-band.
if should_push_immediate_interruption and self._allow_interruptions:
await self.broadcast_interruption()
elif self.interruption_strategies and self._bot_speaking:
logger.debug(
"User started speaking while bot is speaking with interruption config - "
"deferring interruption to aggregator"
)
await self.broadcast_interruption()
elif vad_state == VADState.QUIET:
logger.debug("User stopped speaking")
self._user_speaking = False
await self.broadcast_frame(UserStoppedSpeakingFrame, emulated=emulated)
await self.broadcast_frame(UserStoppedSpeakingFrame)
async def _deprecated_old_handle_vad(
self, audio_frame: InputAudioRawFrame, vad_state: VADState

3
src/pipecat/transports/base_output.py
View File

@@ -517,9 +517,6 @@ class BaseOutputTransport(FrameProcessor):
Args:
_: The start interruption frame (unused).
"""
if not self._transport._allow_interruptions:
return
# Cancel tasks.
await self._cancel_audio_task()
await self._cancel_clock_task()

2
src/pipecat/turns/user_turn_processor.py
View File

@@ -181,7 +181,7 @@ class UserTurnProcessor(FrameProcessor):
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
if params.enable_interruptions and self._allow_interruptions:
if params.enable_interruptions:
await self.broadcast_interruption()
await self._call_event_handler("on_user_turn_started", strategy)

40
src/pipecat/utils/text/pattern_pair_aggregator.py
View File

@@ -161,46 +161,6 @@ class PatternPairAggregator(SimpleTextAggregator):
}
return self
def add_pattern_pair(
self, pattern_id: str, start_pattern: str, end_pattern: str, remove_match: bool = True
):
"""Add a pattern pair to detect in the text.
.. deprecated:: 0.0.95
This function is deprecated and will be removed in a future version.
Use `add_pattern` with a type and MatchAction instead.
This method calls `add_pattern` setting type with the provided pattern_id and action
to either MatchAction.REMOVE or MatchAction.KEEP based on `remove_match`.
Args:
pattern_id: Identifier for this pattern pair. Should be unique and ideally descriptive.
(e.g., 'code', 'speaker', 'custom'). pattern_id can not be 'sentence' or 'word'
as those arereserved for the default behavior.
start_pattern: Pattern that marks the beginning of content.
end_pattern: Pattern that marks the end of content.
remove_match: If True, the matched pattern will be removed from the text. (Same as MatchAction.REMOVE)
If False, it will be kept and treated as normal text. (Same as MatchAction.KEEP)
"""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("once")
warnings.warn(
"add_pattern_pair with a pattern_id or remove_match is deprecated and will be"
" removed in a future version. Use add_pattern with a type and MatchAction instead",
DeprecationWarning,
stacklevel=2,
)
action = MatchAction.REMOVE if remove_match else MatchAction.KEEP
return self.add_pattern(
type=pattern_id,
start_pattern=start_pattern,
end_pattern=end_pattern,
action=action,
)
def on_pattern_match(
self, type: str, handler: Callable[[PatternMatch], Awaitable[None]]
) -> "PatternPairAggregator":

257
src/pipecat/utils/tracing/class_decorators.py
View File

@@ -1,257 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
# Portions Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base OpenTelemetry tracing decorators and utilities for Pipecat.
.. deprecated:: 0.0.103
This module is unused and will be removed in a future release.
Service tracing is handled by the decorators in
:mod:`pipecat.utils.tracing.service_decorators`.
This module provides class and method level tracing capabilities
similar to the original NVIDIA implementation.
"""
import asyncio
import contextlib
import enum
import functools
import inspect
import warnings
from typing import Callable, Optional, TypeVar
warnings.warn(
"pipecat.utils.tracing.class_decorators is deprecated and will be removed in a future "
"release. Use pipecat.utils.tracing.service_decorators instead.",
DeprecationWarning,
stacklevel=2,
)
from pipecat.utils.tracing.setup import is_tracing_available
# Import OpenTelemetry if available
if is_tracing_available():
import opentelemetry.trace
from opentelemetry import metrics, trace
# Type variables for better typing support
T = TypeVar("T")
C = TypeVar("C", bound=type)
class AttachmentStrategy(enum.Enum):
"""Controls how spans are attached to the trace hierarchy.
Parameters:
CHILD: Attached to class span if no parent, otherwise to parent.
LINK: Attached to class span with link to parent.
NONE: Always attached to class span regardless of context.
"""
CHILD = enum.auto()
LINK = enum.auto()
NONE = enum.auto()
class Traceable:
"""Base class for objects that can be traced with OpenTelemetry.
Provides the foundational tracing capabilities used by @traced methods.
"""
def __init__(self, name: str, **kwargs):
"""Initialize a traceable object.
Args:
name: Name of the traceable object for the span.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
if not is_tracing_available():
self._tracer = self._meter = self._parent_span_id = self._span = None
return
self._tracer = trace.get_tracer("pipecat")
self._meter = metrics.get_meter("pipecat")
self._parent_span_id = trace.get_current_span().get_span_context().span_id
self._span = self._tracer.start_span(name)
self._span.end()
@property
def meter(self):
"""Get the OpenTelemetry meter instance.
Returns:
The OpenTelemetry meter instance for this object.
"""
return self._meter
@contextlib.contextmanager
def __traced_context_manager(
self: Traceable, func: Callable, name: str | None, attachment_strategy: AttachmentStrategy
):
"""Internal context manager for the traced decorator.
Args:
self: The Traceable instance.
func: The function being traced.
name: Custom span name or None to use function name.
attachment_strategy: How to attach this span to the trace hierarchy.
Raises:
RuntimeError: If used in a class not inheriting from Traceable.
"""
if not isinstance(self, Traceable):
raise RuntimeError(
"@traced annotation can only be used in classes inheriting from Traceable"
)
stack = contextlib.ExitStack()
try:
current_span = trace.get_current_span()
is_span_class_parent_span = current_span.get_span_context().span_id == self._parent_span_id
match attachment_strategy:
case AttachmentStrategy.CHILD if not is_span_class_parent_span:
stack.enter_context(
self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore
)
case AttachmentStrategy.LINK:
if is_span_class_parent_span:
link = trace.Link(self._span.get_span_context()) # type: ignore
else:
link = trace.Link(current_span.get_span_context())
stack.enter_context(
opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore
)
stack.enter_context(
self._tracer.start_as_current_span( # type: ignore
func.__name__ if name is None else name, links=[link]
)
)
case AttachmentStrategy.NONE | AttachmentStrategy.CHILD:
stack.enter_context(
opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore
)
stack.enter_context(
self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore
)
yield
finally:
stack.close()
def __traced_decorator(func, name, attachment_strategy: AttachmentStrategy):
"""Implementation of the traced decorator.
Args:
func: The function to trace.
name: Custom span name.
attachment_strategy: How to attach this span.
Returns:
The wrapped function with tracing capabilities.
"""
@functools.wraps(func)
async def coroutine_wrapper(self: Traceable, *args, **kwargs):
exception = None
with __traced_context_manager(self, func, name, attachment_strategy):
try:
return await func(self, *args, **kwargs)
except asyncio.CancelledError as e:
exception = e
if exception:
raise exception
@functools.wraps(func)
async def generator_wrapper(self: Traceable, *args, **kwargs):
exception = None
with __traced_context_manager(self, func, name, attachment_strategy):
try:
async for v in func(self, *args, **kwargs):
yield v
except asyncio.CancelledError as e:
exception = e
if exception:
raise exception
if inspect.iscoroutinefunction(func):
return coroutine_wrapper
if inspect.isasyncgenfunction(func):
return generator_wrapper
raise ValueError("@traced annotation can only be used on async or async generator functions")
def traced(
func: Optional[Callable] = None,
*,
name: Optional[str] = None,
attachment_strategy: AttachmentStrategy = AttachmentStrategy.CHILD,
) -> Callable:
"""Add tracing to an async function in a Traceable class.
Args:
func: The async function to trace.
name: Custom span name. Defaults to function name.
attachment_strategy: How to attach this span (CHILD, LINK, NONE).
Returns:
Wrapped async function with tracing.
Raises:
RuntimeError: If used in a class not inheriting from Traceable.
ValueError: If used on a non-async function.
"""
if not is_tracing_available():
# Just return the original function or a simple decorator
def decorator(f):
return f
return decorator if func is None else func
if func is not None:
return __traced_decorator(func, name=name, attachment_strategy=attachment_strategy)
else:
return functools.partial(
__traced_decorator, name=name, attachment_strategy=attachment_strategy
)
def traceable(cls: C) -> C:
"""Make a class traceable for OpenTelemetry.
Creates a new class that inherits from both the original class
and Traceable, enabling tracing for class methods.
Args:
cls: The class to make traceable.
Returns:
A new class with tracing capabilities.
"""
if not is_tracing_available():
return cls
@functools.wraps(cls, updated=())
class TracedClass(cls, Traceable):
def __init__(self, *args, **kwargs):
"""Initialize the traced class instance.
Args:
*args: Positional arguments passed to parent classes.
**kwargs: Keyword arguments passed to parent classes.
"""
cls.__init__(self, *args, **kwargs)
if hasattr(self, "name"):
Traceable.__init__(self, self.name)
else:
Traceable.__init__(self, cls.__name__)
return TracedClass

5
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -100,11 +100,6 @@ def _get_parent_service_context(self):
if not is_tracing_available():
return None
# TODO: Remove this block and delete class_decorators.py once Traceable is removed.
# Legacy: support for classes inheriting from Traceable (currently unused, deprecated).
if hasattr(self, "_span") and self._span:
return trace.set_span_in_context(self._span)
# Use the conversation context set by TurnTraceObserver via TracingContext.
tracing_ctx = getattr(self, "_tracing_context", None)
conversation_context = tracing_ctx.get_conversation_context() if tracing_ctx else None