Model Module

This section documents the model components of the Nextmv Python SDK.

model

Model module for creating and saving decision models in Nextmv Cloud.

This module provides the base classes and functionality for creating decision models that can be deployed and run in Nextmv Cloud. The main components are:

CLASS	DESCRIPTION
Model	Base class for defining decision models.
ModelConfiguration	Configuration for packaging and deploying models.
Models defined using this module can be packaged with their dependencies and
deployed to Nextmv Cloud for execution.

Model

Base class for defining decision models that run in Nextmv Cloud.

You can import the Model class directly from nextmv:

from nextmv import Model

This class serves as a foundation for creating decision models that can be deployed to Nextmv Cloud. Subclasses must implement the solve method, which is the main entry point for processing inputs and producing decisions.

METHOD	DESCRIPTION
solve	Process input data and produce a decision output.
save	Save the model to the filesystem for deployment.

Examples:

>>> import nextroute
>>> import nextmv
>>>
>>> class DecisionModel(nextmv.Model):
...     def solve(self, input: nextmv.Input) -> nextmv.Output:
...         nextroute_input = nextroute.schema.Input.from_dict(input.data)
...         nextroute_options = nextroute.Options.extract_from_dict(input.options.to_dict())
...         nextroute_output = nextroute.solve(nextroute_input, nextroute_options)
...
...         return nextmv.Output(
...             options=input.options,
...             solution=nextroute_output.solutions[0].to_dict(),
...             statistics=nextroute_output.statistics.to_dict(),
...         )

save

save(
    model_self,
    model_dir: str,
    configuration: ModelConfiguration,
) -> None

Save the model to the local filesystem for deployment.

This method packages the model according to the provided configuration, creating all necessary files and dependencies for deployment to Nextmv Cloud.

PARAMETER	DESCRIPTION
model_dir	The directory where the model will be saved. TYPE: str
configuration	The configuration of the model, which defines how the model is saved and loaded. TYPE: ModelConfiguration

RAISES	DESCRIPTION
ImportError	If mlflow is not installed, which is required for model packaging.

Notes

This method uses mlflow for model packaging, creating the necessary files and directory structure for deployment.

Examples:

>>> model = MyDecisionModel()
>>> config = ModelConfiguration(
...     name="routing_model",
...     requirements=["pandas", "numpy"]
... )
>>> model.save("/tmp/my_model", config)

Source code in nextmv/nextmv/model.py

def save(model_self, model_dir: str, configuration: ModelConfiguration) -> None:
    """
    Save the model to the local filesystem for deployment.

    This method packages the model according to the provided configuration,
    creating all necessary files and dependencies for deployment to Nextmv
    Cloud.

    Parameters
    ----------
    model_dir : str
        The directory where the model will be saved.
    configuration : ModelConfiguration
        The configuration of the model, which defines how the model is
        saved and loaded.

    Raises
    ------
    ImportError
        If mlflow is not installed, which is required for model packaging.

    Notes
    -----
    This method uses mlflow for model packaging, creating the necessary
    files and directory structure for deployment.

    Examples
    --------
    >>> model = MyDecisionModel()
    >>> config = ModelConfiguration(
    ...     name="routing_model",
    ...     requirements=["pandas", "numpy"]
    ... )
    >>> model.save("/tmp/my_model", config)
    """

    # mlflow is a big package. We don't want to make it a dependency of
    # `nextmv` because it is not always needed. We only need it if we are
    # working with the "app from model" logic, which involves working with
    # this `Model` class.
    try:
        import mlflow as mlflow
    except ImportError as e:
        raise ImportError(
            "mlflow is not installed. Please install optional dependencies with `pip install nextmv[all]`"
        ) from e

    finally:
        from mlflow.models import infer_signature
        from mlflow.pyfunc import PythonModel, save_model

    class MLFlowModel(PythonModel):
        """
        Transient class to translate a Nextmv Decision Model into an MLflow PythonModel.

        This class complies with the MLflow inference API, implementing a `predict`
        method that calls the user-defined `solve` method of the Nextmv Decision Model.

        Methods
        -------
        predict(context, model_input, params)
            MLflow-compliant predict method that delegates to the Nextmv model's solve method.
        """

        def predict(
            self,
            context,
            model_input,
            params: Optional[dict[str, Any]] = None,
        ) -> Any:
            """
            MLflow-compliant prediction method that calls the Nextmv model's solve method.

            This method enables compatibility with MLflow's python_function model flavor.

            Parameters
            ----------
            context : mlflow.pyfunc.PythonModelContext
                The MLflow model context.
            model_input : Any
                The input data for prediction, passed to the solve method.
            params : Optional[dict[str, Any]], optional
                Additional parameters for prediction.

            Returns
            -------
            Any
                The result from the Nextmv model's solve method.

            Notes
            -----
            This method should not be used or overridden directly. Instead,
            implement the `solve` method in your Nextmv Model subclass.
            """

            return model_self.solve(model_input)

    # Some annoying logging from mlflow must be disabled.
    logging.disable(logging.CRITICAL)

    _cleanup_python_model(model_dir, configuration, verbose=False)

    signature = None
    if configuration.options is not None:
        options_dict = configuration.options.to_dict()
        signature = infer_signature(
            params=options_dict,
        )

    # We use mlflow to save the model to the local filesystem, to be able to
    # load it later on.
    model_path = os.path.join(model_dir, configuration.name)
    save_model(
        path=model_path,  # Customize the name of the model location.
        infer_code_paths=True,  # Makes the imports portable.
        python_model=MLFlowModel(),
        signature=signature,  # Allows us to work with our own `Options` class.
    )

    # Create an auxiliary requirements file with the model dependencies.
    requirements_file = os.path.join(model_dir, _REQUIREMENTS_FILE)
    with open(requirements_file, "w") as file:
        file.write(f"{_MLFLOW_DEPENDENCY}\n")
        reqs = configuration.requirements
        if reqs is not None:
            for req in reqs:
                file.write(f"{req}\n")

    # Adds the main.py file to the app_dir by coping the `entrypoint.py` file
    # which is one level up from this file.
    entrypoint_file = os.path.join(os.path.dirname(__file__), _ENTRYPOINT_FILE)
    shutil.copy2(entrypoint_file, os.path.join(model_dir, "main.py"))

