Skip to main content

​
Overview

ServiceSwitcher is a specialized parallel pipeline that enables dynamic switching between multiple service instances at runtime. This is useful when you need to change between different STT providers, TTS providers, or other frame processors based on user preferences, costs, performance requirements, or other runtime conditions. The switcher uses a strategy pattern to determine which service is active. Currently, a manual switching strategy is provided, allowing explicit control over which service handles frames at any given time.

​
How It Works

ServiceSwitcher wraps multiple services in a parallel pipeline where only the active service processes frames. Each service is “sandwiched” between two filters that check if it’s the currently active service before allowing frames to pass through. When you switch services, the filters update to redirect frame flow to the newly active service.

​
Constructor

from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual

switcher = ServiceSwitcher(
    services=[stt_service1, stt_service2],
    strategy_type=ServiceSwitcherStrategyManual
)
​
services
List[FrameProcessor]
required
List of service instances to switch between. Can be any frame processors (STT, TTS, or custom processors).
​
strategy_type
Type[ServiceSwitcherStrategy]
required
The strategy class to use for switching logic. Pass the class itself, not an instance.

​
Switching Strategies

​
ServiceSwitcherStrategyManual

The manual strategy allows explicit control over which service is active by pushing ManuallySwitchServiceFrame frames into the pipeline. Initial State: The first service in the list is active by default. Switching: Push a ManuallySwitchServiceFrame with the desired service instance.

​
Usage Examples

​
Switching Between TTS Services

from pipecat.frames.frames import ManuallySwitchServiceFrame
from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.cartesia.tts import CartesiaTTSService

# Create TTS services
elevenlabs = ElevenLabsTTSService(api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID"))
cartesia = CartesiaTTSService(api_key=os.getenv("CARTESIA_API_KEY"), voice_id=os.getenv("CARTESIA_VOICE_ID"))

# Create switcher with both services
tts_switcher = ServiceSwitcher(
    services=[elevenlabs, cartesia],
    strategy_type=ServiceSwitcherStrategyManual
)

# Use in pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts_switcher,
    transport.output(),
    context_aggregator.assistant()
])

# Later, switch to Cartesia
await task.queue_frame(ManuallySwitchServiceFrame(service=cartesia))
⌘I