From a4b9db9e073aa21229b6abc2f76aac93419de775 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 6 May 2025 11:37:23 -0700 Subject: [PATCH] fix formatting --- .../foundational/07m-interruptible-aws.py | 38 ++--- .../adapters/services/bedrock_adapter.py | 2 +- src/pipecat/services/aws/llm.py | 134 ++++++++---------- src/pipecat/services/aws/stt.py | 4 +- src/pipecat/services/aws/tts.py | 6 +- 5 files changed, 85 insertions(+), 99 deletions(-) diff --git a/examples/foundational/07m-interruptible-aws.py b/examples/foundational/07m-interruptible-aws.py index d1fae6b5e..ddb8b222e 100644 --- a/examples/foundational/07m-interruptible-aws.py +++ b/examples/foundational/07m-interruptible-aws.py @@ -13,13 +13,13 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.aws.llm import BedrockLLMContext, BedrockLLMService +from pipecat.services.aws.stt import TranscribeSTTService +from pipecat.services.aws.tts import PollyTTSService from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection -from pipecat.services.aws.llm import BedrockLLMService, BedrockLLMContext -from pipecat.services.aws.stt import TranscribeSTTService -from pipecat.services.aws.tts import PollyTTSService load_dotenv(override=True) @@ -42,28 +42,26 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac region="us-west-2", # only specific regions support generative TTS voice_id="Joanna", params=PollyTTSService.InputParams( - engine="generative", - language=Language.EN_US, - rate="1.1" + engine="generative", language=Language.EN_US, rate="1.1" ), ) llm = BedrockLLMService( aws_region="us-west-2", model="us.anthropic.claude-3-5-haiku-20241022-v1:0", - params=BedrockLLMService.InputParams( - temperature=0.8, - latency="optimized" - ) + params=BedrockLLMService.InputParams(temperature=0.8, latency="optimized"), ) messages = [ - { - "role": "system", - "content": [{"text": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way."}], - }, - ] - ) + { + "role": "system", + "content": [ + { + "text": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way." + } + ], + }, + ] context = BedrockLLMContext(messages) context_aggregator = llm.create_context_aggregator(context) @@ -77,8 +75,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac tts, # TTS transport.output(), # Transport bot output context_aggregator.assistant(), # Assistant spoken responses - ] - ) + ] + ) task = PipelineTask( pipeline, @@ -94,7 +92,9 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac async def on_client_connected(transport, client): logger.info(f"Client connected") # Kick off the conversation. - messages.append({"role": "user", "content": [{"text": "Please introduce yourself to the user."}]}) + messages.append( + {"role": "user", "content": [{"text": "Please introduce yourself to the user."}]} + ) await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_client_disconnected") diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index 0aba6aba2..b877f01fc 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -24,7 +24,7 @@ class BedrockLLMAdapter(BaseLLMAdapter): "properties": function.properties, "required": function.required, }, - } + }, } } diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 7c3539f7a..3b9c1fedd 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -135,7 +135,7 @@ class BedrockLLMContext(OpenAILLMContext): """ role = obj.get("role") content = obj.get("content") - + if role == "assistant": if isinstance(content, str): return [{"role": role, "content": [{"type": "text", "text": content}]}] @@ -184,7 +184,7 @@ class BedrockLLMContext(OpenAILLMContext): result_content = json.dumps(content_item["json"]) else: result_content = tool_result["content"] - + tool_items.append( { "role": "tool", @@ -226,26 +226,28 @@ class BedrockLLMContext(OpenAILLMContext): 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('}'): + 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 + "content": tool_result_content, }, }, ], } - + if message.get("tool_calls"): tc = message["tool_calls"] ret = {"role": "assistant", "content": []} @@ -261,7 +263,7 @@ class BedrockLLMContext(OpenAILLMContext): } ret["content"].append(new_tool_use) return ret - + # Handle text content content = message.get("content") if isinstance(content, str): @@ -276,7 +278,7 @@ class BedrockLLMContext(OpenAILLMContext): text_content = item["text"] if item["text"] != "" else "(empty)" new_content.append({"text": text_content}) return {"role": message["role"], "content": new_content} - + return message def add_image_frame_message( @@ -287,15 +289,7 @@ class BedrockLLMContext(OpenAILLMContext): encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") # Image should be the first content block in the message - content = [ - { - "type": "image", - "format": "jpeg", - "source": { - "bytes": encoded_image - } - } - ] + content = [{"type": "image", "format": "jpeg", "source": {"bytes": encoded_image}}] if text: content.append({"text": text}) self.add_message({"role": "user", "content": content}) @@ -309,9 +303,7 @@ class BedrockLLMContext(OpenAILLMContext): # if the last message has just a content string, convert it to a list # in the proper format if isinstance(self.messages[-1]["content"], str): - self.messages[-1]["content"] = [ - {"text": self.messages[-1]["content"]} - ] + self.messages[-1]["content"] = [{"text": self.messages[-1]["content"]}] # if this message has just a content string, convert it to a list # in the proper format if isinstance(message["content"], str): @@ -326,7 +318,7 @@ 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, + """Restructure messages in 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 @@ -338,7 +330,7 @@ class BedrockLLMContext(OpenAILLMContext): system_content = self.messages.pop(0)["content"] if isinstance(system_content, str): system_content = [{"text": system_content}] - + if self.system: if isinstance(self.system, str): self.system = [{"text": self.system}] @@ -366,7 +358,7 @@ class BedrockLLMContext(OpenAILLMContext): merged_messages[-1]["content"].extend(msg["content"]) else: merged_messages.append(msg) - + self.messages.clear() self.messages.extend(merged_messages) @@ -452,7 +444,7 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): "toolUse": { "toolUseId": frame.tool_call_id, "name": frame.function_name, - "input": frame.arguments if frame.arguments else {} + "input": frame.arguments if frame.arguments else {}, } } ], @@ -465,11 +457,7 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): { "toolResult": { "toolUseId": frame.tool_call_id, - "content": [ - { - "text": "IN_PROGRESS" - } - ], + "content": [{"text": "IN_PROGRESS"}], } } ], @@ -517,9 +505,10 @@ class BedrockAssistantContextAggregator(LLMAssistantContextAggregator): class BedrockLLMService(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. """ + class InputParams(BaseModel): max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1) temperature: Optional[float] = Field(default_factory=lambda: 0.7, ge=0.0, le=1.0) @@ -541,34 +530,33 @@ class BedrockLLMService(LLMService): **kwargs, ): super().__init__(**kwargs) - + # Initialize the Bedrock client if not client_config: client_config = Config( connect_timeout=300, # 5 minutes - read_timeout=300, # 5 minutes - retries={'max_attempts': 3} + read_timeout=300, # 5 minutes + retries={"max_attempts": 3}, ) session = boto3.Session( aws_access_key_id=aws_access_key, aws_secret_access_key=aws_secret_key, aws_session_token=aws_session_token, - region_name=aws_region + region_name=aws_region, ) - self._client = session.client( - service_name='bedrock-runtime', - config=client_config - ) - + self._client = session.client(service_name="bedrock-runtime", config=client_config) + self.set_model_name(model) self._settings = { "max_tokens": params.max_tokens, "temperature": params.temperature, "top_p": params.top_p, "latency": params.latency, - "additional_model_request_fields": params.additional_model_request_fields if isinstance(params.additional_model_request_fields, dict) else {}, + "additional_model_request_fields": params.additional_model_request_fields + if isinstance(params.additional_model_request_fields, dict) + else {}, } - + logger.info(f"Using AWS Bedrock model: {model}") def can_generate_metrics(self) -> bool: @@ -603,7 +591,7 @@ class BedrockLLMService(LLMService): if isinstance(context, OpenAILLMContext) and not isinstance(context, BedrockLLMContext): context = BedrockLLMContext.from_openai_context(context) - + user = BedrockUserContextAggregator(context, **user_kwargs) assistant = BedrockAssistantContextAggregator(context, **assistant_kwargs) return BedrockContextAggregatorPair(_user=user, _assistant=assistant) @@ -626,31 +614,29 @@ class BedrockLLMService(LLMService): # ) await self.start_ttfb_metrics() - + # Set up inference config inference_config = { "maxTokens": self._settings["max_tokens"], "temperature": self._settings["temperature"], "topP": self._settings["top_p"], } - + # Prepare request parameters request_params = { "modelId": self.model_name, "messages": context.messages, "inferenceConfig": inference_config, - "additionalModelRequestFields": self._settings["additional_model_request_fields"] + "additionalModelRequestFields": self._settings["additional_model_request_fields"], } - + # Add system message request_params["system"] = context.system - + # Add tools if present if context.tools: - tool_config = { - "tools": context.tools - } - + tool_config = {"tools": context.tools} + # Add tool_choice if specified if context.tool_choice: if context.tool_choice == "auto": @@ -658,32 +644,30 @@ class BedrockLLMService(LLMService): elif context.tool_choice == "none": # Skip adding toolChoice for "none" pass - elif isinstance(context.tool_choice, dict) and "function" in context.tool_choice: + elif ( + isinstance(context.tool_choice, dict) and "function" in context.tool_choice + ): tool_config["toolChoice"] = { - "tool": { - "name": context.tool_choice["function"]["name"] - } + "tool": {"name": context.tool_choice["function"]["name"]} } - + request_params["toolConfig"] = tool_config - + # Add performance config if latency is specified 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}") - + # Call Bedrock with streaming response = self._client.converse_stream(**request_params) - + await self.stop_ttfb_metrics() - + # Process the streaming response tool_use_block = None json_accumulator = "" - + for event in response["stream"]: # Handle text content if "contentBlockDelta" in event: @@ -694,18 +678,20 @@ class BedrockLLMService(LLMService): elif "toolUse" in delta and "input" in delta["toolUse"]: # Handle partial JSON for tool use json_accumulator += delta["toolUse"]["input"] - completion_tokens_estimate += self._estimate_tokens(delta["toolUse"]["input"]) - + completion_tokens_estimate += self._estimate_tokens( + delta["toolUse"]["input"] + ) + # Handle tool use start elif "contentBlockStart" in event: - content_block_start = event["contentBlockStart"]['start'] + content_block_start = event["contentBlockStart"]["start"] if "toolUse" in content_block_start: tool_use_block = { "id": content_block_start["toolUse"].get("toolUseId", ""), - "name": content_block_start["toolUse"].get("name", "") + "name": content_block_start["toolUse"].get("name", ""), } json_accumulator = "" - + # Handle message completion with tool use elif "messageStop" in event and "stopReason" in event["messageStop"]: if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block: @@ -719,7 +705,7 @@ class BedrockLLMService(LLMService): ) except json.JSONDecodeError: logger.error(f"Failed to parse tool arguments: {json_accumulator}") - + # Handle usage metrics if available if "metadata" in event and "usage" in event["metadata"]: usage = event["metadata"]["usage"] @@ -750,7 +736,7 @@ class BedrockLLMService(LLMService): prompt_tokens=prompt_tokens, completion_tokens=comp_tokens, cache_read_input_tokens=cache_read_input_tokens, - cache_creation_input_tokens=cache_creation_input_tokens + cache_creation_input_tokens=cache_creation_input_tokens, ) async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -783,7 +769,7 @@ class BedrockLLMService(LLMService): prompt_tokens: int, completion_tokens: int, cache_read_input_tokens: int, - cache_creation_input_tokens: int + cache_creation_input_tokens: int, ): if prompt_tokens or completion_tokens: tokens = LLMTokenUsage( @@ -791,6 +777,6 @@ class BedrockLLMService(LLMService): completion_tokens=completion_tokens, total_tokens=prompt_tokens + completion_tokens, cache_read_input_tokens=cache_read_input_tokens, - cache_creation_input_tokens=cache_creation_input_tokens + cache_creation_input_tokens=cache_creation_input_tokens, ) await self.start_llm_usage_metrics(tokens) diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 08d74d484..d749eff0c 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -19,7 +19,7 @@ from pipecat.frames.frames import ( Frame, TranscriptionFrame, InterimTranscriptionFrame, - StartFrame + StartFrame, ) from pipecat.services.ai_services import STTService from pipecat.transcriptions.language import Language @@ -597,4 +597,4 @@ class TranscribeSTTService(STTService): except Exception as e: logger.error(f"Unexpected error in receive loop: {e}") finally: - logger.debug("Receive loop ended") \ No newline at end of file + logger.debug("Receive loop ended") diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index ed1230dd7..d61f74ab2 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -17,7 +17,7 @@ from pipecat.frames.frames import ( Frame, TTSAudioRawFrame, TTSStartedFrame, - TTSStoppedFrame + TTSStoppedFrame, ) from pipecat.services.ai_services import TTSService from pipecat.transcriptions.language import Language @@ -187,7 +187,7 @@ class PollyTTSService(TTSService): if self._settings["engine"] == "standard": if self._settings["pitch"]: prosody_attrs.append(f"pitch='{self._settings['pitch']}'") - + if self._settings["rate"]: prosody_attrs.append(f"rate='{self._settings['rate']}'") if self._settings["volume"]: @@ -195,7 +195,7 @@ class PollyTTSService(TTSService): # logger.warning("Prosody tags are not supported for generative engine. Ignoring.") if prosody_attrs: - ssml += f"" + ssml += f"" ssml += text