object config serialization and deserialization logic import importlib import inspect import types import warnings import numpy as np from keras import apiexport from keras import backend from keras apiexport import kerasexport from keras backend common import globalstate from keras saving import objectregistration from keras utils import pythonutils from keras utils moduleutils import tensorflow as tf plaintypes str int float bool list of keras modules with builtin string representations for keras defaults builtinmodules activations constraints initializers losses metrics optimizers regularizers class serializabledict def initself config self config config def serializeself return serializekerasobjectself config class safemodescope disables safe mode globally allowing deserialization of lambdas globalstate setglobalattributesafemodesaving false def insafemode return globalstate getglobalattributesafemodesaving class objectsharingscope retrieve an object previously seen during deserialization idtoobjmap globalstate getglobalattribute sharedobjectsidtoobjmap if idtoobjmap is not none return idtoobjmap getobjid none def recordobjectafterserializationobj config call after deserializing an object to keep track of it in the future idtoobjmap globalstate getglobalattribute sharedobjectsidtoobjmap if idtoobjmap is none return not in a sharing scope idtoobjmapobjid obj kerasexport keras saving serializekerasobject keras utils serializekerasobject def serializekerasobjectobj if obj is none return obj if isinstanceobj plaintypes return obj if isinstanceobj list tuple configarr serializekerasobjectx for x in obj return tupleconfigarr if isinstanceobj tuple else configarr if isinstanceobj dict return serializedictobj special cases if isinstanceobj bytes return classname bytes config value obj decodeutf8 if isinstanceobj slice return classname slice config start serializekerasobjectobj start stop serializekerasobjectobj stop step serializekerasobjectobj step if isinstanceobj backend kerastensor history getattrobj kerashistory none if history history listhistory history0 history0 name return classname kerastensor config shape obj shape dtype obj dtype kerashistory history if tf available and isinstanceobj tf tensorshape return obj aslist if obj dims is not none else none if backend istensorobj return classname tensor config value backend converttonumpyobj tolist dtype backend standardizedtypeobj dtype if typeobj module np name if isinstanceobj np ndarray and obj ndim 0 return classname numpy config value obj tolist dtype backend standardizedtypeobj dtype else treat numpy floats etc as plain types return obj item if tf available and isinstanceobj tf dtype return obj name if isinstanceobj types functiontype and obj name lambda warnings warn the object being serialized includes a lambda this is unsafe in order to reload the object you will have to pass safemodefalse to the loading function please avoid using lambda in the future and use named python functions instead fthis is the lambda being serialized inspect getsourceobj stacklevel2 return classname lambda config value pythonutils funcdumpobj if tf available and isinstanceobj tf typespec tsconfig obj serialize tensorshape and tf dtype conversion tsconfig list map lambda x x aslist if isinstancex tf tensorshape else x name if isinstancex tf dtype else x tsconfig return classname typespec specname obj class name module obj class module config tsconfig registeredname none innerconfig getclassorfnconfigobj configwithpublicclass serializewithpublicclass obj class innerconfig if configwithpublicclass is not none getbuildandcompileconfigobj configwithpublicclass recordobjectafterserializationobj configwithpublicclass return configwithpublicclass any custom object or otherwise nonexported object if isinstanceobj types functiontype module obj module else module obj class module classname obj class name if module builtins registeredname none else if isinstanceobj types functiontype registeredname objectregistration getregisterednameobj else registeredname objectregistration getregisteredname obj class config module module classname classname config innerconfig registeredname registeredname getbuildandcompileconfigobj config recordobjectafterserializationobj config return config def getbuildandcompileconfigobj config if hasattrobj getbuildconfig buildconfig obj getbuildconfig if buildconfig is not none configbuildconfig serializedictbuildconfig if hasattrobj getcompileconfig compileconfig obj getcompileconfig if compileconfig is not none configcompileconfig serializedictcompileconfig return def serializewithpublicclasscls innerconfignone this gets the keras exported name such as keras optimizers adam kerasapiname apiexport getnamefromsymbolcls case of custom or unknown class object if kerasapiname is none registeredname objectregistration getregisterednamecls if registeredname is none return none return custom object config with corresponding registration name return module cls module classname cls name config innerconfig registeredname registeredname split the canonical keras api name into a keras module and class name parts kerasapiname split return module joinparts 1 classname parts1 config innerconfig registeredname none def serializewithpublicfnfn config fnmodulenamenone if fnmodulename return module fnmodulename classname function config config registeredname config kerasapiname apiexport getnamefromsymbolfn if kerasapiname parts kerasapiname split return module joinparts 1 classname function config config registeredname config else registeredname objectregistration getregisterednamefn if not registeredname and not fn module builtins return none return module fn module classname function config config registeredname registeredname def getclassorfnconfigobj functions lambdas all classes retrieve the object by deserializing the config dict the config dict is a python dictionary that consists of a set of keyvalue pairs and represents a keras object such as an optimizer layer metrics etc the saving and loading library uses the following keys to record information of a keras object classname string this is the name of the class as exactly defined in the source code such as lossescontainer config dict librarydefined or userdefined keyvalue pairs that store the configuration of the object as obtained by object getconfig module string the path of the python module builtin keras classes expect to have prefix keras registeredname string the key the class is registered under via keras saving registerkerasserializablepackage name api the key has the format of packagename where package and name are the arguments passed to registerkerasserializable if name is not provided it uses the class name if registeredname successfully resolves to a class that was registered the classname and config values in the dict will not be used registeredname is only used for nonbuiltin classes for example the following dictionary represents the builtin adam optimizer with the relevant config python dictstructure classname adam config amsgrad false beta1 0 8999999761581421 beta2 0 9990000128746033 decay 0 0 epsilon 1e07 learningrate 0 0010000000474974513 name adam module keras optimizers registeredname none returns an adam instance identical to the original one deserializekerasobjectdictstructure if the class does not have an exported keras namespace the library tracks it by its module and classname for example python dictstructure classname metricslist config module keras trainers compileutils registeredname metricslist returns a metricslist instance identical to the original one deserializekerasobjectdictstructure and the following dictionary represents a usercustomized meansquarederror loss python keras saving registerkerasserializablepackage mypackage class modifiedmeansquarederrorkeras losses meansquarederror dictstructure classname modifiedmeansquarederror config fn meansquarederror name meansquarederror reduction auto registeredname mypackagemodifiedmeansquarederror returns the modifiedmeansquarederror object deserializekerasobjectdictstructure args config python dict describing the object customobjects python dict containing a mapping between custom object names the corresponding classes or functions safemode boolean whether to disallow unsafe lambda deserialization when safemodefalse loading an object has the potential to trigger arbitrary code execution this argument is only applicable to the keras v3 model format defaults to true returns the object described by the config dictionary this is to deserialize plain functions which are serialized as string names by legacy saving formats check case where config is function or class and in custom objects case where config is function but not in custom objects case where config is class but not in custom objects return if not found in either module objects or custom objects object has already been deserialized special cases conversion to tensorshape and dtype below classes and functions below handling of all classes first is it a shared object instantiate the class from its config inside a custom object scope so that we can catch any custom objects that the config refers to if there is a custom object registered via registerkerasserializable that takes precedence if it s a keras builtin object we cannot always use direct import because the exported module name might not match the package structure e g experimental symbols configs of keras builtin functions do not contain identifying information other than their name e g acc or tanh this special case searches the keras modules that contain builtins to retrieve the corresponding function from the identifying string retrieval of registered custom function in a package otherwise attempt to retrieve the class object given the module and classname import the module find the class special case for keras metrics metrics list of keras modules with built in string representations for keras defaults scope to propagate safe mode flag to nested deserialization calls disables safe mode globally allowing deserialization of lambdas scope to enable detection and reuse of previously seen objects retrieve an object previously seen during deserialization call after serializing an object to keep track of its config ensures module is none when no module found not in a sharing scope call after deserializing an object to keep track of it in the future not in a sharing scope retrieve the config dict by serializing the keras object serialize_keras_object serializes a keras object to a python dictionary that represents the object and is a reciprocal function of deserialize_keras_object see deserialize_keras_object for more information about the config format args obj the keras object to serialize returns a python dict that represents the object the python dict can be deserialized via deserialize_keras_object special cases treat numpy floats etc as plain types tensorshape and tf dtype conversion any custom object or otherwise non exported object serializes classes from public keras api or object registration called to check and retrieve the config of any class that has a public keras api or has been registered as serializable via keras saving register_keras_serializable this gets the keras exported name such as keras optimizers adam case of custom or unknown class object return custom object config with corresponding registration name split the canonical keras api name into a keras module and class name serializes functions from public keras api or object registration called to check and retrieve the config of any function that has a public keras api or has been registered as serializable via keras saving register_keras_serializable if function s module name is already known returns corresponding config return the object s config depending on its type functions lambdas all classes retrieve the object by deserializing the config dict the config dict is a python dictionary that consists of a set of key value pairs and represents a keras object such as an optimizer layer metrics etc the saving and loading library uses the following keys to record information of a keras object class_name string this is the name of the class as exactly defined in the source code such as lossescontainer config dict library defined or user defined key value pairs that store the configuration of the object as obtained by object get_config module string the path of the python module built in keras classes expect to have prefix keras registered_name string the key the class is registered under via keras saving register_keras_serializable package name api the key has the format of package name where package and name are the arguments passed to register_keras_serializable if name is not provided it uses the class name if registered_name successfully resolves to a class that was registered the class_name and config values in the dict will not be used registered_name is only used for non built in classes for example the following dictionary represents the built in adam optimizer with the relevant config python dict_structure class_name adam config amsgrad false beta_1 0 8999999761581421 beta_2 0 9990000128746033 decay 0 0 epsilon 1e 07 learning_rate 0 0010000000474974513 name adam module keras optimizers registered_name none returns an adam instance identical to the original one deserialize_keras_object dict_structure if the class does not have an exported keras namespace the library tracks it by its module and class_name for example python dict_structure class_name metricslist config module keras trainers compile_utils registered_name metricslist returns a metricslist instance identical to the original one deserialize_keras_object dict_structure and the following dictionary represents a user customized meansquarederror loss python keras saving register_keras_serializable package my_package class modifiedmeansquarederror keras losses meansquarederror dict_structure class_name modifiedmeansquarederror config fn mean_squared_error name mean_squared_error reduction auto registered_name my_package modifiedmeansquarederror returns the modifiedmeansquarederror object deserialize_keras_object dict_structure args config python dict describing the object custom_objects python dict containing a mapping between custom object names the corresponding classes or functions safe_mode boolean whether to disallow unsafe lambda deserialization when safe_mode false loading an object has the potential to trigger arbitrary code execution this argument is only applicable to the keras v3 model format defaults to true returns the object described by the config dictionary enforces safemodescope this is to deserialize plain functions which are serialized as string names by legacy saving formats check case where config is function or class and in custom objects case where config is function but not in custom objects case where config is class but not in custom objects return if not found in either module objects or custom objects object has already been deserialized special cases conversion to tensorshape and dtype below classes and functions below handling of all classes first is it a shared object instantiate the class from its config inside a custom object scope so that we can catch any custom objects that the config refers to if there is a custom object registered via register_keras_serializable that takes precedence if it s a keras built in object we cannot always use direct import because the exported module name might not match the package structure e g experimental symbols configs of keras built in functions do not contain identifying information other than their name e g acc or tanh this special case searches the keras modules that contain built ins to retrieve the corresponding function from the identifying string retrieval of registered custom function in a package otherwise attempt to retrieve the class object given the module and class_name import the module find the class special case for keras metrics metrics
import importlib
import inspect
import types
import warnings

