From 458549f7df9b8e3e1bf9474cdf4aeb5b65560cc1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 6 May 2025 21:07:09 -0700 Subject: [PATCH] AWSBedrockLLMService: fix function calling --- .../foundational/07m-interruptible-aws.py | 5 +- .../foundational/14r-function-calling-aws.py | 139 ++++++++++++++++++ .../adapters/services/anthropic_adapter.py | 2 +- .../adapters/services/bedrock_adapter.py | 2 +- src/pipecat/services/aws/llm.py | 11 +- tests/test_function_calling_adapters.py | 30 ++++ 6 files changed, 178 insertions(+), 11 deletions(-) create mode 100644 examples/foundational/14r-function-calling-aws.py diff --git a/examples/foundational/07m-interruptible-aws.py b/examples/foundational/07m-interruptible-aws.py index 2ccc7b717..bbcfe7313 100644 --- a/examples/foundational/07m-interruptible-aws.py +++ b/examples/foundational/07m-interruptible-aws.py @@ -17,7 +17,6 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.aws.llm import AWSBedrockLLMService from pipecat.services.aws.stt import AWSTranscribeSTTService from pipecat.services.aws.tts import AWSPollyTTSService -from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection @@ -42,9 +41,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac tts = AWSPollyTTSService( region="us-west-2", # only specific regions support generative TTS voice_id="Joanna", - params=AWSPollyTTSService.InputParams( - engine="generative", language=Language.EN_US, rate="1.1" - ), + params=AWSPollyTTSService.InputParams(engine="generative", rate="1.1"), ) llm = AWSBedrockLLMService( diff --git a/examples/foundational/14r-function-calling-aws.py b/examples/foundational/14r-function-calling-aws.py new file mode 100644 index 000000000..cf4859576 --- /dev/null +++ b/examples/foundational/14r-function-calling-aws.py @@ -0,0 +1,139 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +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.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.aws.llm import AWSBedrockLLMService +from pipecat.services.aws.stt import AWSTranscribeSTTService +from pipecat.services.aws.tts import AWSPollyTTSService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting bot") + + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + stt = AWSTranscribeSTTService() + + tts = AWSPollyTTSService( + region="us-west-2", # only specific regions support generative TTS + voice_id="Joanna", + params=AWSPollyTTSService.InputParams(engine="generative", rate="1.1"), + ) + + llm = AWSBedrockLLMService( + aws_region="us-west-2", + model="us.anthropic.claude-3-5-haiku-20241022-v1:0", + params=AWSBedrockLLMService.InputParams(temperature=0.8, latency="optimized"), + ) + + # 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) + + 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"], + ) + tools = ToolsSchema(standard_tools=[weather_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 = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + stt, + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "user", "content": "Please introduce yourself to the user."}) + 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") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index a699469d3..23197d3a8 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, Dict, List, Union +from typing import Any, Dict, List from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.function_schema import FunctionSchema diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index cfb2a5f27..113a6938d 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -11,7 +11,7 @@ from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema -class BedrockLLMAdapter(BaseLLMAdapter): +class AWSBedrockLLMAdapter(BaseLLMAdapter): @staticmethod def _to_bedrock_function_format(function: FunctionSchema) -> Dict[str, Any]: return { diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 00a877c0f..921d3c790 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -17,6 +17,7 @@ from loguru import logger from PIL import Image from pydantic import BaseModel, Field +from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.frames.frames import ( Frame, FunctionCallCancelFrame, @@ -92,7 +93,6 @@ class AWSBedrockLLMContext(OpenAILLMContext): @classmethod def from_openai_context(cls, openai_context: OpenAILLMContext): - logger.debug("from_openai_context called") self = cls( messages=openai_context.messages, tools=openai_context.tools, @@ -105,7 +105,7 @@ class AWSBedrockLLMContext(OpenAILLMContext): @classmethod def from_messages(cls, messages: List[dict]) -> "AWSBedrockLLMContext": self = cls(messages=messages) - # self._restructure_from_openai_messages() + self._restructure_from_openai_messages() return self @classmethod @@ -118,7 +118,7 @@ class AWSBedrockLLMContext(OpenAILLMContext): def set_messages(self, messages: List): self._messages[:] = messages - # self._restructure_from_openai_messages() + self._restructure_from_openai_messages() # convert a message in AWS Bedrock format into one or more messages in OpenAI format def to_standard_messages(self, obj): @@ -334,7 +334,6 @@ class AWSBedrockLLMContext(OpenAILLMContext): """ # Handle system message if present at the beginning - logger.debug(f"_restructure_from_bedrock_messages: {self.messages}") if self.messages and self.messages[0]["role"] == "system": if len(self.messages) == 1: self.messages[0]["role"] = "user" @@ -375,7 +374,6 @@ class AWSBedrockLLMContext(OpenAILLMContext): self.messages.extend(merged_messages) def _restructure_from_openai_messages(self): - logger.debug(f"_restructure_from_openai_messages: {self.messages}") # first, map across self._messages calling self.from_standard_message(m) to modify messages in place try: self._messages[:] = [self.from_standard_message(m) for m in self._messages] @@ -517,6 +515,9 @@ class AWSBedrockLLMService(LLMService): """ + # Overriding the default adapter to use the Anthropic one. + adapter_class = AWSBedrockLLMAdapter + class InputParams(BaseModel): max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1) temperature: Optional[float] = Field(default_factory=lambda: 0.7, ge=0.0, le=1.0) diff --git a/tests/test_function_calling_adapters.py b/tests/test_function_calling_adapters.py index 5d6dafce3..83640bb80 100644 --- a/tests/test_function_calling_adapters.py +++ b/tests/test_function_calling_adapters.py @@ -11,6 +11,7 @@ from openai.types.chat import ChatCompletionToolParam from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter +from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter @@ -174,3 +175,32 @@ class TestFunctionAdapters(unittest.TestCase): tools_def = self.tools_def tools_def.custom_tools = {AdapterType.GEMINI: [search_tool]} assert GeminiLLMAdapter().to_provider_tools_format(tools_def) == expected + + def test_bedrock_adapter(self): + """Test AWS Bedrock adapter format transformation.""" + expected = [ + { + "toolSpec": { + "name": "get_weather", + "description": "Get the weather in a given location", + "inputSchema": { + "json": { + "type": "object", + "properties": { + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use.", + }, + "location": { + "type": "string", + "description": "The city, e.g. San Francisco", + }, + }, + "required": ["location", "format"], + } + }, + } + } + ] + assert AWSBedrockLLMAdapter().to_provider_tools_format(self.tools_def) == expected