diff --git a/CHANGELOG.md b/CHANGELOG.md index be8b5dba7..c882fa7c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Expanded support for universal `LLMContext` to the AWS Bedrock LLM service. + Using the universal `LLMContext` and associated `LLMContextAggregatorPair` is + a pre-requisite for using `LLMSwitcher` to switch between LLMs at runtime. + - Added video streaming support to `LiveKitTransport`. - Added `OpenAIRealtimeLLMService` and `AzureRealtimeLLMService` which provide diff --git a/examples/foundational/12d-describe-video-aws.py b/examples/foundational/12d-describe-video-aws.py index 7c0535169..69c25bc28 100644 --- a/examples/foundational/12d-describe-video-aws.py +++ b/examples/foundational/12d-describe-video-aws.py @@ -13,6 +13,7 @@ from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( Frame, + LLMContextFrame, TextFrame, TTSSpeakFrame, UserImageRawFrame, @@ -21,10 +22,7 @@ from pipecat.frames.frames import ( 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, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.user_response import UserResponseAggregator from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.runner.types import RunnerArguments @@ -73,14 +71,14 @@ class UserImageProcessor(FrameProcessor): if isinstance(frame, UserImageRawFrame): if frame.request and frame.request.context: # Note: AWS Bedrock does not yet support the universal LLMContext - context = OpenAILLMContext() + context = LLMContext() context.add_image_frame_message( image=frame.image, text=frame.request.context, size=frame.size, format=frame.format, ) - frame = OpenAILLMContextFrame(context) + frame = LLMContextFrame(context) await self.push_frame(frame) else: await self.push_frame(frame, direction) @@ -121,6 +119,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 new file mode 100644 index 000000000..87ad20076 --- /dev/null +++ b/examples/foundational/14aa-function-calling-aws-universal-context.py @@ -0,0 +1,214 @@ +# +# Copyright (c) 2024–2025, 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 LLMRunFrame +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.aws.llm import AWSBedrockLLMService +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +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 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 = 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), + ) + llm.register_function("get_weather", get_weather) + llm.register_function("get_image", get_image) + + 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", + }, + }, + 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]) + + 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 two tools: get_weather 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? + +If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise. + """ + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": "Start the conversation by introducing yourself."}, + ] + + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ] + ) + + 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([LLMRunFrame()]) + + @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() 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..2e5c2c62a 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[List[dict[str, Any]]] # [{"text": "system message"}] + messages: List[dict[str, Any]] + tools: List[dict[str, Any]] + tool_choice: LLMContextToolChoice class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): @@ -33,30 +45,239 @@ 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) + + # Convert any subsequent "system"-role messages to "user"-role + # messages, as AWS Bedrock doesn't support system input messages. + for message in messages: + if message["role"] == "system": + message["role"] = "user" + + # 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..b201f43aa 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, @@ -812,14 +815,10 @@ class AWSBedrockLLMService(LLMService): 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"] # [{"text": "system message"}] - raise NotImplementedError( - "Universal LLMContext is not yet supported for AWS Bedrock." - ) + adapter: AWSBedrockLLMAdapter = self.get_llm_adapter() + params: AWSBedrockLLMInvocationParams = adapter.get_llm_invocation_params(context) + messages = params["messages"] + system = params["system"] # [{"text": "system message"}] else: context = AWSBedrockLLMContext.upgrade_to_bedrock(context) messages = context.messages @@ -940,8 +939,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 +974,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 +990,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 +1011,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 +1019,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 +1037,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 +1154,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):