import numpy as np

from keras import api_export
from keras import backend
from keras.api_export import keras_export
from keras.backend.common import global_state
from keras.saving import object_registration
from keras.utils import python_utils
from keras.utils.module_utils import tensorflow as tf

PLAIN_TYPES = (str, int, float, bool)


BUILTIN_MODULES = (
    "activations",
    "constraints",
    "initializers",
    "losses",
    "metrics",
    "optimizers",
    "regularizers",
)


class SerializableDict:
    def __init__(self, **config):
        self.config = config

    def serialize(self):
        return serialize_keras_object(self.config)


class SafeModeScope:
    

    def __init__(self, safe_mode=True):
        self.safe_mode = safe_mode

    def __enter__(self):
        self.original_value = in_safe_mode()
        global_state.set_global_attribute("safe_mode_saving", self.safe_mode)

    def __exit__(self, *args, **kwargs):
        global_state.set_global_attribute(
            "safe_mode_saving", self.original_value
        )


@keras_export("keras.config.enable_unsafe_deserialization")
def enable_unsafe_deserialization():
    
    global_state.set_global_attribute("safe_mode_saving", False)


def in_safe_mode():
    return global_state.get_global_attribute("safe_mode_saving")


class ObjectSharingScope:
    

    def __enter__(self):
        global_state.set_global_attribute("shared_objects/id_to_obj_map", {})
        global_state.set_global_attribute("shared_objects/id_to_config_map", {})

    def __exit__(self, *args, **kwargs):
        global_state.set_global_attribute("shared_objects/id_to_obj_map", None)
        global_state.set_global_attribute(
            "shared_objects/id_to_config_map", None
        )


