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

Merge branch 'pipecat-ai:main' into main

Browse Source
This commit is contained in:
Jessie Wei
2025-09-24 10:23:52 +10:00
committed by GitHub
parent 5b4655f45a d7c8f8df53
commit 88614b312f
148 changed files with 2278 additions and 1630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/pipecat/observers/loggers/llm_log_observer.py
View File

@@ -11,6 +11,7 @@ from loguru import logger
from pipecat.frames.frames import (
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
@@ -79,10 +80,13 @@ class LLMLogObserver(BaseObserver):
f"🧠 {arrow} {dst} LLM MESSAGES FRAME: {frame.messages} at {time_sec:.2f}s"
)
# Log OpenAILLMContextFrame (input)
elif isinstance(frame, OpenAILLMContextFrame):
logger.debug(
f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {frame.context.messages} at {time_sec:.2f}s"
elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
messages = (
frame.context.messages
if isinstance(frame, OpenAILLMContextFrame)
else frame.context.get_messages()
)
logger.debug(f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {messages} at {time_sec:.2f}s")
# Log function call result (input)
elif isinstance(frame, FunctionCallResultFrame):
logger.debug(

32
src/pipecat/observers/loggers/user_bot_latency_log_observer.py
View File

@@ -61,17 +61,29 @@ class UserBotLatencyLogObserver(BaseObserver):
elif isinstance(data.frame, UserStoppedSpeakingFrame):
self._user_stopped_time = time.time()
elif isinstance(data.frame, (EndFrame, CancelFrame)):
if self._latencies:
avg_latency = mean(self._latencies)
min_latency = min(self._latencies)
max_latency = max(self._latencies)
logger.info(
f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING - Avg: {avg_latency:.3f}s, Min: {min_latency:.3f}s, Max: {max_latency:.3f}s"
)
self._log_summary()
elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time:
latency = time.time() - self._user_stopped_time
self._user_stopped_time = 0
self._latencies.append(latency)
logger.debug(
f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency:.3f}s"
)
self._log_latency(latency)
def _log_summary(self):
if not self._latencies:
return
avg_latency = mean(self._latencies)
min_latency = min(self._latencies)
max_latency = max(self._latencies)
logger.info(
f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING - Avg: {avg_latency:.3f}s, Min: {min_latency:.3f}s, Max: {max_latency:.3f}s"
)
def _log_latency(self, latency: float):
"""Log the latency.
Args:
latency: The latency to log.
"""
logger.debug(
f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency:.3f}s"
)

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

@@ -106,13 +106,21 @@ class PipelineRunner(BaseObject):
def _setup_sigint(self):
"""Set up signal handlers for graceful shutdown."""
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGINT, lambda *args: self._sig_handler())
try:
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGINT, lambda *args: self._sig_handler())
except NotImplementedError:
# Windows fallback
signal.signal(signal.SIGINT, lambda s, f: self._sig_handler())
def _setup_sigterm(self):
"""Set up signal handlers for graceful shutdown."""
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, lambda *args: self._sig_handler())
try:
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, lambda *args: self._sig_handler())
except NotImplementedError:
# Windows fallback
signal.signal(signal.SIGTERM, lambda s, f: self._sig_handler())
def _sig_handler(self):
"""Handle interrupt signals by cancelling all tasks."""

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

@@ -777,7 +777,6 @@ class PipelineTask(BasePipelineTask):
"""
running = True
last_frame_time = 0
frame_buffer = deque(maxlen=10) # Store last 10 frames
while running:
try:
@@ -785,9 +784,6 @@ class PipelineTask(BasePipelineTask):
self._idle_queue.get(), timeout=self._idle_timeout_secs
)
if not isinstance(frame, InputAudioRawFrame):
frame_buffer.append(frame)
if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames):
# If we find a StartFrame or one of the frames that prevents a
# time out we update the time.
@@ -798,7 +794,7 @@ class PipelineTask(BasePipelineTask):
# valid frames.
diff_time = time.time() - last_frame_time
if diff_time >= self._idle_timeout_secs:
running = await self._idle_timeout_detected(frame_buffer)
running = await self._idle_timeout_detected()
# Reset `last_frame_time` so we don't trigger another
# immediate idle timeout if we are not cancelling. For
# example, we might want to force the bot to say goodbye
@@ -808,14 +804,11 @@ class PipelineTask(BasePipelineTask):
self._idle_queue.task_done()
except asyncio.TimeoutError:
running = await self._idle_timeout_detected(frame_buffer)
running = await self._idle_timeout_detected()
async def _idle_timeout_detected(self, last_frames: Deque[Frame]) -> bool:
async def _idle_timeout_detected(self) -> bool:
"""Handle idle timeout detection and optional cancellation.
Args:
last_frames: Recent frames received before timeout for debugging.
Returns:
Whether the pipeline task should continue running.
"""
@@ -823,10 +816,7 @@ class PipelineTask(BasePipelineTask):
if self._cancelled:
return True
logger.warning("Idle timeout detected. Last 10 frames received:")
for i, frame in enumerate(last_frames, 1):
logger.warning(f"Frame {i}: {frame}")
logger.warning("Idle timeout detected.")
await self._call_event_handler("on_idle_timeout")
if self._cancel_on_idle_timeout:
logger.warning(f"Idle pipeline detected, cancelling pipeline task...")

18
src/pipecat/processors/aggregators/gated_openai_llm_context.py → src/pipecat/processors/aggregators/gated_llm_context.py
View File

@@ -4,20 +4,20 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Gated OpenAI LLM context aggregator for controlled message flow."""
"""Gated LLM context aggregator for controlled message flow."""
from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame
from pipecat.frames.frames import CancelFrame, EndFrame, Frame, LLMContextFrame, StartFrame
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.sync.base_notifier import BaseNotifier
class GatedOpenAILLMContextAggregator(FrameProcessor):
"""Aggregator that gates OpenAI LLM context frames until notified.
class GatedLLMContextAggregator(FrameProcessor):
"""Aggregator that gates LLM context frames until notified.
This aggregator captures OpenAI LLM context frames and holds them until
a notifier signals that they can be released. This is useful for controlling
the flow of context frames based on external conditions or timing.
This aggregator captures LLM context frames and holds them until a notifier
signals that they can be released. This is useful for controlling the flow
of context frames based on external conditions or timing.
"""
def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
@@ -35,7 +35,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
self._gate_task = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames, gating OpenAI LLM context frames.
"""Process incoming frames, gating LLM context frames.
Args:
frame: The frame to process.
@@ -49,7 +49,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
if isinstance(frame, (EndFrame, CancelFrame)):
await self._stop()
await self.push_frame(frame)
elif isinstance(frame, OpenAILLMContextFrame):
elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
if self._start_open:
self._start_open = False
await self.push_frame(frame, direction)

