diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e330b9c9..2eec61bca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new AWS services `AWSBedrockLLMService` and `AWSTranscribeSTTService`. + - Added `on_active_speaker_changed` event handler to the `DailyTransport` class. - Added `enable_ssml_parsing` and `enable_logging` to `InputParams` in @@ -25,6 +27,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `PollyTTSService` is now deprecated, use `AWSPollyTTSService` instead. + - Observer `on_push_frame(src, dst, frame, direction, timestamp)` is now deprecated, use `on_push_frame(data: FramePushed)` instead. diff --git a/examples/foundational/07m-interruptible-aws.py b/examples/foundational/07m-interruptible-aws.py index c88439c62..2ccc7b717 100644 --- a/examples/foundational/07m-interruptible-aws.py +++ b/examples/foundational/07m-interruptible-aws.py @@ -14,9 +14,9 @@ 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 BedrockLLMService -from pipecat.services.aws.stt import TranscribeSTTService -from pipecat.services.aws.tts import PollyTTSService +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 @@ -37,20 +37,20 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac ), ) - stt = TranscribeSTTService() + stt = AWSTranscribeSTTService() - tts = PollyTTSService( + tts = AWSPollyTTSService( region="us-west-2", # only specific regions support generative TTS voice_id="Joanna", - params=PollyTTSService.InputParams( + params=AWSPollyTTSService.InputParams( engine="generative", language=Language.EN_US, rate="1.1" ), ) - llm = BedrockLLMService( + llm = AWSBedrockLLMService( aws_region="us-west-2", model="us.anthropic.claude-3-5-haiku-20241022-v1:0", - params=BedrockLLMService.InputParams(temperature=0.8, latency="optimized"), + params=AWSBedrockLLMService.InputParams(temperature=0.8, latency="optimized"), ) messages = [ diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index cec0cc2e6..00a877c0f 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -57,18 +57,18 @@ except ModuleNotFoundError as e: @dataclass -class BedrockContextAggregatorPair: - _user: "BedrockUserContextAggregator" - _assistant: "BedrockAssistantContextAggregator" +class AWSBedrockContextAggregatorPair: + _user: "AWSBedrockUserContextAggregator" + _assistant: "AWSBedrockAssistantContextAggregator" - def user(self) -> "BedrockUserContextAggregator": + def user(self) -> "AWSBedrockUserContextAggregator": return self._user - def assistant(self) -> "BedrockAssistantContextAggregator": + def assistant(self) -> "AWSBedrockAssistantContextAggregator": return self._assistant -class BedrockLLMContext(OpenAILLMContext): +class AWSBedrockLLMContext(OpenAILLMContext): def __init__( self, messages: Optional[List[dict]] = None, @@ -81,10 +81,10 @@ class BedrockLLMContext(OpenAILLMContext): self.system = system @staticmethod - def upgrade_to_bedrock(obj: OpenAILLMContext) -> "BedrockLLMContext": - logger.debug(f"Upgrading to Bedrock: {obj}") - if isinstance(obj, OpenAILLMContext) and not isinstance(obj, BedrockLLMContext): - obj.__class__ = BedrockLLMContext + def upgrade_to_bedrock(obj: OpenAILLMContext) -> "AWSBedrockLLMContext": + logger.debug(f"Upgrading to AWS Bedrock: {obj}") + if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AWSBedrockLLMContext): + obj.__class__ = AWSBedrockLLMContext obj._restructure_from_openai_messages() else: obj._restructure_from_bedrock_messages() @@ -103,13 +103,13 @@ class BedrockLLMContext(OpenAILLMContext): return self @classmethod - def from_messages(cls, messages: List[dict]) -> "BedrockLLMContext": + def from_messages(cls, messages: List[dict]) -> "AWSBedrockLLMContext": self = cls(messages=messages) # self._restructure_from_openai_messages() return self @classmethod - def from_image_frame(cls, frame: VisionImageRawFrame) -> "BedrockLLMContext": + def from_image_frame(cls, frame: VisionImageRawFrame) -> "AWSBedrockLLMContext": context = cls() context.add_image_frame_message( format=frame.format, size=frame.size, image=frame.image, text=frame.text @@ -120,14 +120,14 @@ class BedrockLLMContext(OpenAILLMContext): self._messages[:] = messages # self._restructure_from_openai_messages() - # convert a message in Bedrock format into one or more messages in OpenAI format + # convert a message in AWS Bedrock format into one or more messages in OpenAI format def to_standard_messages(self, obj): - """Convert Bedrock message format to standard structured format. + """Convert AWS Bedrock message format to standard structured format. Handles text content and function calls for both user and assistant messages. Args: - obj: Message in Bedrock format: + obj: Message in AWS Bedrock format: { "role": "user/assistant", "content": [{"text": str} | {"toolUse": {...}} | {"toolResult": {...}}] @@ -208,7 +208,7 @@ class BedrockLLMContext(OpenAILLMContext): return messages def from_standard_message(self, message): - """Convert standard format message to Bedrock format. + """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)". @@ -222,7 +222,7 @@ class BedrockLLMContext(OpenAILLMContext): } Returns: - Message in Bedrock format: + Message in AWS Bedrock format: { "role": "user/assistant", "content": [ @@ -306,8 +306,9 @@ class BedrockLLMContext(OpenAILLMContext): def add_message(self, message): try: if self.messages: - # Bedrock requires that roles alternate. If this message's role is the same as the - # last message, we should add this message's content to the last message. + # AWS Bedrock requires that roles alternate. If this message's + # role is the same as the last message, we should add this + # message's content to the last message. if self.messages[-1]["role"] == message["role"]: # if the last message has just a content string, convert it to a list # in the proper format @@ -327,8 +328,10 @@ class BedrockLLMContext(OpenAILLMContext): logger.error(f"Error adding message: {e}") def _restructure_from_bedrock_messages(self): - """Restructure messages in Bedrock format by handling system messages, - merging consecutive messages with the same role, and ensuring proper content formatting. + """Restructure messages in AWS Bedrock format by handling system + messages, merging consecutive messages with the same role, and ensuring + proper content formatting. + """ # Handle system message if present at the beginning logger.debug(f"_restructure_from_bedrock_messages: {self.messages}") @@ -431,13 +434,13 @@ class BedrockLLMContext(OpenAILLMContext): return json.dumps(msgs) -class BedrockUserContextAggregator(LLMUserContextAggregator): +class AWSBedrockUserContextAggregator(LLMUserContextAggregator): pass -class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): +class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator): async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): - # Format tool use according to Bedrock API + # Format tool use according to AWS Bedrock API self._context.add_message( { "role": "assistant", @@ -505,10 +508,13 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): ) -class BedrockLLMService(LLMService): - """This class implements inference with AWS Bedrock models including Amazon Nova and Anthropic Claude. +class AWSBedrockLLMService(LLMService): + """This class implements inference with AWS Bedrock models including Amazon + Nova and Anthropic Claude. + + Requires AWS credentials to be configured in the environment or through + boto3 configuration. - Requires AWS credentials to be configured in the environment or through boto3 configuration. """ class InputParams(BaseModel): @@ -533,7 +539,7 @@ class BedrockLLMService(LLMService): ): super().__init__(**kwargs) - # Initialize the Bedrock client + # Initialize the AWS Bedrock client if not client_config: client_config = Config( connect_timeout=300, # 5 minutes @@ -570,8 +576,8 @@ class BedrockLLMService(LLMService): *, user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), - ) -> BedrockContextAggregatorPair: - """Create an instance of BedrockContextAggregatorPair from an + ) -> AWSBedrockContextAggregatorPair: + """Create an instance of AWSBedrockContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and assistant aggregators can be provided. @@ -583,20 +589,20 @@ class BedrockLLMService(LLMService): aggregator parameters. Returns: - BedrockContextAggregatorPair: A pair of context aggregators, one + AWSBedrockContextAggregatorPair: A pair of context aggregators, one for the user and one for the assistant, encapsulated in an - BedrockContextAggregatorPair. + AWSBedrockContextAggregatorPair. """ context.set_llm_adapter(self.get_llm_adapter()) if isinstance(context, OpenAILLMContext): - context = BedrockLLMContext.from_openai_context(context) + context = AWSBedrockLLMContext.from_openai_context(context) - user = BedrockUserContextAggregator(context, params=user_params) - assistant = BedrockAssistantContextAggregator(context, params=assistant_params) - return BedrockContextAggregatorPair(_user=user, _assistant=assistant) + user = AWSBedrockUserContextAggregator(context, params=user_params) + assistant = AWSBedrockAssistantContextAggregator(context, params=assistant_params) + return AWSBedrockContextAggregatorPair(_user=user, _assistant=assistant) - async def _process_context(self, context: BedrockLLMContext): + async def _process_context(self, context: AWSBedrockLLMContext): # Usage tracking prompt_tokens = 0 completion_tokens = 0 @@ -609,10 +615,6 @@ class BedrockLLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() - # logger.debug( - # f"{self}: Generating chat with Bedrock model {self.model_name} | [{context.get_messages_for_logging()}]" - # ) - await self.start_ttfb_metrics() # Set up inference config @@ -657,9 +659,9 @@ class BedrockLLMService(LLMService): if self._settings["latency"] in ["standard", "optimized"]: request_params["performanceConfig"] = {"latency": self._settings["latency"]} - logger.debug(f"Calling Bedrock model with: {request_params}") + logger.debug(f"Calling AWS Bedrock model with: {request_params}") - # Call Bedrock with streaming + # Call AWS Bedrock with streaming response = self._client.converse_stream(**request_params) await self.stop_ttfb_metrics() @@ -744,15 +746,15 @@ class BedrockLLMService(LLMService): context = None if isinstance(frame, OpenAILLMContextFrame): - context = BedrockLLMContext.upgrade_to_bedrock(frame.context) + context = AWSBedrockLLMContext.upgrade_to_bedrock(frame.context) elif isinstance(frame, LLMMessagesFrame): - context = BedrockLLMContext.from_messages(frame.messages) + context = AWSBedrockLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): # This is only useful in very simple pipelines because it creates # a new context. Generally we want a context manager to catch # UserImageRawFrames coming through the pipeline and add them # to the context. - context = BedrockLLMContext.from_image_frame(frame) + context = AWSBedrockLLMContext.from_image_frame(frame) elif isinstance(frame, LLMUpdateSettingsFrame): await self._update_settings(frame.settings) else: diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 0468ab31b..a02625f81 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -35,7 +35,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class TranscribeSTTService(STTService): +class AWSTranscribeSTTService(STTService): def __init__( self, *, diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index 0fdbb8273..40d746514 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -107,7 +107,7 @@ def language_to_aws_language(language: Language) -> Optional[str]: return language_map.get(language) -class PollyTTSService(TTSService): +class AWSPollyTTSService(TTSService): class InputParams(BaseModel): engine: Optional[str] = None language: Optional[Language] = Language.EN @@ -190,7 +190,6 @@ class PollyTTSService(TTSService): prosody_attrs.append(f"rate='{self._settings['rate']}'") if self._settings["volume"]: prosody_attrs.append(f"volume='{self._settings['volume']}'") - # logger.warning("Prosody tags are not supported for generative engine. Ignoring.") if prosody_attrs: ssml += f"" @@ -269,7 +268,7 @@ class PollyTTSService(TTSService): yield TTSStoppedFrame() -class AWSTTSService(PollyTTSService): +class PollyTTSService(AWSPollyTTSService): def __init__(self, **kwargs): super().__init__(**kwargs) @@ -278,5 +277,6 @@ class AWSTTSService(PollyTTSService): with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( - "'AWSTTSService' is deprecated, use 'PollyTTSService' instead.", DeprecationWarning + "'PollyTTSService' is deprecated, use 'AWSPollyTTSService' instead.", + DeprecationWarning, ) diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index cd84d476d..0f68110ce 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -41,9 +41,9 @@ from pipecat.services.anthropic.llm import ( AnthropicUserContextAggregator, ) from pipecat.services.aws.llm import ( - BedrockAssistantContextAggregator, - BedrockLLMContext, - BedrockUserContextAggregator, + AWSBedrockAssistantContextAggregator, + AWSBedrockLLMContext, + AWSBedrockUserContextAggregator, ) from pipecat.services.google.llm import ( GoogleAssistantContextAggregator, @@ -714,11 +714,11 @@ class TestAnthropicAssistantContextAggregator( # -class TestBedrockUserContextAggregator( +class TestAWSBedrockUserContextAggregator( BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase ): - CONTEXT_CLASS = BedrockLLMContext - AGGREGATOR_CLASS = BedrockUserContextAggregator + CONTEXT_CLASS = AWSBedrockLLMContext + AGGREGATOR_CLASS = AWSBedrockUserContextAggregator def check_message_multi_content( self, context: OpenAILLMContext, content_index: int, index: int, content: str @@ -727,11 +727,11 @@ class TestBedrockUserContextAggregator( assert messages["content"][index]["text"] == content -class TestBedrockAssistantContextAggregator( +class TestAWSBedrockAssistantContextAggregator( BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase ): - CONTEXT_CLASS = BedrockLLMContext - AGGREGATOR_CLASS = BedrockAssistantContextAggregator + CONTEXT_CLASS = AWSBedrockLLMContext + AGGREGATOR_CLASS = AWSBedrockAssistantContextAggregator EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame] def check_message_multi_content(