def get_shared_object(obj_id):
    
    id_to_obj_map = global_state.get_global_attribute(
        "shared_objects/id_to_obj_map"
    )
    if id_to_obj_map is not None:
        return id_to_obj_map.get(obj_id, None)


def record_object_after_serialization(obj, config):
    
    if config["module"] == "__main__":
        config["module"] = None  
    id_to_config_map = global_state.get_global_attribute(
        "shared_objects/id_to_config_map"
    )
    if id_to_config_map is None:
        return  
    obj_id = int(id(obj))
    if obj_id not in id_to_config_map:
        id_to_config_map[obj_id] = config
    else:
        config["shared_object_id"] = obj_id
        prev_config = id_to_config_map[obj_id]
        prev_config["shared_object_id"] = obj_id


def record_object_after_deserialization(obj, obj_id):
    
    id_to_obj_map = global_state.get_global_attribute(
        "shared_objects/id_to_obj_map"
    )
    if id_to_obj_map is None:
        return  
    id_to_obj_map[obj_id] = obj


@keras_export(
    [
        "keras.saving.serialize_keras_object",
        "keras.utils.serialize_keras_object",
    ]
)
def serialize_keras_object(obj):
    
    if obj is None:
        return obj

    if isinstance(obj, PLAIN_TYPES):
        return obj

    if isinstance(obj, (list, tuple)):
        config_arr = [serialize_keras_object(x) for x in obj]
        return tuple(config_arr) if isinstance(obj, tuple) else config_arr
    if isinstance(obj, dict):
        return serialize_dict(obj)

    
    if isinstance(obj, bytes):
        return {
            "class_name": "__bytes__",
            "config": {"value": obj.decode("utf-8")},
        }
    if isinstance(obj, slice):
        return {
            "class_name": "__slice__",
            "config": {
                "start": serialize_keras_object(obj.start),
                "stop": serialize_keras_object(obj.stop),
                "step": serialize_keras_object(obj.step),
            },
        }
    if isinstance(obj, backend.KerasTensor):
        history = getattr(obj, "_keras_history", None)
        if history:
            history = list(history)
            history[0] = history[0].name
        return {
            "class_name": "__keras_tensor__",
            "config": {
                "shape": obj.shape,
                "dtype": obj.dtype,
                "keras_history": history,
            },
        }
    if tf.available and isinstance(obj, tf.TensorShape):
        return obj.as_list() if obj._dims is not None else None
    if backend.is_tensor(obj):
        return {
            "class_name": "__tensor__",
            "config": {
                "value": backend.convert_to_numpy(obj).tolist(),
                "dtype": backend.standardize_dtype(obj.dtype),
            },
        }
    if type(obj).__module__ == np.__name__:
        if isinstance(obj, np.ndarray) and obj.ndim > 0:
            return {
                "class_name": "__numpy__",
                "config": {
                    "value": obj.tolist(),
                    "dtype": backend.standardize_dtype(obj.dtype),
                },
            }
        else:
            
            return obj.item()
    if tf.available and isinstance(obj, tf.DType):
        return obj.name
    if isinstance(obj, types.FunctionType) and obj.__name__ == "<lambda>":
        warnings.warn(
            "The object being serialized includes a `lambda`. This is unsafe. "
            "In order to reload the object, you will have to pass "
            "`safe_mode=False` to the loading function. "
            "Please avoid using `lambda` in the "
            "future, and use named Python functions instead. "
            f"This is the `lambda` being serialized: {inspect.getsource(obj)}",
            stacklevel=2,
        )
        return {
            "class_name": "__lambda__",
            "config": {
                "value": python_utils.func_dump(obj),
            },
        }
    if tf.available and isinstance(obj, tf.TypeSpec):
        ts_config = obj._serialize()
        
        ts_config = list(
            map(
                lambda x: x.as_list()
                if isinstance(x, tf.TensorShape)
                else (x.name if isinstance(x, tf.DType) else x),
                ts_config,
            )
        )
        return {
            "class_name": "__typespec__",
            "spec_name": obj.__class__.__name__,
            "module": obj.__class__.__module__,
            "config": ts_config,
            "registered_name": None,
        }

    inner_config = _get_class_or_fn_config(obj)
    config_with_public_class = serialize_with_public_class(
        obj.__class__, inner_config
    )

    if config_with_public_class is not None:
        get_build_and_compile_config(obj, config_with_public_class)
        record_object_after_serialization(obj, config_with_public_class)
        return config_with_public_class

    
    if isinstance(obj, types.FunctionType):
        module = obj.__module__
    else:
        module = obj.__class__.__module__
    class_name = obj.__class__.__name__

    if module == "builtins":
        registered_name = None
    else:
        if isinstance(obj, types.FunctionType):
            registered_name = object_registration.get_registered_name(obj)
        else:
            registered_name = object_registration.get_registered_name(
                obj.__class__
            )

    config = {
        "module": module,
        "class_name": class_name,
        "config": inner_config,
        "registered_name": registered_name,
    }
    get_build_and_compile_config(obj, config)
    record_object_after_serialization(obj, config)
    return config