12
src/pipecat/processors/aggregators/gated_open_ai_llm_context.py Normal file
View File

@@ -0,0 +1,12 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Gated OpenAI LLM context aggregator for controlled message flow."""
from pipecat.processors.aggregators.gated_llm_context import GatedLLMContextAggregator
# Alias for backward compatibility with the previous name
GatedOpenAILLMContextAggregator = GatedLLMContextAggregator

45
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -229,9 +229,12 @@ class AudioBufferProcessor(FrameProcessor):
# Save time of frame so we can compute silence.
self._last_bot_frame_at = time.time()
if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
if self._buffer_size > 0 and (
len(self._user_audio_buffer) >= self._buffer_size
or len(self._bot_audio_buffer) >= self._buffer_size
):
await self._call_on_audio_data_handler()
self._reset_recording()
self._reset_primary_audio_buffers()
# Process turn recording with preprocessed data.
if self._enable_turn_audio:
@@ -272,9 +275,15 @@ class AudioBufferProcessor(FrameProcessor):
async def _call_on_audio_data_handler(self):
"""Call the audio data event handlers with buffered audio."""
if not self.has_audio() or not self._recording:
if not self._recording:
return
if len(self._user_audio_buffer) == 0 and len(self._bot_audio_buffer) == 0:
return
self._align_track_buffers()
flush_time = time.time()
# Call original handler with merged audio
merged_audio = self.merge_audio_buffers()
await self._call_event_handler(
@@ -290,23 +299,49 @@ class AudioBufferProcessor(FrameProcessor):
self._num_channels,
)
self._last_user_frame_at = flush_time
self._last_bot_frame_at = flush_time
def _buffer_has_audio(self, buffer: bytearray) -> bool:
"""Check if a buffer contains audio data."""
return buffer is not None and len(buffer) > 0
def _reset_recording(self):
"""Reset recording state and buffers."""
self._reset_audio_buffers()
self._reset_all_audio_buffers()
self._last_user_frame_at = time.time()
self._last_bot_frame_at = time.time()
def _reset_audio_buffers(self):
def _reset_all_audio_buffers(self):
"""Reset all audio buffers to empty state."""
self._reset_primary_audio_buffers()
self._reset_turn_audio_buffers()
def _reset_primary_audio_buffers(self):
"""Clear user and bot buffers while preserving turn buffers and timestamps."""
self._user_audio_buffer = bytearray()
self._bot_audio_buffer = bytearray()
def _reset_turn_audio_buffers(self):
"""Clear user and bot turn buffers while preserving primary buffers and timestamps."""
self._user_turn_audio_buffer = bytearray()
self._bot_turn_audio_buffer = bytearray()
def _align_track_buffers(self):
"""Pad the shorter track with silence so both tracks stay in sync."""
user_len = len(self._user_audio_buffer)
bot_len = len(self._bot_audio_buffer)
if user_len == bot_len:
return
target_len = max(user_len, bot_len)
if user_len < target_len:
self._user_audio_buffer.extend(b"\x00" * (target_len - user_len))
self._last_user_frame_at = max(self._last_user_frame_at, self._last_bot_frame_at)
if bot_len < target_len:
self._bot_audio_buffer.extend(b"\x00" * (target_len - bot_len))
self._last_bot_frame_at = max(self._last_bot_frame_at, self._last_user_frame_at)
async def _resample_input_audio(self, frame: InputAudioRawFrame) -> bytes:
"""Resample audio frame to the target sample rate."""
return await self._input_resampler.resample(

10
src/pipecat/processors/frameworks/langchain.py
View File

@@ -12,6 +12,7 @@ from loguru import logger
from pipecat.frames.frames import (
Frame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
TextFrame,
@@ -64,11 +65,16 @@ class LangchainProcessor(FrameProcessor):
"""
await super().process_frame(frame, direction)
if isinstance(frame, OpenAILLMContextFrame):
if isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
# Messages are accumulated on the context as a list of messages.
# The last one by the human is the one we want to send to the LLM.
logger.debug(f"Got transcription frame {frame}")
text: str = frame.context.messages[-1]["content"]
messages = (
frame.context.messages
if isinstance(frame, OpenAILLMContextFrame)
else frame.context.get_messages()
)
text: str = messages[-1]["content"]
await self._ainvoke(text.strip())
else:

169
src/pipecat/processors/frameworks/strands_agents.py Normal file
View File

