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

Merge branch 'main' into feat/gladia-auto-reconnect

Browse Source
This commit is contained in:
Jean-Louis Queguiner
2025-06-18 08:57:21 +02:00
committed by GitHub
parent 3d0ffbc832 1cac94bf97
commit 8071c4ba1c
180 changed files with 11267 additions and 3343 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/pipecat/__init__.py
View File

@@ -4,7 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import sys
from importlib.metadata import version
@@ -12,18 +11,4 @@ from loguru import logger
__version__ = version("pipecat-ai")
event_loop = "asyncio"
if sys.platform in ("linux", "darwin"):
try:
import asyncio
import uvloop
event_loop = f"uvloop {uvloop.__version__}"
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
logger.debug(f"Couldn't find `uvloop`")
pass
logger.info(f"ᓚᘏᗢ Pipecat {__version__} ({event_loop}; Python {sys.version}) ᓚᘏᗢ")
logger.info(f"ᓚᘏᗢ Pipecat {__version__} (Python {sys.version}) ᓚᘏᗢ")

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

38
src/pipecat/audio/interruptions/base_interruption_strategy.py Normal file
View File

@@ -0,0 +1,38 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from abc import ABC, abstractmethod
class BaseInterruptionStrategy(ABC):
"""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):
"""Appends audio to the strategy. Not all strategies handle audio."""
pass
async def append_text(self, text: str):
"""Appends text to the strategy. Not all strategies handle text."""
pass
@abstractmethod
async def should_interrupt(self) -> bool:
"""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.
"""
pass
@abstractmethod
async def reset(self):
"""Reset the current accumulated text and/or audio."""
pass

40
src/pipecat/audio/interruptions/min_words_interruption_strategy.py Normal file
View File

@@ -0,0 +1,40 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from loguru import logger
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
class MinWordsInterruptionStrategy(BaseInterruptionStrategy):
"""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.
"""
def __init__(self, *, min_words: int):
super().__init__()
self._min_words = min_words
self._text = ""
async def append_text(self, text: str):
"""Appends text for later analysis. Not all strategies need to handle
text.
"""
self._text += text
async def should_interrupt(self) -> bool:
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):
self._text = ""

18
src/pipecat/examples/run.py
View File

@@ -64,7 +64,7 @@ async def maybe_capture_participant_screen(
def run_example_daily(
run_example: Callable,
args: argparse.Namespace,
params: DailyParams,
transport_params: Mapping[str, Callable] = {},
):
logger.info("Running example with DailyTransport...")
@@ -75,6 +75,7 @@ def run_example_daily(
(room_url, token) = await configure(session)
# Run example function with DailyTransport transport arguments.
params: DailyParams = transport_params[args.transport]()
transport = DailyTransport(room_url, token, "Pipecat", params=params)
await run_example(transport, args, True)
@@ -84,7 +85,7 @@ def run_example_daily(
def run_example_webrtc(
run_example: Callable,
args: argparse.Namespace,
params: TransportParams,
transport_params: Mapping[str, Callable] = {},
):
logger.info("Running example with SmallWebRTCTransport...")
@@ -130,6 +131,7 @@ def run_example_webrtc(
pcs_map.pop(webrtc_connection.pc_id, None)
# Run example function with SmallWebRTC transport arguments.
params: TransportParams = transport_params[args.transport]()
transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection)
background_tasks.add_task(run_example, transport, args, False)
@@ -142,7 +144,7 @@ def run_example_webrtc(
@asynccontextmanager
async def lifespan(app: FastAPI):
yield # Run app
coros = [pc.close() for pc in pcs_map.values()]
coros = [pc.disconnect() for pc in pcs_map.values()]
await asyncio.gather(*coros)
pcs_map.clear()
@@ -152,7 +154,7 @@ def run_example_webrtc(
def run_example_twilio(
run_example: Callable,
args: argparse.Namespace,
params: FastAPIWebsocketParams,
transport_params: Mapping[str, Callable] = {},
):
logger.info("Running example with FastAPIWebsocketTransport (Twilio)...")
@@ -195,6 +197,7 @@ def run_example_twilio(
call_sid = call_data["start"]["callSid"]
# Create websocket transport and update params.
params: FastAPIWebsocketParams = transport_params[args.transport]()
params.add_wav_header = False
params.serializer = TwilioFrameSerializer(
stream_sid=stream_sid,
@@ -217,14 +220,13 @@ def run_main(
logger.error(f"Transport '{args.transport}' not supported by this example")
return
params = transport_params[args.transport]()
match args.transport:
case "daily":
run_example_daily(run_example, args, params)
run_example_daily(run_example, args, transport_params)
case "webrtc":
run_example_webrtc(run_example, args, params)
run_example_webrtc(run_example, args, transport_params)
case "twilio":
run_example_twilio(run_example, args, params)
run_example_twilio(run_example, args, transport_params)
def main(

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

@@ -15,9 +15,11 @@ from typing import (
Literal,
Mapping,
Optional,
Sequence,
Tuple,
)
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.metrics.metrics import MetricsData
from pipecat.transcriptions.language import Language
@@ -448,6 +450,7 @@ class StartFrame(SystemFrame):
enable_metrics: bool = False
enable_usage_metrics: bool = False
report_only_initial_ttfb: bool = False
interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list)
@dataclass
@@ -524,6 +527,29 @@ class StopTaskFrame(SystemFrame):
pass
@dataclass
class FrameProcessorPauseUrgentFrame(SystemFrame):
"""This processor is used to pause frame processing for the given processor
as fast as possible. Pausing frame processing will keep frames in the
internal queue which will then be processed when frame processing is resumed
with `FrameProcessorResumeFrame`.
"""
processor: str
@dataclass
class FrameProcessorResumeUrgentFrame(SystemFrame):
"""This processor is used to resume frame processing for the given processor
if it was previously paused as fast as possible. After resuming frame
processing all queued frames will be processed in the order received.
"""
processor: str
@dataclass
class StartInterruptionFrame(SystemFrame):
"""Emitted by VAD to indicate that a user has started speaking (i.e. is
@@ -643,6 +669,32 @@ class MetricsFrame(SystemFrame):
data: List[MetricsData]
@dataclass
class FunctionCallFromLLM:
"""Represents a function call returned by the LLM to be registered for execution.
Attributes:
function_name (str): The name of the function.
tool_call_id (str): A unique identifier for the function call.
arguments (Mapping[str, Any]): The arguments for the function.
context (OpenAILLMContext): The LLM context.
"""
function_name: str
tool_call_id: str
arguments: Mapping[str, Any]
context: Any
@dataclass
class FunctionCallsStartedFrame(SystemFrame):
"""A frame signaling that one or more function call execution is going to
start."""
function_calls: Sequence[FunctionCallFromLLM]
@dataclass
class FunctionCallInProgressFrame(SystemFrame):
"""A frame signaling that a function call is in progress."""
@@ -677,6 +729,7 @@ class FunctionCallResultFrame(SystemFrame):
tool_call_id: str
arguments: Any
result: Any
run_llm: Optional[bool] = None
properties: Optional[FunctionCallResultProperties] = None
@@ -824,6 +877,27 @@ class StopFrame(ControlFrame):
pass
@dataclass
class FrameProcessorPauseFrame(ControlFrame):
"""This processor is used to pause frame processing for the given
processor. Pausing frame processing will keep frames in the internal queue
which will then be processed when frame processing is resumed with
`FrameProcessorResumeFrame`."""
processor: str
@dataclass
class FrameProcessorResumeFrame(ControlFrame):
"""This processor is used to resume frame processing for the given processor
if it was previously paused. After resuming frame processing all queued
frames will be processed in the order received.
"""
processor: str
@dataclass
class LLMFullResponseStartFrame(ControlFrame):
"""Used to indicate the beginning of an LLM response. Following by one or

4
src/pipecat/pipeline/runner.py
View File

@@ -59,7 +59,7 @@ class PipelineRunner(BaseObject):
await asyncio.gather(*[t.stop_when_done() for t in self._tasks.values()])
async def cancel(self):
logger.debug(f"Canceling runner {self}")
logger.debug(f"Cancelling runner {self}")
await asyncio.gather(*[t.cancel() for t in self._tasks.values()])
def _setup_sigint(self):
@@ -72,7 +72,7 @@ class PipelineRunner(BaseObject):
self._sig_task = asyncio.create_task(self._sig_cancel())
async def _sig_cancel(self):
logger.warning(f"Interruption detected. Canceling runner {self}")
logger.warning(f"Interruption detected. Cancelling runner {self}")
await self.cancel()
def _gc_collect(self):

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

@@ -6,11 +6,12 @@
import asyncio
import time
from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type
from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Sequence, Tuple, Type
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 (
@@ -54,10 +55,11 @@ class PipelineParams(BaseModel):
enable_metrics: Whether to enable metrics collection.
enable_usage_metrics: Whether to enable usage metrics.
heartbeats_period_secs: Period between heartbeats in seconds.
observers: List of observers for monitoring pipeline execution.
observers: [deprecated] Use `observers` arg in `PipelineTask` class.
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.
interruption_strategies: Strategies for bot interruption behavior.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -73,6 +75,7 @@ class PipelineParams(BaseModel):
report_only_initial_ttfb: bool = False
send_initial_empty_metrics: bool = True
start_metadata: Dict[str, Any] = Field(default_factory=dict)
interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
class PipelineTaskSource(FrameProcessor):
@@ -181,7 +184,9 @@ class PipelineTask(BaseTask):
the idle timeout is reached.
enable_turn_tracking: Whether to enable turn tracking.
enable_turn_tracing: Whether to enable turn tracing.
conversation_id: Optional custom ID for the conversation.
additional_span_attributes: Optional dictionary of attributes to propagate as
OpenTelemetry conversation span attributes.
"""
def __init__(
@@ -202,6 +207,7 @@ class PipelineTask(BaseTask):
enable_turn_tracking: bool = True,
enable_tracing: bool = False,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
):
super().__init__()
self._pipeline = pipeline
@@ -214,6 +220,7 @@ class PipelineTask(BaseTask):
self._enable_turn_tracking = enable_turn_tracking
self._enable_tracing = enable_tracing and is_tracing_available()
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
if self._params.observers:
import warnings
@@ -232,7 +239,9 @@ class PipelineTask(BaseTask):
observers.append(self._turn_tracking_observer)
if self._enable_tracing and self._turn_tracking_observer:
self._turn_trace_observer = TurnTraceObserver(
self._turn_tracking_observer, conversation_id=self._conversation_id
self._turn_tracking_observer,
conversation_id=self._conversation_id,
additional_span_attributes=self._additional_span_attributes,
)
observers.append(self._turn_trace_observer)
self._finished = False
@@ -307,8 +316,8 @@ class PipelineTask(BaseTask):
"""Return the turn trace observer if enabled."""
return self._turn_trace_observer
async def add_observer(self, observer: BaseObserver):
await self._observer.add_observer(observer)
def add_observer(self, observer: BaseObserver):
self._observer.add_observer(observer)
async def remove_observer(self, observer: BaseObserver):
await self._observer.remove_observer(observer)
@@ -518,6 +527,7 @@ class PipelineTask(BaseTask):
enable_metrics=self._params.enable_metrics,
enable_usage_metrics=self._params.enable_usage_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
interruption_strategies=self._params.interruption_strategies,
)
start_frame.metadata = self._params.start_metadata
await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)

