diff --git a/pyrightconfig.json b/pyrightconfig.json index 9370d4387..7cbaf2ce3 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -7,14 +7,10 @@ "ignore": [ "tests", "src/pipecat/adapters/services/aws_nova_sonic_adapter.py", - "src/pipecat/adapters/services/bedrock_adapter.py", - "src/pipecat/adapters/services/gemini_adapter.py", "src/pipecat/adapters/services/grok_realtime_adapter.py", "src/pipecat/adapters/services/inworld_realtime_adapter.py", "src/pipecat/adapters/services/open_ai_adapter.py", - "src/pipecat/adapters/services/open_ai_realtime_adapter.py", "src/pipecat/adapters/services/open_ai_responses_adapter.py", - "src/pipecat/adapters/services/perplexity_adapter.py", "src/pipecat/audio/dtmf/utils.py", "src/pipecat/audio/filters/aic_filter.py", "src/pipecat/audio/filters/krisp_viva_filter.py", diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index bb1223880..9ff8082ed 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -10,7 +10,7 @@ import base64 import copy import json from dataclasses import dataclass -from typing import Any, TypedDict +from typing import Any, TypedDict, cast from loguru import logger @@ -68,16 +68,19 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): system_instruction, discard_context_system=True, ) - return { - "system": [{"text": effective_system}] if effective_system else None, - "messages": converted.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, - } + return cast( + AWSBedrockLLMInvocationParams, + { + "system": [{"text": effective_system}] if effective_system else None, + "messages": converted.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. @@ -213,35 +216,36 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): ] } """ - message = copy.deepcopy(message) - if message["role"] == "tool": + # ChatCompletionMessageParam (input) and the dict shape Bedrock expects + # are different — work with the deepcopied message as a plain dict for + # the transformations below. + msg = cast(dict[str, Any], copy.deepcopy(message)) + if msg["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"]) + if msg["content"].strip().startswith("{") and msg["content"].strip().endswith("}"): + content_json = json.loads(msg["content"]) tool_result_content = [{"json": content_json}] else: - tool_result_content = [{"text": message["content"]}] + tool_result_content = [{"text": msg["content"]}] except (json.JSONDecodeError, ValueError, AttributeError): - tool_result_content = [{"text": message["content"]}] + tool_result_content = [{"text": msg["content"]}] return { "role": "user", "content": [ { "toolResult": { - "toolUseId": message["tool_call_id"], + "toolUseId": msg["tool_call_id"], "content": tool_result_content, }, }, ], } - if message.get("tool_calls"): - tc = message["tool_calls"] - ret = {"role": "assistant", "content": []} + if msg.get("tool_calls"): + tc = msg["tool_calls"] + ret: dict[str, Any] = {"role": "assistant", "content": []} for tool_call in tc: function = tool_call["function"] arguments = json.loads(function["arguments"]) @@ -256,12 +260,12 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): return ret # Handle text content - content = message.get("content") + content = msg.get("content") if isinstance(content, str): if content == "": - return {"role": message["role"], "content": [{"text": "(empty)"}]} + return {"role": msg["role"], "content": [{"text": "(empty)"}]} else: - return {"role": message["role"], "content": [{"text": content}]} + return {"role": msg["role"], "content": [{"text": content}]} elif isinstance(content, list): new_content = [] for item in content: @@ -300,9 +304,9 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): # 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 {"role": msg["role"], "content": new_content} - return message + return msg @staticmethod def _to_bedrock_function_format(function: FunctionSchema) -> dict[str, Any]: diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index aede18e7c..4e9e20e14 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -9,7 +9,7 @@ import base64 import json from dataclasses import dataclass, field -from typing import Any, TypedDict +from typing import Any, TypedDict, cast from loguru import logger from openai import NotGiven @@ -154,9 +154,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): messages = self._from_universal_context_messages(self.get_messages(context)).messages # Sanitize messages for logging - messages_for_logging = [] + messages_for_logging: list[dict[str, Any]] = [] for message in messages: - obj = message.to_json_dict() + # `to_json_dict()` returns `dict[str, object]`; treat as a plain + # dict for the value indexing/mutation below. The broad `except` + # below is the safety net if any item isn't shaped as expected. + obj: dict[str, Any] = cast(dict[str, Any], message.to_json_dict()) try: if "parts" in obj: for part in obj["parts"]: @@ -274,7 +277,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): # Check if we only have function-related messages (no regular text) effective_system = extracted_system or system_instruction has_regular_messages = any( - len(msg.parts) == 1 + msg.parts is not None + and len(msg.parts) == 1 and getattr(msg.parts[0], "text", None) and not getattr(msg.parts[0], "function_call", None) and not getattr(msg.parts[0], "function_response", None) @@ -346,8 +350,11 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))] ) """ - role = message["role"] - content = message.get("content", []) + # ChatCompletionMessageParam (a union of TypedDicts) doesn't allow + # the dict-style key access used below; treat it as a plain dict. + msg = cast(dict[str, Any], message) + role = msg["role"] + content = msg.get("content", []) # Convert non-initial system/developer messages to user role, # as Gemini doesn't support these as input messages. @@ -359,8 +366,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): parts = [] tool_call_id_to_name_mapping = {} - if message.get("tool_calls"): - for tc in message["tool_calls"]: + if msg.get("tool_calls"): + for tc in msg["tool_calls"]: id = tc["id"] name = tc["function"]["name"] tool_call_id_to_name_mapping[id] = name @@ -376,7 +383,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): elif role == "tool": role = "user" try: - response = json.loads(message["content"]) + response = json.loads(msg["content"]) if isinstance(response, dict): response_dict = response else: @@ -384,10 +391,10 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): except Exception as e: # Response might not be JSON-deserializable. # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string. - response_dict = {"value": message["content"]} + response_dict = {"value": msg["content"]} # Get function name from mapping using tool_call_id, or fallback - tool_call_id = message.get("tool_call_id") + tool_call_id = msg.get("tool_call_id") function_name = "tool_call_result" # Default fallback if tool_call_id and tool_call_id in params.tool_call_id_to_name_mapping: function_name = params.tool_call_id_to_name_mapping[tool_call_id] @@ -491,7 +498,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): def is_tool_call_message(msg: Content) -> bool: """Check if message contains only function_call parts.""" - return ( + return bool( msg.role == "model" and msg.parts and all(getattr(part, "function_call", None) for part in msg.parts) @@ -499,6 +506,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): def message_has_thought_signature(msg: Content) -> bool: """Check if any part in the message has a thought_signature.""" + if msg.parts is None: + return False return any(getattr(part, "thought_signature", None) for part in msg.parts) merged_messages = [] @@ -564,6 +573,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): logger.debug(f"Thought signatures to apply: {len(thought_signature_dicts)}") for ts in thought_signature_dicts: bookmark = ts.get("bookmark") + if bookmark is None: + continue if bookmark.get("function_call"): logger.trace(f" - To function call: {bookmark['function_call']}") elif bookmark.get("text"): @@ -665,6 +676,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): if ( hasattr(part, "inline_data") and part.inline_data + and part.inline_data.data is not None + and bookmark_inline_data.data is not None # Comparing length should be good enough for matching inline data, # especially since we're already matching thought signatures in # strict message order. Comparing actual data is expensive. diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 7df7e45c5..fb555b039 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -9,7 +9,7 @@ import copy import json from dataclasses import dataclass -from typing import Any, TypedDict +from typing import Any, TypedDict, cast from loguru import logger @@ -81,7 +81,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages in a format ready for logging about OpenAI Realtime. """ - return self.get_messages(context, truncate_large_values=True) + return cast(list[dict[str, Any]], self.get_messages(context, truncate_large_values=True)) @dataclass class ConvertedMessages: @@ -101,12 +101,24 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): if not universal_context_messages: return self.ConvertedMessages(messages=[]) - messages = copy.deepcopy(universal_context_messages) + # NOTE: This adapter does not yet handle ``LLMSpecificMessage`` — those + # are filtered out below. Other adapters (e.g. Anthropic) dispatch + # LLMSpecific items through a per-provider passthrough. For OpenAI + # Realtime, the strategy here packs a multi-message history into a + # single text message (see comment further down), which doesn't + # compose with opaque per-provider payloads. If/when this adapter + # adopts the per-message strategy, LLMSpecific items can flow + # through `_from_universal_context_message` like in other adapters. + messages: list[dict[str, Any]] = [ + cast(dict[str, Any], m) + for m in copy.deepcopy(universal_context_messages) + if isinstance(m, dict) + ] system_instruction = None # If we have a "system" message as our first message, # pull that out into session "instructions" - if messages[0].get("role") == "system": + if messages and messages[0].get("role") == "system": system = messages.pop(0) content = system.get("content") if isinstance(content, str): @@ -124,7 +136,9 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): # If we have just a single "user" item, we can just send it normally if len(messages) == 1 and messages[0].get("role") == "user": return self.ConvertedMessages( - messages=[self._from_universal_context_message(messages[0])], + messages=[ + self._from_universal_context_message(cast(LLMContextMessage, messages[0])) + ], system_instruction=system_instruction, ) @@ -142,18 +156,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): return self.ConvertedMessages( messages=[ - { - "role": "user", - "type": "message", - "content": [ - { - "type": "input_text", - "text": "\n\n".join( + events.ConversationItem( + role="user", + type="message", + content=[ + events.ItemContent( + type="input_text", + text="\n\n".join( [intro_text, json.dumps(messages, indent=2), trailing_text] ), - } + ) ], - } + ) ], system_instruction=system_instruction, ) @@ -161,31 +175,34 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): def _from_universal_context_message( self, message: LLMContextMessage ) -> events.ConversationItem: - if message.get("role") == "user": - content = message.get("content") - if isinstance(message.get("content"), list): + # NOTE: ``LLMSpecificMessage`` is not yet handled here — see the + # corresponding note in `_from_universal_context_messages`. + msg = cast(dict[str, Any], message) + if msg.get("role") == "user": + content = msg.get("content") + if isinstance(content, list): content = "" - for c in message.get("content"): + for c in msg.get("content", []): if c.get("type") == "text": content += " " + c.get("text") else: logger.error( - f"Unhandled content type in context message: {c.get('type')} - {message}" + f"Unhandled content type in context message: {c.get('type')} - {msg}" ) return events.ConversationItem( role="user", type="message", content=[events.ItemContent(type="input_text", text=content)], ) - if message.get("role") == "assistant" and message.get("tool_calls"): - tc = message.get("tool_calls")[0] + if msg.get("role") == "assistant" and msg.get("tool_calls"): + tc = msg["tool_calls"][0] return events.ConversationItem( type="function_call", call_id=tc["id"], name=tc["function"]["name"], arguments=tc["function"]["arguments"], ) - logger.error(f"Unhandled message type in _from_universal_context_message: {message}") + raise ValueError(f"Unhandled message type in _from_universal_context_message: {msg}") @staticmethod def _to_openai_realtime_function_format(function: FunctionSchema) -> dict[str, Any]: diff --git a/src/pipecat/adapters/services/perplexity_adapter.py b/src/pipecat/adapters/services/perplexity_adapter.py index 188092b78..7e984d83e 100644 --- a/src/pipecat/adapters/services/perplexity_adapter.py +++ b/src/pipecat/adapters/services/perplexity_adapter.py @@ -28,6 +28,7 @@ the messages are sent to Perplexity's API. """ import copy +from typing import Any, cast from openai.types.chat import ChatCompletionMessageParam @@ -116,7 +117,11 @@ class PerplexityLLMAdapter(OpenAILLMAdapter): if not messages: return messages - messages = copy.deepcopy(messages) + # ChatCompletionMessageParam is a union of TypedDicts; the + # transformations below mutate by key/index in ways those TypedDicts + # don't permit. Work against a plain-dict view for the duration of + # the transformation and cast back at the return site. + msgs: list[dict[str, Any]] = cast(list[dict[str, Any]], copy.deepcopy(messages)) # Note: "developer" → "user" conversion is handled by the parent adapter # via the convert_developer_to_user parameter. @@ -125,10 +130,10 @@ class PerplexityLLMAdapter(OpenAILLMAdapter): # Perplexity allows system messages at the start, but rejects them # after any non-system message. in_initial_system_block = True - for i in range(len(messages)): - if messages[i].get("role") == "system": + for i in range(len(msgs)): + if msgs[i].get("role") == "system": if not in_initial_system_block: - messages[i]["role"] = "user" + msgs[i]["role"] = "user" else: in_initial_system_block = False @@ -137,9 +142,9 @@ class PerplexityLLMAdapter(OpenAILLMAdapter): # messages that violate Perplexity's strict alternation requirement. # Skip consecutive system messages at the start — Perplexity allows those. i = 0 - while i < len(messages) - 1: - current = messages[i] - next_msg = messages[i + 1] + while i < len(msgs) - 1: + current = msgs[i] + next_msg = msgs[i + 1] if current["role"] == next_msg["role"] == "system": # Perplexity allows multiple initial system messages, don't merge i += 1 @@ -154,7 +159,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter): next_msg.get("content"), list ): current["content"].extend(next_msg["content"]) - messages.pop(i + 1) + msgs.pop(i + 1) else: i += 1 @@ -162,7 +167,7 @@ class PerplexityLLMAdapter(OpenAILLMAdapter): # Perplexity requires the last message to be "user" or "tool". # OpenAI appears to silently ignore trailing assistant messages # server-side, so removing them preserves equivalent behavior. - while messages and messages[-1].get("role") == "assistant": - messages.pop() + while msgs and msgs[-1].get("role") == "assistant": + msgs.pop() - return messages + return cast(list[ChatCompletionMessageParam], msgs)