From 28be775740f3de74db94972c7c9632f2406daded Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Feb 2026 15:23:35 -0500 Subject: [PATCH] Reduce type: ignore comments by fixing avoidable type mismatches Replace ~20 type: ignore comments with proper type fixes: - Widen set_tools() to accept List[dict] | ToolsSchema | NotGiven - Widen create_task() to accept Coroutine | Awaitable - Fix _turn_params to use BaseTurnParams instead of SmartTurnParams - Make _thought_llm Optional[str] with assertion guard - Add mixer assertion, websocket narrowing, ice_servers cast - Use dict.get() in protobuf serializer - Make remote_participants Optional in Daily transport Co-Authored-By: Claude Opus 4.6 --- changelog/3678.added.md | 2 +- .../processors/aggregators/llm_response.py | 14 +++++----- .../aggregators/llm_response_universal.py | 28 ++++++++++++------- .../aggregators/openai_llm_context.py | 4 +-- src/pipecat/processors/frame_processor.py | 6 ++-- src/pipecat/serializers/protobuf.py | 5 ++-- src/pipecat/transports/base_output.py | 5 ++-- src/pipecat/transports/daily/transport.py | 5 ++-- .../transports/smallwebrtc/connection.py | 6 ++-- src/pipecat/transports/websocket/client.py | 11 +++++--- src/pipecat/utils/asyncio/task_manager.py | 10 +++---- 11 files changed, 55 insertions(+), 41 deletions(-) diff --git a/changelog/3678.added.md b/changelog/3678.added.md index b18d5aac8..1a704b547 100644 --- a/changelog/3678.added.md +++ b/changelog/3678.added.md @@ -1 +1 @@ -- Added pyright basic type checking configuration for the core framework, fixing 276 type errors across 64 files. +- Added pyright basic type checking configuration for the core framework. diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 658bd677c..a274a31ff 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -23,7 +23,7 @@ if TYPE_CHECKING: from loguru import logger from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.turn.base_turn_analyzer import BaseTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( BotStartedSpeakingFrame, @@ -405,7 +405,7 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator): """ self._context.set_messages(messages) - def set_tools(self, tools: List): + def set_tools(self, tools): """Set tools in the context. Args: @@ -470,7 +470,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): super().__init__(context=context, role="user", **kwargs) self._params = params or LLMUserAggregatorParams() self._vad_params: Optional[VADParams] = None - self._turn_params: Optional[SmartTurnParams] = None + self._turn_params: Optional[BaseTurnParams] = None if "aggregation_timeout" in kwargs: with warnings.catch_warnings(): @@ -558,12 +558,12 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): elif isinstance(frame, LLMMessagesUpdateFrame): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) # type: ignore[arg-type] + self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) elif isinstance(frame, SpeechControlParamsFrame): self._vad_params = frame.vad_params - self._turn_params = frame.turn_params # type: ignore[assignment] + self._turn_params = frame.turn_params await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -917,7 +917,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): elif isinstance(frame, LLMMessagesUpdateFrame): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) # type: ignore[arg-type] + self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) elif isinstance(frame, FunctionCallsStartedFrame): @@ -1023,7 +1023,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): # sure we don't block the pipeline. if properties and properties.on_context_updated: task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated" - task = self.create_task(properties.on_context_updated(), task_name) # type: ignore[arg-type] + task = self.create_task(properties.on_context_updated(), task_name) self._context_updated_tasks.add(task) task.add_done_callback(self._context_updated_task_finished) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 570988991..3305efb50 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -20,7 +20,7 @@ from typing import Any, Dict, List, Literal, Optional, Set, Type, cast from loguru import logger -from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.audio.vad.vad_controller import VADController from pipecat.frames.frames import ( @@ -258,12 +258,20 @@ class LLMContextAggregator(FrameProcessor): """ self._context.set_messages(messages) - def set_tools(self, tools: ToolsSchema | NotGiven): + def set_tools(self, tools: ToolsSchema | List | NotGiven): """Set tools in the context. Args: - tools: List of tool definitions to set in the context. + tools: Tool definitions to set in the context. """ + if isinstance(tools, list): + tools = ToolsSchema( + standard_tools=[], + custom_tools=cast( + Dict[AdapterType, List[Dict[str, Any]]], + {AdapterType.SHIM: tools}, + ), + ) self._context.set_tools(tools) def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict): @@ -461,7 +469,7 @@ class LLMUserAggregator(LLMContextAggregator): elif isinstance(frame, LLMMessagesUpdateFrame): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) # type: ignore[arg-type] + self.set_tools(frame.tools) # Push the LLMSetToolsFrame as well, since speech-to-speech LLM # services (like OpenAI Realtime) may need to know about tool # changes; unlike text-based LLM services they won't just "pick up @@ -819,7 +827,7 @@ class LLMAssistantAggregator(LLMContextAggregator): self._assistant_turn_start_timestamp = "" self._thought_append_to_context = False - self._thought_llm: str = "" + self._thought_llm: Optional[str] = "" self._thought_aggregation: List[TextPartForConcatenation] = [] self._thought_start_time: str = "" @@ -881,7 +889,7 @@ class LLMAssistantAggregator(LLMContextAggregator): elif isinstance(frame, LLMMessagesUpdateFrame): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): - self.set_tools(frame.tools) # type: ignore[arg-type] + self.set_tools(frame.tools) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) elif isinstance(frame, FunctionCallsStartedFrame): @@ -1037,7 +1045,7 @@ class LLMAssistantAggregator(LLMContextAggregator): # sure we don't block the pipeline. if properties and properties.on_context_updated: task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated" - task = self.create_task(properties.on_context_updated(), task_name) # type: ignore[arg-type] + task = self.create_task(properties.on_context_updated(), task_name) self._context_updated_tasks.add(task) task.add_done_callback(self._context_updated_task_finished) @@ -1119,7 +1127,7 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._reset_thought_aggregation() self._thought_append_to_context = frame.append_to_context - self._thought_llm = frame.llm # type: ignore[assignment] + self._thought_llm = frame.llm self._thought_start_time = time_now_iso8601() async def _handle_thought_text(self, frame: LLMThoughtTextFrame): @@ -1143,10 +1151,10 @@ class LLMAssistantAggregator(LLMContextAggregator): thought = concatenate_aggregated_text(self._thought_aggregation) if self._thought_append_to_context: - llm = self._thought_llm + assert self._thought_llm is not None, "llm is required when append_to_context is True" self._context.add_message( LLMSpecificMessage( - llm=llm, + llm=self._thought_llm, message={ "type": "thought", "text": thought, diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index dc548c56a..2e443a6bc 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -157,7 +157,7 @@ class OpenAILLMContext: return self._messages @property - def tools(self) -> List[ChatCompletionToolParam] | NotGiven | List[Any]: + def tools(self) -> List[ChatCompletionToolParam] | NotGiven | ToolsSchema | List[Any]: """Get the tools list, converting through adapter if available. Returns: @@ -165,7 +165,7 @@ class OpenAILLMContext: """ if self._llm_adapter: return self._llm_adapter.from_standard_tools(self._tools) - return self._tools # type: ignore[return-value] + return self._tools @property def tool_choice(self) -> ChatCompletionToolChoiceOptionParam | NotGiven: diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 69a1b07eb..4bd331192 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -470,11 +470,13 @@ class FrameProcessor(BaseObject): await self.stop_ttfb_metrics() await self.stop_processing_metrics() - def create_task(self, coroutine: Coroutine, name: Optional[str] = None) -> asyncio.Task: + def create_task( + self, coroutine: Coroutine | Awaitable, name: Optional[str] = None + ) -> asyncio.Task: """Create a new task managed by this processor. Args: - coroutine: The coroutine to run in the task. + coroutine: The coroutine or awaitable to run in the task. name: Optional name for the task. Returns: diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 467b2a55f..01a746742 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -88,12 +88,11 @@ class ProtobufFrameSerializer(FrameSerializer): ) proto_frame = frame_protos.Frame() # type: ignore[attr-defined] - if type(serializable_frame) not in self.SERIALIZABLE_TYPES: + proto_optional_name = self.SERIALIZABLE_TYPES.get(type(serializable_frame)) + if proto_optional_name is None: logger.warning(f"Frame type {type(frame)} is not serializable") return None - # ignoring linter errors; we check that type(serializable_frame) is in this dict above - proto_optional_name = self.SERIALIZABLE_TYPES[type(serializable_frame)] # type: ignore[index] proto_attr = getattr(proto_frame, proto_optional_name) for field in dataclasses.fields(serializable_frame): # type: ignore[arg-type] value = getattr(serializable_frame, field.name) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 4a2a4a16a..7b2448d7b 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -702,13 +702,14 @@ class BaseOutputTransport(FrameProcessor): await self._bot_stopped_speaking() async def with_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]: + assert self._mixer is not None last_frame_time = 0 silence = b"\x00" * self._audio_chunk_size while True: try: frame = self._audio_queue.get_nowait() if isinstance(frame, OutputAudioRawFrame): - frame.audio = await self._mixer.mix(frame.audio) # type: ignore[union-attr] + frame.audio = await self._mixer.mix(frame.audio) last_frame_time = time.time() yield frame self._audio_queue.task_done() @@ -719,7 +720,7 @@ class BaseOutputTransport(FrameProcessor): await self._bot_stopped_speaking() # Generate an audio frame with only the mixer's part. frame = OutputAudioRawFrame( - audio=await self._mixer.mix(silence), # type: ignore[union-attr] + audio=await self._mixer.mix(silence), sample_rate=self._sample_rate, num_channels=self._params.audio_out_channels, ) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index ca13c0497..79b3b39df 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -195,7 +195,7 @@ class DailyUpdateRemoteParticipantsFrame(ControlFrame): remote_participants: See https://reference-python.daily.co/api_reference.html#daily.CallClient.update_remote_participants. """ - remote_participants: Mapping[str, Any] = None # type: ignore[assignment] + remote_participants: Optional[Mapping[str, Any]] = None def __post_init__(self): super().__post_init__() @@ -1959,7 +1959,8 @@ class DailyOutputTransport(BaseOutputTransport): await super().process_frame(frame, direction) if isinstance(frame, DailyUpdateRemoteParticipantsFrame): - await self._client.update_remote_participants(frame.remote_participants) + if frame.remote_participants is not None: + await self._client.update_remote_participants(frame.remote_participants) async def send_message( self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame diff --git a/src/pipecat/transports/smallwebrtc/connection.py b/src/pipecat/transports/smallwebrtc/connection.py index 8a4217995..6702b7edf 100644 --- a/src/pipecat/transports/smallwebrtc/connection.py +++ b/src/pipecat/transports/smallwebrtc/connection.py @@ -15,7 +15,7 @@ import asyncio import json import time import uuid -from typing import Any, List, Literal, Optional, Union +from typing import Any, List, Literal, Optional, Union, cast from loguru import logger from pydantic import BaseModel, TypeAdapter @@ -224,9 +224,9 @@ class SmallWebRTCConnection(BaseObject): if not ice_servers: self.ice_servers: List[IceServer] = [] elif all(isinstance(s, IceServer) for s in ice_servers): - self.ice_servers = ice_servers # type: ignore[assignment] + self.ice_servers = cast(List[IceServer], ice_servers) elif all(isinstance(s, str) for s in ice_servers): - self.ice_servers = [IceServer(urls=s) for s in ice_servers] # type: ignore[misc] + self.ice_servers = [IceServer(urls=cast(str, s)) for s in ice_servers] else: raise TypeError("ice_servers must be either List[str] or List[RTCIceServer]") self._connect_invoked = False diff --git a/src/pipecat/transports/websocket/client.py b/src/pipecat/transports/websocket/client.py index b368a2858..f8dc51d75 100644 --- a/src/pipecat/transports/websocket/client.py +++ b/src/pipecat/transports/websocket/client.py @@ -141,7 +141,8 @@ class WebsocketClientSession: self._client_task_handler(), f"{self._transport_name}::WebsocketClientSession::_client_task_handler", ) - await self._callbacks.on_connected(self._websocket) # type: ignore[arg-type] + assert self._websocket is not None + await self._callbacks.on_connected(self._websocket) except TimeoutError: logger.error(f"Timeout connecting to {self._uri}") @@ -194,13 +195,15 @@ class WebsocketClientSession: """Handle incoming messages from the WebSocket connection.""" try: assert self._websocket is not None + websocket = self._websocket # Handle incoming messages - async for message in self._websocket: - await self._callbacks.on_message(self._websocket, message) # type: ignore[arg-type] + async for message in websocket: + await self._callbacks.on_message(websocket, message) except Exception as e: logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})") - await self._callbacks.on_disconnected(self._websocket) # type: ignore[arg-type] + if self._websocket: + await self._callbacks.on_disconnected(self._websocket) def __str__(self): """String representation of the WebSocket client session.""" diff --git a/src/pipecat/utils/asyncio/task_manager.py b/src/pipecat/utils/asyncio/task_manager.py index c680e19b1..a1da1ddcc 100644 --- a/src/pipecat/utils/asyncio/task_manager.py +++ b/src/pipecat/utils/asyncio/task_manager.py @@ -15,7 +15,7 @@ import asyncio import traceback from abc import ABC, abstractmethod from dataclasses import dataclass -from typing import Coroutine, Dict, Optional, Sequence +from typing import Awaitable, Coroutine, Dict, Optional, Sequence from loguru import logger @@ -56,13 +56,13 @@ class BaseTaskManager(ABC): pass @abstractmethod - def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task: + def create_task(self, coroutine: Coroutine | Awaitable, name: str) -> asyncio.Task: """Creates and schedules a new asyncio Task that runs the given coroutine. The task is added to a global set of created tasks. Args: - coroutine: The coroutine to be executed within the task. + coroutine: The coroutine or awaitable to be executed within the task. name: The name to assign to the task for identification. Returns: @@ -139,13 +139,13 @@ class TaskManager(BaseTaskManager): raise Exception("TaskManager is not setup: unable to get event loop") return self._params.loop - def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task: + def create_task(self, coroutine: Coroutine | Awaitable, name: str) -> asyncio.Task: """Creates and schedules a new asyncio Task that runs the given coroutine. The task is added to a global set of created tasks. Args: - coroutine: The coroutine to be executed within the task. + coroutine: The coroutine or awaitable to be executed within the task. name: The name to assign to the task for identification. Returns: