Merge pull request #2440 from pipecat-ai/pk/prototype-llm-failover-attempt-4

Support for runtime LLM switching
This commit is contained in:
kompfner
2025-08-26 09:55:03 -04:00
committed by GitHub
41 changed files with 3093 additions and 171 deletions

View File

@@ -7,6 +7,78 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- Added a new "universal" (LLM-agnostic) `LLMContext` and accompanying
`LLMContextAggregatorPair`, which will eventually replace `OpenAILLMContext`
(and the other under-the-hood contexts) and the other context aggregators.
The new universal `LLMContext` machinery allows a single context to be shared
between different LLMs, enabling runtime LLM switching and scenarios like
failover.
From the developer's point of view, switching to using the new universal
context machinery will usually be a matter of going from this:
```python
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
```
To this:
```python
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
```
To start, the universal `LLMContext` is supported with the following LLM
services:
- `OpenAILLMService`
- `GoogleLLMService`
- Added a new `LLMSwitcher` class to enable runtime LLM switching, built atop a
new generic `ServiceSwitcher`.
Switchers take a switching strategy. The first available strategy is
`ServiceSwitcherStrategyManual`.
To switch LLMs at runtime, the LLMs must be sharing one instance of the new
universal `LLMContext` (see above bullet).
```python
# Instantiate your LLM services
llm_openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
llm_google = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
# Instantiate a switcher
# (ServiceSwitcherStrategyManual defaults to OpenAI, as it's first in the list)
llm_switcher = LLMSwitcher(
llms=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual
)
# Create your pipeline
pipeline = Pipeline(
[
transport.input(),
stt,
context_aggregator.user(),
llm_switcher,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
# ...
# Whenever is appropriate, switch LLMs!
await task.queue_frames([ManuallySwitchServiceFrame(service=llm_google)])
```
- Added an `LLMService.run_inference()` method to LLM services to enable
direct, out-of-band (i.e. out-of-pipeline) inference.
### Fixed
- Fixed a `CartesiaTTSService` issue that was causing the application to hang
@@ -62,7 +134,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Deprecated
- `FrameProcessor.wait_for_task()` is deprecated. Use `await task` or `await
asyncio.wait_for(task, timeout)` instead.
asyncio.wait_for(task, timeout)` instead.
### Removed

View File

@@ -59,6 +59,9 @@ GOOGLE_VERTEX_TEST_CREDENTIALS=...
LMNT_API_KEY=...
LMNT_VOICE_ID=...
# Perplexity
PERPLEXITY_API_KEY=...
# PlayHT
PLAY_HT_USER_ID=...
PLAY_HT_API_KEY=...

View File

@@ -0,0 +1,170 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
from pipecat.transports.services.daily import DailyParams
load_dotenv(override=True)
async def fetch_weather_from_api(params: FunctionCallParams):
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

View File

@@ -0,0 +1,229 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import (
create_transport,
get_transport_client_id,
maybe_capture_participant_camera,
)
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.google.llm import GoogleLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.services.daily import DailyParams
load_dotenv(override=True)
# Global variable to store the client ID
client_id = ""
async def get_weather(params: FunctionCallParams):
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={client_id}, question={question}")
# Request the image frame
await params.llm.request_image_frame(
user_id=client_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
# Wait a short time for the frame to be processed
await asyncio.sleep(0.5)
# Return a result to complete the function call
await params.result_callback(
f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
llm.register_function("get_weather", get_weather)
llm.register_function("get_image", get_image)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
weather_function = FunctionSchema(
name="get_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
get_image_function = FunctionSchema(
name="get_image",
description="Get an image from the video stream.",
properties={
"question": {
"type": "string",
"description": "The question that the user is asking about the image.",
}
},
required=["question"],
)
tools = ToolsSchema(standard_tools=[weather_function, get_image_function, restaurant_function])
system_prompt = """\
You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions.
Your response will be turned into speech so use only simple words and punctuation.
You have access to three tools: get_weather, get_restaurant_recommendation, and get_image.
You can respond to questions about the weather using the get_weather tool.
You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
indicate you should use the get_image tool are:
- What do you see?
- What's in the video?
- Can you describe the video?
- Tell me about what you see.
- Tell me something interesting about what you see.
- What's happening in the video?
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": "Say hello."},
]
context = LLMContext(messages, tools)
context_aggregator = LLMContextAggregatorPair(context)
pipeline = Pipeline(
[
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected: {client}")
await maybe_capture_participant_camera(transport, client)
global client_id
client_id = get_transport_client_id(transport, client)
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

View File

@@ -11,21 +11,45 @@ adapters that handle tool format conversion and standardization.
"""
from abc import ABC, abstractmethod
from typing import Any, List, Union, cast
from typing import Any, Generic, List, TypeVar, Union, cast
from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
# Should be a TypedDict
TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any])
class BaseLLMAdapter(ABC):
class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
"""Abstract base class for LLM provider adapters.
Provides a standard interface for converting between Pipecat's standardized
tool schemas and provider-specific tool formats. Subclasses must implement
provider-specific conversion logic.
Provides a standard interface for converting to provider-specific formats.
Handles:
- Extracting provider-specific parameters for LLM invocation from a
universal LLM context
- Converting standardized tools schema to provider-specific tool formats.
- Extracting messages from the LLM context for the purposes of logging
about the specific provider.
Subclasses must implement provider-specific conversion logic.
"""
@abstractmethod
def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams:
"""Get provider-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Provider-specific parameters for invoking the LLM.
"""
pass
@abstractmethod
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
"""Convert tools schema to the provider's specific format.
@@ -38,7 +62,20 @@ class BaseLLMAdapter(ABC):
"""
pass
def from_standard_tools(self, tools: Any) -> List[Any]:
@abstractmethod
def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about this provider.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about this
provider.
"""
pass
def from_standard_tools(self, tools: Any) -> List[Any] | NotGiven:
"""Convert tools from standard format to provider format.
Args:

View File

@@ -6,20 +6,58 @@
"""Anthropic LLM adapter for Pipecat."""
from typing import Any, Dict, List
from typing import Any, Dict, List, TypedDict
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
class AnthropicLLMAdapter(BaseLLMAdapter):
class AnthropicLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Anthropic's LLM API.
This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
"""
pass
class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
"""Adapter for converting tool schemas to Anthropic's function-calling format.
This adapter handles the conversion of Pipecat's standard function schemas
to the specific format required by Anthropic's Claude models for function calling.
"""
def get_llm_invocation_params(self, context: LLMContext) -> AnthropicLLMInvocationParams:
"""Get Anthropic-specific LLM invocation parameters from a universal LLM context.
This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for invoking Anthropic's LLM API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about Anthropic.
Removes or truncates sensitive data like image content for safe logging.
This is a placeholder until support for universal LLMContext machinery is added for Anthropic.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about Anthropic.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
@staticmethod
def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a single function schema to Anthropic's format.

View File

@@ -7,20 +7,58 @@
"""AWS Nova Sonic LLM adapter for Pipecat."""
import json
from typing import Any, Dict, List
from typing import Any, Dict, List, TypedDict
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
class AWSNovaSonicLLMAdapter(BaseLLMAdapter):
class AWSNovaSonicLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking AWS Nova Sonic LLM API.
This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
"""
pass
class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
"""Adapter for AWS Nova Sonic language models.
Converts Pipecat's standard function schemas into AWS Nova Sonic's
specific function-calling format, enabling tool use with Nova Sonic models.
"""
def get_llm_invocation_params(self, context: LLMContext) -> AWSNovaSonicLLMInvocationParams:
"""Get AWS Nova Sonic-specific LLM invocation parameters from a universal LLM context.
This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for invoking AWS Nova Sonic's LLM API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.")
def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about AWS Nova Sonic.
Removes or truncates sensitive data like image content for safe logging.
This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about AWS Nova Sonic.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for AWS Nova Sonic.")
@staticmethod
def _to_aws_nova_sonic_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a function schema to AWS Nova Sonic format.

View File

@@ -6,20 +6,58 @@
"""AWS Bedrock LLM adapter for Pipecat."""
from typing import Any, Dict, List
from typing import Any, Dict, List, TypedDict
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
class AWSBedrockLLMAdapter(BaseLLMAdapter):
class AWSBedrockLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking AWS Bedrock's LLM API.
This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
"""
pass
class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
"""Adapter for AWS Bedrock LLM integration with Pipecat.
Provides conversion utilities for transforming Pipecat function schemas
into AWS Bedrock's expected tool format for function calling capabilities.
"""
def get_llm_invocation_params(self, context: LLMContext) -> AWSBedrockLLMInvocationParams:
"""Get AWS Bedrock-specific LLM invocation parameters from a universal LLM context.
This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for invoking AWS Bedrock's LLM API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about AWS Bedrock.
Removes or truncates sensitive data like image content for safe logging.
This is a placeholder until support for universal LLMContext machinery is added for Bedrock.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about AWS Bedrock.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
@staticmethod
def _to_bedrock_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a function schema to Bedrock's tool format.

View File

@@ -6,20 +6,71 @@
"""Gemini LLM adapter for Pipecat."""
from typing import Any, Dict, List, Union
import base64
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, TypedDict
from loguru import logger
from openai import NotGiven
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMSpecificMessage,
LLMStandardMessage,
)
try:
from google.genai.types import (
Blob,
Content,
FunctionCall,
FunctionResponse,
Part,
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
raise Exception(f"Missing module: {e}")
class GeminiLLMAdapter(BaseLLMAdapter):
"""LLM adapter for Google's Gemini service.
class GeminiLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Gemini LLM."""
Provides tool schema conversion functionality to transform standard tool
definitions into Gemini's specific function-calling format for use with
Gemini LLM models.
system_instruction: Optional[str]
messages: List[Content]
tools: List[Any] | NotGiven
class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
"""Gemini-specific adapter for Pipecat.
Handles:
- Extracting parameters for Gemini's API from a universal LLM context
- Converting Pipecat's standardized tools schema to Gemini's function-calling format.
- Extracting and sanitizing messages from the LLM context for logging with Gemini.
"""
def get_llm_invocation_params(self, context: LLMContext) -> GeminiLLMInvocationParams:
"""Get Gemini-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for Gemini's API.
"""
messages = self._from_universal_context_messages(self._get_messages(context))
return {
"system_instruction": messages.system_instruction,
"messages": messages.messages,
# NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
}
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
"""Convert tool schemas to Gemini's function-calling format.
@@ -39,3 +90,223 @@ class GeminiLLMAdapter(BaseLLMAdapter):
custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])
return formatted_standard_tools + custom_gemini_tools
def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about Gemini.
Removes or truncates sensitive data like image content for safe logging.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about Gemini.
"""
# Get messages in Gemini's format
messages = self._from_universal_context_messages(self._get_messages(context)).messages
# Sanitize messages for logging
messages_for_logging = []
for message in messages:
obj = message.to_json_dict()
try:
if "parts" in obj:
for part in obj["parts"]:
if "inline_data" in part:
part["inline_data"]["data"] = "..."
except Exception as e:
logger.debug(f"Error: {e}")
messages_for_logging.append(obj)
return messages_for_logging
def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
return context.get_messages("google")
@dataclass
class ConvertedMessages:
"""Container for Google-formatted messages converted from universal context."""
messages: List[Content]
system_instruction: Optional[str] = None
def _from_universal_context_messages(
self, universal_context_messages: List[LLMContextMessage]
) -> ConvertedMessages:
"""Restructures messages to ensure proper Google format and message ordering.
This method handles conversion of OpenAI-formatted messages to Google format,
with special handling for function calls, function responses, and system messages.
System messages are added back to the context as user messages when needed.
The final message order is preserved as:
1. Function calls (from model)
2. Function responses (from user)
3. Text messages (converted from system messages)
Note::
System messages are only added back when there are no regular text
messages in the context, ensuring proper conversation continuity
after function calls.
"""
system_instruction = None
messages = []
# Process each message, preserving Google-formatted messages and converting others
for message in universal_context_messages:
if isinstance(message, LLMSpecificMessage):
# Assume that LLMSpecificMessage wraps a message in Google format
messages.append(message.message)
continue
# Convert standard format to Google format
converted = self._from_standard_message(
message, already_have_system_instruction=bool(system_instruction)
)
if isinstance(converted, Content):
# Regular (non-system) message
messages.append(converted)
else:
# System instruction
system_instruction = converted
# Check if we only have function-related messages (no regular text)
has_regular_messages = any(
len(msg.parts) == 1
and getattr(msg.parts[0], "text", None)
and not getattr(msg.parts[0], "function_call", None)
and not getattr(msg.parts[0], "function_response", None)
for msg in messages
)
# Add system instruction back as a user message if we only have function messages
if system_instruction and not has_regular_messages:
messages.append(Content(role="user", parts=[Part(text=system_instruction)]))
# Remove any empty messages
messages = [m for m in messages if m.parts]
return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
def _from_standard_message(
self, message: LLMStandardMessage, already_have_system_instruction: bool
) -> Content | str:
"""Convert universal context message to Google Content object.
Handles conversion of text, images, and function calls to Google's
format.
System instructions are returned as a plain string.
Args:
message: Message in universal context format.
already_have_system_instruction: Whether we already have a system instruction
Returns:
Content object with role and parts, or a plain string for system
messages.
Examples:
Standard text message::
{
"role": "user",
"content": "Hello there"
}
Converts to Google Content with::
Content(
role="user",
parts=[Part(text="Hello there")]
)
Standard function call message::
{
"role": "assistant",
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "test"}'
}
}
]
}
Converts to Google Content with::
Content(
role="model",
parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
)
"""
role = message["role"]
content = message.get("content", [])
if role == "system":
if already_have_system_instruction:
role = "user" # Convert system message to user role if we already have a system instruction
else:
# System instructions are returned as plain text
if isinstance(content, str):
return content
elif isinstance(content, list):
# If content is a list, we assume it's a list of text parts, per the standard
return " ".join(part["text"] for part in content if part.get("type") == "text")
elif role == "assistant":
role = "model"
parts = []
if message.get("tool_calls"):
for tc in message["tool_calls"]:
parts.append(
Part(
function_call=FunctionCall(
name=tc["function"]["name"],
args=json.loads(tc["function"]["arguments"]),
)
)
)
elif role == "tool":
role = "model"
try:
response = json.loads(message["content"])
if isinstance(response, dict):
response_dict = response
else:
response_dict = {"value": response}
except Exception as e:
# Response might not be JSON-deserializable.
# This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
response_dict = {"value": message["content"]}
parts.append(
Part(
function_response=FunctionResponse(
name="tool_call_result", # seems to work to hard-code the same name every time
response=response_dict,
)
)
)
elif isinstance(content, str):
parts.append(Part(text=content))
elif isinstance(content, list):
for c in content:
if c["type"] == "text":
parts.append(Part(text=c["text"]))
elif c["type"] == "image_url":
parts.append(
Part(
inline_data=Blob(
mime_type="image/jpeg",
data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
)
)
)
elif c["type"] == "input_audio":
input_audio = c["input_audio"]
audio_bytes = base64.b64decode(input_audio["data"])
parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
message = Content(role=role, parts=parts)
return message

View File

@@ -6,22 +6,63 @@
"""OpenAI LLM adapter for Pipecat."""
from typing import List
import copy
import json
from typing import Any, List, TypedDict
from openai.types.chat import ChatCompletionToolParam
from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import (
ChatCompletionMessageParam,
ChatCompletionToolChoiceOptionParam,
ChatCompletionToolParam,
)
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMContextToolChoice,
NotGiven,
)
class OpenAILLMAdapter(BaseLLMAdapter):
"""Adapter for converting tool schemas to OpenAI's format.
class OpenAILLMInvocationParams(TypedDict):
"""Context-based parameters for invoking OpenAI ChatCompletion API."""
Provides conversion utilities for transforming Pipecat's standard tool
schemas into the format expected by OpenAI's ChatCompletion API for
function calling capabilities.
messages: List[ChatCompletionMessageParam]
tools: List[ChatCompletionToolParam] | OpenAINotGiven
tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven
class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
"""OpenAI-specific adapter for Pipecat.
Handles:
- Extracting parameters for OpenAI's ChatCompletion API from a universal
LLM context
- Converting Pipecat's standardized tools schema to OpenAI's function-calling format.
- Extracting and sanitizing messages from the LLM context for logging about OpenAI.
"""
def get_llm_invocation_params(self, context: LLMContext) -> OpenAILLMInvocationParams:
"""Get OpenAI-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for OpenAI's ChatCompletion API.
"""
return {
"messages": self._from_universal_context_messages(self._get_messages(context)),
# NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools),
"tool_choice": context.tool_choice,
}
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]:
"""Convert function schemas to OpenAI's function-calling format.
@@ -37,3 +78,43 @@ class OpenAILLMAdapter(BaseLLMAdapter):
ChatCompletionToolParam(type="function", function=func.to_default_dict())
for func in functions_schema
]
def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about OpenAI.
Removes or truncates sensitive data like image content for safe logging.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about OpenAI.
"""
msgs = []
for message in self._get_messages(context):
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image_url":
if item["image_url"]["url"].startswith("data:image/"):
item["image_url"]["url"] = "data:image/..."
if "mime_type" in msg and msg["mime_type"].startswith("image/"):
msg["data"] = "..."
msgs.append(msg)
return json.dumps(msgs, ensure_ascii=False)
def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
return context.get_messages("openai")
def _from_universal_context_messages(
self, messages: List[LLMContextMessage]
) -> List[ChatCompletionMessageParam]:
# Just a pass-through: messages are already the right type
return messages
def _from_standard_tool_choice(
self, tool_choice: LLMContextToolChoice | NotGiven
) -> ChatCompletionToolChoiceOptionParam | OpenAINotGiven:
# Just a pass-through: tool_choice is already the right type
return tool_choice

View File

@@ -6,11 +6,21 @@
"""OpenAI Realtime LLM adapter for Pipecat."""
from typing import Any, Dict, List, Union
from typing import Any, Dict, List, TypedDict, Union
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
class OpenAIRealtimeLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking OpenAI Realtime API.
This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
"""
pass
class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@@ -20,6 +30,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
OpenAI's Realtime API for function calling capabilities.
"""
def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
"""Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for invoking OpenAI Realtime's API.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
def get_messages_for_logging(self, context) -> List[dict[str, Any]]:
"""Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime.
Removes or truncates sensitive data like image content for safe logging.
This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about OpenAI Realtime.
"""
raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
@staticmethod
def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
"""Convert a function schema to OpenAI Realtime format.

View File

@@ -36,6 +36,7 @@ from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id
if TYPE_CHECKING:
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameProcessor
@@ -403,6 +404,11 @@ class OpenAILLMContextAssistantTimestampFrame(DataFrame):
timestamp: str
# A more universal (LLM-agnostic) name for
# OpenAILLMContextAssistantTimestampFrame, matching LLMContext
LLMContextAssistantTimestampFrame = OpenAILLMContextAssistantTimestampFrame
@dataclass
class TranscriptionMessage:
"""A message in a conversation transcript.
@@ -474,6 +480,20 @@ class TranscriptionUpdateFrame(DataFrame):
return f"{self.name}(pts: {pts}, messages: {len(self.messages)})"
@dataclass
class LLMContextFrame(Frame):
"""Frame containing a universal LLM context.
Used as a signal to LLM services to ingest the provided context and
generate a response based on it.
Parameters:
context: The LLM context containing messages, tools, and configuration.
"""
context: "LLMContext"
@dataclass
class LLMMessagesFrame(DataFrame):
"""Frame containing LLM messages for chat completion.
@@ -1445,3 +1465,20 @@ class MixerEnableFrame(MixerControlFrame):
"""
enable: bool
@dataclass
class ServiceSwitcherFrame(ControlFrame):
"""A base class for frames that control ServiceSwitcher behavior."""
pass
@dataclass
class ManuallySwitchServiceFrame(ServiceSwitcherFrame):
"""A frame to request a manual switch in the active service in a ServiceSwitcher.
Handled by ServiceSwitcherStrategyManual to switch the active service.
"""
service: "FrameProcessor"

View File

@@ -0,0 +1,84 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LLM switcher for switching between different LLMs at runtime, with different switching strategies."""
from typing import Any, List, Optional, Type
from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.services.llm_service import LLMService
class LLMSwitcher(ServiceSwitcher[StrategyType]):
"""A pipeline that switches between different LLMs at runtime."""
def __init__(self, llms: List[LLMService], strategy_type: Type[StrategyType]):
"""Initialize the service switcher with a list of LLMs and a switching strategy."""
super().__init__(llms, strategy_type)
@property
def llms(self) -> List[LLMService]:
"""Get the list of LLMs managed by this switcher."""
return self.services
@property
def active_llm(self) -> Optional[LLMService]:
"""Get the currently active LLM, if any."""
return self.strategy.active_service
async def run_inference(
self, context: LLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context. If both are provided, the
one in the context takes precedence.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
if self.active_llm:
return await self.active_llm.run_inference(
context=context, system_instruction=system_instruction
)
return None
def register_function(
self,
function_name: Optional[str],
handler: Any,
start_callback=None,
*,
cancel_on_interruption: bool = True,
):
"""Register a function handler for LLM function calls, on all LLMs, active or not.
Args:
function_name: The name of the function to handle. Use None to handle
all function calls with a catch-all handler.
handler: The function handler. Should accept a single FunctionCallParams
parameter.
start_callback: Legacy callback function (deprecated). Put initialization
code at the top of your handler instead.
.. deprecated:: 0.0.59
The `start_callback` parameter is deprecated and will be removed in a future version.
cancel_on_interruption: Whether to cancel this function call when an
interruption occurs. Defaults to True.
"""
for llm in self.llms:
llm.register_function(
function_name=function_name,
handler=handler,
start_callback=start_callback,
cancel_on_interruption=cancel_on_interruption,
)

View File

@@ -0,0 +1,144 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Service switcher for switching between different services at runtime, with different switching strategies."""
from typing import Any, Generic, List, Optional, Type, TypeVar
from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, ServiceSwitcherFrame
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class ServiceSwitcherStrategy:
"""Base class for service switching strategies."""
def __init__(self, services: List[FrameProcessor]):
"""Initialize the service switcher strategy with a list of services."""
self.services = services
self.active_service: Optional[FrameProcessor] = None
def is_active(self, service: FrameProcessor) -> bool:
"""Determine if the given service is the currently active one.
This method should be overridden by subclasses to implement specific logic.
Args:
service: The service to check.
Returns:
True if the given service is the active one, False otherwise.
"""
raise NotImplementedError("Subclasses must implement this method.")
def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
"""Handle a frame that controls service switching.
This method can be overridden by subclasses to implement specific logic
for handling frames that control service switching.
Args:
frame: The frame to handle.
direction: The direction of the frame (upstream or downstream).
"""
raise NotImplementedError("Subclasses must implement this method.")
class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
"""A strategy for switching between services manually.
This strategy allows the user to manually select which service is active.
The initial active service is the first one in the list.
"""
def __init__(self, services: List[FrameProcessor]):
"""Initialize the manual service switcher strategy with a list of services."""
super().__init__(services)
self.active_service = services[0] if services else None
def is_active(self, service: FrameProcessor) -> bool:
"""Check if the given service is the currently active one.
Args:
service: The service to check.
Returns:
True if the given service is the active one, False otherwise.
"""
return service == self.active_service
def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
"""Handle a frame that controls service switching.
Args:
frame: The frame to handle.
direction: The direction of the frame (upstream or downstream).
"""
if isinstance(frame, ManuallySwitchServiceFrame):
self._set_active(frame.service)
else:
raise ValueError(f"Unsupported frame type: {type(frame)}")
def _set_active(self, service: FrameProcessor):
"""Set the active service to the given one.
Args:
service: The service to set as active.
"""
if service in self.services:
self.active_service = service
else:
raise ValueError(f"Service {service} is not in the list of available services.")
StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy)
class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
"""A pipeline that switches between different services at runtime."""
def __init__(self, services: List[FrameProcessor], strategy_type: Type[StrategyType]):
"""Initialize the service switcher with a list of services and a switching strategy."""
strategy = strategy_type(services)
super().__init__(*self._make_pipeline_definitions(services, strategy))
self.services = services
self.strategy = strategy
@staticmethod
def _make_pipeline_definitions(
services: List[FrameProcessor], strategy: ServiceSwitcherStrategy
) -> List[Any]:
pipelines = []
for service in services:
pipelines.append(ServiceSwitcher._make_pipeline_definition(service, strategy))
return pipelines
@staticmethod
def _make_pipeline_definition(
service: FrameProcessor, strategy: ServiceSwitcherStrategy
) -> Any:
async def filter(frame) -> bool:
_ = frame
return strategy.is_active(service)
return [
FunctionFilter(filter, direction=FrameDirection.DOWNSTREAM),
service,
FunctionFilter(filter, direction=FrameDirection.UPSTREAM),
]
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process a frame, handling frames which affect service switching.
Args:
frame: The frame to process.
direction: The direction of the frame (upstream or downstream).
"""
await super().process_frame(frame, direction)
if isinstance(frame, ServiceSwitcherFrame):
self.strategy.handle_frame(frame, direction)

View File

@@ -0,0 +1,277 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Universal LLM context management for LLM services in Pipecat.
Context contents are represented in a universal format (based on OpenAI)
that supports a union of known Pipecat LLM service functionality.
Whenever an LLM service needs to access context, it does a just-in-time
translation from this universal context into whatever format it needs, using a
service-specific adapter.
"""
import base64
import io
from dataclasses import dataclass
from typing import Any, List, Optional, TypeAlias, Union
from loguru import logger
from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import (
ChatCompletionMessageParam,
ChatCompletionToolChoiceOptionParam,
)
from PIL import Image
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import AudioRawFrame
# "Re-export" types from OpenAI that we're using as universal context types.
# NOTE: if universal message types need to someday diverge from OpenAI's, we
# should consider managing our own definitions. But we should do so carefully,
# as the OpenAI messages are somewhat of a standard and we want to continue
# supporting them.
LLMStandardMessage = ChatCompletionMessageParam
LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
NOT_GIVEN = OPEN_AI_NOT_GIVEN
NotGiven = OpenAINotGiven
@dataclass
class LLMSpecificMessage:
"""A container for a context message that is specific to a particular LLM service.
Enables the use of service-specific message types while maintaining
compatibility with the universal LLM context format.
"""
llm: str
message: Any
LLMContextMessage: TypeAlias = Union[LLMStandardMessage, LLMSpecificMessage]
class LLMContext:
"""Manages conversation context for LLM interactions.
Handles message history, tool definitions, tool choices, and multimedia
content for LLM conversations. Provides methods for message manipulation,
and content formatting.
"""
def __init__(
self,
messages: Optional[List[LLMContextMessage]] = None,
tools: ToolsSchema | NotGiven = NOT_GIVEN,
tool_choice: LLMContextToolChoice | NotGiven = NOT_GIVEN,
):
"""Initialize the LLM context.
Args:
messages: Initial list of conversation messages.
tools: Available tools for the LLM to use.
tool_choice: Tool selection strategy for the LLM.
"""
self._messages: List[LLMContextMessage] = messages if messages else []
self._tools: ToolsSchema | NotGiven = LLMContext._normalize_and_validate_tools(tools)
self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice
def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]:
"""Get the current messages list.
Args:
llm_specific_filter: Optional filter to return LLM-specific
messages for the given LLM, in addition to the standard
messages. If messages end up being filtered, an error will be
logged.
Returns:
List of conversation messages.
"""
if llm_specific_filter is None:
return self._messages
filtered_messages = [
msg
for msg in self._messages
if not isinstance(msg, LLMSpecificMessage) or msg.llm == llm_specific_filter
]
if len(filtered_messages) < len(self._messages):
logger.error(
f"Attempted to use incompatible LLMSpecificMessages with LLM '{llm_specific_filter}'."
)
return filtered_messages
@property
def tools(self) -> ToolsSchema | NotGiven:
"""Get the tools list.
Returns:
Tools list.
"""
return self._tools
@property
def tool_choice(self) -> LLMContextToolChoice | NotGiven:
"""Get the current tool choice setting.
Returns:
The tool choice configuration.
"""
return self._tool_choice
def add_message(self, message: LLMContextMessage):
"""Add a single message to the context.
Args:
message: The message to add to the conversation history.
"""
self._messages.append(message)
def add_messages(self, messages: List[LLMContextMessage]):
"""Add multiple messages to the context.
Args:
messages: List of messages to add to the conversation history.
"""
self._messages.extend(messages)
def set_messages(self, messages: List[LLMContextMessage]):
"""Replace all messages in the context.
Args:
messages: New list of messages to replace the current history.
"""
self._messages[:] = messages
def set_tools(self, tools: ToolsSchema | NotGiven = NOT_GIVEN):
"""Set the available tools for the LLM.
Args:
tools: A ToolsSchema or NOT_GIVEN to disable tools.
"""
self._tools = LLMContext._normalize_and_validate_tools(tools)
def set_tool_choice(self, tool_choice: LLMContextToolChoice | NotGiven):
"""Set the tool choice configuration.
Args:
tool_choice: Tool selection strategy for the LLM.
"""
self._tool_choice = tool_choice
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add a message containing an image frame.
Args:
format: Image format (e.g., 'RGB', 'RGBA').
size: Image dimensions as (width, height) tuple.
image: Raw image bytes.
text: Optional text to include with the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
content = []
if text:
content.append({"type": "text", "text": text})
content.append(
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}},
)
self.add_message({"role": "user", "content": content})
def add_audio_frames_message(
self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
):
"""Add a message containing audio frames.
Args:
audio_frames: List of audio frame objects to include.
text: Optional text to include with the audio.
"""
if not audio_frames:
return
sample_rate = audio_frames[0].sample_rate
num_channels = audio_frames[0].num_channels
content = []
content.append({"type": "text", "text": text})
data = b"".join(frame.audio for frame in audio_frames)
data = bytes(
self._create_wav_header(
sample_rate,
num_channels,
16,
len(data),
)
+ data
)
encoded_audio = base64.b64encode(data).decode("utf-8")
content.append(
{
"type": "input_audio",
"input_audio": {"data": encoded_audio, "format": "wav"},
}
)
self.add_message({"role": "user", "content": content})
def _create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
"""Create a WAV file header for audio data.
Args:
sample_rate: Audio sample rate in Hz.
num_channels: Number of audio channels.
bits_per_sample: Bits per audio sample.
data_size: Size of audio data in bytes.
Returns:
WAV header as a bytearray.
"""
# RIFF chunk descriptor
header = bytearray()
header.extend(b"RIFF") # ChunkID
header.extend((data_size + 36).to_bytes(4, "little")) # ChunkSize: total size - 8
header.extend(b"WAVE") # Format
# "fmt " sub-chunk
header.extend(b"fmt ") # Subchunk1ID
header.extend((16).to_bytes(4, "little")) # Subchunk1Size (16 for PCM)
header.extend((1).to_bytes(2, "little")) # AudioFormat (1 for PCM)
header.extend(num_channels.to_bytes(2, "little")) # NumChannels
header.extend(sample_rate.to_bytes(4, "little")) # SampleRate
# Calculate byte rate and block align
byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
block_align = num_channels * (bits_per_sample // 8)
header.extend(byte_rate.to_bytes(4, "little")) # ByteRate
header.extend(block_align.to_bytes(2, "little")) # BlockAlign
header.extend(bits_per_sample.to_bytes(2, "little")) # BitsPerSample
# "data" sub-chunk
header.extend(b"data") # Subchunk2ID
header.extend(data_size.to_bytes(4, "little")) # Subchunk2Size
return header
@staticmethod
def _normalize_and_validate_tools(tools: ToolsSchema | NotGiven) -> ToolsSchema | NotGiven:
"""Normalize and validate the given tools.
Raises:
TypeError: If tools are not a ToolsSchema or NotGiven.
"""
if isinstance(tools, ToolsSchema):
if not tools.standard_tools and not tools.custom_tools:
return NOT_GIVEN
return tools
elif tools is NOT_GIVEN:
return NOT_GIVEN
else:
raise TypeError(
f"In LLMContext, tools must be a ToolsSchema object or NOT_GIVEN. Got type: {type(tools)}",
)

View File

@@ -0,0 +1,827 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LLM response aggregators for handling conversation context and message aggregation.
This module provides aggregators that process and accumulate LLM responses, user inputs,
and conversation context. These aggregators handle the flow between speech-to-text,
LLM processing, and text-to-speech components in conversational AI pipelines.
"""
import asyncio
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Literal, Optional, Set
from loguru import logger
from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import (
BotInterruptionFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
EndFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMContextAssistantTimestampFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
LLMMessagesUpdateFrame,
LLMSetToolChoiceFrame,
LLMSetToolsFrame,
SpeechControlParamsFrame,
StartFrame,
StartInterruptionFrame,
TextFrame,
TranscriptionFrame,
UserImageRawFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMSpecificMessage,
)
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
class LLMContextAggregator(FrameProcessor):
"""Base LLM aggregator that uses an LLMContext for conversation storage.
This aggregator maintains conversation state using an LLMContext and
pushes LLMContextFrame objects as aggregation frames. It provides
common functionality for context-based conversation management.
"""
def __init__(self, *, context: LLMContext, role: str, **kwargs):
"""Initialize the context response aggregator.
Args:
context: The LLM context to use for conversation storage.
role: The role this aggregator represents (e.g. "user", "assistant").
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._context = context
self._role = role
self._aggregation: str = ""
@property
def messages(self) -> List[LLMContextMessage]:
"""Get messages from the LLM context.
Returns:
List of message dictionaries from the context.
"""
return self._context.get_messages()
@property
def role(self) -> str:
"""Get the role for this aggregator.
Returns:
The role string for this aggregator.
"""
return self._role
@property
def context(self):
"""Get the LLM context.
Returns:
The LLMContext instance used by this aggregator.
"""
return self._context
def get_context_frame(self) -> LLMContextFrame:
"""Create a context frame with the current context.
Returns:
LLMContextFrame containing the current context.
"""
return LLMContextFrame(context=self._context)
async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Push a context frame in the specified direction.
Args:
direction: The direction to push the frame (upstream or downstream).
"""
frame = self.get_context_frame()
await self.push_frame(frame, direction)
def add_messages(self, messages):
"""Add messages to the context.
Args:
messages: Messages to add to the conversation context.
"""
self._context.add_messages(messages)
def set_messages(self, messages):
"""Set the context messages.
Args:
messages: Messages to replace the current context messages.
"""
self._context.set_messages(messages)
def set_tools(self, tools: List):
"""Set tools in the context.
Args:
tools: List of tool definitions to set in the context.
"""
self._context.set_tools(tools)
def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict):
"""Set tool choice in the context.
Args:
tool_choice: Tool choice configuration for the context.
"""
self._context.set_tool_choice(tool_choice)
async def reset(self):
"""Reset the aggregation state."""
self._aggregation = ""
class LLMUserAggregator(LLMContextAggregator):
"""User LLM aggregator that processes speech-to-text transcriptions.
This aggregator handles the complex logic of aggregating user speech transcriptions
from STT services. It manages multiple scenarios including:
- Transcriptions received between VAD events
- Transcriptions received outside VAD events
- Interim vs final transcriptions
- User interruptions during bot speech
- Emulated VAD for whispered or short utterances
The aggregator uses timeouts to handle cases where transcriptions arrive
after VAD events or when no VAD is available.
"""
def __init__(
self,
context: LLMContext,
*,
params: Optional[LLMUserAggregatorParams] = None,
**kwargs,
):
"""Initialize the user context aggregator.
Args:
context: The LLM context for conversation storage.
params: Configuration parameters for aggregation behavior.
**kwargs: Additional arguments. Supports deprecated 'aggregation_timeout'.
"""
super().__init__(context=context, role="user", **kwargs)
self._params = params or LLMUserAggregatorParams()
self._vad_params: Optional[VADParams] = None
self._turn_params: Optional[SmartTurnParams] = None
if "aggregation_timeout" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'aggregation_timeout' is deprecated, use 'params' instead.",
DeprecationWarning,
)
self._params.aggregation_timeout = kwargs["aggregation_timeout"]
self._user_speaking = False
self._bot_speaking = False
self._was_bot_speaking = False
self._emulating_vad = False
self._seen_interim_results = False
self._waiting_for_aggregation = False
self._aggregation_event = asyncio.Event()
self._aggregation_task = None
async def reset(self):
"""Reset the aggregation state and interruption strategies."""
await super().reset()
self._was_bot_speaking = False
self._seen_interim_results = False
self._waiting_for_aggregation = False
[await s.reset() for s in self._interruption_strategies]
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for user speech aggregation and context management.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
# Push StartFrame before start(), because we want StartFrame to be
# processed by every processor before any other frame is processed.
await self.push_frame(frame, direction)
await self._start(frame)
elif isinstance(frame, EndFrame):
# Push EndFrame before stop(), because stop() waits on the task to
# finish and the task finishes when EndFrame is processed.
await self.push_frame(frame, direction)
await self._stop(frame)
elif isinstance(frame, CancelFrame):
await self._cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, InputAudioRawFrame):
await self._handle_input_audio(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStartedSpeakingFrame):
await self._handle_user_started_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStoppedSpeakingFrame):
await self._handle_user_stopped_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, BotStartedSpeakingFrame):
await self._handle_bot_started_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._handle_bot_stopped_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, TranscriptionFrame):
await self._handle_transcription(frame)
elif isinstance(frame, InterimTranscriptionFrame):
await self._handle_interim_transcription(frame)
elif isinstance(frame, LLMMessagesAppendFrame):
await self._handle_llm_messages_append(frame)
elif isinstance(frame, LLMMessagesUpdateFrame):
await self._handle_llm_messages_update(frame)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, SpeechControlParamsFrame):
self._vad_params = frame.vad_params
self._turn_params = frame.turn_params
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
async def _process_aggregation(self):
"""Process the current aggregation and push it downstream."""
aggregation = self._aggregation
await self.reset()
self._context.add_message({"role": self.role, "content": aggregation})
frame = LLMContextFrame(self._context)
await self.push_frame(frame)
async def _push_aggregation(self):
"""Push the current aggregation based on interruption strategies and conditions."""
if len(self._aggregation) > 0:
if self.interruption_strategies and self._bot_speaking:
should_interrupt = await self._should_interrupt_based_on_strategies()
if should_interrupt:
logger.debug(
"Interruption conditions met - pushing BotInterruptionFrame and aggregation"
)
await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
await self._process_aggregation()
else:
logger.debug("Interruption conditions not met - not pushing aggregation")
# Don't process aggregation, just reset it
await self.reset()
else:
# No interruption config - normal behavior (always push aggregation)
await self._process_aggregation()
# Handles the case where both the user and the bot are not speaking,
# and the bot was previously speaking before the user interruption.
# Normally, when the user stops speaking, new text is expected,
# which triggers the bot to respond. However, if no new text
# is received, this safeguard ensures
# the bot doesn't hang indefinitely while waiting to speak again.
elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
logger.warning("User stopped speaking but no new aggregation received.")
# Resetting it so we don't trigger this twice
self._was_bot_speaking = False
# TODO: we are not enabling this for now, due to some STT services which can take as long as 2 seconds two return a transcription
# So we need more tests and probably make this feature configurable, disabled it by default.
# We are just pushing the same previous context to be processed again in this case
# await self.push_frame(LLMContextFrame(self._context))
async def _should_interrupt_based_on_strategies(self) -> bool:
"""Check if interruption should occur based on configured strategies.
Returns:
True if any interruption strategy indicates interruption should occur.
"""
async def should_interrupt(strategy: BaseInterruptionStrategy):
await strategy.append_text(self._aggregation)
return await strategy.should_interrupt()
return any([await should_interrupt(s) for s in self._interruption_strategies])
async def _start(self, frame: StartFrame):
self._create_aggregation_task()
async def _stop(self, frame: EndFrame):
await self._cancel_aggregation_task()
async def _cancel(self, frame: CancelFrame):
await self._cancel_aggregation_task()
async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
self.add_messages(frame.messages)
if frame.run_llm:
await self.push_context_frame()
async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
self.set_messages(frame.messages)
if frame.run_llm:
await self.push_context_frame()
async def _handle_input_audio(self, frame: InputAudioRawFrame):
for s in self.interruption_strategies:
await s.append_audio(frame.audio, frame.sample_rate)
async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
self._user_speaking = True
self._waiting_for_aggregation = True
self._was_bot_speaking = self._bot_speaking
# If we get a non-emulated UserStartedSpeakingFrame but we are in the
# middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
# EmulateUserStoppedSpeakingFrame).
if not frame.emulated and self._emulating_vad:
self._emulating_vad = False
async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
self._user_speaking = False
# We just stopped speaking. Let's see if there's some aggregation to
# push. If the last thing we saw is an interim transcription, let's wait
# pushing the aggregation as we will probably get a final transcription.
if len(self._aggregation) > 0:
if not self._seen_interim_results:
await self._push_aggregation()
# Handles the case where both the user and the bot are not speaking,
# and the bot was previously speaking before the user interruption.
# So in this case we are resetting the aggregation timer
elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
# Reset aggregation timer.
self._aggregation_event.set()
async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):
self._bot_speaking = True
async def _handle_bot_stopped_speaking(self, _: BotStoppedSpeakingFrame):
self._bot_speaking = False
async def _handle_transcription(self, frame: TranscriptionFrame):
text = frame.text
# Make sure we really have some text.
if not text.strip():
return
self._aggregation += f" {text}" if self._aggregation else text
# We just got a final result, so let's reset interim results.
self._seen_interim_results = False
# Reset aggregation timer.
self._aggregation_event.set()
async def _handle_interim_transcription(self, _: InterimTranscriptionFrame):
self._seen_interim_results = True
def _create_aggregation_task(self):
if not self._aggregation_task:
self._aggregation_task = self.create_task(self._aggregation_task_handler())
async def _cancel_aggregation_task(self):
if self._aggregation_task:
await self.cancel_task(self._aggregation_task)
self._aggregation_task = None
async def _aggregation_task_handler(self):
while True:
try:
# The _aggregation_task_handler handles two distinct timeout scenarios:
#
# 1. When emulating_vad=True: Wait for emulated VAD timeout before
# pushing aggregation (simulating VAD behavior when no actual VAD
# detection occurred).
#
# 2. When emulating_vad=False: Use aggregation_timeout as a buffer
# to wait for potential late-arriving transcription frames after
# a real VAD event.
#
# For emulated VAD scenarios, the timeout strategy depends on whether
# a turn analyzer is configured:
#
# - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because
# the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech
# chunking to feed the turn analyzer. This low value is too fast
# for emulated VAD scenarios where we need to allow users time to
# finish speaking (e.g. 0.8s).
#
# - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain
# consistent user experience between real VAD detection and
# emulated VAD scenarios.
if not self._emulating_vad:
timeout = self._params.aggregation_timeout
elif self._turn_params:
timeout = self._params.turn_emulated_vad_timeout
else:
# Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params
timeout = (
self._vad_params.stop_secs
if self._vad_params
else self._params.turn_emulated_vad_timeout
)
await asyncio.wait_for(self._aggregation_event.wait(), timeout=timeout)
await self._maybe_emulate_user_speaking()
except asyncio.TimeoutError:
if not self._user_speaking:
await self._push_aggregation()
# If we are emulating VAD we still need to send the user stopped
# speaking frame.
if self._emulating_vad:
await self.push_frame(
EmulateUserStoppedSpeakingFrame(), FrameDirection.UPSTREAM
)
self._emulating_vad = False
finally:
self._aggregation_event.clear()
async def _maybe_emulate_user_speaking(self):
"""Maybe emulate user speaking based on transcription.
Emulate user speaking if we got a transcription but it was not
detected by VAD. Behavior when bot is speaking depends on the
enable_emulated_vad_interruptions parameter.
"""
# Check if we received a transcription but VAD was not able to detect
# voice (e.g. when you whisper a short utterance). In that case, we need
# to emulate VAD (i.e. user start/stopped speaking), but we do it only
# if the bot is not speaking. If the bot is speaking and we really have
# a short utterance we don't really want to interrupt the bot.
if (
not self._user_speaking
and not self._waiting_for_aggregation
and len(self._aggregation) > 0
):
if self._bot_speaking and not self._params.enable_emulated_vad_interruptions:
# If emulated VAD interruptions are disabled and bot is speaking, ignore
logger.debug("Ignoring user speaking emulation, bot is speaking.")
await self.reset()
else:
# Either bot is not speaking, or emulated VAD interruptions are enabled
# - trigger user speaking emulation.
await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
self._emulating_vad = True
class LLMAssistantAggregator(LLMContextAggregator):
"""Assistant LLM aggregator that processes bot responses and function calls.
This aggregator handles the complex logic of processing assistant responses including:
- Text frame aggregation between response start/end markers
- Function call lifecycle management
- Context updates with timestamps
- Tool execution and result handling
- Interruption handling during responses
The aggregator manages function calls in progress and coordinates between
text generation and tool execution phases of LLM responses.
"""
def __init__(
self,
context: LLMContext,
*,
params: Optional[LLMAssistantAggregatorParams] = None,
**kwargs,
):
"""Initialize the assistant context aggregator.
Args:
context: The OpenAI LLM context for conversation storage.
params: Configuration parameters for aggregation behavior.
**kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'.
"""
super().__init__(context=context, role="assistant", **kwargs)
self._params = params or LLMAssistantAggregatorParams()
if "expect_stripped_words" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'expect_stripped_words' is deprecated, use 'params' instead.",
DeprecationWarning,
)
self._params.expect_stripped_words = kwargs["expect_stripped_words"]
self._started = 0
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
@property
def has_function_calls_in_progress(self) -> bool:
"""Check if there are any function calls currently in progress.
Returns:
True if function calls are in progress, False otherwise.
"""
return bool(self._function_calls_in_progress)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for assistant response aggregation and function call management.
Args:
frame: The frame to process.
direction: The direction of frame flow in the pipeline.
"""
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
await self._handle_llm_end(frame)
elif isinstance(frame, TextFrame):
await self._handle_text(frame)
elif isinstance(frame, LLMMessagesAppendFrame):
await self._handle_llm_messages_append(frame)
elif isinstance(frame, LLMMessagesUpdateFrame):
await self._handle_llm_messages_update(frame)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, FunctionCallsStartedFrame):
await self._handle_function_calls_started(frame)
elif isinstance(frame, FunctionCallInProgressFrame):
await self._handle_function_call_in_progress(frame)
elif isinstance(frame, FunctionCallResultFrame):
await self._handle_function_call_result(frame)
elif isinstance(frame, FunctionCallCancelFrame):
await self._handle_function_call_cancel(frame)
elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
await self._handle_user_image_frame(frame)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._push_aggregation()
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
async def _push_aggregation(self):
"""Push the current assistant aggregation with timestamp."""
if not self._aggregation:
return
aggregation = self._aggregation.strip()
await self.reset()
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
# Push context frame
await self.push_context_frame()
# Push timestamp frame with current time
timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
self.add_messages(frame.messages)
if frame.run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
self.set_messages(frame.messages)
if frame.run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_interruptions(self, frame: StartInterruptionFrame):
await self._push_aggregation()
self._started = 0
await self.reset()
async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
for function_call in frame.function_calls:
self._function_calls_in_progress[function_call.tool_call_id] = None
async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
logger.debug(
f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
# Update context with the in-progress function call
self._context.add_message(
{
"role": "assistant",
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments),
},
"type": "function",
}
],
}
)
self._context.add_message(
{
"role": "tool",
"content": "IN_PROGRESS",
"tool_call_id": frame.tool_call_id,
}
)
self._function_calls_in_progress[frame.tool_call_id] = frame
async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
logger.debug(
f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
if frame.tool_call_id not in self._function_calls_in_progress:
logger.warning(
f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running"
)
return
del self._function_calls_in_progress[frame.tool_call_id]
properties = frame.properties
# Update context with the function call result
if frame.result:
result = json.dumps(frame.result)
self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
else:
self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED")
run_llm = False
# Run inference if the function call result requires it.
if frame.result:
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it.
run_llm = properties.run_llm
elif frame.run_llm is not None:
# If the frame is indicating we should run the LLM, do it.
run_llm = frame.run_llm
else:
# If this is the last function call in progress, run the LLM.
run_llm = not bool(self._function_calls_in_progress)
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Call the `on_context_updated` callback once the function call result
# is added to the context. Also, run this in a separate task to make
# sure we don't block the pipeline.
if properties and properties.on_context_updated:
task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated"
task = self.create_task(properties.on_context_updated(), task_name)
self._context_updated_tasks.add(task)
task.add_done_callback(self._context_updated_task_finished)
async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
logger.debug(
f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
if frame.tool_call_id not in self._function_calls_in_progress:
return
if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption:
# Update context with the function call cancellation
self._update_function_call_result(frame.function_name, frame.tool_call_id, "CANCELLED")
del self._function_calls_in_progress[frame.tool_call_id]
def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any):
for message in self._context.get_messages():
if (
not isinstance(message, LLMSpecificMessage)
and message["role"] == "tool"
and message["tool_call_id"]
and message["tool_call_id"] == tool_call_id
):
message["content"] = result
async def _handle_user_image_frame(self, frame: UserImageRawFrame):
logger.debug(
f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]"
)
if frame.request.tool_call_id not in self._function_calls_in_progress:
logger.warning(
f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running"
)
return
del self._function_calls_in_progress[frame.request.tool_call_id]
# Update context with the image frame
self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
self._context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.request.context,
)
await self._push_aggregation()
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started += 1
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started -= 1
await self._push_aggregation()
async def _handle_text(self, frame: TextFrame):
if not self._started:
return
if self._params.expect_stripped_words:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
else:
self._aggregation += frame.text
def _context_updated_task_finished(self, task: asyncio.Task):
self._context_updated_tasks.discard(task)
class LLMContextAggregatorPair:
"""Pair of LLM context aggregators for updating context with user and assistant messages."""
def __init__(
self,
context: LLMContext,
*,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
):
"""Initialize the LLM context aggregator pair.
Args:
context: The context to be managed by the aggregators.
user_params: Parameters for the user context aggregator.
assistant_params: Parameters for the assistant context aggregator.
"""
self._user = LLMUserAggregator(context, params=user_params)
self._assistant = LLMAssistantAggregator(context, params=assistant_params)
def user(self) -> LLMUserAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> LLMAssistantAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant

View File

@@ -42,6 +42,7 @@ from pipecat.frames.frames import (
FunctionCallResultFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
@@ -916,7 +917,10 @@ class RTVIObserver(BaseObserver):
and self._params.user_transcription_enabled
):
await self._handle_user_transcriptions(frame)
elif isinstance(frame, OpenAILLMContextFrame) and self._params.user_llm_enabled:
elif (
isinstance(frame, (OpenAILLMContextFrame, LLMContextFrame))
and self._params.user_llm_enabled
):
await self._handle_context(frame)
elif isinstance(frame, LLMFullResponseStartFrame) and self._params.bot_llm_enabled:
await self.push_transport_message_urgent(RTVIBotLLMStartedMessage())
@@ -1017,16 +1021,20 @@ class RTVIObserver(BaseObserver):
if message:
await self.push_transport_message_urgent(message)
async def _handle_context(self, frame: OpenAILLMContextFrame):
async def _handle_context(self, frame: OpenAILLMContextFrame | LLMContextFrame):
"""Process LLM context frames to extract user messages for the RTVI client."""
try:
messages = frame.context.messages
if isinstance(frame, OpenAILLMContextFrame):
messages = frame.context.messages
else:
messages = frame.context.get_messages()
if not messages:
return
message = messages[-1]
# Handle Google LLM format (protobuf objects with attributes)
# Note: not possible if frame is a universal LLMContextFrame
if hasattr(message, "role") and message.role == "user" and hasattr(message, "parts"):
text = "".join(part.text for part in message.parts if hasattr(part, "text"))
if text:

View File

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMContextFrame,
LLMEnablePromptCachingFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
@@ -41,6 +42,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
@@ -197,6 +199,46 @@ class AnthropicLLMService(LLMService):
response = await api_call(**params)
return response
async def run_inference(
self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context. If both are provided, the
one in the context takes precedence.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
messages = []
system = []
if isinstance(context, LLMContext):
# Future code will be something like this:
# adapter = self.get_llm_adapter()
# params: AnthropicLLMInvocationParams = adapter.get_llm_invocation_params(context)
# messages = params["messages"]
# system = params["system_instruction"]
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
else:
context = AnthropicLLMContext.upgrade_to_anthropic(context)
messages = context.messages
system = getattr(context, "system", None) or system_instruction
# LLM completion
response = await self._client.messages.create(
model=self.model_name,
messages=messages,
system=system,
max_tokens=8192,
stream=False,
)
return response.content[0].text
@property
def enable_prompt_caching_beta(self) -> bool:
"""Check if prompt caching beta feature is enabled.
@@ -408,6 +450,8 @@ class AnthropicLLMService(LLMService):
context = None
if isinstance(frame, OpenAILLMContextFrame):
context: "AnthropicLLMContext" = AnthropicLLMContext.upgrade_to_anthropic(frame.context)
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError("Universal LLMContext is not yet supported for Anthropic.")
elif isinstance(frame, LLMMessagesFrame):
context = AnthropicLLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):

View File

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
FunctionCallFromLLM,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
@@ -40,6 +41,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
@@ -789,6 +791,81 @@ class AWSBedrockLLMService(LLMService):
"""
return True
async def run_inference(
self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context. If both are provided, the
one in the context takes precedence.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
try:
messages = []
system = []
if isinstance(context, LLMContext):
# Future code will be something like this:
# adapter = self.get_llm_adapter()
# params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context)
# messages = params["messages"]
# system = params["system_instruction"]
raise NotImplementedError(
"Universal LLMContext is not yet supported for AWS Bedrock."
)
else:
context = AWSBedrockLLMContext.upgrade_to_bedrock(context)
messages = context.messages
system = getattr(context, "system", None) or system_instruction
# Determine if we're using Claude or Nova based on model ID
model_id = self.model_name
# Prepare request parameters
request_params = {
"modelId": model_id,
"messages": messages,
"inferenceConfig": {
"maxTokens": 8192,
"temperature": 0.7,
"topP": 0.9,
},
}
if system:
request_params["system"] = [{"text": system}]
async with self._aws_session.client(
service_name="bedrock-runtime", **self._aws_params
) as client:
# Call Bedrock without streaming
response = await client.converse(**request_params)
# Extract the response text
if (
"output" in response
and "message" in response["output"]
and "content" in response["output"]["message"]
):
content = response["output"]["message"]["content"]
if isinstance(content, list):
for item in content:
if item.get("text"):
return item["text"]
elif isinstance(content, str):
return content
return None
except Exception as e:
logger.error(f"Bedrock summary generation failed: {e}", exc_info=True)
return None
async def _create_converse_stream(self, client, request_params):
"""Create converse stream with optional timeout and retry.
@@ -1044,6 +1121,8 @@ class AWSBedrockLLMService(LLMService):
context = None
if isinstance(frame, OpenAILLMContextFrame):
context = AWSBedrockLLMContext.upgrade_to_bedrock(frame.context)
if isinstance(frame, LLMContextFrame):
raise NotImplementedError("Universal LLMContext is not yet supported for AWS Bedrock.")
elif isinstance(frame, LLMMessagesFrame):
context = AWSBedrockLLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):

View File

@@ -34,6 +34,7 @@ from pipecat.frames.frames import (
FunctionCallFromLLM,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMTextFrame,
@@ -322,6 +323,10 @@ class AWSNovaSonicLLMService(LLMService):
if isinstance(frame, OpenAILLMContextFrame):
await self._handle_context(frame.context)
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError(
"Universal LLMContext is not yet supported for AWS Nova Sonic."
)
elif isinstance(frame, InputAudioRawFrame):
await self._handle_input_audio_frame(frame)
elif isinstance(frame, BotStoppedSpeakingFrame):

View File

@@ -60,3 +60,12 @@ class AzureLLMService(OpenAILLMService):
azure_endpoint=self._endpoint,
api_version=self._api_version,
)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as Azure service does yet not support universal LLMContext.
"""
return False

View File

@@ -9,9 +9,8 @@
from typing import List
from loguru import logger
from openai.types.chat import ChatCompletionMessageParam
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.services.openai.llm import OpenAILLMService
@@ -54,25 +53,40 @@ class CerebrasLLMService(OpenAILLMService):
logger.debug(f"Creating Cerebras client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for Cerebras chat completion request.
Cerebras supports a subset of OpenAI parameters, focusing on core
completion settings without advanced features like frequency/presence penalties.
Args:
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
"""
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"tools": context.tools,
"tool_choice": context.tool_choice,
"seed": self._settings["seed"],
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
"max_completion_tokens": self._settings["max_completion_tokens"],
}
# Messages, tools, tool_choice
params.update(params_from_context)
params.update(self._settings["extra"])
return params
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as Cerebras service does not yet support universal LLMContext.
"""
return False

View File

@@ -9,9 +9,8 @@
from typing import List
from loguru import logger
from openai.types.chat import ChatCompletionMessageParam
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.services.openai.llm import OpenAILLMService
@@ -54,19 +53,22 @@ class DeepSeekLLMService(OpenAILLMService):
logger.debug(f"Creating DeepSeek client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
def _build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def _build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for DeepSeek chat completion request.
DeepSeek doesn't support some OpenAI parameters like seed and max_completion_tokens.
Args:
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
"""
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"tools": context.tools,
"tool_choice": context.tool_choice,
"stream_options": {"include_usage": True},
"frequency_penalty": self._settings["frequency_penalty"],
"presence_penalty": self._settings["presence_penalty"],
@@ -75,5 +77,17 @@ class DeepSeekLLMService(OpenAILLMService):
"max_tokens": self._settings["max_tokens"],
}
# Messages, tools, tool_choice
params.update(params_from_context)
params.update(self._settings["extra"])
return params
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as DeepSeekLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -9,9 +9,8 @@
from typing import List
from loguru import logger
from openai.types.chat import ChatCompletionMessageParam
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.services.openai.llm import OpenAILLMService
@@ -54,20 +53,23 @@ class FireworksLLMService(OpenAILLMService):
logger.debug(f"Creating Fireworks client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for Fireworks chat completion request.
Fireworks doesn't support some OpenAI parameters like seed, max_completion_tokens,
and stream_options.
Args:
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
"""
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"tools": context.tools,
"tool_choice": context.tool_choice,
"frequency_penalty": self._settings["frequency_penalty"],
"presence_penalty": self._settings["presence_penalty"],
"temperature": self._settings["temperature"],
@@ -75,5 +77,17 @@ class FireworksLLMService(OpenAILLMService):
"max_tokens": self._settings["max_tokens"],
}
# Messages, tools, tool_choice
params.update(params_from_context)
params.update(self._settings["extra"])
return params
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as FireworksLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -16,19 +16,20 @@ import json
import os
import uuid
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from typing import Any, AsyncIterator, Dict, List, Optional
from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
@@ -38,6 +39,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
@@ -67,6 +69,7 @@ try:
FunctionCall,
FunctionResponse,
GenerateContentConfig,
GenerateContentResponse,
HttpOptions,
Part,
)
@@ -418,7 +421,14 @@ class GoogleLLMContext(OpenAILLMContext):
role = message["role"]
content = message.get("content", [])
if role == "system":
self.system_message = content
# System instructions are returned as plain text
if isinstance(content, str):
self.system_message = content
elif isinstance(content, list):
# If content is a list, we assume it's a list of text parts, per the standard
self.system_message = " ".join(
part["text"] for part in content if part.get("type") == "text"
)
return None
elif role == "assistant":
role = "model"
@@ -436,11 +446,20 @@ class GoogleLLMContext(OpenAILLMContext):
)
elif role == "tool":
role = "model"
try:
response = json.loads(message["content"])
if isinstance(response, dict):
response_dict = response
else:
response_dict = {"value": response}
except Exception as e:
# Response might not be JSON-deserializable (e.g. plain text).
response_dict = {"value": message["content"]}
parts.append(
Part(
function_response=FunctionResponse(
name="tool_call_result", # seems to work to hard-code the same name every time
response=json.loads(message["content"]),
response=response_dict,
)
)
)
@@ -636,9 +655,8 @@ class GoogleLLMService(LLMService):
"""Google AI (Gemini) LLM service implementation.
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.
from an OpenAILLMContext or a universal LLMContext to the messages format
expected by the Google AI model.
"""
# Overriding the default adapter to use the Gemini one.
@@ -715,6 +733,50 @@ class GoogleLLMService(LLMService):
def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None):
self._client = genai.Client(api_key=api_key, http_options=http_options)
async def run_inference(
self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context. If both are provided, the
one in the context takes precedence.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
messages = []
system = []
if isinstance(context, LLMContext):
adapter = self.get_llm_adapter()
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
system = params["system_instruction"]
else:
context = GoogleLLMContext.upgrade_to_google(context)
messages = context.messages
system = getattr(context, "system_message", None) or system_instruction
generation_config = GenerateContentConfig(system_instruction=system)
# Use the new google-genai client's async method
response = await self._client.aio.models.generate_content(
model=self._model_name,
contents=messages,
config=generation_config,
)
# Extract text from response
if response.candidates and response.candidates[0].content:
for part in response.candidates[0].content.parts:
if part.text:
return part.text
return None
def needs_mcp_alternate_schema(self) -> bool:
"""Check if this LLM service requires alternate MCP schema.
@@ -740,8 +802,89 @@ class GoogleLLMService(LLMService):
except Exception as e:
logger.exception(f"Failed to unset thinking budget: {e}")
async def _stream_content(
self, params_from_context: GeminiLLMInvocationParams
) -> AsyncIterator[GenerateContentResponse]:
messages = params_from_context["messages"]
if (
params_from_context["system_instruction"]
and self._system_instruction != params_from_context["system_instruction"]
):
logger.debug(f"System instruction changed: {params_from_context['system_instruction']}")
self._system_instruction = params_from_context["system_instruction"]
tools = []
if params_from_context["tools"]:
tools = params_from_context["tools"]
elif self._tools:
tools = self._tools
tool_config = None
if self._tool_config:
tool_config = self._tool_config
# Filter out None values and create GenerationContentConfig
generation_params = {
k: v
for k, v in {
"system_instruction": self._system_instruction,
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
"top_k": self._settings["top_k"],
"max_output_tokens": self._settings["max_tokens"],
"tools": tools,
"tool_config": tool_config,
}.items()
if v is not None
}
if self._settings["extra"]:
generation_params.update(self._settings["extra"])
# possibly modify generation_params (in place) to set thinking to off by default
self._maybe_unset_thinking_budget(generation_params)
generation_config = (
GenerateContentConfig(**generation_params) if generation_params else None
)
await self.start_ttfb_metrics()
return await self._client.aio.models.generate_content_stream(
model=self._model_name,
contents=messages,
config=generation_config,
)
async def _stream_content_specific_context(
self, context: OpenAILLMContext
) -> AsyncIterator[GenerateContentResponse]:
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]"
)
params = GeminiLLMInvocationParams(
messages=context.messages,
system_instruction=context.system_message,
tools=context.tools,
)
return await self._stream_content(params)
async def _stream_content_universal_context(
self, context: LLMContext
) -> AsyncIterator[GenerateContentResponse]:
adapter = self.get_llm_adapter()
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]"
)
params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(context)
return await self._stream_content(params)
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
async def _process_context(self, context: OpenAILLMContext | LLMContext):
await self.push_frame(LLMFullResponseStartFrame())
prompt_tokens = 0
@@ -754,55 +897,11 @@ class GoogleLLMService(LLMService):
search_result = ""
try:
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
f"{self}: Generating chat [{context.get_messages_for_logging()}]"
)
messages = context.messages
if context.system_message and self._system_instruction != context.system_message:
logger.debug(f"System instruction changed: {context.system_message}")
self._system_instruction = context.system_message
tools = []
if context.tools:
tools = context.tools
elif self._tools:
tools = self._tools
tool_config = None
if self._tool_config:
tool_config = self._tool_config
# Filter out None values and create GenerationContentConfig
generation_params = {
k: v
for k, v in {
"system_instruction": self._system_instruction,
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
"top_k": self._settings["top_k"],
"max_output_tokens": self._settings["max_tokens"],
"tools": tools,
"tool_config": tool_config,
}.items()
if v is not None
}
if self._settings["extra"]:
generation_params.update(self._settings["extra"])
# possibly modify generation_params (in place) to set thinking to off by default
self._maybe_unset_thinking_budget(generation_params)
generation_config = (
GenerateContentConfig(**generation_params) if generation_params else None
)
await self.start_ttfb_metrics()
response = await self._client.aio.models.generate_content_stream(
model=self._model_name,
contents=messages,
config=generation_config,
# Generate content using either OpenAILLMContext or universal LLMContext
response = await (
self._stream_content_specific_context(context)
if isinstance(context, OpenAILLMContext)
else self._stream_content_universal_context(context)
)
function_calls = []
@@ -915,9 +1014,18 @@ class GoogleLLMService(LLMService):
if isinstance(frame, OpenAILLMContextFrame):
context = GoogleLLMContext.upgrade_to_google(frame.context)
elif isinstance(frame, LLMContextFrame):
# Handle universal (LLM-agnostic) LLM context frames
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
# NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal
# LLMContext with it
context = GoogleLLMContext(frame.messages)
elif isinstance(frame, VisionImageRawFrame):
# This is only useful in very simple pipelines because it creates
# a new context. Generally we want a context manager to catch
# UserImageRawFrames coming through the pipeline and add them
# to the context.
context = GoogleLLMContext()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text

View File

@@ -39,6 +39,10 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
Note: This service includes a workaround for a Google API bug where function
call indices may be incorrectly set to None, resulting in empty function names.
.. deprecated:: 0.0.82
GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version.
Use GoogleLLMService instead for better integration with Google's native API.
Reference:
https://ai.google.dev/gemini-api/docs/openai
"""
@@ -59,8 +63,26 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
model: Google model name to use (e.g., "gemini-2.0-flash").
**kwargs: Additional arguments passed to the parent OpenAILLMService.
"""
import warnings
warnings.warn(
"GoogleLLMOpenAIBetaService is deprecated and will be removed in a future version. "
"Use GoogleLLMService instead for better integration with Google's native API.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as GoogleLLMOpenAIBetaService does not yet support universal LLMContext.
"""
return False
async def _process_context(self, context: OpenAILLMContext):
functions_list = []
arguments_list = []
@@ -72,9 +94,9 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
await self.start_ttfb_metrics()
chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
context
)
chunk_stream: AsyncStream[
ChatCompletionChunk
] = await self._stream_chat_completions_specific_context(context)
async for chunk in chunk_stream:
if chunk.usage:

View File

@@ -139,3 +139,12 @@ class GoogleVertexLLMService(OpenAILLMService):
creds.refresh(Request()) # Ensure token is up-to-date, lifetime is 1 hour.
return creds.token
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as GoogleVertexLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -190,3 +190,12 @@ class GrokLLMService(OpenAILLMService):
user = OpenAIUserContextAggregator(context, params=user_params)
assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
return GrokContextAggregatorPair(_user=user, _assistant=assistant)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as GrokLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -49,3 +49,12 @@ class GroqLLMService(OpenAILLMService):
"""
logger.debug(f"Creating Groq client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as GroqLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -14,6 +14,7 @@ from typing import (
Awaitable,
Callable,
Dict,
List,
Mapping,
Optional,
Protocol,
@@ -40,6 +41,7 @@ from pipecat.frames.frames import (
StartInterruptionFrame,
UserImageRequestFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
@@ -88,7 +90,7 @@ class FunctionCallParams:
tool_call_id: str
arguments: Mapping[str, Any]
llm: "LLMService"
context: OpenAILLMContext
context: OpenAILLMContext | LLMContext
result_callback: FunctionCallResultCallback
@@ -129,7 +131,7 @@ class FunctionCallRunnerItem:
function_name: str
tool_call_id: str
arguments: Mapping[str, Any]
context: OpenAILLMContext
context: OpenAILLMContext | LLMContext
run_llm: Optional[bool] = None
@@ -189,6 +191,24 @@ class LLMService(AIService):
"""
return self._adapter
async def run_inference(
self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
Must be implemented by subclasses.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
raise NotImplementedError(f"run_inference() not supported by {self.__class__.__name__}")
def create_context_aggregator(
self,
context: OpenAILLMContext,
@@ -432,7 +452,9 @@ class LLMService(AIService):
else:
await self._sequential_runner_queue.put(runner_item)
async def _call_start_function(self, context: OpenAILLMContext, function_name: str):
async def _call_start_function(
self, context: OpenAILLMContext | LLMContext, function_name: str
):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](function_name, self, context)
elif None in self._start_callbacks.keys():

View File

@@ -47,6 +47,15 @@ class NimLLMService(OpenAILLMService):
self._has_reported_prompt_tokens = False
self._is_processing = False
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as NimLLMService does not yet support universal LLMContext.
"""
return False
async def _process_context(self, context: OpenAILLMContext):
"""Process a context through the LLM and accumulate token usage metrics.

View File

@@ -43,3 +43,12 @@ class OLLamaLLMService(OpenAILLMService):
"""
logger.debug(f"Creating Ollama client with api {base_url}")
return super().create_client(base_url=base_url, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as OLLamaLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Base OpenAI LLM service implementation."""
"""Base LLM service implementation for services that use the AsyncOpenAI client."""
import asyncio
import base64
@@ -23,8 +23,10 @@ from openai import (
from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
from pydantic import BaseModel, Field
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.frames.frames import (
Frame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
@@ -33,6 +35,7 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -45,10 +48,11 @@ from pipecat.utils.tracing.service_decorators import traced_llm
class BaseOpenAILLMService(LLMService):
"""Base class for all services that use the AsyncOpenAI client.
This service consumes OpenAILLMContextFrame frames, which contain a reference
to an OpenAILLMContext object. The context defines what is sent to the LLM for
completion, including user, assistant, and system messages, as well as tool
choices and function call configurations.
This service consumes OpenAILLMContextFrame or LLMContextFrame frames,
which contain a reference to an OpenAILLMContext or LLMContext object. The
context defines what is sent to the LLM for completion, including user,
assistant, and system messages, as well as tool choices and function call
configurations.
"""
class InputParams(BaseModel):
@@ -180,18 +184,19 @@ class BaseOpenAILLMService(LLMService):
return True
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
self, params_from_context: OpenAILLMInvocationParams
) -> AsyncStream[ChatCompletionChunk]:
"""Get streaming chat completions from OpenAI API with optional timeout and retry.
Args:
context: The LLM context containing tools and configuration.
messages: List of chat completion messages to send.
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Async stream of chat completion chunks.
"""
params = self.build_chat_completion_params(context, messages)
params = self.build_chat_completion_params(params_from_context)
if self._retry_on_timeout:
try:
@@ -208,16 +213,15 @@ class BaseOpenAILLMService(LLMService):
chunks = await self._client.chat.completions.create(**params)
return chunks
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for chat completion request.
Subclasses can override this to customize parameters for different providers.
Args:
context: The LLM context containing tools and configuration.
messages: List of chat completion messages to send.
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
@@ -225,9 +229,6 @@ class BaseOpenAILLMService(LLMService):
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"tools": context.tools,
"tool_choice": context.tool_choice,
"stream_options": {"include_usage": True},
"frequency_penalty": self._settings["frequency_penalty"],
"presence_penalty": self._settings["presence_penalty"],
@@ -238,13 +239,48 @@ class BaseOpenAILLMService(LLMService):
"max_completion_tokens": self._settings["max_completion_tokens"],
}
# Messages, tools, tool_choice
params.update(params_from_context)
params.update(self._settings["extra"])
return params
async def _stream_chat_completions(
async def run_inference(
self, context: LLMContext | OpenAILLMContext, system_instruction: Optional[str] = None
) -> Optional[str]:
"""Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
Args:
context: The LLM context containing conversation history.
system_instruction: Optional system instruction to guide the LLM's
behavior. You could also (again, optionally) provide a system
instruction directly in the context.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
if isinstance(context, LLMContext):
adapter = self.get_llm_adapter()
params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
messages = params["messages"]
else:
messages = context.messages
# LLM completion
response = await self._client.chat.completions.create(
model=self.model_name,
messages=messages,
stream=False,
)
return response.choices[0].message.content
async def _stream_chat_completions_specific_context(
self, context: OpenAILLMContext
) -> AsyncStream[ChatCompletionChunk]:
logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]")
logger.debug(
f"{self}: Generating chat from OpenAI context [{context.get_messages_for_logging()}]"
)
messages: List[ChatCompletionMessageParam] = context.get_messages()
@@ -263,12 +299,28 @@ class BaseOpenAILLMService(LLMService):
del message["data"]
del message["mime_type"]
chunks = await self.get_chat_completions(context, messages)
params = OpenAILLMInvocationParams(
messages=messages, tools=context.tools, tool_choice=context.tool_choice
)
chunks = await self.get_chat_completions(params)
return chunks
async def _stream_chat_completions_universal_context(
self, context: LLMContext
) -> AsyncStream[ChatCompletionChunk]:
adapter = self.get_llm_adapter()
logger.debug(
f"{self}: Generating chat from universal context [{adapter.get_messages_for_logging(context)}]"
)
params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
chunks = await self.get_chat_completions(params)
return chunks
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
async def _process_context(self, context: OpenAILLMContext | LLMContext):
functions_list = []
arguments_list = []
tool_id_list = []
@@ -279,8 +331,11 @@ class BaseOpenAILLMService(LLMService):
await self.start_ttfb_metrics()
chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
context
# Generate chat completions using either OpenAILLMContext or universal LLMContext
chunk_stream = await (
self._stream_chat_completions_specific_context(context)
if isinstance(context, OpenAILLMContext)
else self._stream_chat_completions_universal_context(context)
)
async for chunk in chunk_stream:
@@ -364,11 +419,24 @@ class BaseOpenAILLMService(LLMService):
await self.run_function_calls(function_calls)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
Whether service supports universal LLMContext.
"""
# Return True in subclasses that support universal LLMContext
# This property lets us gradually roll out support for universal
# LLMContext to OpenAI-like services in a controlled manner.
return False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for LLM completion requests.
Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
Handles OpenAILLMContextFrame, LLMContextFrame, LLMMessagesFrame,
VisionImageRawFrame, and LLMUpdateSettingsFrame to trigger LLM
completions and manage settings.
Args:
frame: The frame to process.
@@ -378,10 +446,26 @@ class BaseOpenAILLMService(LLMService):
context = None
if isinstance(frame, OpenAILLMContextFrame):
context: OpenAILLMContext = frame.context
# Handle OpenAI-specific context frames
context = frame.context
elif isinstance(frame, LLMContextFrame):
# Handle universal (LLM-agnostic) LLM context frames
if self.supports_universal_context:
context = frame.context
else:
raise NotImplementedError(
f"Universal LLMContext is not yet supported for {self.__class__.__name__}."
)
elif isinstance(frame, LLMMessagesFrame):
# NOTE: LLMMessagesFrame is deprecated, so we don't support the newer universal
# LLMContext with it
context = OpenAILLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):
# This is only useful in very simple pipelines because it creates
# a new context. Generally we want a context manager to catch
# UserImageRawFrames coming through the pipeline and add them
# to the context.
# TODO: support the newer universal LLMContext with a VisionImageRawFrame equivalent?
context = OpenAILLMContext()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text

View File

@@ -107,6 +107,15 @@ class OpenAILLMService(BaseOpenAILLMService):
assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
True, as OpenAI service supports universal LLMContext.
"""
return True
class OpenAIUserContextAggregator(LLMUserContextAggregator):
"""OpenAI-specific user context aggregator.

View File

@@ -23,6 +23,7 @@ from pipecat.frames.frames import (
Frame,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
@@ -343,6 +344,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
await self.reset_conversation()
# Run the LLM at next opportunity
await self._create_response()
elif isinstance(frame, LLMContextFrame):
raise NotImplementedError(
"Universal LLMContext is not yet supported for OpenAI Realtime."
)
elif isinstance(frame, InputAudioRawFrame):
if not self._audio_input_paused:
await self._send_user_audio(frame)

View File

@@ -13,9 +13,8 @@ enabling integration with OpenPipe's fine-tuning and monitoring capabilities.
from typing import Dict, List, Optional
from loguru import logger
from openai.types.chat import ChatCompletionMessageParam
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.services.openai.llm import OpenAILLMService
try:
@@ -86,22 +85,21 @@ class OpenPipeLLMService(OpenAILLMService):
)
return client
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for OpenPipe chat completion request.
Adds OpenPipe-specific logging and tagging parameters.
Args:
context: The LLM context containing tools and configuration.
messages: List of chat completion messages to send.
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
"""
# Start with base parameters
params = super().build_chat_completion_params(context, messages)
params = super().build_chat_completion_params(params_from_context)
# Add OpenPipe-specific parameters
params["openpipe"] = {
@@ -110,3 +108,12 @@ class OpenPipeLLMService(OpenAILLMService):
}
return params
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as OpenPipeLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -61,3 +61,12 @@ class OpenRouterLLMService(OpenAILLMService):
"""
logger.debug(f"Creating OpenRouter client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as OpenRouterLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -11,11 +11,9 @@ 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
from openai.types.chat import ChatCompletionMessageParam
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.llm import OpenAILLMService
@@ -53,17 +51,23 @@ class PerplexityLLMService(OpenAILLMService):
self._has_reported_prompt_tokens = False
self._is_processing = False
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for Perplexity chat completion request.
Perplexity uses a subset of OpenAI parameters and doesn't support tools.
Args:
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
"""
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"messages": params_from_context["messages"],
}
# Add OpenAI-compatible parameters if they're set
@@ -80,6 +84,15 @@ class PerplexityLLMService(OpenAILLMService):
return params
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as PerplexityLLMService does not yet support universal LLMContext.
"""
return False
async def _process_context(self, context: OpenAILLMContext):
"""Process a context through the LLM and accumulate token usage metrics.

View File

@@ -50,3 +50,12 @@ class QwenLLMService(OpenAILLMService):
"""
logger.debug(f"Creating Qwen client with base URL: {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as QwenLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -7,12 +7,13 @@
"""SambaNova LLM service implementation using OpenAI-compatible interface."""
import json
from typing import Any, Dict, List, Optional
from typing import Any, Dict, Optional
from loguru import logger
from openai import AsyncStream
from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
from openai.types.chat import ChatCompletionChunk
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.frames.frames import (
LLMTextFrame,
)
@@ -67,17 +68,16 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
logger.debug(f"Creating SambaNova client with API {base_url}")
return super().create_client(api_key, base_url, **kwargs)
def build_chat_completion_params(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> dict:
def build_chat_completion_params(self, params_from_context: OpenAILLMInvocationParams) -> dict:
"""Build parameters for SambaNova chat completion request.
SambaNova doesn't support some OpenAI parameters like frequency_penalty,
presence_penalty, and seed.
Args:
context: The LLM context containing tools and configuration.
messages: List of chat completion messages to send.
params_from_context: Parameters, derived from the LLM context, to
use for the chat completion. Contains messages, tools, and tool
choice.
Returns:
Dictionary of parameters for the chat completion request.
@@ -85,9 +85,6 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
params = {
"model": self.model_name,
"stream": True,
"messages": messages,
"tools": context.tools,
"tool_choice": context.tool_choice,
"stream_options": {"include_usage": True},
"temperature": self._settings["temperature"],
"top_p": self._settings["top_p"],
@@ -95,6 +92,9 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
"max_completion_tokens": self._settings["max_completion_tokens"],
}
# Messages, tools, tool_choice
params.update(params_from_context)
params.update(self._settings["extra"])
return params
@@ -122,9 +122,9 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
await self.start_ttfb_metrics()
chunk_stream: AsyncStream[ChatCompletionChunk] = await self._stream_chat_completions(
context
)
chunk_stream: AsyncStream[
ChatCompletionChunk
] = await self._stream_chat_completions_specific_context(context)
async for chunk in chunk_stream:
if chunk.usage:
@@ -210,3 +210,12 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore
)
await self.run_function_calls(function_calls)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as SambaNovaLLMService does not yet support universal LLMContext.
"""
return False

View File

@@ -49,3 +49,12 @@ class TogetherLLMService(OpenAILLMService):
"""
logger.debug(f"Creating Together.ai client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@property
def supports_universal_context(self) -> bool:
"""Check if this service supports universal LLMContext.
Returns:
False, as TogetherLLMService does not yet support universal LLMContext.
"""
return False