wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aws: use AWS prefix for all services

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-05-06 14:52:29 -07:00
parent ce1a72850b
commit a8405649d0
6 changed files with 74 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added ### Added
- Added new AWS services `AWSBedrockLLMService` and `AWSTranscribeSTTService`.
- Added `on_active_speaker_changed` event handler to the `DailyTransport` class. - Added `on_active_speaker_changed` event handler to the `DailyTransport` class.
- Added `enable_ssml_parsing` and `enable_logging` to `InputParams` in - 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 ### Deprecated
- `PollyTTSService` is now deprecated, use `AWSPollyTTSService` instead.
- Observer `on_push_frame(src, dst, frame, direction, timestamp)` is now - Observer `on_push_frame(src, dst, frame, direction, timestamp)` is now
deprecated, use `on_push_frame(data: FramePushed)` instead. deprecated, use `on_push_frame(data: FramePushed)` instead.

16
examples/foundational/07m-interruptible-aws.py
View File

@@ -14,9 +14,9 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.aws.llm import BedrockLLMService from pipecat.services.aws.llm import AWSBedrockLLMService
from pipecat.services.aws.stt import TranscribeSTTService from pipecat.services.aws.stt import AWSTranscribeSTTService
from pipecat.services.aws.tts import PollyTTSService from pipecat.services.aws.tts import AWSPollyTTSService
from pipecat.transcriptions.language import Language from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import TransportParams from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport 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 region="us-west-2", # only specific regions support generative TTS
voice_id="Joanna", voice_id="Joanna",
params=PollyTTSService.InputParams( params=AWSPollyTTSService.InputParams(
engine="generative", language=Language.EN_US, rate="1.1" engine="generative", language=Language.EN_US, rate="1.1"
), ),
) )
llm = BedrockLLMService( llm = AWSBedrockLLMService(
aws_region="us-west-2", aws_region="us-west-2",
model="us.anthropic.claude-3-5-haiku-20241022-v1:0", 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 = [ messages = [

94
src/pipecat/services/aws/llm.py
View File

@@ -57,18 +57,18 @@ except ModuleNotFoundError as e:
@dataclass @dataclass
class BedrockContextAggregatorPair: class AWSBedrockContextAggregatorPair:
_user: "BedrockUserContextAggregator" _user: "AWSBedrockUserContextAggregator"
_assistant: "BedrockAssistantContextAggregator" _assistant: "AWSBedrockAssistantContextAggregator"
def user(self) -> "BedrockUserContextAggregator": def user(self) -> "AWSBedrockUserContextAggregator":
return self._user return self._user
def assistant(self) -> "BedrockAssistantContextAggregator": def assistant(self) -> "AWSBedrockAssistantContextAggregator":
return self._assistant return self._assistant
class BedrockLLMContext(OpenAILLMContext): class AWSBedrockLLMContext(OpenAILLMContext):
def __init__( def __init__(
self, self,
messages: Optional[List[dict]] = None, messages: Optional[List[dict]] = None,
@@ -81,10 +81,10 @@ class BedrockLLMContext(OpenAILLMContext):
self.system = system self.system = system
@staticmethod @staticmethod
def upgrade_to_bedrock(obj: OpenAILLMContext) -> "BedrockLLMContext": def upgrade_to_bedrock(obj: OpenAILLMContext) -> "AWSBedrockLLMContext":
logger.debug(f"Upgrading to Bedrock: {obj}") logger.debug(f"Upgrading to AWS Bedrock: {obj}")
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, BedrockLLMContext): if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AWSBedrockLLMContext):
obj.__class__ = BedrockLLMContext obj.__class__ = AWSBedrockLLMContext
obj._restructure_from_openai_messages() obj._restructure_from_openai_messages()
else: else:
obj._restructure_from_bedrock_messages() obj._restructure_from_bedrock_messages()
@@ -103,13 +103,13 @@ class BedrockLLMContext(OpenAILLMContext):
return self return self
@classmethod @classmethod
def from_messages(cls, messages: List[dict]) -> "BedrockLLMContext": def from_messages(cls, messages: List[dict]) -> "AWSBedrockLLMContext":
self = cls(messages=messages) self = cls(messages=messages)
# self._restructure_from_openai_messages() # self._restructure_from_openai_messages()
return self return self
@classmethod @classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "BedrockLLMContext": def from_image_frame(cls, frame: VisionImageRawFrame) -> "AWSBedrockLLMContext":
context = cls() context = cls()
context.add_image_frame_message( context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text format=frame.format, size=frame.size, image=frame.image, text=frame.text
@@ -120,14 +120,14 @@ class BedrockLLMContext(OpenAILLMContext):
self._messages[:] = messages self._messages[:] = messages
# self._restructure_from_openai_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): 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. Handles text content and function calls for both user and assistant messages.
Args: Args:
obj: Message in Bedrock format: obj: Message in AWS Bedrock format:
{ {
"role": "user/assistant", "role": "user/assistant",
"content": [{"text": str} | {"toolUse": {...}} | {"toolResult": {...}}] "content": [{"text": str} | {"toolUse": {...}} | {"toolResult": {...}}]
@@ -208,7 +208,7 @@ class BedrockLLMContext(OpenAILLMContext):
return messages return messages
def from_standard_message(self, message): 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. Handles conversion of text content, tool calls, and tool results.
Empty text content is converted to "(empty)". Empty text content is converted to "(empty)".
@@ -222,7 +222,7 @@ class BedrockLLMContext(OpenAILLMContext):
} }
Returns: Returns:
Message in Bedrock format: Message in AWS Bedrock format:
{ {
"role": "user/assistant", "role": "user/assistant",
"content": [ "content": [
@@ -306,8 +306,9 @@ class BedrockLLMContext(OpenAILLMContext):
def add_message(self, message): def add_message(self, message):
try: try:
if self.messages: if self.messages:
# Bedrock requires that roles alternate. If this message's role is the same as the # AWS Bedrock requires that roles alternate. If this message's
# last message, we should add this message's content to the last message. # 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 self.messages[-1]["role"] == message["role"]:
# if the last message has just a content string, convert it to a list # if the last message has just a content string, convert it to a list
# in the proper format # in the proper format
@@ -327,8 +328,10 @@ class BedrockLLMContext(OpenAILLMContext):
logger.error(f"Error adding message: {e}") logger.error(f"Error adding message: {e}")
def _restructure_from_bedrock_messages(self): def _restructure_from_bedrock_messages(self):
"""Restructure messages in Bedrock format by handling system messages, """Restructure messages in AWS Bedrock format by handling system
merging consecutive messages with the same role, and ensuring proper content formatting. messages, merging consecutive messages with the same role, and ensuring
proper content formatting.
""" """
# Handle system message if present at the beginning # Handle system message if present at the beginning
logger.debug(f"_restructure_from_bedrock_messages: {self.messages}") logger.debug(f"_restructure_from_bedrock_messages: {self.messages}")
@@ -431,13 +434,13 @@ class BedrockLLMContext(OpenAILLMContext):
return json.dumps(msgs) return json.dumps(msgs)
class BedrockUserContextAggregator(LLMUserContextAggregator): class AWSBedrockUserContextAggregator(LLMUserContextAggregator):
pass pass
class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): class AWSBedrockAssistantContextAggregator(LLMAssistantContextAggregator):
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): 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( self._context.add_message(
{ {
"role": "assistant", "role": "assistant",
@@ -505,10 +508,13 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator):
) )
class BedrockLLMService(LLMService): class AWSBedrockLLMService(LLMService):
"""This class implements inference with AWS Bedrock models including Amazon Nova and Anthropic Claude. """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): class InputParams(BaseModel):
@@ -533,7 +539,7 @@ class BedrockLLMService(LLMService):
): ):
super().__init__(**kwargs) super().__init__(**kwargs)
# Initialize the Bedrock client # Initialize the AWS Bedrock client
if not client_config: if not client_config:
client_config = Config( client_config = Config(
connect_timeout=300, # 5 minutes connect_timeout=300, # 5 minutes
@@ -570,8 +576,8 @@ class BedrockLLMService(LLMService):
*, *,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> BedrockContextAggregatorPair: ) -> AWSBedrockContextAggregatorPair:
"""Create an instance of BedrockContextAggregatorPair from an """Create an instance of AWSBedrockContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided. assistant aggregators can be provided.
@@ -583,20 +589,20 @@ class BedrockLLMService(LLMService):
aggregator parameters. aggregator parameters.
Returns: 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 for the user and one for the assistant, encapsulated in an
BedrockContextAggregatorPair. AWSBedrockContextAggregatorPair.
""" """
context.set_llm_adapter(self.get_llm_adapter()) context.set_llm_adapter(self.get_llm_adapter())
if isinstance(context, OpenAILLMContext): if isinstance(context, OpenAILLMContext):
context = BedrockLLMContext.from_openai_context(context) context = AWSBedrockLLMContext.from_openai_context(context)
user = BedrockUserContextAggregator(context, params=user_params) user = AWSBedrockUserContextAggregator(context, params=user_params)
assistant = BedrockAssistantContextAggregator(context, params=assistant_params) assistant = AWSBedrockAssistantContextAggregator(context, params=assistant_params)
return BedrockContextAggregatorPair(_user=user, _assistant=assistant) return AWSBedrockContextAggregatorPair(_user=user, _assistant=assistant)
async def _process_context(self, context: BedrockLLMContext): async def _process_context(self, context: AWSBedrockLLMContext):
# Usage tracking # Usage tracking
prompt_tokens = 0 prompt_tokens = 0
completion_tokens = 0 completion_tokens = 0
@@ -609,10 +615,6 @@ class BedrockLLMService(LLMService):
await self.push_frame(LLMFullResponseStartFrame()) await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics() 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() await self.start_ttfb_metrics()
# Set up inference config # Set up inference config
@@ -657,9 +659,9 @@ class BedrockLLMService(LLMService):
if self._settings["latency"] in ["standard", "optimized"]: if self._settings["latency"] in ["standard", "optimized"]:
request_params["performanceConfig"] = {"latency": self._settings["latency"]} 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) response = self._client.converse_stream(**request_params)
await self.stop_ttfb_metrics() await self.stop_ttfb_metrics()
@@ -744,15 +746,15 @@ class BedrockLLMService(LLMService):
context = None context = None
if isinstance(frame, OpenAILLMContextFrame): if isinstance(frame, OpenAILLMContextFrame):
context = BedrockLLMContext.upgrade_to_bedrock(frame.context) context = AWSBedrockLLMContext.upgrade_to_bedrock(frame.context)
elif isinstance(frame, LLMMessagesFrame): elif isinstance(frame, LLMMessagesFrame):
context = BedrockLLMContext.from_messages(frame.messages) context = AWSBedrockLLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame): elif isinstance(frame, VisionImageRawFrame):
# This is only useful in very simple pipelines because it creates # This is only useful in very simple pipelines because it creates
# a new context. Generally we want a context manager to catch # a new context. Generally we want a context manager to catch
# UserImageRawFrames coming through the pipeline and add them # UserImageRawFrames coming through the pipeline and add them
# to the context. # to the context.
context = BedrockLLMContext.from_image_frame(frame) context = AWSBedrockLLMContext.from_image_frame(frame)
elif isinstance(frame, LLMUpdateSettingsFrame): elif isinstance(frame, LLMUpdateSettingsFrame):
await self._update_settings(frame.settings) await self._update_settings(frame.settings)
else: else:

2
src/pipecat/services/aws/stt.py
View File

@@ -35,7 +35,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}") raise Exception(f"Missing module: {e}")
class TranscribeSTTService(STTService): class AWSTranscribeSTTService(STTService):
def __init__( def __init__(
self, self,
*, *,

8
src/pipecat/services/aws/tts.py
View File

@@ -107,7 +107,7 @@ def language_to_aws_language(language: Language) -> Optional[str]:
return language_map.get(language) return language_map.get(language)
class PollyTTSService(TTSService): class AWSPollyTTSService(TTSService):
class InputParams(BaseModel): class InputParams(BaseModel):
engine: Optional[str] = None engine: Optional[str] = None
language: Optional[Language] = Language.EN language: Optional[Language] = Language.EN
@@ -190,7 +190,6 @@ class PollyTTSService(TTSService):
prosody_attrs.append(f"rate='{self._settings['rate']}'") prosody_attrs.append(f"rate='{self._settings['rate']}'")
if self._settings["volume"]: if self._settings["volume"]:
prosody_attrs.append(f"volume='{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: if prosody_attrs:
ssml += f"<prosody {' '.join(prosody_attrs)}>" ssml += f"<prosody {' '.join(prosody_attrs)}>"
@@ -269,7 +268,7 @@ class PollyTTSService(TTSService):
yield TTSStoppedFrame() yield TTSStoppedFrame()
class AWSTTSService(PollyTTSService): class PollyTTSService(AWSPollyTTSService):
def __init__(self, **kwargs): def __init__(self, **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
@@ -278,5 +277,6 @@ class AWSTTSService(PollyTTSService):
with warnings.catch_warnings(): with warnings.catch_warnings():
warnings.simplefilter("always") warnings.simplefilter("always")
warnings.warn( warnings.warn(
"'AWSTTSService' is deprecated, use 'PollyTTSService' instead.", DeprecationWarning "'PollyTTSService' is deprecated, use 'AWSPollyTTSService' instead.",
DeprecationWarning,
) )

18
tests/test_context_aggregators.py
View File

@@ -41,9 +41,9 @@ from pipecat.services.anthropic.llm import (
AnthropicUserContextAggregator, AnthropicUserContextAggregator,
) )
from pipecat.services.aws.llm import ( from pipecat.services.aws.llm import (
BedrockAssistantContextAggregator, AWSBedrockAssistantContextAggregator,
BedrockLLMContext, AWSBedrockLLMContext,
BedrockUserContextAggregator, AWSBedrockUserContextAggregator,
) )
from pipecat.services.google.llm import ( from pipecat.services.google.llm import (
GoogleAssistantContextAggregator, GoogleAssistantContextAggregator,
@@ -714,11 +714,11 @@ class TestAnthropicAssistantContextAggregator(
# #
class TestBedrockUserContextAggregator( class TestAWSBedrockUserContextAggregator(
BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase BaseTestUserContextAggregator, unittest.IsolatedAsyncioTestCase
): ):
CONTEXT_CLASS = BedrockLLMContext CONTEXT_CLASS = AWSBedrockLLMContext
AGGREGATOR_CLASS = BedrockUserContextAggregator AGGREGATOR_CLASS = AWSBedrockUserContextAggregator
def check_message_multi_content( def check_message_multi_content(
self, context: OpenAILLMContext, content_index: int, index: int, content: str self, context: OpenAILLMContext, content_index: int, index: int, content: str
@@ -727,11 +727,11 @@ class TestBedrockUserContextAggregator(
assert messages["content"][index]["text"] == content assert messages["content"][index]["text"] == content
class TestBedrockAssistantContextAggregator( class TestAWSBedrockAssistantContextAggregator(
BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase BaseTestAssistantContextAggreagator, unittest.IsolatedAsyncioTestCase
): ):
CONTEXT_CLASS = BedrockLLMContext CONTEXT_CLASS = AWSBedrockLLMContext
AGGREGATOR_CLASS = BedrockAssistantContextAggregator AGGREGATOR_CLASS = AWSBedrockAssistantContextAggregator
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame] EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
def check_message_multi_content( def check_message_multi_content(