def get_build_and_compile_config(obj, config):
    if hasattr(obj, "get_build_config"):
        build_config = obj.get_build_config()
        if build_config is not None:
            config["build_config"] = serialize_dict(build_config)
    if hasattr(obj, "get_compile_config"):
        compile_config = obj.get_compile_config()
        if compile_config is not None:
            config["compile_config"] = serialize_dict(compile_config)
    return


def serialize_with_public_class(cls, inner_config=None):
    
    
    
    keras_api_name = api_export.get_name_from_symbol(cls)

    
    if keras_api_name is None:
        registered_name = object_registration.get_registered_name(cls)
        if registered_name is None:
            return None

        
        return {
            "module": cls.__module__,
            "class_name": cls.__name__,
            "config": inner_config,
            "registered_name": registered_name,
        }

    
    parts = keras_api_name.split(".")
    return {
        "module": ".".join(parts[:-1]),
        "class_name": parts[-1],
        "config": inner_config,
        "registered_name": None,
    }


def serialize_with_public_fn(fn, config, fn_module_name=None):
    
    if fn_module_name:
        return {
            "module": fn_module_name,
            "class_name": "function",
            "config": config,
            "registered_name": config,
        }
    keras_api_name = api_export.get_name_from_symbol(fn)
    if keras_api_name:
        parts = keras_api_name.split(".")
        return {
            "module": ".".join(parts[:-1]),
            "class_name": "function",
            "config": config,
            "registered_name": config,
        }
    else:
        registered_name = object_registration.get_registered_name(fn)
        if not registered_name and not fn.__module__ == "builtins":
            return None
        return {
            "module": fn.__module__,
            "class_name": "function",
            "config": config,
            "registered_name": registered_name,
        }


