orchestrator.ModelRegistry

class orchestrator.ModelRegistry[source]

Bases: object

Central registry for all available models.

Manages model registration, selection, and performance tracking using multi-armed bandit algorithms.

__init__()[source]: Initialize model registry.

Methods

`__init__`()	Initialize model registry.
`get_available_models`()	Get list of available (healthy) models.
`get_model`(model_name[, provider])	Get a model by name.
`get_model_statistics`()	Get registry statistics.
`list_models`([provider])	List all registered models.
`list_providers`()	List all providers.
`register_model`(model)	Register a new model.
`reset_statistics`()	Reset all performance statistics.
`select_model`(requirements)	Select best model for given requirements.
`unregister_model`(model_name[, provider])	Unregister a model.
`update_model_performance`(model, success[, ...])	Update model performance metrics.

__init__()[source]: Initialize model registry.

register_model(model)[source]

Register a new model.

Parameters:: model (Model) – Model to register
Raises:: ValueError – If model with same name already exists
Return type:: None

unregister_model(model_name, provider='')[source]

Unregister a model.

Parameters:

model_name (str) – Name of model to unregister
provider (str) – Provider name (optional)

Return type:

None

get_model(model_name, provider='')[source]

Get a model by name.

Parameters:

model_name (str) – Model name
provider (str) – Provider name (optional)

Return type:

Model

Returns:

Model instance

Raises:

ModelNotFoundError – If model not found

list_models(provider=None)[source]

List all registered models.

Parameters:: provider (Optional[str]) – Filter by provider (optional)
Return type:: List[str]
Returns:: List of model names

list_providers()[source]

List all providers.

Return type:: List[str]
Returns:: List of provider names

async get_available_models()[source]

Get list of available (healthy) models.

Return type:: List[str]
Returns:: List of available model keys

async select_model(requirements)[source]

Select best model for given requirements.

Parameters:: requirements (Dict[str, Any]) – Requirements dictionary
Return type:: Model
Returns:: Selected model
Raises:: NoEligibleModelsError – If no models meet requirements

update_model_performance(model, success, latency=0.0, cost=0.0)[source]

Update model performance metrics.

Parameters:

model (Model) – Model that was used
success (bool) – Whether the operation was successful
latency (float) – Operation latency in seconds
cost (float) – Operation cost in USD

Return type:

None

get_model_statistics()[source]

Get registry statistics.

Return type:: Dict[str, Any]
Returns:: Dictionary of statistics

reset_statistics()[source]

Reset all performance statistics.

Return type:: None