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

Fix classes that subclass BaseLLMAdapter by adding placeholder stuff until support for universal LLMContext machinery comes to all LLM services

Browse Source
This commit is contained in:
Paul Kompfner
2025-08-18 10:18:09 -04:00
parent fa1f6f1c51
commit e3019261a5
6 changed files with 160 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/pipecat/adapters/base_llm_adapter.py
View File

@@ -22,7 +22,6 @@ from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any])
# TODO: fix everywhere we subclass BaseLLMAdapter...
class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
"""Abstract base class for LLM provider adapters.

42
src/pipecat/adapters/services/anthropic_adapter.py
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.

42
src/pipecat/adapters/services/aws_nova_sonic_adapter.py
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.

42
src/pipecat/adapters/services/bedrock_adapter.py
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.

2
src/pipecat/adapters/services/open_ai_adapter.py
View File

@@ -36,7 +36,7 @@ class OpenAILLMInvocationParams(TypedDict):
tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven
class OpenAILLMAdapter(BaseLLMAdapter):
class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
"""OpenAI-specific adapter for Pipecat.
Handles:

40
src/pipecat/adapters/services/open_ai_realtime_adapter.py
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.