@@ -0,0 +1,169 @@
"""Strands Agent integration for Pipecat.
This module provides integration with Strands Agents for handling conversational AI
interactions. It supports both single agent and multi-agent graphs.
"""
from typing import Optional
from loguru import logger
from pipecat.frames.frames import (
Frame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMTextFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
try:
from strands import Agent
from strands.multiagent.graph import Graph
except ModuleNotFoundError as e:
logger.exception("In order to use Strands Agents, you need to `pip install strands-agents`.")
raise Exception(f"Missing module: {e}")
class StrandsAgentsProcessor(FrameProcessor):
"""Processor that integrates Strands Agents with Pipecat's frame pipeline.
This processor takes LLM message frames, extracts the latest user message,
and processes it through either a single Strands Agent or a multi-agent Graph.
The response is streamed back as text frames with appropriate response markers.
Supports both single agent streaming and graph-based multi-agent workflows.
"""
def __init__(
self,
agent: Optional[Agent] = None,
graph: Optional[Graph] = None,
graph_exit_node: Optional[str] = None,
):
"""Initialize the Strands Agents processor.
Args:
agent: The Strands Agent to use for single-agent processing.
graph: The Strands multi-agent Graph to use for graph-based processing.
graph_exit_node: The exit node name when using graph-based processing.
Raises:
AssertionError: If neither agent nor graph is provided, or if graph is
provided without a graph_exit_node.
"""
super().__init__()
self.agent = agent
self.graph = graph
self.graph_exit_node = graph_exit_node
assert self.agent or self.graph, "Either agent or graph must be provided"
if self.graph:
assert self.graph_exit_node, "graph_exit_node must be provided if graph is provided"
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and handle LLM message frames.
Args:
frame: The incoming frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, LLMContextFrame):
messages = frame.context.get_messages()
if messages:
last_message = messages[-1]
await self._ainvoke(str(last_message["content"]).strip())
else:
await self.push_frame(frame, direction)
async def _ainvoke(self, text: str):
"""Invoke the Strands agent with the provided text and stream results as Pipecat frames.
Args:
text: The user input text to process through the agent or graph.
"""
logger.debug(f"Invoking Strands agent with: {text}")
ttfb_tracking = True
try:
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
await self.start_ttfb_metrics()
if self.graph:
# Graph does not stream; await full result then emit assistant text
graph_result = await self.graph.invoke_async(text)
if ttfb_tracking:
await self.stop_ttfb_metrics()
ttfb_tracking = False
try:
node_result = graph_result.results[self.graph_exit_node]
logger.debug(f"Node result: {node_result}")
for agent_result in node_result.get_agent_results():
# Push to TTS service
message = getattr(agent_result, "message", None)
if isinstance(message, dict) and "content" in message:
for block in message["content"]:
if isinstance(block, dict) and "text" in block:
await self.push_frame(LLMTextFrame(str(block["text"])))
# Update usage metrics
await self._report_usage_metrics(
agent_result.metrics.accumulated_usage.get("inputTokens", 0),
agent_result.metrics.accumulated_usage.get("outputTokens", 0),
agent_result.metrics.accumulated_usage.get("totalTokens", 0),
)
except Exception as parse_err:
logger.warning(f"Failed to extract messages from GraphResult: {parse_err}")
else:
# Agent supports streaming events via async iterator
async for event in self.agent.stream_async(text):
# Push to TTS service
if isinstance(event, dict) and "data" in event:
await self.push_frame(LLMTextFrame(str(event["data"])))
if ttfb_tracking:
await self.stop_ttfb_metrics()
ttfb_tracking = False
# Update usage metrics
if (
isinstance(event, dict)
and "event" in event
and "metadata" in event["event"]
):
if "usage" in event["event"]["metadata"]:
usage = event["event"]["metadata"]["usage"]
await self._report_usage_metrics(
usage.get("inputTokens", 0),
usage.get("outputTokens", 0),
usage.get("totalTokens", 0),
)
except GeneratorExit:
logger.warning(f"{self} generator was closed prematurely")
except Exception as e:
logger.exception(f"{self} an unknown error occurred: {e}")
finally:
if ttfb_tracking:
await self.stop_ttfb_metrics()
ttfb_tracking = False
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
def can_generate_metrics(self) -> bool:
"""Check if this service can generate performance metrics.
Returns:
True as this service supports metrics generation.
"""
return True
async def _report_usage_metrics(
self, prompt_tokens: int, completion_tokens: int, total_tokens: int
):
tokens = LLMTokenUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
)
await self.start_llm_usage_metrics(tokens)

339
src/pipecat/services/elevenlabs/stt.py Normal file
View File

