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

Reduce type: ignore comments by fixing avoidable type mismatches

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Mark Backman
2026-02-08 15:23:35 -05:00
parent bc730e4069
commit 28be775740
11 changed files with 55 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
changelog/3678.added.md
View File

@@ -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.

14
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -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)

28
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -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,

4
src/pipecat/processors/aggregators/openai_llm_context.py
View File

@@ -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:

6
src/pipecat/processors/frame_processor.py
View File

@@ -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:

5
src/pipecat/serializers/protobuf.py
View File

@@ -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)

5
src/pipecat/transports/base_output.py
View File

@@ -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,
)

5
src/pipecat/transports/daily/transport.py
View File

@@ -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

6
src/pipecat/transports/smallwebrtc/connection.py
View File

@@ -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

11
src/pipecat/transports/websocket/client.py
View File

@@ -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."""

10
src/pipecat/utils/asyncio/task_manager.py
View File

@@ -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: