Skip to content

Keras API Reference 📖

Complete API reference for MLPotion's Keras components.

Auto-Generated Documentation

This page is automatically populated with API documentation from the source code.

Extensibility

These components are built using protocol-based design, making MLPotion easy to extend. Want to add new data sources, training methods, or integrations? See Contributing Guide.

Data Loading

mlpotion.frameworks.keras.data.loaders

Keras data loaders.

Classes

CSVDataLoader dataclass

Bases: DataLoader[CSVSequence]

Loader for CSV files into a Keras-ready Sequence.

This class provides a high-level interface for loading data from CSV files into a CSVSequence compatible with Keras models. It handles file globbing, reading, column selection, and label separation.

Attributes:

Name Type Description
file_pattern str

Glob pattern matching the CSV files to load (e.g., "data/*.csv").
batch_size int

Number of samples per batch in the resulting sequence.
column_names list[str] | None

List of column names to use as features. If None, all columns except the label are used.
label_name str | None

Name of the column to use as labels. If None, no labels are returned (inference mode).
shuffle bool

Whether to shuffle the data between epochs.
dtype np.dtype | str

Data type to use for the loaded data (default: "float32").
Example 
import keras
from mlpotion.frameworks.keras.data.loaders import CSVDataLoader

# Define a simple model
model = keras.Sequential([
    keras.layers.Dense(10, input_shape=(5,), activation='relu'),
    keras.layers.Dense(1, activation='sigmoid')
])
model.compile(optimizer='adam', loss='binary_crossentropy')

# Create loader
loader = CSVDataLoader(
    file_pattern="data/train_*.csv",
    label_name="target_class",
    batch_size=64,
    shuffle=True
)

# Load data
train_sequence = loader.load()

# Train model
model.fit(train_sequence, epochs=10)
Functions
load 
load() -> CSVSequence

Load CSV files and return a CSVSequence.

Returns:

Type Description
CSVSequence

CSVSequence that can be passed directly to model.fit(...).

Raises:

Type Description
DataLoadingError

If files cannot be found or read.
Source code in mlpotion/frameworks/keras/data/loaders.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@trycatch(
    error=DataLoadingError,
    success_msg="✅ Successfully created Keras CSV Sequence",
)
def load(self) -> CSVSequence:
    """Load CSV files and return a CSVSequence.

    Returns:
        CSVSequence that can be passed directly to `model.fit(...)`.

    Raises:
        DataLoadingError: If files cannot be found or read.
    """
    files = self._get_files()
    df = self._load_dataframe(files)
    df = self._select_columns(df)

    features_np, labels_np = self._split_features_labels(df)

    logger.info(
        "Creating CSVSequence: n_samples={n}, n_features={d}, labels={labels}",
        n=features_np.shape[0],
        d=features_np.shape[1],
        labels="yes" if labels_np is not None else "no",
    )

    sequence = CSVSequence(
        features=features_np,
        labels=labels_np,
        batch_size=self.batch_size,
        shuffle=self.shuffle,
        dtype=self.dtype,
    )
    return sequence

CSVSequence 

CSVSequence(
    features: np.ndarray,
    labels: np.ndarray | None,
    batch_size: int = 32,
    shuffle: bool = True,
    dtype: np.dtype | str = "float32",
) -> None

Bases: Sequence

Keras Sequence for CSV data backed by NumPy arrays.

This class implements the keras.utils.Sequence interface, allowing it to be used directly with model.fit(), model.evaluate(), and model.predict(). It handles data batching and shuffling efficiently in memory.

Attributes:

Name Type Description
features np.ndarray

The feature data as a 2D NumPy array.
labels np.ndarray | None

The label data as a NumPy array, or None if not available.
batch_size int

The size of each data batch.
shuffle bool

Whether to shuffle the data at the end of each epoch.
dtype np.dtype | str

The data type of the features and labels.
Example 
from mlpotion.frameworks.keras.data.loaders import CSVSequence
import numpy as np