@@ -0,0 +1,339 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""ElevenLabs speech-to-text service implementation.
This module provides integration with ElevenLabs' Speech-to-Text API for transcription
using segmented audio processing. The service uploads audio files and receives
transcription results directly.
"""
import io
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_elevenlabs_language(language: Language) -> Optional[str]:
"""Convert a Language enum to ElevenLabs language code.
Source:
https://elevenlabs.io/docs/capabilities/speech-to-text
Args:
language: The Language enum value to convert.
Returns:
The corresponding ElevenLabs language code, or None if not supported.
"""
BASE_LANGUAGES = {
Language.AF: "afr", # Afrikaans
Language.AM: "amh", # Amharic
Language.AR: "ara", # Arabic
Language.HY: "hye", # Armenian
Language.AS: "asm", # Assamese
Language.AST: "ast", # Asturian
Language.AZ: "aze", # Azerbaijani
Language.BE: "bel", # Belarusian
Language.BN: "ben", # Bengali
Language.BS: "bos", # Bosnian
Language.BG: "bul", # Bulgarian
Language.MY: "mya", # Burmese
Language.YUE: "yue", # Cantonese
Language.CA: "cat", # Catalan
Language.CEB: "ceb", # Cebuano
Language.NY: "nya", # Chichewa
Language.HR: "hrv", # Croatian
Language.CS: "ces", # Czech
Language.DA: "dan", # Danish
Language.NL: "nld", # Dutch
Language.EN: "eng", # English
Language.ET: "est", # Estonian
Language.FIL: "fil", # Filipino
Language.FI: "fin", # Finnish
Language.FR: "fra", # French
Language.FF: "ful", # Fulah
Language.GL: "glg", # Galician
Language.LG: "lug", # Ganda
Language.KA: "kat", # Georgian
Language.DE: "deu", # German
Language.EL: "ell", # Greek
Language.GU: "guj", # Gujarati
Language.HA: "hau", # Hausa
Language.HE: "heb", # Hebrew
Language.HI: "hin", # Hindi
Language.HU: "hun", # Hungarian
Language.IS: "isl", # Icelandic
Language.IG: "ibo", # Igbo
Language.ID: "ind", # Indonesian
Language.GA: "gle", # Irish
Language.IT: "ita", # Italian
Language.JA: "jpn", # Japanese
Language.JV: "jav", # Javanese
Language.KEA: "kea", # Kabuverdianu
Language.KN: "kan", # Kannada
Language.KK: "kaz", # Kazakh
Language.KM: "khm", # Khmer
Language.KO: "kor", # Korean
Language.KU: "kur", # Kurdish
Language.KY: "kir", # Kyrgyz
Language.LO: "lao", # Lao
Language.LV: "lav", # Latvian
Language.LN: "lin", # Lingala
Language.LT: "lit", # Lithuanian
Language.LUO: "luo", # Luo
Language.LB: "ltz", # Luxembourgish
Language.MK: "mkd", # Macedonian
Language.MS: "msa", # Malay
Language.ML: "mal", # Malayalam
Language.MT: "mlt", # Maltese
Language.ZH: "zho", # Mandarin Chinese
Language.MI: "mri", # Māori
Language.MR: "mar", # Marathi
Language.MN: "mon", # Mongolian
Language.NE: "nep", # Nepali
Language.NSO: "nso", # Northern Sotho
Language.NO: "nor", # Norwegian
Language.OC: "oci", # Occitan
Language.OR: "ori", # Odia
Language.PS: "pus", # Pashto
Language.FA: "fas", # Persian
Language.PL: "pol", # Polish
Language.PT: "por", # Portuguese
Language.PA: "pan", # Punjabi
Language.RO: "ron", # Romanian
Language.RU: "rus", # Russian
Language.SR: "srp", # Serbian
Language.SN: "sna", # Shona
Language.SD: "snd", # Sindhi
Language.SK: "slk", # Slovak
Language.SL: "slv", # Slovenian
Language.SO: "som", # Somali
Language.ES: "spa", # Spanish
Language.SW: "swa", # Swahili
Language.SV: "swe", # Swedish
Language.TA: "tam", # Tamil
Language.TG: "tgk", # Tajik
Language.TE: "tel", # Telugu
Language.TH: "tha", # Thai
Language.TR: "tur", # Turkish
Language.UK: "ukr", # Ukrainian
Language.UMB: "umb", # Umbundu
Language.UR: "urd", # Urdu
Language.UZ: "uzb", # Uzbek
Language.VI: "vie", # Vietnamese
Language.CY: "cym", # Welsh
Language.WO: "wol", # Wolof
Language.XH: "xho", # Xhosa
Language.ZU: "zul", # Zulu
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
class ElevenLabsSTTService(SegmentedSTTService):
"""Speech-to-text service using ElevenLabs' file-based API.
This service uses ElevenLabs' Speech-to-Text API to perform transcription on audio
segments. It inherits from SegmentedSTTService to handle audio buffering and speech detection.
The service uploads audio files to ElevenLabs and receives transcription results directly.
"""
class InputParams(BaseModel):
"""Configuration parameters for ElevenLabs STT API.
Parameters:
language: Target language for transcription.
tag_audio_events: Whether to include audio events like (laughter), (coughing), in the transcription.
"""
language: Optional[Language] = None
tag_audio_events: bool = True
def __init__(
self,
*,
api_key: str,
aiohttp_session: aiohttp.ClientSession,
base_url: str = "https://api.elevenlabs.io",
model: str = "scribe_v1",
sample_rate: Optional[int] = None,
params: Optional[InputParams] = None,
**kwargs,
):
"""Initialize the ElevenLabs STT service.
Args:
api_key: ElevenLabs API key for authentication.
aiohttp_session: aiohttp ClientSession for HTTP requests.
base_url: Base URL for ElevenLabs API.
model: Model ID for transcription. Defaults to "scribe_v1".
sample_rate: Audio sample rate in Hz. If not provided, uses the pipeline's rate.
params: Configuration parameters for the STT service.
**kwargs: Additional arguments passed to SegmentedSTTService.
"""
super().__init__(
sample_rate=sample_rate,
**kwargs,
)
params = params or ElevenLabsSTTService.InputParams()
self._api_key = api_key
self._base_url = base_url
self._session = aiohttp_session
self._model_id = model
self._tag_audio_events = params.tag_audio_events
self._settings = {
"language": self.language_to_service_language(params.language)
if params.language
else "eng",
}
def can_generate_metrics(self) -> bool:
"""Check if the service can generate processing metrics.
Returns:
True, as ElevenLabs STT service supports metrics generation.
"""
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert a Language enum to ElevenLabs service-specific language code.
Args:
language: The language to convert.
Returns:
The ElevenLabs-specific language code, or None if not supported.
"""
return language_to_elevenlabs_language(language)
async def set_language(self, language: Language):
"""Set the transcription language.
Args:
language: The language to use for speech-to-text transcription.
"""
logger.info(f"Switching STT language to: [{language}]")
self._settings["language"] = self.language_to_service_language(language)
async def set_model(self, model: str):
"""Set the STT model.
Args:
model: The model name to use for transcription.
Note:
ElevenLabs STT API does not currently support model selection.
This method is provided for interface compatibility.
"""
await super().set_model(model)
logger.info(f"Model setting [{model}] noted, but ElevenLabs STT uses default model")
async def _transcribe_audio(self, audio_data: bytes) -> dict:
"""Upload audio data to ElevenLabs and get transcription result.
Args:
audio_data: Raw audio bytes in WAV format.
Returns:
The transcription result data.
Raises:
Exception: If transcription fails or returns an error.
"""
url = f"{self._base_url}/v1/speech-to-text"
headers = {"xi-api-key": self._api_key}
# Create form data with the audio file
data = aiohttp.FormData()
data.add_field(
"file",
io.BytesIO(audio_data),
filename="audio.wav",
content_type="audio/x-wav",
)
# Add required model_id, language_code, and tag_audio_events
data.add_field("model_id", self._model_id)
data.add_field("language_code", self._settings["language"])
data.add_field("tag_audio_events", str(self._tag_audio_events).lower())
async with self._session.post(url, data=data, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"ElevenLabs transcription error: {error_text}")
raise Exception(f"Transcription failed with status {response.status}: {error_text}")
result = await response.json()
return result
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[str] = None
):
"""Handle a transcription result with tracing."""
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Transcribe an audio segment using ElevenLabs' STT API.
Args:
audio: Raw audio bytes in WAV format (already converted by base class).
Yields:
Frame: TranscriptionFrame containing the transcribed text, or ErrorFrame on failure.
Note:
The audio is already in WAV format from the SegmentedSTTService.
Only non-empty transcriptions are yielded.
"""
try:
await self.start_processing_metrics()
await self.start_ttfb_metrics()
# Upload audio and get transcription result directly
result = await self._transcribe_audio(audio)
# Extract transcription text
text = result.get("text", "").strip()
if text:
# Use the language_code returned by the API
detected_language = result.get("language_code", "eng")
await self._handle_transcription(text, True, detected_language)
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(
text,
self._user_id,
time_now_iso8601(),
detected_language,
result=result,
)
except Exception as e:
logger.error(f"ElevenLabs STT error: {e}")
yield ErrorFrame(f"ElevenLabs STT error: {str(e)}")

11
src/pipecat/services/mem0/memory.py
View File

@@ -16,7 +16,8 @@ from typing import Any, Dict, List, Optional
from loguru import logger
from pydantic import BaseModel, Field
from pipecat.frames.frames import ErrorFrame, Frame, LLMMessagesFrame
from pipecat.frames.frames import ErrorFrame, Frame, LLMContextFrame, LLMMessagesFrame
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -180,11 +181,11 @@ class Mem0MemoryService(FrameProcessor):
logger.error(f"Error retrieving memories from Mem0: {e}")
return []
def _enhance_context_with_memories(self, context: OpenAILLMContext, query: str):
def _enhance_context_with_memories(self, context: LLMContext | OpenAILLMContext, query: str):
"""Enhance the LLM context with relevant memories.
Args:
context: The OpenAILLMContext to enhance with memory information.
context: The LLM context to enhance with memory information.
query: The query to search for relevant memories.
"""
# Skip if this is the same query we just processed
@@ -222,11 +223,11 @@ class Mem0MemoryService(FrameProcessor):
context = None
messages = None
if isinstance(frame, OpenAILLMContextFrame):
if isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
messages = frame.messages
context = OpenAILLMContext.from_messages(messages)
context = LLMContext(messages)
if context:
try:

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

@@ -281,8 +281,10 @@ class BaseOpenAILLMService(LLMService):
# base64 encode any images
for message in messages:
if message.get("mime_type") == "image/jpeg":
encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
text = message["content"]
# Avoid .getvalue() which makes a full copy of BytesIO
raw_bytes = message["data"].read()
encoded_image = base64.b64encode(raw_bytes).decode("utf-8")
text = message.get("content", "")
message["content"] = [
{"type": "text", "text": text},
{
@@ -290,6 +292,7 @@ class BaseOpenAILLMService(LLMService):
"image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"},
},
]
# Explicit cleanup
del message["data"]
del message["mime_type"]

42
src/pipecat/transcriptions/language.py
View File

@@ -68,6 +68,9 @@ class Language(StrEnum):
AS = "as"
AS_IN = "as-IN"
# Asturian
AST = "ast"
# Azerbaijani
AZ = "az"
AZ_AZ = "az-AZ"
@@ -101,6 +104,9 @@ class Language(StrEnum):
CA = "ca"
CA_ES = "ca-ES"
# Cebuano
CEB = "ceb"
# Mandarin Chinese
CMN = "cmn"
CMN_CN = "cmn-CN"
@@ -185,6 +191,9 @@ class Language(StrEnum):
FA = "fa"
FA_IR = "fa-IR"
# Fulah
FF = "ff"
# Finnish
FI = "fi"
FI_FI = "fi-FI"
@@ -251,6 +260,9 @@ class Language(StrEnum):
ID = "id"
ID_ID = "id-ID"
# Igbo
IG = "ig"
# Icelandic
IS = "is"
IS_IS = "is-IS"
@@ -279,6 +291,9 @@ class Language(StrEnum):
KA = "ka"
KA_GE = "ka-GE"
# Kabuverdianu
KEA = "kea"
# Kazakh
KK = "kk"
KK_KZ = "kk-KZ"
@@ -295,6 +310,13 @@ class Language(StrEnum):
KO = "ko"
KO_KR = "ko-KR"
# Kurdish
KU = "ku"
# Kyrgyz
KY = "ky"
KY_KG = "ky-KG"
# Latin
LA = "la"
@@ -312,6 +334,12 @@ class Language(StrEnum):
LT = "lt"
LT_LT = "lt-LT"
# Ganda
LG = "lg"
# Luo
LUO = "luo"
# Latvian
LV = "lv"
LV_LV = "lv-LV"
@@ -366,6 +394,12 @@ class Language(StrEnum):
NL_BE = "nl-BE"
NL_NL = "nl-NL"
# Northern Sotho
NSO = "nso"
# Chichewa
NY = "ny"
# Occitan
OC = "oc"
@@ -484,6 +518,9 @@ class Language(StrEnum):
UK = "uk"
UK_UA = "uk-UA"
# Umbundu
UMB = "umb"
# Urdu
UR = "ur"
UR_IN = "ur-IN"
@@ -497,6 +534,9 @@ class Language(StrEnum):
VI = "vi"
VI_VN = "vi-VN"
# Wolof
WO = "wo"
# Wu Chinese
WUU = "wuu"
WUU_CN = "wuu-CN"
@@ -507,7 +547,7 @@ class Language(StrEnum):
# Yoruba
YO = "yo"
# Yue Chinese
# Yue Chinese (Cantonese)
YUE = "yue"
YUE_CN = "yue-CN"

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

@@ -202,21 +202,27 @@ class BaseOutputTransport(FrameProcessor):
"""
pass
async def write_video_frame(self, frame: OutputImageRawFrame):
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the transport.
Args:
frame: The output video frame to write.
"""
pass
async def write_audio_frame(self, frame: OutputAudioRawFrame):
Returns:
True if the video frame was written successfully, False otherwise.
"""
return False
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the transport.
Args:
frame: The output audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
pass
return False
async def write_dtmf(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame):
"""Write a DTMF tone using the transport's preferred method.
@@ -659,6 +665,7 @@ class BaseOutputTransport(FrameProcessor):
self._audio_queue.get(), timeout=vad_stop_secs
)
yield frame
self._audio_queue.task_done()
except asyncio.TimeoutError:
# Notify the bot stopped speaking upstream if necessary.
await self._bot_stopped_speaking()
@@ -673,6 +680,7 @@ class BaseOutputTransport(FrameProcessor):
frame.audio = await self._mixer.mix(frame.audio)
last_frame_time = time.time()
yield frame
self._audio_queue.task_done()
except asyncio.QueueEmpty:
# Notify the bot stopped speaking upstream if necessary.
diff_time = time.time() - last_frame_time
@@ -738,12 +746,22 @@ class BaseOutputTransport(FrameProcessor):
# Handle frame.
await self._handle_frame(frame)
# Also, push frame downstream in case anyone else needs it.
await self._transport.push_frame(frame)
# If we are not able to write to the transport we shouldn't
# pushb downstream.
push_downstream = True
# Send audio.
if isinstance(frame, OutputAudioRawFrame):
await self._transport.write_audio_frame(frame)
# Try to send audio to the transport.
try:
if isinstance(frame, OutputAudioRawFrame):
push_downstream = await self._transport.write_audio_frame(frame)
except Exception as e:
logger.error(f"{self} Error writing {frame} to transport: {e}")
push_downstream = False
# If we were able to send to the transport, push the frame
# downstream in case anyone else needs it.
if push_downstream:
await self._transport.push_frame(frame)
#
# Video handling

39
src/pipecat/transports/daily/transport.py
View File

@@ -506,11 +506,14 @@ class DailyTransportClient(EventHandler):
self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination)
self._client.update_publishing({"customAudio": {destination: True}})
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the appropriate audio track.
Args:
frame: The audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
future = self._get_event_loop().create_future()
@@ -526,18 +529,24 @@ class DailyTransportClient(EventHandler):
audio_source.write_frames(frame.audio, completion=completion_callback(future))
else:
logger.warning(f"{self} unable to write audio frames to destination [{destination}]")
future.set_result(None)
future.set_result(0)
await future
num_frames = await future
return num_frames > 0
async def write_video_frame(self, frame: OutputImageRawFrame):
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the camera device.
Args:
frame: The image frame to write.
Returns:
True if the video frame was written successfully, False otherwise.
"""
if not frame.transport_destination and self._camera:
self._camera.write_frame(frame.image)
return True
return False
async def setup(self, setup: FrameProcessorSetup):
"""Setup the client with task manager and event queues.
@@ -1835,24 +1844,33 @@ class DailyOutputTransport(BaseOutputTransport):
Args:
destination: The destination identifier to register.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
await self._client.register_audio_destination(destination)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the Daily call.
Args:
frame: The audio frame to write.
"""
await self._client.write_audio_frame(frame)
async def write_video_frame(self, frame: OutputImageRawFrame):
Returns:
True if the audio frame was written successfully, False otherwise.
"""
return await self._client.write_audio_frame(frame)
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the Daily call.
Args:
frame: The video frame to write.
Returns:
True if the video frame was written successfully, False otherwise.
"""
await self._client.write_video_frame(frame)
return await self._client.write_video_frame(frame)
def _supports_native_dtmf(self) -> bool:
"""Daily supports native DTMF via telephone events.
@@ -1974,9 +1992,6 @@ class DailyTransport(BaseTransport):
self._register_event_handler("on_recording_stopped")
self._register_event_handler("on_recording_error")
self._register_event_handler("on_before_leave", sync=True)
# Deprecated
self._register_event_handler("on_joined")
self._register_event_handler("on_left")
#
# BaseTransport

13
src/pipecat/transports/livekit/transport.py
View File

@@ -329,19 +329,21 @@ class LiveKitTransportClient:
except Exception as e:
logger.error(f"Error sending DTMF tone {digit}: {e}")
async def publish_audio(self, audio_frame: rtc.AudioFrame):
async def publish_audio(self, audio_frame: rtc.AudioFrame) -> bool:
"""Publish an audio frame to the room.
Args:
audio_frame: The LiveKit audio frame to publish.
"""
if not self._connected or not self._audio_source:
return
return False
try:
await self._audio_source.capture_frame(audio_frame)
return True
except Exception as e:
logger.error(f"Error publishing audio: {e}")
return False
def get_participants(self) -> List[str]:
"""Get list of participant IDs in the room.
@@ -849,14 +851,17 @@ class LiveKitOutputTransport(BaseOutputTransport):
else:
await self._client.send_data(message.encode())
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the LiveKit room.
Args:
frame: The audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
livekit_audio = self._convert_pipecat_audio_to_livekit(frame.audio)
await self._client.publish_audio(livekit_audio)
return await self._client.publish_audio(livekit_audio)
def _supports_native_dtmf(self) -> bool:
"""LiveKit supports native DTMF via telephone events.

7
src/pipecat/transports/local/audio.py
View File

@@ -172,16 +172,21 @@ class LocalAudioOutputTransport(BaseOutputTransport):
self._out_stream.close()
self._out_stream = None
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the output stream.
Args:
frame: The audio frame to write to the output device.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if self._out_stream:
await self.get_event_loop().run_in_executor(
self._executor, self._out_stream.write, frame.audio
)
return True
return False
class LocalAudioTransport(BaseTransport):

13
src/pipecat/transports/local/tk.py
View File

@@ -191,24 +191,33 @@ class TkOutputTransport(BaseOutputTransport):
self._out_stream.close()
self._out_stream = None
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the output stream.
Args:
frame: The audio frame to write to the output device.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if self._out_stream:
await self.get_event_loop().run_in_executor(
self._executor, self._out_stream.write, frame.audio
)
return True
return False
async def write_video_frame(self, frame: OutputImageRawFrame):
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the Tkinter display.
Args:
frame: The video frame to display in the Tkinter window.
Returns:
True if the video frame was written successfully, False otherwise.
"""
self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
return True
def _write_frame_to_tk(self, frame: OutputImageRawFrame):
"""Write frame data to the Tkinter image label."""

46
src/pipecat/transports/smallwebrtc/connection.py
View File

@@ -206,11 +206,16 @@ class SmallWebRTCConnection(BaseObject):
for real-time audio/video communication.
"""
def __init__(self, ice_servers: Optional[Union[List[str], List[IceServer]]] = None):
def __init__(
self,
ice_servers: Optional[Union[List[str], List[IceServer]]] = None,
connection_timeout_secs: int = 60,
):
"""Initialize the WebRTC connection.
Args:
ice_servers: List of ICE servers as URLs or IceServer objects.
connection_timeout_secs: Timeout in seconds for connecting to the peer.
Raises:
TypeError: If ice_servers contains mixed types or unsupported types.
@@ -231,6 +236,7 @@ class SmallWebRTCConnection(BaseObject):
VIDEO_TRANSCEIVER_INDEX: self.video_input_track,
SCREEN_VIDEO_TRANSCEIVER_INDEX: self.screen_video_input_track,
}
self.connection_timeout_secs = connection_timeout_secs
self._initialize()
@@ -279,6 +285,7 @@ class SmallWebRTCConnection(BaseObject):
self._last_received_time = None
self._message_queue = []
self._pending_app_messages = []
self._connecting_timeout_task = None
def _setup_listeners(self):
"""Set up event listeners for the peer connection."""
@@ -499,6 +506,7 @@ class SmallWebRTCConnection(BaseObject):
self._message_queue.clear()
self._pending_app_messages.clear()
self._track_map = {}
self._cancel_monitoring_connecting_state()
def get_answer(self):
"""Get the SDP answer for the current connection.
@@ -516,9 +524,45 @@ class SmallWebRTCConnection(BaseObject):
"pc_id": self._pc_id,
}
def _monitoring_connecting_state(self) -> None:
"""Start monitoring the peer connection while it is in the *connecting* state.
This method schedules a timeout task that will automatically close the
connection if it remains in the connecting state for more than the specified
timeout, default to 60 seconds.
"""
logger.debug("Monitoring connecting state")
async def timeout_handler():
# We will close the connection in case we have remained in the connecting state for over 1 minute
await asyncio.sleep(self.connection_timeout_secs)
logger.warning("Timeout establishing the connection to the remote peer. Closing.")
await self._close()
# Create and store the timeout task
self._connecting_timeout_task = asyncio.create_task(timeout_handler())
def _cancel_monitoring_connecting_state(self) -> None:
"""Cancel the ongoing connecting-state timeout task, if any.
This method should be called once the connection has either succeeded or
transitioned out of the connecting state. If the timeout task is still
pending, it will be canceled and the reference cleared.
"""
if self._connecting_timeout_task and not self._connecting_timeout_task.done():
logger.debug("Cancelling the connecting timeout task")
self._connecting_timeout_task.cancel()
self._connecting_timeout_task = None
async def _handle_new_connection_state(self):
"""Handle changes in the peer connection state."""
state = self._pc.connectionState
if state == "connecting":
self._monitoring_connecting_state()
else:
self._cancel_monitoring_connecting_state()
if state == "connected" and not self._connect_invoked:
# We are going to wait until the pipeline is ready before triggering the event
return

56
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -309,7 +309,7 @@ class SmallWebRTCClient:
# self._webrtc_connection.ask_to_renegotiate()
frame = None
except MediaStreamError:
logger.warning("Received an unexpected media stream error while reading the audio.")
logger.warning("Received an unexpected media stream error while reading the video.")
frame = None
if frame is None or not isinstance(frame, VideoFrame):
@@ -321,15 +321,21 @@ class SmallWebRTCClient:
# Convert frame to NumPy array in its native format
frame_array = frame.to_ndarray(format=format_name)
frame_rgb = self._convert_frame(frame_array, format_name)
del frame_array # free intermediate array immediately
image_bytes = frame_rgb.tobytes()
del frame_rgb # free RGB array immediately
image_frame = UserImageRawFrame(
user_id=self._webrtc_connection.pc_id,
image=frame_rgb.tobytes(),
image=image_bytes,
size=(frame.width, frame.height),
format="RGB",
)
image_frame.transport_source = video_source
del frame # free original VideoFrame
del image_bytes # reference kept in image_frame
yield image_frame
async def read_audio_frame(self):
@@ -364,40 +370,62 @@ class SmallWebRTCClient:
resampled_frames = self._pipecat_resampler.resample(frame)
for resampled_frame in resampled_frames:
# 16-bit PCM bytes
pcm_bytes = resampled_frame.to_ndarray().astype(np.int16).tobytes()
pcm_array = resampled_frame.to_ndarray().astype(np.int16)
pcm_bytes = pcm_array.tobytes()
del pcm_array # free NumPy array immediately
audio_frame = InputAudioRawFrame(
audio=pcm_bytes,
sample_rate=resampled_frame.sample_rate,
num_channels=self._audio_in_channels,
)
del pcm_bytes # reference kept in audio_frame
yield audio_frame
else:
# 16-bit PCM bytes
pcm_bytes = frame.to_ndarray().astype(np.int16).tobytes()
pcm_array = frame.to_ndarray().astype(np.int16)
pcm_bytes = pcm_array.tobytes()
del pcm_array # free NumPy array immediately
audio_frame = InputAudioRawFrame(
audio=pcm_bytes,
sample_rate=frame.sample_rate,
num_channels=self._audio_in_channels,
)
del pcm_bytes # reference kept in audio_frame
yield audio_frame
async def write_audio_frame(self, frame: OutputAudioRawFrame):
del frame # free original AudioFrame
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the WebRTC connection.
Args:
frame: The audio frame to transmit.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if self._can_send() and self._audio_output_track:
await self._audio_output_track.add_audio_bytes(frame.audio)
return True
return False
async def write_video_frame(self, frame: OutputImageRawFrame):
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the WebRTC connection.
Args:
frame: The video frame to transmit.
Returns:
True if the video frame was written successfully, False otherwise.
"""
if self._can_send() and self._video_output_track:
self._video_output_track.add_video_frame(frame)
return True
return False
async def setup(self, _params: TransportParams, frame):
"""Set up the client with transport parameters.
@@ -800,21 +828,27 @@ class SmallWebRTCOutputTransport(BaseOutputTransport):
"""
await self._client.send_message(frame)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the WebRTC connection.
Args:
frame: The output audio frame to transmit.
"""
await self._client.write_audio_frame(frame)
async def write_video_frame(self, frame: OutputImageRawFrame):
Returns:
True if the audio frame was written successfully, False otherwise.
"""
return await self._client.write_audio_frame(frame)
async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
"""Write a video frame to the WebRTC connection.
Args:
frame: The output video frame to transmit.
Returns:
True if the video frame was written successfully, False otherwise.
"""
await self._client.write_video_frame(frame)
return await self._client.write_video_frame(frame)
class SmallWebRTCTransport(BaseTransport):

16
src/pipecat/transports/tavus/transport.py
View File

@@ -395,15 +395,18 @@ class TavusTransportClient:
participant_settings=participant_settings, profile_settings=profile_settings
)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the transport.
Args:
frame: The audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if not self._client:
return
await self._client.write_audio_frame(frame)
return False
return await self._client.write_audio_frame(frame)
async def register_audio_destination(self, destination: str):
"""Register an audio destination for output.
@@ -625,15 +628,18 @@ class TavusOutputTransport(BaseOutputTransport):
"""Handle interruption events by sending interrupt message."""
await self._client.send_interrupt_message()
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the Tavus transport.
Args:
frame: The audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
# This is the custom track destination expected by Tavus
frame.transport_destination = self._transport_destination
await self._client.write_audio_frame(frame)
return await self._client.write_audio_frame(frame)
async def register_audio_destination(self, destination: str):
"""Register an audio destination.

38
src/pipecat/transports/websocket/client.py
View File

@@ -150,17 +150,39 @@ class WebsocketClientSession:
await self._websocket.close()
self._websocket = None
async def send(self, message: websockets.Data):
async def send(self, message: websockets.Data) -> bool:
"""Send a message through the WebSocket connection.
Args:
message: The message data to send.
"""
result = False
try:
if self._websocket:
await self._websocket.send(message)
result = True
except Exception as e:
logger.error(f"{self} exception sending data: {e.__class__.__name__} ({e})")
finally:
return result
@property
def is_connected(self) -> bool:
"""Check if the WebSocket is currently connected.
Returns:
True if the WebSocket is in connected state.
"""
return self._websocket.state == websockets.State.OPEN if self._websocket else False
@property
def is_closing(self) -> bool:
"""Check if the WebSocket is currently closing.
Returns:
True if the WebSocket is in the process of closing.
"""
return self._websocket.state == websockets.State.CLOSING if self._websocket else False
async def _client_task_handler(self):
"""Handle incoming messages from the WebSocket connection."""
@@ -371,12 +393,18 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
"""
await self._write_frame(frame)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the WebSocket with optional WAV header.
Args:
frame: The output audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if self._session.is_closing or not self._session.is_connected:
return False
frame = OutputAudioRawFrame(
audio=frame.audio,
sample_rate=self.sample_rate,
@@ -402,10 +430,16 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
# Simulate audio playback with a sleep.
await self._write_audio_sleep()
return True
async def _write_frame(self, frame: Frame):
"""Write a frame to the WebSocket after serialization."""
if self._session.is_closing or not self._session.is_connected:
return
if not self._params.serializer:
return
payload = await self._params.serializer.serialize(frame)
if payload:
await self._session.send(payload)

9
src/pipecat/transports/websocket/fastapi.py
View File

@@ -410,14 +410,17 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
"""
await self._write_frame(frame)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the WebSocket with timing simulation.
Args:
frame: The output audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if self._client.is_closing or not self._client.is_connected:
return
return False
frame = OutputAudioRawFrame(
audio=frame.audio,
@@ -444,6 +447,8 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
# Simulate audio playback with a sleep.
await self._write_audio_sleep()
return True
async def _write_frame(self, frame: Frame):
"""Serialize and send a frame through the WebSocket."""
if self._client.is_closing or not self._client.is_connected:

9
src/pipecat/transports/websocket/server.py
View File

@@ -346,14 +346,17 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
"""
await self._write_frame(frame)
async def write_audio_frame(self, frame: OutputAudioRawFrame):
async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
"""Write an audio frame to the WebSocket client with timing control.
Args:
frame: The output audio frame to write.
Returns:
True if the audio frame was written successfully, False otherwise.
"""
if not self._websocket:
return
return False
frame = OutputAudioRawFrame(
audio=frame.audio,
@@ -380,6 +383,8 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
# Simulate audio playback with a sleep.
await self._write_audio_sleep()
return True
async def _write_frame(self, frame: Frame):
"""Serialize and send a frame to the WebSocket client."""
if not self._params.serializer:

2
src/pipecat/utils/base_object.py
View File

@@ -139,7 +139,7 @@ class BaseObject(ABC):
name=event_name, handlers=[], is_sync=sync
)
else:
logger.warning(f"Event handler {event_name} not registered")
logger.warning(f"Event handler {event_name} already registered")
async def _call_event_handler(self, event_name: str, *args, **kwargs):
"""Call all registered handlers for the specified event.