Merge pull request #4327 from pipecat-ai/filipi/pyright_service_switcher

Fixing typecheck for service switcher.

scripts/fix-ruff-and-typecheck.sh

@@ -6,5 +6,9 @@ PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
 
 echo "Running ruff format..."
 uv run ruff format "$PROJECT_ROOT"
+
 echo "Running ruff check..."
 uv run ruff check --fix "$PROJECT_ROOT"
+
+echo "Running pyright check..."
+uv run pyright

src/pipecat/pipeline/llm_switcher.py

@@ -15,6 +15,7 @@ from pipecat.pipeline.service_switcher import (
     StrategyType,
 )
 from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.services.llm_service import LLMService
 
 
@@ -38,7 +39,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
             strategy_type: The strategy class to use for switching between LLMs.
                 Defaults to ``ServiceSwitcherStrategyManual``.
         """
-        super().__init__(llms, strategy_type)
+        super().__init__(cast(list[FrameProcessor], llms), strategy_type)
 
     @property
     def llms(self) -> list[LLMService]:

src/pipecat/pipeline/service_switcher.py

@@ -6,7 +6,6 @@
 
 """Service switcher for switching between different services at runtime, with different switching strategies."""
 
-from collections.abc import Sequence
 from typing import Any, Generic, TypeVar
 
 from loguru import logger
@@ -43,7 +42,7 @@ class ServiceSwitcherStrategy(BaseObject):
             ...
     """
 
-    def __init__(self, services: Sequence[FrameProcessor]):
+    def __init__(self, services: list[FrameProcessor]):
         """Initialize the service switcher strategy with a list of services.
 
         Note:
@@ -57,7 +56,7 @@ class ServiceSwitcherStrategy(BaseObject):
         if len(services) == 0:
             raise Exception(f"ServiceSwitcherStrategy needs at least one service")
 
-        self._services = list(services)
+        self._services = services
         self._active_service = services[0]
 
         self._register_event_handler("on_service_switched")
@@ -224,7 +223,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
     def __init__(
         self,
-        services: Sequence[FrameProcessor],
+        services: list[FrameProcessor],
         strategy_type: type[StrategyType] = ServiceSwitcherStrategyManual,
     ):
         """Initialize the service switcher with a list of services and a switching strategy.
@@ -236,7 +235,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
         """
         _strategy = strategy_type(services)
         super().__init__(*self._make_pipeline_definitions(services, _strategy))
-        self._services = list(services)
+        self._services = services
         self._strategy = _strategy
 
     @property
@@ -251,7 +250,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
     @staticmethod
     def _make_pipeline_definitions(
-        services: Sequence[FrameProcessor], strategy: ServiceSwitcherStrategy
+        services: list[FrameProcessor], strategy: ServiceSwitcherStrategy
     ) -> list[Any]:
         pipelines = []
         for service in services: