From 75f9914f49044fc518c028472b18123ac8bf0518 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 9 Sep 2025 12:15:56 -0400 Subject: [PATCH] Add support for universal LLMContext to AWS Bedrock LLM service --- .../foundational/12d-describe-video-aws.py | 3 + ...-function-calling-aws-universal-context.py | 5 +- .../adapters/services/anthropic_adapter.py | 11 +- .../adapters/services/bedrock_adapter.py | 241 +++++++++++++++++- src/pipecat/services/aws/llm.py | 63 +++-- 5 files changed, 283 insertions(+), 40 deletions(-) diff --git a/examples/foundational/12d-describe-video-aws.py b/examples/foundational/12d-describe-video-aws.py index 7c0535169..1a3105e8a 100644 --- a/examples/foundational/12d-describe-video-aws.py +++ b/examples/foundational/12d-describe-video-aws.py @@ -121,6 +121,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): aws = AWSBedrockLLMService( aws_region="us-west-2", model="us.anthropic.claude-3-7-sonnet-20250219-v1:0", + # Note: usually, prefer providing latency="optimized" param. + # Here we can't because AWS Bedrock doesn't support it for Claude 3.7, + # which we need for image input. params=AWSBedrockLLMService.InputParams(temperature=0.8), ) diff --git a/examples/foundational/14aa-function-calling-aws-universal-context.py b/examples/foundational/14aa-function-calling-aws-universal-context.py index 0095bde80..87ad20076 100644 --- a/examples/foundational/14aa-function-calling-aws-universal-context.py +++ b/examples/foundational/14aa-function-calling-aws-universal-context.py @@ -98,7 +98,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = AWSBedrockLLMService( aws_region="us-west-2", model="us.anthropic.claude-3-7-sonnet-20250219-v1:0", - params=AWSBedrockLLMService.InputParams(temperature=0.8, latency="optimized"), + # Note: usually, prefer providing latency="optimized" param. + # Here we can't because AWS Bedrock doesn't support it for Claude 3.7, + # which we need for image input. + params=AWSBedrockLLMService.InputParams(temperature=0.8), ) llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index b8761e552..a98475016 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -9,7 +9,7 @@ import copy import json from dataclasses import dataclass -from typing import Any, Dict, List, Optional, TypedDict +from typing import Any, Dict, List, TypedDict from anthropic import NOT_GIVEN, NotGiven from anthropic.types.message_param import MessageParam @@ -28,10 +28,7 @@ from pipecat.processors.aggregators.llm_context import ( 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. - """ + """Context-based parameters for invoking Anthropic's LLM API.""" system: str | NotGiven messages: List[MessageParam] @@ -50,8 +47,6 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): ) -> 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. enable_prompt_caching: Whether prompt caching should be enabled. @@ -76,8 +71,6 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): 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. diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index 0be2527b1..7f1b5138b 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -6,21 +6,33 @@ """AWS Bedrock LLM adapter for Pipecat.""" -from typing import Any, Dict, List, TypedDict +import base64 +import copy +import json +from dataclasses import dataclass +from typing import Any, Dict, List, Literal, Optional, TypedDict + +from loguru import logger 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 +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMContextToolChoice, + LLMSpecificMessage, + LLMStandardMessage, +) class AWSBedrockLLMInvocationParams(TypedDict): - """Context-based parameters for invoking AWS Bedrock's LLM API. + """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 + system: Optional[str] + messages: List[dict[str, Any]] + tools: List[dict[str, Any]] + tool_choice: LLMContextToolChoice class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): @@ -33,30 +45,233 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): 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.") + messages = self._from_universal_context_messages(self._get_messages(context)) + return { + "system": messages.system, + "messages": messages.messages, + # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) + "tools": self.from_standard_tools(context.tools) or [], + # To avoid refactoring in AWSBedrockLLMService, we just pass through tool_choice. + # Eventually (when we don't have to maintain the non-LLMContext code path) we should do + # the conversion to Bedrock's expected format here rather than in AWSBedrockLLMService. + "tool_choice": context.tool_choice, + } 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.") + # Get messages in Anthropic's format + messages = self._from_universal_context_messages(self._get_messages(context)).messages + + # Sanitize messages for logging + messages_for_logging = [] + for message in messages: + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item.get("image"): + item["image"]["source"]["bytes"] = "..." + messages_for_logging.append(msg) + return messages_for_logging + + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("anthropic") + + @dataclass + class ConvertedMessages: + """Container for Anthropic-formatted messages converted from universal context.""" + + messages: List[dict[str, Any]] + system: Optional[str] + + def _from_universal_context_messages( + self, universal_context_messages: List[LLMContextMessage] + ) -> ConvertedMessages: + system = None + messages = [] + + # first, map messages using self._from_universal_context_message(m) + try: + messages = [self._from_universal_context_message(m) for m in universal_context_messages] + except Exception as e: + logger.error(f"Error mapping messages: {e}") + + # See if we should pull the system message out of our messages list + if messages and messages[0]["role"] == "system": + system = messages[0]["content"] + messages.pop(0) + + # Merge consecutive messages with the same role. + i = 0 + while i < len(messages) - 1: + current_message = messages[i] + next_message = messages[i + 1] + if current_message["role"] == next_message["role"]: + # Convert content to list of dictionaries if it's a string + if isinstance(current_message["content"], str): + current_message["content"] = [ + {"type": "text", "text": current_message["content"]} + ] + if isinstance(next_message["content"], str): + next_message["content"] = [{"type": "text", "text": next_message["content"]}] + # Concatenate the content + current_message["content"].extend(next_message["content"]) + # Remove the next message from the list + messages.pop(i + 1) + else: + i += 1 + + # Avoid empty content in messages + for message in messages: + if isinstance(message["content"], str) and message["content"] == "": + message["content"] = "(empty)" + elif isinstance(message["content"], list) and len(message["content"]) == 0: + message["content"] = [{"type": "text", "text": "(empty)"}] + + return self.ConvertedMessages(messages=messages, system=system) + + def _from_universal_context_message(self, message: LLMContextMessage) -> dict[str, Any]: + if isinstance(message, LLMSpecificMessage): + return copy.deepcopy(message.message) + return self._from_standard_message(message) + + def _from_standard_message(self, message: LLMStandardMessage) -> dict[str, Any]: + """Convert standard format message to AWS Bedrock format. + + Handles conversion of text content, tool calls, and tool results. + Empty text content is converted to "(empty)". + + Args: + message: Message in standard format. + + Returns: + Message in AWS Bedrock format. + + Examples: + Standard format input:: + + { + "role": "assistant", + "tool_calls": [ + { + "id": "123", + "function": {"name": "search", "arguments": '{"q": "test"}'} + } + ] + } + + AWS Bedrock format output:: + + { + "role": "assistant", + "content": [ + { + "toolUse": { + "toolUseId": "123", + "name": "search", + "input": {"q": "test"} + } + } + ] + } + """ + message = copy.deepcopy(message) + if message["role"] == "tool": + # Try to parse the content as JSON if it looks like JSON + try: + if message["content"].strip().startswith("{") and message[ + "content" + ].strip().endswith("}"): + content_json = json.loads(message["content"]) + tool_result_content = [{"json": content_json}] + else: + tool_result_content = [{"text": message["content"]}] + except: + tool_result_content = [{"text": message["content"]}] + + return { + "role": "user", + "content": [ + { + "toolResult": { + "toolUseId": message["tool_call_id"], + "content": tool_result_content, + }, + }, + ], + } + + if message.get("tool_calls"): + tc = message["tool_calls"] + ret = {"role": "assistant", "content": []} + for tool_call in tc: + function = tool_call["function"] + arguments = json.loads(function["arguments"]) + new_tool_use = { + "toolUse": { + "toolUseId": tool_call["id"], + "name": function["name"], + "input": arguments, + } + } + ret["content"].append(new_tool_use) + return ret + + # Handle text content + content = message.get("content") + if isinstance(content, str): + if content == "": + return {"role": message["role"], "content": [{"text": "(empty)"}]} + else: + return {"role": message["role"], "content": [{"text": content}]} + elif isinstance(content, list): + new_content = [] + for item in content: + # fix empty text + if item.get("type", "") == "text": + text_content = item["text"] if item["text"] != "" else "(empty)" + new_content.append({"text": text_content}) + # handle image_url -> image conversion + if item["type"] == "image_url": + new_item = { + "image": { + "format": "jpeg", + "source": { + "bytes": base64.b64decode(item["image_url"]["url"].split(",")[1]) + }, + } + } + new_content.append(new_item) + # In the case where there's a single image in the list (like what + # would result from a UserImageRawFrame), ensure that the image + # comes before text + image_indices = [i for i, item in enumerate(new_content) if "image" in item] + text_indices = [i for i, item in enumerate(new_content) if "text" in item] + if len(image_indices) == 1 and text_indices: + img_idx = image_indices[0] + first_txt_idx = text_indices[0] + if img_idx > first_txt_idx: + # Move image before the first text + image_item = new_content.pop(img_idx) + new_content.insert(first_txt_idx, image_item) + return {"role": message["role"], "content": new_content} + + return message @staticmethod def _to_bedrock_function_format(function: FunctionSchema) -> Dict[str, Any]: diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 824b2b4fa..781ae6322 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -25,7 +25,10 @@ from loguru import logger from PIL import Image from pydantic import BaseModel, Field -from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter +from pipecat.adapters.services.bedrock_adapter import ( + AWSBedrockLLMAdapter, + AWSBedrockLLMInvocationParams, +) from pipecat.frames.frames import ( Frame, FunctionCallCancelFrame, @@ -940,8 +943,25 @@ class AWSBedrockLLMService(LLMService): } } + def _get_llm_invocation_params( + self, context: OpenAILLMContext | LLMContext + ) -> AWSBedrockLLMInvocationParams: + # Universal LLMContext + if isinstance(context, LLMContext): + adapter: AWSBedrockLLMAdapter = self.get_llm_adapter() + params = adapter.get_llm_invocation_params(context) + return params + + # AWS Bedrock-specific context + return AWSBedrockLLMInvocationParams( + system=getattr(context, "system", None), + messages=context.messages, + tools=context.tools or [], + tool_choice=context.tool_choice, + ) + @traced_llm - async def _process_context(self, context: AWSBedrockLLMContext): + async def _process_context(self, context: AWSBedrockLLMContext | LLMContext): # Usage tracking prompt_tokens = 0 completion_tokens = 0 @@ -958,6 +978,12 @@ class AWSBedrockLLMService(LLMService): await self.start_ttfb_metrics() + params_from_context = self._get_llm_invocation_params(context) + messages = params_from_context["messages"] + system = params_from_context["system"] + tools = params_from_context["tools"] + tool_choice = params_from_context["tool_choice"] + # Set up inference config inference_config = { "maxTokens": self._settings["max_tokens"], @@ -968,19 +994,18 @@ class AWSBedrockLLMService(LLMService): # Prepare request parameters request_params = { "modelId": self.model_name, - "messages": context.messages, + "messages": messages, "inferenceConfig": inference_config, "additionalModelRequestFields": self._settings["additional_model_request_fields"], } # Add system message - system = getattr(context, "system", None) if system: request_params["system"] = system # Check if messages contain tool use or tool result content blocks has_tool_content = False - for message in context.messages: + for message in messages: if isinstance(message.get("content"), list): for content_item in message["content"]: if "toolUse" in content_item or "toolResult" in content_item: @@ -990,7 +1015,6 @@ class AWSBedrockLLMService(LLMService): break # Handle tools: use current tools, or no-op if tool content exists but no current tools - tools = context.tools or [] if has_tool_content and not tools: tools = [self._create_no_op_tool()] using_noop_tool = True @@ -999,17 +1023,15 @@ class AWSBedrockLLMService(LLMService): tool_config = {"tools": tools} # Only add tool_choice if we have real tools (not just no-op) - if not using_noop_tool and context.tool_choice: - if context.tool_choice == "auto": + if not using_noop_tool and tool_choice: + if tool_choice == "auto": tool_config["toolChoice"] = {"auto": {}} - elif context.tool_choice == "none": + elif tool_choice == "none": # Skip adding toolChoice for "none" pass - elif ( - isinstance(context.tool_choice, dict) and "function" in context.tool_choice - ): + elif isinstance(tool_choice, dict) and "function" in tool_choice: tool_config["toolChoice"] = { - "tool": {"name": context.tool_choice["function"]["name"]} + "tool": {"name": tool_choice["function"]["name"]} } request_params["toolConfig"] = tool_config @@ -1019,9 +1041,16 @@ class AWSBedrockLLMService(LLMService): request_params["performanceConfig"] = {"latency": self._settings["latency"]} # Log request params with messages redacted for logging - log_params = dict(request_params) - log_params["messages"] = context.get_messages_for_logging() - logger.debug(f"Calling AWS Bedrock model with: {log_params}") + if isinstance(context, LLMContext): + adapter = self.get_llm_adapter() + context_type_for_logging = "universal" + messages_for_logging = adapter.get_messages_for_logging(context) + else: + context_type_for_logging = "LLM-specific" + messages_for_logging = context.get_messages_for_logging() + logger.debug( + f"{self}: Generating chat from {context_type_for_logging} context [{system}] | {messages_for_logging}" + ) async with self._aws_session.client( service_name="bedrock-runtime", **self._aws_params @@ -1129,7 +1158,7 @@ class AWSBedrockLLMService(LLMService): 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.") + context = frame.context elif isinstance(frame, LLMMessagesFrame): context = AWSBedrockLLMContext.from_messages(frame.messages) elif isinstance(frame, LLMUpdateSettingsFrame):