llm_switcher

LLM switcher for switching between different LLMs at runtime, with different switching strategies.

class pipecat.pipeline.llm_switcher.LLMSwitcher(llms: list[~pipecat.services.llm_service.LLMService], strategy_type: type[~pipecat.pipeline.service_switcher.StrategyType] = <class 'pipecat.pipeline.service_switcher.ServiceSwitcherStrategyManual'>)[source]

Bases: ServiceSwitcher[StrategyType]

A pipeline that switches between different LLMs at runtime.

Example:

llm_switcher = LLMSwitcher(llms=[openai_llm, anthropic_llm])
__init__(llms: list[~pipecat.services.llm_service.LLMService], strategy_type: type[~pipecat.pipeline.service_switcher.StrategyType] = <class 'pipecat.pipeline.service_switcher.ServiceSwitcherStrategyManual'>)[source]

Initialize the service switcher with a list of LLMs and a switching strategy.

Parameters:

  • llms – List of LLM services to switch between.

  • strategy_type – The strategy class to use for switching between LLMs. Defaults to ServiceSwitcherStrategyManual.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process a frame, syncing context tool handlers on all member LLMs.

On an LLMContextFrame, the handlers advertised in the context are synced on every member LLM — active or not — so that tools listed via LLMContext(tools=[...]) keep working across service switches.

This is needed because member LLMs sit behind per-branch filters: only the active LLM receives the context frame and would otherwise sync its handlers, leaving inactive LLMs out of step with the advertised tools.

Parameters:

  • frame – The frame to process.

  • direction – The direction of frame flow.

property llms: list[LLMService]

Get the list of LLMs managed by this switcher.

Returns:

List of LLM services managed by this switcher.

property active_llm: LLMService

Get the currently active LLM.

Returns:

The currently active LLM service, or None if no LLM is active.

async run_inference(context: LLMContext, **kwargs) → str | None[source]

Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.

Parameters:

  • context – The LLM context containing conversation history.

  • **kwargs – Additional arguments forwarded to the active LLM’s run_inference (e.g. max_tokens, system_instruction).

Returns:

The LLM’s response as a string, or None if no response is generated.

register_function(function_name: str | None, handler: Any, *, cancel_on_interruption: bool | None = None, timeout_secs: float | None = None)[source]

Register a function handler for LLM function calls, on all LLMs, active or not.

Parameters:

  • function_name – The name of the function to handle. Use None to handle all function calls with a catch-all handler.

  • handler – The function handler. Should accept a single FunctionCallParams parameter.

  • cancel_on_interruption – Whether to cancel this function call when an interruption occurs. Defaults to None (fall back to the @tool_options decorator value on the handler, then to True).

  • timeout_secs – Optional timeout in seconds for the function call.

register_direct_function(handler: Callable[[...], Awaitable[Any]], *, cancel_on_interruption: bool | None = None, timeout_secs: float | None = None)[source]

Register a direct function handler for LLM function calls, on all LLMs, active or not.

Deprecated since version 1.4.0: Use LLMContext with tools=[...] instead. Direct functions listed in the context are registered on every member LLM automatically — at session start, or push an LLMSetToolsFrame to change tools mid-session. Will be removed in 2.0.0.

Parameters:

  • handler – The direct function to register. Must follow DirectFunction protocol.

  • cancel_on_interruption – Whether to cancel this function call when an interruption occurs. Defaults to None (fall back to the @tool_options decorator value on the handler, then to True).

  • timeout_secs – Optional timeout in seconds for the function call.