23
src/pipecat/pipeline/task_observer.py
View File

@@ -49,21 +49,31 @@ class TaskObserver(BaseObserver):
super().__init__(**kwargs)
self._observers = observers or []
self._task_manager = task_manager
self._proxies: Dict[BaseObserver, Proxy] = {}
self._proxies: Optional[Dict[BaseObserver, Proxy]] = (
None # Becomes a dict after start() is called
)
async def add_observer(self, observer: BaseObserver):
proxy = self._create_proxy(observer)
self._proxies[observer] = proxy
def add_observer(self, observer: BaseObserver):
# Add the observer to the list.
self._observers.append(observer)
# If we already started, create a new proxy for the observer.
# Otherwise, it will be created in start().
if self._started():
proxy = self._create_proxy(observer)
self._proxies[observer] = proxy
async def remove_observer(self, observer: BaseObserver):
# If the observer has a proxy, remove it.
if observer in self._proxies:
proxy = self._proxies[observer]
# Remove the proxy so it doesn't get called anymore.
del self._proxies[observer]
# Cancel the proxy task right away.
await self._task_manager.cancel_task(proxy.task)
# Remove the observer.
# Remove the observer from the list.
if observer in self._observers:
self._observers.remove(observer)
async def start(self):
@@ -79,6 +89,9 @@ class TaskObserver(BaseObserver):
for proxy in self._proxies.values():
await proxy.queue.put(data)
def _started(self) -> bool:
return self._proxies is not None
def _create_proxy(self, observer: BaseObserver) -> Proxy:
queue = asyncio.Queue()
task = self._task_manager.create_task(

104
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -11,7 +11,9 @@ from typing import Dict, List, Literal, Optional, Set
from loguru import logger
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.frames.frames import (
BotInterruptionFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -22,6 +24,8 @@ from pipecat.frames.frames import (
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
@@ -159,7 +163,7 @@ class BaseLLMResponseAggregator(FrameProcessor):
pass
@abstractmethod
def reset(self):
async def reset(self):
"""Reset the internals of this aggregator. This should not modify the
internal messages.
"""
@@ -193,7 +197,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
self._context = context
self._role = role
self._aggregation = ""
self._aggregation: str = ""
@property
def messages(self) -> List[dict]:
@@ -226,7 +230,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict):
self._context.set_tool_choice(tool_choice)
def reset(self):
async def reset(self):
self._aggregation = ""
@@ -269,10 +273,11 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self._aggregation_event = asyncio.Event()
self._aggregation_task = None
def reset(self):
super().reset()
async def reset(self):
await super().reset()
self._seen_interim_results = False
self._waiting_for_aggregation = False
[await s.reset() for s in self._interruption_strategies]
async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": self.role, "content": aggregation})
@@ -293,6 +298,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
elif isinstance(frame, CancelFrame):
await self._cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, InputAudioRawFrame):
await self._handle_input_audio(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStartedSpeakingFrame):
await self._handle_user_started_speaking(frame)
await self.push_frame(frame, direction)
@@ -320,18 +328,42 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
else:
await self.push_frame(frame, direction)
async def _process_aggregation(self):
"""Process the current aggregation and push it downstream."""
aggregation = self._aggregation
await self.reset()
await self.handle_aggregation(aggregation)
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
async def push_aggregation(self):
"""Pushes the current aggregation based on interruption strategies and conditions."""
if len(self._aggregation) > 0:
aggregation = self._aggregation
if self.interruption_strategies and self._bot_speaking:
should_interrupt = await self._should_interrupt_based_on_strategies()
# 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.reset()
if should_interrupt:
logger.debug(
"Interruption conditions met - pushing BotInterruptionFrame and aggregation"
)
await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
await self._process_aggregation()
else:
logger.debug("Interruption conditions not met - not pushing aggregation")
# Don't process aggregation, just reset it
await self.reset()
else:
# No interruption config - normal behavior (always push aggregation)
await self._process_aggregation()
await self.handle_aggregation(aggregation)
async def _should_interrupt_based_on_strategies(self) -> bool:
"""Check if interruption should occur based on configured strategies."""
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
async def should_interrupt(strategy: BaseInterruptionStrategy):
await strategy.append_text(self._aggregation)
return await strategy.should_interrupt()
return any([await should_interrupt(s) for s in self._interruption_strategies])
async def _start(self, frame: StartFrame):
self._create_aggregation_task()
@@ -342,6 +374,10 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def _cancel(self, frame: CancelFrame):
await self._cancel_aggregation_task()
async def _handle_input_audio(self, frame: InputAudioRawFrame):
for s in self.interruption_strategies:
await s.append_audio(frame.audio, frame.sample_rate)
async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
self._user_speaking = True
self._waiting_for_aggregation = True
@@ -427,7 +463,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
# If we reached this case and the bot is speaking, let's ignore
# what the user said.
logger.debug("Ignoring user speaking emulation, bot is speaking.")
self.reset()
await self.reset()
else:
# The bot is not speaking so, let's trigger user speaking
# emulation.
@@ -465,9 +501,18 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
self._params.expect_stripped_words = kwargs["expect_stripped_words"]
self._started = 0
self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
@property
def has_function_calls_in_progress(self) -> bool:
"""Check if there are any function calls currently in progress.
Returns:
bool: True if function calls are in progress, False otherwise
"""
return bool(self._function_calls_in_progress)
async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": "assistant", "content": aggregation})
@@ -503,6 +548,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
self.set_tools(frame.tools)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, FunctionCallsStartedFrame):
await self._handle_function_calls_started(frame)
elif isinstance(frame, FunctionCallInProgressFrame):
await self._handle_function_call_in_progress(frame)
elif isinstance(frame, FunctionCallResultFrame):
@@ -522,7 +569,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
return
aggregation = self._aggregation.strip()
self.reset()
await self.reset()
if aggregation:
await self.handle_aggregation(aggregation)
@@ -537,7 +584,13 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
async def _handle_interruptions(self, frame: StartInterruptionFrame):
await self.push_aggregation()
self._started = 0
self.reset()
await self.reset()
async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
for function_call in frame.function_calls:
self._function_calls_in_progress[function_call.tool_call_id] = None
async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
logger.debug(
@@ -562,19 +615,22 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
await self.handle_function_call_result(frame)
run_llm = False
# Run inference if the function call result requires it.
if frame.result:
run_llm = False
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
# If the tool call result has a run_llm property, use it.
run_llm = properties.run_llm
elif frame.run_llm is not None:
# If the frame is indicating we should run the LLM, do it.
run_llm = frame.run_llm
else:
# Default behavior is to run the LLM if there are no function calls in progress
# If this is the last function call in progress, run the LLM.
run_llm = not bool(self._function_calls_in_progress)
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Call the `on_context_updated` callback once the function call result
# is added to the context. Also, run this in a separate task to make
@@ -653,7 +709,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
# 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.reset()
await self.reset()
frame = LLMMessagesFrame(self._context.messages)
await self.push_frame(frame)
@@ -675,7 +731,7 @@ class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
# 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.reset()
await self.reset()
frame = LLMMessagesFrame(self._context.messages)
await self.push_frame(frame)

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

@@ -106,7 +106,7 @@ class OpenAILLMContext:
if "mime_type" in msg and msg["mime_type"].startswith("image/"):
msg["data"] = "..."
msgs.append(msg)
return json.dumps(msgs)
return json.dumps(msgs, ensure_ascii=False)
def from_standard_message(self, message):
"""Convert from OpenAI message format to OpenAI message format (passthrough).

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

@@ -23,4 +23,4 @@ class UserResponseAggregator(LLMUserResponseAggregator):
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
await self.reset()

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

@@ -7,15 +7,20 @@
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Awaitable, Callable, Coroutine, Optional
from typing import Awaitable, Callable, Coroutine, List, Optional, Sequence
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,
ErrorFrame,
Frame,
FrameProcessorPauseFrame,
FrameProcessorPauseUrgentFrame,
FrameProcessorResumeFrame,
FrameProcessorResumeUrgentFrame,
StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
@@ -67,6 +72,7 @@ class FrameProcessor(BaseObject):
self._enable_metrics = False
self._enable_usage_metrics = False
self._report_only_initial_ttfb = False
self._interruption_strategies: List[BaseInterruptionStrategy] = []
# Indicates whether we have received the StartFrame.
self.__started = False
@@ -119,6 +125,10 @@ class FrameProcessor(BaseObject):
def report_only_initial_ttfb(self):
return self._report_only_initial_ttfb
@property
def interruption_strategies(self) -> Sequence[BaseInterruptionStrategy]:
return self._interruption_strategies
def can_generate_metrics(self) -> bool:
return False
@@ -253,6 +263,10 @@ class FrameProcessor(BaseObject):
self._should_report_ttfb = True
elif isinstance(frame, CancelFrame):
await self.__cancel(frame)
elif isinstance(frame, (FrameProcessorPauseFrame, FrameProcessorPauseUrgentFrame)):
await self.__pause(frame)
elif isinstance(frame, (FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame)):
await self.__resume(frame)
async def push_error(self, error: ErrorFrame):
await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -272,6 +286,7 @@ class FrameProcessor(BaseObject):
self._enable_metrics = frame.enable_metrics
self._enable_usage_metrics = frame.enable_usage_metrics
self._report_only_initial_ttfb = frame.report_only_initial_ttfb
self._interruption_strategies = frame.interruption_strategies
self.__create_input_task()
self.__create_push_task()
@@ -280,6 +295,14 @@ class FrameProcessor(BaseObject):
await self.__cancel_input_task()
await self.__cancel_push_task()
async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
if frame.name == self.name:
await self.pause_processing_frames()
async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
if frame.name == self.name:
await self.resume_processing_frames()
#
# Handle interruptions
#

161
src/pipecat/serializers/exotel.py Normal file
View File

@@ -0,0 +1,161 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import json
from typing import Optional
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_default_resampler
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
InputAudioRawFrame,
InputDTMFFrame,
KeypadEntry,
StartFrame,
StartInterruptionFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
class ExotelFrameSerializer(FrameSerializer):
"""Serializer for Exotel Media Streams WebSocket protocol.
This serializer handles converting between Pipecat frames and Exotel's WebSocket
media streams protocol. It supports audio conversion, DTMF events, and automatic
call termination.
Ref Doc for events - https://support.exotel.com/support/solutions/articles/3000108630-working-with-the-stream-and-voicebot-applet
"""
class InputParams(BaseModel):
"""Configuration parameters for ExotelFrameSerializer.
Attributes:
exotel_sample_rate: Sample rate used by Exotel, defaults to 8000 Hz.
sample_rate: Optional override for pipeline input sample rate.
"""
exotel_sample_rate: int = 8000
sample_rate: Optional[int] = None
def __init__(
self, stream_sid: str, call_sid: Optional[str] = None, params: Optional[InputParams] = None
):
"""Initialize the ExotelFrameSerializer.
Args:
stream_sid: The Exotel Media Stream SID.
call_sid: The associated Exotel Call SID (optional, not used in this implementation).
params: Configuration parameters.
"""
self._stream_sid = stream_sid
self._call_sid = call_sid
self._params = params or ExotelFrameSerializer.InputParams()
self._exotel_sample_rate = self._params.exotel_sample_rate
self._sample_rate = 0 # Pipeline input rate
self._resampler = create_default_resampler()
@property
def type(self) -> FrameSerializerType:
"""Gets the serializer type.
Returns:
The serializer type, either TEXT or BINARY.
"""
return FrameSerializerType.TEXT
async def setup(self, frame: StartFrame):
"""Sets up the serializer with pipeline configuration.
Args:
frame: The StartFrame containing pipeline configuration.
"""
self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate
async def serialize(self, frame: Frame) -> str | bytes | None:
"""Serializes a Pipecat frame to Exotel WebSocket format.
Handles conversion of various frame types to Exotel WebSocket messages.
Args:
frame: The Pipecat frame to serialize.
Returns:
Serialized data as string or bytes, or None if the frame isn't handled.
"""
if isinstance(frame, StartInterruptionFrame):
answer = {"event": "clear", "streamSid": self._stream_sid}
return json.dumps(answer)
elif isinstance(frame, AudioRawFrame):
data = frame.audio
# Output: Exotel outputs PCM audio, but we need to resample to match requested sample_rate
serialized_data = await self._resampler.resample(
data, frame.sample_rate, self._exotel_sample_rate
)
payload = base64.b64encode(serialized_data).decode("ascii")
answer = {
"event": "media",
"streamSid": self._stream_sid,
"media": {"payload": payload},
}
return json.dumps(answer)
elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)):
return json.dumps(frame.message)
return None
async def deserialize(self, data: str | bytes) -> Frame | None:
"""Deserializes Exotel WebSocket data to Pipecat frames.
Handles conversion of Exotel media events to appropriate Pipecat frames.
Args:
data: The raw WebSocket data from Exotel.
Returns:
A Pipecat frame corresponding to the Exotel event, or None if unhandled.
"""
message = json.loads(data)
if message["event"] == "media":
payload_base64 = message["media"]["payload"]
payload = base64.b64decode(payload_base64)
deserialized_data = await self._resampler.resample(
payload,
self._exotel_sample_rate,
self._sample_rate,
)
# Input: Exotel takes PCM data, so just resample to match sample_rate
audio_frame = InputAudioRawFrame(
audio=deserialized_data,
num_channels=1, # Assuming mono audio from Exotel
sample_rate=self._sample_rate, # Use the configured pipeline input rate
)
return audio_frame
elif message["event"] == "dtmf":
digit = message.get("dtmf", {}).get("digit")
try:
return InputDTMFFrame(KeypadEntry(digit))
except ValueError:
# Handle case where string doesn't match any enum value
logger.info(f"Invalid DTMF digit: {digit}")
return None
return None

15
src/pipecat/serializers/protobuf.py
View File

@@ -41,6 +41,7 @@ class ProtobufFrameSerializer(FrameSerializer):
TextFrame: "text",
InputAudioRawFrame: "audio",
TranscriptionFrame: "transcription",
MessageFrame: "message",
}
DESERIALIZABLE_FIELDS = {v: k for k, v in DESERIALIZABLE_TYPES.items()}
@@ -97,8 +98,18 @@ class ProtobufFrameSerializer(FrameSerializer):
if "pts" in args_dict:
del args_dict["pts"]
# Create the instance
instance = class_name(**args_dict)
# Special handling for MessageFrame -> TransportMessageUrgentFrame
if class_name == MessageFrame:
try:
msg = json.loads(args_dict["data"])
instance = TransportMessageUrgentFrame(message=msg)
logger.debug(f"ProtobufFrameSerializer: Transport message {instance}")
except Exception as e:
logger.error(f"Error parsing MessageFrame data: {e}")
return None
else:
# Normal deserialization, create the instance
instance = class_name(**args_dict)
# Set special fields
if id:

20
src/pipecat/services/anthropic/llm.py
View File

@@ -45,7 +45,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.utils.tracing.service_decorators import traced_llm
try:
@@ -202,9 +202,8 @@ class AnthropicLLMService(LLMService):
tool_use_block = None
json_accumulator = ""
function_calls = []
async for event in response:
# logger.debug(f"Anthropic LLM event: {event}")
# Aggregate streaming content, create frames, trigger events
if event.type == "content_block_delta":
@@ -226,11 +225,14 @@ class AnthropicLLMService(LLMService):
and event.delta.stop_reason == "tool_use"
):
if tool_use_block:
await self.call_function(
context=context,
tool_call_id=tool_use_block.id,
function_name=tool_use_block.name,
arguments=json.loads(json_accumulator) if json_accumulator else dict(),
args = json.loads(json_accumulator) if json_accumulator else {}
function_calls.append(
FunctionCallFromLLM(
context=context,
tool_call_id=tool_use_block.id,
function_name=tool_use_block.name,
arguments=args,
)
)
# Calculate usage. Do this here in its own if statement, because there may be usage
@@ -277,6 +279,8 @@ class AnthropicLLMService(LLMService):
if total_input_tokens >= 1024:
context.turns_above_cache_threshold += 1
await self.run_function_calls(function_calls)
except asyncio.CancelledError:
# If we're interrupted, we won't get a complete usage report. So set our flag to use the
# token estimate. The reraise the exception so all the processors running in this task

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

@@ -98,7 +98,7 @@ class AssemblyAISTTService(STTService):
self._audio_buffer = self._audio_buffer[self._chunk_size_bytes :]
await self._websocket.send(chunk)
yield Frame()
yield None
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)

16
src/pipecat/services/aws/llm.py
View File

@@ -21,6 +21,7 @@ from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter
from pipecat.frames.frames import (
Frame,
FunctionCallCancelFrame,
FunctionCallFromLLM,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMFullResponseEndFrame,
@@ -708,6 +709,7 @@ class AWSBedrockLLMService(LLMService):
tool_use_block = None
json_accumulator = ""
function_calls = []
for event in response["stream"]:
# Handle text content
if "contentBlockDelta" in event:
@@ -740,11 +742,13 @@ class AWSBedrockLLMService(LLMService):
# Only call function if it's not the no_operation tool
if not using_noop_tool:
await self.call_function(
context=context,
tool_call_id=tool_use_block["id"],
function_name=tool_use_block["name"],
arguments=arguments,
function_calls.append(
FunctionCallFromLLM(
context=context,
tool_call_id=tool_use_block["id"],
function_name=tool_use_block["name"],
arguments=arguments,
)
)
else:
logger.debug("Ignoring no_operation tool call")
@@ -758,7 +762,7 @@ class AWSBedrockLLMService(LLMService):
completion_tokens += usage.get("outputTokens", 0)
cache_read_input_tokens += usage.get("cacheReadInputTokens", 0)
cache_creation_input_tokens += usage.get("cacheWriteInputTokens", 0)
await self.run_function_calls(function_calls)
except asyncio.CancelledError:
# If we're interrupted, we won't get a complete usage report. So set our flag to use the
# token estimate. The reraise the exception so all the processors running in this task

1
src/pipecat/services/aws/stt.py
View File

@@ -266,6 +266,7 @@ class AWSTranscribeSTTService(STTService):
Language.JA: "ja-JP",
Language.KO: "ko-KR",
Language.ZH: "zh-CN",
Language.PL: "pl-PL",
}
return language_map.get(language)

3
src/pipecat/services/aws/tts.py
View File

@@ -253,7 +253,8 @@ class AWSPollyTTSService(TTSService):
yield TTSStartedFrame()
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
for i in range(0, len(audio_data), CHUNK_SIZE):
chunk = audio_data[i : i + CHUNK_SIZE]
if len(chunk) > 0:

3
src/pipecat/services/cartesia/__init__.py
View File

@@ -8,6 +8,7 @@ import sys
from pipecat.services import DeprecatedModuleProxy
from .stt import *
from .tts import *
sys.modules[__name__] = DeprecatedModuleProxy(globals(), "cartesia", "cartesia.tts")
sys.modules[__name__] = DeprecatedModuleProxy(globals(), "cartesia", "cartesia.[stt,tts]")

239
src/pipecat/services/cartesia/stt.py Normal file
View File

@@ -0,0 +1,239 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import json
import urllib.parse
from typing import AsyncGenerator, Optional
import websockets
from loguru import logger
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
InterimTranscriptionFrame,
StartFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
class CartesiaLiveOptions:
def __init__(
self,
*,
model: str = "ink-whisper",
language: str = Language.EN.value,
encoding: str = "pcm_s16le",
sample_rate: int = 16000,
**kwargs,
):
self.model = model
self.language = language
self.encoding = encoding
self.sample_rate = sample_rate
self.additional_params = kwargs
def to_dict(self):
params = {
"model": self.model,
"language": self.language if isinstance(self.language, str) else self.language.value,
"encoding": self.encoding,
"sample_rate": str(self.sample_rate),
}
return params
def items(self):
return self.to_dict().items()
def get(self, key, default=None):
if hasattr(self, key):
return getattr(self, key)
return self.additional_params.get(key, default)
@classmethod
def from_json(cls, json_str: str) -> "CartesiaLiveOptions":
return cls(**json.loads(json_str))
class CartesiaSTTService(STTService):
def __init__(
self,
*,
api_key: str,
base_url: str = "",
sample_rate: int = 16000,
live_options: Optional[CartesiaLiveOptions] = None,
**kwargs,
):
sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
super().__init__(sample_rate=sample_rate, **kwargs)
default_options = CartesiaLiveOptions(
model="ink-whisper",
language=Language.EN.value,
encoding="pcm_s16le",
sample_rate=sample_rate,
)
merged_options = default_options
if live_options:
merged_options_dict = default_options.to_dict()
merged_options_dict.update(live_options.to_dict())
merged_options = CartesiaLiveOptions(
**{
k: v
for k, v in merged_options_dict.items()
if not isinstance(v, str) or v != "None"
}
)
self._settings = merged_options
self.set_model_name(merged_options.model)
self._api_key = api_key
self._base_url = base_url or "api.cartesia.ai"
self._connection = None
self._receiver_task = None
def can_generate_metrics(self) -> bool:
return True
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
# If the connection is closed, due to timeout, we need to reconnect when the user starts speaking again
if not self._connection or self._connection.closed:
await self._connect()
await self._connection.send(audio)
yield None
async def _connect(self):
params = self._settings.to_dict()
ws_url = f"wss://{self._base_url}/stt/websocket?{urllib.parse.urlencode(params)}"
logger.debug(f"Connecting to Cartesia: {ws_url}")
headers = {"Cartesia-Version": "2025-04-16", "X-API-Key": self._api_key}
try:
self._connection = await websockets.connect(ws_url, extra_headers=headers)
# Setup the receiver task to handle the incoming messages from the Cartesia server
if self._receiver_task is None or self._receiver_task.done():
self._receiver_task = asyncio.create_task(self._receive_messages())
logger.debug(f"Connected to Cartesia")
except Exception as e:
logger.error(f"{self}: unable to connect to Cartesia: {e}")
async def _receive_messages(self):
try:
while True:
if not self._connection or self._connection.closed:
break
message = await self._connection.recv()
try:
data = json.loads(message)
await self._process_response(data)
except json.JSONDecodeError:
logger.warning(f"Received non-JSON message: {message}")
except asyncio.CancelledError:
pass
except websockets.exceptions.ConnectionClosed as e:
logger.debug(f"WebSocket connection closed: {e}")
except Exception as e:
logger.error(f"Error in message receiver: {e}")
async def _process_response(self, data):
if "type" in data:
if data["type"] == "transcript":
await self._on_transcript(data)
elif data["type"] == "error":
logger.error(f"Cartesia error: {data.get('message', 'Unknown error')}")
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def _on_transcript(self, data):
if "text" not in data:
return
transcript = data.get("text", "")
is_final = data.get("is_final", False)
language = None
if "language" in data:
try:
language = Language(data["language"])
except (ValueError, KeyError):
pass
if len(transcript) > 0:
await self.stop_ttfb_metrics()
if is_final:
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), language)
)
await self._handle_transcription(transcript, is_final, language)
await self.stop_processing_metrics()
else:
# For interim transcriptions, just push the frame without tracing
await self.push_frame(
InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language)
)
async def _disconnect(self):
if self._receiver_task:
self._receiver_task.cancel()
try:
await self._receiver_task
except asyncio.CancelledError:
pass
except Exception as e:
logger.exception(f"Unexpected exception while cancelling task: {e}")
self._receiver_task = None
if self._connection and self._connection.open:
logger.debug("Disconnecting from Cartesia")
await self._connection.close()
self._connection = None
async def start_metrics(self):
await self.start_ttfb_metrics()
await self.start_processing_metrics()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, UserStartedSpeakingFrame):
await self.start_metrics()
elif isinstance(frame, UserStoppedSpeakingFrame):
# Send finalize command to flush the transcription session
if self._connection and self._connection.open:
await self._connection.send("finalize")

31
src/pipecat/services/elevenlabs/tts.py
View File

@@ -279,9 +279,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._disconnect()
async def flush_audio(self):
if self._websocket and self._context_id:
msg = {"context_id": self._context_id, "flush": True}
await self._websocket.send(json.dumps(msg))
if not self._context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
msg = {"context_id": self._context_id, "flush": True}
await self._websocket.send(json.dumps(msg))
self._context_id = None
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
@@ -389,14 +392,9 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async for message in self._get_websocket():
msg = json.loads(message)
# Check if this message belongs to the current context
# The default context may return null/None for context_id
received_ctx_id = msg.get("contextId")
if (
self._context_id is not None
and received_ctx_id is not None
and received_ctx_id != self._context_id
):
logger.trace(f"Ignoring message from different context: {received_ctx_id}")
if not self.audio_context_available(received_ctx_id):
logger.trace(f"Ignoring message from unavailable context: {received_ctx_id}")
continue
if msg.get("audio"):
@@ -405,14 +403,15 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
audio = base64.b64decode(msg["audio"])
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
await self.push_frame(frame)
await self.append_to_audio_context(received_ctx_id, frame)
if msg.get("alignment"):
word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
await self.add_word_timestamps(word_times)
self._cumulative_time = word_times[-1][1]
if msg.get("isFinal"):
logger.trace(f"Received final message for context {received_ctx_id}")
# Context has finished
await self.remove_audio_context(received_ctx_id)
# Reset context tracking if this was our active context
if self._context_id == received_ctx_id:
self._context_id = None
self._started = False
@@ -464,6 +463,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._websocket.send(
json.dumps({"context_id": self._context_id, "close_context": True})
)
await self.remove_audio_context(self._context_id)
self._context_id = None
if not self._started:
@@ -471,6 +471,9 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
yield TTSStartedFrame()
self._started = True
self._cumulative_time = 0
# Create new context ID and register it
self._context_id = str(uuid.uuid4())
await self.create_audio_context(self._context_id)
await self._send_text(text)
await self.start_tts_usage_metrics(text)
@@ -478,7 +481,9 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
logger.error(f"{self} error sending message: {e}")
yield TTSStoppedFrame()
self._started = False
self._context_id = None
if self._context_id:
await self.remove_audio_context(self._context_id)
self._context_id = None
return
yield None
except Exception as e:

68
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -52,7 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
@@ -60,6 +60,7 @@ from pipecat.services.openai.llm import (
from pipecat.transcriptions.language import Language
from pipecat.utils.string import match_endofsentence
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt, traced_tts
from . import events
@@ -335,7 +336,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
*,
api_key: str,
base_url: str = "generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent",
model="models/gemini-2.5-flash-preview-native-audio-dialog",
model="models/gemini-2.0-flash-live-001",
voice_id: str = "Charon",
start_audio_paused: bool = False,
start_video_paused: bool = False,
@@ -378,6 +379,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._last_transcription_sent = ""
self._bot_audio_buffer = bytearray()
self._bot_text_buffer = ""
self._llm_output_buffer = ""
self._sample_rate = 24000
@@ -471,6 +473,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
async def _handle_user_stopped_speaking(self, frame):
self._user_is_speaking = False
self._user_audio_buffer = bytearray()
await self.start_ttfb_metrics()
if self._needs_turn_complete_message:
self._needs_turn_complete_message = False
evt = events.ClientContentMessage.model_validate(
@@ -752,6 +755,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
logger.debug(f"Creating initial response: {messages}")
await self.start_ttfb_metrics()
evt = events.ClientContentMessage.model_validate(
{
"clientContent": {
@@ -793,6 +798,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
return
logger.debug(f"Creating response: {messages}")
await self.start_ttfb_metrics()
evt = events.ClientContentMessage.model_validate(
{
"clientContent": {
@@ -803,6 +810,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
await self.send_client_event(evt)
@traced_gemini_live(operation="llm_tool_result")
async def _tool_result(self, tool_result_message):
# For now we're shoving the name into the tool_call_id field, so this
# will work until we revisit that.
@@ -827,6 +835,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self._websocket.send(response_message)
# await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}}))
@traced_gemini_live(operation="llm_setup")
async def _handle_evt_setup_complete(self, evt):
# If this is our first context frame, run the LLM
self._api_session_ready = True
@@ -840,6 +849,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
if not part:
return
await self.stop_ttfb_metrics()
# part.text is added when `modalities` is set to TEXT; otherwise, it's None
text = part.text
if text:
@@ -873,26 +884,48 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
await self.push_frame(frame)
@traced_gemini_live(operation="llm_tool_call")
async def _handle_evt_tool_call(self, evt):
function_calls = evt.toolCall.functionCalls
if not function_calls:
return
if not self._context:
logger.error("Function calls are not supported without a context object.")
for call in function_calls:
await self.call_function(
context=self._context,
tool_call_id=call.id,
function_name=call.name,
arguments=call.args,
)
function_calls_llm = [
FunctionCallFromLLM(
context=self._context,
tool_call_id=f.id,
function_name=f.name,
arguments=f.args,
)
for f in function_calls
]
await self.run_function_calls(function_calls_llm)
@traced_gemini_live(operation="llm_response")
async def _handle_evt_turn_complete(self, evt):
self._bot_is_speaking = False
text = self._bot_text_buffer
self._bot_text_buffer = ""
# Only push the TTSStoppedFrame the bot is outputting audio
# Determine output and modality for tracing
if text:
# TEXT modality
output_text = text
output_modality = "TEXT"
else:
# AUDIO modality
output_text = self._llm_output_buffer
output_modality = "AUDIO"
# Trace the complete LLM response (this will be handled by the decorator)
# The decorator will extract the output text and usage metadata from the event
self._bot_text_buffer = ""
self._llm_output_buffer = ""
# Only push the TTSStoppedFrame if the bot is outputting audio
# when text is found, modalities is set to TEXT and no audio
# is produced.
if not text:
@@ -900,6 +933,13 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self.push_frame(LLMFullResponseEndFrame())
@traced_stt
async def _handle_user_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def _handle_evt_input_transcription(self, evt):
"""Handle the input transcription event.
@@ -935,6 +975,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Send a TranscriptionFrame with the complete sentence
logger.debug(f"[Transcription:user] [{complete_sentence}]")
await self._handle_user_transcription(
complete_sentence, True, self._settings["language"]
)
await self.push_frame(
TranscriptionFrame(
text=complete_sentence,
@@ -957,6 +1000,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
if not text:
return
# Collect text for tracing
self._llm_output_buffer += text
await self.push_frame(LLMTextFrame(text=text))
await self.push_frame(TTSTextFrame(text=text))

2
src/pipecat/services/gladia/config.py
View File

@@ -77,6 +77,7 @@ class TranslationConfig(BaseModel):
lipsync: Whether to enable lip-sync optimization for translations
context_adaptation: Whether to enable context-aware translation adaptation
context: Additional context to help with translation accuracy
informal: Force informal language forms when available
"""
target_languages: Optional[List[str]] = None
@@ -85,6 +86,7 @@ class TranslationConfig(BaseModel):
lipsync: Optional[bool] = None
context_adaptation: Optional[bool] = None
context: Optional[str] = None
informal: Optional[bool] = None
class RealtimeProcessingConfig(BaseModel):

22
src/pipecat/services/google/llm.py
View File

@@ -42,7 +42,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchResponseFrame
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
@@ -83,7 +83,7 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
await self.reset()
class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
@@ -555,9 +555,11 @@ class GoogleLLMService(LLMService):
contents=messages,
config=generation_config,
)
await self.stop_ttfb_metrics()
function_calls = []
async for chunk in response:
# Stop TTFB metrics after the first chunk
await self.stop_ttfb_metrics()
if chunk.usage_metadata:
prompt_tokens += chunk.usage_metadata.prompt_token_count or 0
completion_tokens += chunk.usage_metadata.candidates_token_count or 0
@@ -576,11 +578,13 @@ class GoogleLLMService(LLMService):
function_call = part.function_call
id = function_call.id or str(uuid.uuid4())
logger.debug(f"Function call: {function_call.name}:{id}")
await self.call_function(
context=context,
tool_call_id=id,
function_name=function_call.name,
arguments=function_call.args or {},
function_calls.append(
FunctionCallFromLLM(
context=context,
tool_call_id=id,
function_name=function_call.name,
arguments=function_call.args or {},
)
)
if (
@@ -621,6 +625,8 @@ class GoogleLLMService(LLMService):
"rendered_content": rendered_content,
"origins": origins,
}
await self.run_function_calls(function_calls)
except DeadlineExceeded:
await self._call_event_handler("on_completion_timeout")
except Exception as e:

28
src/pipecat/services/google/llm_openai.py
View File

@@ -10,6 +10,8 @@ import os
from openai import AsyncStream
from openai.types.chat import ChatCompletionChunk
from pipecat.services.llm_service import FunctionCallFromLLM
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
@@ -18,7 +20,6 @@ from loguru import logger
from pipecat.frames.frames import LLMTextFrame
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.base_llm import OpenAIUnhandledFunctionException
from pipecat.services.openai.llm import OpenAILLMService
@@ -112,25 +113,26 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
logger.debug(
f"Function list: {functions_list}, Arguments list: {arguments_list}, Tool ID list: {tool_id_list}"
)
for index, (function_name, arguments, tool_id) in enumerate(
zip(functions_list, arguments_list, tool_id_list), start=1
function_calls = []
for function_name, arguments, tool_id in zip(
functions_list, arguments_list, tool_id_list
):
if function_name == "":
# TODO: Remove the _process_context method once Google resolves the bug
# where the index is incorrectly set to None instead of returning the actual index,
# which currently results in an empty function name('').
continue
if self.has_function(function_name):
run_llm = False
arguments = json.loads(arguments)
await self.call_function(
arguments = json.loads(arguments)
function_calls.append(
FunctionCallFromLLM(
context=context,
tool_call_id=tool_id,
function_name=function_name,
arguments=arguments,
tool_call_id=tool_id,
run_llm=run_llm,
)
else:
raise OpenAIUnhandledFunctionException(
f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."
)
)
await self.run_function_calls(function_calls)

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

@@ -747,6 +747,11 @@ class GoogleSTTService(STTService):
try:
while True:
try:
if self._request_queue.empty():
# wait for 10ms in case we don't have audio
await asyncio.sleep(0.01)
continue
# Start bi-directional streaming
streaming_recognize = await self._client.streaming_recognize(
requests=self._request_generator()

7
src/pipecat/services/google/tts.py
View File

@@ -362,8 +362,8 @@ class GoogleHttpTTSService(TTSService):
# Skip the first 44 bytes to remove the WAV header
audio_content = response.audio_content[44:]
# Read and yield audio data in chunks
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
for i in range(0, len(audio_content), CHUNK_SIZE):
chunk = audio_content[i : i + CHUNK_SIZE]
if not chunk:
@@ -505,9 +505,10 @@ class GoogleTTSService(TTSService):
yield TTSStartedFrame()
audio_buffer = b""
CHUNK_SIZE = 1024
first_chunk_for_ttfb = False
CHUNK_SIZE = self.chunk_size
async for response in streaming_responses:
chunk = response.audio_content
if not chunk:

38
src/pipecat/services/groq/tts.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import io
import wave
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -78,22 +80,26 @@ class GroqTTSService(TTSService):
await self.start_ttfb_metrics()
yield TTSStartedFrame()
response = await self._client.audio.speech.create(
model=self._model_name,
voice=self._voice_id,
response_format=self._output_format,
input=text,
)
try:
response = await self._client.audio.speech.create(
model=self._model_name,
voice=self._voice_id,
response_format=self._output_format,
input=text,
)
async for data in response.iter_bytes():
if measuring_ttfb:
await self.stop_ttfb_metrics()
measuring_ttfb = False
# remove wav header if present
if data.startswith(b"RIFF"):
data = data[44:]
if len(data) == 0:
continue
yield TTSAudioRawFrame(data, self.sample_rate, 1)
async for data in response.iter_bytes():
if measuring_ttfb:
await self.stop_ttfb_metrics()
measuring_ttfb = False
with wave.open(io.BytesIO(data)) as w:
channels = w.getnchannels()
frame_rate = w.getframerate()
num_frames = w.getnframes()
bytes = w.readframes(num_frames)
yield TTSAudioRawFrame(bytes, frame_rate, channels)
except Exception as e:
logger.error(f"{self} exception: {e}")
yield TTSStoppedFrame()

279
src/pipecat/services/llm_service.py
View File

@@ -7,18 +7,23 @@
import asyncio
import inspect
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Set, Tuple, Type
from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Sequence, Type
from loguru import logger
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallFromLLM,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallResultProperties,
FunctionCallsStartedFrame,
StartFrame,
StartInterruptionFrame,
UserImageRequestFrame,
)
@@ -41,22 +46,6 @@ class FunctionCallResultCallback(Protocol):
) -> None: ...
@dataclass
class FunctionCallEntry:
"""Represents an internal entry for a function call.
Attributes:
function_name (Optional[str]): The name of the function.
handler (FunctionCallHandler): The handler for processing function call parameters.
cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption.
"""
function_name: Optional[str]
handler: FunctionCallHandler
cancel_on_interruption: bool
@dataclass
class FunctionCallParams:
"""Parameters for a function call.
@@ -79,20 +68,78 @@ class FunctionCallParams:
result_callback: FunctionCallResultCallback
@dataclass
class FunctionCallRegistryItem:
"""Represents an entry in our function call registry. This is what the user
registers.
Attributes:
function_name (Optional[str]): The name of the function.
handler (FunctionCallHandler): The handler for processing function call parameters.
cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption.
"""
function_name: Optional[str]
handler: FunctionCallHandler
cancel_on_interruption: bool
@dataclass
class FunctionCallRunnerItem:
"""Represents an internal function call entry to our function call
runner. The runner executes function calls in order.
Attributes:
registry_name (Optional[str]): The function call name registration (could be None).
function_name (str): The name of the function.
tool_call_id (str): A unique identifier for the function call.
arguments (Mapping[str, Any]): The arguments for the function.
context (OpenAILLMContext): The LLM context.
"""
registry_item: FunctionCallRegistryItem
function_name: str
tool_call_id: str
arguments: Mapping[str, Any]
context: OpenAILLMContext
run_llm: Optional[bool] = None
class LLMService(AIService):
"""This class is a no-op but serves as a base class for LLM services."""
"""This is the base class for all LLM services. It handles function calling
registration and execution. The class also provides event handlers.
An event to know when an LLM service completion timeout occurs:
@task.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
...
And an event to know that function calls have been received from the LLM
service and that we are going to start executing them:
@task.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls: Sequence[FunctionCallFromLLM]):
...
"""
# OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations.
# However, subclasses should override this with a more specific adapter when necessary.
adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter
def __init__(self, **kwargs):
def __init__(self, run_in_parallel: bool = True, **kwargs):
super().__init__(**kwargs)
self._functions = {}
self._run_in_parallel = run_in_parallel
self._start_callbacks = {}
self._adapter = self.adapter_class()
self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set()
self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {}
self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {}
self._sequential_runner_task: Optional[asyncio.Task] = None
self._register_event_handler("on_function_calls_started")
self._register_event_handler("on_completion_timeout")
def get_llm_adapter(self) -> BaseLLMAdapter:
@@ -107,13 +154,28 @@ class LLMService(AIService):
) -> Any:
pass
async def start(self, frame: StartFrame):
await super().start(frame)
if not self._run_in_parallel:
await self._create_sequential_runner_task()
async def stop(self, frame: EndFrame):
await super().stop(frame)
if not self._run_in_parallel:
await self._cancel_sequential_runner_task()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
if not self._run_in_parallel:
await self._cancel_sequential_runner_task()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self._handle_interruptions(frame)
async def _handle_interruptions(self, frame: StartInterruptionFrame):
async def _handle_interruptions(self, _: StartInterruptionFrame):
for function_name, entry in self._functions.items():
if entry.cancel_on_interruption:
await self._cancel_function_call(function_name)
@@ -124,11 +186,11 @@ class LLMService(AIService):
handler: Any,
start_callback=None,
*,
cancel_on_interruption: bool = False,
cancel_on_interruption: bool = True,
):
# Registering a function with the function_name set to None will run
# that handler for all functions
self._functions[function_name] = FunctionCallEntry(
self._functions[function_name] = FunctionCallRegistryItem(
function_name=function_name,
handler=handler,
cancel_on_interruption=cancel_on_interruption,
@@ -157,25 +219,43 @@ class LLMService(AIService):
return True
return function_name in self._functions.keys()
async def call_function(
self,
*,
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: Mapping[str, Any],
run_llm: bool = True,
):
if not function_name in self._functions.keys() and not None in self._functions.keys():
async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]):
if len(function_calls) == 0:
return
task = self.create_task(
self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
)
await self._call_event_handler("on_function_calls_started", function_calls)
self._function_call_tasks.add((task, tool_call_id, function_name))
# Push frame both downstream and upstream
started_frame_downstream = FunctionCallsStartedFrame(function_calls=function_calls)
started_frame_upstream = FunctionCallsStartedFrame(function_calls=function_calls)
await self.push_frame(started_frame_downstream, FrameDirection.DOWNSTREAM)
await self.push_frame(started_frame_upstream, FrameDirection.UPSTREAM)
task.add_done_callback(self._function_call_task_finished)
for function_call in function_calls:
if function_call.function_name in self._functions.keys():
item = self._functions[function_call.function_name]
elif None in self._functions.keys():
item = self._functions[None]
else:
logger.warning(
f"{self} is calling '{function_call.function_name}', but it's not registered."
)
continue
runner_item = FunctionCallRunnerItem(
registry_item=item,
function_name=function_call.function_name,
tool_call_id=function_call.tool_call_id,
arguments=function_call.arguments,
context=function_call.context,
)
if self._run_in_parallel:
task = self.create_task(self._run_function_call(runner_item))
self._function_call_tasks[task] = runner_item
task.add_done_callback(self._function_call_task_finished)
else:
await self._sequential_runner_queue.put(runner_item)
async def call_start_function(self, context: OpenAILLMContext, function_name: str):
if function_name in self._start_callbacks.keys():
@@ -203,43 +283,57 @@ class LLMService(AIService):
FrameDirection.UPSTREAM,
)
async def _run_function_call(
self,
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: Mapping[str, Any],
run_llm: bool = True,
):
if function_name in self._functions.keys():
entry = self._functions[function_name]
async def _create_sequential_runner_task(self):
if not self._sequential_runner_task:
self._sequential_runner_queue = asyncio.Queue()
self._sequential_runner_task = self.create_task(self._sequential_runner_handler())
async def _cancel_sequential_runner_task(self):
if self._sequential_runner_task:
await self.cancel_task(self._sequential_runner_task)
self._sequential_runner_task = None
async def _sequential_runner_handler(self):
while True:
runner_item = await self._sequential_runner_queue.get()
task = self.create_task(self._run_function_call(runner_item))
self._function_call_tasks[task] = runner_item
# Since we run tasks sequentially we don't need to call
# task.add_done_callback(self._function_call_task_finished).
await self.wait_for_task(task)
del self._function_call_tasks[task]
async def _run_function_call(self, runner_item: FunctionCallRunnerItem):
if runner_item.function_name in self._functions.keys():
item = self._functions[runner_item.function_name]
elif None in self._functions.keys():
entry = self._functions[None]
item = self._functions[None]
else:
return
logger.debug(
f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}"
f"{self} Calling function [{runner_item.function_name}:{runner_item.tool_call_id}] with arguments {runner_item.arguments}"
)
# NOTE(aleix): This needs to be removed after we remove the deprecation.
await self.call_start_function(context, function_name)
await self.call_start_function(runner_item.context, runner_item.function_name)
# Push a SystemFrame downstream. This frame will let our assistant context aggregator
# know that we are in the middle of a function call. Some contexts/aggregators may
# not need this. But some definitely do (Anthropic, for example).
# Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
# Push a function call in-progress downstream. This frame will let our
# assistant context aggregator know that we are in the middle of a
# function call. Some contexts/aggregators may not need this. But some
# definitely do (Anthropic, for example). Also push it upstream for use
# by other processors, like STTMuteFilter.
progress_frame_downstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
cancel_on_interruption=entry.cancel_on_interruption,
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
cancel_on_interruption=item.cancel_on_interruption,
)
progress_frame_upstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
cancel_on_interruption=entry.cancel_on_interruption,
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
cancel_on_interruption=item.cancel_on_interruption,
)
# Push frame both downstream and upstream
@@ -251,24 +345,26 @@ class LLMService(AIService):
result: Any, *, properties: Optional[FunctionCallResultProperties] = None
):
result_frame_downstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
result=result,
run_llm=runner_item.run_llm,
properties=properties,
)
result_frame_upstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
result=result,
run_llm=runner_item.run_llm,
properties=properties,
)
await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
signature = inspect.signature(entry.handler)
signature = inspect.signature(item.handler)
if len(signature.parameters) > 1:
import warnings
@@ -279,24 +375,32 @@ class LLMService(AIService):
DeprecationWarning,
)
await entry.handler(
function_name, tool_call_id, arguments, self, context, function_call_result_callback
await item.handler(
runner_item.function_name,
runner_item.tool_call_id,
runner_item.arguments,
self,
runner_item.context,
function_call_result_callback,
)
else:
params = FunctionCallParams(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
function_name=runner_item.function_name,
tool_call_id=runner_item.tool_call_id,
arguments=runner_item.arguments,
llm=self,
context=context,
context=runner_item.context,
result_callback=function_call_result_callback,
)
await entry.handler(params)
await item.handler(params)
async def _cancel_function_call(self, function_name: str):
async def _cancel_function_call(self, function_name: Optional[str]):
cancelled_tasks = set()
for task, tool_call_id, name in self._function_call_tasks:
if name == function_name:
for task, runner_item in self._function_call_tasks.items():
if runner_item.registry_item.function_name == function_name:
name = runner_item.function_name
tool_call_id = runner_item.tool_call_id
# We remove the callback because we are going to cancel the task
# now, otherwise we will be removing it from the set while we
# are iterating.
@@ -306,23 +410,20 @@ class LLMService(AIService):
await self.cancel_task(task)
frame = FunctionCallCancelFrame(
function_name=function_name, tool_call_id=tool_call_id
)
frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id)
await self.push_frame(frame)
logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
cancelled_tasks.add(task)
logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
# Remove all cancelled tasks from our set.
for task in cancelled_tasks:
self._function_call_task_finished(task)
def _function_call_task_finished(self, task: asyncio.Task):
tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None)
if tuple_to_remove:
self._function_call_tasks.discard(tuple_to_remove)
if task in self._function_call_tasks:
del self._function_call_tasks[task]
# The task is finished so this should exit immediately. We need to
# do this because otherwise the task manager would report a dangling
# task if we don't remove it.