# Create dummy data
X = np.random.rand(100, 10)
y = np.random.randint(0, 2, 100)

# Create sequence
sequence = CSVSequence(
    features=X,
    labels=y,
    batch_size=32,
    shuffle=True
)

# Use in training
model.fit(sequence, epochs=5)
Source code in mlpotion/frameworks/keras/data/loaders.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    features: np.ndarray,
    labels: np.ndarray | None,
    batch_size: int = 32,
    shuffle: bool = True,
    dtype: np.dtype | str = "float32",
) -> None:
    if features.ndim != 2:
        raise ValueError(
            f"features must be 2D (n_samples, n_features), got shape {features.shape}"
        )

    if labels is not None and len(labels) != len(features):
        raise ValueError(
            f"features and labels must have same length, "
            f"got {len(features)} != {len(labels)}"
        )

    self._features = features.astype(dtype, copy=False)
    self._labels = labels.astype(dtype, copy=False) if labels is not None else None
    self._batch_size = int(batch_size)
    self._shuffle = bool(shuffle)
    self._indices = np.arange(len(self._features))

    if self._shuffle:
        np.random.shuffle(self._indices)

    logger.info(
        "Initialized CSVSequence: "
        f"n_samples={len(self._features)}, "
        f"batch_size={self._batch_size}, "
        f"shuffle={self._shuffle}, "
        f"labels={'yes' if self._labels is not None else 'no'}"
    )
Functions
__getitem__ 
__getitem__(idx: int) -> Any

Get batch by index.

Returns:

Type Description
Any
  • (x_batch, y_batch) if labels are available
Any
  • x_batch otherwise
Source code in mlpotion/frameworks/keras/data/loaders.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __getitem__(self, idx: int) -> Any:
    """Get batch by index.

    Returns:
        - (x_batch, y_batch) if labels are available
        - x_batch otherwise
    """
    if idx < 0 or idx >= len(self):
        raise IndexError(f"Batch index out of range: {idx}")

    start = idx * self._batch_size
    end = min(start + self._batch_size, len(self._features))
    batch_idx = self._indices[start:end]

    x_batch = self._features[batch_idx]
    if self._labels is not None:
        y_batch = self._labels[batch_idx]
        return x_batch, y_batch

    return x_batch
__len__ 
__len__() -> int

Number of batches per epoch.

Source code in mlpotion/frameworks/keras/data/loaders.py 
89
90
91
92
def __len__(self) -> int:
    """Number of batches per epoch."""
    n_samples = len(self._features)
    return int(np.ceil(n_samples / self._batch_size))
on_epoch_end 
on_epoch_end() -> None

Shuffle indices between epochs if enabled.

Source code in mlpotion/frameworks/keras/data/loaders.py 
115
116
117
118
def on_epoch_end(self) -> None:
    """Shuffle indices between epochs if enabled."""
    if self._shuffle:
        np.random.shuffle(self._indices)

Training

mlpotion.frameworks.keras.training.trainers

Classes

ModelTrainer dataclass

Bases: ModelTrainerProtocol[Model, Sequence]

Generic trainer for Keras 3 models.

This class implements the ModelTrainerProtocol for Keras models, providing a standardized interface for training. It wraps the standard model.fit() method but adds flexibility and consistency checks.

It supports: - Automatic model compilation if compile_params are provided. - Handling of various data formats (tuples, dicts, generators). - Standardized return format (dictionary of history metrics).

Example 
import keras
import numpy as np
from mlpotion.frameworks.keras import ModelTrainer

# Prepare data
X_train = np.random.rand(100, 10)
y_train = np.random.randint(0, 2, 100)

# Define model
model = keras.Sequential([
    keras.layers.Dense(1, activation='sigmoid')
])

# Initialize trainer
trainer = ModelTrainer()