def _get_class_or_fn_config(obj):
    
    
    if isinstance(obj, types.FunctionType):
        return obj.__name__
    
    if hasattr(obj, "get_config"):
        config = obj.get_config()
        if not isinstance(config, dict):
            raise TypeError(
                f"The `get_config()` method of {obj} should return "
                f"a dict. It returned: {config}"
            )
        return serialize_dict(config)
    elif hasattr(obj, "__name__"):
        return object_registration.get_registered_name(obj)
    else:
        raise TypeError(
            f"Cannot serialize object {obj} of type {type(obj)}. "
            "To be serializable, "
            "a class must implement the `get_config()` method."
        )


def serialize_dict(obj):
    return {key: serialize_keras_object(value) for key, value in obj.items()}


@keras_export(
    [
        "keras.saving.deserialize_keras_object",
        "keras.utils.deserialize_keras_object",
    ]
)
def deserialize_keras_object(
    config, custom_objects=None, safe_mode=True, **kwargs
):
    
    safe_scope_arg = in_safe_mode()  
    safe_mode = safe_scope_arg if safe_scope_arg is not None else safe_mode

    module_objects = kwargs.pop("module_objects", None)
    custom_objects = custom_objects or {}
    tlco = global_state.get_global_attribute("custom_objects_scope_dict", {})
    gco = object_registration.GLOBAL_CUSTOM_OBJECTS
    custom_objects = {**custom_objects, **tlco, **gco}

    if config is None:
        return None

    if (
        isinstance(config, str)
        and custom_objects
        and custom_objects.get(config) is not None
    ):
        
        
        return custom_objects[config]

    if isinstance(config, (list, tuple)):
        return [
            deserialize_keras_object(
                x, custom_objects=custom_objects, safe_mode=safe_mode
            )
            for x in config
        ]

    if module_objects is not None:
        inner_config, fn_module_name, has_custom_object = None, None, False

        if isinstance(config, dict):
            if "config" in config:
                inner_config = config["config"]
            if "class_name" not in config:
                raise ValueError(
                    f"Unknown `config` as a `dict`, config={config}"
                )

            
            if custom_objects and (
                config["class_name"] in custom_objects
                or config.get("registered_name") in custom_objects
                or (
                    isinstance(inner_config, str)
                    and inner_config in custom_objects
                )
            ):
                has_custom_object = True

            
            elif config["class_name"] == "function":
                fn_module_name = config["module"]
                if fn_module_name == "builtins":
                    config = config["config"]
                else:
                    config = config["registered_name"]

            
            else:
                if config.get("module", "_") is None:
                    raise TypeError(
                        "Cannot deserialize object of type "
                        f"`{config['class_name']}`. If "
                        f"`{config['class_name']}` is a custom class, please "
                        "register it using the "
                        "`@keras.saving.register_keras_serializable()` "
                        "decorator."
                    )
                config = config["class_name"]

        if not has_custom_object:
            
            if config not in module_objects:
                
                return config
            if isinstance(module_objects[config], types.FunctionType):
                return deserialize_keras_object(
                    serialize_with_public_fn(
                        module_objects[config], config, fn_module_name
                    ),
                    custom_objects=custom_objects,
                )
            return deserialize_keras_object(
                serialize_with_public_class(
                    module_objects[config], inner_config=inner_config
                ),
                custom_objects=custom_objects,
            )

    if isinstance(config, PLAIN_TYPES):
        return config
    if not isinstance(config, dict):
        raise TypeError(f"Could not parse config: {config}")

    if "class_name" not in config or "config" not in config:
        return {
            key: deserialize_keras_object(
                value, custom_objects=custom_objects, safe_mode=safe_mode
            )
            for key, value in config.items()
        }

    class_name = config["class_name"]
    inner_config = config["config"] or {}
    custom_objects = custom_objects or {}

    
    if class_name == "__keras_tensor__":
        obj = backend.KerasTensor(
            inner_config["shape"], dtype=inner_config["dtype"]
        )
        obj._pre_serialization_keras_history = inner_config["keras_history"]
        return obj

    if class_name == "__tensor__":
        return backend.convert_to_tensor(
            inner_config["value"], dtype=inner_config["dtype"]
        )
    if class_name == "__numpy__":
        return np.array(inner_config["value"], dtype=inner_config["dtype"])
    if config["class_name"] == "__bytes__":
        return inner_config["value"].encode("utf-8")
    if config["class_name"] == "__slice__":
        return slice(
            deserialize_keras_object(
                inner_config["start"],
                custom_objects=custom_objects,
                safe_mode=safe_mode,
            ),
            deserialize_keras_object(
                inner_config["stop"],
                custom_objects=custom_objects,
                safe_mode=safe_mode,
            ),
            deserialize_keras_object(
                inner_config["step"],
                custom_objects=custom_objects,
                safe_mode=safe_mode,
            ),
        )
    if config["class_name"] == "__lambda__":
        if safe_mode:
            raise ValueError(
                "Requested the deserialization of a `lambda` object. "
                "This carries a potential risk of arbitrary code execution "
                "and thus it is disallowed by default. If you trust the "
                "source of the saved model, you can pass `safe_mode=False` to "
                "the loading function in order to allow `lambda` loading, "
                "or call `keras.config.enable_unsafe_deserialization()`."
            )
        return python_utils.func_load(inner_config["value"])
    if tf is not None and config["class_name"] == "__typespec__":
        obj = _retrieve_class_or_fn(
            config["spec_name"],
            config["registered_name"],
            config["module"],
            obj_type="class",
            full_config=config,
            custom_objects=custom_objects,
        )
        
        inner_config = map(
            lambda x: tf.TensorShape(x)
            if isinstance(x, list)
            else (getattr(tf, x) if hasattr(tf.dtypes, str(x)) else x),
            inner_config,
        )
        return obj._deserialize(tuple(inner_config))

    
    module = config.get("module", None)
    registered_name = config.get("registered_name", class_name)

    if class_name == "function":
        fn_name = inner_config
        return _retrieve_class_or_fn(
            fn_name,
            registered_name,
            module,
            obj_type="function",
            full_config=config,
            custom_objects=custom_objects,
        )

    
    
    if "shared_object_id" in config:
        obj = get_shared_object(config["shared_object_id"])
        if obj is not None:
            return obj

    cls = _retrieve_class_or_fn(
        class_name,
        registered_name,
        module,
        obj_type="class",
        full_config=config,
        custom_objects=custom_objects,
    )

    if isinstance(cls, types.FunctionType):
        return cls
    if not hasattr(cls, "from_config"):
        raise TypeError(
            f"Unable to reconstruct an instance of '{class_name}' because "
            f"the class is missing a `from_config()` method. "
            f"Full object config: {config}"
        )

    
    
    custom_obj_scope = object_registration.CustomObjectScope(custom_objects)
    safe_mode_scope = SafeModeScope(safe_mode)
    with custom_obj_scope, safe_mode_scope:
        try:
            instance = cls.from_config(inner_config)
        except TypeError as e:
            raise TypeError(
                f"{cls} could not be deserialized properly. Please"
                " ensure that components that are Python object"
                " instances (layers, models, etc.) returned by"
                " `get_config()` are explicitly deserialized in the"
                " model's `from_config()` method."
                f"\n\nconfig={config}.\n\nException encountered: {e}"
            )
        build_config = config.get("build_config", None)
        if build_config and not instance.built:
            instance.build_from_config(build_config)
            instance.built = True
        compile_config = config.get("compile_config", None)
        if compile_config:
            instance.compile_from_config(compile_config)
            instance.compiled = True

    if "shared_object_id" in config:
        record_object_after_deserialization(
            instance, config["shared_object_id"]
        )
    return instance


