from keras.src import backend from keras.src import tree from keras.src import utils from keras.src.api_export import keras_export from keras.src.layers import Input from keras.src.layers import InputLayer from keras.src.models.functional import Functional from keras.src.models.functional import functional_like_constructor from keras.src.models.sequential import Sequential from keras.src.saving import serialization_lib @keras_export("keras.models.clone_model") def clone_model( model, input_tensors=None, clone_function=None, call_function=None, recursive=False, **kwargs, ): """Clone a Functional or Sequential `Model` instance. Model cloning is similar to calling a model on new inputs, except that it creates new layers (and thus new weights) instead of sharing the weights of the existing layers. Note that `clone_model` will not preserve the uniqueness of shared objects within the model (e.g. a single variable attached to two distinct layers will be restored as two separate variables). Args: model: Instance of `Model` (could be a Functional model or a Sequential model). input_tensors: optional list of input tensors or InputLayer objects to build the model upon. If not provided, new `Input` objects will be created. clone_function: Callable with signature `fn(layer)` to be used to clone each layer in the target model (except `Input` instances). It takes as argument the layer instance to be cloned, and returns the corresponding layer instance to be used in the model copy. If unspecified, this callable defaults to the following serialization/deserialization function: `lambda layer: layer.__class__.from_config(layer.get_config())`. By passing a custom callable, you can customize your copy of the model, e.g. by wrapping certain layers of interest (you might want to replace all `LSTM` instances with equivalent `Bidirectional(LSTM(...))` instances, for example). Defaults to `None`. call_function: Callable with signature `fn(layer, *args, **kwargs)` to be used to call each cloned layer and a set of inputs. It takes the layer instance, the call arguments and keyword arguments, and returns the call outputs. If unspecified, this callable defaults to the regular `__call__()` method: `def fn(layer, *args, **kwargs): return layer(*args, **kwargs)`. By passing a custom callable, you can insert new layers before or after a given layer. Note: this argument can only be used with Functional models. recursive: Boolean. Whether to recursively clone any Sequential or Functional models encountered in the original Sequential/Functional model. If `False`, then inner models are cloned by calling `clone_function()`. If `True`, then inner models are cloned by calling `clone_model()` with the same `clone_function`, `call_function`, and `recursive` arguments. Note that in this case, `call_function` will not be propagated to any Sequential model (since it is not applicable to Sequential models). Returns: An instance of `Model` reproducing the behavior of the original model, on top of new inputs tensors, using newly instantiated weights. The cloned model may behave differently from the original model if a custom `clone_function` or `call_function` modifies a layer or layer call. Example: ```python # Create a test Sequential model. model = keras.Sequential([ keras.layers.Input(shape=(728,)), keras.layers.Dense(32, activation='relu'), keras.layers.Dense(1, activation='sigmoid'), ]) # Create a copy of the test model (with freshly initialized weights). new_model = clone_model(model) ``` Using a `clone_function` to make a model deterministic by setting the random seed everywhere: ```python def clone_function(layer): config = layer.get_config() if "seed" in config: config["seed"] = 1337 return layer.__class__.from_config(config) new_model = clone_model(model, clone_function=clone_function) ``` Using a `call_function` to add a `Dropout` layer after each `Dense` layer (without recreating new layers): ```python def call_function(layer, *args, **kwargs): out = layer(*args, **kwargs) if isinstance(layer, keras.layers.Dense): out = keras.layers.Dropout(0.5)(out) return out new_model = clone_model( model, clone_function=lambda x: x, # Reuse the same layers. call_function=call_function, ) ``` Note that subclassed models cannot be cloned by default, since their internal layer structure is not known. To achieve equivalent functionality as `clone_model` in the case of a subclassed model, simply make sure that the model class implements `get_config()` (and optionally `from_config()`), and call: ```python new_model = model.__class__.from_config(model.get_config()) ``` In the case of a subclassed model, you cannot using a custom `clone_function`. """ cache = kwargs.pop("cache", None) if kwargs: raise ValueError( f"Unexpected keyword argument(s): {tuple(kwargs.keys())}" ) if isinstance(model, Sequential): # Wrap clone_function to handle recursiveness and layer sharing. clone_function = _wrap_clone_function( clone_function, call_function=call_function, recursive=recursive, cache=cache, ) if call_function is not None: raise ValueError( "`call_function` argument is not supported with Sequential " "models. In a Sequential model, layers aren't called " "at model-construction time (they're merely listed). " "Use `call_function` with Functional models only. " "Received model of " f"type '{model.__class__.__name__}', with " f"call_function={clone_function}" ) return _clone_sequential_model( model, clone_function=clone_function, input_tensors=input_tensors, ) if isinstance(model, Functional): # Wrap clone_function to handle recursiveness and layer sharing. clone_function = _wrap_clone_function( clone_function, call_function=call_function, recursive=recursive, cache=cache, ) # If the get_config() method is the same as a regular Functional # model, we're safe to use _clone_functional_model (which relies # on a Functional constructor). In the case where the get_config # is custom, this may not necessarily work, but if clone_function # or input_tensors are passed, we attempt it anyway # in order to preserve backwards compatibility. if utils.is_default(model.get_config) or ( clone_function or input_tensors ): return _clone_functional_model( model, clone_function=clone_function, call_function=call_function, input_tensors=input_tensors, ) # Case of a custom model class if clone_function or input_tensors: raise ValueError( "Arguments `clone_function` and `input_tensors` " "are only supported for Sequential models " "or Functional models. Received model of " f"type '{model.__class__.__name__}', with " f"clone_function={clone_function} and " f"input_tensors={input_tensors}" ) if call_function is not None: raise ValueError( "Argument `call_function` is only supported " "for Functional models. Received model of " f"type '{model.__class__.__name__}', with " f"call_function={clone_function}" ) config = serialization_lib.serialize_keras_object(model) return serialization_lib.deserialize_keras_object( config, custom_objects={model.__class__.__name__: model.__class__} ) def _wrap_clone_function( clone_function, call_function=None, recursive=False, cache=None ): """Wrapper to handle recursiveness and layer sharing.""" if clone_function is None: def _clone_layer(layer): return layer.__class__.from_config(layer.get_config()) clone_function = _clone_layer if cache is None: cache = {} def wrapped_clone_function(layer): if id(layer) in cache: return cache[id(layer)] if recursive: if isinstance(layer, Sequential): # Note: Sequential doesn't support call_function. clone = clone_model( layer, clone_function=clone_function, cache=cache, ) cache[id(layer)] = clone return clone elif isinstance(layer, Functional): clone = clone_model( layer, clone_function=clone_function, call_function=call_function, cache=cache, ) cache[id(layer)] = clone return clone clone = clone_function(layer) cache[id(layer)] = clone return clone return wrapped_clone_function def _clone_sequential_model(model, clone_function, input_tensors=None): """Clone a `Sequential` model instance. Model cloning is similar to calling a model on new inputs, except that it creates new layers (and thus new weights) instead of sharing the weights of the existing layers. Args: model: Instance of `Sequential`. input_tensors: optional list of input tensors to build the model upon. If not provided, placeholders will be created. clone_function: callable to be applied on non-input layers in the model. By default, it clones the layer (without copying the weights). Returns: An instance of `Sequential` reproducing the behavior of the original model, on top of new inputs tensors, using newly instantiated weights. """ if not isinstance(model, Sequential): raise ValueError( "Expected `model` argument " "to be a `Sequential` model instance. " f"Received: model={model}" ) if not callable(clone_function): raise ValueError( "Expected `clone_function` argument to be a callable. " f"Received: clone_function={clone_function}" ) new_layers = [clone_function(layer) for layer in model.layers] if isinstance(model._layers[0], InputLayer): ref_input_layer = model._layers[0] input_name = ref_input_layer.name input_batch_shape = ref_input_layer.batch_shape input_dtype = ref_input_layer._dtype else: input_name = None input_dtype = None input_batch_shape = None if input_tensors is not None: if isinstance(input_tensors, (list, tuple)): if len(input_tensors) != 1: raise ValueError( "Argument `input_tensors` must contain a single tensor." ) input_tensors = input_tensors[0] if not isinstance(input_tensors, backend.KerasTensor): raise ValueError( "Argument `input_tensors` must be a KerasTensor. " f"Received invalid value: input_tensors={input_tensors}" ) inputs = Input( tensor=input_tensors, name=input_name, ) new_layers = [inputs] + new_layers else: if input_batch_shape is not None: inputs = Input( batch_shape=input_batch_shape, dtype=input_dtype, name=input_name, ) new_layers = [inputs] + new_layers cloned_model = Sequential( new_layers, name=model.name, trainable=model.trainable ) # If model compiled already then set same to cloned model if model.compiled: compiled_config = model.get_compile_config() cloned_model.compile_from_config(compiled_config) return cloned_model def _clone_functional_model( model, clone_function, input_tensors=None, call_function=None ): """Clone a `Functional` model instance. Model cloning is similar to calling a model on new inputs, except that it creates new layers (and thus new weights) instead of sharing the weights of the existing layers. Input layers are always cloned. Args: model: Instance of `Functional`. input_tensors: optional list of input tensors to build the model upon. If not provided, placeholders will be created. clone_function: callable to be applied on non-input layers in the model. By default, it clones the layer (without copying the weights). Returns: An instance of `Functional` reproducing the behavior of the original model, on top of new inputs tensors, using newly instantiated weights. """ if not callable(clone_function): raise ValueError( "Expected `clone_function` argument to be a callable. " f"Received: clone_function={clone_function}" ) if not isinstance(model, Functional): raise ValueError( "Expected `model` argument " f"to be a Functional Model instance. Received: model={model}" ) if input_tensors is not None: if not all( isinstance(x, backend.KerasTensor) for x in tree.flatten(input_tensors) ): raise ValueError( "All entries in `input_tensors` must be KerasTensors. " f"Received invalid values: inputs_tensors={input_tensors}" ) try: tree.assert_same_structure(input_tensors, model.input) except ValueError as e: raise ValueError( "`input_tensors` must have the same structure as model.input" f"\nReference structure: {model.input}" f"\nReceived structure: {input_tensors}" ) from e else: input_tensors = tree.map_structure( lambda x: Input(batch_shape=x.shape, dtype=x.dtype, name=x.name), model.input, ) def operation_fn(layer): new_layer = clone_function(layer) return new_layer output_tensors = model._run_through_graph( input_tensors, operation_fn=operation_fn, call_fn=call_function, ) if functional_like_constructor(model.__class__): new_model = model.__class__( input_tensors, output_tensors, name=model.name ) else: # This may be incorrect: the new model will end up having a different # class than the original. However various existing models rely # on this behavior, so we keep it. new_model = Functional(input_tensors, output_tensors, name=model.name) if model.compiled: compiled_config = model.get_compile_config() new_model.compile_from_config(compiled_config) return new_model