# Train
history = trainer.train(
    model=model,
    data=(X_train, y_train),
    compile_params={
        "optimizer": "adam",
        "loss": "binary_crossentropy",
        "metrics": ["accuracy"]
    },
    fit_params={
        "epochs": 5,
        "batch_size": 32,
        "verbose": 1
    }
)

print(history['loss'])
Functions
train 
train(
    model: Model,
    dataset: Any,
    config: ModelTrainingConfig,
    validation_dataset: Any | None = None,
) -> TrainingResult[Model]

Train a Keras model using the provided dataset and configuration.

Parameters:

Name Type Description Default
model Model

The Keras model to train.

 required
dataset Any

The training data. Can be a tuple (x, y), a dictionary, a Sequence, or a generator.

 required
config ModelTrainingConfig

Configuration object containing training parameters.

 required
validation_dataset Any | None

Optional validation data.

 None

Returns:

Type Description
TrainingResult[Model]

TrainingResult[Model]: An object containing the trained model, training history, and metrics.
Source code in mlpotion/frameworks/keras/training/trainers.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@trycatch(
    error=ModelTrainerError,
    success_msg="✅ Successfully trained Keras model",
)
def train(
    self,
    model: Model,
    dataset: Any,
    config: ModelTrainingConfig,
    validation_dataset: Any | None = None,
) -> TrainingResult[Model]:
    """Train a Keras model using the provided dataset and configuration.

    Args:
        model: The Keras model to train.
        dataset: The training data. Can be a tuple `(x, y)`, a dictionary, a `Sequence`, or a generator.
        config: Configuration object containing training parameters.
        validation_dataset: Optional validation data.

    Returns:
        TrainingResult[Model]: An object containing the trained model, training history, and metrics.
    """
    self._validate_model(model)

    # Prepare compile parameters from config
    compile_params = {
        "optimizer": self._get_optimizer(config),
        "loss": config.loss,
        "metrics": config.metrics,
    }

    # Compile if needed or if forced by config (though we usually respect existing compilation)
    # Here we'll ensure it's compiled. If the user wants to use their own compilation,
    # they should probably compile it before passing it, but our config implies we control it.
    # However, to be safe and flexible:
    if not self._is_compiled(model):
        if not config.optimizer or not config.loss:
            raise RuntimeError(
                "Model is not compiled and config does not provide optimizer and loss. "
                "Either compile the model beforehand or provide optimizer and loss in config."
            )
        logger.info("Compiling model with config parameters.")
        model.compile(**compile_params)
    else:
        logger.info("Model already compiled. Using existing compilation settings.")

    # Prepare fit parameters
    fit_kwargs = {
        "epochs": config.epochs,
        "batch_size": config.batch_size,
        "verbose": config.verbose,
        "shuffle": config.shuffle,
        "validation_split": config.validation_split,
        "callbacks": self._prepare_callbacks(config),
    }

    if validation_dataset is not None:
        fit_kwargs["validation_data"] = validation_dataset

    # Add any framework-specific options
    fit_kwargs.update(config.framework_options)

    logger.info("Starting Keras model training...")
    logger.debug(f"Training data type: {type(dataset)!r}")
    logger.debug(f"Fit parameters: {fit_kwargs}")

    import time

    start_time = time.time()

    history_obj = self._call_fit(model=model, data=dataset, fit_kwargs=fit_kwargs)

    training_time = time.time() - start_time

    # Convert History object to dict[str, list[float]]
    history_dict = self._history_to_dict(history_obj)

    # Extract final metrics
    final_metrics = {}
    for k, v in history_dict.items():
        if v:
            final_metrics[k] = v[-1]

    logger.info("Training completed.")
    logger.debug(f"Training history: {history_dict}")

    return TrainingResult(
        model=model,
        history=history_dict,
        metrics=final_metrics,
        config=config,
        training_time=training_time,
        best_epoch=None,  # Keras history doesn't explicitly track "best" unless using callbacks
    )

