Skip to main content

​
Overview

LLMSwitcher is a specialized version of ServiceSwitcher designed specifically for managing multiple LLM services. Beyond basic service switching, it provides convenient methods for running ad-hoc inferences and registering function handlers across all LLMs simultaneously. This is particularly useful when you need to switch between different LLM providers based on task complexity, cost optimization, or specific model capabilities while maintaining a consistent function calling interface.

​
Constructor

from pipecat.pipeline.llm_switcher import LLMSwitcher

# Uses ServiceSwitcherStrategyManual by default
switcher = LLMSwitcher(llms=[openai_service, gemini_service])
​
llms
List[LLMService]
required
List of LLM service instances to switch between.
​
strategy_type
Type[ServiceSwitcherStrategy]
default:"ServiceSwitcherStrategyManual"
The strategy class to use for switching logic. Pass the class itself, not an instance. Defaults to ServiceSwitcherStrategyManual.

​
Properties

​
llms
List[LLMService]
The list of LLM services managed by this switcher.
​
active_llm
LLMService | None
The currently active LLM service, or None if no LLMs are configured.

​
Methods

​
run_inference()

Run a one-shot inference with the currently active LLM, outside of the normal pipeline flow. 
result = await llm_switcher.run_inference(
    context=llm_context,
    max_tokens=1000,
    system_instruction="You are a helpful assistant."
)
​
context
LLMContext
required
The LLM context containing conversation history and messages.
​
**kwargs
Any
Additional arguments forwarded to the active LLM’s run_inference method. Common options include max_tokens (maximum tokens to generate) and system_instruction (override the system prompt for this inference).
Returns: str | None - The LLM’s response as a string, or None if no LLM is active.

​
register_function()

You don’t need register_function when your tool’s handler is bundled on its FunctionSchema: listing it in LLMContext(tools=[...]) or adding it via LLMSetToolsFrame is enough.
Register a function handler with all LLMs in the switcher, regardless of which is currently active. 
llm_switcher.register_function(
    function_name="get_weather",
    handler=handle_weather,
    cancel_on_interruption=True
)
​
function_name
str | None
required
The name of the function to handle. Use None for a catch-all handler that processes all function calls.
​
handler
Callable
required
The async function handler. Should accept a single FunctionCallParams parameter.
​
cancel_on_interruption
bool | None
default:"None"
Whether to cancel this function call when a user interruption occurs. When False, the call is treated as asynchronous: the LLM continues the conversation immediately without waiting for the result, and the result is injected later via a developer message. Defaults to None, which falls back to the handler’s @tool_options value, then to True.
​
timeout_secs
float | None
default:"None"
Optional per-tool timeout in seconds. Overrides the global function_call_timeout_secs for this specific function.

​
register_direct_function()

Deprecated since 1.4.0. Direct functions now register automatically on every member LLM (active or not) when you list them in LLMContext(tools=[...]), so explicit registration is no longer needed — the tools keep working across service switches. Push an LLMSetToolsFrame to change tools mid-session. This method will be removed in a future version.
Register a direct function handler with all LLMs in the switcher, regardless of which is currently active. 
llm_switcher.register_direct_function(
    handler=my_direct_function,
    cancel_on_interruption=True,
    timeout_secs=30.0
)
​
handler
DirectFunction
required
The direct function to register. Must follow the DirectFunction protocol.
​
cancel_on_interruption
bool | None
default:"None"
Whether to cancel this function call when a user interruption occurs. When False, the call is treated as asynchronous: the LLM continues the conversation immediately without waiting for the result, and the result is injected later via a developer message. Defaults to None, which falls back to the handler’s @tool_options value, then to True.
​
timeout_secs
float | None
default:"None"
Optional per-tool timeout in seconds. Overrides the global function_call_timeout_secs for this specific function.

​
Usage Examples

​
Basic LLM Switching

from pipecat.pipeline.llm_switcher import LLMSwitcher
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.google.llm import GoogleLLMService
from pipecat.frames.frames import ManuallySwitchServiceFrame

# Create LLM services
openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
gemini = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))

# Create switcher (uses ServiceSwitcherStrategyManual by default)
llm_switcher = LLMSwitcher(llms=[openai, gemini])

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

# Switch to cheaper model for simple tasks
await worker.queue_frame(ManuallySwitchServiceFrame(service=gpt35))