Repository: caikit/caikit
Pull request #307 (2023-07-07 19:13:41)

Update example and documentation
Along with a couple of doc edits for readability, this PR addresses #235 to ensure that the text-sentiment example can be followed successfully without specifying a directory to run the `start_runtime.py` script from by overwriting the default value for `local_models_dir` to be a deduced absolute path of the script.


Patch:
diff --git a/README.md b/README.md
index 5b1acfa5..d303f093 100644
--- a/README.md
+++ b/README.md
@@ -37,11 +37,11 @@ There are 2 key things to define upfront when using Caikit to manage your AI mod
 
 The `module` defines the entry points for Caikit to manage your model. In other words, it tells Caikit how to load, infer and train your model. An example is the [text sentiment module](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/text_sentiment/runtime_model/hf_module.py). The `data model` defines the input and outputs of the model task. An example is the [text sentiment data model](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/text_sentiment/data_model/classification.py).
 
-The model is served by a [gRPC](https://grpc.io) server which can run as is or in any container runtime, including [Knative](https://knative.dev/docs/) and [KServe](https://www.kubeflow.org/docs/external-add-ons/kserve/kserve/). Here is an example of the [text sentiment server code for gRPC](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/start_runtime.py). This references the module configuration [here](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/models/text_sentiment/config.yml). This configuration specifies the module(s) (which wraps the model(s)) to serve.
+The model is served by a [gRPC](https://grpc.io) server which can run as is or in any container runtime, including [Knative](https://knative.dev/docs/) and [KServe](https://www.kubeflow.org/docs/external-add-ons/kserve/kserve/). Here is an example of the [text sentiment server code for gRPC](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/start_runtime.py). This references the module configuration [here](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/models/text_sentiment/config.yml). This configuration specifies the module(s), which wrap the model(s), to serve.
 
 There is an example of a client [here](https://github.com/caikit/caikit/blob/main/examples/text-sentiment/client.py) which is a simple Python CLI which calls the model and queries it for sentiment analysis on 2 different pieces of text. The client also references the module configuration.
 
-Check out the full [Text Sentiment example](examples/text-sentiment/) to understand how to load and infer a model using Caikit. If you want to get started with developing and integrating your AI model algorithm using Caikit, checkout the [GitHub template](https://github.com/caikit/caikit-template). In the template repository when you click on the green `Use this template` button, it generates a repository in your GitHub account with a simple customized module which is wrapped to be served by the Caikit runtime. This template is designed to be extended for module implementations.
+Check out the full [Text Sentiment example](examples/text-sentiment/) to understand how to load and infer a model using Caikit.
 
 ## User Profiles
 
@@ -64,6 +64,6 @@ Get going with [Getting Started](#getting-started) or jump into more details wit
 
 Check out our [contributing](CONTRIBUTING.md) guide to learn how to contribute to Caikit.
 
-## Code of conduct
+## Code of Conduct
 
 Participation in the Caikit community is governed by the [Code of Conduct](code-of-conduct.md).
diff --git a/caikit/config/config.yml b/caikit/config/config.yml
index f6dc6a5b..dd646f4f 100644
--- a/caikit/config/config.yml
+++ b/caikit/config/config.yml
@@ -22,28 +22,31 @@ merge_strategy: "merge"
 enable_error_checks: true
 # Maximum number of times that a single exception is logged
 max_exception_log_messages: 4
-
-# Configuration for managing the lifecycle of models
-model_management:
-    # How models are trained through the ModelManager. Each section is a key/val
-    # mapping from a user-defined label to an object with a "type" and
-    # (optional) "config" key. The "default" key is treated as a special entry
-    # that is used if no version is given explicitly at train() or load() time.
-    trainers:
-        default:
-            type: LOCAL
-    finders:
-        default:
-            type: LOCAL
-            config:
-                load_path: null
-    initializers:
-        default:
-            type: LOCAL
-            config:
-                # List of module backend configurations in priority order
-                backend_priority:
-                    - type: LOCAL
+load_path: null
+
+# Configuration for training/loading via module backends
+module_backends:
+    # Configuration for distributed loading and training. Each has an ordered
+    # list of backends in priority order. Each entry is a mapping with a "type"
+    # field and optionally a "config" field that maps to a blob of config for
+    # the backend instance. Entries may also have a "name" field that must be
+    # unique among entries, allowing for multiple instances of the same type.
+    # For example:
+    #
+    # - type: FOO
+    #   config:
+    #       foo: 123
+    # - type: BAR
+    #   config:
+    #       bar: 456
+    # - type: FOO
+    #   name: BackupFoo
+    #   config:
+    #       bar: 42
+    load_priority:
+        - type: LOCAL
+    train_priority:
+        - type: LOCAL
 
 log:
     # Default level for all python loggers
diff --git a/caikit/core/__init__.py b/caikit/core/__init__.py
index d6a50e9d..2b52fdcf 100644
--- a/caikit/core/__init__.py
+++ b/caikit/core/__init__.py
@@ -33,6 +33,7 @@ _warnings.filterwarnings("ignore")
 from .data_model import DataObjectBase, dataobject
 from .model_manager import *
 from .module_backends import *
+from .module_backends.module_backend_config import configure as backend_configure
 from .modules import ModuleBase, ModuleConfig, ModuleLoader, ModuleSaver, module
 from .task import TaskBase, task
 from .toolkit import *
diff --git a/caikit/core/model_management/__init__.py b/caikit/core/model_management/__init__.py
deleted file mode 100644
index abd77e75..00000000
--- a/caikit/core/model_management/__init__.py
+++ /dev/null
@@ -1,30 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-The model management abstractions manage the key lifecycle phases of a concrete
-model instance and are used by the ModelManager to handle each phase of the
-model lifecycle.
-
-For architectural details, see docs/adrs/020-model-management-abstractions.md
-"""
-
-# Local
-from .factories import (
-    model_finder_factory,
-    model_initializer_factory,
-    model_trainer_factory,
-)
-from .model_finder_base import ModelFinderBase
-from .model_initializer_base import ModelInitializerBase
-from .model_trainer_base import ModelTrainerBase
diff --git a/caikit/core/model_management/factories.py b/caikit/core/model_management/factories.py
deleted file mode 100644
index 866239ac..00000000
--- a/caikit/core/model_management/factories.py
+++ /dev/null
@@ -1,36 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-Global factories for model management
-"""
-
-# Local
-from ..toolkit.factory import Factory
-from .local_model_finder import LocalModelFinder
-from .local_model_initializer import LocalModelInitializer
-
-# Model trainer factory. A trainer is responsible for performing the train
-# operation against a configured framework connection.
-model_trainer_factory = Factory("ModelTrainer")
-
-# Model finder factory. A finder is responsible for locating a well defined
-# configuration for a model based on a unique path or id.
-model_finder_factory = Factory("ModelFinder")
-model_finder_factory.register(LocalModelFinder)
-
-# Model initializer factory. An initializer is responsible for taking a model
-# configuration and preparing the model to be run in a configured runtime
-# location.
-model_initializer_factory = Factory("ModelInitializer")
-model_initializer_factory.register(LocalModelInitializer)
diff --git a/caikit/core/model_management/local_model_finder.py b/caikit/core/model_management/local_model_finder.py
deleted file mode 100644
index c1f08795..00000000
--- a/caikit/core/model_management/local_model_finder.py
+++ /dev/null
@@ -1,65 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-The LocalModelFinder locates models locally on disk that contain a caikit-native
-config.yml file.
-
-Configuration for LocalModelFinder lives under the config as follows:
-
-model_management:
-    finders:
-        <finder name>:
-            type: LOCAL
-            config:
-                # Path to a local directory that holds models for relative paths
-                load_path: <null or str>
-"""
-# Standard
-from typing import Optional
-import os
-
-# First Party
-import aconfig
-import alog
-
-# Local
-from ..modules import ModuleConfig
-from ..toolkit.errors import error_handler
-from .model_finder_base import ModelFinderBase
-
-log = alog.use_channel("LFIND")
-error = error_handler.get(log)
-
-
-class LocalModelFinder(ModelFinderBase):
-    __doc__ = __doc__
-
-    name = "LOCAL"
-
-    def __init__(self, config: aconfig.Config):
-        """Initialize with an optional path prefix"""
-        self._load_path = config.load_path
-
-    def find_model(
-        self,
-        model_path: str,
-        **__,
-    ) -> Optional[ModuleConfig]:
-        """Find a model at the local path or with the configured prefix"""
-        full_model_path = model_path
-        if not os.path.exists(model_path) and self._load_path:
-            full_model_path = os.path.join(self._load_path, model_path)
-            log.debug2("Looking for %s in %s", model_path, full_model_path)
-        full_model_path = os.path.normpath(full_model_path)
-        return ModuleConfig.load(full_model_path)
diff --git a/caikit/core/model_management/local_model_initializer.py b/caikit/core/model_management/local_model_initializer.py
deleted file mode 100644
index 630a0b65..00000000
--- a/caikit/core/model_management/local_model_initializer.py
+++ /dev/null
@@ -1,257 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-The LocalModelInitializer loads a model locally, optionally with a non-local backend
-if the given module provides multiple backend-specific implementations.
-
-Configuration for LocalModelInitializer lives under the config as follows:
-
-model_management:
-    initializers:
-        <initializer name>:
-            type: LOCAL
-            config:
-                # List of module backend configurations in priority order
-                backend_priority:
-                    - type: LOCAL
-                      config: {}
-"""
-# Standard
-from typing import Callable, Optional
-import inspect
-
-# First Party
-import aconfig
-import alog
-
-# Local
-from ..module_backends import BackendBase, backend_types
-from ..modules import ModuleBase, ModuleConfig
-from ..modules.decorator import SUPPORTED_LOAD_BACKENDS_VAR_NAME
-from ..registries import (
-    module_backend_classes,
-    module_backend_registry,
-    module_backend_types,
-)
-from ..toolkit.errors import error_handler
-from .model_initializer_base import ModelInitializerBase
-
-log = alog.use_channel("LLOAD")
-error = error_handler.get(log)
-
-
-class LocalModelInitializer(ModelInitializerBase):
-    __doc__ = __doc__
-
-    name = "LOCAL"
-
-    def __init__(self, config: aconfig.Config):
-        """Construct with the config"""
-        self._backends = []
-        backend_priority = config.get(
-            "backend_priority",
-            [aconfig.Config({"type": "LOCAL"}, override_env_vars=False)],
-        )
-        for i, backend_config in enumerate(backend_priority):
-            error.value_check(
-                "<COR48633635E>",
-                "type" in backend_config,
-                "All backend priority configs must have a 'type' field",
-            )
-            backend_type = backend_config.type
-            error.value_check(
-                "<COR72281596E>",
-                backend_type in module_backend_types(),
-                "Invalid backend [{}] found at backend_priority index [{}]",
-                backend_type,
-                i,
-            )
-
-            log.debug("Configuring backend (%d)[%s]", i, backend_type)
-            backend_instance_config = backend_config.get("config", {})
-            log.debug3(
-                "Backend (%d)[%s] config: %s", i, backend_type, backend_instance_config
-            )
-
-            backend_class = module_backend_classes().get(backend_type)
-            error.value_check(
-                "<COR64618509E>",
-                len(self._backends) == i,
-                "({})[{}] already configured",
-                i,
-                backend_type,
-            )
-            error.value_check(
-                "<COR39517372E>",
-                backend_class is not None,
-                "Unsupported backend type {}",
-                backend_type,
-            )
-            if not isinstance(backend_class, type) and issubclass(
-                backend_class, BackendBase
-            ):
-                error(
-                    "<COR05184600E>",
-                    TypeError(
-                        f"Backend {backend_class} is not derived from BackendBase"
-                    ),
-                )
-
-            log.debug2("Performing config for (%d)[%s]", i, backend_type)
-            backend_instance = backend_class(backend_instance_config)
-
-            # Add configuration to backends as per individual module requirements
-            self._configure_backend_overrides(backend_type, backend_instance)
-
-            # Add the instance to the registry
-            self._backends.append(backend_instance)
-
-        log.debug2("All configured backends: %s", self._backends)
-
-    def init(self, model_config: ModuleConfig, **kwargs) -> Optional[ModuleBase]:
-        """Given a ModelConfig, attempt to initialize it locally, possibly using
-        a non-local backend
-
-        Args:
-            model_config (ModuleConfig): The in-memory model config object for
-                the model to be initialized
-
-        Returns:
-            model (Optional[ModuleBase]): The in-memory ModuleBase instance that
-                is ready to run
-        """
-        module_id = model_config.module_id
-        model_path = model_config.model_path
-        module_implementations = module_backend_registry().get(module_id, {})
-        log.debug2(
-            "Number of available backend implementations for %s found: %d",
-            module_id,
-            len(module_implementations),
-        )
-        # Look up the backend that this model was created with
-        model_creation_backend = model_config.get("model_backend", backend_types.LOCAL)
-        log.debug2("Model creation backend: %s", model_creation_backend)
-
-        # Iterate through each backend in priority order and see if this module
-        # can load with it
-        loaded_model = None
-        for load_backend in self._backends:
-
-            # Look in the module's implementations for this backend type
-            backend_impl_obj = module_implementations.get(load_backend.backend_type)
-            if backend_impl_obj is None:
-                log.debug3(
-                    "Module %s does not support loading with %s",
-                    module_id,
-                    load_backend.backend_type,
-                )
-                continue
-
-            # Grab the concrete module class for this backend and check to
-            # see if this model's artifacts were created with a version of
-            # the module that can be loaded with this backend.
-            module_backend_impl = backend_impl_obj.impl_class
-            supported_load_backends = self._get_supported_load_backends(
-                module_backend_impl
-            )
-            if model_creation_backend in supported_load_backends:
-                log.debug3(
-                    "Attempting to load %s (module_id %s) with backend %s and class %s",
-                    model_path,
-                    module_id,
-                    load_backend.backend_type,
-                    module_backend_impl.__name__,
-                )
-                extra_kwargs = {}
-                if self._supports_load_backend_kwarg(module_backend_impl.load):
-                    extra_kwargs["load_backend"] = load_backend
-                loaded_model = module_backend_impl.load(
-                    model_path,
-                    **extra_kwargs,
-                    **kwargs,
-                )
-                error.type_check("<COR40080753E>", ModuleBase, model=loaded_model)
-                if loaded_model is not None:
-                    log.debug2(
-                        "Successfully loaded %s with backend %s",
-                        model_path,
-                        load_backend.backend_type,
-                    )
-                    loaded_model.set_load_backend(load_backend)
-                    break
-
-        # Return the loaded model if it was able to load
-        return loaded_model
-
-    ## Implementation Details ##################################################
-
-    @staticmethod
-    def _supports_load_backend_kwarg(load_fn: Callable) -> bool:
-        """A load function supports the load_backend kwarg IFF it has an arg
-        explicitly named load_backend or it has a ** kwarg capture
-        """
-        sig = inspect.signature(load_fn)
-        return "load_backend" in sig.parameters or any(
-            param.kind == inspect.Parameter.VAR_KEYWORD
-            for param in sig.parameters.values()
-        )
-
-    def _get_supported_load_backends(self, backend_impl: ModuleBase):
-        """Function to get a list of supported load backends
-        that the module supports
-
-        Args:
-            backend_impl: caikit.core.ModuleBase
-                Module implementing the backend
-        Returns:
-            list(backend_types)
-                list of backends that are supported for model load
-        """
-
-        # Get list of backends that are supported for load
-        # NOTE: since code in a module can change anytime, its support
-        # for various backend might also change, in which case,
-        # it would be better to keep the backend information in the model itself
-        # If module_backend is None, then we will assume that this model is not loadable in
-        # any other backend
-        return getattr(backend_impl, SUPPORTED_LOAD_BACKENDS_VAR_NAME, [])
-
-    @staticmethod
-    def _configure_backend_overrides(backend: str, backend_instance: object):
-        """Function to go over all the modules registered in the MODULE_BACKEND_REGISTRY
-        for a particular backend and configure their backend overrides
-
-        Args:
-            backend: str
-                Name of the backend to select from registry
-            backend_instance: object
-                Initialized backend instance. This object should
-                implement the `register_config` function which will be
-                used to merge / iteratively configure the backend
-        """
-        # Go through all the modules registered with particular backend
-        for module_id, module_type_mapping in module_backend_registry().items():
-            if backend in module_type_mapping:
-                # check if it contains any special config
-                config = module_type_mapping[backend].backend_config_override
-                error.type_check("<COR61136899E>", dict, config=config)
-                if len(config) != 0:
-                    # TODO: Add a check here to see if the backend has already started
-                    backend_instance.register_config(config)
-                else:
-                    log.debug2(
-                        "No backend overrides configured for %s module and %s backend",
-                        module_id,
-                        backend,
-                    )
diff --git a/caikit/core/model_management/model_finder_base.py b/caikit/core/model_management/model_finder_base.py
deleted file mode 100644
index 3d513003..00000000
--- a/caikit/core/model_management/model_finder_base.py
+++ /dev/null
@@ -1,60 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-A ModelFinder is responsible for finding models based on a unique input string
-which may be a path, id, or other identifier.
-
-Configuration for ModelFinders lives under the config as follows:
-
-model_management:
-    finders:
-        <finder name>:
-            type: <finder type name>
-            config:
-                <config option>: <value>
-"""
-
-# Standard
-from typing import Optional
-import abc
-
-# Local
-from ..modules import ModuleConfig
-from ..toolkit.factory import FactoryConstructible
-
-
-class ModelFinderBase(FactoryConstructible):
-    __doc__ = __doc__
-
-    @abc.abstractmethod
-    def find_model(
-        self,
-        model_path: str,
-        **kwargs,
-    ) -> Optional[ModuleConfig]:
-        """Find any model that can be uniquely identified by the given (logical)
-        path. If found, return the in-memory ModuleConfig.
-
-        Args:
-            model_path (str): The logical path to the model (name, id, file
-                path, etc...)
-            **kwargs: All finders must allow additional kwargs through so that
-                specific finders and loaders can support additional optional
-                arguments.
-
-        Returns:
-            result (Optional[ModuleConfig]): If found, the in-memory config
-                object for the model. If the model cannot be found, an exception
-                may be raised or None returned.
-        """
diff --git a/caikit/core/model_management/model_initializer_base.py b/caikit/core/model_management/model_initializer_base.py
deleted file mode 100644
index 227e7e0c..00000000
--- a/caikit/core/model_management/model_initializer_base.py
+++ /dev/null
@@ -1,59 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-A ModelInitializer is responsible for taking an in-memory ModuleConfig and
-producing a usable Module instance.
-
-Configuration for ModelInitializers lives under the config as follows:
-
-model_management:
-    initializers:
-        <initializer name>:
-            type: <initializer type name>
-            config:
-                <config option>: <value>
-"""
-
-# Standard
-from typing import Optional
-import abc
-
-# Local
-from ..modules import ModuleBase, ModuleConfig
-from ..toolkit.factory import FactoryConstructible
-
-
-class ModelInitializerBase(FactoryConstructible):
-    __doc__ = __doc__
-
-    @abc.abstractmethod
-    def init(
-        self,
-        model_config: ModuleConfig,
-        **kwargs,
-    ) -> Optional[ModuleBase]:
-        """Given a ModelConfig, attempt to initialize it for running
-
-        Args:
-            model_config (ModuleConfig): The in-memory model config object for
-                the model to be loaded
-            **kwargs: All loaders must allow additional kwargs through so that
-                specific finders and loaders can support additional optional
-                arguments.
-
-        Returns:
-            model (Optional[ModuleBase]): The in-memory ModuleBase instance that
-                is ready to run if successful. If unable to load, an exception
-                may be raised, or None may be returned.
-        """
diff --git a/caikit/core/model_management/model_trainer_base.py b/caikit/core/model_management/model_trainer_base.py
deleted file mode 100644
index 032dbd6f..00000000
--- a/caikit/core/model_management/model_trainer_base.py
+++ /dev/null
@@ -1,88 +0,0 @@
-# Copyright The Caikit Authors
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""
-A Trainer is responsible for managing execution of a training job for a given
-module class
-"""
-
-# Standard
-from enum import Enum
-from typing import Optional, Type
-import abc
-
-# Local
-from ..modules import ModuleBase
-from ..toolkit.factory import FactoryConstructible
-
-
-class ModelTrainerBase(FactoryConstructible):
-    __doc__ = __doc__
-
-    class TrainingStatus(Enum):
-        """A given training job must be in exactly one of these states at all
-        times
-        """
-
-        QUEUED = 1
-        RUNNING = 2
-        COMPLETED = 3
-        CANCELED = 4
-        ERRORED = 5
-
-    class ModelFutureBase(abc.ABC):
-        """Every Trainer must implement a ModelFuture class that can access the
-        training job in the infrastructure managed by the trainer.
-        """
-
-        @property
-        @abc.abstractmethod
-        def id(self) -> str:
-            """Every model future must have a unique ID that can be used to look
-            up the in-flight training
-            """
-
-        @property
-        @abc.abstractmethod
-        def save_path(self) -> Optional[str]:
-            """If created with a save path, the future must expose it"""
-
-        @abc.abstractmethod
-        def get_status(self) -> "TrainingStatus":
-            """Every model future must be able to poll the status of the
-            training job
-            """
-
-        @abc.abstractmethod
-        def cancel(self):
-            """Terminate the given training"""
-
-        @abc.abstractmethod
-        def wait(self):
-            """Block until the job reaches a terminal state"""
-
-        @abc.abstractmethod
-        def load(self) -> ModuleBase:
-            """A model future must be loadable with no additional arguments"""
-
-    @abc.abstractmethod
-    def train(
-        self,
-        module_class: Type[ModuleBase],
-        *args,
-        save_path: Optional[str] = None,
-        **kwargs,
-    ) -> "ModelFutureBase":
-        """Start training the given module and return a future to the trained
-        model instance
-        """
diff --git a/caikit/core/model_manager.py b/caikit/core/model_manager.py
index c552da7f..2a89b59a 100644
--- a/caikit/core/model_manager.py
+++ b/caikit/core/model_manager.py
@@ -20,7 +20,8 @@
 from contextlib import contextmanager
 from io import BytesIO
 from threading import Lock
-from typing import Dict, Union
+from typing import Union
+import errno
 import os
 import tempfile
 import zipfile
@@ -29,16 +30,13 @@ import zipfile
 import alog
 
 # Local
-from .model_management import (
-    ModelFinderBase,
-    ModelInitializerBase,
-    model_finder_factory,
-    model_initializer_factory,
-)
+from .module_backends import backend_types, module_backend_config
+from .module_backends.base import SharedLoadBackendBase
 from .modules.base import ModuleBase
-from .registries import module_registry
+from .modules.config import ModuleConfig
+from .modules.decorator import SUPPORTED_LOAD_BACKENDS_VAR_NAME
+from .registries import module_backend_registry, module_registry
 from .toolkit.errors import error_handler
-from .toolkit.factory import Factory, FactoryConstructible
 from caikit.config import get_config
 
 log = alog.use_channel("MDLMNG")
@@ -69,53 +67,30 @@ class ModelManager:
         # Map to store module caches, to be used for singleton model lookups
         self.singleton_module_cache = {}
         self._singleton_lock = Lock()
-        self._finders = {}
-        self._initializers = {}
 
     # make load function available from top-level of library
-    def load(
-        self,
-        module_path: str,
-        *,
-        load_singleton: bool = False,
-        finder: Union[str, ModelFinderBase] = "default",
-        initializer: Union[str, ModelInitializerBase] = "default",
-        **kwargs,
-    ):
-        """Load a model and return an instantiated object on which we can run
-        inference.
+    def load(self, module_path, *args, load_singleton=False, **kwargs):
+        """Load a model and return an instantiated object on which we can run inference.
 
         Args:
-            module_path (str | BytesIO | bytes): A module path (identifier) to
-                one of the following:
-                1. A directory containing a yaml config file in the top level.
-                2. A zip archive containing either a yaml config file in the
-                    top level when extracted, or a directory containing a yaml
-                    config file in the top level.
-                3. A BytesIO object corresponding to a zip archive containing
-                    either a yaml config file in the top level when extracted,
-                    or a directory containing a yaml config file in the top
-                    level.
-                4. A bytes object corresponding to a zip archive containing
-                    either a yaml config file in the top level when extracted,
-                    or a directory containing a yaml config file in the top
-                    level.
-                5. A string that is understood by the configured
-                    finder/initializer
-
-        Kwargs:
-            load_singleton (bool): Load this model as a singleton
-            finder (Union[str, ModelFinderBase]): Finder to use when loading
-                this model. If passed as a string, this names the finder in the
-                global config model_management.finders section.
-            initializer (Union[str, ModelInitializerBase]): Loader to use when
-                initializint this model. If passed as a string, this is the name
-                of the initializer in the global
-                config model_management.initializers section.
+            module_path (str | BytesIO | bytes): A module path to one of the
+                following. 1. A module path to a directory containing a yaml
+                config file in the top level. 2. A module path to a zip archive
+                containing either a yaml config file in the top level when
+                extracted, or a directory containing a yaml config file in the
+                top level. 3. A BytesIO object corresponding to a zip archive
+                containing either a yaml config file in the top level when
+                extracted, or a directory containing a yaml config file in the
+                top level. 4. A bytes object corresponding to a zip archive
+                containing either a yaml config file in the top level when
+                extracted, or a directory containing a yaml config file in the
+                top level.
+            load_singleton: bool (Defaults to False)
+                Indicates whether this model should be loaded as a singleton.
 
         Returns:
-            model (ModuleBase) Model object that is loaded, configured, and
-                ready for prediction.
+            subclass of caikit.core.modules.ModuleBase: Model object that is
+                loaded, configured, and ready for prediction.
         """
         error.type_check("<COR98255724E>", bool, load_singleton=load_singleton)
 
@@ -133,16 +108,11 @@ class ModelManager:
         # If we have bytes, convert to a buffer, since we already handle in memory binary streams.
         elif isinstance(module_path, bytes):
             module_path = BytesIO(module_path)
-
         # Now that we have a file like object | str we can try to load as an archive.
         if zipfile.is_zipfile(module_path):
-            return self._load_from_zipfile(
-                module_path, load_singleton, finder, initializer, **kwargs
-            )
+            return self._load_from_zipfile(module_path, load_singleton, *args, **kwargs)
         try:
-            return self._load_from_dir(
-                module_path, load_singleton, finder, initializer, **kwargs
-            )
+            return self._load_from_dir(module_path, load_singleton, *args, **kwargs)
         except FileNotFoundError:
             error(
                 "<COR80419785E>",
@@ -153,60 +123,164 @@ class ModelManager:
                 ),
             )
 
-    def _load_from_dir(
-        self, module_path, load_singleton, finder, initializer, **kwargs
-    ):
+    def _load_from_dir(self, module_path, load_singleton, *args, **kwargs):
         """Load a model from a directory.
 
         Args:
             module_path (str): Path to directory. At the top level of directory
                 is `config.yml` which holds info about the model.
-            load_singleton (bool): Load this model as a singleton
-            finder (Union[str, ModelFinderBase]): Finder to use when loading
-                this model. If passed as a string, this names the finder in the
-                global config model_management.finders section.
-            initializer (Union[str, ModelInitializerBase]): Loader to use when
-                loading this model. If passed as a string, this is the name of
-                the initializer in the global
-                config model_management.initializers section.
+            load_singleton (bool): Indicates whether this model should be loaded
+                as a singleton.
 
         Returns:
             subclass of caikit.core.modules.ModuleBase: Model object that is
                 loaded, configured, and ready for prediction.
         """
+        # Short-circuit the loading process if the path does not exist
+        if not os.path.exists(module_path):
+            raise FileNotFoundError(
+                errno.ENOENT, os.strerror(errno.ENOENT), module_path
+            )
+
+        # If this is a singleton load, the entire body of this function needs to
+        # be locked to avoid concurrent loads on the same model. Otherwise, we
+        # can freely load in parallel.
         with self.singleton_lock(load_singleton):
+
+            # Using the module_path as a key, look for an instance preloaded in the
+            # singleton cache if desired
+            # 🌶🌶🌶 This doesn't work for nested modules
+            # TODO: think about bringing back the `unique_hash` or `tracking_id`
             if singleton_entry := (
                 load_singleton and self.singleton_module_cache.get(module_path)
             ):
                 log.debug("Found %s in the singleton cache", module_path)
                 return singleton_entry
 
-            # Use the given finder to try to find the module config for this
-            # module_path
-            #
-            # NOTE: This will lazily construct named finders if needed
-            log.debug("Attempting to find [%s] with finder %s", module_path, finder)
-            finder = self._get_finder(finder)
-            model_config = finder.find_model(module_path, **kwargs)
-            error.value_check(
-                "<COR92173495E>",
-                model_config is not None,
-                "Unable to find a ModuleConfig for {}",
-                module_path,
-            )
+            # Get the set of configured loaders
+            configured_load_backends = module_backend_config.configured_load_backends()
+            if not configured_load_backends:
+                log.info(
+                    "<COR56759744I>",
+                    "No backends configured! Configuring backends with current configuration",
+                )
+                module_backend_config.configure()
+                configured_load_backends = (
+                    module_backend_config.configured_load_backends()
+                )
 
-            # Use the given initializer to try to load the model
-            #
-            # NOTE: This will lazily construct named initializers if needed
-            initializer = self._get_initializer(initializer)
-            loaded_model = initializer.init(model_config, **kwargs)
-            error.value_check(
-                "<COR50207494E>",
-                loaded_model is not None,
-                "Unable to load model from {} with MODULE_ID {}",
-                module_path,
-                model_config.module_id,
-            )
+            # Pre-initialize variables that will be parsed lazily from the
+            # ModuleConfig if needed. This is done lazily so that loaders which
+            # don't require a config.yml can take precedence over those that do
+            # require one.
+            module_id = None
+            module_implementations = None
+            model_creation_backend = None
+
+            # For each backend, if it's a shared loader, attempt to load the model
+            # directly. If not, parse the module config and look to see if there is
+            # a version of the module available for the given backend
+            loaded_model = None
+            log.debug("Available load backends: %s", configured_load_backends)
+            for i, load_backend in enumerate(configured_load_backends):
+                # If this is a shared loader, try loading the model directly
+                if isinstance(load_backend, SharedLoadBackendBase):
+                    log.debug("Trying shared backend loader")
+                    model = load_backend.load(module_path, *args, **kwargs)
+                    if model is not None:
+                        log.debug2(
+                            "Successfully loaded %s with loader (%d)%s",
+                            module_path,
+                            i,
+                            load_backend.backend_type,
+                        )
+                        error.type_check(
+                            "<COR76726077E>",
+                            ModuleBase,
+                            model=model,
+                        )
+
+                        loaded_model = model
+                        model.set_load_backend(load_backend)
+                        break
+                    log.debug3(
+                        "Could not load %s with loader (%d)%s",
+                        module_path,
+                        i,
+                        load_backend.backend_type,
+                    )
+
+                # If this is not a shared loader, look for an implementation of the
+                # model's module that works with this backend
+                else:
+                    # If this is the first time parsing the module config, do so
+                    if module_id is None:
+                        log.debug2("Loading ModuleConfig from %s", module_path)
+                        module_config = ModuleConfig.load(module_path)
+                        module_id = module_config.module_id
+                        module_implementations = module_backend_registry().get(
+                            module_id, {}
+                        )
+                        log.debug2(
+                            "Number of available backend implementations for %s found: %d",
+                            module_id,
+                            len(module_implementations),
+                        )
+                        # Look up the backend that this model was created with
+                        model_creation_backend = module_config.get(
+                            "model_backend", backend_types.LOCAL
+                        )
+
+                    # Look in the module's implementations for this backend type
+                    backend_impl_obj = module_implementations.get(
+                        load_backend.backend_type
+                    )
+                    if backend_impl_obj is None:
+                        log.debug3(
+                            "Module %s does not support loading with %s",
+                            module_id,
+                            load_backend.backend_type,
+                        )
+                        continue
+
+                    # Grab the concrete module class for this backend and check to
+                    # see if this model's artifacts were created with a version of
+                    # the module that can be loaded with this backend.
+                    module_backend_impl = backend_impl_obj.impl_class
+                    supported_load_backends = self._get_supported_load_backends(
+                        module_backend_impl
+                    )
+                    if model_creation_backend in supported_load_backends:
+                        log.debug3(
+                            "Attempting to load %s (module_id %s) with backend %s and class %s",
+                            module_path,
+                            module_id,
+                            load_backend.backend_type,
+                            module_backend_impl.__name__,
+                        )
+                        loaded_model = module_backend_impl.load(
+                            module_path,
+                            *args,
+                            load_backend=load_backend,
+                            **kwargs,
+                        )
+                        if loaded_model is not None:
+                            log.debug2(
+                                "Successfully loaded %s with backend %s",
+                                module_path,
+                                load_backend.backend_type,
+                            )
+                            loaded_model.set_load_backend(load_backend)
+                            break
+
+            # If no model successfully loaded, it's an error
+            if loaded_model is None:
+                error(
+                    "<COR50207494E>",
+                    ValueError(
+                        f"Unable to load model from {module_path} with MODULE_ID {module_id}"
+                    ),
+                )
 
             # If loading as a singleton, populate the cache
             if load_singleton:
@@ -215,22 +289,14 @@ class ModelManager:
             # Return successfully!
             return loaded_model
 
-    def _load_from_zipfile(
-        self, module_path, load_singleton, finder, initializer, **kwargs
-    ):
+    def _load_from_zipfile(self, module_path, load_singleton, *args, **kwargs):
         """Load a model from a zip archive.
 
         Args:
             module_path (str): Path to directory. At the top level of directory
                 is `config.yml` which holds info about the model.
-            load_singleton (bool): Load this model as a singleton
-            finder (Union[str, ModelFinderBase]): Finder to use when loading
-                this model. If passed as a string, this names the finder in the
-                global config model_management.finders section.
-            initializer (Union[str, ModelInitializerBase]): Loader to use when
-                loading this model. If passed as a string, this is the name of
-                the initializer in the global
-                config model_management.initializers section.
+            load_singleton (bool): Indicates whether this model should be loaded
+                as a singleton.
 
         Returns:
             subclass of caikit.core.modules.ModuleBase: Model object that is
@@ -244,7 +310,7 @@ class ModelManager:
             # We expect the former, but fall back to the second if we can't find the config.
             try:
                 model = self._load_from_dir(
-                    extract_path, load_singleton, finder, initializer, **kwargs
+                    extract_path, load_singleton, *args, **kwargs
                 )
             # NOTE: Error handling is a little gross here, the main reason being that we
             # only want to log to error() if something is fatal, and there are a good amount
@@ -274,7 +340,7 @@ class ModelManager:
                 # create one potential extra layer of nesting around the model directory.
                 try:
                     model = self._load_from_dir(
-                        nested_dirs[0], load_singleton, finder, initializer, **kwargs
+                        nested_dirs[0], load_singleton, *args, **kwargs
                     )
                 except FileNotFoundError:
                     error(
@@ -398,57 +464,22 @@ class ModelManager:
         else:
             yield
 
-    @staticmethod
-    def _get_component(
-        component: Union[str, FactoryConstructible],
-        component_dict: Dict[str, FactoryConstructible],
-        component_factory: Factory,
-        component_name: str,
-        component_cfg: dict,
-        component_type: type,
-    ) -> FactoryConstructible:
-        """Common logic for resolving components from config
-
-        NOTE: This is done lazily to avoid relying on import order and to allow
-            for dynamic config changes
+    def _get_supported_load_backends(self, backend_impl: ModuleBase):
+        """Function to get a list of supported load backends
+        that the module supports
+
+        Args:
+            backend_impl (caikit.core.ModuleBase): Module implementing the
+                backend
+        Returns:
+            list(backend_types): list of backends that are supported for model
+                load
         """
-        error.type_check(
-            "<COR45466249E>", str, component_type, **{component_name: component}
-        )
-        if isinstance(component, component_type):
-            return component
-        if component not in component_dict:
-            cfg = component_cfg.get(component)
-            error.value_check(
-                "<COR55057389E>",
-                isinstance(cfg, dict),
-                "Unknown {}: {}",
-                component_name,
-                component,
-            )
-            component_dict[component] = component_factory.construct(cfg)
-        return component_dict[component]
-
-    def _get_finder(self, finder: Union[str, ModelFinderBase]) -> ModelFinderBase:
-        """Get the configured model finder or the one passed by value"""
-        return self._get_component(
-            component=finder,
-            component_dict=self._finders,
-            component_factory=model_finder_factory,
-            component_name="finder",
-            component_cfg=get_config().model_management.finders,
-            component_type=ModelFinderBase,
-        )
 
-    def _get_initializer(
-        self, initializer: Union[str, ModelInitializerBase]
-    ) -> ModelInitializerBase:
-        """Get the configured model initializer or the one passed by value"""
-        return self._get_component(
-            component=initializer,
-            component_dict=self._initializers,
-            component_factory=model_initializer_factory,
-            component_name="initializer",
-            component_cfg=get_config().model_management.initializers,
-            component_type=ModelInitializerBase,
-        )
+        # Get list of backends that are supported for load
+        # NOTE: since code in a module can change anytime, its support
+        # for various backend might also change, in which case,
+        # it would be better to keep the backend information in the model itself
+        # If module_backend is None, then we will assume that this model is not loadable in
+        # any other backend
+        return getattr(backend_impl, SUPPORTED_LOAD_BACKENDS_VAR_NAME, [])
diff --git a/caikit/core/module_backends/base.py b/caikit/core/module_backends/base.py
index c3ebc321..04266ae1 100644
--- a/caikit/core/module_backends/base.py
+++ b/caikit/core/module_backends/base.py
@@ -18,12 +18,15 @@
 # Standard
 from contextlib import contextmanager
 from threading import Lock
-from typing import Optional
+from typing import Optional, Type
 import abc
 
 # First Party
 import aconfig
 
+# Local
+from ..modules import ModuleBase
+
 
 class BackendBase(abc.ABC):
     """Interface for creating configuration setup for backends"""
@@ -66,3 +69,66 @@ class BackendBase(abc.ABC):
     def start_lock(self):
         with self._start_lock:
             yield
+
+
+class SharedTrainBackendBase(BackendBase, abc.ABC):
+    """Interface for a backend that can perform train on any given module
+
+    A Shared backend is one that treats the given module as a black box and
+    delegates the execution of that module's functionality to an alternate
+    execution engine.
+    """
+
+    @abc.abstractmethod
+    def train(self, module_class: Type[ModuleBase], *args, **kwargs) -> ModuleBase:
+        """Perform the given module's train operation and return the trained
+        module instance.
+
+        TODO: The return type here might be problematic in the case where the
+            server performing the train operation is just a proxy for both train
+            and inference. Consider some kind of lazy load proxy that would not
+            require the model to be held in memory.
+
+        Args:
+            module_class (Type[ModuleBase]): The module class to train
+            *args, **kwargs: The args to pass through to training
+        Returns:
+            model (ModuleBase): The in-memory instance of the trained
+                module
+        """
+
+
+class SharedLoadBackendBase(BackendBase, abc.ABC):
+    """Interface for a backend that can perform load/unload on any given model
+
+    A Shared backend is one that treats the given module as a black box and
+    delegates the execution of that module's functionality to an alternate
+    execution engine.
+
+    The module returned by a universal manager must be capable of having run
+    called locally and delegating the execution of the underlying module to the
+    backend's framework.
+    """
+
+    @abc.abstractmethod
+    def load(self, model_path: str, *args, **kwargs) -> Optional[ModuleBase]:
+        """Load the model stored at the given path into the backend
+
+        This function is responsible for loading a model in a way that the
+        backend is then able to execute it.
+
+        Shared loaders will be configured in a priority sequence. If a higher
+        priority loader fails to load a given model, the next one is attempted
+        until the model is loaded or no loaders are left.
+
+        Args:
+            model_path (str): Path to directory or zip file holding the model
+                with the config.yml and any artifacts
+            *args, **kwargs: Additional args to pass through to the module's
+                load function
+
+        Returns:
+            model (Optional[ModuleBase]): A runnable model if one could be
+                loaded, otherwise None. Such a model may be a wrapper that
+                delegates execution to the concrete model loaded elsewhere.
+        """
diff --git a/caikit/core/module_backends/local_backend.py b/caikit/core/module_backends/local_backend.py
index 567de1bb..e6768141 100644
--- a/caikit/core/module_backends/local_backend.py
+++ b/caikit/core/module_backends/local_backend.py
@@ -14,21 +14,29 @@
 
 """This module implements a LOCAL backend configuration
 """
+# Standard
+from typing import Optional, Type
+
+# Third Party
+import yaml
 
 # First Party
 import alog
 
 # Local
+from ..modules.base import ModuleBase
+from ..modules.config import ModuleConfig
+from ..registries import module_registry
 from ..toolkit.errors import error_handler
 from ..toolkit.wip_decorator import TempDisableWIP
 from .backend_types import register_backend_type
-from .base import BackendBase
+from .base import SharedLoadBackendBase, SharedTrainBackendBase
 
 log = alog.use_channel("LCLBKND")
 error = error_handler.get(log)
 
 
-class LocalBackend(BackendBase):
+class LocalBackend(SharedLoadBackendBase, SharedTrainBackendBase):
     backend_type = "LOCAL"
 
     def register_config(self, config) -> None:
@@ -48,6 +56,29 @@ class LocalBackend(BackendBase):
         """Stop local backend. This is a no-op"""
         self._started = False
 
+    def train(self, module_class: Type[ModuleBase], *args, **kwargs) -> ModuleBase:
+        """Perform a local training on the given class"""
+        with alog.ContextTimer(log.info, "Finished local training in: "):
+            return module_class.train(*args, **kwargs)
+
+    def load(self, model_path: str, *args, **kwargs) -> Optional[ModuleBase]:
+        """Look up the given module in the module registry and load it if found"""
+        try:
+            model_config = ModuleConfig.load(model_path)
+        except (FileNotFoundError, KeyError, yaml.parser.ParserError):
+            log.debug("Unable to load local model config from [%s]", model_path)
+            return None
+        module_id = model_config.module_id
+        module_class = module_registry().get(module_id)
+        if module_class is not None:
+            return module_class.load(model_path, *args, **kwargs)
+        log.warning(
+            "<COR14661026W>",
+            "Could not load MODULE_ID %s with %s",
+            module_id,
+            self.backend_type,
+        )
+
 
 # Register local backend
 with TempDisableWIP():
diff --git a/caikit/core/module_backends/module_backend_config.py b/caikit/core/module_backends/module_backend_config.py
new file mode 100644
index 00000000..d98cc2be
--- /dev/null
+++ b/caikit/core/module_backends/module_backend_config.py
@@ -0,0 +1,164 @@
+# Copyright The Caikit Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Standard
+from typing import List
+import copy
+
+# First Party
+import alog
+
+# Local
+from ..registries import (
+    module_backend_classes,
+    module_backend_registry,
+    module_backend_types,
+)
+from ..toolkit.errors import error_handler
+from .base import BackendBase
+from caikit.config import get_config
+
+log = alog.use_channel("CONF")
+error = error_handler.get(log)
+
+
+def start_backends() -> None:
+    """This function kicks off the `start` functions for configured backends
+    Returns:
+        None
+    """
+    for backend in _CONFIGURED_LOAD_BACKENDS:
+        with backend.start_lock():
+            if not backend.is_started:
+                backend.start()
+    for backend in _CONFIGURED_TRAIN_BACKENDS:
+        with backend.start_lock():
+            if not backend.is_started:
+                backend.start()
+
+
+def configured_load_backends() -> List[BackendBase]:
+    """This function returns the list of configured load backends"""
+    return copy.copy(_CONFIGURED_LOAD_BACKENDS)
+
+
+def configured_train_backends() -> List[BackendBase]:
+    """This function returns the list of configured train backends"""
+    return copy.copy(_CONFIGURED_TRAIN_BACKENDS)
+
+
+def configure():
+    """Configure the backend environment
+
+    NOTE: This function is NOT thread safe!
+    """
+    config_object = get_config().module_backends
+    log.debug3("Full Config: %s", config_object)
+
+    # Configure both train and load backends
+    for backend_priority, registry, registry_name in [
+        (config_object.load_priority, _CONFIGURED_LOAD_BACKENDS, "load"),
+        (config_object.train_priority, _CONFIGURED_TRAIN_BACKENDS, "train"),
+    ]:
+        backend_priority = backend_priority or []
+        error.type_check("<COR46006487E>", list, backend_priority=backend_priority)
+
+        # Configure each backend instance
+        for i, backend_config in enumerate(backend_priority):
+            error.value_check(
+                "<COR48633635E>",
+                "type" in backend_config,
+                "All backend priority configs must have a 'type' field",
+            )
+            backend_type = backend_config.type
+            error.value_check(
+                "<COR72281596E>",
+                backend_type in module_backend_types(),
+                "Invalid backend [{}] found at backend_priority index [{}]",
+                backend_type,
+                i,
+            )
+
+            log.debug("Configuring backend (%d)[%s]", i, backend_type)
+            backend_instance_config = backend_config.get("config", {})
+            log.debug3(
+                "Backend (%d)[%s] config: %s", i, backend_type, backend_instance_config
+            )
+
+            backend_class = module_backend_classes().get(backend_type)
+            error.value_check(
+                "<COR64618509E>",
+                len(registry) == i,
+                "({})[{}] already configured",
+                i,
+                backend_type,
+            )
+            error.value_check(
+                "<COR39517372E>",
+                backend_class is not None,
+                "Unsupported backend type {}",
+                backend_type,
+            )
+            if not isinstance(backend_class, type) and issubclass(
+                backend_class, BackendBase
+            ):
+                error(
+                    "<COR05184600E>",
+                    TypeError(
+                        f"Backend {backend_class} is not derived from BackendBase"
+                    ),
+                )
+
+            log.debug2("Performing config for (%d)[%s]", i, backend_type)
+            backend_instance = backend_class(backend_instance_config)
+
+            # Add configuration to backends as per individual module requirements
+            _configure_backend_overrides(backend_type, backend_instance)
+
+            # Add the instance to the registry
+            registry.append(backend_instance)
+
+        log.debug2("All configured %s backends: %s", registry_name, registry)
+
+
+## Implementation Details ######################################################
+
+# Singleton registries for load and train backends in priority order
+_CONFIGURED_LOAD_BACKENDS = []
+_CONFIGURED_TRAIN_BACKENDS = []
+
+
+def _configure_backend_overrides(backend: str, backend_instance: object):
+    """Function to go over all the modules registered in the MODULE_BACKEND_REGISTRY
+    for a particular backend and configure their backend overrides
+
+    Args:
+        backend (str): Name of the backend to select from registry
+        backend_instance (object): Initialized backend instance. This object
+            should implement the `register_config` function which will be used
+            to merge / iteratively configure the backend
+    """
+    # Go through all the modules registered with particular backend
+    for module_id, module_type_mapping in module_backend_registry().items():
+        if backend in module_type_mapping:
+            # check if it contains any special config
+            config = module_type_mapping[backend].backend_config_override
+            error.type_check("<COR61136899E>", dict, config=config)
+            if len(config) != 0:
+                # TODO: Add a check here to see if the backend has already started
+                backend_instance.register_config(config)
+            else:
+                log.debug2(
+                    f"No backend overrides configured for {module_id} module and {backend} backend"
+                )
diff --git a/caikit/core/modules/meta.py b/caikit/core/modules/meta.py
index 45f68118..d3d1ac05 100644
--- a/caikit/core/modules/meta.py
+++ b/caikit/core/modules/meta.py
@@ -67,7 +67,6 @@ is known.
 
 # Standard
 import abc
-import functools
 
 # First Party
 import alog
@@ -143,11 +142,6 @@ class _ModuleBaseMeta(abc.ABCMeta):
 
                 return module
 
-            # Wrap the load function so that the final method appears the same
-            # as the original
-            metadata_injecting_load = functools.wraps(real_load.__func__)(
-                metadata_injecting_load
-            )
             attrs["load"] = classmethod(metadata_injecting_load)
 
         return super().__new__(mcs, name, bases, attrs)
diff --git a/docs/adrs/018-shared-backends.md b/docs/adrs/018-shared-backends.md
index bf82b12d..1302eaa6 100644
--- a/docs/adrs/018-shared-backends.md
+++ b/docs/adrs/018-shared-backends.md
@@ -16,7 +16,7 @@ One particular decision here is that the responsibility for delegating `inferenc
 
 ## Status
 
-choose one: [Superseded by 020-model-management-abstractions](020-model-management-abstractions.md)
+choose one: Accepted
 
 ## Consequences
 
diff --git a/docs/adrs/019-loader-stack.md b/docs/adrs/019-loader-stack.md
index 4f45b974..d87e9883 100644
--- a/docs/adrs/019-loader-stack.md
+++ b/docs/adrs/019-loader-stack.md
@@ -12,7 +12,7 @@ When loading a model in a non-local backend, there are several ways to map a giv
 
 ## Status
 
-choose one: [Superseded by 020-model-management-abstractions](020-model-management-abstractions.md)
+choose one: Accepted
 
 ## Consequences
 
diff --git a/docs/adrs/020-model-management-abstractions.md b/docs/adrs/020-model-management-abstractions.md
deleted file mode 100644
index 7e42e889..00000000
--- a/docs/adrs/020-model-management-abstractions.md
+++ /dev/null
@@ -1,58 +0,0 @@
-# ADR 020: Model Management Abstractions
-
-When managing models, there are three (and possibly eventually more) central abstract tasks:
-
-* `training`: Take raw materials (including pretrained models) and create usable concrete model artifacts
-* `finding`: Given some unique identifier for a concrete model, identify how the model should be loaded in `caikit`
-* `initialization`: Given the information about a model that can be loaded in `caikit`, prepare the artifacts for online inference
-
-The default implementation of each of these tasks is to perform them locally in the python process using the standard `caikit` model format, but each could be implemented using alternate technology. Here are some examples:
-
-* `training`:
-    * Train in a remote `ray` cluster where the `def train` function is executed as a `ray` job
-    * Train from a local `caikit` library python process using a hosted training API in a running instance of `caikit.runtime` in a remote kubernetes cluster
-* `finding`:
-    * Find models in an alternate saving format such as `transformers`
-    * Find models hosted in a remote storage service such as S3
-* `initializing`:
-    * Load the artifacts using `ray` serving and proxy `run` calls to the instance running in the ray cluster
-    * Delegate the model to a hosted `caikit.runtime` instance in a remote cluster
-
-The previous ADRs for [018-shared-backends](018-shared-backends.md) and [019-loader-stack](019-loader-stack.md) had several key problems:
-
-* They attempted to solve these problems by merging them with the `module_backend` abstraction which has caused a lot of confusion since it was adopted.
-* They didn't differentiate between `finding` and `initialization`
-* They relied on an ordered list of configured objects to discover the right loader instance which was hard to configure in a running server since lists are not well supported with environment variables
-    * This also did not allow explicit specification of a loader for a model where mulitple loaders in the list could work, so you could not load one model one way and another a different way
-
-Prior to `018` and `019`, the `module_backend` abstraction was explicitly tasked with performing "internal distribution" (distributing the work of a given module to some framework, requiring specific knowledge of the algorithm implemented by the module). With `018` and `019`, `module_backend` was extended to also support "external distribution" (distributing the whole operation of a given module as a black box without knowledge of the module's algorithm). This was at the heart of what lead to the confusion, so in this ADR we walk that back and split those responsibilities between `module_backend` (responsible for "internal distribution") and `initializer` (responsible for "external distribution").
-
-## Decision
-
-* The `SharedTrainBackend` an `SharedLoadBackend` abstract classes will be removed
-* The logic in `ModelManager` that handles `SharedLoadBackend`s will be removed
-* A new set of abstract classes will be introduced in `caikit.core.model_management`, one each for `ModelTrainer`, `ModelFinder`, and `ModelLoader`
-* The `ModelManager.load` implementation will be refactored to do the following:
-    * Get a unique `finder` by name or value, falling back to a `"default"` instance and attempt to find the model based on `model_path`. A successful finder returns a `ModuleConfig` containing the `module_id` of the `module` class that should be used to initialize the model.
-    * If the model is found, get a unique `initializer` by name or value, falling back to a `"default"` instance and attempt to initialize the found model using it.
-* All of the responsibility for managing `module_backend` mapping will move to the `LOCAL` implementation of the `ModelLoader` abstraction
-* The global `config` will be refactored as follows:
-    * `module_backends` will be removed
-    * A new section for `model_management` will be added with `trainers`, `finders`, and `initializers` under it
-    * Each section will be a key/value map where the key is a unique string name and the value is a factory-constructible object with `"type"` and (optionally) `"config"` keys
-    * The `backend_priority` logic will move to the `"config"` key of the `LOCAL` `initializer` configuration blob
-    * Each section will be required to have a `"default"` key for the instance to use when none is passed by name
-* (FUTURE) `ModelManager` will introduce a `train` function that uses the `ModelTrainer` abstraction in the same way that `load` uses the `ModelFinder`/`ModelLoader` abstractions
-* Each abstraction will be managed using an extensible factory that can be extended in external libraries
-
-## Status
-
-choose one: Accepted
-
-
-## Consequences
-
-* New abstraction implementations can be added in external libraries to implement platform-specific logic (e.g. `S3BucketModuleFinder`)
-* The confusion around `module_backends` will be removed and they will go back to being implementation details of specific modules
-* Users will be able to explicitly name the `finder`/`initializer` to use for a given model without relying on global configuration
-* The `config` API will break, causing churn with current user deployment topology
diff --git a/examples/text-sentiment/README.md b/examples/text-sentiment/README.md
index 22c22d9f..d1beaf75 100644
--- a/examples/text-sentiment/README.md
+++ b/examples/text-sentiment/README.md
@@ -9,7 +9,7 @@ The following tools are required:
 - [python](https://www.python.org) (v3.8+)
 - [pip](https://pypi.org/project/pip/) (v23.0+)
 
-**Note: Before installing dependencies and to avoid conflicts in your environment, it is advisable to use a [virtual environment(venv)](https://docs.python.org/3/library/venv.html).**
+**Note: Before installing dependencies and to avoid conflicts in your environment, it is advisable to use a [virtual environment (venv)](https://docs.python.org/3/library/venv.html).**
 
 Install the dependencies: `pip install -r requirements.txt`
 
diff --git a/examples/text-sentiment/start_runtime.py b/examples/text-sentiment/start_runtime.py
index 0440fbb4..93fefe45 100644
--- a/examples/text-sentiment/start_runtime.py
+++ b/examples/text-sentiment/start_runtime.py
@@ -19,6 +19,17 @@ import sys
 # First Party
 import alog
 
+# Local
+import caikit
+
+models_directory = path.abspath(path.join(path.dirname(__file__), "models"))
+caikit.config.configure(
+    config_dict={
+        "merge_strategy": "merge",
+        "runtime": {"local_models_dir": models_directory},
+    }
+)
+
 sys.path.append(
     path.abspath(path.join(path.dirname(__file__), "../"))
 )  # Here we assume that `start_runtime` file is at the same level of the `text_sentiment` package


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 5831ac5d..31b7f529 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -4,7 +4,6 @@ This sets up global test configs when pytest starts
 
 # Standard
 from contextlib import contextmanager
-from typing import List, Union
 from unittest.mock import patch
 import copy
 import json
@@ -165,26 +164,6 @@ def toolkit_fixtures_dir(fixtures_dir):
     yield os.path.join(fixtures_dir, "toolkit")
 
 
-@contextmanager
-def backend_priority(backend_cfg: Union[List[dict], dict]):
-    if isinstance(backend_cfg, dict):
-        backend_cfg = [backend_cfg]
-    with temp_config(
-        {
-            "model_management": {
-                "finders": {"default": {"type": "LOCAL"}},
-                "initializers": {
-                    "default": {
-                        "type": "LOCAL",
-                        "config": {"backend_priority": backend_cfg},
-                    }
-                },
-            }
-        }
-    ):
-        yield
-
-
 # IMPLEMENTATION DETAILS ############################################################
 def _random_id():
     return str(uuid.uuid4())
diff --git a/tests/core/helpers.py b/tests/core/helpers.py
index 5a83e15f..a32f3e95 100644
--- a/tests/core/helpers.py
+++ b/tests/core/helpers.py
@@ -20,14 +20,17 @@ import copy
 import pytest
 
 # Local
-from caikit.core import MODEL_MANAGER
+from caikit.core import LocalBackend
+from caikit.core.module_backends import BackendBase, backend_types
 
 # Add mock backend
 # This is set in the base test config's load_priority list
-from caikit.core.model_management import ModelInitializerBase, model_initializer_factory
-from caikit.core.model_management.local_model_initializer import LocalModelInitializer
-from caikit.core.module_backends import BackendBase, backend_types
-from caikit.core.modules import ModuleBase, ModuleConfig
+from caikit.core.module_backends.base import SharedLoadBackendBase
+from caikit.core.module_backends.module_backend_config import (
+    _CONFIGURED_LOAD_BACKENDS,
+    _CONFIGURED_TRAIN_BACKENDS,
+)
+from caikit.core.modules.base import ModuleBase
 from caikit.core.registries import (
     module_backend_classes,
     module_backend_registry,
@@ -57,37 +60,31 @@ backend_types.register_backend_type(MockBackend)
 
 
 # Add a new shared load backend that tests can use
-class TestInitializer(ModelInitializerBase):
-    name = "TestInitializer"
+class TestLoader(SharedLoadBackendBase):
+    backend_type = "TESTLOADER"
     __test__ = False
 
-    def __init__(self, config):
-        self.config = config
-        self.loaded_models = []
-        self.local_initializer = model_initializer_factory.construct({"type": "LOCAL"})
-
-    def init(self, model_config: ModuleConfig, *args, **kwargs) -> Optional[ModuleBase]:
+    def load(self, model_path: str, *args, **kwargs) -> Optional[ModuleBase]:
         # allow config.model_type to control whether this loader barfs
         if "model_type" in self.config and "model_type" in kwargs:
             if self.config["model_type"] != kwargs["model_type"]:
                 # Don't load in this loader
                 return None
         # use the "Local" loader to actually load the model
-        model = self.local_initializer.init(model_config)
-        self.loaded_models.append(model)
+        model = LocalBackend().load(model_path)
         return model
 
+    def register_config(self, config):
+        pass
+
+    def stop(self):
+        pass
 
-model_initializer_factory.register(TestInitializer)
+    def start(self):
+        pass
 
 
-def configured_backends():
-    local_initializers = [
-        loader
-        for loader in MODEL_MANAGER._initializers.values()
-        if isinstance(loader, LocalModelInitializer)
-    ]
-    return [backend for loader in local_initializers for backend in loader._backends]
+backend_types.register_backend_type(TestLoader)
 
 
 @pytest.fixture
@@ -123,21 +120,24 @@ def reset_module_registry():
 
 
 @pytest.fixture
-def reset_model_manager():
-    prev_finders = MODEL_MANAGER._finders
-    prev_initializers = MODEL_MANAGER._initializers
-    MODEL_MANAGER._finders = {}
-    MODEL_MANAGER._initializers = {}
+def reset_configured_backends():
+    """Fixture that will reset the configured backends"""
+    load_backends_list = copy.copy(_CONFIGURED_LOAD_BACKENDS)
+    train_backends_list = copy.copy(_CONFIGURED_TRAIN_BACKENDS)
+    _CONFIGURED_LOAD_BACKENDS.clear()
+    _CONFIGURED_TRAIN_BACKENDS.clear()
     yield
-    MODEL_MANAGER._finders = prev_finders
-    MODEL_MANAGER._initializers = prev_initializers
+    _CONFIGURED_LOAD_BACKENDS.clear()
+    _CONFIGURED_LOAD_BACKENDS.extend(load_backends_list)
+    _CONFIGURED_TRAIN_BACKENDS.clear()
+    _CONFIGURED_TRAIN_BACKENDS.extend(train_backends_list)
 
 
 @pytest.fixture
 def reset_globals(
     reset_backend_types,
-    reset_model_manager,
-    reset_module_backend_registry,
+    reset_configured_backends,
     reset_module_registry,
+    reset_module_backend_registry,
 ):
     """Fixture that will reset the backend types and module registries if a test modifies them"""
diff --git a/tests/core/module_backends/test_module_backend_config.py b/tests/core/module_backends/test_module_backend_config.py
new file mode 100644
index 00000000..7b575f01
--- /dev/null
+++ b/tests/core/module_backends/test_module_backend_config.py
@@ -0,0 +1,235 @@
+# Copyright The Caikit Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Tests for the backend configuration framework
+"""
+
+# Local
+from caikit.core.module_backends.module_backend_config import (
+    configure,
+    configured_load_backends,
+    configured_train_backends,
+    start_backends,
+)
+from caikit.core.modules import base, module
+from sample_lib.data_model import SampleTask
+from tests.conftest import temp_config
+from tests.core.helpers import *
+
+# Setup #########################################################################
+
+foo_cfg = {"mock": 1}
+
+
+def _get_backend(backend_type, backend_list):
+    """Get the single configured backend of the given type. An error is raised
+    if the number of matches != 1
+    """
+    matches = [
+        backend for backend in backend_list if backend.backend_type == backend_type
+    ]
+    assert len(matches) == 1
+    return matches[0]
+
+
+def get_load_backend(backend_type):
+    return _get_backend(backend_type, configured_load_backends())
+
+
+def get_train_backend(backend_type):
+    return _get_backend(backend_type, configured_train_backends())
+
+
+## Tests #######################################################################
+
+
+def test_configure_with_module(reset_globals):
+    """Test that configuring with a configured backend type that has a
+    configuration module obj works
+    """
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {
+                        "type": backend_types.MOCK,
+                        "config": foo_cfg,
+                    }
+                ],
+                "train_priority": [
+                    {
+                        "type": backend_types.MOCK,
+                        "config": foo_cfg,
+                    }
+                ],
+            }
+        }
+    ):
+        configure()
+
+        # Test load backend config
+        mock_load_backend = get_load_backend(backend_types.MOCK)
+        assert mock_load_backend.backend_type == backend_types.MOCK
+        assert foo_cfg == mock_load_backend.config
+
+        # Test train backend config
+        mock_train_backend = get_train_backend(backend_types.MOCK)
+        assert mock_train_backend.backend_type == backend_types.MOCK
+        assert foo_cfg == mock_train_backend.config
+
+
+def test_configure_load_only(reset_globals):
+    """Test that train and load backends can be configured independently"""
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {
+                        "type": backend_types.MOCK,
+                        "config": foo_cfg,
+                    }
+                ],
+                "train_priority": [],
+            }
+        }
+    ):
+        configure()
+
+        # Test load backend config
+        mock_load_backend = get_load_backend(backend_types.MOCK)
+        assert mock_load_backend.backend_type == backend_types.MOCK
+        assert foo_cfg == mock_load_backend.config
+
+        # Test train backend config
+        with pytest.raises(AssertionError):
+            get_train_backend(backend_types.MOCK)
+
+
+def test_non_supported_backend_raises():
+    """Test that backend provided as priority to configure raises
+    if not registered"""
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": "unsupported"}],
+            }
+        }
+    ):
+        with pytest.raises(ValueError):
+            configure()
+
+
+def test_duplicate_config_raises(reset_globals):
+    """Test that duplicate configuration of a backend raises"""
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {"type": backend_types.MOCK},
+                ],
+            }
+        }
+    ):
+        # The mock backend is already registered for tests in the core
+        # see tests/core/helpers.py
+        configure()
+        with pytest.raises(ValueError):
+            configure()
+
+
+def test_duplicate_explicit_instances_allowed(reset_globals):
+    """Test that duplicate entries are allowed"""
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {"type": backend_types.MOCK},
+                    {"type": backend_types.MOCK},
+                ],
+            }
+        }
+    ):
+        configure()
+
+
+def test_one_configured_backend_can_start(reset_globals):
+    """Test that the configured backend can be started"""
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {
+                        "type": backend_types.MOCK,
+                        "config": foo_cfg,
+                    }
+                ],
+            }
+        }
+    ):
+        configure()
+        start_backends()
+
+        # This is configured to be True in helpers
+        mock_load_backend = get_load_backend(backend_types.MOCK)
+        assert mock_load_backend.backend_type == backend_types.MOCK
+        assert mock_load_backend.is_started
+
+
+def test_multiple_module_same_backend_configures(reset_globals):
+    """Test to check if multiple modules for same backend
+    can override backend configurations"""
+    # Register backend type
+
+    @module(id="foo", name="dummy base", version="0.0.1")
+    class DummyFoo(base.ModuleBase):
+        pass
+
+    # Create dummy classes
+    @module(
+        base_module=DummyFoo,
+        backend_type=backend_types.MOCK,
+        backend_config_override={"bar1": 1},
+    )
+    class DummyBar(base.ModuleBase):
+        pass
+
+    @module(id="foo2", name="dummy base", version="0.0.1")
+    class DummyFoo2(base.ModuleBase):
+        pass
+
+    # Create dummy classes
+    @module(
+        base_module=DummyFoo2,
+        backend_type=backend_types.MOCK,
+        backend_config_override={"bar2": 2},
+    )
+    class DummyBar(base.ModuleBase):
+        pass
+
+    # Initiate configuration
+
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
+        mock_load_backend = get_load_backend(backend_types.MOCK)
+        assert mock_load_backend.backend_type == backend_types.MOCK
+        assert "bar1" in mock_load_backend.config
+        assert "bar2" in mock_load_backend.config
+        assert mock_load_backend.config["bar1"] == 1
diff --git a/tests/core/modules/test_module.py b/tests/core/modules/test_module.py
index d7189c9a..0b5e6233 100644
--- a/tests/core/modules/test_module.py
+++ b/tests/core/modules/test_module.py
@@ -22,6 +22,7 @@ import aconfig
 
 # Local
 from caikit.core import ModuleConfig, ModuleLoader
+from caikit.core.module_backends.module_backend_config import configured_load_backends
 from caikit.core.modules.decorator import SUPPORTED_LOAD_BACKENDS_VAR_NAME
 
 # pylint: disable=import-error
@@ -71,7 +72,7 @@ def configure_alternate_backend_impl():
         def test_fetching_backend(self):
             return [
                 backend
-                for backend in configured_backends()
+                for backend in configured_load_backends()
                 if backend.backend_type == backend_types.MOCK
             ][0]
 
@@ -328,30 +329,20 @@ def test_class_attributes(reset_globals):
     assert DummyBar.MODULE_CLASS
     assert DummyBar.PRODUCER_ID
 
+    # No configured modules before configure
+    assert not configured_load_backends()
+
     # Configure and make sure it can be fetched by the class
     with temp_config(
         {
-            "model_management": {
-                "initializers": {
-                    "default": {
-                        "type": "LOCAL",
-                        "config": {
-                            "backend_priority": [
-                                {"type": backend_types.MOCK},
-                            ]
-                        },
-                    },
-                }
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
             }
         }
     ):
+        caikit.core.module_backends.module_backend_config.configure()
         assert DummyBar.BACKEND_TYPE == backend_types.MOCK
 
-        # Normally non-local backend modules are loaded through caikit.load, so
-        # we need to make sure the local loader has been properly configured to
-        # enable local instance instantiation.
-        assert MODEL_MANAGER._get_initializer("default")
-
         # Make sure an instance can fetch via get_backend()
         inst = DummyBar()
         assert inst.test_fetching_backend().backend_type == backend_types.MOCK
diff --git a/tests/core/test_model_manager.py b/tests/core/test_model_manager.py
index 2d285fd6..a3bba692 100644
--- a/tests/core/test_model_manager.py
+++ b/tests/core/test_model_manager.py
@@ -21,16 +21,20 @@ from unittest import mock
 from unittest.mock import MagicMock
 import os
 import tempfile
+import uuid
 
 # Local
-from caikit.core import MODEL_MANAGER, LocalBackend
-from caikit.core.model_management import ModelFinderBase, model_finder_factory
-from caikit.core.modules import ModuleConfig
+from caikit.core.module_backends import module_backend_config
+from caikit.core.module_backends.module_backend_config import (
+    configure,
+    configured_load_backends,
+)
+from sample_lib.data_model import SampleTask
 
 # Unit Test Infrastructure
 from sample_lib.modules.sample_task import SampleModule
 from tests.base import TestCaseBase
-from tests.conftest import backend_priority, temp_config
+from tests.conftest import temp_config
 
 # NOTE: We do need to import `reset_backend_types` and `reset_module_distribution_registry` for `reset_globals` to work
 from tests.core.helpers import *
@@ -241,39 +245,19 @@ def temp_saved_model(model):
         yield model_path
 
 
-class NoYamlFinder(ModelFinderBase):
-    name = "NOYAML"
-
-    def __init__(self, *_, **__):
-        pass
-
-    def find_model(self, model_path):
-        """This load function doesn't read from model_path, so it definitely does not read the config.yml file"""
-        return ModuleConfig({"module_id": SampleModule.MODULE_ID})
-
-
-class NoYamlInitializer(ModelInitializerBase):
-    name = "NOYAML"
-
-    def __init__(self, *_, **__):
-        pass
-
-    def init(self, model_config):
-        """This load function doesn't read from model_path, so it definitely does not read the config.yml file"""
-        assert model_config.module_id == SampleModule.MODULE_ID
-        return SampleModule()
-
-
-model_finder_factory.register(NoYamlFinder)
-model_initializer_factory.register(NoYamlInitializer)
-
 ## Tests #######################################################################
 
 
 def test_backend_supported_model_load_successfully(reset_globals):
     """Test backend supported model can load successfully"""
+
     _, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    # Configure backend
+    with temp_config(
+        {"module_backends": {"load_priority": [{"type": backend_types.MOCK}]}}
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_BACKEND_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
         assert isinstance(model, DummyBar)
@@ -282,9 +266,20 @@ def test_backend_supported_model_load_successfully(reset_globals):
 def test_local_model_load_successfully(reset_globals):
     """Test regular / local model can load successfully"""
     DummyFoo, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.LOCAL}):
+
+    # Configure backend
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.LOCAL}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_LOCAL_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
+
         assert isinstance(model, DummyFoo)
         assert not isinstance(model, DummyBar)
 
@@ -294,7 +289,16 @@ def test_local_model_loaded_backend_successfully(reset_globals):
     if available in SUPPORTED_LOAD_BACKENDS
     """
     _, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    # Configure backend
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
+
         DummyBar.SUPPORTED_LOAD_BACKENDS.append("LOCAL")
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_LOCAL_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
@@ -305,7 +309,16 @@ def test_backend_model_loaded_as_singleton(reset_globals):
     """Test that backend specific model can be loaded as a singleton"""
 
     _, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    # Configure backend
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_BACKEND_MODEL_NAME)
         model1 = caikit.core.load(dummy_model_path, load_singleton=True)
         model2 = caikit.core.load(dummy_model_path, load_singleton=True)
@@ -332,7 +345,16 @@ def test_singleton_cache_with_different_backend(reset_globals):
     """Test singleton cache doesn't stop different backend models"""
 
     _, _ = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    # Configure backend
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_BACKEND_MODEL_NAME)
         _ = caikit.core.load(dummy_model_path, load_singleton=True)
 
@@ -357,7 +379,16 @@ def test_load_fails_on_no_supported_backend(reset_globals):
     loading fails.
     """
     _ = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    with temp_config(
+        {
+            "merge_strategy": "override",
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+                "train_priority": [],
+            },
+        }
+    ):
+        configure()
         with temp_saved_model(NonDistributedModule()) as model_path:
             with pytest.raises(ValueError):
                 caikit.core.load(model_path)
@@ -368,7 +399,15 @@ def test_preferred_backend_enabled(reset_globals):
     loaded with an enabled non-local backend if given as a preferred_backend.
     """
     _, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_LOCAL_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
         assert isinstance(model, DummyBar)
@@ -379,13 +418,20 @@ def test_module_backend_instance_is_passed_to_load_classmethod(reset_globals):
     instance of the module backend is passed to .load via the load_backend kwarg.
     """
     _, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.MOCK}):
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK}],
+            }
+        }
+    ):
+        configure()
         with mock.patch.object(DummyBar, "load", MagicMock()) as mock_load:
             mock_load.return_value = DummyBar()
             dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_LOCAL_MODEL_NAME)
-            caikit.core.load(dummy_model_path)
+            model = caikit.core.load(dummy_model_path)
 
-            load_backends = configured_backends()
+            load_backends = module_backend_config.configured_load_backends()
             expected_load_backend = [
                 be for be in load_backends if be.backend_type == backend_types.MOCK
             ][0]
@@ -400,7 +446,15 @@ def test_preferred_backend_disabled(reset_globals):
     local even with a preferred_backend when the preferred backend is disabled.
     """
     DummyFoo, DummyBar = setup_saved_model(MockBackend)
-    with backend_priority({"type": backend_types.LOCAL}):
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.LOCAL}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_LOCAL_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
         assert isinstance(model, DummyFoo)
@@ -435,7 +489,15 @@ def test_non_local_supported_backend(reset_globals):
         def load(self, *args, **kwargs):
             return DummyBaz()
 
-    with backend_priority({"type": backend_types.MOCK2}):
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [{"type": backend_types.MOCK2}],
+            }
+        }
+    ):
+        configure()
+
         dummy_model_path = os.path.join(TEST_DATA_PATH, DUMMY_BACKEND_MODEL_NAME)
         model = caikit.core.load(dummy_model_path)
         assert isinstance(model, DummyBaz)
@@ -460,104 +522,90 @@ def test_load_must_return_model():
             caikit.core.load(tempdir)
 
 
-def test_load_with_two_shared_initializers_of_the_same_type(
-    good_model_path, reset_globals
-):
-    """Multiple instances of one shared loader type can be configured"""
+def test_load_with_new_shared_backend(good_model_path, reset_globals):
+    """If a shared laod backend has higher priority than LOCAL, it is used"""
+    loader_name = str(uuid.uuid4())
+    with temp_config(
+        {
+            "module_backends": {
+                "load_priority": [
+                    {"type": TestLoader.backend_type},
+                    {"type": backend_types.LOCAL},
+                ],
+            }
+        }
+    ):
+        configure()
+        model = caikit.core.load(good_model_path)
+        assert model.load_backend.backend_type == TestLoader.backend_type
 
+
+def test_load_with_two_shared_loaders_of_the_same_type(good_model_path, reset_globals):
+    """Multiple instances of one shared loader type can be configured"""
     with temp_config(
         {
-            "model_management": {
-                "finders": {"default": {"type": "LOCAL"}},
-                "initializers": {
-                    "model-one": {
-                        "type": TestInitializer.name,
+            "module_backends": {
+                "load_priority": [
+                    {
+                        "type": TestLoader.backend_type,
                         "config": {"model_type": "model one"},
                     },
-                    "model-two": {
-                        "type": TestInitializer.name,
+                    {
+                        "type": TestLoader.backend_type,
                         "config": {"model_type": "model two"},
                     },
-                    "default": {"type": "LOCAL"},
-                },
+                    {"type": backend_types.LOCAL},
+                ],
             }
         }
     ):
-        model_one_loader = MODEL_MANAGER._get_initializer("model-one")
-        model_two_loader = MODEL_MANAGER._get_initializer("model-two")
-
+        configure()
+        backends = configured_load_backends()
+        assert len(backends) == 3
         # plain model load should use first loader
-        model = caikit.core.load(good_model_path, initializer="model-one")
-        assert model_one_loader
-        assert model in model_one_loader.loaded_models
-        assert model not in model_two_loader.loaded_models
+        model = caikit.core.load(good_model_path)
+        assert model.load_backend is backends[0]
 
         # model load that fails in the first loader will use the second
-        model = caikit.core.load(good_model_path, initializer="model-two")
-        assert model not in model_one_loader.loaded_models
-        assert model in model_two_loader.loaded_models
+        model = caikit.core.load(good_model_path, model_type="model two")
+        assert model.load_backend is backends[1]
 
 
 def test_load_does_not_read_config_yml_if_loader_does_not_require_it(
     reset_globals, tmp_path
 ):
     tmpdir = str(tmp_path)
+
     with open(os.path.join(tmpdir, "config.yml"), "w") as f:
         f.write("{this is not yaml} !!@#$%^")
 
-    with temp_config(
-        {
-            "model_management": {
-                "finders": {
-                    "default": {"type": NoYamlFinder.name},
-                },
-                "initializers": {
-                    "default": {"type": NoYamlInitializer.name},
-                },
-            }
-        }
-    ):
-        model = caikit.core.load(tmpdir)
-        assert isinstance(model, SampleModule)
+    class NoYamlLoader(SharedLoadBackendBase):
+        backend_type = "NOYAML"
 
+        def stop(self):
+            pass
 
-def test_load_loader_finder_by_name(reset_globals):
-    """Make sure that non-default loaders and finders can be referenced by name
-    when calling load
-    """
-    with temp_config(
-        {
-            "model_management": {
-                "finders": {
-                    "default": {"type": "LOCAL"},
-                    "other": {"type": NoYamlFinder.name},
-                },
-                "initializers": {
-                    "default": {"type": "LOCAL"},
-                    "other": {"type": NoYamlInitializer.name},
-                },
-            }
-        }
-    ):
-        model = caikit.core.load("unused", finder="other", initializer="other")
-        assert isinstance(model, SampleModule)
+        def start(self):
+            pass
 
+        def register_config(self, config):
+            pass
+
+        def load(self, model_path, *args, **kwargs):
+            """This load function doesn't read from model_path, so it definitely does not read the config.yml file"""
+            return SampleModule()
+
+    backend_types.register_backend_type(NoYamlLoader)
 
-def test_load_loader_finder_by_value(reset_globals):
-    """Make sure that non-default loaders and finders can be passed by value"""
     with temp_config(
         {
-            "model_management": {
-                "finders": {
-                    "default": {"type": "LOCAL"},
-                },
-                "initializers": {
-                    "default": {"type": "LOCAL"},
-                },
+            "module_backends": {
+                "load_priority": [
+                    {"type": NoYamlLoader.backend_type},
+                ]
             }
         }
     ):
-        model = caikit.core.load(
-            "unused", finder=NoYamlFinder(), initializer=NoYamlInitializer()
-        )
+        configure()
+        model = caikit.core.load(tmpdir)
         assert isinstance(model, SampleModule)
diff --git a/tests/fixtures/config/config.yml b/tests/fixtures/config/config.yml
index a86dac27..b05eac11 100644
--- a/tests/fixtures/config/config.yml
+++ b/tests/fixtures/config/config.yml
@@ -18,15 +18,8 @@ inference_plugin:
     model_mesh:
         runtime_version: mock
 
-model_management:
-    finders:
-        default:
-            type: LOCAL
-    initializers:
-        default:
-            type: LOCAL
-            config:
-                backend_priority:
-                    # There is a 'MOCK' backend type defined in our test helpers
-                    - type: MOCK
-                    - type: LOCAL
+module_backends:
+    load_priority:
+        # There is a 'MOCK' backend type defined in our test helpers
+        - type: MOCK
+        - type: LOCAL