From ab91047300161fabbcca96d96bb8ffee65f3ac8e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Apr 2026 21:47:11 -0400 Subject: [PATCH] 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 | 3 ++- src/pipecat/pipeline/llm_switcher.py | 6 +++--- src/pipecat/pipeline/pipeline.py | 8 ++++---- src/pipecat/pipeline/service_switcher.py | 11 ++++++----- src/pipecat/pipeline/task.py | 2 +- src/pipecat/pipeline/task_observer.py | 2 ++ 6 files changed, 18 insertions(+), 14 deletions(-) diff --git a/pyrightconfig.json b/pyrightconfig.json index 4016fd6e9..7176ab774 100644 --- a/pyrightconfig.json +++ b/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", diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py index 71a7c7974..031f9e875 100644 --- a/src/pipecat/pipeline/llm_switcher.py +++ b/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. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 325cedb82..d7999b4fe 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/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() diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py index a7f98f9b1..dd5e330e9 100644 --- a/src/pipecat/pipeline/service_switcher.py +++ b/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: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 394b3d2e2..08e08ea00 100644 --- a/src/pipecat/pipeline/task.py +++ b/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. diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index c6603c1d8..5b29850a6 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/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)