Evaluation

mlpotion.frameworks.keras.evaluation.evaluators

Classes

ModelEvaluator dataclass

Bases: ModelEvaluatorProtocol[Model, Sequence]

Generic evaluator for Keras 3 models.

This class implements the ModelEvaluatorProtocol for Keras models. It wraps the model.evaluate() method to provide a consistent evaluation interface.

It ensures that the evaluation result is always returned as a dictionary of metric names to values, regardless of how the model was compiled or what arguments were passed.

Example 
import keras
import numpy as np
from mlpotion.frameworks.keras import ModelEvaluator

# Prepare data
X_test = np.random.rand(20, 10)
y_test = np.random.randint(0, 2, 20)

# Define model
model = keras.Sequential([
    keras.layers.Dense(1, activation='sigmoid')
])

# Initialize evaluator
evaluator = ModelEvaluator()

# Evaluate
metrics = evaluator.evaluate(
    model=model,
    data=(X_test, y_test),
    compile_params={
        "optimizer": "adam",
        "loss": "binary_crossentropy",
        "metrics": ["accuracy"]
    },
    eval_params={"batch_size": 32}
)

print(metrics)  # {'loss': 0.693..., 'accuracy': 0.5...}
Functions
evaluate 
evaluate(
    model: Model,
    dataset: Any,
    config: ModelEvaluationConfig,
) -> EvaluationResult

Evaluate a Keras model on the given data.

Parameters:

Name Type Description Default
model Model

The Keras model to evaluate.

 required
dataset Any

The evaluation data. Can be a tuple (x, y), a dictionary, or a Sequence.

 required
config ModelEvaluationConfig

Configuration object containing evaluation parameters.

 required

Returns:

Name Type Description
EvaluationResult EvaluationResult

An object containing the evaluation metrics.
Source code in mlpotion/frameworks/keras/evaluation/evaluators.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@trycatch(
    error=ModelEvaluatorError,
    success_msg="✅ Successfully evaluated Keras model",
)
def evaluate(
    self,
    model: Model,
    dataset: Any,
    config: ModelEvaluationConfig,
) -> EvaluationResult:
    """Evaluate a Keras model on the given data.

    Args:
        model: The Keras model to evaluate.
        dataset: The evaluation data. Can be a tuple `(x, y)`, a dictionary, or a `Sequence`.
        config: Configuration object containing evaluation parameters.

    Returns:
        EvaluationResult: An object containing the evaluation metrics.
    """
    self._validate_model(model)

    # Prepare eval parameters
    eval_kwargs = {
        "batch_size": config.batch_size,
        "verbose": config.verbose,
        "return_dict": True,
    }

    # Add any framework-specific options
    eval_kwargs.update(config.framework_options)

    # We assume the model is already compiled. If not, Keras will raise an error
    # unless we provide compile params, but EvaluationConfig doesn't typically carry them.
    # The user should ensure the model is compiled (e.g. after loading or training).
    if not self._is_compiled(model):
        logger.warning(
            "Model is not compiled. Evaluation might fail if loss/metrics are not defined."
        )

    logger.info("Evaluating Keras model...")
    logger.debug(f"Evaluation data type: {type(dataset)!r}")
    logger.debug(f"Evaluation parameters: {eval_kwargs}")

    import time

    start_time = time.time()

    result = self._call_evaluate(model=model, data=dataset, eval_kwargs=eval_kwargs)

    evaluation_time = time.time() - start_time

    # At this point, result should be a dict[str, float]
    if not isinstance(result, dict):
        # Defensive fallback if user or Keras changed behavior
        logger.warning(
            f"`model.evaluate` did not return a dict (got {type(result)!r}). "
            "Wrapping into a dict under key 'metric_0'."
        )
        result = {"metric_0": float(result)}

    metrics = {str(k): float(v) for k, v in result.items()}
    logger.info(f"Evaluation result: {metrics}")

    return EvaluationResult(
        metrics=metrics,
        config=config,
        evaluation_time=evaluation_time,
    )

