diff --git a/src/pipecat/adapters/__init__.py b/src/pipecat/adapters/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py new file mode 100644 index 000000000..c26722604 --- /dev/null +++ b/src/pipecat/adapters/base_llm_adapter.py @@ -0,0 +1,22 @@ +from abc import ABC, abstractmethod +from typing import Any, List, Union, cast + +from loguru import logger + +from pipecat.adapters.schemas.tools_schema import ToolsSchema + + +class BaseLLMAdapter(ABC): + @abstractmethod + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]: + """Converts tools to the provider's format.""" + pass + + def from_standard_tools(self, tools: Any) -> List[Any]: + if isinstance(tools, ToolsSchema): + logger.debug(f"Retrieving the tools using the adapter: {type(self)}") + return self.to_provider_tools_format(tools) + # Fallback to return the same tools in case they are not in a standard format + return tools + + # TODO: we can move the logic to also handle the Messages here diff --git a/src/pipecat/adapters/schemas/__init__.py b/src/pipecat/adapters/schemas/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/adapters/schemas/function_schema.py b/src/pipecat/adapters/schemas/function_schema.py new file mode 100644 index 000000000..f6e59cef1 --- /dev/null +++ b/src/pipecat/adapters/schemas/function_schema.py @@ -0,0 +1,55 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Any, Dict, List + + +class FunctionSchema: + def __init__( + self, name: str, description: str, properties: Dict[str, Any], required: List[str] + ) -> None: + """Standardized function schema representation. + + :param name: Name of the function. + :param description: Description of the function. + :param properties: Dictionary defining properties types and descriptions. + :param required: List of required parameters. + """ + self._name = name + self._description = description + self._properties = properties + self._required = required + + def to_default_dict(self) -> Dict[str, Any]: + """Converts the function schema to a dictionary. + + :return: Dictionary representation of the function schema. + """ + return { + "name": self._name, + "description": self._description, + "parameters": { + "type": "object", + "properties": self._properties, + "required": self._required, + }, + } + + @property + def name(self) -> str: + return self._name + + @property + def description(self) -> str: + return self._description + + @property + def properties(self) -> Dict[str, Any]: + return self._properties + + @property + def required(self) -> List[str]: + return self._required diff --git a/src/pipecat/adapters/schemas/tools_schema.py b/src/pipecat/adapters/schemas/tools_schema.py new file mode 100644 index 000000000..5720535c5 --- /dev/null +++ b/src/pipecat/adapters/schemas/tools_schema.py @@ -0,0 +1,43 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from enum import Enum +from typing import Any, Dict, List + +from pipecat.adapters.schemas.function_schema import FunctionSchema + + +class AdapterType(Enum): + GEMINI = "gemini" # that is the only service where we are able to add custom tools for now + + +class ToolsSchema: + def __init__( + self, + standard_tools: List[FunctionSchema], + custom_tools: Dict[AdapterType, List[Dict[str, Any]]] = None, + ) -> None: + """ + A schema for tools that includes both standardized function schemas + and custom tools that do not follow the FunctionSchema format. + + :param standard_tools: List of tools following FunctionSchema. + :param custom_tools: List of tools in a custom format (e.g., search_tool). + """ + self._standard_tools = standard_tools + self._custom_tools = custom_tools + + @property + def standard_tools(self) -> List[FunctionSchema]: + return self._standard_tools + + @property + def custom_tools(self) -> Dict[AdapterType, List[Dict[str, Any]]]: + return self._custom_tools + + @custom_tools.setter + def custom_tools(self, value: Dict[AdapterType, List[Dict[str, Any]]]) -> None: + self._custom_tools = value diff --git a/src/pipecat/adapters/services/__init__.py b/src/pipecat/adapters/services/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py new file mode 100644 index 000000000..a699469d3 --- /dev/null +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -0,0 +1,34 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Any, Dict, List, 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 + + +class AnthropicLLMAdapter(BaseLLMAdapter): + @staticmethod + def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]: + return { + "name": function.name, + "description": function.description, + "input_schema": { + "type": "object", + "properties": function.properties, + "required": function.required, + }, + } + + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]: + """Converts function schemas to Anthropic's function-calling format. + + :return: Anthropic formatted function call definition. + """ + + functions_schema = tools_schema.standard_tools + return [self._to_anthropic_function_format(func) for func in functions_schema] diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py new file mode 100644 index 000000000..8efca5189 --- /dev/null +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -0,0 +1,28 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Any, Dict, List, Union + +from pipecat.adapters.base_llm_adapter import BaseLLMAdapter +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema + + +class GeminiLLMAdapter(BaseLLMAdapter): + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]: + """Converts function schemas to Gemini's function-calling format. + + :return: Gemini formatted function call definition. + """ + + functions_schema = tools_schema.standard_tools + formatted_standard_tools = [ + {"function_declarations": [func.to_default_dict() for func in functions_schema]} + ] + custom_gemini_tools = [] + if tools_schema.custom_tools: + custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, []) + + return formatted_standard_tools + custom_gemini_tools diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py new file mode 100644 index 000000000..909e5103a --- /dev/null +++ b/src/pipecat/adapters/services/open_ai_adapter.py @@ -0,0 +1,24 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# +from typing import List + +from openai.types.chat import ChatCompletionToolParam + +from pipecat.adapters.base_llm_adapter import BaseLLMAdapter +from pipecat.adapters.schemas.tools_schema import ToolsSchema + + +class OpenAILLMAdapter(BaseLLMAdapter): + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]: + """Converts function schemas to OpenAI's function-calling format. + + :return: OpenAI formatted function call definition. + """ + functions_schema = tools_schema.standard_tools + return [ + ChatCompletionToolParam(type="function", function=func.to_default_dict()) + for func in functions_schema + ] diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py new file mode 100644 index 000000000..b7eafaa81 --- /dev/null +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -0,0 +1,34 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# +from typing import Any, Dict, List, 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 + + +class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): + @staticmethod + def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]: + return { + "type": "function", + "name": function.name, + "description": function.description, + "parameters": { + "type": "object", + "properties": function.properties, + "required": function.required, + }, + } + + def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]: + """Converts function schemas to Openai Realtime function-calling format. + + :return: Openai Realtime formatted function call definition. + """ + + functions_schema = tools_schema.standard_tools + return [self._to_openai_realtime_function_format(func) for func in functions_schema] diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 5ef8c090f..e8391d62b 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -20,6 +20,8 @@ from openai.types.chat import ( ) from PIL import Image +from pipecat.adapters.base_llm_adapter import BaseLLMAdapter +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.frames.frames import ( AudioRawFrame, Frame, @@ -44,13 +46,20 @@ class OpenAILLMContext: def __init__( self, messages: Optional[List[ChatCompletionMessageParam]] = None, - tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN, + tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = NOT_GIVEN, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN, ): self._messages: List[ChatCompletionMessageParam] = messages if messages else [] self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice - self._tools: List[ChatCompletionToolParam] | NotGiven = tools + self._tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = tools self._user_image_request_context = {} + self._llm_adapter: Optional[BaseLLMAdapter] = None + + def get_llm_adapter(self) -> Optional[BaseLLMAdapter]: + return self._llm_adapter + + def set_llm_adapter(self, llm_adapter: BaseLLMAdapter): + self._llm_adapter = llm_adapter @staticmethod def from_messages(messages: List[dict]) -> "OpenAILLMContext": @@ -67,7 +76,9 @@ class OpenAILLMContext: return self._messages @property - def tools(self) -> List[ChatCompletionToolParam] | NotGiven: + def tools(self) -> List[ChatCompletionToolParam] | NotGiven | List[Any]: + if self._llm_adapter: + return self._llm_adapter.from_standard_tools(self._tools) return self._tools @property @@ -152,7 +163,7 @@ class OpenAILLMContext: def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven): self._tool_choice = tool_choice - def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN): + def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = NOT_GIVEN): if tools != NOT_GIVEN and len(tools) == 0: tools = NOT_GIVEN self._tools = tools diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 136afb47a..65c9b5d92 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -8,10 +8,12 @@ import asyncio import io import wave from abc import abstractmethod -from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Tuple +from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Tuple, Type from loguru import logger +from pipecat.adapters.base_llm_adapter import BaseLLMAdapter +from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, @@ -137,10 +139,23 @@ class AIService(FrameProcessor): class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" + # OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations. + # However, subclasses should override this with a more specific adapter when necessary. + adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter + def __init__(self, **kwargs): super().__init__(**kwargs) self._callbacks = {} self._start_callbacks = {} + self._adapter = self.adapter_class() + + def get_llm_adapter(self) -> BaseLLMAdapter: + return self._adapter + + def create_context_aggregator( + self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + ) -> Any: + pass self._register_event_handler("on_completion_timeout") diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index bae780e62..10a2ab7b7 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -18,6 +18,7 @@ from loguru import logger from PIL import Image from pydantic import BaseModel, Field +from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter from pipecat.frames.frames import ( Frame, FunctionCallInProgressFrame, @@ -85,6 +86,9 @@ class AnthropicLLMService(LLMService): use `AsyncAnthropicBedrock` and `AsyncAnthropicVertex` clients """ + # Overriding the default adapter to use the Anthropic one. + adapter_class = AnthropicLLMAdapter + class InputParams(BaseModel): enable_prompt_caching_beta: Optional[bool] = False max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1) @@ -123,8 +127,8 @@ class AnthropicLLMService(LLMService): def enable_prompt_caching_beta(self) -> bool: return self._enable_prompt_caching_beta - @staticmethod def create_context_aggregator( + self, context: OpenAILLMContext, *, user_kwargs: Mapping[str, Any] = {}, @@ -149,6 +153,8 @@ class AnthropicLLMService(LLMService): AnthropicContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) + if isinstance(context, OpenAILLMContext): context = AnthropicLLMContext.from_openai_context(context) user = AnthropicUserContextAggregator(context, **user_kwargs) @@ -382,6 +388,7 @@ class AnthropicLLMContext(OpenAILLMContext): tools=openai_context.tools, tool_choice=openai_context.tool_choice, ) + self.set_llm_adapter(openai_context.get_llm_adapter()) self._restructure_from_openai_messages() return self diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 934117c52..ef49df329 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -9,12 +9,14 @@ import base64 import json from dataclasses import dataclass from enum import Enum -from typing import Any, Dict, List, Mapping, Optional +from typing import Any, Dict, List, Mapping, Optional, Union import websockets from loguru import logger from pydantic import BaseModel, Field +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, @@ -152,6 +154,9 @@ class InputParams(BaseModel): class GeminiMultimodalLiveLLMService(LLMService): + # Overriding the default adapter to use the Gemini one. + adapter_class = GeminiLLMAdapter + def __init__( self, *, @@ -162,7 +167,7 @@ class GeminiMultimodalLiveLLMService(LLMService): start_audio_paused: bool = False, start_video_paused: bool = False, system_instruction: Optional[str] = None, - tools: Optional[List[dict]] = None, + tools: Optional[Union[List[dict], ToolsSchema]] = None, transcribe_user_audio: bool = False, transcribe_model_audio: bool = False, params: InputParams = InputParams(), @@ -435,7 +440,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) if self._tools: logger.debug(f"Gemini is configuring to use tools{self._tools}") - config.setup.tools = self._tools + config.setup.tools = self.get_llm_adapter().from_standard_tools(self._tools) await self.send_client_event(config) except Exception as e: @@ -726,6 +731,8 @@ class GeminiMultimodalLiveLLMService(LLMService): encapsulated in an GeminiMultimodalLiveContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) + GeminiMultimodalLiveContext.upgrade(context) user = GeminiMultimodalLiveUserContextAggregator(context, **user_kwargs) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index cbbc73b47..1d914a9bb 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -15,6 +15,8 @@ from google.api_core.exceptions import DeadlineExceeded from openai import AsyncStream from openai.types.chat import ChatCompletionChunk +from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter + # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" @@ -950,6 +952,9 @@ class GoogleLLMService(LLMService): franca for all LLM services, so that it is easy to switch between different LLMs. """ + # Overriding the default adapter to use the Gemini one. + adapter_class = GeminiLLMAdapter + class InputParams(BaseModel): max_tokens: Optional[int] = Field(default=4096, ge=1) temperature: Optional[float] = Field(default=None, ge=0.0, le=2.0) @@ -1180,8 +1185,8 @@ class GoogleLLMService(LLMService): if context: await self._process_context(context) - @staticmethod def create_context_aggregator( + self, context: OpenAILLMContext, *, user_kwargs: Mapping[str, Any] = {}, @@ -1206,6 +1211,8 @@ class GoogleLLMService(LLMService): GoogleContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) + if isinstance(context, OpenAILLMContext): context = GoogleLLMContext.upgrade_to_google(context) user = GoogleUserContextAggregator(context, **user_kwargs) diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py index 1f1661cf4..cf7d74f59 100644 --- a/src/pipecat/services/grok.py +++ b/src/pipecat/services/grok.py @@ -206,8 +206,8 @@ class GrokLLMService(OpenAILLMService): if tokens.completion_tokens > self._completion_tokens: self._completion_tokens = tokens.completion_tokens - @staticmethod def create_context_aggregator( + self, context: OpenAILLMContext, *, user_kwargs: Mapping[str, Any] = {}, @@ -232,6 +232,8 @@ class GrokLLMService(OpenAILLMService): GrokContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) + user = OpenAIUserContextAggregator(context, **user_kwargs) assistant = GrokAssistantContextAggregator(context, **assistant_kwargs) return GrokContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 425882d6f..5a3a993aa 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -343,8 +343,8 @@ class OpenAILLMService(BaseOpenAILLMService): ): super().__init__(model=model, params=params, **kwargs) - @staticmethod def create_context_aggregator( + self, context: OpenAILLMContext, *, user_kwargs: Mapping[str, Any] = {}, @@ -369,6 +369,7 @@ class OpenAILLMService(BaseOpenAILLMService): OpenAIContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) user = OpenAIUserContextAggregator(context, **user_kwargs) assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs) return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 44ce45dd7..00f8cd840 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -12,6 +12,8 @@ from typing import Any, Mapping from loguru import logger +from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter + try: import websockets except ModuleNotFoundError as e: @@ -76,6 +78,9 @@ class OpenAIUnhandledFunctionException(Exception): class OpenAIRealtimeBetaLLMService(LLMService): + # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one. + adapter_class = OpenAIRealtimeLLMAdapter + def __init__( self, *, @@ -596,6 +601,8 @@ class OpenAIRealtimeBetaLLMService(LLMService): OpenAIContextAggregatorPair. """ + context.set_llm_adapter(self.get_llm_adapter()) + OpenAIRealtimeLLMContext.upgrade_to_realtime(context) user = OpenAIRealtimeUserContextAggregator(context, **user_kwargs)