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

Add ServiceSwitcherStrategyFailover for automatic failover on service errors (#3870)

Browse Source 
* Add ServiceSwitcherStrategyFailover for automatic error-based service switching

Introduce a strategy hierarchy: ServiceSwitcherStrategy (base) →
ServiceSwitcherStrategyManual (handles ManuallySwitchServiceFrame) →
ServiceSwitcherStrategyFailover (adds error-based failover). ServiceSwitcher
now defaults to ServiceSwitcherStrategyManual with strategy_type optional.
Non-fatal ErrorFrames are forwarded to the strategy via handle_error().

* Move metadata request into _set_active_if_available

Requesting metadata is part of making a service active, so it belongs
alongside setting _active_service and firing on_service_switched. This
removes the duplicate queue_frame calls from ServiceSwitcher push_frame
and process_frame.
This commit is contained in:
kollaikal-rupesh
2026-03-10 12:37:30 -07:00
committed by GitHub
parent 43a9e9a1b5
commit 80bd935c19
6 changed files with 264 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/pipecat/pipeline/llm_switcher.py
View File

@@ -9,7 +9,11 @@
from typing import Any, List, Optional, Type
from pipecat.adapters.schemas.direct_function import DirectFunction
from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType
from pipecat.pipeline.service_switcher import (
ServiceSwitcher,
ServiceSwitcherStrategyManual,
StrategyType,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.services.llm_service import LLMService
@@ -19,18 +23,20 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
Example::
llm_switcher = LLMSwitcher(
llms=[openai_llm, anthropic_llm],
strategy_type=ServiceSwitcherStrategyManual
)
llm_switcher = LLMSwitcher(llms=[openai_llm, anthropic_llm])
"""
def __init__(self, llms: List[LLMService], strategy_type: Type[StrategyType]):
def __init__(
self,
llms: List[LLMService],
strategy_type: Type[StrategyType] = ServiceSwitcherStrategyManual,
):
"""Initialize the service switcher with a list of LLMs and a switching strategy.
Args:
llms: List of LLM services to switch between.
strategy_type: The strategy class to use for switching between LLMs.
Defaults to ``ServiceSwitcherStrategyManual``.
"""
super().__init__(llms, strategy_type)

121
src/pipecat/pipeline/service_switcher.py
View File

@@ -6,10 +6,12 @@
"""Service switcher for switching between different services at runtime, with different switching strategies."""
from abc import abstractmethod
from typing import Any, Generic, List, Optional, Type, TypeVar
from loguru import logger
from pipecat.frames.frames import (
ErrorFrame,
Frame,
ManuallySwitchServiceFrame,
ServiceMetadataFrame,
@@ -69,13 +71,13 @@ class ServiceSwitcherStrategy(BaseObject):
"""Return the currently active service."""
return self._active_service
@abstractmethod
async def handle_frame(
self, frame: ServiceSwitcherFrame, direction: FrameDirection
) -> Optional[FrameProcessor]:
"""Handle a frame that controls service switching.
Subclasses implement this to decide whether a switch should occur.
The base implementation returns ``None`` for all frames. Subclasses
override this to implement specific switching behaviors.
Args:
frame: The frame to handle.
@@ -84,7 +86,41 @@ class ServiceSwitcherStrategy(BaseObject):
Returns:
The newly active service if a switch occurred, or None otherwise.
"""
pass
return None
async def handle_error(self, error: ErrorFrame) -> Optional[FrameProcessor]:
"""Handle an error from the active service.
Called by ``ServiceSwitcher`` when a non-fatal ``ErrorFrame`` is pushed
upstream by the currently active service. Subclasses can override this
to implement automatic failover.
Args:
error: The error frame pushed by the active service.
Returns:
The newly active service if a switch occurred, or None otherwise.
"""
return None
async def _set_active_if_available(self, service: FrameProcessor) -> Optional[FrameProcessor]:
"""Set the active service to the given one, if it is in the list of available services.
If it's not in the list, the request is ignored, as it may have been
intended for another ServiceSwitcher in the pipeline.
Args:
service: The service to set as active.
Returns:
The newly active service, or None if the service was not found.
"""
if service in self.services:
self._active_service = service
await service.queue_frame(ServiceSwitcherRequestMetadataFrame(service=service))
await self._call_event_handler("on_service_switched", service)
return service
return None
class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
@@ -118,23 +154,54 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
return None
async def _set_active_if_available(self, service: FrameProcessor) -> Optional[FrameProcessor]:
"""Set the active service to the given one, if it is in the list of available services.
If it's not in the list, the request is ignored, as it may have been
intended for another ServiceSwitcher in the pipeline.
class ServiceSwitcherStrategyFailover(ServiceSwitcherStrategyManual):
"""A strategy that automatically switches to a backup service on failure.
When the active service produces a non-fatal error, this strategy switches
to the next available service in the list. Recovery and fallback policies
are left to application code via the ``on_service_switched`` event.
Event handlers available:
- on_service_switched: Called when the active service changes.
Example::
switcher = ServiceSwitcher(
services=[primary_stt, backup_stt],
strategy_type=ServiceSwitcherStrategyFailover,
)
@switcher.strategy.event_handler("on_service_switched")
async def on_switched(strategy, service):
# App decides when/how to recover the failed service
...
"""
async def handle_error(self, error: ErrorFrame) -> Optional[FrameProcessor]:
"""Handle an error from the active service by failing over.
Switches to the next service in the list. The failed service remains
in the list and can be switched back to manually or via application
logic in the ``on_service_switched`` event handler.
Args:
service: The service to set as active.
error: The error frame pushed by the active service.
Returns:
The newly active service, or None if the service was not found.
The newly active service if a switch occurred, or None if no
other service is available.
"""
if service in self.services:
self._active_service = service
await self._call_event_handler("on_service_switched", service)
return service
return None
logger.warning(f"Service {self._active_service.name} reported an error: {error.error}")
if len(self._services) <= 1:
logger.error("No other service available to switch to")
return None
current_idx = self._services.index(self._active_service)
next_idx = (current_idx + 1) % len(self._services)
return await self._set_active_if_available(self._services[next_idx])
StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy)
@@ -150,18 +217,20 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
Example::
switcher = ServiceSwitcher(
services=[stt_1, stt_2],
strategy_type=ServiceSwitcherStrategyManual,
)
switcher = ServiceSwitcher(services=[stt_1, stt_2])
"""
def __init__(self, services: List[FrameProcessor], strategy_type: Type[StrategyType]):
def __init__(
self,
services: List[FrameProcessor],
strategy_type: Type[StrategyType] = ServiceSwitcherStrategyManual,
):
"""Initialize the service switcher with a list of services and a switching strategy.
Args:
services: List of frame processors to switch between.
strategy_type: The strategy class to use for switching between services.
Defaults to ``ServiceSwitcherStrategyManual``.
"""
_strategy = strategy_type(services)
super().__init__(*self._make_pipeline_definitions(services, _strategy))
@@ -227,6 +296,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
all the filters let it pass, and `StartFrame` causes the service to
generate `ServiceMetadataFrame`.
Non-fatal ``ErrorFrame`` instances are forwarded to the strategy via
``handle_error`` so strategies like ``ServiceSwitcherStrategyFailover``
can perform failover. The error frame is still propagated upstream so
that application-level error handlers can observe it.
"""
# Consume ServiceSwitcherRequestMetadataFrame once the targeted service
# has handled it (i.e. the active service).
@@ -239,6 +312,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
if frame.service_name != self.strategy.active_service.name:
return
# Let the strategy react to non-fatal errors from the active service.
if isinstance(frame, ErrorFrame) and not frame.fatal:
await self.strategy.handle_error(frame)
await super().push_frame(frame, direction)
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -255,9 +332,5 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
# frame. If we switched, we just swallow the frame.
if not service:
await super().process_frame(frame, direction)
# If we switched to a new service, request its metadata.
if service:
await service.queue_frame(ServiceSwitcherRequestMetadataFrame(service=service))
else:
await super().process_frame(frame, direction)