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

Merge pull request #2071 from pipecat-ai/mb/services-docstrings-update

Browse Source 
Add/update docstrings to LLM services
This commit is contained in:
Mark Backman
2025-06-26 14:42:32 -04:00
committed by GitHub
parent 7947497f7e 0e4d2be98c
commit 01040ddcdd
27 changed files with 2234 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Anthropic AI service integration for Pipecat.
This module provides LLM services and context management for Anthropic's Claude models,
including support for function calling, vision, and prompt caching features.
"""
import asyncio
import base64
import copy
@@ -59,27 +65,66 @@ except ModuleNotFoundError as e:
@dataclass
class AnthropicContextAggregatorPair:
"""Pair of context aggregators for Anthropic conversations.
Encapsulates both user and assistant context aggregators
to manage conversation flow and message formatting.
Parameters:
_user: The user context aggregator.
_assistant: The assistant context aggregator.
"""
_user: "AnthropicUserContextAggregator"
_assistant: "AnthropicAssistantContextAggregator"
def user(self) -> "AnthropicUserContextAggregator":
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> "AnthropicAssistantContextAggregator":
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class AnthropicLLMService(LLMService):
"""This class implements inference with Anthropic's AI models.
"""LLM service for Anthropic's Claude models.
Can provide a custom client via the `client` kwarg, allowing you to
use `AsyncAnthropicBedrock` and `AsyncAnthropicVertex` clients
Provides inference capabilities with Claude models including support for
function calling, vision processing, streaming responses, and prompt caching.
Can use custom clients like AsyncAnthropicBedrock and AsyncAnthropicVertex.
Args:
api_key: Anthropic API key for authentication.
model: Model name to use. Defaults to "claude-sonnet-4-20250514".
params: Optional model parameters for inference.
client: Optional custom Anthropic client instance.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the Anthropic one.
adapter_class = AnthropicLLMAdapter
class InputParams(BaseModel):
"""Input parameters for Anthropic model inference.
Parameters:
enable_prompt_caching_beta: Whether to enable beta prompt caching feature.
max_tokens: Maximum tokens to generate. Must be at least 1.
temperature: Sampling temperature between 0.0 and 1.0.
top_k: Top-k sampling parameter.
top_p: Top-p sampling parameter between 0.0 and 1.0.
extra: Additional parameters to pass to the API.
"""
enable_prompt_caching_beta: Optional[bool] = False
max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1)
temperature: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=1.0)
@@ -112,10 +157,20 @@ class AnthropicLLMService(LLMService):
}
def can_generate_metrics(self) -> bool:
"""Check if this service can generate usage metrics.
Returns:
True, as Anthropic provides detailed token usage metrics.
"""
return True
@property
def enable_prompt_caching_beta(self) -> bool:
"""Check if prompt caching beta feature is enabled.
Returns:
True if prompt caching is enabled.
"""
return self._enable_prompt_caching_beta
def create_context_aggregator(
@@ -125,22 +180,19 @@ class AnthropicLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AnthropicContextAggregatorPair:
"""Create an instance of AnthropicContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create Anthropic-specific context aggregators.
Creates a pair of context aggregators optimized for Anthropic'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.
user_params: User aggregator parameters.
assistant_params: Assistant aggregator parameters.
Returns:
AnthropicContextAggregatorPair: A pair of context aggregators, one
for the user and one for the assistant, encapsulated in an
AnthropicContextAggregatorPair.
A pair of context aggregators, one for the user and one for the assistant,
encapsulated in an AnthropicContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())
@@ -308,6 +360,15 @@ class AnthropicLLMService(LLMService):
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and route them appropriately.
Handles various frame types including context frames, message frames,
vision frames, and settings updates.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None
@@ -359,6 +420,19 @@ class AnthropicLLMService(LLMService):
class AnthropicLLMContext(OpenAILLMContext):
"""LLM context specialized for Anthropic's message format and features.
Extends OpenAILLMContext to handle Anthropic-specific features like
system messages, prompt caching, and message format conversions.
Manages conversation state and message history formatting.
Args:
messages: Initial list of conversation messages.
tools: Available function calling tools.
tool_choice: Tool selection preference.
system: System message content.
"""
def __init__(
self,
messages: Optional[List[dict]] = None,
@@ -379,6 +453,16 @@ class AnthropicLLMContext(OpenAILLMContext):
@staticmethod
def upgrade_to_anthropic(obj: OpenAILLMContext) -> "AnthropicLLMContext":
"""Upgrade an OpenAI context to Anthropic format.
Converts message format and restructures content for Anthropic compatibility.
Args:
obj: The OpenAI context to upgrade.
Returns:
The upgraded Anthropic context.
"""
logger.debug(f"Upgrading to Anthropic: {obj}")
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AnthropicLLMContext):
obj.__class__ = AnthropicLLMContext
@@ -387,6 +471,14 @@ class AnthropicLLMContext(OpenAILLMContext):
@classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
"""Create Anthropic context from OpenAI context.
Args:
openai_context: The OpenAI context to convert.
Returns:
New Anthropic context with converted messages.
"""
self = cls(
messages=openai_context.messages,
tools=openai_context.tools,
@@ -398,12 +490,28 @@ class AnthropicLLMContext(OpenAILLMContext):
@classmethod
def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext":
"""Create context from a list of messages.
Args:
messages: List of conversation messages.
Returns:
New Anthropic context with the provided messages.
"""
self = cls(messages=messages)
self._restructure_from_openai_messages()
return self
@classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext":
"""Create context from a vision image frame.
Args:
frame: The vision image frame to process.
Returns:
New Anthropic context with the image message.
"""
context = cls()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text
@@ -411,11 +519,15 @@ class AnthropicLLMContext(OpenAILLMContext):
return context
def set_messages(self, messages: List):
"""Set the messages list and reset cache tracking.
Args:
messages: New list of messages to set.
"""
self.turns_above_cache_threshold = 0
self._messages[:] = messages
self._restructure_from_openai_messages()
# convert a message in Anthropic format into one or more messages in OpenAI format
def to_standard_messages(self, obj):
"""Convert Anthropic message format to standard structured format.
@@ -556,6 +668,17 @@ class AnthropicLLMContext(OpenAILLMContext):
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add an image message to the context.
Converts the image to base64 JPEG format and adds it as a user message
with optional accompanying text.
Args:
format: The image format (e.g., 'RGB', 'RGBA').
size: Image dimensions as (width, height).
image: Raw image bytes.
text: Optional text to accompany the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
@@ -576,6 +699,14 @@ class AnthropicLLMContext(OpenAILLMContext):
self.add_message({"role": "user", "content": content})
def add_message(self, message):
"""Add a message to the context, merging with previous message if same role.
Anthropic requires alternating roles, so consecutive messages from the same
role are merged together.
Args:
message: The message to add to the context.
"""
try:
if self.messages:
# Anthropic requires that roles alternate. If this message's role is the same as the
@@ -601,6 +732,14 @@ class AnthropicLLMContext(OpenAILLMContext):
logger.error(f"Error adding message: {e}")
def get_messages_with_cache_control_markers(self) -> List[dict]:
"""Get messages with prompt caching markers applied.
Adds cache control markers to appropriate messages based on the
number of turns above the cache threshold.
Returns:
List of messages with cache control markers added.
"""
try:
messages = copy.deepcopy(self.messages)
if self.turns_above_cache_threshold >= 1 and messages[-1]["role"] == "user":
@@ -668,12 +807,26 @@ class AnthropicLLMContext(OpenAILLMContext):
message["content"] = [{"type": "text", "text": "(empty)"}]
def get_messages_for_persistent_storage(self):
"""Get messages formatted for persistent storage.
Includes system message at the beginning if present.
Returns:
List of messages suitable for storage.
"""
messages = super().get_messages_for_persistent_storage()
if self.system:
messages.insert(0, {"role": "system", "content": self.system})
return messages
def get_messages_for_logging(self) -> str:
"""Get messages formatted for logging with sensitive data redacted.
Replaces image data with placeholder text for cleaner logs.
Returns:
JSON string representation of messages for logging.
"""
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
@@ -687,6 +840,12 @@ class AnthropicLLMContext(OpenAILLMContext):
class AnthropicUserContextAggregator(LLMUserContextAggregator):
"""Anthropic-specific user context aggregator.
Handles aggregation of user messages for Anthropic LLM services.
Inherits all functionality from the base LLMUserContextAggregator.
"""
pass
@@ -701,7 +860,20 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
"""Context aggregator for assistant messages in Anthropic conversations.
Handles function call lifecycle management including in-progress tracking,
result handling, and cancellation for Anthropic's tool use format.
"""
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle a function call that is starting.
Creates tool use message and placeholder tool result for tracking.
Args:
frame: Frame containing function call details.
"""
assistant_message = {"role": "assistant", "content": []}
assistant_message["content"].append(
{
@@ -726,6 +898,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle the result of a completed function call.
Updates the tool result with actual return value or completion status.
Args:
frame: Frame containing 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)
@@ -735,6 +914,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle cancellation of a function call.
Updates the tool result to indicate cancellation.
Args:
frame: Frame containing function call cancellation details.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
@@ -753,6 +939,14 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
content["content"] = result
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle a user image frame with function call context.
Marks the associated function call as completed and adds the image
to the conversation context.
Args:
frame: User image frame with request context.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)

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

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""AWS Bedrock integration for Large Language Model services.
This module provides AWS Bedrock LLM service implementation with support for
Amazon Nova and Anthropic Claude models, including vision capabilities and
function calling.
"""
import asyncio
import base64
import copy
@@ -61,17 +68,50 @@ except ModuleNotFoundError as e:
@dataclass
class AWSBedrockContextAggregatorPair:
"""Container for AWS Bedrock context aggregators.
Provides convenient access to both user and assistant context aggregators
for AWS Bedrock LLM operations.
Parameters:
_user: The user context aggregator instance.
_assistant: The assistant context aggregator instance.
"""
_user: "AWSBedrockUserContextAggregator"
_assistant: "AWSBedrockAssistantContextAggregator"
def user(self) -> "AWSBedrockUserContextAggregator":
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> "AWSBedrockAssistantContextAggregator":
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class AWSBedrockLLMContext(OpenAILLMContext):
"""AWS Bedrock-specific LLM context implementation.
Extends OpenAI LLM context to handle AWS Bedrock's specific message format
and system message handling. Manages conversion between OpenAI and Bedrock
message formats.
Args:
messages: List of conversation messages in OpenAI format.
tools: List of available function calling tools.
tool_choice: Tool selection strategy or specific tool choice.
system: System message content for AWS Bedrock.
"""
def __init__(
self,
messages: Optional[List[dict]] = None,
@@ -85,6 +125,14 @@ class AWSBedrockLLMContext(OpenAILLMContext):
@staticmethod
def upgrade_to_bedrock(obj: OpenAILLMContext) -> "AWSBedrockLLMContext":
"""Upgrade an OpenAI LLM context to AWS Bedrock format.
Args:
obj: The OpenAI LLM context to upgrade.
Returns:
The upgraded AWS Bedrock LLM context.
"""
logger.debug(f"Upgrading to AWS Bedrock: {obj}")
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AWSBedrockLLMContext):
obj.__class__ = AWSBedrockLLMContext
@@ -95,6 +143,14 @@ class AWSBedrockLLMContext(OpenAILLMContext):
@classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
"""Create AWS Bedrock context from OpenAI context.
Args:
openai_context: The OpenAI LLM context to convert.
Returns:
New AWS Bedrock LLM context instance.
"""
self = cls(
messages=openai_context.messages,
tools=openai_context.tools,
@@ -106,12 +162,28 @@ class AWSBedrockLLMContext(OpenAILLMContext):
@classmethod
def from_messages(cls, messages: List[dict]) -> "AWSBedrockLLMContext":
"""Create AWS Bedrock context from message list.
Args:
messages: List of messages in OpenAI format.
Returns:
New AWS Bedrock LLM context instance.
"""
self = cls(messages=messages)
self._restructure_from_openai_messages()
return self
@classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "AWSBedrockLLMContext":
"""Create AWS Bedrock context from vision image frame.
Args:
frame: The vision image frame to convert.
Returns:
New AWS Bedrock LLM context instance.
"""
context = cls()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text
@@ -119,10 +191,14 @@ class AWSBedrockLLMContext(OpenAILLMContext):
return context
def set_messages(self, messages: List):
"""Set the messages list and restructure for Bedrock format.
Args:
messages: List of messages to set.
"""
self._messages[:] = messages
self._restructure_from_openai_messages()
# convert a message in AWS Bedrock format into one or more messages in OpenAI format
def to_standard_messages(self, obj):
"""Convert AWS Bedrock message format to standard structured format.
@@ -295,6 +371,14 @@ class AWSBedrockLLMContext(OpenAILLMContext):
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add an image message to the context.
Args:
format: The image format (e.g., 'RGB', 'RGBA').
size: The image dimensions as (width, height).
image: The raw image data as bytes.
text: Optional text to accompany the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
@@ -306,6 +390,14 @@ class AWSBedrockLLMContext(OpenAILLMContext):
self.add_message({"role": "user", "content": content})
def add_message(self, message):
"""Add a message to the context, merging with previous message if same role.
AWS Bedrock requires alternating roles, so consecutive messages from the
same role are merged together.
Args:
message: The message to add to the context.
"""
try:
if self.messages:
# AWS Bedrock requires that roles alternate. If this message's
@@ -330,10 +422,10 @@ class AWSBedrockLLMContext(OpenAILLMContext):
logger.error(f"Error adding message: {e}")
def _restructure_from_bedrock_messages(self):
"""Restructure messages in AWS Bedrock format by handling system
messages, merging consecutive messages with the same role, and ensuring
proper content formatting.
"""Restructure messages in AWS Bedrock format.
Handles system messages, merging consecutive messages with the same role,
and ensuring proper content formatting.
"""
# Handle system message if present at the beginning
if self.messages and self.messages[0]["role"] == "system":
@@ -416,12 +508,22 @@ class AWSBedrockLLMContext(OpenAILLMContext):
message["content"] = [{"type": "text", "text": "(empty)"}]
def get_messages_for_persistent_storage(self):
"""Get messages formatted for persistent storage.
Returns:
List of messages including system message if present.
"""
messages = super().get_messages_for_persistent_storage()
if self.system:
messages.insert(0, {"role": "system", "content": self.system})
return messages
def get_messages_for_logging(self) -> str:
"""Get messages formatted for logging with sensitive data redacted.
Returns:
JSON string representation of messages with image data redacted.
"""
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
@@ -435,11 +537,36 @@ class AWSBedrockLLMContext(OpenAILLMContext):
class AWSBedrockUserContextAggregator(LLMUserContextAggregator):
"""User context aggregator for AWS Bedrock LLM service.
Handles aggregation of user messages and frames for AWS Bedrock format.
Inherits all functionality from the base LLM user context aggregator.
Args:
context: The LLM context to aggregate messages into.
params: Configuration parameters for the aggregator.
"""
pass
class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
"""Assistant context aggregator for AWS Bedrock LLM service.
Handles aggregation of assistant responses and function calls for AWS Bedrock
format, including tool use and tool result handling.
Args:
context: The LLM context to aggregate messages into.
params: Configuration parameters for the aggregator.
"""
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle function call in progress frame.
Args:
frame: The function call in progress frame to handle.
"""
# Format tool use according to AWS Bedrock API
self._context.add_message(
{
@@ -470,6 +597,11 @@ class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle function call result frame.
Args:
frame: The function call result frame to handle.
"""
if frame.result:
result = json.dumps(frame.result)
await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
@@ -479,6 +611,11 @@ class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle function call cancel frame.
Args:
frame: The function call cancel frame to handle.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
@@ -497,6 +634,11 @@ class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
content["toolResult"]["content"] = [{"text": result}]
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle user image frame.
Args:
frame: The user image frame to handle.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
@@ -509,18 +651,38 @@ class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
class AWSBedrockLLMService(LLMService):
"""This class implements inference with AWS Bedrock models including Amazon
Nova and Anthropic Claude.
"""AWS Bedrock Large Language Model service implementation.
Requires AWS credentials to be configured in the environment or through
boto3 configuration.
Provides inference capabilities for AWS Bedrock models including Amazon Nova
and Anthropic Claude. Supports streaming responses, function calling, and
vision capabilities.
Args:
model: The AWS Bedrock model identifier to use.
aws_access_key: AWS access key ID. If None, uses default credentials.
aws_secret_key: AWS secret access key. If None, uses default credentials.
aws_session_token: AWS session token for temporary credentials.
aws_region: AWS region for the Bedrock service.
params: Model parameters and configuration.
client_config: Custom boto3 client configuration.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the Anthropic one.
adapter_class = AWSBedrockLLMAdapter
class InputParams(BaseModel):
"""Input parameters for AWS Bedrock LLM service.
Parameters:
max_tokens: Maximum number of tokens to generate.
temperature: Sampling temperature between 0.0 and 1.0.
top_p: Nucleus sampling parameter between 0.0 and 1.0.
stop_sequences: List of strings that stop generation.
latency: Performance mode - "standard" or "optimized".
additional_model_request_fields: Additional model-specific parameters.
"""
max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1)
temperature: Optional[float] = Field(default_factory=lambda: 0.7, ge=0.0, le=1.0)
top_p: Optional[float] = Field(default_factory=lambda: 0.999, ge=0.0, le=1.0)
@@ -573,6 +735,11 @@ class AWSBedrockLLMService(LLMService):
logger.info(f"Using AWS Bedrock model: {model}")
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True if metrics generation is supported.
"""
return True
def create_context_aggregator(
@@ -582,21 +749,21 @@ class AWSBedrockLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AWSBedrockContextAggregatorPair:
"""Create an instance of AWSBedrockContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create AWS Bedrock-specific context aggregators.
Creates a pair of context aggregators optimized for AWS Bedrocks'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:
AWSBedrockContextAggregatorPair: A pair of context aggregators, one
for the user and one for the assistant, encapsulated in an
AWSBedrockContextAggregatorPair: A pair of context aggregators, one for
the user and one for the assistant, encapsulated in an
AWSBedrockContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())
@@ -792,6 +959,12 @@ class AWSBedrockLLMService(LLMService):
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and handle LLM-specific frame types.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None

97
src/pipecat/services/aws_nova_sonic/aws.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""AWS Nova Sonic LLM service implementation for Pipecat AI framework.
This module provides a speech-to-speech LLM service using AWS Nova Sonic, which supports
bidirectional audio streaming, text generation, and function calling capabilities.
"""
import asyncio
import base64
import json
@@ -83,22 +89,37 @@ except ModuleNotFoundError as e:
class AWSNovaSonicUnhandledFunctionException(Exception):
"""Exception raised when the LLM attempts to call an unregistered function."""
pass
class ContentType(Enum):
"""Content types supported by AWS Nova Sonic."""
AUDIO = "AUDIO"
TEXT = "TEXT"
TOOL = "TOOL"
class TextStage(Enum):
"""Text generation stages in AWS Nova Sonic responses."""
FINAL = "FINAL" # what has been said
SPECULATIVE = "SPECULATIVE" # what's planned to be said
@dataclass
class CurrentContent:
"""Represents content currently being received from AWS Nova Sonic.
Parameters:
type: The type of content (audio, text, or tool).
role: The role generating the content (user, assistant, etc.).
text_stage: The stage of text generation (final or speculative).
text_content: The actual text content if applicable.
"""
type: ContentType
role: Role
text_stage: TextStage # None if not text
@@ -115,6 +136,20 @@ class CurrentContent:
class Params(BaseModel):
"""Configuration parameters for AWS Nova Sonic.
Attributes:
input_sample_rate: Audio input sample rate in Hz.
input_sample_size: Audio input sample size in bits.
input_channel_count: Number of input audio channels.
output_sample_rate: Audio output sample rate in Hz.
output_sample_size: Audio output sample size in bits.
output_channel_count: Number of output audio channels.
max_tokens: Maximum number of tokens to generate.
top_p: Nucleus sampling parameter.
temperature: Sampling temperature for text generation.
"""
# Audio input
input_sample_rate: Optional[int] = Field(default=16000)
input_sample_size: Optional[int] = Field(default=16)
@@ -132,6 +167,24 @@ class Params(BaseModel):
class AWSNovaSonicLLMService(LLMService):
"""AWS Nova Sonic speech-to-speech LLM service.
Provides bidirectional audio streaming, real-time transcription, text generation,
and function calling capabilities using AWS Nova Sonic model.
Args:
secret_access_key: AWS secret access key for authentication.
access_key_id: AWS access key ID for authentication.
region: AWS region where the service is hosted.
model: Model identifier. Defaults to "amazon.nova-sonic-v1:0".
voice_id: Voice ID for speech synthesis. Options: matthew, tiffany, amy.
params: Model parameters for audio configuration and inference.
system_instruction: System-level instruction for the model.
tools: Available tools/functions for the model to use.
send_transcription_frames: Whether to emit transcription frames.
**kwargs: Additional arguments passed to the parent LLMService.
"""
# Override the default adapter to use the AWSNovaSonicLLMAdapter one
adapter_class = AWSNovaSonicLLMAdapter
@@ -188,16 +241,31 @@ class AWSNovaSonicLLMService(LLMService):
#
async def start(self, frame: StartFrame):
"""Start the service and initiate connection to AWS Nova Sonic.
Args:
frame: The start frame triggering service initialization.
"""
await super().start(frame)
self._wants_connection = True
await self._start_connecting()
async def stop(self, frame: EndFrame):
"""Stop the service and close connections.
Args:
frame: The end frame triggering service shutdown.
"""
await super().stop(frame)
self._wants_connection = False
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the service and close connections.
Args:
frame: The cancel frame triggering service cancellation.
"""
await super().cancel(frame)
self._wants_connection = False
await self._disconnect()
@@ -207,6 +275,11 @@ class AWSNovaSonicLLMService(LLMService):
#
async def reset_conversation(self):
"""Reset the conversation state while preserving context.
Handles bot stopped speaking event, disconnects from the service,
and reconnects with the preserved context.
"""
logger.debug("Resetting conversation")
await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=False)
@@ -222,6 +295,12 @@ class AWSNovaSonicLLMService(LLMService):
#
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and handle service-specific logic.
Args:
frame: The frame to process.
direction: The direction the frame is traveling.
"""
await super().process_frame(frame, direction)
if isinstance(frame, OpenAILLMContextFrame):
@@ -960,6 +1039,16 @@ class AWSNovaSonicLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AWSNovaSonicContextAggregatorPair:
"""Create context aggregator pair for managing conversation context.
Args:
context: The OpenAI LLM context to upgrade.
user_params: Parameters for the user context aggregator.
assistant_params: Parameters for the assistant context aggregator.
Returns:
A pair of user and assistant context aggregators.
"""
context.set_llm_adapter(self.get_llm_adapter())
user = AWSNovaSonicUserContextAggregator(context=context, params=user_params)
@@ -978,6 +1067,14 @@ class AWSNovaSonicLLMService(LLMService):
)
async def trigger_assistant_response(self):
"""Trigger an assistant response by sending audio cue.
Sends a pre-recorded "ready" audio trigger to prompt the assistant
to start speaking. This is useful for controlling conversation flow.
Returns:
False if already triggering a response, True otherwise.
"""
if self._triggering_assistant_response:
return False

131
src/pipecat/services/aws_nova_sonic/context.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Context management for AWS Nova Sonic LLM service.
This module provides specialized context aggregators and message handling for AWS Nova Sonic,
including conversation history management and role-specific message processing.
"""
import copy
from dataclasses import dataclass, field
from enum import Enum
@@ -35,6 +41,8 @@ from pipecat.services.openai.llm import (
class Role(Enum):
"""Roles supported in AWS Nova Sonic conversations."""
SYSTEM = "SYSTEM"
USER = "USER"
ASSISTANT = "ASSISTANT"
@@ -43,17 +51,42 @@ class Role(Enum):
@dataclass
class AWSNovaSonicConversationHistoryMessage:
"""A single message in AWS Nova Sonic conversation history.
Parameters:
role: The role of the message sender (USER or ASSISTANT only).
text: The text content of the message.
"""
role: Role # only USER and ASSISTANT
text: str
@dataclass
class AWSNovaSonicConversationHistory:
"""Complete conversation history for AWS Nova Sonic initialization.
Parameters:
system_instruction: System-level instruction for the conversation.
messages: List of conversation messages between user and assistant.
"""
system_instruction: str = None
messages: list[AWSNovaSonicConversationHistoryMessage] = field(default_factory=list)
class AWSNovaSonicLLMContext(OpenAILLMContext):
"""Specialized LLM context for AWS Nova Sonic service.
Extends OpenAI context with Nova Sonic-specific message handling,
conversation history management, and text buffering capabilities.
Args:
messages: Initial messages for the context.
tools: Available tools for the context.
**kwargs: Additional arguments passed to parent class.
"""
def __init__(self, messages=None, tools=None, **kwargs):
super().__init__(messages=messages, tools=tools, **kwargs)
self.__setup_local()
@@ -67,6 +100,15 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
def upgrade_to_nova_sonic(
obj: OpenAILLMContext, system_instruction: str
) -> "AWSNovaSonicLLMContext":
"""Upgrade an OpenAI context to AWS Nova Sonic context.
Args:
obj: The OpenAI context to upgrade.
system_instruction: System instruction for the context.
Returns:
The upgraded AWS Nova Sonic context.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AWSNovaSonicLLMContext):
obj.__class__ = AWSNovaSonicLLMContext
obj.__setup_local(system_instruction)
@@ -74,6 +116,14 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
# NOTE: this method has the side-effect of updating _system_instruction from messages
def get_messages_for_initializing_history(self) -> AWSNovaSonicConversationHistory:
"""Get conversation history for initializing AWS Nova Sonic session.
Processes stored messages and extracts system instruction and conversation
history in the format expected by AWS Nova Sonic.
Returns:
Formatted conversation history with system instruction and messages.
"""
history = AWSNovaSonicConversationHistory(system_instruction=self._system_instruction)
# Bail if there are no messages
@@ -103,6 +153,11 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
return history
def get_messages_for_persistent_storage(self):
"""Get messages formatted for persistent storage.
Returns:
List of messages including system instruction if present.
"""
messages = super().get_messages_for_persistent_storage()
# If we have a system instruction and messages doesn't already contain it, add it
if self._system_instruction and not (messages and messages[0].get("role") == "system"):
@@ -110,6 +165,14 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
return messages
def from_standard_message(self, message) -> AWSNovaSonicConversationHistoryMessage:
"""Convert standard message format to Nova Sonic format.
Args:
message: Standard message dictionary to convert.
Returns:
Nova Sonic conversation history message, or None if not convertible.
"""
role = message.get("role")
if message.get("role") == "user" or message.get("role") == "assistant":
content = message.get("content")
@@ -131,10 +194,20 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
# Sonic conversation history
def buffer_user_text(self, text):
"""Buffer user text for later flushing to context.
Args:
text: User text to buffer.
"""
self._user_text += f" {text}" if self._user_text else text
# logger.debug(f"User text buffered: {self._user_text}")
def flush_aggregated_user_text(self) -> str:
"""Flush buffered user text to context as a complete message.
Returns:
The flushed user text, or empty string if no text was buffered.
"""
if not self._user_text:
return ""
user_text = self._user_text
@@ -148,10 +221,16 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
return user_text
def buffer_assistant_text(self, text):
"""Buffer assistant text for later flushing to context.
Args:
text: Assistant text to buffer.
"""
self._assistant_text += text
# logger.debug(f"Assistant text buffered: {self._assistant_text}")
def flush_aggregated_assistant_text(self):
"""Flush buffered assistant text to context as a complete message."""
if not self._assistant_text:
return
message = {
@@ -165,13 +244,31 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
@dataclass
class AWSNovaSonicMessagesUpdateFrame(DataFrame):
"""Frame containing updated AWS Nova Sonic context.
Parameters:
context: The updated AWS Nova Sonic LLM context.
"""
context: AWSNovaSonicLLMContext
class AWSNovaSonicUserContextAggregator(OpenAIUserContextAggregator):
"""Context aggregator for user messages in AWS Nova Sonic conversations.
Extends the OpenAI user context aggregator to emit Nova Sonic-specific
context update frames.
"""
async def process_frame(
self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
):
"""Process frames and emit Nova Sonic-specific context updates.
Args:
frame: The frame to process.
direction: The direction the frame is traveling.
"""
await super().process_frame(frame, direction)
# Parent does not push LLMMessagesUpdateFrame
@@ -180,7 +277,19 @@ class AWSNovaSonicUserContextAggregator(OpenAIUserContextAggregator):
class AWSNovaSonicAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Context aggregator for assistant messages in AWS Nova Sonic conversations.
Provides specialized handling for assistant responses and function calls
in AWS Nova Sonic context, with custom frame processing logic.
"""
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with Nova Sonic-specific logic.
Args:
frame: The frame to process.
direction: The direction the frame is traveling.
"""
# HACK: For now, disable the context aggregator by making it just pass through all frames
# that the parent handles (except the function call stuff, which we still need).
# For an explanation of this hack, see
@@ -205,6 +314,11 @@ class AWSNovaSonicAssistantContextAggregator(OpenAIAssistantContextAggregator):
await super().process_frame(frame, direction)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle function call results for AWS Nova Sonic.
Args:
frame: The function call result frame to handle.
"""
await super().handle_function_call_result(frame)
# The standard function callback code path pushes the FunctionCallResultFrame from the LLM
@@ -217,11 +331,28 @@ class AWSNovaSonicAssistantContextAggregator(OpenAIAssistantContextAggregator):
@dataclass
class AWSNovaSonicContextAggregatorPair:
"""Pair of user and assistant context aggregators for AWS Nova Sonic.
Parameters:
_user: The user context aggregator.
_assistant: The assistant context aggregator.
"""
_user: AWSNovaSonicUserContextAggregator
_assistant: AWSNovaSonicAssistantContextAggregator
def user(self) -> AWSNovaSonicUserContextAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> AWSNovaSonicAssistantContextAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant

11
src/pipecat/services/aws_nova_sonic/frames.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Custom frames for AWS Nova Sonic LLM service."""
from dataclasses import dataclass
from pipecat.frames.frames import DataFrame, FunctionCallResultFrame
@@ -11,4 +13,13 @@ from pipecat.frames.frames import DataFrame, FunctionCallResultFrame
@dataclass
class AWSNovaSonicFunctionCallResultFrame(DataFrame):
"""Frame containing function call result for AWS Nova Sonic processing.
This frame wraps a standard function call result frame to enable
AWS Nova Sonic-specific handling and context updates.
Parameters:
result_frame: The underlying function call result frame.
"""
result_frame: FunctionCallResultFrame

23
src/pipecat/services/azure/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Azure OpenAI service implementation for the Pipecat AI framework."""
from loguru import logger
from openai import AsyncAzureOpenAI
@@ -17,11 +19,11 @@ class AzureLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Azure OpenAI
endpoint (str): The Azure endpoint URL
model (str): The model identifier to use
api_version (str, optional): Azure API version. Defaults to "2024-09-01-preview"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Azure OpenAI.
endpoint: The Azure endpoint URL.
model: The model identifier to use.
api_version: Azure API version. Defaults to "2024-09-01-preview".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -40,7 +42,16 @@ class AzureLLMService(OpenAILLMService):
super().__init__(api_key=api_key, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Azure OpenAI endpoint."""
"""Create OpenAI-compatible client for Azure OpenAI endpoint.
Args:
api_key: API key for authentication. Uses instance key if None.
base_url: Base URL for the client. Ignored for Azure implementation.
**kwargs: Additional keyword arguments. Ignored for Azure implementation.
Returns:
AsyncAzureOpenAI: Configured Azure OpenAI client instance.
"""
logger.debug(f"Creating Azure OpenAI client with endpoint {self._endpoint}")
return AsyncAzureOpenAI(
api_key=api_key,

33
src/pipecat/services/cerebras/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Cerebras LLM service implementation using OpenAI-compatible interface."""
from typing import List
from loguru import logger
@@ -21,10 +23,10 @@ class CerebrasLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Cerebras's API
base_url (str, optional): The base URL for Cerebras API. Defaults to "https://api.cerebras.ai/v1"
model (str, optional): The model identifier to use. Defaults to "llama-3.3-70b"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Cerebras's API.
base_url: The base URL for Cerebras API. Defaults to "https://api.cerebras.ai/v1".
model: The model identifier to use. Defaults to "llama-3.3-70b".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -38,7 +40,16 @@ class CerebrasLLMService(OpenAILLMService):
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Cerebras API endpoint."""
"""Create OpenAI-compatible client for Cerebras API endpoint.
Args:
api_key: The API key for authentication. If None, uses instance key.
base_url: The base URL for the API. If None, uses instance URL.
**kwargs: Additional arguments passed to the client constructor.
Returns:
An OpenAI-compatible client configured for Cerebras API.
"""
logger.debug(f"Creating Cerebras client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@@ -48,14 +59,14 @@ class CerebrasLLMService(OpenAILLMService):
"""Create a streaming chat completion using Cerebras's API.
Args:
context (OpenAILLMContext): The context object containing tools configuration
and other settings for the chat completion.
messages (List[ChatCompletionMessageParam]): The list of messages comprising
the conversation history and current request.
context: The context object containing tools configuration
and other settings for the chat completion.
messages: The list of messages comprising
the conversation history and current request.
Returns:
AsyncStream[ChatCompletionChunk]: A streaming response of chat completion
chunks that can be processed asynchronously.
A streaming response of chat completion
chunks that can be processed asynchronously.
"""
params = {
"model": self.model_name,

34
src/pipecat/services/deepseek/llm.py
View File

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""DeepSeek LLM service implementation using OpenAI-compatible interface."""
from typing import List
@@ -22,10 +23,10 @@ class DeepSeekLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing DeepSeek's API
base_url (str, optional): The base URL for DeepSeek API. Defaults to "https://api.deepseek.com/v1"
model (str, optional): The model identifier to use. Defaults to "deepseek-chat"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing DeepSeek's API.
base_url: The base URL for DeepSeek API. Defaults to "https://api.deepseek.com/v1".
model: The model identifier to use. Defaults to "deepseek-chat".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -39,24 +40,33 @@ class DeepSeekLLMService(OpenAILLMService):
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for DeepSeek API endpoint."""
"""Create OpenAI-compatible client for DeepSeek API endpoint.
Args:
api_key: The API key for authentication. If None, uses instance default.
base_url: The base URL for the API. If None, uses instance default.
**kwargs: Additional keyword arguments for client configuration.
Returns:
An OpenAI-compatible client configured for DeepSeek's API.
"""
logger.debug(f"Creating DeepSeek client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:
"""Create a streaming chat completion using Cerebras's API.
"""Create a streaming chat completion using DeepSeek's API.
Args:
context (OpenAILLMContext): The context object containing tools configuration
and other settings for the chat completion.
messages (List[ChatCompletionMessageParam]): The list of messages comprising
the conversation history and current request.
context: The context object containing tools configuration
and other settings for the chat completion.
messages: The list of messages comprising the conversation
history and current request.
Returns:
AsyncStream[ChatCompletionChunk]: A streaming response of chat completion
chunks that can be processed asynchronously.
A streaming response of chat completion chunks that can be
processed asynchronously.
"""
params = {
"model": self.model_name,

30
src/pipecat/services/fireworks/llm.py
View File

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Fireworks AI service implementation using OpenAI-compatible interface."""
from typing import List
@@ -21,10 +22,10 @@ class FireworksLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Fireworks AI
model (str, optional): The model identifier to use. Defaults to "accounts/fireworks/models/firefunction-v2"
base_url (str, optional): The base URL for Fireworks API. Defaults to "https://api.fireworks.ai/inference/v1"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Fireworks AI.
model: The model identifier to use. Defaults to "accounts/fireworks/models/firefunction-v2".
base_url: The base URL for Fireworks API. Defaults to "https://api.fireworks.ai/inference/v1".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -38,7 +39,16 @@ class FireworksLLMService(OpenAILLMService):
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Fireworks API endpoint."""
"""Create OpenAI-compatible client for Fireworks API endpoint.
Args:
api_key: API key for authentication. If None, uses instance default.
base_url: Base URL for the API. If None, uses instance default.
**kwargs: Additional arguments passed to the client constructor.
Returns:
Configured OpenAI client instance for Fireworks API.
"""
logger.debug(f"Creating Fireworks client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@@ -47,7 +57,15 @@ class FireworksLLMService(OpenAILLMService):
):
"""Get chat completions from Fireworks API.
Removes OpenAI-specific parameters not supported by Fireworks.
Removes OpenAI-specific parameters not supported by Fireworks and
configures the request with Fireworks-compatible settings.
Args:
context: The OpenAI LLM context containing tools and settings.
messages: List of chat completion message parameters.
Returns:
Async generator yielding chat completion chunks from Fireworks API.
"""
params = {
"model": self.model_name,

224
src/pipecat/services/gemini_multimodal_live/events.py
View File

@@ -3,7 +3,8 @@
#
# SPDX-License-Identifier: BSD 2-Clause License
#
#
"""Event models and utilities for Google Gemini Multimodal Live API."""
import base64
import io
@@ -22,16 +23,37 @@ from pipecat.frames.frames import ImageRawFrame
class MediaChunk(BaseModel):
"""Represents a chunk of media data for transmission.
Parameters:
mimeType: MIME type of the media content.
data: Base64-encoded media data.
"""
mimeType: str
data: str
class ContentPart(BaseModel):
"""Represents a part of content that can contain text or media.
Parameters:
text: Text content. Defaults to None.
inlineData: Inline media data. Defaults to None.
"""
text: Optional[str] = Field(default=None, validate_default=False)
inlineData: Optional[MediaChunk] = Field(default=None, validate_default=False)
class Turn(BaseModel):
"""Represents a conversational turn in the dialogue.
Parameters:
role: The role of the speaker, either "user" or "model". Defaults to "user".
parts: List of content parts that make up the turn.
"""
role: Literal["user", "model"] = "user"
parts: List[ContentPart]
@@ -53,7 +75,15 @@ class EndSensitivity(str, Enum):
class AutomaticActivityDetection(BaseModel):
"""Configures automatic detection of activity."""
"""Configures automatic detection of voice activity.
Parameters:
disabled: Whether automatic activity detection is disabled. Defaults to None.
start_of_speech_sensitivity: Sensitivity for detecting speech start. Defaults to None.
prefix_padding_ms: Padding before speech start in milliseconds. Defaults to None.
end_of_speech_sensitivity: Sensitivity for detecting speech end. Defaults to None.
silence_duration_ms: Duration of silence to detect speech end. Defaults to None.
"""
disabled: Optional[bool] = None
start_of_speech_sensitivity: Optional[StartSensitivity] = None
@@ -63,25 +93,57 @@ class AutomaticActivityDetection(BaseModel):
class RealtimeInputConfig(BaseModel):
"""Configures the realtime input behavior."""
"""Configures the realtime input behavior.
Parameters:
automatic_activity_detection: Voice activity detection configuration. Defaults to None.
"""
automatic_activity_detection: Optional[AutomaticActivityDetection] = None
class RealtimeInput(BaseModel):
"""Contains realtime input media chunks.
Parameters:
mediaChunks: List of media chunks for realtime processing.
"""
mediaChunks: List[MediaChunk]
class ClientContent(BaseModel):
"""Content sent from client to the Gemini Live API.
Parameters:
turns: List of conversation turns. Defaults to None.
turnComplete: Whether the client's turn is complete. Defaults to False.
"""
turns: Optional[List[Turn]] = None
turnComplete: bool = False
class AudioInputMessage(BaseModel):
"""Message containing audio input data.
Parameters:
realtimeInput: Realtime input containing audio chunks.
"""
realtimeInput: RealtimeInput
@classmethod
def from_raw_audio(cls, raw_audio: bytes, sample_rate: int) -> "AudioInputMessage":
"""Create an audio input message from raw audio data.
Args:
raw_audio: Raw audio bytes.
sample_rate: Audio sample rate in Hz.
Returns:
AudioInputMessage instance with encoded audio data.
"""
data = base64.b64encode(raw_audio).decode("utf-8")
return cls(
realtimeInput=RealtimeInput(
@@ -91,10 +153,24 @@ class AudioInputMessage(BaseModel):
class VideoInputMessage(BaseModel):
"""Message containing video/image input data.
Parameters:
realtimeInput: Realtime input containing video/image chunks.
"""
realtimeInput: RealtimeInput
@classmethod
def from_image_frame(cls, frame: ImageRawFrame) -> "VideoInputMessage":
"""Create a video input message from an image frame.
Args:
frame: Image frame to encode.
Returns:
VideoInputMessage instance with encoded image data.
"""
buffer = io.BytesIO()
Image.frombytes(frame.format, frame.size, frame.image).save(buffer, format="JPEG")
data = base64.b64encode(buffer.getvalue()).decode("utf-8")
@@ -104,18 +180,44 @@ class VideoInputMessage(BaseModel):
class ClientContentMessage(BaseModel):
"""Message containing client content for the API.
Parameters:
clientContent: The client content to send.
"""
clientContent: ClientContent
class SystemInstruction(BaseModel):
"""System instruction for the model.
Parameters:
parts: List of content parts that make up the system instruction.
"""
parts: List[ContentPart]
class AudioTranscriptionConfig(BaseModel):
"""Configuration for audio transcription."""
pass
class Setup(BaseModel):
"""Setup configuration for the Gemini Live session.
Parameters:
model: Model identifier to use.
system_instruction: System instruction for the model. Defaults to None.
tools: List of available tools/functions. Defaults to None.
generation_config: Generation configuration parameters. Defaults to None.
input_audio_transcription: Input audio transcription config. Defaults to None.
output_audio_transcription: Output audio transcription config. Defaults to None.
realtime_input_config: Realtime input configuration. Defaults to None.
"""
model: str
system_instruction: Optional[SystemInstruction] = None
tools: Optional[List[dict]] = None
@@ -126,6 +228,12 @@ class Setup(BaseModel):
class Config(BaseModel):
"""Configuration message for session setup.
Parameters:
setup: Setup configuration for the session.
"""
setup: Setup
@@ -135,36 +243,86 @@ class Config(BaseModel):
class SetupComplete(BaseModel):
"""Indicates that session setup is complete."""
pass
class InlineData(BaseModel):
"""Inline data embedded in server responses.
Parameters:
mimeType: MIME type of the data.
data: Base64-encoded data content.
"""
mimeType: str
data: str
class Part(BaseModel):
"""Part of a server response containing data or text.
Parameters:
inlineData: Inline binary data. Defaults to None.
text: Text content. Defaults to None.
"""
inlineData: Optional[InlineData] = None
text: Optional[str] = None
class ModelTurn(BaseModel):
"""Represents a turn from the model in the conversation.
Parameters:
parts: List of content parts in the model's response.
"""
parts: List[Part]
class ServerContentInterrupted(BaseModel):
"""Indicates server content was interrupted.
Parameters:
interrupted: Whether the content was interrupted.
"""
interrupted: bool
class ServerContentTurnComplete(BaseModel):
"""Indicates the server's turn is complete.
Parameters:
turnComplete: Whether the turn is complete.
"""
turnComplete: bool
class BidiGenerateContentTranscription(BaseModel):
"""Transcription data from bidirectional content generation.
Parameters:
text: The transcribed text content.
"""
text: str
class ServerContent(BaseModel):
"""Content sent from server to client.
Parameters:
modelTurn: Model's conversational turn. Defaults to None.
interrupted: Whether content was interrupted. Defaults to None.
turnComplete: Whether the turn is complete. Defaults to None.
inputTranscription: Transcription of input audio. Defaults to None.
outputTranscription: Transcription of output audio. Defaults to None.
"""
modelTurn: Optional[ModelTurn] = None
interrupted: Optional[bool] = None
turnComplete: Optional[bool] = None
@@ -173,12 +331,26 @@ class ServerContent(BaseModel):
class FunctionCall(BaseModel):
"""Represents a function call from the model.
Parameters:
id: Unique identifier for the function call.
name: Name of the function to call.
args: Arguments to pass to the function.
"""
id: str
name: str
args: dict
class ToolCall(BaseModel):
"""Contains one or more function calls.
Parameters:
functionCalls: List of function calls to execute.
"""
functionCalls: List[FunctionCall]
@@ -193,14 +365,32 @@ class Modality(str, Enum):
class ModalityTokenCount(BaseModel):
"""Token count for a specific modality."""
"""Token count for a specific modality.
Parameters:
modality: The modality type.
tokenCount: Number of tokens for this modality.
"""
modality: Modality
tokenCount: int
class UsageMetadata(BaseModel):
"""Usage metadata about the response."""
"""Usage metadata about the API response.
Parameters:
promptTokenCount: Number of tokens in the prompt. Defaults to None.
cachedContentTokenCount: Number of cached content tokens. Defaults to None.
responseTokenCount: Number of tokens in the response. Defaults to None.
toolUsePromptTokenCount: Number of tokens for tool use prompts. Defaults to None.
thoughtsTokenCount: Number of tokens for model thoughts. Defaults to None.
totalTokenCount: Total number of tokens used. Defaults to None.
promptTokensDetails: Detailed breakdown of prompt tokens by modality. Defaults to None.
cacheTokensDetails: Detailed breakdown of cache tokens by modality. Defaults to None.
responseTokensDetails: Detailed breakdown of response tokens by modality. Defaults to None.
toolUsePromptTokensDetails: Detailed breakdown of tool use tokens by modality. Defaults to None.
"""
promptTokenCount: Optional[int] = None
cachedContentTokenCount: Optional[int] = None
@@ -215,6 +405,15 @@ class UsageMetadata(BaseModel):
class ServerEvent(BaseModel):
"""Server event received from the Gemini Live API.
Parameters:
setupComplete: Setup completion notification. Defaults to None.
serverContent: Content from the server. Defaults to None.
toolCall: Tool/function call request. Defaults to None.
usageMetadata: Token usage metadata. Defaults to None.
"""
setupComplete: Optional[SetupComplete] = None
serverContent: Optional[ServerContent] = None
toolCall: Optional[ToolCall] = None
@@ -222,6 +421,14 @@ class ServerEvent(BaseModel):
def parse_server_event(str):
"""Parse a server event from JSON string.
Args:
str: JSON string containing the server event.
Returns:
ServerEvent instance if parsing succeeds, None otherwise.
"""
try:
evt = json.loads(str)
return ServerEvent.model_validate(evt)
@@ -231,7 +438,12 @@ def parse_server_event(str):
class ContextWindowCompressionConfig(BaseModel):
"""Configuration for context window compression."""
"""Configuration for context window compression.
Parameters:
sliding_window: Whether to use sliding window compression. Defaults to True.
trigger_tokens: Token count threshold to trigger compression. Defaults to None.
"""
sliding_window: Optional[bool] = Field(default=True)
trigger_tokens: Optional[int] = Field(default=None)

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

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Google Gemini Multimodal Live API service implementation.
This module provides real-time conversational AI capabilities using Google's
Gemini Multimodal Live API, supporting both text and audio modalities with
voice transcription, streaming responses, and tool usage.
"""
import base64
import json
import time
@@ -79,7 +86,11 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
Source:
https://ai.google.dev/api/generate-content#MediaResolution
Returns None if the language is not supported by Gemini Live.
Args:
language: The language enum value to convert.
Returns:
The Gemini language code string, or None if the language is not supported.
"""
language_map = {
# Arabic
@@ -166,8 +177,22 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
class GeminiMultimodalLiveContext(OpenAILLMContext):
"""Extended OpenAI context for Gemini Multimodal Live API.
Provides Gemini-specific context management including system instruction
extraction and message format conversion for the Live API.
"""
@staticmethod
def upgrade(obj: OpenAILLMContext) -> "GeminiMultimodalLiveContext":
"""Upgrade an OpenAI context to Gemini context.
Args:
obj: The OpenAI context to upgrade.
Returns:
The upgraded Gemini context instance.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GeminiMultimodalLiveContext):
logger.debug(f"Upgrading to Gemini Multimodal Live Context: {obj}")
obj.__class__ = GeminiMultimodalLiveContext
@@ -178,6 +203,11 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
pass
def extract_system_instructions(self):
"""Extract system instructions from context messages.
Returns:
Combined system instruction text from all system messages.
"""
system_instruction = ""
for item in self.messages:
if item.get("role") == "system":
@@ -189,6 +219,11 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
return system_instruction
def get_messages_for_initializing_history(self):
"""Get messages formatted for Gemini history initialization.
Returns:
List of messages in Gemini format for conversation history.
"""
messages = []
for item in self.messages:
role = item.get("role")
@@ -216,7 +251,19 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
"""User context aggregator for Gemini Multimodal Live.
Extends OpenAI user aggregator to handle Gemini-specific message passing
while maintaining compatibility with the standard aggregation pipeline.
"""
async def process_frame(self, frame, direction):
"""Process incoming frames for user context aggregation.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
await super().process_frame(frame, direction)
# kind of a hack just to pass the LLMMessagesAppendFrame through, but it's fine for now
if isinstance(frame, LLMMessagesAppendFrame):
@@ -224,15 +271,33 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
# The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
# but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
# need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
# are process. This ensures that the context gets only one set of messages.
"""Assistant context aggregator for Gemini Multimodal Live.
Handles assistant response aggregation while filtering out LLMTextFrames
to prevent duplicate context entries, as Gemini Live pushes both
LLMTextFrames and TTSTextFrames.
"""
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames for assistant context aggregation.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
# The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
# but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
# need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
# are process. This ensures that the context gets only one set of messages.
if not isinstance(frame, LLMTextFrame):
await super().process_frame(frame, direction)
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle user image frames.
Args:
frame: The user image frame to handle.
"""
# We don't want to store any images in the context. Revisit this later
# when the API evolves.
pass
@@ -240,17 +305,36 @@ class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggre
@dataclass
class GeminiMultimodalLiveContextAggregatorPair:
"""Pair of user and assistant context aggregators for Gemini Multimodal Live.
Parameters:
_user: The user context aggregator instance.
_assistant: The assistant context aggregator instance.
"""
_user: GeminiMultimodalLiveUserContextAggregator
_assistant: GeminiMultimodalLiveAssistantContextAggregator
def user(self) -> GeminiMultimodalLiveUserContextAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> GeminiMultimodalLiveAssistantContextAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class GeminiMultimodalModalities(Enum):
"""Supported modalities for Gemini Multimodal Live."""
TEXT = "TEXT"
AUDIO = "AUDIO"
@@ -265,7 +349,15 @@ class GeminiMediaResolution(str, Enum):
class GeminiVADParams(BaseModel):
"""Voice Activity Detection parameters."""
"""Voice Activity Detection parameters for Gemini Live.
Parameters:
disabled: Whether to disable VAD. Defaults to None.
start_sensitivity: Sensitivity for speech start detection. Defaults to None.
end_sensitivity: Sensitivity for speech end detection. Defaults to None.
prefix_padding_ms: Prefix padding in milliseconds. Defaults to None.
silence_duration_ms: Silence duration threshold in milliseconds. Defaults to None.
"""
disabled: Optional[bool] = Field(default=None)
start_sensitivity: Optional[events.StartSensitivity] = Field(default=None)
@@ -275,7 +367,12 @@ class GeminiVADParams(BaseModel):
class ContextWindowCompressionParams(BaseModel):
"""Parameters for context window compression."""
"""Parameters for context window compression in Gemini Live.
Parameters:
enabled: Whether compression is enabled. Defaults to False.
trigger_tokens: Token count to trigger compression. None uses 80% of context window.
"""
enabled: bool = Field(default=False)
trigger_tokens: Optional[int] = Field(
@@ -284,6 +381,23 @@ class ContextWindowCompressionParams(BaseModel):
class InputParams(BaseModel):
"""Input parameters for Gemini Multimodal Live generation.
Parameters:
frequency_penalty: Frequency penalty for generation (0.0-2.0). Defaults to None.
max_tokens: Maximum tokens to generate. Must be >= 1. Defaults to 4096.
presence_penalty: Presence penalty for generation (0.0-2.0). Defaults to None.
temperature: Sampling temperature (0.0-2.0). Defaults to None.
top_k: Top-k sampling parameter. Must be >= 0. Defaults to None.
top_p: Top-p sampling parameter (0.0-1.0). Defaults to None.
modalities: Response modalities. Defaults to AUDIO.
language: Language for generation. Defaults to EN_US.
media_resolution: Media resolution setting. Defaults to UNSPECIFIED.
vad: Voice activity detection parameters. Defaults to None.
context_window_compression: Context compression settings. Defaults to None.
extra: Additional parameters. Defaults to empty dict.
"""
frequency_penalty: Optional[float] = Field(default=None, ge=0.0, le=2.0)
max_tokens: Optional[int] = Field(default=4096, ge=1)
presence_penalty: Optional[float] = Field(default=None, ge=0.0, le=2.0)
@@ -310,23 +424,18 @@ class GeminiMultimodalLiveLLMService(LLMService):
responses, and tool usage.
Args:
api_key (str): Google AI API key
base_url (str, optional): API endpoint base URL. Defaults to
"generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent".
model (str, optional): Model identifier to use. Defaults to
"models/gemini-2.0-flash-live-001".
voice_id (str, optional): TTS voice identifier. Defaults to "Charon".
start_audio_paused (bool, optional): Whether to start with audio input paused.
Defaults to False.
start_video_paused (bool, optional): Whether to start with video input paused.
Defaults to False.
system_instruction (str, optional): System prompt for the model. Defaults to None.
tools (Union[List[dict], ToolsSchema], optional): Tools/functions available to the model.
Defaults to None.
params (InputParams, optional): Configuration parameters for the model.
Defaults to InputParams().
inference_on_context_initialization (bool, optional): Whether to generate a response
when context is first set. Defaults to True.
api_key: Google AI API key for authentication.
base_url: API endpoint base URL. Defaults to the official Gemini Live endpoint.
model: Model identifier to use. Defaults to "models/gemini-2.0-flash-live-001".
voice_id: TTS voice identifier. Defaults to "Charon".
start_audio_paused: Whether to start with audio input paused. Defaults to False.
start_video_paused: Whether to start with video input paused. Defaults to False.
system_instruction: System prompt for the model. Defaults to None.
tools: Tools/functions available to the model. Defaults to None.
params: Configuration parameters for the model. Defaults to InputParams().
inference_on_context_initialization: Whether to generate a response when context
is first set. Defaults to True.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the Gemini one.
@@ -408,19 +517,43 @@ class GeminiMultimodalLiveLLMService(LLMService):
}
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True as Gemini Live supports token usage metrics.
"""
return True
def set_audio_input_paused(self, paused: bool):
"""Set the audio input pause state.
Args:
paused: Whether to pause audio input.
"""
self._audio_input_paused = paused
def set_video_input_paused(self, paused: bool):
"""Set the video input pause state.
Args:
paused: Whether to pause video input.
"""
self._video_input_paused = paused
def set_model_modalities(self, modalities: GeminiMultimodalModalities):
"""Set the model response modalities.
Args:
modalities: The modalities to use for responses.
"""
self._settings["modalities"] = modalities
def set_language(self, language: Language):
"""Set the language for generation."""
"""Set the language for generation.
Args:
language: The language to use for generation.
"""
self._language = language
self._language_code = language_to_gemini_language(language) or "en-US"
self._settings["language"] = self._language_code
@@ -433,6 +566,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
way to trigger the pipeline. This sends the history to the server. The `inference_on_context_initialization`
flag controls whether to set the turnComplete flag when we do this. Without that flag, the model will
not respond. This is often what we want when setting the context at the beginning of a conversation.
Args:
context: The OpenAI LLM context to set.
"""
if self._context:
logger.error(
@@ -447,14 +583,29 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def start(self, frame: StartFrame):
"""Start the service and establish websocket connection.
Args:
frame: The start frame.
"""
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the service and close connections.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the service and close connections.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._disconnect()
@@ -489,6 +640,12 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames for the Gemini Live service.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
@@ -544,6 +701,11 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def send_client_event(self, event):
"""Send a client event to the Gemini Live API.
Args:
event: The event to send.
"""
await self._ws_send(event.model_dump(exclude_none=True))
async def _connect(self):
@@ -1031,22 +1193,19 @@ class GeminiMultimodalLiveLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GeminiMultimodalLiveContextAggregatorPair:
"""Create an instance of GeminiMultimodalLiveContextAggregatorPair from
an OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create an instance of GeminiMultimodalLiveContextAggregatorPair from an OpenAILLMContext.
Constructor keyword arguments for both the user and assistant aggregators can be provided.
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 use.
user_params: User aggregator parameters. Defaults to LLMUserAggregatorParams().
assistant_params: Assistant aggregator parameters. Defaults to LLMAssistantAggregatorParams().
Returns:
GeminiMultimodalLiveContextAggregatorPair: A pair of context
aggregators, one for the user and one for the assistant,
encapsulated in an GeminiMultimodalLiveContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())

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

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Google Gemini integration for Pipecat.
This module provides Google Gemini integration for the Pipecat framework,
including LLM services, context management, and message aggregation.
"""
import base64
import io
import json
@@ -71,7 +77,14 @@ except ModuleNotFoundError as e:
class GoogleUserContextAggregator(OpenAIUserContextAggregator):
"""Google-specific user context aggregator.
Extends OpenAI user context aggregator to handle Google AI's specific
Content and Part message format for user messages.
"""
async def push_aggregation(self):
"""Push aggregated user text as a Google Content message."""
if len(self._aggregation) > 0:
self._context.add_message(Content(role="user", parts=[Part(text=self._aggregation)]))
@@ -88,10 +101,26 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Google-specific assistant context aggregator.
Extends OpenAI assistant context aggregator to handle Google AI's specific
Content and Part message format for assistant responses and function calls.
"""
async def handle_aggregation(self, aggregation: str):
"""Handle aggregated assistant text response.
Args:
aggregation: The aggregated text response from the assistant.
"""
self._context.add_message(Content(role="model", parts=[Part(text=aggregation)]))
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle function call in progress frame.
Args:
frame: Frame containing function call details.
"""
self._context.add_message(
Content(
role="model",
@@ -120,6 +149,11 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle function call result frame.
Args:
frame: Frame containing function call result.
"""
if frame.result:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, frame.result
@@ -130,6 +164,11 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle function call cancellation frame.
Args:
frame: Frame containing function call cancellation details.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
@@ -144,6 +183,11 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
part.function_response.response = {"value": json.dumps(result)}
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle user image frame.
Args:
frame: Frame containing user image data and request context.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
@@ -157,17 +201,45 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
@dataclass
class GoogleContextAggregatorPair:
"""Pair of Google context aggregators for user and assistant messages.
Parameters:
_user: User context aggregator for handling user messages.
_assistant: Assistant context aggregator for handling assistant responses.
"""
_user: GoogleUserContextAggregator
_assistant: GoogleAssistantContextAggregator
def user(self) -> GoogleUserContextAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> GoogleAssistantContextAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class GoogleLLMContext(OpenAILLMContext):
"""Google AI LLM context that extends OpenAI context for Google-specific formatting.
This class handles conversion between OpenAI-style messages and Google AI's
Content/Part format, including system messages, function calls, and media.
Args:
messages: Initial messages in OpenAI format.
tools: Available tools/functions for the model.
tool_choice: Tool choice configuration.
"""
def __init__(
self,
messages: Optional[List[dict]] = None,
@@ -179,6 +251,14 @@ class GoogleLLMContext(OpenAILLMContext):
@staticmethod
def upgrade_to_google(obj: OpenAILLMContext) -> "GoogleLLMContext":
"""Upgrade an OpenAI context to a Google context.
Args:
obj: OpenAI LLM context to upgrade.
Returns:
GoogleLLMContext instance with converted messages.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GoogleLLMContext):
logger.debug(f"Upgrading to Google: {obj}")
obj.__class__ = GoogleLLMContext
@@ -186,10 +266,20 @@ class GoogleLLMContext(OpenAILLMContext):
return obj
def set_messages(self, messages: List):
"""Set messages and restructure them for Google format.
Args:
messages: List of messages to set.
"""
self._messages[:] = messages
self._restructure_from_openai_messages()
def add_messages(self, messages: List):
"""Add messages to the context, converting to Google format as needed.
Args:
messages: List of messages to add (can be mixed formats).
"""
# Convert each message individually
converted_messages = []
for msg in messages:
@@ -206,6 +296,11 @@ class GoogleLLMContext(OpenAILLMContext):
self._messages.extend(converted_messages)
def get_messages_for_logging(self):
"""Get messages formatted for logging with sensitive data redacted.
Returns:
List of message dictionaries with inline data redacted.
"""
msgs = []
for message in self.messages:
obj = message.to_json_dict()
@@ -222,6 +317,14 @@ class GoogleLLMContext(OpenAILLMContext):
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add an image message to the context.
Args:
format: Image format (e.g., 'RGB', 'RGBA').
size: Image dimensions as (width, height).
image: Raw image bytes.
text: Optional text to accompany the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
@@ -235,6 +338,12 @@ class GoogleLLMContext(OpenAILLMContext):
def add_audio_frames_message(
self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
):
"""Add audio frames as a message to the context.
Args:
audio_frames: List of audio frames to add.
text: Text description of the audio content.
"""
if not audio_frames:
return
@@ -448,17 +557,37 @@ class GoogleLLMContext(OpenAILLMContext):
class GoogleLLMService(LLMService):
"""This class implements inference with Google's AI models.
"""Google AI (Gemini) LLM service implementation.
This service translates internally from OpenAILLMContext to the messages format
expected by the Google AI model. We are using the OpenAILLMContext as a lingua
franca for all LLM services, so that it is easy to switch between different LLMs.
This class implements inference with Google's AI models, translating internally
from OpenAILLMContext to the messages format expected by the Google AI model.
We use OpenAILLMContext as a lingua franca for all LLM services to enable
easy switching between different LLMs.
Args:
api_key: Google AI API key for authentication.
model: Model name to use. Defaults to "gemini-2.0-flash".
params: Input parameters for the model.
system_instruction: System instruction/prompt for the model.
tools: List of available tools/functions.
tool_config: Configuration for tool usage.
**kwargs: Additional arguments passed to parent class.
"""
# Overriding the default adapter to use the Gemini one.
adapter_class = GeminiLLMAdapter
class InputParams(BaseModel):
"""Input parameters for Google AI models.
Parameters:
max_tokens: Maximum number of tokens to generate.
temperature: Sampling temperature between 0.0 and 2.0.
top_k: Top-k sampling parameter.
top_p: Top-p sampling parameter between 0.0 and 1.0.
extra: Additional parameters as a dictionary.
"""
max_tokens: Optional[int] = Field(default=4096, ge=1)
temperature: Optional[float] = Field(default=None, ge=0.0, le=2.0)
top_k: Optional[int] = Field(default=None, ge=0)
@@ -495,6 +624,11 @@ class GoogleLLMService(LLMService):
self._tool_config = tool_config
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True, as Google AI provides token usage metrics.
"""
return True
def _create_client(self, api_key: str):
@@ -651,6 +785,12 @@ class GoogleLLMService(LLMService):
await self.push_frame(LLMFullResponseEndFrame())
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and handle different frame types.
Args:
frame: The frame to process.
direction: Direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None
@@ -679,16 +819,15 @@ class GoogleLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GoogleContextAggregatorPair:
"""Create an instance of GoogleContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create Google-specific context aggregators.
Creates a pair of context aggregators optimized for Google'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:
GoogleContextAggregatorPair: A pair of context aggregators, one for

72
src/pipecat/services/grok/llm.py
View File

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Grok LLM service implementation using OpenAI-compatible interface.
This module provides a service for interacting with Grok's API through an
OpenAI-compatible interface, including specialized token usage tracking
and context aggregation functionality.
"""
from dataclasses import dataclass
from loguru import logger
@@ -23,13 +30,33 @@ from pipecat.services.openai.llm import (
@dataclass
class GrokContextAggregatorPair:
"""Pair of context aggregators for user and assistant interactions.
Provides a convenient container for managing both user and assistant
context aggregators together for Grok LLM interactions.
Parameters:
_user: The user context aggregator instance.
_assistant: The assistant context aggregator instance.
"""
_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
@@ -38,12 +65,14 @@ class GrokLLMService(OpenAILLMService):
This service extends OpenAILLMService to connect to Grok's API endpoint while
maintaining full compatibility with OpenAI's interface and functionality.
Includes specialized token usage tracking that accumulates metrics during
processing and reports final totals.
Args:
api_key (str): The API key for accessing Grok's API
base_url (str, optional): The base URL for Grok API. Defaults to "https://api.x.ai/v1"
model (str, optional): The model identifier to use. Defaults to "grok-3-beta"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Grok's API.
base_url: The base URL for Grok API. Defaults to "https://api.x.ai/v1".
model: The model identifier to use. Defaults to "grok-3-beta".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -63,7 +92,16 @@ class GrokLLMService(OpenAILLMService):
self._is_processing = False
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Grok API endpoint."""
"""Create OpenAI-compatible client for Grok API endpoint.
Args:
api_key: The API key to use. If None, uses instance default.
base_url: The base URL to use. If None, uses instance default.
**kwargs: Additional arguments passed to client creation.
Returns:
The configured client instance for Grok API.
"""
logger.debug(f"Creating Grok client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@@ -75,8 +113,8 @@ class GrokLLMService(OpenAILLMService):
them once at the end of processing.
Args:
context (OpenAILLMContext): The context to process, containing messages
and other information needed for the LLM interaction.
context: The context to process, containing messages and other
information needed for the LLM interaction.
"""
# Reset all counters and flags at the start of processing
self._prompt_tokens = 0
@@ -107,8 +145,8 @@ class GrokLLMService(OpenAILLMService):
The final accumulated totals are reported at the end of processing.
Args:
tokens (LLMTokenUsage): The token usage metrics for the current chunk
of processing, containing prompt_tokens and completion_tokens counts.
tokens: The token usage metrics for the current chunk of processing,
containing prompt_tokens and completion_tokens counts.
"""
# Only accumulate metrics during active processing
if not self._is_processing:
@@ -130,22 +168,20 @@ class GrokLLMService(OpenAILLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GrokContextAggregatorPair:
"""Create an instance of GrokContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create an instance of GrokContextAggregatorPair from an OpenAILLMContext.
Constructor keyword arguments for both the user and assistant aggregators
can be provided.
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 configuring the user aggregator.
assistant_params: Parameters for configuring the assistant aggregator.
Returns:
GrokContextAggregatorPair: A pair of context aggregators, one for
the user and one for the assistant, encapsulated in an
GrokContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())

21
src/pipecat/services/groq/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Groq LLM Service implementation using OpenAI-compatible interface."""
from loguru import logger
from pipecat.services.openai.llm import OpenAILLMService
@@ -16,10 +18,10 @@ class GroqLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Groq's API
base_url (str, optional): The base URL for Groq API. Defaults to "https://api.groq.com/openai/v1"
model (str, optional): The model identifier to use. Defaults to "llama-3.3-70b-versatile"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Groq's API.
base_url: The base URL for Groq API. Defaults to "https://api.groq.com/openai/v1".
model: The model identifier to use. Defaults to "llama-3.3-70b-versatile".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -33,6 +35,15 @@ class GroqLLMService(OpenAILLMService):
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Groq API endpoint."""
"""Create OpenAI-compatible client for Groq API endpoint.
Args:
api_key: API key for authentication. If None, uses instance api_key.
base_url: Base URL for the API. If None, uses instance base_url.
**kwargs: Additional arguments passed to the client constructor.
Returns:
An OpenAI-compatible client configured for Groq's API.
"""
logger.debug(f"Creating Groq client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)

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

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""NVIDIA NIM API service implementation.
This module provides a service for interacting with NVIDIA's NIM (NVIDIA Inference
Microservice) API while maintaining compatibility with the OpenAI-style interface.
"""
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.llm import OpenAILLMService
@@ -17,10 +23,10 @@ class NimLLMService(OpenAILLMService):
in token usage reporting between NIM (incremental) and OpenAI (final summary).
Args:
api_key (str): The API key for accessing NVIDIA's NIM API
base_url (str, optional): The base URL for NIM API. Defaults to "https://integrate.api.nvidia.com/v1"
model (str, optional): The model identifier to use. Defaults to "nvidia/llama-3.1-nemotron-70b-instruct"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing NVIDIA's NIM API.
base_url: The base URL for NIM API. Defaults to "https://integrate.api.nvidia.com/v1".
model: The model identifier to use. Defaults to "nvidia/llama-3.1-nemotron-70b-instruct".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -47,8 +53,8 @@ class NimLLMService(OpenAILLMService):
them once at the end of processing.
Args:
context (OpenAILLMContext): The context to process, containing messages
and other information needed for the LLM interaction.
context: The context to process, containing messages and other information
needed for the LLM interaction.
"""
# Reset all counters and flags at the start of processing
self._prompt_tokens = 0
@@ -79,8 +85,8 @@ class NimLLMService(OpenAILLMService):
The final accumulated totals are reported at the end of processing.
Args:
tokens (LLMTokenUsage): The token usage metrics for the current chunk
of processing, containing prompt_tokens and completion_tokens counts.
tokens: The token usage metrics for the current chunk of processing,
containing prompt_tokens and completion_tokens counts.
"""
# Only accumulate metrics during active processing
if not self._is_processing:

13
src/pipecat/services/ollama/llm.py
View File

@@ -4,9 +4,22 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OLLama LLM service implementation for Pipecat AI framework."""
from pipecat.services.openai.llm import OpenAILLMService
class OLLamaLLMService(OpenAILLMService):
"""OLLama LLM service that provides local language model capabilities.
This service extends OpenAILLMService to work with locally hosted OLLama models,
providing a compatible interface for running large language models locally.
Args:
model: The OLLama model to use. Defaults to "llama2".
base_url: The base URL for the OLLama API endpoint.
Defaults to "http://localhost:11434/v1".
"""
def __init__(self, *, model: str = "llama2", base_url: str = "http://localhost:11434/v1"):
super().__init__(model=model, base_url=base_url, api_key="ollama")

25
src/pipecat/services/openai_realtime_beta/azure.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Azure OpenAI Realtime Beta LLM service implementation."""
from loguru import logger
from .openai import OpenAIRealtimeBetaLLMService
@@ -19,7 +21,18 @@ except ModuleNotFoundError as e:
class AzureRealtimeBetaLLMService(OpenAIRealtimeBetaLLMService):
"""Subclass of OpenAI Realtime API Service with adjustments for Azure's wss connection."""
"""Azure OpenAI Realtime Beta LLM service with Azure-specific authentication.
Extends the OpenAI Realtime service to work with Azure OpenAI endpoints,
using Azure's authentication headers and endpoint format. Provides the same
real-time audio and text communication capabilities as the base OpenAI service.
Args:
api_key: The API key for the Azure OpenAI service.
base_url: The full Azure WebSocket endpoint URL including api-version and deployment.
Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
**kwargs: Additional arguments passed to parent OpenAIRealtimeBetaLLMService.
"""
def __init__(
self,
@@ -28,16 +41,6 @@ class AzureRealtimeBetaLLMService(OpenAIRealtimeBetaLLMService):
base_url: str,
**kwargs,
):
"""Constructor takes the same arguments as the parent class, OpenAIRealtimeBetaLLMService.
Note that the following are required arguments:
api_key: The API key for the Azure OpenAI service.
base_url: The base URL for the Azure OpenAI service.
base_url should be set to the full Azure endpoint URL including the api-version and the deployment name. For example,
wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment
"""
super().__init__(base_url=base_url, api_key=api_key, **kwargs)
self.api_key = api_key
self.base_url = base_url

85
src/pipecat/services/openai_realtime_beta/context.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Realtime LLM context and aggregator implementations."""
import copy
import json
@@ -30,6 +32,18 @@ from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
class OpenAIRealtimeLLMContext(OpenAILLMContext):
"""OpenAI Realtime LLM context with session management and message conversion.
Extends the standard OpenAI LLM context to support real-time session properties,
instruction management, and conversion between standard message formats and
realtime conversation items.
Args:
messages: Initial conversation messages. Defaults to None.
tools: Available function tools. Defaults to None.
**kwargs: Additional arguments passed to parent OpenAILLMContext.
"""
def __init__(self, messages=None, tools=None, **kwargs):
super().__init__(messages=messages, tools=tools, **kwargs)
self.__setup_local()
@@ -43,6 +57,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
@staticmethod
def upgrade_to_realtime(obj: OpenAILLMContext) -> "OpenAIRealtimeLLMContext":
"""Upgrade a standard OpenAI LLM context to a realtime context.
Args:
obj: The OpenAILLMContext instance to upgrade.
Returns:
The upgraded OpenAIRealtimeLLMContext instance.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, OpenAIRealtimeLLMContext):
obj.__class__ = OpenAIRealtimeLLMContext
obj.__setup_local()
@@ -52,6 +74,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
# - finish implementing all frames
def from_standard_message(self, message):
"""Convert a standard message format to a realtime conversation item.
Args:
message: The standard message dictionary to convert.
Returns:
A ConversationItem instance for the realtime API.
"""
if message.get("role") == "user":
content = message.get("content")
if isinstance(message.get("content"), list):
@@ -79,6 +109,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
logger.error(f"Unhandled message type in from_standard_message: {message}")
def get_messages_for_initializing_history(self):
"""Get conversation items for initializing the realtime session history.
Converts the context's messages to a format suitable for the realtime API,
handling system instructions and conversation history packaging.
Returns:
List of conversation items for session initialization.
"""
# We can't load a long conversation history into the openai realtime api yet. (The API/model
# forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So
# our general strategy until this is fixed is just to put everything into a first "user"
@@ -133,6 +171,11 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
]
def add_user_content_item_as_message(self, item):
"""Add a user content item as a standard message to the context.
Args:
item: The conversation item to add as a user message.
"""
message = {
"role": "user",
"content": [{"type": "text", "text": item.content[0].transcript}],
@@ -141,9 +184,25 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
"""User context aggregator for OpenAI Realtime API.
Handles user input frames and generates appropriate context updates
for the realtime conversation, including message updates and tool settings.
Args:
context: The OpenAI realtime LLM context.
**kwargs: Additional arguments passed to parent aggregator.
"""
async def process_frame(
self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
):
"""Process incoming frames and handle realtime-specific frame types.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
# Parent does not push LLMMessagesUpdateFrame. This ensures that in a typical pipeline,
# messages are only processed by the user context aggregator, which is generally what we want. But
@@ -157,6 +216,11 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
await self.push_frame(frame, direction)
async def push_aggregation(self):
"""Push user input aggregation.
Currently ignores all user input coming into the pipeline as realtime
audio input is handled directly by the service.
"""
# for the moment, ignore all user input coming into the pipeline.
# todo: think about whether/how to fix this to allow for text input from
# upstream (transport/transcription, or other sources)
@@ -164,6 +228,16 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Assistant context aggregator for OpenAI Realtime API.
Handles assistant output frames from the realtime service, filtering
out duplicate text frames and managing function call results.
Args:
context: The OpenAI realtime LLM context.
**kwargs: Additional arguments passed to parent aggregator.
"""
# The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
# but the OpenAIRealtimeLLMService pushes LLMTextFrames and TTSTextFrames. We
# need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
@@ -171,10 +245,21 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator)
# OpenAIRealtimeLLMService also pushes TranscriptionFrames and InterimTranscriptionFrames,
# so we need to ignore pushing those as well, as they're also TextFrames.
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process assistant frames, filtering out duplicate text content.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
if not isinstance(frame, (LLMTextFrame, TranscriptionFrame, InterimTranscriptionFrame)):
await super().process_frame(frame, direction)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle function call result and notify the realtime service.
Args:
frame: The function call result frame to handle.
"""
await super().handle_function_call_result(frame)
# The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,

498
src/pipecat/services/openai_realtime_beta/events.py
View File

@@ -3,7 +3,8 @@
#
# SPDX-License-Identifier: BSD 2-Clause License
#
#
"""Event models and data structures for OpenAI Realtime API communication."""
import json
import uuid
@@ -19,7 +20,7 @@ from pydantic import BaseModel, ConfigDict, Field
class InputAudioTranscription(BaseModel):
"""Configuration for audio transcription settings.
Attributes:
Parameters:
model: Transcription model to use (e.g., "gpt-4o-transcribe", "whisper-1").
language: Optional language code for transcription.
prompt: Optional transcription hint text.
@@ -39,6 +40,15 @@ class InputAudioTranscription(BaseModel):
class TurnDetection(BaseModel):
"""Server-side voice activity detection configuration.
Parameters:
type: Detection type, must be "server_vad".
threshold: Voice activity detection threshold (0.0-1.0). Defaults to 0.5.
prefix_padding_ms: Padding before speech starts in milliseconds. Defaults to 300.
silence_duration_ms: Silence duration to detect speech end in milliseconds. Defaults to 800.
"""
type: Optional[Literal["server_vad"]] = "server_vad"
threshold: Optional[float] = 0.5
prefix_padding_ms: Optional[int] = 300
@@ -46,6 +56,15 @@ class TurnDetection(BaseModel):
class SemanticTurnDetection(BaseModel):
"""Semantic-based turn detection configuration.
Parameters:
type: Detection type, must be "semantic_vad".
eagerness: Turn detection eagerness level. Can be "low", "medium", "high", or "auto".
create_response: Whether to automatically create responses on turn detection.
interrupt_response: Whether to interrupt ongoing responses on turn detection.
"""
type: Optional[Literal["semantic_vad"]] = "semantic_vad"
eagerness: Optional[Literal["low", "medium", "high", "auto"]] = None
create_response: Optional[bool] = None
@@ -53,10 +72,33 @@ class SemanticTurnDetection(BaseModel):
class InputAudioNoiseReduction(BaseModel):
"""Input audio noise reduction configuration.
Parameters:
type: Noise reduction type for different microphone scenarios.
"""
type: Optional[Literal["near_field", "far_field"]]
class SessionProperties(BaseModel):
"""Configuration properties for an OpenAI Realtime session.
Parameters:
modalities: Communication modalities to enable (text, audio, or both).
instructions: System instructions for the assistant.
voice: Voice ID for text-to-speech output.
input_audio_format: Format for input audio data.
output_audio_format: Format for output audio data.
input_audio_transcription: Configuration for input audio transcription.
input_audio_noise_reduction: Configuration for input audio noise reduction.
turn_detection: Turn detection configuration or False to disable.
tools: Available function tools for the assistant.
tool_choice: Tool usage strategy ("auto", "none", or "required").
temperature: Sampling temperature for response generation.
max_response_output_tokens: Maximum tokens in response or "inf" for unlimited.
"""
modalities: Optional[List[Literal["text", "audio"]]] = None
instructions: Optional[str] = None
voice: Optional[str] = None
@@ -80,6 +122,15 @@ class SessionProperties(BaseModel):
class ItemContent(BaseModel):
"""Content within a conversation item.
Parameters:
type: Content type (text, audio, input_text, or input_audio).
text: Text content for text-based items.
audio: Base64-encoded audio data for audio items.
transcript: Transcribed text for audio items.
"""
type: Literal["text", "audio", "input_text", "input_audio"]
text: Optional[str] = None
audio: Optional[str] = None # base64-encoded audio
@@ -87,6 +138,21 @@ class ItemContent(BaseModel):
class ConversationItem(BaseModel):
"""A conversation item in the realtime session.
Parameters:
id: Unique identifier for the item, auto-generated if not provided.
object: Object type identifier for the realtime API.
type: Item type (message, function_call, or function_call_output).
status: Current status of the item.
role: Speaker role for message items (user, assistant, or system).
content: Content list for message items.
call_id: Function call identifier for function_call items.
name: Function name for function_call items.
arguments: Function arguments as JSON string for function_call items.
output: Function output as JSON string for function_call_output items.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4().hex))
object: Optional[Literal["realtime.item"]] = None
type: Literal["message", "function_call", "function_call_output"]
@@ -102,11 +168,31 @@ class ConversationItem(BaseModel):
class RealtimeConversation(BaseModel):
"""A realtime conversation session.
Parameters:
id: Unique identifier for the conversation.
object: Object type identifier, always "realtime.conversation".
"""
id: str
object: Literal["realtime.conversation"]
class ResponseProperties(BaseModel):
"""Properties for configuring assistant responses.
Parameters:
modalities: Output modalities for the response. Defaults to ["audio", "text"].
instructions: Specific instructions for this response.
voice: Voice ID for text-to-speech in this response.
output_audio_format: Audio format for this response.
tools: Available tools for this response.
tool_choice: Tool usage strategy for this response.
temperature: Sampling temperature for this response.
max_response_output_tokens: Maximum tokens for this response.
"""
modalities: Optional[List[Literal["text", "audio"]]] = ["audio", "text"]
instructions: Optional[str] = None
voice: Optional[str] = None
@@ -121,6 +207,16 @@ class ResponseProperties(BaseModel):
# error class
#
class RealtimeError(BaseModel):
"""Error information from the realtime API.
Parameters:
type: Error type identifier.
code: Specific error code.
message: Human-readable error message.
param: Parameter name that caused the error, if applicable.
event_id: Event ID associated with the error, if applicable.
"""
type: str
code: Optional[str] = ""
message: str
@@ -134,14 +230,38 @@ class RealtimeError(BaseModel):
class ClientEvent(BaseModel):
"""Base class for client events sent to the realtime API.
Parameters:
event_id: Unique identifier for the event, auto-generated if not provided.
"""
event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class SessionUpdateEvent(ClientEvent):
"""Event to update session properties.
Parameters:
type: Event type, always "session.update".
session: Updated session properties.
"""
type: Literal["session.update"] = "session.update"
session: SessionProperties
def model_dump(self, *args, **kwargs) -> Dict[str, Any]:
"""Serialize the event to a dictionary.
Handles special serialization for turn_detection where False becomes null.
Args:
*args: Positional arguments passed to parent model_dump.
**kwargs: Keyword arguments passed to parent model_dump.
Returns:
Dictionary representation of the event.
"""
dump = super().model_dump(*args, **kwargs)
# Handle turn_detection so that False is serialized as null
@@ -153,25 +273,61 @@ class SessionUpdateEvent(ClientEvent):
class InputAudioBufferAppendEvent(ClientEvent):
"""Event to append audio data to the input buffer.
Parameters:
type: Event type, always "input_audio_buffer.append".
audio: Base64-encoded audio data to append.
"""
type: Literal["input_audio_buffer.append"] = "input_audio_buffer.append"
audio: str # base64-encoded audio
class InputAudioBufferCommitEvent(ClientEvent):
"""Event to commit the current input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.commit".
"""
type: Literal["input_audio_buffer.commit"] = "input_audio_buffer.commit"
class InputAudioBufferClearEvent(ClientEvent):
"""Event to clear the input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.clear".
"""
type: Literal["input_audio_buffer.clear"] = "input_audio_buffer.clear"
class ConversationItemCreateEvent(ClientEvent):
"""Event to create a new conversation item.
Parameters:
type: Event type, always "conversation.item.create".
previous_item_id: ID of the item to insert after, if any.
item: The conversation item to create.
"""
type: Literal["conversation.item.create"] = "conversation.item.create"
previous_item_id: Optional[str] = None
item: ConversationItem
class ConversationItemTruncateEvent(ClientEvent):
"""Event to truncate a conversation item's audio content.
Parameters:
type: Event type, always "conversation.item.truncate".
item_id: ID of the item to truncate.
content_index: Index of the content to truncate within the item.
audio_end_ms: End time in milliseconds for the truncated audio.
"""
type: Literal["conversation.item.truncate"] = "conversation.item.truncate"
item_id: str
content_index: int
@@ -179,21 +335,48 @@ class ConversationItemTruncateEvent(ClientEvent):
class ConversationItemDeleteEvent(ClientEvent):
"""Event to delete a conversation item.
Parameters:
type: Event type, always "conversation.item.delete".
item_id: ID of the item to delete.
"""
type: Literal["conversation.item.delete"] = "conversation.item.delete"
item_id: str
class ConversationItemRetrieveEvent(ClientEvent):
"""Event to retrieve a conversation item by ID.
Parameters:
type: Event type, always "conversation.item.retrieve".
item_id: ID of the item to retrieve.
"""
type: Literal["conversation.item.retrieve"] = "conversation.item.retrieve"
item_id: str
class ResponseCreateEvent(ClientEvent):
"""Event to create a new assistant response.
Parameters:
type: Event type, always "response.create".
response: Optional response configuration properties.
"""
type: Literal["response.create"] = "response.create"
response: Optional[ResponseProperties] = None
class ResponseCancelEvent(ClientEvent):
"""Event to cancel the current assistant response.
Parameters:
type: Event type, always "response.cancel".
"""
type: Literal["response.cancel"] = "response.cancel"
@@ -203,6 +386,13 @@ class ResponseCancelEvent(ClientEvent):
class ServerEvent(BaseModel):
"""Base class for server events received from the realtime API.
Parameters:
event_id: Unique identifier for the event.
type: Type of the server event.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
event_id: str
@@ -210,27 +400,65 @@ class ServerEvent(BaseModel):
class SessionCreatedEvent(ServerEvent):
"""Event indicating a session has been created.
Parameters:
type: Event type, always "session.created".
session: The created session properties.
"""
type: Literal["session.created"]
session: SessionProperties
class SessionUpdatedEvent(ServerEvent):
"""Event indicating a session has been updated.
Parameters:
type: Event type, always "session.updated".
session: The updated session properties.
"""
type: Literal["session.updated"]
session: SessionProperties
class ConversationCreated(ServerEvent):
"""Event indicating a conversation has been created.
Parameters:
type: Event type, always "conversation.created".
conversation: The created conversation.
"""
type: Literal["conversation.created"]
conversation: RealtimeConversation
class ConversationItemCreated(ServerEvent):
"""Event indicating a conversation item has been created.
Parameters:
type: Event type, always "conversation.item.created".
previous_item_id: ID of the previous item, if any.
item: The created conversation item.
"""
type: Literal["conversation.item.created"]
previous_item_id: Optional[str] = None
item: ConversationItem
class ConversationItemInputAudioTranscriptionDelta(ServerEvent):
"""Event containing incremental input audio transcription.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.delta".
item_id: ID of the conversation item being transcribed.
content_index: Index of the content within the item.
delta: Incremental transcription text.
"""
type: Literal["conversation.item.input_audio_transcription.delta"]
item_id: str
content_index: int
@@ -238,6 +466,15 @@ class ConversationItemInputAudioTranscriptionDelta(ServerEvent):
class ConversationItemInputAudioTranscriptionCompleted(ServerEvent):
"""Event indicating input audio transcription is complete.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.completed".
item_id: ID of the conversation item that was transcribed.
content_index: Index of the content within the item.
transcript: Complete transcription text.
"""
type: Literal["conversation.item.input_audio_transcription.completed"]
item_id: str
content_index: int
@@ -245,6 +482,15 @@ class ConversationItemInputAudioTranscriptionCompleted(ServerEvent):
class ConversationItemInputAudioTranscriptionFailed(ServerEvent):
"""Event indicating input audio transcription failed.
Parameters:
type: Event type, always "conversation.item.input_audio_transcription.failed".
item_id: ID of the conversation item that failed transcription.
content_index: Index of the content within the item.
error: Error details for the transcription failure.
"""
type: Literal["conversation.item.input_audio_transcription.failed"]
item_id: str
content_index: int
@@ -252,6 +498,15 @@ class ConversationItemInputAudioTranscriptionFailed(ServerEvent):
class ConversationItemTruncated(ServerEvent):
"""Event indicating a conversation item has been truncated.
Parameters:
type: Event type, always "conversation.item.truncated".
item_id: ID of the truncated conversation item.
content_index: Index of the content within the item.
audio_end_ms: End time in milliseconds for the truncated audio.
"""
type: Literal["conversation.item.truncated"]
item_id: str
content_index: int
@@ -259,26 +514,63 @@ class ConversationItemTruncated(ServerEvent):
class ConversationItemDeleted(ServerEvent):
"""Event indicating a conversation item has been deleted.
Parameters:
type: Event type, always "conversation.item.deleted".
item_id: ID of the deleted conversation item.
"""
type: Literal["conversation.item.deleted"]
item_id: str
class ConversationItemRetrieved(ServerEvent):
"""Event containing a retrieved conversation item.
Parameters:
type: Event type, always "conversation.item.retrieved".
item: The retrieved conversation item.
"""
type: Literal["conversation.item.retrieved"]
item: ConversationItem
class ResponseCreated(ServerEvent):
"""Event indicating an assistant response has been created.
Parameters:
type: Event type, always "response.created".
response: The created response object.
"""
type: Literal["response.created"]
response: "Response"
class ResponseDone(ServerEvent):
"""Event indicating an assistant response is complete.
Parameters:
type: Event type, always "response.done".
response: The completed response object.
"""
type: Literal["response.done"]
response: "Response"
class ResponseOutputItemAdded(ServerEvent):
"""Event indicating an output item has been added to a response.
Parameters:
type: Event type, always "response.output_item.added".
response_id: ID of the response.
output_index: Index of the output item.
item: The added conversation item.
"""
type: Literal["response.output_item.added"]
response_id: str
output_index: int
@@ -286,6 +578,15 @@ class ResponseOutputItemAdded(ServerEvent):
class ResponseOutputItemDone(ServerEvent):
"""Event indicating an output item is complete.
Parameters:
type: Event type, always "response.output_item.done".
response_id: ID of the response.
output_index: Index of the output item.
item: The completed conversation item.
"""
type: Literal["response.output_item.done"]
response_id: str
output_index: int
@@ -293,6 +594,17 @@ class ResponseOutputItemDone(ServerEvent):
class ResponseContentPartAdded(ServerEvent):
"""Event indicating a content part has been added to a response.
Parameters:
type: Event type, always "response.content_part.added".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
part: The added content part.
"""
type: Literal["response.content_part.added"]
response_id: str
item_id: str
@@ -302,6 +614,17 @@ class ResponseContentPartAdded(ServerEvent):
class ResponseContentPartDone(ServerEvent):
"""Event indicating a content part is complete.
Parameters:
type: Event type, always "response.content_part.done".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
part: The completed content part.
"""
type: Literal["response.content_part.done"]
response_id: str
item_id: str
@@ -311,6 +634,17 @@ class ResponseContentPartDone(ServerEvent):
class ResponseTextDelta(ServerEvent):
"""Event containing incremental text from a response.
Parameters:
type: Event type, always "response.text.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
delta: Incremental text content.
"""
type: Literal["response.text.delta"]
response_id: str
item_id: str
@@ -320,6 +654,17 @@ class ResponseTextDelta(ServerEvent):
class ResponseTextDone(ServerEvent):
"""Event indicating text content is complete.
Parameters:
type: Event type, always "response.text.done".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
text: Complete text content.
"""
type: Literal["response.text.done"]
response_id: str
item_id: str
@@ -329,6 +674,17 @@ class ResponseTextDone(ServerEvent):
class ResponseAudioTranscriptDelta(ServerEvent):
"""Event containing incremental audio transcript from a response.
Parameters:
type: Event type, always "response.audio_transcript.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
delta: Incremental transcript text.
"""
type: Literal["response.audio_transcript.delta"]
response_id: str
item_id: str
@@ -338,6 +694,17 @@ class ResponseAudioTranscriptDelta(ServerEvent):
class ResponseAudioTranscriptDone(ServerEvent):
"""Event indicating audio transcript is complete.
Parameters:
type: Event type, always "response.audio_transcript.done".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
transcript: Complete transcript text.
"""
type: Literal["response.audio_transcript.done"]
response_id: str
item_id: str
@@ -347,6 +714,17 @@ class ResponseAudioTranscriptDone(ServerEvent):
class ResponseAudioDelta(ServerEvent):
"""Event containing incremental audio data from a response.
Parameters:
type: Event type, always "response.audio.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
delta: Base64-encoded incremental audio data.
"""
type: Literal["response.audio.delta"]
response_id: str
item_id: str
@@ -356,6 +734,16 @@ class ResponseAudioDelta(ServerEvent):
class ResponseAudioDone(ServerEvent):
"""Event indicating audio content is complete.
Parameters:
type: Event type, always "response.audio.done".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
content_index: Index of the content part.
"""
type: Literal["response.audio.done"]
response_id: str
item_id: str
@@ -364,6 +752,17 @@ class ResponseAudioDone(ServerEvent):
class ResponseFunctionCallArgumentsDelta(ServerEvent):
"""Event containing incremental function call arguments.
Parameters:
type: Event type, always "response.function_call_arguments.delta".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
call_id: ID of the function call.
delta: Incremental function arguments as JSON.
"""
type: Literal["response.function_call_arguments.delta"]
response_id: str
item_id: str
@@ -373,6 +772,17 @@ class ResponseFunctionCallArgumentsDelta(ServerEvent):
class ResponseFunctionCallArgumentsDone(ServerEvent):
"""Event indicating function call arguments are complete.
Parameters:
type: Event type, always "response.function_call_arguments.done".
response_id: ID of the response.
item_id: ID of the conversation item.
output_index: Index of the output item.
call_id: ID of the function call.
arguments: Complete function arguments as JSON string.
"""
type: Literal["response.function_call_arguments.done"]
response_id: str
item_id: str
@@ -382,38 +792,90 @@ class ResponseFunctionCallArgumentsDone(ServerEvent):
class InputAudioBufferSpeechStarted(ServerEvent):
"""Event indicating speech has started in the input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.speech_started".
audio_start_ms: Start time of speech in milliseconds.
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_started"]
audio_start_ms: int
item_id: str
class InputAudioBufferSpeechStopped(ServerEvent):
"""Event indicating speech has stopped in the input audio buffer.
Parameters:
type: Event type, always "input_audio_buffer.speech_stopped".
audio_end_ms: End time of speech in milliseconds.
item_id: ID of the associated conversation item.
"""
type: Literal["input_audio_buffer.speech_stopped"]
audio_end_ms: int
item_id: str
class InputAudioBufferCommitted(ServerEvent):
"""Event indicating the input audio buffer has been committed.
Parameters:
type: Event type, always "input_audio_buffer.committed".
previous_item_id: ID of the previous item, if any.
item_id: ID of the committed conversation item.
"""
type: Literal["input_audio_buffer.committed"]
previous_item_id: Optional[str] = None
item_id: str
class InputAudioBufferCleared(ServerEvent):
"""Event indicating the input audio buffer has been cleared.
Parameters:
type: Event type, always "input_audio_buffer.cleared".
"""
type: Literal["input_audio_buffer.cleared"]
class ErrorEvent(ServerEvent):
"""Event indicating an error occurred.
Parameters:
type: Event type, always "error".
error: Error details.
"""
type: Literal["error"]
error: RealtimeError
class RateLimitsUpdated(ServerEvent):
"""Event indicating rate limits have been updated.
Parameters:
type: Event type, always "rate_limits.updated".
rate_limits: List of rate limit information.
"""
type: Literal["rate_limits.updated"]
rate_limits: List[Dict[str, Any]]
class TokenDetails(BaseModel):
"""Detailed token usage information.
Parameters:
cached_tokens: Number of cached tokens used. Defaults to 0.
text_tokens: Number of text tokens used. Defaults to 0.
audio_tokens: Number of audio tokens used. Defaults to 0.
"""
cached_tokens: Optional[int] = 0
text_tokens: Optional[int] = 0
audio_tokens: Optional[int] = 0
@@ -423,6 +885,16 @@ class TokenDetails(BaseModel):
class Usage(BaseModel):
"""Token usage statistics for a response.
Parameters:
total_tokens: Total number of tokens used.
input_tokens: Number of input tokens used.
output_tokens: Number of output tokens used.
input_token_details: Detailed breakdown of input token usage.
output_token_details: Detailed breakdown of output token usage.
"""
total_tokens: int
input_tokens: int
output_tokens: int
@@ -431,6 +903,17 @@ class Usage(BaseModel):
class Response(BaseModel):
"""A complete assistant response.
Parameters:
id: Unique identifier for the response.
object: Object type, always "realtime.response".
status: Current status of the response.
status_details: Additional status information.
output: List of conversation items in the response.
usage: Token usage statistics for the response.
"""
id: str
object: Literal["realtime.response"]
status: Literal["completed", "in_progress", "incomplete", "cancelled", "failed"]
@@ -474,6 +957,17 @@ _server_event_types = {
def parse_server_event(str):
"""Parse a server event from JSON string.
Args:
str: JSON string containing the server event.
Returns:
Parsed server event object of the appropriate type.
Raises:
Exception: If the event type is unimplemented or parsing fails.
"""
try:
event = json.loads(str)
event_type = event["type"]

15
src/pipecat/services/openai_realtime_beta/frames.py
View File

@@ -4,16 +4,31 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Custom frame types for OpenAI Realtime API integration."""
from dataclasses import dataclass
from pipecat.frames.frames import DataFrame, FunctionCallResultFrame
from pipecat.services.openai_realtime_beta.context import OpenAIRealtimeLLMContext
@dataclass
class RealtimeMessagesUpdateFrame(DataFrame):
"""Frame indicating that the realtime context messages have been updated.
Parameters:
context: The updated OpenAI realtime LLM context.
"""
context: "OpenAIRealtimeLLMContext"
@dataclass
class RealtimeFunctionCallResultFrame(DataFrame):
"""Frame containing function call results for the realtime service.
Parameters:
result_frame: The function call result frame to send to the realtime API.
"""
result_frame: FunctionCallResultFrame

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

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Realtime Beta LLM service implementation with WebSocket support."""
import base64
import json
import time
@@ -73,6 +75,15 @@ except ModuleNotFoundError as e:
@dataclass
class CurrentAudioResponse:
"""Tracks the current audio response from the assistant.
Parameters:
item_id: Unique identifier for the audio response item.
content_index: Index of the audio content within the item.
start_time_ms: Timestamp when the audio response started in milliseconds.
total_size: Total size of audio data received in bytes. Defaults to 0.
"""
item_id: str
content_index: int
start_time_ms: int
@@ -80,6 +91,24 @@ class CurrentAudioResponse:
class OpenAIRealtimeBetaLLMService(LLMService):
"""OpenAI Realtime Beta LLM service providing real-time audio and text communication.
Implements the OpenAI Realtime API Beta with WebSocket communication for low-latency
bidirectional audio and text interactions. Supports function calling, conversation
management, and real-time transcription.
Args:
api_key: OpenAI API key for authentication.
model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03".
base_url: WebSocket base URL for the realtime API.
Defaults to "wss://api.openai.com/v1/realtime".
session_properties: Configuration properties for the realtime session.
If None, uses default SessionProperties.
start_audio_paused: Whether to start with audio input paused. Defaults to False.
send_transcription_frames: Whether to emit transcription frames. Defaults to True.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one.
adapter_class = OpenAIRealtimeLLMAdapter
@@ -125,12 +154,30 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self._retrieve_conversation_item_futures = {}
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True if metrics generation is supported.
"""
return True
def set_audio_input_paused(self, paused: bool):
"""Set whether audio input is paused.
Args:
paused: True to pause audio input, False to resume.
"""
self._audio_input_paused = paused
async def retrieve_conversation_item(self, item_id: str):
"""Retrieve a conversation item by ID from the server.
Args:
item_id: The ID of the conversation item to retrieve.
Returns:
The retrieved conversation item.
"""
future = self.get_event_loop().create_future()
retrieval_in_flight = False
if not self._retrieve_conversation_item_futures.get(item_id):
@@ -154,14 +201,29 @@ class OpenAIRealtimeBetaLLMService(LLMService):
#
async def start(self, frame: StartFrame):
"""Start the service and establish WebSocket connection.
Args:
frame: The start frame triggering service initialization.
"""
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the service and close WebSocket connection.
Args:
frame: The end frame triggering service shutdown.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the service and close WebSocket connection.
Args:
frame: The cancel frame triggering service cancellation.
"""
await super().cancel(frame)
await self._disconnect()
@@ -247,6 +309,12 @@ class OpenAIRealtimeBetaLLMService(LLMService):
#
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames from the pipeline.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
@@ -304,6 +372,11 @@ class OpenAIRealtimeBetaLLMService(LLMService):
#
async def send_client_event(self, event: events.ClientEvent):
"""Send a client event to the OpenAI Realtime API.
Args:
event: The client event to send.
"""
await self._ws_send(event.model_dump(exclude_none=True))
async def _connect(self):
@@ -476,6 +549,11 @@ class OpenAIRealtimeBetaLLMService(LLMService):
pass
async def handle_evt_input_audio_transcription_completed(self, evt):
"""Handle completion of input audio transcription.
Args:
evt: The transcription completed event.
"""
await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
if self._send_transcription_frames:
@@ -556,7 +634,9 @@ class OpenAIRealtimeBetaLLMService(LLMService):
await self.push_frame(UserStoppedSpeakingFrame())
async def _maybe_handle_evt_retrieve_conversation_item_error(self, evt: events.ErrorEvent):
"""If the given error event is an error retrieving a conversation item:
"""Maybe handle an error event related to retrieving a conversation item.
If the given error event is an error retrieving a conversation item:
- set an exception on the future that retrieve_conversation_item() is waiting on
- return true
Otherwise:
@@ -603,8 +683,11 @@ class OpenAIRealtimeBetaLLMService(LLMService):
#
async def reset_conversation(self):
# Disconnect/reconnect is the safest way to start a new conversation.
# Note that this will fail if called from the receive task.
"""Reset the conversation by disconnecting and reconnecting.
This is the safest way to start a new conversation. Note that this will
fail if called from the receive task.
"""
logger.debug("Resetting conversation")
await self._disconnect()
if self._context:
@@ -652,22 +735,19 @@ class OpenAIRealtimeBetaLLMService(LLMService):
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 an instance of OpenAIContextAggregatorPair from an OpenAILLMContext.
Constructor keyword arguments for both the user and assistant aggregators can be provided.
Args:
context (OpenAILLMContext): The LLM context.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
context: The LLM context.
user_params: User aggregator parameters.
assistant_params: Assistant aggregator parameters.
Returns:
OpenAIContextAggregatorPair: A pair of context aggregators, one for
the user and one for the assistant, encapsulated in an
OpenAIContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())

41
src/pipecat/services/openpipe/llm.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenPipe LLM service implementation for Pipecat.
This module provides an OpenPipe-specific implementation of the OpenAI LLM service,
enabling integration with OpenPipe's fine-tuning and monitoring capabilities.
"""
from typing import Dict, List, Optional
from loguru import logger
@@ -22,6 +28,22 @@ except ModuleNotFoundError as e:
class OpenPipeLLMService(OpenAILLMService):
"""OpenPipe-powered Large Language Model service.
Extends OpenAI's LLM service to integrate with OpenPipe's fine-tuning and
monitoring platform. Provides enhanced request logging and tagging capabilities
for model training and evaluation.
Args:
model: The model name to use. Defaults to "gpt-4.1".
api_key: OpenAI API key for authentication. If None, reads from environment.
base_url: Custom OpenAI API endpoint URL. Uses default if None.
openpipe_api_key: OpenPipe API key for enhanced features. If None, reads from environment.
openpipe_base_url: OpenPipe API endpoint URL. Defaults to "https://app.openpipe.ai/api/v1".
tags: Optional dictionary of tags to apply to all requests for tracking.
**kwargs: Additional arguments passed to parent OpenAILLMService.
"""
def __init__(
self,
*,
@@ -44,6 +66,16 @@ class OpenPipeLLMService(OpenAILLMService):
self._tags = tags
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create an OpenPipe client instance.
Args:
api_key: OpenAI API key for authentication.
base_url: OpenAI API base URL.
**kwargs: Additional arguments including openpipe_api_key and openpipe_base_url.
Returns:
Configured OpenPipe AsyncOpenAI client instance.
"""
openpipe_api_key = kwargs.get("openpipe_api_key") or ""
openpipe_base_url = kwargs.get("openpipe_base_url") or ""
client = OpenPipeAI(
@@ -56,6 +88,15 @@ class OpenPipeLLMService(OpenAILLMService):
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:
"""Generate streaming chat completions with OpenPipe logging.
Args:
context: The OpenAI LLM context containing conversation state.
messages: List of chat completion message parameters.
Returns:
Async stream of chat completion chunks.
"""
chunks = await self._client.chat.completions.create(
model=self.model_name,
stream=True,

25
src/pipecat/services/openrouter/llm.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenRouter LLM service implementation.
This module provides an OpenAI-compatible interface for interacting with OpenRouter's API,
extending the base OpenAI LLM service functionality.
"""
from typing import Optional
from loguru import logger
@@ -18,10 +24,11 @@ class OpenRouterLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing OpenRouter's API
base_url (str, optional): The base URL for OpenRouter API. Defaults to "https://openrouter.ai/api/v1"
model (str, optional): The model identifier to use. Defaults to "openai/gpt-4o-2024-11-20"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing OpenRouter's API. If None, will attempt
to read from environment variables.
model: The model identifier to use. Defaults to "openai/gpt-4o-2024-11-20".
base_url: The base URL for OpenRouter API. Defaults to "https://openrouter.ai/api/v1".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -40,5 +47,15 @@ class OpenRouterLLMService(OpenAILLMService):
)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create an OpenRouter API client.
Args:
api_key: The API key to use for authentication. If None, uses instance default.
base_url: The base URL for the API. If None, uses instance default.
**kwargs: Additional arguments passed to the parent client creation method.
Returns:
The configured OpenRouter API client instance.
"""
logger.debug(f"Creating OpenRouter client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)

28
src/pipecat/services/perplexity/llm.py
View File

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Perplexity LLM service implementation.
This module provides a service for interacting with Perplexity's API using
an OpenAI-compatible interface. It handles Perplexity's unique token usage
reporting patterns while maintaining compatibility with the Pipecat framework.
"""
from typing import List
from openai import NOT_GIVEN, AsyncStream
@@ -22,10 +29,10 @@ class PerplexityLLMService(OpenAILLMService):
in token usage reporting between Perplexity (incremental) and OpenAI (final summary).
Args:
api_key (str): The API key for accessing Perplexity's API
base_url (str, optional): The base URL for Perplexity's API. Defaults to "https://api.perplexity.ai"
model (str, optional): The model identifier to use. Defaults to "sonar"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Perplexity's API.
base_url: The base URL for Perplexity's API. Defaults to "https://api.perplexity.ai".
model: The model identifier to use. Defaults to "sonar".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -50,11 +57,11 @@ class PerplexityLLMService(OpenAILLMService):
"""Get chat completions from Perplexity API using OpenAI-compatible parameters.
Args:
context: The context containing conversation history and settings
messages: The messages to send to the API
context: The context containing conversation history and settings.
messages: The messages to send to the API.
Returns:
A stream of chat completion chunks
A stream of chat completion chunks from the Perplexity API.
"""
params = {
"model": self.model_name,
@@ -85,8 +92,8 @@ class PerplexityLLMService(OpenAILLMService):
and reporting them once at the end of processing.
Args:
context (OpenAILLMContext): The context to process, containing messages
and other information needed for the LLM interaction.
context: The context to process, containing messages and other
information needed for the LLM interaction.
"""
# Reset all counters and flags at the start of processing
self._prompt_tokens = 0
@@ -115,6 +122,9 @@ class PerplexityLLMService(OpenAILLMService):
Perplexity reports token usage incrementally during streaming,
unlike OpenAI which provides a final summary. We accumulate the
counts and report the total at the end of processing.
Args:
tokens: Token usage information to accumulate.
"""
if not self._is_processing:
return

21
src/pipecat/services/qwen/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Qwen LLM service implementation using OpenAI-compatible interface."""
from loguru import logger
from pipecat.services.openai.llm import OpenAILLMService
@@ -16,10 +18,10 @@ class QwenLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Qwen's API (DashScope API key)
base_url (str, optional): Base URL for Qwen API. Defaults to "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"
model (str, optional): The model identifier to use. Defaults to "qwen-plus".
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Qwen's API (DashScope API key).
base_url: Base URL for Qwen API. Defaults to "https://dashscope-intl.aliyuncs.com/compatible-mode/v1".
model: The model identifier to use. Defaults to "qwen-plus".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -34,6 +36,15 @@ class QwenLLMService(OpenAILLMService):
logger.info(f"Initialized Qwen LLM service with model: {model}")
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Qwen API endpoint."""
"""Create OpenAI-compatible client for Qwen API endpoint.
Args:
api_key: API key for authentication. If None, uses instance default.
base_url: Base URL for the API. If None, uses instance default.
**kwargs: Additional arguments passed to the parent client creation.
Returns:
An OpenAI-compatible client configured for Qwen's API.
"""
logger.debug(f"Creating Qwen client with base URL: {base_url}")
return super().create_client(api_key, base_url, **kwargs)

41
src/pipecat/services/sambanova/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""SambaNova LLM service implementation using OpenAI-compatible interface."""
import json
from typing import Any, Dict, List, Optional
@@ -24,12 +26,14 @@ from pipecat.utils.tracing.service_decorators import traced_llm
class SambaNovaLLMService(OpenAILLMService): # type: ignore
"""A service for interacting with SambaNova using the OpenAI-compatible interface.
This service extends OpenAILLMService to connect to SambaNova's API endpoint while
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing SambaNova API.
model (str, optional): The model identifier to use. Defaults to "Meta-Llama-3.3-70B-Instruct".
base_url (str, optional): The base URL for SambaNova API. Defaults to "https://api.sambanova.ai/v1".
api_key: The API key for accessing SambaNova API.
model: The model identifier to use. Defaults to "Llama-4-Maverick-17B-128E-Instruct".
base_url: The base URL for SambaNova API. Defaults to "https://api.sambanova.ai/v1".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
@@ -49,16 +53,31 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
base_url: Optional[str] = None,
**kwargs: Dict[Any, Any],
) -> Any:
"""Create OpenAI-compatible client for SambaNova API endpoint."""
"""Create OpenAI-compatible client for SambaNova API endpoint.
Args:
api_key: API key for authentication. If None, uses instance default.
base_url: Base URL for the API endpoint. If None, uses instance default.
**kwargs: Additional keyword arguments for client configuration.
Returns:
Configured OpenAI-compatible client instance.
"""
logger.debug(f"Creating SambaNova client with API {base_url}")
return super().create_client(api_key, base_url, **kwargs)
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> Any:
"""Get chat completions from SambaNova API endpoint."""
"""Get chat completions from SambaNova API endpoint.
Args:
context: OpenAI LLM context containing tools and configuration.
messages: List of chat completion message parameters.
Returns:
Chat completion response stream from SambaNova API.
"""
params = {
"model": self.model_name,
"stream": True,
@@ -79,8 +98,18 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
@traced_llm # type: ignore
async def _process_context(self, context: OpenAILLMContext) -> AsyncStream[ChatCompletionChunk]:
"""Redefine this method until SambaNova API introduces indexing in tool calls."""
"""Process OpenAI LLM context and stream chat completion chunks.
This method handles the streaming response from SambaNova API, including
function call processing and text frame generation. It includes special
handling for SambaNova's API limitations with tool call indexing.
Args:
context: OpenAI LLM context containing conversation state and tools.
Returns:
Async stream of chat completion chunks.
"""
functions_list = []
arguments_list = []
tool_id_list = []

21
src/pipecat/services/together/llm.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Together.ai LLM service implementation using OpenAI-compatible interface."""
from loguru import logger
from pipecat.services.openai.llm import OpenAILLMService
@@ -16,10 +18,10 @@ class TogetherLLMService(OpenAILLMService):
maintaining full compatibility with OpenAI's interface and functionality.
Args:
api_key (str): The API key for accessing Together.ai's API
base_url (str, optional): The base URL for Together.ai API. Defaults to "https://api.together.xyz/v1"
model (str, optional): The model identifier to use. Defaults to "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo"
**kwargs: Additional keyword arguments passed to OpenAILLMService
api_key: The API key for accessing Together.ai's API.
base_url: The base URL for Together.ai API. Defaults to "https://api.together.xyz/v1".
model: The model identifier to use. Defaults to "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo".
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
def __init__(
@@ -33,6 +35,15 @@ class TogetherLLMService(OpenAILLMService):
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
def create_client(self, api_key=None, base_url=None, **kwargs):
"""Create OpenAI-compatible client for Together.ai API endpoint."""
"""Create OpenAI-compatible client for Together.ai API endpoint.
Args:
api_key: The API key to use for the client. If None, uses instance api_key.
base_url: The base URL for the API. If None, uses instance base_url.
**kwargs: Additional keyword arguments passed to the parent create_client method.
Returns:
An OpenAI-compatible client configured for Together.ai's API.
"""
logger.debug(f"Creating Together.ai client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)