solve

solve(input: Input) -> Output

Process input data and produce a decision output.

This is the main entry point of your model that you must implement in subclasses. It receives input data and should process it to produce an output containing the solution to the decision problem.

PARAMETER	DESCRIPTION
input	The input data that the model will use to make a decision. TYPE: Input

RETURNS	DESCRIPTION
Output	The output of the model, which is the solution to the decision model/problem.

RAISES	DESCRIPTION
NotImplementedError	When called on the base Model class, as this method must be implemented by subclasses.

Examples:

>>> def solve(self, input: Input) -> Output:
...     # Process input data
...     result = self._process_data(input.data)
...
...     # Return formatted output
...     return Output(
...         options=input.options,
...         solution=result,
...         statistics={"processing_time": 0.5}
...     )

Source code in nextmv/nextmv/model.py

def solve(self, input: Input) -> Output:
    """
    Process input data and produce a decision output.

    This is the main entry point of your model that you must implement in
    subclasses. It receives input data and should process it to produce an
    output containing the solution to the decision problem.

    Parameters
    ----------
    input : Input
        The input data that the model will use to make a decision.

    Returns
    -------
    Output
        The output of the model, which is the solution to the decision
        model/problem.

    Raises
    ------
    NotImplementedError
        When called on the base Model class, as this method must be
        implemented by subclasses.

    Examples
    --------
    >>> def solve(self, input: Input) -> Output:
    ...     # Process input data
    ...     result = self._process_data(input.data)
    ...
    ...     # Return formatted output
    ...     return Output(
    ...         options=input.options,
    ...         solution=result,
    ...         statistics={"processing_time": 0.5}
    ...     )
    """

    raise NotImplementedError

ModelConfiguration dataclass

ModelConfiguration(
    name: str,
    requirements: Optional[list[str]] = None,
    options: Optional[Options] = None,
)

Configuration class for Nextmv models.

You can import the ModelConfiguration class directly from nextmv:

from nextmv import ModelConfiguration

This class holds the configuration for a model, defining how a Python model is encoded and loaded for use in Nextmv Cloud.

PARAMETER	DESCRIPTION
name	A personalized name for the model. This is required. TYPE: str
requirements	A list of Python dependencies that the decision model requires, formatted as they would appear in a requirements.txt file. TYPE: list[str] DEFAULT: None
options	Options that the decision model requires. TYPE: Options DEFAULT: None

Examples:

>>> from nextmv import ModelConfiguration, Options
>>> config = ModelConfiguration(
...     name="my_routing_model",
...     requirements=["nextroute>=1.0.0"],
...     options=Options({"max_time": 60})
... )

name instance-attribute

name: str

The name of the decision model.

options class-attribute instance-attribute

options: Optional[Options] = None

Options that the decision model requires.

requirements class-attribute instance-attribute

requirements: Optional[list[str]] = None

A list of Python dependencies that the decision model requires.