def _retrieve_class_or_fn(
    name, registered_name, module, obj_type, full_config, custom_objects=None
):
    
    
    if obj_type == "function":
        custom_obj = object_registration.get_registered_object(
            name, custom_objects=custom_objects
        )
    else:
        custom_obj = object_registration.get_registered_object(
            registered_name, custom_objects=custom_objects
        )
    if custom_obj is not None:
        return custom_obj

    if module:
        
        
        
        
        if module == "keras" or module.startswith("keras."):
            api_name = module + "." + name

            obj = api_export.get_symbol_from_name(api_name)
            if obj is not None:
                return obj

        
        
        
        
        if obj_type == "function" and module == "builtins":
            for mod in BUILTIN_MODULES:
                obj = api_export.get_symbol_from_name(
                    "keras." + mod + "." + name
                )
                if obj is not None:
                    return obj

            
            filtered_dict = {
                k: v
                for k, v in custom_objects.items()
                if k.endswith(full_config["config"])
            }
            if filtered_dict:
                return next(iter(filtered_dict.values()))

        
        
        try:
            mod = importlib.import_module(module)
        except ModuleNotFoundError:
            raise TypeError(
                f"Could not deserialize {obj_type} '{name}' because "
                f"its parent module {module} cannot be imported. "
                f"Full object config: {full_config}"
            )
        obj = vars(mod).get(name, None)

        
        if obj is None and registered_name is not None:
            obj = vars(mod).get(registered_name, None)

        if obj is not None:
            return obj

    raise TypeError(
        f"Could not locate {obj_type} '{name}'. "
        "Make sure custom classes are decorated with "
        "`@keras.saving.register_keras_serializable()`. "
        f"Full object config: {full_config}"
    )