Fix type errors in pipeline and add to pyright checked set

Use Sequence[FrameProcessor] instead of list[FrameProcessor] in Pipeline,
ServiceSwitcher, and ServiceSwitcherStrategy parameters to accept subtype
lists. Add cast() in LLMSwitcher for narrowed return types. Guard against
None in task_observer._send_to_proxy and replace hasattr with truthiness
check in task._cleanup.

pyrightconfig.json

@@ -9,7 +9,8 @@
     "src/pipecat/frames",
     "src/pipecat/observers",
     "src/pipecat/extensions",
-    "src/pipecat/turns"
+    "src/pipecat/turns",
+    "src/pipecat/pipeline"
   ],
   "exclude": [
     "**/*_pb2.py",

src/pipecat/pipeline/llm_switcher.py

@@ -6,7 +6,7 @@
 
 """LLM switcher for switching between different LLMs at runtime, with different switching strategies."""
 
-from typing import Any
+from typing import Any, cast
 
 from pipecat.adapters.schemas.direct_function import DirectFunction
 from pipecat.pipeline.service_switcher import (
@@ -47,7 +47,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
         Returns:
             List of LLM services managed by this switcher.
         """
-        return self.services
+        return cast(list[LLMService], self.services)
 
     @property
     def active_llm(self) -> LLMService:
@@ -56,7 +56,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
         Returns:
             The currently active LLM service, or None if no LLM is active.
         """
-        return self.strategy.active_service
+        return cast(LLMService, self.strategy.active_service)
 
     async def run_inference(self, context: LLMContext, **kwargs) -> str | None:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.

src/pipecat/pipeline/pipeline.py

@@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes
 for pipeline source and sink operations.
 """
 
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Sequence
 
 from pipecat.frames.frames import Frame
 from pipecat.pipeline.base_pipeline import BasePipeline
@@ -98,7 +98,7 @@ class Pipeline(BasePipeline):
 
     def __init__(
         self,
-        processors: list[FrameProcessor],
+        processors: Sequence[FrameProcessor],
         *,
         source: FrameProcessor | None = None,
         sink: FrameProcessor | None = None,
@@ -106,7 +106,7 @@ class Pipeline(BasePipeline):
         """Initialize the pipeline with a list of processors.
 
         Args:
-            processors: List of frame processors to connect in sequence.
+            processors: Sequence of frame processors to connect in sequence.
             source: An optional pipeline source processor.
             sink: An optional pipeline sink processor.
         """
@@ -116,7 +116,7 @@ class Pipeline(BasePipeline):
         # downstream outside of the pipeline.
         self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source")
         self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink")
-        self._processors: list[FrameProcessor] = [self._source] + processors + [self._sink]
+        self._processors: list[FrameProcessor] = [self._source, *processors, self._sink]
 
         self._link_processors()
 

src/pipecat/pipeline/service_switcher.py

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

src/pipecat/pipeline/task.py

@@ -742,7 +742,7 @@ class PipelineTask(BasePipelineTask):
         await self._observer.cleanup()
 
         # End conversation tracing if it's active - this will also close any active turn span
-        if self._enable_tracing and hasattr(self, "_turn_trace_observer"):
+        if self._enable_tracing and self._turn_trace_observer:
             self._turn_trace_observer.end_conversation_tracing()
 
         # Cleanup pipeline processors.

src/pipecat/pipeline/task_observer.py

@@ -173,6 +173,8 @@ class TaskObserver(BaseObserver):
         return proxies
 
     async def _send_to_proxy(self, data: Any):
+        if not self._proxies:
+            return
         for proxy in self._proxies.values():
             await proxy.queue.put(data)