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

Merge pull request #2067 from pipecat-ai/mb/update-docstrings-for-ref-docs

Browse Source 
Update base service class docstrings for better docs auto-generation
This commit is contained in:
Mark Backman
2025-06-26 07:07:59 -04:00
committed by GitHub
parent ba43558049 951c8d34da
commit 1cd42066a6
16 changed files with 1151 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
CONTRIBUTING.md
View File

@@ -41,36 +41,76 @@ We use Ruff for code linting and formatting. Please ensure your code passes all
We follow Google-style docstrings with these specific conventions:
- Class docstrings should fully document all parameters used in `__init__`
- We don't require separate docstrings for `__init__` methods when parameters are documented in the class docstring
- Property methods should have docstrings explaining their purpose and return value
**Regular Classes:**
Example of correctly documented class:
- Class docstring describes the class purpose and documents all `__init__` parameters in an `Args:` section
- No separate `__init__` docstring needed
- All public methods must have docstrings with `Args:` and `Returns:` sections as appropriate
**Dataclasses:**
- Class docstring describes the purpose and documents all fields in a `Parameters:` section
- No `__init__` docstring (auto-generated)
**Properties:**
- Must have docstrings with `Returns:` section
**Abstract Methods:**
- Must have docstrings explaining what subclasses should implement
#### Examples:
```python
class MyClass:
"""Class description.
Additional details about the class.
# Regular class
class MyService(BaseService):
"""Description of what the service does.
Args:
param1: Description of first parameter.
param2: Description of second parameter.
param1: Description of param1.
param2: Description of param2. Defaults to True.
**kwargs: Additional arguments passed to parent.
"""
def __init__(self, param1, param2):
# No docstring required here as parameters are documented above
self.param1 = param1
self.param2 = param2
def __init__(self, param1: str, param2: bool = True, **kwargs):
# No docstring - parameters documented above
super().__init__(**kwargs)
@property
def some_property(self) -> str:
"""Get the formatted property value.
def sample_rate(self) -> int:
"""Get the current sample rate.
Returns:
A string representation of the property.
The sample rate in Hz.
"""
return f"Property: {self.param1}"
return self._sample_rate
async def process_data(self, data: str) -> bool:
"""Process the provided data.
Args:
data: The data to process.
Returns:
True if processing succeeded.
"""
pass
# Dataclass
@dataclass
class ConfigParams:
"""Configuration parameters for the service.
Parameters:
host: The host address.
port: The port number. Defaults to 8080.
timeout: Connection timeout in seconds.
"""
host: str
port: int = 8080
timeout: float = 30.0
```
# Contributor Covenant Code of Conduct

34
docs/api/conf.py
View File

@@ -1,5 +1,6 @@
import logging
import sys
from datetime import datetime
from pathlib import Path
# Configure logging
@@ -13,7 +14,8 @@ sys.path.insert(0, str(project_root / "src"))
# Project information
project = "pipecat-ai"
copyright = "2024, Daily"
current_year = datetime.now().year
copyright = f"2024-{current_year}, Daily" if current_year > 2024 else "2024, Daily"
author = "Daily"
# General configuration
@@ -27,15 +29,14 @@ extensions = [
# Napoleon settings
napoleon_google_docstring = True
napoleon_numpy_docstring = False
napoleon_include_init_with_doc = True
napoleon_include_init_with_doc = False
# AutoDoc settings
autodoc_default_options = {
"members": True,
"member-order": "bysource",
"special-members": "__init__",
"undoc-members": True,
"exclude-members": "__weakref__",
"exclude-members": "__weakref__,__init__",
"no-index": True,
"show-inheritance": True,
}
@@ -145,6 +146,28 @@ autodoc_mock_imports = [
"transformers.AutoFeatureExtractor",
# Also add specific classes that are imported
"AutoFeatureExtractor",
# Sentry dependencies
"sentry_sdk",
# AWS Nova Sonic dependencies
"aws_sdk_bedrock_runtime",
"aws_sdk_bedrock_runtime.client",
"aws_sdk_bedrock_runtime.config",
"aws_sdk_bedrock_runtime.models",
"smithy_aws_core",
"smithy_aws_core.credentials_resolvers",
"smithy_aws_core.credentials_resolvers.static",
"smithy_aws_core.identity",
"smithy_core",
"smithy_core.aio",
"smithy_core.aio.eventstream",
# MCP dependencies (you may already have these)
"mcp",
"mcp.client",
"mcp.client.session_group",
"mcp.client.sse",
"mcp.client.stdio",
"mcp.ClientSession",
"mcp.StdioServerParameters",
]
# HTML output settings
@@ -249,6 +272,9 @@ def clean_title(title: str) -> str:
"playht": "PlayHT",
"xtts": "XTTS",
"lmnt": "LMNT",
"stt": "STT",
"tts": "TTS",
"llm": "LLM",
}
# Check if the entire title is a special case

3
pyproject.toml
View File

@@ -123,8 +123,7 @@ select = [
"D", # Docstring rules
"I", # Import rules
]
# We ignore D107 because class docstrings already document __init__ parameters
# and our Sphinx configuration uses napoleon_include_init_with_doc=True
# Ignore requirement for __init__ docstrings
ignore = ["D107"]
[tool.ruff.lint.pydocstyle]

68
src/pipecat/services/ai_service.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base AI service implementation.
Provides the foundation for all AI services in the Pipecat framework, including
model management, settings handling, and frame processing lifecycle methods.
"""
from typing import Any, AsyncGenerator, Dict, Mapping
from loguru import logger
@@ -20,6 +26,17 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class AIService(FrameProcessor):
"""Base class for all AI services.
Provides common functionality for AI services including model management,
settings handling, session properties, and frame processing lifecycle.
Subclasses should implement specific AI functionality while leveraging
this base infrastructure.
Args:
**kwargs: Additional arguments passed to the parent FrameProcessor.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._model_name: str = ""
@@ -28,19 +45,53 @@ class AIService(FrameProcessor):
@property
def model_name(self) -> str:
"""Get the current model name.
Returns:
The name of the AI model being used.
"""
return self._model_name
def set_model_name(self, model: str):
"""Set the AI model name and update metrics.
Args:
model: The name of the AI model to use.
"""
self._model_name = model
self.set_core_metrics_data(MetricsData(processor=self.name, model=self._model_name))
async def start(self, frame: StartFrame):
"""Start the AI service.
Called when the service should begin processing. Subclasses should
override this method to perform service-specific initialization.
Args:
frame: The start frame containing initialization parameters.
"""
pass
async def stop(self, frame: EndFrame):
"""Stop the AI service.
Called when the service should stop processing. Subclasses should
override this method to perform cleanup operations.
Args:
frame: The end frame.
"""
pass
async def cancel(self, frame: CancelFrame):
"""Cancel the AI service.
Called when the service should cancel all operations. Subclasses should
override this method to handle cancellation logic.
Args:
frame: The cancel frame.
"""
pass
async def _update_settings(self, settings: Mapping[str, Any]):
@@ -87,6 +138,15 @@ class AIService(FrameProcessor):
logger.warning(f"Unknown setting for {self.name} service: {key}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames and handle service lifecycle.
Automatically handles StartFrame, EndFrame, and CancelFrame by calling
the appropriate lifecycle methods.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
@@ -97,6 +157,14 @@ class AIService(FrameProcessor):
await self.stop(frame)
async def process_generator(self, generator: AsyncGenerator[Frame | None, None]):
"""Process frames from an async generator.
Takes an async generator that yields frames and processes each one,
handling error frames specially by pushing them as errors.
Args:
generator: An async generator that yields Frame objects or None.
"""
async for f in generator:
if f:
if isinstance(f, ErrorFrame):

11
src/pipecat/services/ai_services.py
View File

@@ -4,6 +4,17 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Deprecated AI services module.
This module is deprecated. Import services directly from their respective modules:
- pipecat.services.ai_service
- pipecat.services.image_service
- pipecat.services.llm_service
- pipecat.services.stt_service
- pipecat.services.tts_service
- pipecat.services.vision_service
"""
import sys
from pipecat.services import DeprecatedModuleProxy

143
src/pipecat/services/cartesia/tts.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Cartesia text-to-speech service implementations."""
import base64
import json
import uuid
@@ -43,6 +45,14 @@ except ModuleNotFoundError as e:
def language_to_cartesia_language(language: Language) -> Optional[str]:
"""Convert a Language enum to Cartesia language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Cartesia language code, or None if not supported.
"""
BASE_LANGUAGES = {
Language.DE: "de",
Language.EN: "en",
@@ -75,7 +85,35 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
class CartesiaTTSService(AudioContextWordTTSService):
"""Cartesia TTS service with WebSocket streaming and word timestamps.
Provides text-to-speech using Cartesia's streaming WebSocket API.
Supports word-level timestamps, audio context management, and various voice
customization options including speed and emotion controls.
Args:
api_key: Cartesia API key for authentication.
voice_id: ID of the voice to use for synthesis.
cartesia_version: API version string for Cartesia service.
url: WebSocket URL for Cartesia TTS API.
model: TTS model to use (e.g., "sonic-2").
sample_rate: Audio sample rate. If None, uses default.
encoding: Audio encoding format.
container: Audio container format.
params: Additional input parameters for voice customization.
text_aggregator: Custom text aggregator for processing input text.
**kwargs: Additional arguments passed to the parent service.
"""
class InputParams(BaseModel):
"""Input parameters for Cartesia TTS configuration.
Parameters:
language: Language to use for synthesis.
speed: Voice speed control (string or float).
emotion: List of emotion controls (deprecated).
"""
language: Optional[Language] = Language.EN
speed: Optional[Union[str, float]] = ""
emotion: Optional[List[str]] = []
@@ -138,14 +176,32 @@ class CartesiaTTSService(AudioContextWordTTSService):
self._receive_task = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
Returns:
True, as Cartesia service supports metrics generation.
"""
return True
async def set_model(self, model: str):
"""Set the TTS model.
Args:
model: The model name to use for synthesis.
"""
self._model_id = model
await super().set_model(model)
logger.info(f"Switching TTS model to: [{model}]")
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert a Language enum to Cartesia language format.
Args:
language: The language to convert.
Returns:
The Cartesia-specific language code, or None if not supported.
"""
return language_to_cartesia_language(language)
def _build_msg(
@@ -183,15 +239,30 @@ class CartesiaTTSService(AudioContextWordTTSService):
return json.dumps(msg)
async def start(self, frame: StartFrame):
"""Start the Cartesia TTS service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._settings["output_format"]["sample_rate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the Cartesia TTS service.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Stop the Cartesia TTS service.
Args:
frame: The end frame.
"""
await super().cancel(frame)
await self._disconnect()
@@ -248,6 +319,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
self._context_id = None
async def flush_audio(self):
"""Flush any pending audio and finalize the current context."""
if not self._context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
@@ -290,6 +362,14 @@ class CartesiaTTSService(AudioContextWordTTSService):
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Cartesia's streaming API.
Args:
text: The text to synthesize into speech.
Yields:
Frame: Audio frames containing the synthesized speech.
"""
logger.debug(f"{self}: Generating TTS [{text}]")
try:
@@ -319,7 +399,34 @@ class CartesiaTTSService(AudioContextWordTTSService):
class CartesiaHttpTTSService(TTSService):
"""Cartesia HTTP-based TTS service.
Provides text-to-speech using Cartesia's HTTP API for simpler, non-streaming
synthesis. Suitable for use cases where streaming is not required and simpler
integration is preferred.
Args:
api_key: Cartesia API key for authentication.
voice_id: ID of the voice to use for synthesis.
model: TTS model to use (e.g., "sonic-2").
base_url: Base URL for Cartesia HTTP API.
cartesia_version: API version string for Cartesia service.
sample_rate: Audio sample rate. If None, uses default.
encoding: Audio encoding format.
container: Audio container format.
params: Additional input parameters for voice customization.
**kwargs: Additional arguments passed to the parent TTSService.
"""
class InputParams(BaseModel):
"""Input parameters for Cartesia HTTP TTS configuration.
Parameters:
language: Language to use for synthesis.
speed: Voice speed control (string or float).
emotion: List of emotion controls (deprecated).
"""
language: Optional[Language] = Language.EN
speed: Optional[Union[str, float]] = ""
emotion: Optional[List[str]] = Field(default_factory=list)
@@ -366,25 +473,61 @@ class CartesiaHttpTTSService(TTSService):
)
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
Returns:
True, as Cartesia HTTP service supports metrics generation.
"""
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert a Language enum to Cartesia language format.
Args:
language: The language to convert.
Returns:
The Cartesia-specific language code, or None if not supported.
"""
return language_to_cartesia_language(language)
async def start(self, frame: StartFrame):
"""Start the Cartesia HTTP TTS service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._settings["output_format"]["sample_rate"] = self.sample_rate
async def stop(self, frame: EndFrame):
"""Stop the Cartesia HTTP TTS service.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._client.close()
async def cancel(self, frame: CancelFrame):
"""Cancel the Cartesia HTTP TTS service.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._client.close()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Cartesia's HTTP API.
Args:
text: The text to synthesize into speech.
Yields:
Frame: Audio frames containing the synthesized speech.
"""
logger.debug(f"{self}: Generating TTS [{text}]")
try:

68
src/pipecat/services/deepgram/stt.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Deepgram speech-to-text service implementation."""
from typing import AsyncGenerator, Dict, Optional
from loguru import logger
@@ -41,6 +43,22 @@ except ModuleNotFoundError as e:
class DeepgramSTTService(STTService):
"""Deepgram speech-to-text service.
Provides real-time speech recognition using Deepgram's WebSocket API.
Supports configurable models, languages, VAD events, and various audio
processing options.
Args:
api_key: Deepgram API key for authentication.
url: Deprecated. Use base_url instead.
base_url: Custom Deepgram API base URL.
sample_rate: Audio sample rate. If None, uses default or live_options value.
live_options: Deepgram LiveOptions for detailed configuration.
addons: Additional Deepgram features to enable.
**kwargs: Additional arguments passed to the parent STTService.
"""
def __init__(
self,
*,
@@ -108,12 +126,27 @@ class DeepgramSTTService(STTService):
@property
def vad_enabled(self):
"""Check if Deepgram VAD events are enabled.
Returns:
True if VAD events are enabled in the current settings.
"""
return self._settings["vad_events"]
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
Returns:
True, as Deepgram service supports metrics generation.
"""
return True
async def set_model(self, model: str):
"""Set the Deepgram model and reconnect.
Args:
model: The Deepgram model name to use.
"""
await super().set_model(model)
logger.info(f"Switching STT model to: [{model}]")
self._settings["model"] = model
@@ -121,25 +154,53 @@ class DeepgramSTTService(STTService):
await self._connect()
async def set_language(self, language: Language):
"""Set the recognition language and reconnect.
Args:
language: The language to use for speech recognition.
"""
logger.info(f"Switching STT language to: [{language}]")
self._settings["language"] = language
await self._disconnect()
await self._connect()
async def start(self, frame: StartFrame):
"""Start the Deepgram STT service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the Deepgram STT service.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the Deepgram STT service.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._disconnect()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Send audio data to Deepgram for transcription.
Args:
audio: Raw audio bytes to transcribe.
Yields:
Frame: None (transcription results come via WebSocket callbacks).
"""
await self._connection.send(audio)
yield None
@@ -172,6 +233,7 @@ class DeepgramSTTService(STTService):
await self._connection.finish()
async def start_metrics(self):
"""Start TTFB and processing metrics collection."""
await self.start_ttfb_metrics()
await self.start_processing_metrics()
@@ -235,6 +297,12 @@ class DeepgramSTTService(STTService):
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with Deepgram-specific handling.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, UserStartedSpeakingFrame) and not self.vad_enabled:

37
src/pipecat/services/image_service.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Image generation service implementation.
Provides base functionality for AI-powered image generation services that convert
text prompts into images.
"""
from abc import abstractmethod
from typing import AsyncGenerator
@@ -13,15 +19,46 @@ from pipecat.services.ai_service import AIService
class ImageGenService(AIService):
"""Base class for image generation services.
Processes TextFrames by using their content as prompts for image generation.
Subclasses must implement the run_image_gen method to provide actual image
generation functionality using their specific AI service.
Args:
**kwargs: Additional arguments passed to the parent AIService.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Renders the image. Returns an Image object.
@abstractmethod
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
"""Generate an image from a text prompt.
This method must be implemented by subclasses to provide actual image
generation functionality using their specific AI service.
Args:
prompt: The text prompt to generate an image from.
Yields:
Frame: Frames containing the generated image (typically ImageRawFrame
or URLImageRawFrame).
"""
pass
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for image generation.
TextFrames are used as prompts for image generation, while other frames
are passed through unchanged.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TextFrame):

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

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base classes for Large Language Model services with function calling support."""
import asyncio
import inspect
from dataclasses import dataclass
@@ -41,23 +43,34 @@ FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]]
# Type alias for a callback function that handles the result of an LLM function call.
class FunctionCallResultCallback(Protocol):
"""Protocol for function call result callbacks.
Handles the result of an LLM function call execution.
"""
async def __call__(
self, result: Any, *, properties: Optional[FunctionCallResultProperties] = None
) -> None: ...
) -> None:
"""Call the result callback.
Args:
result: The result of the function call.
properties: Optional properties for the result.
"""
...
@dataclass
class FunctionCallParams:
"""Parameters for a function call.
Attributes:
function_name (str): The name of the function being called.
arguments (Mapping[str, Any]): The arguments for the function.
tool_call_id (str): A unique identifier for the function call.
llm (LLMService): The LLMService instance being used.
context (OpenAILLMContext): The LLM context.
result_callback (FunctionCallResultCallback): Callback to handle the result of the function call.
Parameters:
function_name: The name of the function being called.
tool_call_id: A unique identifier for the function call.
arguments: The arguments for the function.
llm: The LLMService instance being used.
context: The LLM context.
result_callback: Callback to handle the result of the function call.
"""
function_name: str
@@ -70,14 +83,14 @@ class FunctionCallParams:
@dataclass
class FunctionCallRegistryItem:
"""Represents an entry in our function call registry. This is what the user
registers.
"""Represents an entry in the function call registry.
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.
This is what the user registers when calling register_function.
Parameters:
function_name: The name of the function (None for catch-all handler).
handler: The handler for processing function call parameters.
cancel_on_interruption: Whether to cancel the call on interruption.
"""
function_name: Optional[str]
@@ -87,16 +100,17 @@ class FunctionCallRegistryItem:
@dataclass
class FunctionCallRunnerItem:
"""Represents an internal function call entry to our function call
runner. The runner executes function calls in order.
"""Internal function call entry for the function call runner.
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.
The runner executes function calls in order.
Parameters:
registry_item: The registry item containing handler information.
function_name: The name of the function.
tool_call_id: A unique identifier for the function call.
arguments: The arguments for the function.
context: The LLM context.
run_llm: Optional flag to control LLM execution after function call.
"""
registry_item: FunctionCallRegistryItem
@@ -108,22 +122,32 @@ class FunctionCallRunnerItem:
class LLMService(AIService):
"""This is the base class for all LLM services. It handles function calling
registration and execution. The class also provides event handlers.
"""Base class for all LLM services.
An event to know when an LLM service completion timeout occurs:
Handles function calling registration and execution with support for both
parallel and sequential execution modes. Provides event handlers for
completion timeouts and function call lifecycle events.
@task.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
...
Args:
run_in_parallel: Whether to run function calls in parallel or sequentially.
Defaults to True.
**kwargs: Additional arguments passed to the parent AIService.
And an event to know that function calls have been received from the LLM
service and that we are going to start executing them:
Event handlers:
on_completion_timeout: Called when an LLM completion timeout occurs.
on_function_calls_started: Called when function calls are received and
execution is about to start.
@task.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls: Sequence[FunctionCallFromLLM]):
...
Example:
```python
@task.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
logger.warning("LLM completion timed out")
@task.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
logger.info(f"Starting {len(function_calls)} function calls")
```
"""
# OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations.
@@ -143,6 +167,11 @@ class LLMService(AIService):
self._register_event_handler("on_completion_timeout")
def get_llm_adapter(self) -> BaseLLMAdapter:
"""Get the LLM adapter instance.
Returns:
The adapter instance used for LLM communication.
"""
return self._adapter
def create_context_aggregator(
@@ -152,24 +181,57 @@ class LLMService(AIService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> Any:
"""Create a context aggregator for managing LLM conversation context.
Must be implemented by subclasses.
Args:
context: The LLM context to create an aggregator for.
user_params: Parameters for user message aggregation.
assistant_params: Parameters for assistant message aggregation.
Returns:
A context aggregator instance.
"""
pass
async def start(self, frame: StartFrame):
"""Start the LLM service.
Args:
frame: The start frame.
"""
await super().start(frame)
if not self._run_in_parallel:
await self._create_sequential_runner_task()
async def stop(self, frame: EndFrame):
"""Stop the LLM service.
Args:
frame: The end frame.
"""
await super().stop(frame)
if not self._run_in_parallel:
await self._cancel_sequential_runner_task()
async def cancel(self, frame: CancelFrame):
"""Cancel the LLM service.
Args:
frame: The cancel frame.
"""
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):
"""Process a frame.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
@@ -188,6 +250,18 @@ class LLMService(AIService):
*,
cancel_on_interruption: bool = True,
):
"""Register a function handler for LLM function calls.
Args:
function_name: The name of the function to handle. Use None to handle
all function calls with a catch-all handler.
handler: The function handler. Should accept a single FunctionCallParams
parameter.
start_callback: Legacy callback function (deprecated). Put initialization
code at the top of your handler instead.
cancel_on_interruption: Whether to cancel this function call when an
interruption occurs. Defaults to True.
"""
# Registering a function with the function_name set to None will run
# that handler for all functions
self._functions[function_name] = FunctionCallRegistryItem(
@@ -210,16 +284,38 @@ class LLMService(AIService):
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name: Optional[str]):
"""Remove a registered function handler.
Args:
function_name: The name of the function handler to remove.
"""
del self._functions[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
def has_function(self, function_name: str):
"""Check if a function handler is registered.
Args:
function_name: The name of the function to check.
Returns:
True if the function is registered or if a catch-all handler (None)
is registered.
"""
if None in self._functions.keys():
return True
return function_name in self._functions.keys()
async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]):
"""Execute a sequence of function calls from the LLM.
Triggers the on_function_calls_started event and executes functions
either in parallel or sequentially based on the run_in_parallel setting.
Args:
function_calls: The function calls to execute.
"""
if len(function_calls) == 0:
return
@@ -257,7 +353,7 @@ class LLMService(AIService):
else:
await self._sequential_runner_queue.put(runner_item)
async def call_start_function(self, context: OpenAILLMContext, function_name: str):
async def _call_start_function(self, context: OpenAILLMContext, function_name: str):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](function_name, self, context)
elif None in self._start_callbacks.keys():
@@ -272,6 +368,18 @@ class LLMService(AIService):
text_content: Optional[str] = None,
video_source: Optional[str] = None,
):
"""Request an image from a user.
Pushes a UserImageRequestFrame upstream to request an image from the
specified user.
Args:
user_id: The ID of the user to request an image from.
function_name: Optional function name associated with the request.
tool_call_id: Optional tool call ID associated with the request.
text_content: Optional text content/context for the image request.
video_source: Optional video source identifier.
"""
await self.push_frame(
UserImageRequestFrame(
user_id=user_id,
@@ -316,7 +424,7 @@ class LLMService(AIService):
)
# NOTE(aleix): This needs to be removed after we remove the deprecation.
await self.call_start_function(runner_item.context, runner_item.function_name)
await self._call_start_function(runner_item.context, runner_item.function_name)
# Push a function call in-progress downstream. This frame will let our
# assistant context aggregator know that we are in the middle of a

37
src/pipecat/services/mcp_service.py
View File

@@ -1,3 +1,11 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""MCP (Model Context Protocol) client for integrating external tools with LLMs."""
import json
from typing import Any, Dict, List, Optional, Union
@@ -19,6 +27,20 @@ except ModuleNotFoundError as e:
class MCPClient(BaseObject):
"""Client for Model Context Protocol (MCP) servers.
Enables integration with MCP servers to provide external tools and resources
to LLMs. Supports both stdio and SSE server connections with automatic tool
registration and schema conversion.
Args:
server_params: Server connection parameters (stdio or SSE).
**kwargs: Additional arguments passed to the parent BaseObject.
Raises:
TypeError: If server_params is not a supported parameter type.
"""
def __init__(
self,
server_params: Union[StdioServerParameters, SseServerParameters],
@@ -39,6 +61,17 @@ class MCPClient(BaseObject):
)
async def register_tools(self, llm) -> ToolsSchema:
"""Register all available MCP tools with an LLM service.
Connects to the MCP server, discovers available tools, converts their
schemas to Pipecat format, and registers them with the LLM service.
Args:
llm: The Pipecat LLM service to register tools with.
Returns:
A ToolsSchema containing all successfully registered tools.
"""
tools_schema = await self._register_tools(llm)
return tools_schema
@@ -46,13 +79,13 @@ class MCPClient(BaseObject):
self, tool_name: str, tool_schema: Dict[str, Any]
) -> FunctionSchema:
"""Convert an mcp tool schema to Pipecat's FunctionSchema format.
Args:
tool_name: The name of the tool
tool_schema: The mcp tool schema
Returns:
A FunctionSchema instance
"""
logger.debug(f"Converting schema for tool '{tool_name}'")
logger.trace(f"Original schema: {json.dumps(tool_schema, indent=2)}")
@@ -72,6 +105,7 @@ class MCPClient(BaseObject):
async def _sse_register_tools(self, llm) -> ToolsSchema:
"""Register all available mcp.run tools with the LLM service.
Args:
llm: The Pipecat LLM service to register tools with
Returns:
@@ -120,6 +154,7 @@ class MCPClient(BaseObject):
async def _stdio_register_tools(self, llm) -> ToolsSchema:
"""Register all available mcp.run tools with the LLM service.
Args:
llm: The Pipecat LLM service to register tools with
Returns:

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

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base OpenAI LLM service implementation."""
import base64
import json
from typing import Any, Dict, List, Mapping, Optional
@@ -40,16 +42,39 @@ from pipecat.utils.watchdog_async_iterator import WatchdogAsyncIterator
class BaseOpenAILLMService(LLMService):
"""This is the base for all services that use the AsyncOpenAI client.
"""Base class for all services that use the AsyncOpenAI client.
This service consumes OpenAILLMContextFrame frames, which contain a reference
to an OpenAILLMContext frame. The OpenAILLMContext object defines the context
sent to the LLM for a completion. This includes user, assistant and system messages
as well as tool choices and the tool, which is used if requesting function
calls from the LLM.
to an OpenAILLMContext object. The context defines what is sent to the LLM for
completion, including user, assistant, and system messages, as well as tool
choices and function call configurations.
Args:
model: The OpenAI model name to use (e.g., "gpt-4.1", "gpt-4o").
api_key: OpenAI API key. If None, uses environment variable.
base_url: Custom base URL for OpenAI API. If None, uses default.
organization: OpenAI organization ID.
project: OpenAI project ID.
default_headers: Additional HTTP headers to include in requests.
params: Input parameters for model configuration and behavior.
**kwargs: Additional arguments passed to the parent LLMService.
"""
class InputParams(BaseModel):
"""Input parameters for OpenAI model configuration.
Parameters:
frequency_penalty: Penalty for frequent tokens (-2.0 to 2.0).
presence_penalty: Penalty for new tokens (-2.0 to 2.0).
seed: Random seed for deterministic outputs.
temperature: Sampling temperature (0.0 to 2.0).
top_k: Top-k sampling parameter (currently ignored by OpenAI).
top_p: Top-p (nucleus) sampling parameter (0.0 to 1.0).
max_tokens: Maximum tokens in response (deprecated, use max_completion_tokens).
max_completion_tokens: Maximum completion tokens to generate.
extra: Additional model-specific parameters.
"""
frequency_penalty: Optional[float] = Field(
default_factory=lambda: NOT_GIVEN, ge=-2.0, le=2.0
)
@@ -111,6 +136,19 @@ class BaseOpenAILLMService(LLMService):
default_headers=None,
**kwargs,
):
"""Create an AsyncOpenAI client instance.
Args:
api_key: OpenAI API key.
base_url: Custom base URL for the API.
organization: OpenAI organization ID.
project: OpenAI project ID.
default_headers: Additional HTTP headers.
**kwargs: Additional client configuration arguments.
Returns:
Configured AsyncOpenAI client instance.
"""
return AsyncOpenAI(
api_key=api_key,
base_url=base_url,
@@ -125,11 +163,25 @@ class BaseOpenAILLMService(LLMService):
)
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
Returns:
True, as OpenAI service supports metrics generation.
"""
return True
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:
"""Get streaming chat completions from OpenAI API.
Args:
context: The LLM context containing tools and configuration.
messages: List of chat completion messages to send.
Returns:
Async stream of chat completion chunks.
"""
params = {
"model": self.model_name,
"stream": True,
@@ -277,6 +329,15 @@ class BaseOpenAILLMService(LLMService):
await self.run_function_calls(function_calls)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for LLM completion requests.
Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None

88
src/pipecat/services/openai/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI LLM service implementation with context aggregators."""
import json
from dataclasses import dataclass
from typing import Any, Optional
@@ -26,17 +28,46 @@ from pipecat.services.openai.base_llm import BaseOpenAILLMService
@dataclass
class OpenAIContextAggregatorPair:
"""Pair of OpenAI context aggregators for user and assistant messages.
Parameters:
_user: User context aggregator for processing user messages.
_assistant: Assistant context aggregator for processing assistant messages.
"""
_user: "OpenAIUserContextAggregator"
_assistant: "OpenAIAssistantContextAggregator"
def user(self) -> "OpenAIUserContextAggregator":
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> "OpenAIAssistantContextAggregator":
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class OpenAILLMService(BaseOpenAILLMService):
"""OpenAI LLM service implementation.
Provides a complete OpenAI LLM service with context aggregation support.
Uses the BaseOpenAILLMService for core functionality and adds OpenAI-specific
context aggregator creation.
Args:
model: The OpenAI model name to use. Defaults to "gpt-4.1".
params: Input parameters for model configuration.
**kwargs: Additional arguments passed to the parent BaseOpenAILLMService.
"""
def __init__(
self,
*,
@@ -53,14 +84,15 @@ class OpenAILLMService(BaseOpenAILLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> OpenAIContextAggregatorPair:
"""Create an instance of OpenAIContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create OpenAI-specific context aggregators.
Creates a pair of context aggregators optimized for OpenAI's message format,
including support for function calls, tool usage, and image handling.
Args:
context (OpenAILLMContext): The LLM context.
user_params (LLMUserAggregatorParams, optional): User aggregator parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User aggregator parameters.
context: The LLM context to create aggregators for.
user_params: Parameters for user message aggregation.
assistant_params: Parameters for assistant message aggregation.
Returns:
OpenAIContextAggregatorPair: A pair of context aggregators, one for
@@ -75,11 +107,32 @@ class OpenAILLMService(BaseOpenAILLMService):
class OpenAIUserContextAggregator(LLMUserContextAggregator):
"""OpenAI-specific user context aggregator.
Handles aggregation of user messages for OpenAI LLM services.
Inherits all functionality from the base LLMUserContextAggregator.
"""
pass
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
"""OpenAI-specific assistant context aggregator.
Handles aggregation of assistant messages for OpenAI LLM services,
with specialized support for OpenAI's function calling format,
tool usage tracking, and image message handling.
"""
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle a function call in progress.
Adds the function call to the context with an IN_PROGRESS status
to track ongoing function execution.
Args:
frame: Frame containing function call progress information.
"""
self._context.add_message(
{
"role": "assistant",
@@ -104,6 +157,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle the result of a function call.
Updates the context with the function call result, replacing any
previous IN_PROGRESS status.
Args:
frame: Frame containing the function call result.
"""
if frame.result:
result = json.dumps(frame.result)
await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
@@ -113,6 +174,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle a cancelled function call.
Updates the context to mark the function call as cancelled.
Args:
frame: Frame containing the function call cancellation information.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
@@ -129,6 +197,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
message["content"] = result
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle a user image frame from a function call request.
Marks the associated function call as completed and adds the image
to the context for processing.
Args:
frame: Frame containing the user image and request context.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)

99
src/pipecat/services/stt_service.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base classes for Speech-to-Text services with continuous and segmented processing."""
import io
import wave
from abc import abstractmethod
@@ -26,7 +28,19 @@ from pipecat.transcriptions.language import Language
class STTService(AIService):
"""STTService is a base class for speech-to-text services."""
"""Base class for speech-to-text services.
Provides common functionality for STT services including audio passthrough,
muting, settings management, and audio processing. Subclasses must implement
the run_stt method to provide actual speech recognition.
Args:
audio_passthrough: Whether to pass audio frames downstream after processing.
Defaults to True.
sample_rate: The sample rate for audio input. If None, will be determined
from the start frame.
**kwargs: Additional arguments passed to the parent AIService.
"""
def __init__(
self,
@@ -44,25 +58,59 @@ class STTService(AIService):
@property
def is_muted(self) -> bool:
"""Returns whether the STT service is currently muted."""
"""Check if the STT service is currently muted.
Returns:
True if the service is muted and will not process audio.
"""
return self._muted
@property
def sample_rate(self) -> int:
"""Get the current sample rate for audio processing.
Returns:
The sample rate in Hz.
"""
return self._sample_rate
async def set_model(self, model: str):
"""Set the speech recognition model.
Args:
model: The name of the model to use for speech recognition.
"""
self.set_model_name(model)
async def set_language(self, language: Language):
"""Set the language for speech recognition.
Args:
language: The language to use for speech recognition.
"""
pass
@abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
"""Run speech-to-text on the provided audio data.
This method must be implemented by subclasses to provide actual speech
recognition functionality.
Args:
audio: Raw audio bytes to transcribe.
Yields:
Frame: Frames containing transcription results (typically TextFrame).
"""
pass
async def start(self, frame: StartFrame):
"""Start the STT service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate
@@ -80,13 +128,24 @@ class STTService(AIService):
logger.warning(f"Unknown setting for STT service: {key}")
async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
"""Process an audio frame for speech recognition.
Args:
frame: The audio frame to process.
direction: The direction of frame processing.
"""
if self._muted:
return
await self.process_generator(self.run_stt(frame.audio))
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
"""Process frames, handling VAD events and audio segmentation.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, AudioRawFrame):
@@ -106,14 +165,19 @@ class STTService(AIService):
class SegmentedSTTService(STTService):
"""SegmentedSTTService is an STTService that uses VAD events to detect
speech and will run speech-to-text on speech segments only, instead of a
continous stream. Since it uses VAD it means that VAD needs to be enabled in
the pipeline.
"""STT service that processes speech in segments using VAD events.
This service always keeps a small audio buffer to take into account that VAD
events are delayed from when the user speech really starts.
Uses Voice Activity Detection (VAD) events to detect speech segments and runs
speech-to-text only on those segments, rather than continuously.
Requires VAD to be enabled in the pipeline to function properly. Maintains a
small audio buffer to account for the delay between actual speech start and
VAD detection.
Args:
sample_rate: The sample rate for audio input. If None, will be determined
from the start frame.
**kwargs: Additional arguments passed to the parent STTService.
"""
def __init__(self, *, sample_rate: Optional[int] = None, **kwargs):
@@ -125,10 +189,16 @@ class SegmentedSTTService(STTService):
self._user_speaking = False
async def start(self, frame: StartFrame):
"""Start the segmented STT service and initialize audio buffer.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._audio_buffer_size_1s = self.sample_rate * 2
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames, handling VAD events and audio segmentation."""
await super().process_frame(frame, direction)
if isinstance(frame, UserStartedSpeakingFrame):
@@ -162,6 +232,15 @@ class SegmentedSTTService(STTService):
self._audio_buffer.clear()
async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
"""Process audio frames by buffering them for segmented transcription.
Continuously buffers audio, growing the buffer while user is speaking and
maintaining a small buffer when not speaking to account for VAD delay.
Args:
frame: The audio frame to process.
direction: The direction of frame processing.
"""
# If the user is speaking the audio buffer will keep growing.
self._audio_buffer += frame.audio

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

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base classes for Text-to-speech services."""
import asyncio
from abc import abstractmethod
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Tuple
@@ -43,6 +45,28 @@ from pipecat.utils.watchdog_queue import WatchdogQueue
class TTSService(AIService):
"""Base class for text-to-speech services.
Provides common functionality for TTS services including text aggregation,
filtering, audio generation, and frame management. Supports configurable
sentence aggregation, silence insertion, and frame processing control.
Args:
aggregate_sentences: Whether to aggregate text into sentences before synthesis.
push_text_frames: Whether to push TextFrames and LLMFullResponseEndFrames.
push_stop_frames: Whether to automatically push TTSStoppedFrames.
stop_frame_timeout_s: Idle time before pushing TTSStoppedFrame when push_stop_frames is True.
push_silence_after_stop: Whether to push silence audio after TTSStoppedFrame.
silence_time_s: Duration of silence to push when push_silence_after_stop is True.
pause_frame_processing: Whether to pause frame processing during audio generation.
sample_rate: Output sample rate for generated audio.
text_aggregator: Custom text aggregator for processing incoming text.
text_filters: Sequence of text filters to apply after aggregation.
text_filter: Single text filter (deprecated, use text_filters).
transport_destination: Destination for generated audio frames.
**kwargs: Additional arguments passed to the parent AIService.
"""
def __init__(
self,
*,
@@ -105,54 +129,113 @@ class TTSService(AIService):
@property
def sample_rate(self) -> int:
"""Get the current sample rate for audio output.
Returns:
The sample rate in Hz.
"""
return self._sample_rate
@property
def chunk_size(self) -> int:
"""This property indicates how much audio we download (from TTS services
"""Get the recommended chunk size for audio streaming.
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.
Returns:
The recommended chunk size in bytes.
"""
CHUNK_SECONDS = 0.5
return int(self.sample_rate * CHUNK_SECONDS * 2) # 2 bytes/sample
async def set_model(self, model: str):
"""Set the TTS model to use.
Args:
model: The name of the TTS model.
"""
self.set_model_name(model)
def set_voice(self, voice: str):
"""Set the voice for speech synthesis.
Args:
voice: The voice identifier or name.
"""
self._voice_id = voice
# Converts the text to audio.
@abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Run text-to-speech synthesis on the provided text.
This method must be implemented by subclasses to provide actual TTS functionality.
Args:
text: The text to synthesize into speech.
Yields:
Frame: Audio frames containing the synthesized speech.
"""
pass
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert a language to the service-specific language format.
Args:
language: The language to convert.
Returns:
The service-specific language identifier, or None if not supported.
"""
return Language(language)
async def update_setting(self, key: str, value: Any):
"""Update a service-specific setting.
Args:
key: The setting key to update.
value: The new value for the setting.
"""
pass
async def flush_audio(self):
"""Flush any buffered audio data."""
pass
async def start(self, frame: StartFrame):
"""Start the TTS service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
if self._push_stop_frames and not self._stop_frame_task:
self._stop_frame_task = self.create_task(self._stop_frame_handler())
async def stop(self, frame: EndFrame):
"""Stop the TTS service.
Args:
frame: The end frame.
"""
await super().stop(frame)
if self._stop_frame_task:
await self.cancel_task(self._stop_frame_task)
self._stop_frame_task = None
async def cancel(self, frame: CancelFrame):
"""Cancel the TTS service.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
if self._stop_frame_task:
await self.cancel_task(self._stop_frame_task)
@@ -176,9 +259,23 @@ class TTSService(AIService):
logger.warning(f"Unknown setting for TTS service: {key}")
async def say(self, text: str):
"""Immediately speak the provided text.
Args:
text: The text to speak.
"""
await self.queue_frame(TTSSpeakFrame(text))
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for text-to-speech conversion.
Handles TextFrames for synthesis, interruption frames, settings updates,
and various control frames.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if (
@@ -223,6 +320,12 @@ class TTSService(AIService):
await self.push_frame(frame, direction)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Push a frame downstream with TTS-specific handling.
Args:
frame: The frame to push.
direction: The direction to push the frame.
"""
if self._push_silence_after_stop and isinstance(frame, TTSStoppedFrame):
silence_num_bytes = int(self._silence_time_s * self.sample_rate * 2) # 16-bit
silence_frame = TTSAudioRawFrame(
@@ -321,10 +424,13 @@ class TTSService(AIService):
class WordTTSService(TTSService):
"""This is a base class for TTS services that support word timestamps. Word
timestamps are useful to synchronize audio with text of the spoken
"""Base class for TTS services that support word timestamps.
Word timestamps are useful to synchronize audio with text of the spoken
words. This way only the spoken words are added to the conversation context.
Args:
**kwargs: Additional arguments passed to the parent TTSService.
"""
def __init__(self, **kwargs):
@@ -334,29 +440,57 @@ class WordTTSService(TTSService):
self._llm_response_started: bool = False
def start_word_timestamps(self):
"""Start tracking word timestamps from the current time."""
if self._initial_word_timestamp == -1:
self._initial_word_timestamp = self.get_clock().get_time()
def reset_word_timestamps(self):
"""Reset word timestamp tracking."""
self._initial_word_timestamp = -1
async def add_word_timestamps(self, word_times: List[Tuple[str, float]]):
"""Add word timestamps to the processing queue.
Args:
word_times: List of (word, timestamp) tuples where timestamp is in seconds.
"""
for word, timestamp in word_times:
await self._words_queue.put((word, seconds_to_nanoseconds(timestamp)))
async def start(self, frame: StartFrame):
"""Start the word TTS service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._create_words_task()
async def stop(self, frame: EndFrame):
"""Stop the word TTS service.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._stop_words_task()
async def cancel(self, frame: CancelFrame):
"""Cancel the word TTS service.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._stop_words_task()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with word timestamp awareness.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, LLMFullResponseStartFrame):
@@ -403,15 +537,24 @@ class WordTTSService(TTSService):
class WebsocketTTSService(TTSService, WebsocketService):
"""This is a base class for websocket-based TTS services.
"""Base class for websocket-based TTS services.
If an error occurs with the websocket, an "on_connection_error" event will
be triggered:
Combines TTS functionality with websocket connectivity, providing automatic
error handling and reconnection capabilities.
@tts.event_handler("on_connection_error")
async def on_connection_error(tts: TTSService, error: str):
...
Args:
reconnect_on_error: Whether to automatically reconnect on websocket errors.
**kwargs: Additional arguments passed to parent classes.
Event handlers:
on_connection_error: Called when a websocket connection error occurs.
Example:
```python
@tts.event_handler("on_connection_error")
async def on_connection_error(tts: TTSService, error: str):
logger.error(f"TTS connection error: {error}")
```
"""
def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
@@ -425,10 +568,13 @@ class WebsocketTTSService(TTSService, WebsocketService):
class InterruptibleTTSService(WebsocketTTSService):
"""This is a base class for websocket-based TTS services that don't support
word timestamps and that don't offer a way to correlate the generated audio
to the requested text.
"""Websocket-based TTS service that handles interruptions without word timestamps.
Designed for TTS services that don't support word timestamps. Handles interruptions
by reconnecting the websocket when the bot is speaking and gets interrupted.
Args:
**kwargs: Additional arguments passed to the parent WebsocketTTSService.
"""
def __init__(self, **kwargs):
@@ -446,6 +592,12 @@ class InterruptibleTTSService(WebsocketTTSService):
await self._connect()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with bot speaking state tracking.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, BotStartedSpeakingFrame):
@@ -455,16 +607,23 @@ class InterruptibleTTSService(WebsocketTTSService):
class WebsocketWordTTSService(WordTTSService, WebsocketService):
"""This is a base class for websocket-based TTS services that support word
timestamps.
"""Base class for websocket-based TTS services that support word timestamps.
If an error occurs with the websocket a "on_connection_error" event will be
triggered:
Combines word timestamp functionality with websocket connectivity.
@tts.event_handler("on_connection_error")
async def on_connection_error(tts: TTSService, error: str):
...
Args:
reconnect_on_error: Whether to automatically reconnect on websocket errors.
**kwargs: Additional arguments passed to parent classes.
Event handlers:
on_connection_error: Called when a websocket connection error occurs.
Example:
```python
@tts.event_handler("on_connection_error")
async def on_connection_error(tts: TTSService, error: str):
logger.error(f"TTS connection error: {error}")
```
"""
def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
@@ -478,10 +637,13 @@ class WebsocketWordTTSService(WordTTSService, WebsocketService):
class InterruptibleWordTTSService(WebsocketWordTTSService):
"""This is a base class for websocket-based TTS services that support word
timestamps but don't offer a way to correlate the generated audio to the
requested text.
"""Websocket-based TTS service with word timestamps that handles interruptions.
For TTS services that support word timestamps but can't correlate generated
audio with requested text. Handles interruptions by reconnecting when needed.
Args:
**kwargs: Additional arguments passed to the parent WebsocketWordTTSService.
"""
def __init__(self, **kwargs):
@@ -499,6 +661,12 @@ class InterruptibleWordTTSService(WebsocketWordTTSService):
await self._connect()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with bot speaking state tracking.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, BotStartedSpeakingFrame):
@@ -508,7 +676,9 @@ class InterruptibleWordTTSService(WebsocketWordTTSService):
class AudioContextWordTTSService(WebsocketWordTTSService):
"""This is a base class for websocket-based TTS services that support word
"""Websocket-based TTS service with word timestamps and audio context management.
This is a base class for websocket-based TTS services that support word
timestamps and also allow correlating the generated audio with the requested
text.
@@ -520,6 +690,8 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
we requested audio for a context "A" and then audio for context "B", the
audio from context ID "A" will be played first.
Args:
**kwargs: Additional arguments passed to the parent WebsocketWordTTSService.
"""
def __init__(self, **kwargs):
@@ -528,13 +700,22 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
self._audio_context_task = None
async def create_audio_context(self, context_id: str):
"""Create a new audio context."""
"""Create a new audio context for grouping related audio.
Args:
context_id: Unique identifier for the audio context.
"""
await self._contexts_queue.put(context_id)
self._contexts[context_id] = asyncio.Queue()
logger.trace(f"{self} created audio context {context_id}")
async def append_to_audio_context(self, context_id: str, frame: TTSAudioRawFrame):
"""Append audio to an existing context."""
"""Append audio to an existing context.
Args:
context_id: The context to append audio to.
frame: The audio frame to append.
"""
if self.audio_context_available(context_id):
logger.trace(f"{self} appending audio {frame} to audio context {context_id}")
await self._contexts[context_id].put(frame)
@@ -542,7 +723,11 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
logger.warning(f"{self} unable to append audio to context {context_id}")
async def remove_audio_context(self, context_id: str):
"""Remove an existing audio context."""
"""Remove an existing audio context.
Args:
context_id: The context to remove.
"""
if self.audio_context_available(context_id):
# We just mark the audio context for deletion by appending
# None. Once we reach None while handling audio we know we can
@@ -553,14 +738,31 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
logger.warning(f"{self} unable to remove context {context_id}")
def audio_context_available(self, context_id: str) -> bool:
"""Checks whether the given audio context is registered."""
"""Check whether the given audio context is registered.
Args:
context_id: The context ID to check.
Returns:
True if the context exists and is available.
"""
return context_id in self._contexts
async def start(self, frame: StartFrame):
"""Start the audio context TTS service.
Args:
frame: The start frame containing initialization parameters.
"""
await super().start(frame)
self._create_audio_context_task()
async def stop(self, frame: EndFrame):
"""Stop the audio context TTS service.
Args:
frame: The end frame.
"""
await super().stop(frame)
if self._audio_context_task:
# Indicate no more audio contexts are available. this will end the
@@ -570,6 +772,11 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
self._audio_context_task = None
async def cancel(self, frame: CancelFrame):
"""Cancel the audio context TTS service.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._stop_audio_context_task()

39
src/pipecat/services/vision_service.py
View File

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Vision service implementation.
Provides base classes and implementations for computer vision services that can
analyze images and generate textual descriptions or answers to questions about
visual content.
"""
from abc import abstractmethod
from typing import AsyncGenerator
@@ -13,7 +20,15 @@ from pipecat.services.ai_service import AIService
class VisionService(AIService):
"""VisionService is a base class for vision services."""
"""Base class for vision services.
Provides common functionality for vision services that process images and
generate textual responses. Handles image frame processing and integrates
with the AI service infrastructure for metrics and lifecycle management.
Args:
**kwargs: Additional arguments passed to the parent AIService.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@@ -21,9 +36,31 @@ class VisionService(AIService):
@abstractmethod
async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]:
"""Process a vision image frame and generate results.
This method must be implemented by subclasses to provide actual computer
vision functionality such as image description, object detection, or
visual question answering.
Args:
frame: The vision image frame to process, containing image data.
Yields:
Frame: Frames containing the vision analysis results, typically TextFrame
objects with descriptions or answers.
"""
pass
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames, handling vision image frames for analysis.
Automatically processes VisionImageRawFrame objects by calling run_vision
and handles metrics tracking. Other frames are passed through unchanged.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
if isinstance(frame, VisionImageRawFrame):

66
src/pipecat/services/websocket_service.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base websocket service with automatic reconnection and error handling."""
import asyncio
from abc import ABC, abstractmethod
from typing import Awaitable, Callable, Optional
@@ -17,18 +19,26 @@ from pipecat.utils.network import exponential_backoff_time
class WebsocketService(ABC):
"""Base class for websocket-based services with reconnection logic."""
"""Base class for websocket-based services with automatic reconnection.
Provides websocket connection management, automatic reconnection with
exponential backoff, connection verification, and error handling.
Subclasses implement service-specific connection and message handling logic.
Args:
reconnect_on_error: Whether to automatically reconnect on connection errors.
**kwargs: Additional arguments (unused, for compatibility).
"""
def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
"""Initialize websocket attributes."""
self._websocket: Optional[websockets.WebSocketClientProtocol] = None
self._reconnect_on_error = reconnect_on_error
async def _verify_connection(self) -> bool:
"""Verify websocket connection is working.
"""Verify the websocket connection is active and responsive.
Returns:
bool: True if connection is verified working, False otherwise
True if connection is verified working, False otherwise.
"""
try:
if not self._websocket or self._websocket.closed:
@@ -40,13 +50,13 @@ class WebsocketService(ABC):
return False
async def _reconnect_websocket(self, attempt_number: int) -> bool:
"""Reconnect the websocket.
"""Reconnect the websocket with the current attempt number.
Args:
attempt_number: Current retry attempt number
attempt_number: Current retry attempt number for logging.
Returns:
bool: True if reconnection and verification successful, False otherwise
True if reconnection and verification successful, False otherwise.
"""
logger.warning(f"{self} reconnecting (attempt: {attempt_number})")
await self._disconnect_websocket()
@@ -54,10 +64,14 @@ class WebsocketService(ABC):
return await self._verify_connection()
async def _receive_task_handler(self, report_error: Callable[[ErrorFrame], Awaitable[None]]):
"""Handles WebSocket message receiving with automatic retry logic.
"""Handle websocket message receiving with automatic retry logic.
Continuously receives messages with automatic reconnection on errors.
Uses exponential backoff between retry attempts and reports fatal errors
after maximum retries are exhausted.
Args:
report_error: Callback to report errors
report_error: Callback function to report connection errors.
"""
retry_count = 0
MAX_RETRIES = 3
@@ -98,33 +112,45 @@ class WebsocketService(ABC):
@abstractmethod
async def _connect(self):
"""Implement service-specific connection logic. This function will
connect to the websocket via _connect_websocket() among other connection
logic."""
"""Connect to the service.
Implement service-specific connection logic including websocket connection
via _connect_websocket() and any additional setup required.
"""
pass
@abstractmethod
async def _disconnect(self):
"""Implement service-specific disconnection logic. This function will
disconnect to the websocket via _connect_websocket() among other
connection logic.
"""Disconnect from the service.
Implement service-specific disconnection logic including websocket
disconnection via _disconnect_websocket() and any cleanup required.
"""
pass
@abstractmethod
async def _connect_websocket(self):
"""Implement service-specific websocket connection logic. This function
should only connect to the websocket."""
"""Establish the websocket connection.
Implement the low-level websocket connection logic specific to the service.
Should only handle websocket connection, not additional service setup.
"""
pass
@abstractmethod
async def _disconnect_websocket(self):
"""Implement service-specific websocket disconnection logic. This
function should only disconnect from the websocket."""
"""Close the websocket connection.
Implement the low-level websocket disconnection logic specific to the service.
Should only handle websocket disconnection, not additional service cleanup.
"""
pass
@abstractmethod
async def _receive_messages(self):
"""Implement service-specific message receiving logic."""
"""Receive and process websocket messages.
Implement service-specific logic for receiving and handling messages
from the websocket connection. Called continuously by the receive task handler.
"""
pass