Merge pull request #2148 from pipecat-ai/filipi/aws_bedrock

Refactoring AWSBedrockLLMService to work async

CHANGELOG.md

@@ -25,6 +25,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Refactored `AWSBedrockLLMService` and `AWSPollyTTSService` to work
+  asynchronously using `aioboto3` instead of the `boto3` library.
+
 - The `UserIdleProcessor` now handles the scenario where function calls take
   longer than the idle timeout duration. This allows you to use the
   `UserIdleProcessor` in conjunction with function calls that take a while to

pyproject.toml

@@ -42,7 +42,7 @@ Website = "https://pipecat.ai"
 [project.optional-dependencies]
 anthropic = [ "anthropic~=0.49.0" ]
 assemblyai = [ "websockets~=13.1" ]
-aws = [ "boto3~=1.37.16", "websockets~=13.1" ]
+aws = [ "aioboto3~=15.0.0", "websockets~=13.1" ]
 aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2" ]
 azure = [ "azure-cognitiveservices-speech~=1.42.0"]
 cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ]

src/pipecat/services/aws/llm.py

@@ -55,7 +55,7 @@ from pipecat.services.llm_service import LLMService
 from pipecat.utils.tracing.service_decorators import traced_llm
 
 try:
-    import boto3
+    import aioboto3
     import httpx
     from botocore.config import Config
 except ModuleNotFoundError as e:
@@ -749,13 +749,17 @@ class AWSBedrockLLMService(LLMService):
                 read_timeout=300,  # 5 minutes
                 retries={"max_attempts": 3},
             )
-        session = boto3.Session(
+
-            aws_access_key_id=aws_access_key,
+        self._aws_session = aioboto3.Session()
-            aws_secret_access_key=aws_secret_key,
+
-            aws_session_token=aws_session_token,
+        # Store AWS session parameters for creating client in async context
-            region_name=aws_region,
+        self._aws_params = {
-        )
+            "aws_access_key_id": aws_access_key,
-        self._client = session.client(service_name="bedrock-runtime", config=client_config)
+            "aws_secret_access_key": aws_secret_key,
+            "aws_session_token": aws_session_token,
+            "region_name": aws_region,
+            "config": client_config,
+        }
 
         self.set_model_name(model)
         self._settings = {
@@ -903,8 +907,11 @@ class AWSBedrockLLMService(LLMService):
 
             logger.debug(f"Calling AWS Bedrock model with: {request_params}")
 
+            async with self._aws_session.client(
+                service_name="bedrock-runtime", **self._aws_params
+            ) as client:
                 # Call AWS Bedrock with streaming
-            response = self._client.converse_stream(**request_params)
+                response = await client.converse_stream(**request_params)
 
                 await self.stop_ttfb_metrics()
 
@@ -913,7 +920,8 @@ class AWSBedrockLLMService(LLMService):
                 json_accumulator = ""
 
                 function_calls = []
-            for event in response["stream"]:
+
+                async for event in response["stream"]:
                     self.reset_watchdog()
 
                     # Handle text content

src/pipecat/services/aws/tts.py

@@ -30,7 +30,7 @@ from pipecat.transcriptions.language import Language
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
-    import boto3
+    import aioboto3
     from botocore.exceptions import BotoCoreError, ClientError
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -177,13 +177,25 @@ class AWSPollyTTSService(TTSService):
 
         params = params or AWSPollyTTSService.InputParams()
 
-        self._polly_client = boto3.client(
+        # Get credentials from environment variables if not provided
-            "polly",
+        self._aws_params = {
-            aws_access_key_id=aws_access_key_id,
+            "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"),
-            aws_secret_access_key=api_key,
+            "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"),
-            aws_session_token=aws_session_token,
+            "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"),
-            region_name=region,
+            "region_name": region or os.getenv("AWS_REGION", "us-east-1"),
+        }
+
+        # Validate that we have the required credentials
+        if (
+            not self._aws_params["aws_access_key_id"]
+            or not self._aws_params["aws_secret_access_key"]
+        ):
+            raise ValueError(
+                "AWS credentials not found. Please provide them either through constructor parameters "
+                "or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables."
             )
+
+        self._aws_session = aioboto3.Session()
         self._settings = {
             "engine": params.engine,
             "language": self.language_to_service_language(params.language)
@@ -199,24 +211,6 @@ class AWSPollyTTSService(TTSService):
 
         self.set_voice(voice_id)
 
-        # Get credentials from environment variables if not provided
-        self._credentials = {
-            "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"),
-            "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"),
-            "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"),
-            "region": region or os.getenv("AWS_REGION", "us-east-1"),
-        }
-
-        # Validate that we have the required credentials
-        if (
-            not self._credentials["aws_access_key_id"]
-            or not self._credentials["aws_secret_access_key"]
-        ):
-            raise ValueError(
-                "AWS credentials not found. Please provide them either through constructor parameters "
-                "or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables."
-            )
-
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
 
@@ -279,14 +273,6 @@ class AWSPollyTTSService(TTSService):
         Yields:
             Frame: Audio frames containing the synthesized speech.
         """
-
-        def read_audio_data(**args):
-            response = self._polly_client.synthesize_speech(**args)
-            if "AudioStream" in response:
-                audio_data = response["AudioStream"].read()
-                return audio_data
-            return None
-
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
@@ -309,12 +295,15 @@ class AWSPollyTTSService(TTSService):
             # Filter out None values
             filtered_params = {k: v for k, v in params.items() if v is not None}
 
-            audio_data = await asyncio.to_thread(read_audio_data, **filtered_params)
+            async with self._aws_session.client("polly", **self._aws_params) as polly:
-
+                response = await polly.synthesize_speech(**filtered_params)
-            if not audio_data:
+                if "AudioStream" in response:
-                logger.error(f"{self} No audio data returned")
+                    # Get the streaming body and read it
-                yield None
+                    stream = response["AudioStream"]
-                return
+                    audio_data = await stream.read()
+                else:
+                    logger.error(f"{self} No audio stream in response")
+                    audio_data = None
 
                 audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate)
 
@@ -332,7 +321,6 @@ class AWSPollyTTSService(TTSService):
                         yield frame
 
                 yield TTSStoppedFrame()
-
         except (BotoCoreError, ClientError) as error:
             logger.exception(f"{self} error generating TTS: {error}")
             error_message = f"AWS Polly TTS error: {str(error)}"