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.

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 = True, 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 True.

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

register_direct_function(handler: DirectFunction, *, cancel_on_interruption: bool = True, timeout_secs: float | None = None)[source]

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

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 True.

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