Persistence

mlpotion.frameworks.keras.deployment.persistence

Classes

ModelPersistence 

ModelPersistence(
    path: str | Path, model: Model | None = None
) -> None

Bases: ModelPersistenceProtocol[Model]

Persistence helper for Keras models.

This class manages saving and loading of Keras models. It supports standard Keras formats (.keras, .h5) and SavedModel directories. It also integrates with ModelInspector to provide model metadata upon loading.

Attributes:

Name Type Description
path Path

The file path for the model artifact.
model Model | None

The Keras model instance (optional).
Example 
import keras
from mlpotion.frameworks.keras import ModelPersistence

# Define model
model = keras.Sequential([keras.layers.Dense(1)])

# Save
saver = ModelPersistence(path="models/my_model.keras", model=model)
saver.save()

# Load
loader = ModelPersistence(path="models/my_model.keras")
loaded_model, metadata = loader.load(inspect=True)
print(metadata['parameters'])
Source code in mlpotion/frameworks/keras/deployment/persistence.py 
44
45
46
def __init__(self, path: str | Path, model: Model | None = None) -> None:
    self._path = Path(path)
    self._model = model
Attributes
model property writable 
model: Model | None

Currently attached Keras model (may be None before loading).

path property writable 
path: Path

Filesystem path where the model is saved/loaded.

Functions
load 
load(
    *, inspect: bool = True, **kwargs: Any
) -> tuple[Model, dict[str, Any] | None]

Load a Keras model from disk.

Parameters:

Name Type Description Default
inspect bool

Whether to inspect the loaded model and return metadata.

 True
**kwargs Any

Additional arguments passed to keras.models.load_model().

 {}

Returns:

Type Description
Model

tuple[Model, dict[str, Any] | None]: A tuple containing the loaded model and
dict[str, Any] | None

optional inspection metadata.

Raises:

Type Description
ModelPersistenceError

If the model file cannot be found or loaded.
Source code in mlpotion/frameworks/keras/deployment/persistence.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@trycatch(
    error=ModelPersistenceError,
    success_msg="✅ Successfully loaded Keras model",
)
def load(
    self,
    *,
    inspect: bool = True,
    **kwargs: Any,
) -> tuple[Model, dict[str, Any] | None]:
    """Load a Keras model from disk.

    Args:
        inspect: Whether to inspect the loaded model and return metadata.
        **kwargs: Additional arguments passed to `keras.models.load_model()`.

    Returns:
        tuple[Model, dict[str, Any] | None]: A tuple containing the loaded model and
        optional inspection metadata.

    Raises:
        ModelPersistenceError: If the model file cannot be found or loaded.
    """
    path = self._ensure_path_exists()

    logger.info(f"Loading Keras model from: {path!s}")
    model = keras.models.load_model(path.as_posix(), **kwargs)

    self._model = model  # keep instance in sync

    inspection_result: dict[str, Any] | None = None
    if inspect:
        logger.info("Inspecting loaded Keras model with ModelInspector.")
        inspector = ModelInspector()
        inspection_result = inspector.inspect(model)

    return model, inspection_result
save 
save(overwrite: bool = True, **kwargs: Any) -> None

Save the attached model to disk.

Parameters:

Name Type Description Default
overwrite bool

Whether to overwrite the file if it already exists.

 True
**kwargs Any

Additional arguments passed to model.save().

 {}

Raises:

Type Description
ModelPersistenceError

If no model is attached or if the file exists and overwrite is False.
Source code in mlpotion/frameworks/keras/deployment/persistence.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@trycatch(
    error=ModelPersistenceError,
    success_msg="✅ Successfully saved Keras model",
)
def save(
    self,
    overwrite: bool = True,
    **kwargs: Any,
) -> None:
    """Save the attached model to disk.

    Args:
        overwrite: Whether to overwrite the file if it already exists.
        **kwargs: Additional arguments passed to `model.save()`.

    Raises:
        ModelPersistenceError: If no model is attached or if the file exists and `overwrite` is False.
    """
    model = self._ensure_model()
    target = self._path

    if target.exists() and not overwrite:
        raise ModelPersistenceError(
            f"Target path already exists and overwrite=False: {target!s}"
        )

    logger.info(f"Saving Keras model to: {target!s}")
    target.parent.mkdir(parents=True, exist_ok=True)

    # Keras 3 generally infers format from the path; `save_format` is
    # deprecated / discouraged in newer APIs, so we do NOT pass it.
    model.save(target.as_posix(), **kwargs)
    logger.info("Keras model saved successfully.")

Export

mlpotion.frameworks.keras.deployment.exporters

Classes

ModelExporter

Bases: ModelExporterProtocol[Model]

Generic exporter for Keras 3 models.

This class implements ModelExporterProtocol and supports exporting Keras models to various formats, including native Keras formats (.keras, .h5) and inference formats like TensorFlow SavedModel or ONNX (via model.export).

It also supports creating export archives with custom endpoints using keras.export.ExportArchive.

Example 
import keras
from mlpotion.frameworks.keras import ModelExporter

model = keras.Sequential([keras.layers.Dense(1)])
exporter = ModelExporter()

# Export as standard Keras file
exporter.export(model, "models/model.keras")

# Export for serving (TF SavedModel)
exporter.export(model, "models/serving", export_format="tf_saved_model")
Functions
export 
export(model: Model, path: str, **kwargs: Any) -> None

Export a Keras model to disk.

Parameters:

Name Type Description Default
model Model

The Keras model to export.

 required
path str

The destination path or directory.

 required
**kwargs Any

Additional export options: - export_format (str): "keras", "h5", "tf_saved_model", "onnx", etc. - dataset (Iterable): Optional data for model warmup. - endpoint_name (str): Name for custom endpoint (uses ExportArchive). - input_specs (list[InputSpec]): Input signatures for custom endpoint. - config (dict): Extra arguments for the underlying save/export method.

 {}

Raises:

Type Description
ModelExporterError

If export fails.
Source code in mlpotion/frameworks/keras/deployment/exporters.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@trycatch(
    error=ModelExporterError,
    success_msg="✅ Successfully Exported model",
)
def export(self, model: Model, path: str, **kwargs: Any) -> None:
    """Export a Keras model to disk.

    Args:
        model: The Keras model to export.
        path: The destination path or directory.
        **kwargs: Additional export options:
            - `export_format` (str): "keras", "h5", "tf_saved_model", "onnx", etc.
            - `dataset` (Iterable): Optional data for model warmup.
            - `endpoint_name` (str): Name for custom endpoint (uses ExportArchive).
            - `input_specs` (list[InputSpec]): Input signatures for custom endpoint.
            - `config` (dict): Extra arguments for the underlying save/export method.

    Raises:
        ModelExporterError: If export fails.
    """
    export_path = Path(path)

    export_format: str | None = kwargs.pop("export_format", None)
    dataset: Iterable[Any] | None = kwargs.pop("dataset", None)
    endpoint_name: str | None = kwargs.pop("endpoint_name", None)
    input_specs: Sequence[InputSpec] | None = kwargs.pop("input_specs", None)
    config: Mapping[str, Any] | None = kwargs.pop("config", None)

    if kwargs:
        logger.warning(
            "Unused export kwargs passed to ModelExporter: "
            f"{list(kwargs.keys())}"
        )

    self._validate_model(model)
    self._validate_config(config)

    # Determine mode if export_format isn't explicitly set
    if export_format is None:
        export_format = self._infer_export_format_from_path(export_path)

    logger.info(
        f"Exporting Keras model '{model.name}' to {export_path!s} "
        f"with format '{export_format}'"
    )

    # Optional warm-up pass
    self._warmup_if_needed(model=model, dataset=dataset)

    # Choose strategy
    try:
        if self._is_native_keras_format(export_format):
            self._save_native_keras(model=model, path=export_path, config=config)
        elif endpoint_name is not None or input_specs is not None:
            self._export_with_export_archive(
                model=model,
                path=export_path,
                endpoint_name=endpoint_name or self.default_endpoint_name,
                input_specs=input_specs,
                export_format=export_format,
            )
        else:
            self._export_with_model_export(
                model=model,
                path=export_path,
                export_format=export_format,
                config=config,
            )
    except ValueError as err:
        logger.warning(
            f"Export error: {err} "
            "(you may need to build the model by calling it on example data "
            "before exporting)"
        )

    logger.info(f"Model export completed: {export_path!s}")

Model Inspection

mlpotion.frameworks.keras.models.inspection

Classes

ModelInspector dataclass

Bases: ModelInspectorProtocol[ModelLike]

Inspector for Keras models.

This class analyzes Keras models to extract metadata such as input/output shapes, parameter counts, layer details, and signatures. It is useful for validating models before training or deployment, and for generating model reports.

Attributes:

Name Type Description
include_layers bool

Whether to include detailed information about each layer.
include_signatures bool

Whether to include model signatures (if available).
Example 
import keras
from mlpotion.frameworks.keras import ModelInspector

model = keras.Sequential([keras.layers.Dense(1, input_shape=(10,))])
inspector = ModelInspector()

info = inspector.inspect(model)
print(f"Total params: {info['parameters']['total']}")
print(f"Inputs: {info['inputs']}")
Functions
inspect 
inspect(model: ModelLike) -> dict[str, Any]

Inspect a Keras model and return structured metadata.

Parameters:

Name Type Description Default
model ModelLike

The Keras model to inspect.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: A dictionary containing model metadata: - name: Model name. - backend: Keras backend used. - trainable: Whether the model is trainable. - inputs: List of input specifications. - outputs: List of output specifications. - parameters: Dictionary of parameter counts. - layers: List of layer details (if include_layers=True). - signatures: Model signatures (if include_signatures=True).
Source code in mlpotion/frameworks/keras/models/inspection.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@trycatch(
    error=ModelInspectorError,
    success_msg="✅ Successfully inspected Keras model",
)
def inspect(self, model: ModelLike) -> dict[str, Any]:
    """Inspect a Keras model and return structured metadata.

    Args:
        model: The Keras model to inspect.

    Returns:
        dict[str, Any]: A dictionary containing model metadata:
            - `name`: Model name.
            - `backend`: Keras backend used.
            - `trainable`: Whether the model is trainable.
            - `inputs`: List of input specifications.
            - `outputs`: List of output specifications.
            - `parameters`: Dictionary of parameter counts.
            - `layers`: List of layer details (if `include_layers=True`).
            - `signatures`: Model signatures (if `include_signatures=True`).
    """
    if not isinstance(model, keras.Model):
        raise TypeError(
            f"ModelInspector expects a keras.Model, got {type(model)!r}"
        )

    logger.info("Inspecting Keras model...")

    backend_name = self._get_backend_name()

    info: dict[str, Any] = {
        "name": model.name,
        "backend": backend_name,
        "trainable": model.trainable,
    }

    info["inputs"] = self._get_inputs(model)
    info["input_names"] = [input["name"] for input in info["inputs"]]
    info["outputs"] = self._get_outputs(model)
    info["output_names"] = [output["name"] for output in info["outputs"]]
    info["parameters"] = self._get_param_counts(model)

    if self.include_signatures:
        info["signatures"] = self._get_signatures(model)

    if self.include_layers:
        info["layers"] = self._get_layers_summary(model)

    logger.debug(f"Keras model inspection result: {info}")
    return info

See the Keras Guide for usage examples