9
src/pipecat/services/minimax/tts.py
View File

@@ -227,7 +227,8 @@ class MiniMaxHttpTTSService(TTSService):
# Process the streaming response
buffer = bytearray()
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
if not chunk:
@@ -279,10 +280,8 @@ class MiniMaxHttpTTSService(TTSService):
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(
audio=audio_chunk,
sample_rate=self._settings["audio_setting"][
"sample_rate"
],
num_channels=self._settings["audio_setting"]["channel"],
sample_rate=self.sample_rate,
num_channels=1,
)
except ValueError as e:
logger.error(f"Error converting hex to binary: {e}")

29
src/pipecat/services/openai/base_llm.py
View File

@@ -34,14 +34,10 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.utils.tracing.service_decorators import traced_llm
class OpenAIUnhandledFunctionException(Exception):
pass
class BaseOpenAILLMService(LLMService):
"""This is the base for all services that use the AsyncOpenAI client.
@@ -260,23 +256,22 @@ class BaseOpenAILLMService(LLMService):
arguments_list.append(arguments)
tool_id_list.append(tool_call_id)
for index, (function_name, arguments, tool_id) in enumerate(
zip(functions_list, arguments_list, tool_id_list), start=1
function_calls = []
for function_name, arguments, tool_id in zip(
functions_list, arguments_list, tool_id_list
):
if self.has_function(function_name):
run_llm = False
arguments = json.loads(arguments)
await self.call_function(
arguments = json.loads(arguments)
function_calls.append(
FunctionCallFromLLM(
context=context,
tool_call_id=tool_id,
function_name=function_name,
arguments=arguments,
tool_call_id=tool_id,
run_llm=run_llm,
)
else:
raise OpenAIUnhandledFunctionException(
f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."
)
)
await self.run_function_calls(function_calls)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)

2
src/pipecat/services/openai/tts.py
View File

@@ -125,7 +125,7 @@ class OpenAITTSService(TTSService):
await self.start_tts_usage_metrics(text)
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
yield TTSStartedFrame()
async for chunk in r.iter_bytes(CHUNK_SIZE):

49
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -48,9 +48,11 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import OpenAIContextAggregatorPair
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt, traced_tts
from . import events
from .context import (
@@ -76,10 +78,6 @@ class CurrentAudioResponse:
total_size: int = 0
class OpenAIUnhandledFunctionException(Exception):
pass
class OpenAIRealtimeBetaLLMService(LLMService):
# Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one.
adapter_class = OpenAIRealtimeLLMAdapter
@@ -100,6 +98,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self.api_key = api_key
self.base_url = full_url
self.set_model_name(model)
self._session_properties: events.SessionProperties = (
session_properties or events.SessionProperties()
@@ -402,6 +401,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# errors are fatal, so exit the receive loop
return
@traced_openai_realtime(operation="llm_setup")
async def _handle_evt_session_created(self, evt):
# session.created is received right after connecting. Send a message
# to configure the session properties.
@@ -467,6 +467,13 @@ class OpenAIRealtimeBetaLLMService(LLMService):
InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
)
@traced_stt
async def _handle_user_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def handle_evt_input_audio_transcription_completed(self, evt):
await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
@@ -475,6 +482,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# no way to get a language code?
TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
)
await self._handle_user_transcription(evt.transcript, True, Language.EN)
pair = self._user_and_response_message_tuple
if pair:
user, assistant = pair
@@ -493,6 +501,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
for future in futures:
future.set_result(evt.item)
@traced_openai_realtime(operation="llm_response")
async def _handle_evt_response_done(self, evt):
# todo: figure out whether there's anything we need to do for "cancelled" events
# usage metrics
@@ -574,25 +583,18 @@ class OpenAIRealtimeBetaLLMService(LLMService):
await self._handle_function_call_items(function_calls)
async def _handle_function_call_items(self, items):
total_items = len(items)
for index, item in enumerate(items):
function_name = item.name
tool_id = item.call_id
arguments = json.loads(item.arguments)
if self.has_function(function_name):
run_llm = index == total_items - 1
if function_name in self._functions.keys() or None in self._functions.keys():
await self.call_function(
context=self._context,
tool_call_id=tool_id,
function_name=function_name,
arguments=arguments,
run_llm=run_llm,
)
else:
raise OpenAIUnhandledFunctionException(
f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."
function_calls = []
for item in items:
args = json.loads(item.arguments)
function_calls.append(
FunctionCallFromLLM(
context=self._context,
tool_call_id=item.call_id,
function_name=item.name,
arguments=args,
)
)
await self.run_function_calls(function_calls)
#
# state and client events for the current conversation
@@ -609,6 +611,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self._context.llm_needs_initial_messages = True
await self._connect()
@traced_openai_realtime(operation="llm_request")
async def _create_response(self):
if not self._api_session_ready:
self._run_llm_when_api_session_ready = True

9
src/pipecat/services/piper/tts.py
View File

@@ -74,19 +74,18 @@ class PiperTTSService(TTSService):
async with self._session.post(self._base_url, data=text, headers=headers) as response:
if response.status != 200:
eror = await response.text()
error = await response.text()
logger.error(
f"{self} error getting audio (status: {response.status}, error: {eror})"
f"{self} error getting audio (status: {response.status}, error: {error})"
)
yield ErrorFrame(
f"Error getting audio (status: {response.status}, error: {eror})"
f"Error getting audio (status: {response.status}, error: {error})"
)
return
await self.start_tts_usage_metrics(text)
# Process the streaming response
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
yield TTSStartedFrame()
async for chunk in response.content.iter_chunked(CHUNK_SIZE):

14
src/pipecat/services/rime/tts.py
View File

@@ -26,6 +26,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
from pipecat.transcriptions import language
from pipecat.transcriptions.language import Language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
@@ -49,6 +50,8 @@ def language_to_rime_language(language: Language) -> str:
str: Three-letter language code used by Rime (e.g., 'eng' for English).
"""
LANGUAGE_MAP = {
Language.DE: "ger",
Language.FR: "fra",
Language.EN: "eng",
Language.ES: "spa",
}
@@ -352,6 +355,7 @@ class RimeTTSService(AudioContextWordTTSService):
class RimeHttpTTSService(TTSService):
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
pause_between_brackets: Optional[bool] = False
phonemize_between_brackets: Optional[bool] = False
inline_speed_alpha: Optional[str] = None
@@ -377,6 +381,9 @@ class RimeHttpTTSService(TTSService):
self._session = aiohttp_session
self._base_url = "https://users.rime.ai/v1/rime-tts"
self._settings = {
"lang": self.language_to_service_language(params.language)
if params.language
else "eng",
"speedAlpha": params.speed_alpha,
"reduceLatency": params.reduce_latency,
"pauseBetweenBrackets": params.pause_between_brackets,
@@ -391,6 +398,10 @@ class RimeHttpTTSService(TTSService):
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> str | None:
"""Convert pipecat language to Rime language code."""
return language_to_rime_language(language)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -430,8 +441,7 @@ class RimeHttpTTSService(TTSService):
yield TTSStartedFrame()
# Process the streaming response
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
if need_to_strip_wav_header and chunk.startswith(b"RIFF"):

13
src/pipecat/services/tts_service.py
View File

@@ -106,6 +106,19 @@ class TTSService(AIService):
def sample_rate(self) -> int:
return self._sample_rate
@property
def chunk_size(self) -> int:
"""This property indicates how much audio we download (from TTS services
that require chunking) before we start pushing the first audio
frame. This will make sure we download the rest of the audio while audio
is being played without causing audio glitches (specially at the
beginning). Of course, this will also depend on how fast the TTS service
generates bytes.
"""
CHUNK_SECONDS = 0.5
return int(self.sample_rate * CHUNK_SECONDS * 2) # 2 bytes/sample
async def set_model(self, model: str):
self.set_model_name(model)

2
src/pipecat/services/xtts/tts.py
View File

@@ -152,7 +152,7 @@ class XTTSService(TTSService):
yield TTSStartedFrame()
CHUNK_SIZE = 1024
CHUNK_SIZE = self.chunk_size
buffer = bytearray()
async for chunk in r.content.iter_chunked(CHUNK_SIZE):

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

@@ -17,6 +17,8 @@ from pipecat.audio.turn.base_turn_analyzer import (
from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
from pipecat.frames.frames import (
BotInterruptionFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
@@ -51,6 +53,9 @@ class BaseInputTransport(FrameProcessor):
# Input sample rate. It will be initialized on StartFrame.
self._sample_rate = 0
# Track bot speaking state for interruption logic
self._bot_speaking = False
# We read audio from a single queue one at a time and we then run VAD in
# a thread. Therefore, only one thread should be necessary.
self._executor = ThreadPoolExecutor(max_workers=1)
@@ -189,6 +194,12 @@ class BaseInputTransport(FrameProcessor):
await self.push_frame(frame, direction)
elif isinstance(frame, BotInterruptionFrame):
await self._handle_bot_interruption(frame)
elif isinstance(frame, BotStartedSpeakingFrame):
await self._handle_bot_started_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._handle_bot_stopped_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, EmulateUserStartedSpeakingFrame):
logger.debug("Emulating user started speaking")
await self._handle_user_interruption(UserStartedSpeakingFrame(emulated=True))
@@ -230,13 +241,26 @@ class BaseInputTransport(FrameProcessor):
if isinstance(frame, UserStartedSpeakingFrame):
logger.debug("User started speaking")
await self.push_frame(frame)
# Only push StartInterruptionFrame 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
)
# Make sure we notify about interruptions quickly out-of-band.
if self.interruptions_allowed:
if should_push_immediate_interruption and self.interruptions_allowed:
await self._start_interruption()
# Push an out-of-band frame (i.e. not using the ordered push
# frame task) to stop everything, specially at the output
# transport.
await self.push_frame(StartInterruptionFrame())
elif self.interruption_strategies and self._bot_speaking:
logger.debug(
"User started speaking while bot is speaking with interruption config - "
"deferring interruption to aggregator"
)
elif isinstance(frame, UserStoppedSpeakingFrame):
logger.debug("User stopped speaking")
await self.push_frame(frame)
@@ -244,6 +268,16 @@ class BaseInputTransport(FrameProcessor):
await self._stop_interruption()
await self.push_frame(StopInterruptionFrame())
#
# Handle bot speaking state
#
async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame):
self._bot_speaking = True
async def _handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame):
self._bot_speaking = False
#
# Audio input
#

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

