​
ModelProcessingError

class ModelProcessingError(Exception):

Raised when an error occurs during model processing.

​
ModelManager

class ModelManager:

ModelManager choosing a model from provided list. Models are picked according to defined strategy.

Parameters:

  • models (Union[BaseModelBackend, List[BaseModelBackend]]): model backend or list of model backends (e.g., model instances, APIs)
  • scheduling_strategy (str): name of function that defines how to select the next model. (default: :str:round_robin)

​
init

def __init__(
    self,
    models: Union[BaseModelBackend, List[BaseModelBackend]],
    scheduling_strategy: str = 'round_robin'
):

​
model_type

def model_type(self):

Returns:

Union[ModelType, str]: Current model type.

​
model_config_dict

def model_config_dict(self):

Returns:

Dict[str, Any]: Config dictionary of the current model.

​
model_config_dict

def model_config_dict(self, model_config_dict: Dict[str, Any]):

Set model_config_dict to the current model.

Parameters:

  • model_config_dict (Dict[str, Any]): Config dictionary to be set at current model.

​
current_model_index

def current_model_index(self):

Returns:

int: index of current model in given list of models.

​
num_models

def num_models(self):

Returns:

int: The number of models available in the model manager.

​
token_limit

def token_limit(self):

Returns:

int: The maximum token limit for the given model.

​
token_counter

def token_counter(self):

Returns:

BaseTokenCounter: The token counter following the model’s tokenization style.

​
add_strategy

def add_strategy(self, name: str, strategy_fn: Callable):

Add a scheduling strategy method provided by user in case when none of existent strategies fits. When custom strategy is provided, it will be set as “self.scheduling_strategy” attribute.

Parameters:

  • name (str): The name of the strategy.
  • strategy_fn (Callable): The scheduling strategy function.

​
round_robin

def round_robin(self):

Returns:

BaseModelBackend for processing incoming messages.

​
always_first

def always_first(self):

Returns:

BaseModelBackend for processing incoming messages.

​
random_model

def random_model(self):

Returns:

BaseModelBackend for processing incoming messages.

​
run

def run(
    self,
    messages: List[OpenAIMessage],
    response_format: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Dict[str, Any]]] = None
):

Process a list of messages by selecting a model based on the scheduling strategy. Sends the entire list of messages to the selected model, and returns a single response.

Parameters:

  • messages (List[OpenAIMessage]): Message list with the chat history in OpenAI API format.

Returns:

Union[ChatCompletion, Stream[ChatCompletionChunk]]: ChatCompletion in the non-stream mode, or Stream[ChatCompletionChunk] in the stream mode.