@@ -351,6 +351,7 @@ class BaseOutputTransport(FrameProcessor):
sample_rate=self._sample_rate,
num_channels=frame.num_channels,
)
chunk.transport_destination = self._destination
await self._audio_queue.put(chunk)
self._audio_buffer = self._audio_buffer[self._audio_chunk_size :]

4
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -92,7 +92,7 @@ class FastAPIWebsocketClient:
async def trigger_client_connected(self):
await self._callbacks.on_client_connected(self._websocket)
async def trigger_client_timout(self):
async def trigger_client_timeout(self):
await self._callbacks.on_session_timeout(self._websocket)
def _can_send(self):
@@ -188,7 +188,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
async def _monitor_websocket(self):
"""Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
await asyncio.sleep(self._params.session_timeout)
await self._client.trigger_client_timout()
await self._client.trigger_client_timeout()
class FastAPIWebsocketOutputTransport(BaseOutputTransport):

2
src/pipecat/transports/services/livekit.py
View File

@@ -363,8 +363,6 @@ class LiveKitInputTransport(BaseInputTransport):
self._audio_in_task = None
self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer
self._resampler = create_default_resampler()
if self._initialized:
return
# Whether we have seen a StartFrame already.
self._initialized = False

206
src/pipecat/utils/tracing/service_attributes.py
View File

@@ -6,7 +6,7 @@
"""Functions for adding attributes to OpenTelemetry spans."""
from typing import TYPE_CHECKING, Any, Dict, Optional
from typing import TYPE_CHECKING, Any, Dict, List, Optional
# Import for type checking only
if TYPE_CHECKING:
@@ -256,3 +256,207 @@ def add_llm_span_attributes(
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_gemini_live_span_attributes(
span: "Span",
service_name: str,
model: str,
operation_name: str,
voice_id: Optional[str] = None,
language: Optional[str] = None,
modalities: Optional[str] = None,
settings: Optional[Dict[str, Any]] = None,
tools: Optional[List[Dict]] = None,
tools_serialized: Optional[str] = None,
transcript: Optional[str] = None,
is_input: Optional[bool] = None,
text_output: Optional[str] = None,
audio_data_size: Optional[int] = None,
**kwargs,
) -> None:
"""Add Gemini Live specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the service
model: Model name/identifier
operation_name: Name of the operation (setup, model_turn, tool_call, etc.)
voice_id: Voice identifier used for output
language: Language code for the session
modalities: Supported modalities (e.g., "AUDIO", "TEXT")
settings: Service configuration settings
tools: Available tools/functions list
tools_serialized: JSON-serialized tools for detailed inspection
transcript: Transcription text
is_input: Whether transcript is input (True) or output (False)
text_output: Text output from model
audio_data_size: Size of audio data in bytes
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("gen_ai.system", "gcp.gemini")
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("service.operation", operation_name)
# Add optional attributes
if voice_id:
span.set_attribute("voice_id", voice_id)
if language:
span.set_attribute("language", language)
if modalities:
span.set_attribute("modalities", modalities)
if transcript:
span.set_attribute("transcript", transcript)
if is_input is not None:
span.set_attribute("transcript.is_input", is_input)
if text_output:
span.set_attribute("text_output", text_output)
if audio_data_size is not None:
span.set_attribute("audio.data_size_bytes", audio_data_size)
if tools:
span.set_attribute("tools.count", len(tools))
span.set_attribute("tools.available", True)
# Add individual tool names for easier filtering
tool_names = []
for tool in tools:
if isinstance(tool, dict) and "name" in tool:
tool_names.append(tool["name"])
elif hasattr(tool, "name"):
tool_name = getattr(tool, "name", None)
if tool_name is not None:
tool_names.append(tool_name)
if tool_names:
span.set_attribute("tools.names", ",".join(tool_names))
if tools_serialized:
span.set_attribute("tools.definitions", tools_serialized)
# Add settings if provided
if settings:
for key, value in settings.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"settings.{key}", value)
elif key == "vad" and value:
# Handle VAD settings specially
if hasattr(value, "disabled") and value.disabled is not None:
span.set_attribute("settings.vad.disabled", value.disabled)
if hasattr(value, "start_sensitivity") and value.start_sensitivity:
span.set_attribute(
"settings.vad.start_sensitivity", value.start_sensitivity.value
)
if hasattr(value, "end_sensitivity") and value.end_sensitivity:
span.set_attribute("settings.vad.end_sensitivity", value.end_sensitivity.value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_openai_realtime_span_attributes(
span: "Span",
service_name: str,
model: str,
operation_name: str,
session_properties: Optional[Dict[str, Any]] = None,
transcript: Optional[str] = None,
is_input: Optional[bool] = None,
context_messages: Optional[str] = None,
function_calls: Optional[List[Dict]] = None,
tools: Optional[List[Dict]] = None,
tools_serialized: Optional[str] = None,
audio_data_size: Optional[int] = None,
**kwargs,
) -> None:
"""Add OpenAI Realtime specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the service
model: Model name/identifier
operation_name: Name of the operation (setup, transcription, response, etc.)
session_properties: Session configuration properties
transcript: Transcription text
is_input: Whether transcript is input (True) or output (False)
context_messages: JSON-serialized context messages
function_calls: Function calls being made
tools: Available tools/functions list
tools_serialized: JSON-serialized tools for detailed inspection
audio_data_size: Size of audio data in bytes
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("gen_ai.system", "openai")
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("service.operation", operation_name)
# Add optional attributes
if transcript:
span.set_attribute("transcript", transcript)
if is_input is not None:
span.set_attribute("transcript.is_input", is_input)
if context_messages:
span.set_attribute("input", context_messages)
if audio_data_size is not None:
span.set_attribute("audio.data_size_bytes", audio_data_size)
if tools:
span.set_attribute("tools.count", len(tools))
span.set_attribute("tools.available", True)
# Add individual tool names for easier filtering
tool_names = []
for tool in tools:
if isinstance(tool, dict) and "name" in tool:
tool_names.append(tool["name"])
elif hasattr(tool, "name"):
tool_names.append(tool.name)
elif isinstance(tool, dict) and "function" in tool and "name" in tool["function"]:
tool_names.append(tool["function"]["name"])
if tool_names:
span.set_attribute("tools.names", ",".join(tool_names))
if tools_serialized:
span.set_attribute("tools.definitions", tools_serialized)
if function_calls:
span.set_attribute("function_calls.count", len(function_calls))
if function_calls:
call = function_calls[0]
if hasattr(call, "name"):
span.set_attribute("function_calls.first_name", call.name)
elif isinstance(call, dict) and "name" in call:
span.set_attribute("function_calls.first_name", call["name"])
# Add session properties if provided
if session_properties:
for key, value in session_properties.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"session.{key}", value)
elif key == "turn_detection" and value is not None:
if isinstance(value, bool):
span.set_attribute("session.turn_detection.enabled", value)
elif isinstance(value, dict):
span.set_attribute("session.turn_detection.enabled", True)
for td_key, td_value in value.items():
if isinstance(td_value, (str, int, float, bool)):
span.set_attribute(f"session.turn_detection.{td_key}", td_value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)

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

@@ -24,7 +24,9 @@ if TYPE_CHECKING:
from opentelemetry import trace
from pipecat.utils.tracing.service_attributes import (
add_gemini_live_span_attributes,
add_llm_span_attributes,
add_openai_realtime_span_attributes,
add_stt_span_attributes,
add_tts_span_attributes,
)
@@ -477,3 +479,525 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
if func is not None:
return decorator(func)
return decorator
def traced_gemini_live(operation: str) -> Callable:
"""Traces Gemini Live service methods with operation-specific attributes.
This decorator automatically captures relevant information based on the operation type:
- llm_setup: Configuration, tools definitions, and system instructions
- llm_tool_call: Function call information
- llm_tool_result: Function execution results
- llm_response: Complete LLM response with usage and output
Args:
operation: The operation name (matches the event type being handled)
Returns:
Wrapped method with Gemini Live specific tracing.
"""
if not is_tracing_available():
return _noop_decorator
def decorator(func):
@functools.wraps(func)
async def wrapper(self, *args, **kwargs):
try:
if not is_tracing_available():
return await func(self, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# Base service attributes
model_name = getattr(
self, "model_name", getattr(self, "_model_name", "unknown")
)
voice_id = getattr(self, "_voice_id", None)
language_code = getattr(self, "_language_code", None)
settings = getattr(self, "_settings", {})
# Get modalities if available
modalities = None
if hasattr(self, "_settings") and "modalities" in self._settings:
modality_obj = self._settings["modalities"]
if hasattr(modality_obj, "value"):
modalities = modality_obj.value
else:
modalities = str(modality_obj)
# Operation-specific attribute collection
operation_attrs = {}
if operation == "llm_setup":
# Capture detailed tool information
tools = getattr(self, "_tools", None)
if tools:
# Handle different tool formats
tools_list = []
tools_serialized = None
try:
if hasattr(tools, "standard_tools"):
# ToolsSchema object
tools_list = tools.standard_tools
# Serialize the tools for detailed inspection
tools_serialized = json.dumps(
[
{
"name": tool.name
if hasattr(tool, "name")
else tool.get("name", "unknown"),
"description": tool.description
if hasattr(tool, "description")
else tool.get("description", ""),
"properties": tool.properties
if hasattr(tool, "properties")
else tool.get("properties", {}),
"required": tool.required
if hasattr(tool, "required")
else tool.get("required", []),
}
for tool in tools_list
]
)
elif isinstance(tools, list):
# List of tool dictionaries or objects
tools_list = tools
tools_serialized = json.dumps(
[
{
"name": tool.get("name", "unknown")
if isinstance(tool, dict)
else getattr(tool, "name", "unknown"),
"description": tool.get("description", "")
if isinstance(tool, dict)
else getattr(tool, "description", ""),
"properties": tool.get("properties", {})
if isinstance(tool, dict)
else getattr(tool, "properties", {}),
"required": tool.get("required", [])
if isinstance(tool, dict)
else getattr(tool, "required", []),
}
for tool in tools_list
]
)
if tools_list:
operation_attrs["tools"] = tools_list
operation_attrs["tools_serialized"] = tools_serialized
except Exception as e:
logging.warning(f"Error serializing tools for tracing: {e}")
# Fallback to basic tool count
if tools_list:
operation_attrs["tools"] = tools_list
# Capture system instruction information
system_instruction = getattr(self, "_system_instruction", None)
if system_instruction:
operation_attrs["system_instruction"] = system_instruction[
:500
] # Truncate if very long
# Capture context system instructions if available
if hasattr(self, "_context") and self._context:
try:
context_system = self._context.extract_system_instructions()
if context_system:
operation_attrs["context_system_instruction"] = (
context_system[:500]
) # Truncate if very long
except Exception as e:
logging.warning(
f"Error extracting context system instructions: {e}"
)
elif operation == "llm_tool_call" and args:
# Extract tool call information
evt = args[0] if args else None
if evt and hasattr(evt, "toolCall") and evt.toolCall.functionCalls:
function_calls = evt.toolCall.functionCalls
if function_calls:
# Add information about the first function call
call = function_calls[0]
operation_attrs["tool.function_name"] = call.name
operation_attrs["tool.call_id"] = call.id
operation_attrs["tool.calls_count"] = len(function_calls)
# Add all function names being called
all_function_names = [c.name for c in function_calls]
operation_attrs["tool.all_function_names"] = ",".join(
all_function_names
)
# Add arguments for the first call (truncated if too long)
try:
args_str = json.dumps(call.args) if call.args else "{}"
if len(args_str) > 1000:
args_str = args_str[:1000] + "..."
operation_attrs["tool.arguments"] = args_str
except Exception:
operation_attrs["tool.arguments"] = str(call.args)[:1000]
elif operation == "llm_tool_result" and args:
# Extract tool result information
tool_result_message = args[0] if args else None
if tool_result_message and isinstance(tool_result_message, dict):
# Extract the tool call information
tool_call_id = tool_result_message.get("tool_call_id")
tool_call_name = tool_result_message.get("tool_call_name")
result_content = tool_result_message.get("content")
if tool_call_id:
operation_attrs["tool.call_id"] = tool_call_id
if tool_call_name:
operation_attrs["tool.function_name"] = tool_call_name
# Parse and capture the result
if result_content:
try:
result = json.loads(result_content)
# Serialize the result, truncating if too long
result_str = json.dumps(result)
if len(result_str) > 2000: # Larger limit for results
result_str = result_str[:2000] + "..."
operation_attrs["tool.result"] = result_str
# Add result status/success indicator if present
if isinstance(result, dict):
if "error" in result:
operation_attrs["tool.result_status"] = "error"
elif "success" in result:
operation_attrs["tool.result_status"] = "success"
else:
operation_attrs["tool.result_status"] = "completed"
except json.JSONDecodeError as e:
operation_attrs["tool.result"] = (
f"Invalid JSON: {str(result_content)[:500]}"
)
operation_attrs["tool.result_status"] = "parse_error"
except Exception as e:
operation_attrs["tool.result"] = (
f"Error processing result: {str(e)}"
)
operation_attrs["tool.result_status"] = "processing_error"
elif operation == "llm_response" and args:
# Extract usage and response metadata from turn complete event
evt = args[0] if args else None
if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
usage = evt.usageMetadata
# Token usage - basic attributes for span visibility
if hasattr(usage, "promptTokenCount"):
operation_attrs["tokens.prompt"] = usage.promptTokenCount or 0
if hasattr(usage, "responseTokenCount"):
operation_attrs["tokens.completion"] = (
usage.responseTokenCount or 0
)
if hasattr(usage, "totalTokenCount"):
operation_attrs["tokens.total"] = usage.totalTokenCount or 0
# Get output text and modality from service state
text = getattr(self, "_bot_text_buffer", "")
audio_text = getattr(self, "_llm_output_buffer", "")
if text:
# TEXT modality
operation_attrs["output"] = text
operation_attrs["output_modality"] = "TEXT"
elif audio_text:
# AUDIO modality
operation_attrs["output"] = audio_text
operation_attrs["output_modality"] = "AUDIO"
# Add turn completion status
if (
evt
and hasattr(evt, "serverContent")
and evt.serverContent.turnComplete
):
operation_attrs["turn_complete"] = True
# Add all attributes to the span
add_gemini_live_span_attributes(
span=current_span,
service_name=service_class_name,
model=model_name,
operation_name=operation,
voice_id=voice_id,
language=language_code,
modalities=modalities,
settings=settings,
**operation_attrs,
)
# For llm_response operation, also handle token usage metrics
if operation == "llm_response" and hasattr(self, "start_llm_usage_metrics"):
evt = args[0] if args else None
if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
usage = evt.usageMetadata
# Create LLMTokenUsage object
from pipecat.metrics.metrics import LLMTokenUsage
tokens = LLMTokenUsage(
prompt_tokens=usage.promptTokenCount or 0,
completion_tokens=usage.responseTokenCount or 0,
total_tokens=usage.totalTokenCount or 0,
)
_add_token_usage_to_span(current_span, tokens)
# Capture TTFB metric if available
ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None)
if ttfb is not None:
current_span.set_attribute("metrics.ttfb", ttfb)
# Run the original function
result = await func(self, *args, **kwargs)
return result
except Exception as e:
current_span.record_exception(e)
current_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
raise
except Exception as e:
logging.error(f"Error in Gemini Live tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await func(self, *args, **kwargs)
return wrapper
return decorator
def traced_openai_realtime(operation: str) -> Callable:
"""Traces OpenAI Realtime service methods with operation-specific attributes.
This decorator automatically captures relevant information based on the operation type:
- llm_setup: Session configuration and tools
- llm_request: Context and input messages
- llm_response: Usage metadata, output, and function calls
Args:
operation: The operation name (matches the event type being handled)
Returns:
Wrapped method with OpenAI Realtime specific tracing.
"""
if not is_tracing_available():
return _noop_decorator
def decorator(func):
@functools.wraps(func)
async def wrapper(self, *args, **kwargs):
try:
if not is_tracing_available():
return await func(self, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# Base service attributes
model_name = getattr(
self, "model_name", getattr(self, "_model_name", "unknown")
)
# Operation-specific attribute collection
operation_attrs = {}
if operation == "llm_setup":
# Capture session properties and tools
session_properties = getattr(self, "_session_properties", None)
if session_properties:
try:
# Convert to dict for easier processing
if hasattr(session_properties, "model_dump"):
props_dict = session_properties.model_dump()
elif hasattr(session_properties, "__dict__"):
props_dict = session_properties.__dict__
else:
props_dict = {}
operation_attrs["session_properties"] = props_dict
# Extract tools if available
tools = props_dict.get("tools")
if tools:
operation_attrs["tools"] = tools
try:
operation_attrs["tools_serialized"] = json.dumps(tools)
except Exception as e:
logging.warning(f"Error serializing OpenAI tools: {e}")
# Extract instructions
instructions = props_dict.get("instructions")
if instructions:
operation_attrs["instructions"] = instructions[:500]
except Exception as e:
logging.warning(f"Error processing session properties: {e}")
# Also check context for tools
if hasattr(self, "_context") and self._context:
try:
context_tools = getattr(self._context, "tools", None)
if context_tools and not operation_attrs.get("tools"):
operation_attrs["tools"] = context_tools
operation_attrs["tools_serialized"] = json.dumps(
context_tools
)
except Exception as e:
logging.warning(f"Error extracting context tools: {e}")
elif operation == "llm_request":
# Capture context messages being sent
if hasattr(self, "_context") and self._context:
try:
messages = self._context.get_messages_for_logging()
if messages:
operation_attrs["context_messages"] = json.dumps(messages)
except Exception as e:
logging.warning(f"Error getting context messages: {e}")
elif operation == "llm_response" and args:
# Extract usage and response metadata
evt = args[0] if args else None
if evt and hasattr(evt, "response"):
response = evt.response
# Token usage - basic attributes for span visibility
if hasattr(response, "usage"):
usage = response.usage
if hasattr(usage, "input_tokens"):
operation_attrs["tokens.prompt"] = usage.input_tokens
if hasattr(usage, "output_tokens"):
operation_attrs["tokens.completion"] = usage.output_tokens
if hasattr(usage, "total_tokens"):
operation_attrs["tokens.total"] = usage.total_tokens
# Response status and metadata
if hasattr(response, "status"):
operation_attrs["response.status"] = response.status
if hasattr(response, "id"):
operation_attrs["response.id"] = response.id
# Output items and extract transcript for output field
if hasattr(response, "output") and response.output:
operation_attrs["response.output_items"] = len(response.output)
# Extract assistant transcript and function calls
assistant_transcript = ""
function_calls = []
for item in response.output:
if (
hasattr(item, "content")
and item.content
and hasattr(item, "role")
and item.role == "assistant"
):
for content in item.content:
if (
hasattr(content, "transcript")
and content.transcript
):
assistant_transcript += content.transcript + " "
elif hasattr(item, "type") and item.type == "function_call":
function_call_info = {
"name": getattr(item, "name", "unknown"),
"call_id": getattr(item, "call_id", "unknown"),
}
if hasattr(item, "arguments"):
args_str = item.arguments
if len(args_str) > 500:
args_str = args_str[:500] + "..."
function_call_info["arguments"] = args_str
function_calls.append(function_call_info)
if assistant_transcript.strip():
operation_attrs["output"] = assistant_transcript.strip()
if function_calls:
operation_attrs["function_calls"] = function_calls
operation_attrs["function_calls.count"] = len(
function_calls
)
all_names = [call["name"] for call in function_calls]
operation_attrs["function_calls.all_names"] = ",".join(
all_names
)
# Add all attributes to the span
add_openai_realtime_span_attributes(
span=current_span,
service_name=service_class_name,
model=model_name,
operation_name=operation,
**operation_attrs,
)
# For llm_response operation, also handle token usage metrics
if operation == "llm_response" and hasattr(self, "start_llm_usage_metrics"):
evt = args[0] if args else None
if evt and hasattr(evt, "response") and hasattr(evt.response, "usage"):
usage = evt.response.usage
# Create LLMTokenUsage object
from pipecat.metrics.metrics import LLMTokenUsage
tokens = LLMTokenUsage(
prompt_tokens=getattr(usage, "input_tokens", 0),
completion_tokens=getattr(usage, "output_tokens", 0),
total_tokens=getattr(usage, "total_tokens", 0),
)
_add_token_usage_to_span(current_span, tokens)
# Capture TTFB metric if available
ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None)
if ttfb is not None:
current_span.set_attribute("metrics.ttfb", ttfb)
# Run the original function
result = await func(self, *args, **kwargs)
return result
except Exception as e:
current_span.record_exception(e)
current_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
raise
except Exception as e:
logging.error(f"Error in OpenAI Realtime tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await func(self, *args, **kwargs)
return wrapper
return decorator

10
src/pipecat/utils/tracing/turn_trace_observer.py
View File

@@ -35,7 +35,11 @@ class TurnTraceObserver(BaseObserver):
"""
def __init__(
self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None, **kwargs
self,
turn_tracker: TurnTrackingObserver,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
**kwargs,
):
super().__init__(**kwargs)
self._turn_tracker = turn_tracker
@@ -47,6 +51,7 @@ class TurnTraceObserver(BaseObserver):
# Conversation tracking properties
self._conversation_span: Optional["Span"] = None
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
if turn_tracker:
@@ -89,6 +94,9 @@ class TurnTraceObserver(BaseObserver):
# Set span attributes
self._conversation_span.set_attribute("conversation.id", conversation_id)
self._conversation_span.set_attribute("conversation.type", "voice")
# Set custom otel attributes if provided
for k, v in (self._additional_span_attributes or {}).items():
self._conversation_span.set_attribute(k, v)
# Update the conversation context provider
context_provider.set_current_conversation_context(