以下为脚本文件yuanweihua.py
from __future__ import print_function
import os
os.environ['TF_USE_LEGACY_KERAS'] = '1'
from tensorflow.python.keras.models import Sequential
from tensorflow.python.keras.layers import Dense, Activation
import tensorflow as tf
from tensorflow.python.keras.regularizers import l2
from tensorflow.python.keras import backend as K, Input
from tensorflow.python.keras.optimizer_v2.adam import Adam
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import OneHotEncoder
from sklearn.metrics import roc_curve, auc
import numpy as np
import matplotlib.pyplot as plt
from tensorflow.python.keras.models import Model
from importance_sampling.training import ConstantTimeImportanceTraining
from example_utils import get_parser
import tensorflow as tf
tf.compat.v1.disable_eager_execution()

def plot_history(history):
    plt.figure(figsize=(12, 5))

    # 绘制训练集和验证集的准确率
    plt.subplot(121)
    plt.plot(history.history['accuracy'])
    plt.plot(history.history['val_accuracy'])
    plt.title('Model accuracy')
    plt.ylabel('Accuracy')
    plt.xlabel('Epoch')
    plt.legend(['Train', 'Validation'], loc='upper left')

    # 绘制训练集和验证集的损失
    plt.subplot(122)
    plt.plot(history.history['loss'])
    plt.plot(history.history['val_loss'])
    plt.title('Model loss')
    plt.ylabel('Loss')
    plt.xlabel('Epoch')
    plt.legend(['Train', 'Validation'], loc='upper left')

    plt.tight_layout()
    plt.show()


def plot_roc_curve(y_test, y_pred):
    n_classes = y_test.shape[1]
    fpr = dict()
    tpr = dict()
    roc_auc = dict()

    for i in range(n_classes):
        fpr[i], tpr[i], _ = roc_curve(y_test[:, i], y_pred[:, i])
        roc_auc[i] = auc(fpr[i], tpr[i])

    # Compute micro-average ROC curve and ROC area
    fpr["micro"], tpr["micro"], _ = roc_curve(y_test.ravel(), y_pred.ravel())
    roc_auc["micro"] = auc(fpr["micro"], tpr["micro"])

    plt.figure()
    plt.plot(fpr["micro"], tpr["micro"],
             label='micro-average ROC curve (area = {0:0.2f})'
                   ''.format(roc_auc["micro"]))
    for i in range(n_classes):
        plt.plot(fpr[i], tpr[i],
                 label='ROC curve of class {0} (area = {1:0.2f})'
                       ''.format(i, roc_auc[i]))

    plt.plot([0, 1], [0, 1], 'k--')
    plt.xlim([0.0, 1.0])
    plt.ylim([0.0, 1.05])
    plt.xlabel('False Positive Rate')
    plt.ylabel('True Positive Rate')
    plt.title('Receiver Operating Characteristic (ROC) Curve')
    plt.legend(loc="lower right")
    plt.show()


def calculate_accuracy_fluctuation(history):
    # 获取每个 epoch 的验证集准确率
    val_accuracies = history.history['val_accuracy']

    # 计算平均准确率
    mean_accuracy = np.mean(val_accuracies)

    # 计算每个 epoch 的准确率波动（与均值的绝对差值）
    fluctuation = np.sum(np.abs(val_accuracies - mean_accuracy))

    return mean_accuracy, fluctuation


if __name__ == "__main__":
    parser = get_parser("Train a model on Iris dataset")
    args = parser.parse_args()

    # 导入数据，分别为输入特征和标签
    iris = datasets.load_iris()
    x_data = iris.data
    y_data = iris.target.reshape(-1, 1)

    # 使用OneHotEncoder进行标签的独热编码
    encoder = OneHotEncoder(sparse_output=False)
    y_data = encoder.fit_transform(y_data)

    # 分割数据集为训练集和测试集
    x_train, x_test, y_train, y_test = train_test_split(x_data, y_data, test_size=0.3, random_state=116)

    # 定义神经网络的超参数
    learning_rate = 0.1
    epochs = 100
    batch_size = 8
    num_classes = 3  # 鸢尾花数据集有3个类别

    model = Sequential()
    model.add(Dense(10, activation='relu', input_shape=(4,), kernel_regularizer=l2(1e-5)))
    model.add(Dense(num_classes, kernel_regularizer=l2(1e-5)))
    model.add(Activation('softmax'))

    # **在包装之前编译模型**
    model.compile(
        optimizer=Adam(learning_rate=learning_rate),
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )

    # 在初始化 ConstantTimeImportanceTraining 之前构建模型
    model.predict(np.zeros((1, 4)))  # 这里用模型的输入形状(1, 4)

    # 使用重要性采样训练或常规训练
    if args.importance_training:
        wrapped = ConstantTimeImportanceTraining(model)
    else:
        wrapped = model

    history = wrapped.fit(x_train, y_train,
                          batch_size=batch_size,
                          epochs=epochs,
                          verbose=1,
                          validation_data=(x_test, y_test))

    score = model.evaluate(x_test, y_test, verbose=0)
    print('Test loss:', score[0])
    print('Test accuracy:', score[1])

    # 计算并输出准确率波动
    mean_accuracy, fluctuation = calculate_accuracy_fluctuation(history)
    print('Mean validation accuracy:', mean_accuracy)
    print('Accuracy fluctuation:', fluctuation)

    # 绘制训练历史
    plot_history(history)

    # 计算预测概率
    y_pred = model.predict(x_test)

    # 绘制ROC曲线
    plot_roc_curve(y_test, y_pred)
以下为训练文件training.py
import sys
import numpy as np
import logging
from collections.abc import Iterable
from tensorflow.python.keras.callbacks import BaseLogger, CallbackList, History, ProgbarLogger
from tensorflow.python.keras.layers import Dropout
import tensorflow as tf
#from keras.src.utils import Sequence

import numpy as np
from blinker import signal

from .datasets import InMemoryDataset, GeneratorDataset
from .model_wrappers import OracleWrapper
from .samplers import ConditionalStartSampler, VarianceReductionCondition, \
    AdaptiveAdditiveSmoothingSampler, AdditiveSmoothingSampler, ModelSampler, \
     ConstantVarianceSampler, ConstantTimeSampler, SCSGSampler
from .reweighting import BiasedReweightingPolicy, NoReweightingPolicy
from .utils.functional import ___, compose, partial
class _UnbiasedImportanceTraining(_BaseImportanceTraining):
    def __init__(self, model, score="gnorm", layer=None):
        self._reweighting = BiasedReweightingPolicy(1.0)  # no bias

        super(_UnbiasedImportanceTraining, self).__init__(model, score, layer)

    @property
    def reweighting(self):
        return self._reweighting


class ImportanceTraining(_UnbiasedImportanceTraining):
    """Train a model with exact importance sampling.

    Arguments
    ---------
        model: The Keras model to train
        presample: float, the number of samples to presample for scoring
                   given as a factor of the batch size
        tau_th: float or None, the variance reduction threshold after which we
                enable importance sampling
        forward_batch_size: int or None, the batch size to use for the forward
                            pass during scoring
        score: {"gnorm", "loss", "full_gnorm"}, the importance metric to use
               for importance sampling
        layer: None or int or Layer, the layer to compute the gnorm with
    """

    def __init__(self, model, presample=3.0, tau_th=None,
                 forward_batch_size=None, score="gnorm", layer=None):
        self._presample = presample
        self._tau_th = tau_th
        self._forward_batch_size = forward_batch_size

        # Call the parent to wrap the model
        super(ImportanceTraining, self).__init__(model, score, layer)

    def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
        # Configure the sampler
        large_batch = int(self._presample * batch_size)
        forward_batch_size = self._forward_batch_size or batch_size
        # Compute the threshold using eq. 29 in
        # https://arxiv.org/abs/1803.00942
        B = large_batch
        b = batch_size
        tau_th = float(B + 3 * b) / (3 * b)
        tau_th = self._tau_th if self._tau_th is not None else tau_th

        return ConditionalStartSampler(
            ModelSampler(
                dataset,
                self.reweighting,
                self.model,
                large_batch=large_batch,
                forward_batch_size=forward_batch_size
            ),
            VarianceReductionCondition(tau_th)
        )


class ConstantVarianceImportanceTraining(_UnbiasedImportanceTraining):
    """Train a model faster by keeping the per iteration variance constant but
    decreasing the time.

    Arguments
    ---------
        model: The Keras model to train
        score: {"gnorm", "loss", "full_gnorm"}, the importance metric to use
               for importance sampling
        layer: None or int or Layer, the layer to compute the gnorm with
    """

    def __init__(self, model, backward_time=2.0, extra_samples=0.2,
                 score="gnorm", layer=None):
        self._backward_time = backward_time
        self._extra_samples = extra_samples

        super(ConstantVarianceImportanceTraining, self).__init__(
            model,
            score,
            layer
        )

    def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
        return ConstantVarianceSampler(
            dataset,
            self.reweighting,
            self.model,
            backward_time=self._backward_time,
            extra_samples=self._extra_samples
        )
class _BaseImportanceTraining(object):
    def __init__(self, model, score="gnorm", layer=None):
        """Abstract base class for training a model using importance sampling.

        Arguments
        ---------
            model: The Keras model to train
        """
        # Wrap and transform the model so that it outputs the importance scores
        # and can be used in an importance sampling training scheme

        self._check_model(model)
        self.original_model = model
        self.model = OracleWrapper(
            model,
            self.reweighting,
            score=score,
            layer=layer
        )
        self.original_model.optimizer = self.model.optimizer
        signal("is.sample").connect(self._on_sample)
        signal("is.score").connect(self._on_scores)

    def _on_sample(self, event):
        self._latest_sample_event = event

    def _on_scores(self, scores):
        self._latest_scores = scores

    def _check_model(self, model):
        """Check if the model uses Dropout and BatchNorm and warn that it may
        affect the importance calculations."""
        if any(
                isinstance(l, (tf.keras.layers.BatchNormalization, Dropout))
                for l in model.layers
        ):
            sys.stderr.write(("[NOTICE]: You are using BatchNormalization "
                              "and/or Dropout.\nThose layers may affect the "
                              "importance calculations and you are advised "
                              "to exchange them for LayerNormalization or "
                              "BatchNormalization in test mode and L2 "
                              "regularization.\n"))

    @property
    def reweighting(self):
        """The reweighting policy that controls the bias of the estimation when
        using importance sampling."""
        raise NotImplementedError()

    def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
        """Create a new sampler to sample from the given dataset using
        importance sampling."""
        raise NotImplementedError()

    def fit(self, x, y, batch_size=32, epochs=1, verbose=1, callbacks=None,
            validation_split=0.0, validation_data=None, steps_per_epoch=None,
            on_sample=None, on_scores=None):
        """Create an `InMemoryDataset` instance with the given data and train
        the model using importance sampling for a given number of epochs.

        Arguments
        ---------
            x: Numpy array of training data or list of numpy arrays
            y: Numpy array of target data
            batch_size: int, number of samples per gradient update
            epochs: int, number of times to iterate over the entire
                    training set
            verbose: {0, >0}, whether to employ the progress bar Keras
                     callback or not
            callbacks: list of Keras callbacks to be called during training
            validation_split: float in [0, 1), percentage of data to use for
                              evaluation
            validation_data: tuple of numpy arrays, Data to evaluate the
                             trained model on without ever training on them
            steps_per_epoch: int or None, number of gradient updates before
                             considering an epoch has passed
            on_sample: callable that accepts the sampler, idxs, w, scores
            on_scores: callable that accepts the sampler and scores
        Returns
        -------
            A Keras `History` object that contains information collected during
            training.
        """
        # Create two data tuples from the given x, y, validation_*
        if validation_data is not None:
            x_train, y_train = x, y
            x_test, y_test = validation_data

        elif validation_split > 0:
            assert validation_split < 1, "100% of the data used for testing"
            n = int(round(validation_split * len(x)))
            idxs = np.arange(len(x))
            np.random.shuffle(idxs)
            x_train, y_train = x[idxs[n:]], y[idxs[n:]]
            x_test, y_test = x[idxs[:n]], y[idxs[:n]]

        else:
            x_train, y_train = x, y
            x_test, y_test = np.empty(shape=(0, 1)), np.empty(shape=(0, 1))

        # Make the dataset to train on
        dataset = InMemoryDataset(
            x_train,
            y_train,
            x_test,
            y_test,
            categorical=False  # this means use the targets as is
        )

        return self.fit_dataset(
            dataset=dataset,
            batch_size=batch_size,
            epochs=epochs,
            steps_per_epoch=steps_per_epoch,
            verbose=verbose,
            callbacks=callbacks,
            on_sample=on_sample,
            on_scores=on_scores
        )

    def fit_generator(self, train, steps_per_epoch=None, batch_size=32,
                      epochs=1, verbose=1, callbacks=None,
                      validation_data=None, validation_steps=None,
                      on_sample=None, on_scores=None):
        """Create a GeneratorDataset instance and train the model using
        importance sampling for a given number of epochs.

        Arguments
        ---------
            train: A generator or iterable that returns tuples (inputs, targets)
            steps_per_epoch: int, number of gradient updates before considering
                             an epoch has passed
            batch_size: int, the number of samples per gradient update
            epochs: int, multiplied by steps_per_epoch denotes the total number
                    of gradient updates
            verbose: {0, >0}, whether to use the progress bar Keras callback
            validation_data: generator, iterable or tuple (inputs, targets)
            validation_steps: None or int, used only if validation_data is a
                              generator
            on_sample: callable that accepts the sampler, idxs, w, scores
            on_scores: callable that accepts the sampler and scores
        """

        def infinite_iterator(iterable):
            while True:
                yield from iterable

        # 创建验证数据集
        if validation_data is not None:
            if isinstance(validation_data, (tuple, list)):
                test = validation_data
                test_len = None
            elif isinstance(validation_data, Iterable):
                test = infinite_iterator(validation_data)
                try:
                    test_len = len(validation_data) * batch_size
                except TypeError:
                    test_len = validation_steps * batch_size if validation_steps else None
            else:
                test = validation_data
                test_len = validation_steps * batch_size if validation_steps else None
        else:
            test = (np.empty(shape=(0, 1)), np.empty(shape=(0, 1)))
            test_len = None
            logging.info("No validation data provided")

        # 设置训练数据和每个 epoch 的步数
        if isinstance(train, Iterable):
            if steps_per_epoch is None:
                try:
                    steps_per_epoch = len(train)
                except TypeError:
                    raise ValueError("steps_per_epoch must be set when using a generator without a __len__ method")
            train = infinite_iterator(train)
        elif steps_per_epoch is None:
            raise ValueError("steps_per_epoch is not set")

        dataset = GeneratorDataset(train, test, test_len)
        try:
            print("dataset.train_data=", len(dataset.train_data))
        except Exception as e:
            print(f"Exception occurred: {e}")

        return self.fit_dataset(
            dataset=dataset,
            batch_size=batch_size,
            epochs=epochs,
            steps_per_epoch=steps_per_epoch,
            verbose=verbose,
            callbacks=callbacks,
            on_sample=on_sample,
            on_scores=on_scores
        )

    def fit_dataset(self, dataset, steps_per_epoch=None, batch_size=32,
                    epochs=1, verbose=1, callbacks=None, on_sample=None,
                    on_scores=None):
        """Train the model on the given dataset for a given number of epochs.

        Arguments
        ---------
            dataset: Instance of `BaseDataset` that provides the data
                     to train on.
            steps_per_epoch: int or None, number of gradient updates before
                             considering an epoch has passed. If None it is set
                             to be `len(dataset.train_data) / batch_size`.
            batch_size: int, number of samples per gradient update
            epochs: int, number of times to iterate `steps_per_epoch` times
            verbose: {0, >0}, whether to employ the progress bar Keras
                     callback or not
            callbacks: list of Keras callbacks to be called during training
            on_sample: callable that accepts the sampler, idxs, w, scores
            on_scores: callable that accepts the sampler and scores
        """
        try:
            if len(dataset.train_data) < batch_size:
                raise ValueError(("The model cannot be trained with "
                                  "batch_size > training set"))
        except RuntimeError as e:
            assert "no size" in str(e)

        # Set steps_per_epoch properly
        if steps_per_epoch is None:
            steps_per_epoch = len(dataset.train_data) // batch_size

        # Create the callbacks list
        # 这段代码的目的是配置和初始化训练过程中将要使用的一组回调函数（callbacks）。
        # 回调函数是在训练的不同阶段被自动调用的函数，用于执行特定的任务，如记录训练指标、
        # 保存模型、早停（early stopping）等
        self.history = History()
        # History是一个回调（Callback），用于在训练过程中记录各种指标（如损失、准确率等）的值
        callbacks = [BaseLogger()] + (callbacks or []) + [self.history]
        if verbose > 0:
            # 添加 ProgbarLogger 回调。ProgbarLogger 用于在终端输出一个进度条，显示训练过程的进度。
            callbacks += [ProgbarLogger(count_mode="steps")]
        callbacks = CallbackList(callbacks)
        callbacks.set_model(self.original_model)
        callbacks.set_params({
            # 设置训练过程的参数
            "epochs": epochs,  # 总共的训练轮数
            "steps": steps_per_epoch,  # 每个 epoch 的步数
            "verbose": verbose,  # 详细程度
            "do_validation": len(dataset.test_data) > 0,  # 是否进行验证
            "metrics": self.metrics_names + [  # 评估指标
                "val_" + n for n in self.metrics_names
            ]
        })

        # 创建采样器
        sampler = self.sampler(dataset, batch_size, steps_per_epoch, epochs)

        # Start the training loop
        epoch = 0
        self.original_model.stop_training = False
        callbacks.on_train_begin()
        while epoch < epochs:
            # 在每个 epoch 开始时执行一些操作 ，
            # 它会遍历所有在 CallbackList 中注册的回调函数，并调用它们的 on_epoch_begin 方法。
            callbacks.on_epoch_begin(epoch)

            for step in range(steps_per_epoch):
                # epoch中的每一step都需要做如下循环
                batch_logs = {"batch": step, "size": batch_size}  # 存储当前批次（batch）训练过程中的相关信息。
                callbacks.on_batch_begin(step, batch_logs)

                # Importance sampling is done here
                idxs, (x, y), w = sampler.sample(batch_size)  # 返回被采样的样本
                # Train on the sampled data
                loss, metrics, scores = self.model.train_batch(x, y, w)
                # Update the sampler
                sampler.update(idxs, scores)

                values = map(lambda x: x.mean(), [loss] + metrics)
                for l, o in zip(self.metrics_names, values):
                    batch_logs[l] = o
                callbacks.on_batch_end(step, batch_logs)

                if on_scores is not None and hasattr(self, "_latest_scores"):
                    on_scores(
                        sampler,
                        self._latest_scores
                    )

                if on_sample is not None:
                    on_sample(
                        sampler,
                        self._latest_sample_event["idxs"],
                        self._latest_sample_event["w"],
                        self._latest_sample_event["predicted_scores"]
                    )

                if self.original_model.stop_training:
                    break

            # Evaluate now that an epoch passed
            epoch_logs = {}
            if len(dataset.test_data) > 0:
                val = self.model.evaluate(
                    *dataset.test_data[:],
                    batch_size=batch_size
                )
                epoch_logs = {
                    "val_" + l: o
                    for l, o in zip(self.metrics_names, val)
                }
            callbacks.on_epoch_end(epoch, epoch_logs)
            if self.original_model.stop_training:
                break
            epoch += 1
        callbacks.on_train_end()

        return self.history

    @property
    def metrics_names(self):
        def name(x):
            if isinstance(x, str):
                return x
            if hasattr(x, "name"):
                return x.name
            if hasattr(x, "__name__"):
                return x.__name__
            return str(x)

        metrics = self.original_model.metrics or []
        return (
                ["loss"] +
                [name(m) for m in metrics] +
                ["score"]
        )
以下是包装文件model_wrappers.py
import sys

from blinker import signal
from tensorflow.python.keras import backend as K
from tensorflow.python.keras import Input
from tensorflow.python.keras.layers import Layer, multiply
from tensorflow.python.keras import Model

import numpy as np
from tensorflow.python.keras.metrics import categorical_accuracy

from .layers import GradientNormLayer, LossLayer, MetricLayer
class OracleWrapper(ModelWrapper):
    AVG_LOSS = 0
    LOSS = 1
    WEIGHTED_LOSS = 2
    SCORE = 3
    METRIC0 = 4

    FUSED_ACTIVATION_WARNING = ("[WARNING]: The last layer has a fused "
                                "activation i.e. Dense(..., "
                                "activation=\"sigmoid\").\nIn order for the "
                                "preactivation to be automatically extracted "
                                "use a separate activation layer (see "
                                "examples).\n")

    def __init__(self, model, reweighting, score="loss", layer=None):
        self.reweighting = reweighting
        self.layer = self._gnorm_layer(model, layer)

        # Augment the model with reweighting, scoring etc
        # Save the new model and the training functions in member variables
        self._augment_model(model, score, reweighting)

    def _gnorm_layer(self, model, layer):
        # If we were given a layer then use it directly
        if isinstance(layer, Layer):
            return layer

        # If we were given a layer index extract the layer
        if isinstance(layer, int):
            return model.layers[layer]

        try:
            # Get the last or the previous to last layer depending on wether
            # the last has trainable weights
            skip_one = not bool(model.layers[-1].trainable_weights)
            last_layer = -2 if skip_one else -1

            # If the last layer has trainable weights that means that we cannot
            # automatically extract the preactivation tensor so we have to warn
            # them because they might be missing out or they might not even
            # have noticed
            if last_layer == -1:
                config = model.layers[-1].get_config()
                if config.get("activation", "linear") != "linear":
                    sys.stderr.write(self.FUSED_ACTIVATION_WARNING)

            return model.layers[last_layer]
        except:
            # In case of an error then probably we are not using the gnorm
            # importance
            return None

    def _augment_model(self, model, score, reweighting):
        print("模型摘要:")
        model.summary()
        print("输出形状:", model.output_shape)

        # 从模型中提取一些信息
        loss = model.loss
        optimizer = model.optimizer.__class__(**model.optimizer.get_config())
        output_shape = model.output_shape[1:]
        if isinstance(loss, str) and loss.startswith("sparse"):
            output_shape = output_shape[:-1] + (1,)

        # 确保某些内容看起来正常
        assert not isinstance(loss, list)

        # 我们需要创建两个额外的输入
        # 1. 目标
        # 2. 预测得分
        y_true = Input(shape=output_shape)
        pred_score = Input(shape=(reweighting.weight_size,))

        # 创建损失层和得分层
        loss_tensor = LossLayer(loss)([y_true, model.output])
        score_tensor = _get_scoring_layer(
            score,
            y_true,
            model.output,
            loss,
            self.layer,
            model
        )

        # 创建样本权重
        weights = reweighting.weight_layer()([score_tensor, pred_score])

        # 创建输出
        weighted_loss = weighted_loss_model = multiply([loss_tensor, weights])
        for l in model.losses:
            weighted_loss += l
        weighted_loss_mean = K.mean(weighted_loss)

        # 创建指标层
        metrics = model.metrics or []
        metrics = [
            MetricLayer(metric)([y_true, model.output])
            for metric in metrics
        ]

        # 创建一个用于绘图和访问 trainable_weights 等的模型
        new_model = Model(
            inputs=_tolist(model.input) + [y_true, pred_score],
            outputs=[weighted_loss_model]
        )

        # 构建单独的 on_batch keras 函数用于评分和训练
        updates = optimizer.get_updates(
            weighted_loss_mean,
            new_model.trainable_weights
        )
        metrics_updates = []
        if hasattr(model, "metrics_updates"):
            metrics_updates = model.metrics_updates

        # 移除学习阶段的处理
        # learning_phase = []
        # if weighted_loss_model._uses_learning_phase:
        #     learning_phase.append(K.learning_phase())

        # 更新输入，不包括学习阶段
        inputs = _tolist(model.input) + [y_true, pred_score]  # 已移除 + learning_phase
        outputs = [
                      weighted_loss_mean,
                      loss_tensor,
                      weighted_loss,
                      score_tensor
                  ] + metrics

        train_on_batch = K.function(
            inputs=inputs,
            outputs=outputs,
            updates=updates + model.updates + metrics_updates
        )
        evaluate_on_batch = K.function(
            inputs=inputs,
            outputs=outputs,
            updates=model.state_updates + metrics_updates
        )

        self.model = new_model
        self.optimizer = optimizer
        self.model.optimizer = optimizer
        self._train_on_batch = train_on_batch
        self._evaluate_on_batch = evaluate_on_batch

    def evaluate_batch(self, x, y):
        if len(y.shape) == 1:
            y = np.expand_dims(y, axis=1)
        dummy_weights = np.ones((y.shape[0], self.reweighting.weight_size))
        inputs = _tolist(x) + [y, dummy_weights] + [0]
        outputs = self._evaluate_on_batch(inputs)

        signal("is.evaluate_batch").send(outputs)

        return np.hstack([outputs[self.LOSS]] + outputs[self.METRIC0:])

    def score_batch(self, x, y):
        if len(y.shape) == 1:
            y = np.expand_dims(y, axis=1)
        dummy_weights = np.ones((y.shape[0], self.reweighting.weight_size))
        inputs = _tolist(x) + [y, dummy_weights] + [0]
        outputs = self._evaluate_on_batch(inputs)

        return outputs[self.SCORE].ravel()

    def train_batch(self, x, y, w):
        if len(y.shape) == 1:
            y = np.expand_dims(y, axis=1)

        # train on a single batch
        outputs = self._train_on_batch(_tolist(x) + [y, w, 1])

        # Add the outputs in a tuple to send to whoever is listening
        result = (
            outputs[self.WEIGHTED_LOSS],
            outputs[self.METRIC0:],
            outputs[self.SCORE]
        )
        signal("is.training").send(result)

        return result
def _tolist(x, acceptable_iterables=(list, tuple)):
    if not isinstance(x, acceptable_iterables):
        x = [x]
    return x


def _get_scoring_layer(score, y_true, y_pred, loss="categorical_crossentropy",
                       layer=None, model=None):
    """Get a scoring layer that computes the score for each pair of y_true,
    y_pred"""
    assert score in ["loss", "gnorm", "full_gnorm", "acc"]

    if score == "loss":
        return LossLayer(loss)([
            y_true,
            y_pred
        ])
    elif score == "gnorm":
        return GradientNormLayer(
            layer.output,
            loss,
            fast=False
        )([
            y_true,
            y_pred
        ])
    elif score == "full_gnorm":
        return GradientNormLayer(
            model.trainable_weights,
            loss,
            fast=False
        )([
            y_true,
            y_pred
        ])
    elif score == "acc":
        return LossLayer(categorical_accuracy)([
            y_true,
            y_pred
        ])


class ModelWrapper(object):
    """The goal of the ModelWrapper is to take a NN and add some extra layers
    that produce a score, a loss and the sample weights to perform importance
    sampling."""
    def _iterate_batches(self, x, y, batch_size):
        bs = batch_size
        for s in range(0, len(y), bs):
            yield [xi[s:s+bs] for xi in _tolist(x)], y[s:s+bs]

    def evaluate(self, x, y, batch_size=128):
        result = np.mean(
            np.vstack([
                self.evaluate_batch(xi, yi)
                for xi, yi in self._iterate_batches(x, y, batch_size)
            ]),
            axis=0
        )

        signal("is.evaluation").send(result)
        return result

    def score(self, x, y, batch_size=128):
        bs = batch_size
        result = np.hstack([
            self.score_batch(xi, yi).T
            for xi, yi in self._iterate_batches(x, y, batch_size)
        ]).T

        signal("is.score").send(result)
        return result

    def set_lr(self, lr):
        """Set the learning rate of the wrapped models.

        We try to set the learning rate on a member variable model and a member
        variable small. If we do not find a member variable model we raise a
        NotImplementedError
        """
        try:
            K.set_value(
                self.optimizer.lr,
                lr
            )
        except AttributeError:
            try:
                K.set_value(
                    self.model.optimizer.lr,
                    lr
                )
            except AttributeError:
                raise NotImplementedError()

        try:
            K.set_value(
                self.small.optimizer.lr,
                lr
            )
        except AttributeError:
            pass

    def evaluate_batch(self, x, y):
        raise NotImplementedError()

    def score_batch(self, x, y):
        raise NotImplementedError()

    def train_batch(self, x, y, w):
        raise NotImplementedError()


class ModelWrapperDecorator(ModelWrapper):
    def __init__(self, model_wrapper, implemented_attributes=set()):
        self.model_wrapper = model_wrapper
        self.implemented_attributes = (
            implemented_attributes | set(["model_wrapper"])
        )

    def __getattribute__(self, name):
        _getattr = object.__getattribute__
        implemented_attributes = _getattr(self, "implemented_attributes")
        if name in implemented_attributes:
            return _getattr(self, name)
        else:
            model_wrapper = _getattr(self, "model_wrapper")
            return getattr(model_wrapper, name)
以下是数据处理文件datasets.py
from bisect import bisect_left
from collections import  deque
import gzip
import os
from os import path
import pickle
import sys
from threading import Condition, Lock, Thread
import tensorflow as tf
class BaseDataset(object):
    class _DataProxy(object):
        def __init__(self, dataset, subset):
            self.data = getattr(dataset, "_%s_data" % (subset,))
            self.size = getattr(dataset, "_%s_size" % (subset,))

        def __getitem__(self, idxs):
            return self.data(idxs)

        def __len__(self):
            return self.size()

    def _train_data(self, idxs=slice(None)):
        """Return the training data in the form (x, y)"""
        raise NotImplementedError()

    def _train_size(self):
        """Training data length"""
        x, y = self._train_data()
        if isinstance(x, (list, tuple)):
            return len(x[0])
        else:
            return len(x)

    def _test_data(self, idxs=slice(None)):
        """Return the testing data in the form (x, y)"""
        raise NotImplementedError()

    def _test_size(self):
        """Test data length"""
        x, y = self._test_data()
        if isinstance(x, (list, tuple)):
            return len(x[0])
        else:
            return len(x)

    def _slice_data(self, x, y, idxs):
        if isinstance(x, (list, tuple)):
            return [xi[idxs] for xi in x], y[idxs]
        else:
            return x[idxs], y[idxs]

    def _extract_shape(self, x):
        if isinstance(x, (list, tuple)):
            return [xi.shape[1:] for xi in x]
        return x.shape[1:]

    @property
    def train_data(self):
        return self._DataProxy(self, "train")

    @property
    def test_data(self):
        return self._DataProxy(self, "test")

    @property
    def shape(self):
        """Return the shape of the samples"""
        raise NotImplementedError()

    @property
    def output_size(self):
        """Return the number of outputs"""
        raise NotImplementedError()

    @property
    def output_shape(self):
        """Return the shape of the output (it could differ for seq models for
        instance)."""
        return (self.output_size,)


class InMemoryDataset(BaseDataset):
    """A dataset that fits in memory and is simply 4 numpy arrays (x, y) *
    (train, test) where x can also be a list of numpy arrays"""
    def __init__(self, X_train, y_train, X_test, y_test, categorical=True):
        self._x_train = X_train
        self._x_test = X_test

        # are the targets to be made one hot vectors
        if categorical:
            self._y_train = np_utils.to_categorical(y_train)
            self._y_test = np_utils.to_categorical(y_test)
            self._output_size = self._y_train.shape[1]

        # handle sparse output classification
        elif issubclass(y_train.dtype.type, np.integer):
            self._y_train = y_train
            self._y_test = y_test
            self._output_size = self._y_train.max() + 1  # assume 0 based idxs

        # not classification, just copy them
        else:
            self._y_train = y_train
            self._y_test = y_test
            self._output_size = self._y_train.shape[1]

    def _train_data(self, idxs=slice(None)):
        return self._slice_data(self._x_train, self._y_train, idxs)

    def _test_data(self, idxs=slice(None)):
        return self._slice_data(self._x_test, self._y_test, idxs)

    @property
    def shape(self):
        return self._extract_shape(self._x_train)

    @property
    def output_size(self):
        return self._output_size

    @classmethod
    def from_loadable(cls, dataset):
        (a, b), (c, d) = dataset.load_data()
        return cls(a, b, c, d)
如上述代码所示,目前所用的模型是经典神经网络模型,我想要将其变换为量子神经网络模型,并且所有的训练逻辑不能发生改变,也就是在不改变训练逻辑的情况下去将经典神经网络更改为量子神经网络.你需要使用pennylane来构建量子神经网络,使用tensorflow来与pennylane共同搭建训练架构.如下是我的环境具体情况:python=3.9
absl-py                      2.1.0
appdirs                      1.4.4
array-record                 0.4.1
astunparse                   1.6.3
autograd                     1.7.0
blinker                      1.8.2
cached-property              1.5.2
cachetools                   5.5.0
certifi                      2024.8.30
charset-normalizer           3.3.2
click                        8.1.7
colorama                     0.4.6
contourpy                    1.3.0
cycler                       0.12.1
dill                         0.3.8
dm-tree                      0.1.8
docplex                      2.28.240
etils                        1.5.2
flatbuffers                  24.3.25
fonttools                    4.54.1
fsspec                       2024.9.0
gast                         0.4.0
google-auth                  2.35.0
google-auth-oauthlib         0.4.6
google-pasta                 0.2.0
grpcio                       1.66.1
h5py                         3.11.0
idna                         3.10
importlib_metadata           8.5.0
importlib_resources          6.4.5
joblib                       1.4.2
keras                        2.10.0
Keras-Preprocessing          1.1.2
kiwisolver                   1.4.7
libclang                     18.1.1
Markdown                     3.7
markdown-it-py               3.0.0
MarkupSafe                   2.1.5
matplotlib                   3.9.1
mdurl                        0.1.2
ml-dtypes                    0.3.2
mpmath                       1.3.0
munkres                      1.1.4
namex                        0.0.8
networkx                     3.2.1
numpy                        1.26.4
oauthlib                     3.2.2
olefile                      0.47
opt-einsum                   3.3.0
optree                       0.12.1
packaging                    24.1
pbr                          6.1.0
PennyLane                    0.15.1
Pillow                       8.4.0
pip                          24.2
promise                      2.3
protobuf                     3.20.3
psutil                       6.0.0
pyasn1                       0.6.1
pyasn1_modules               0.4.1
Pygments                     2.18.0
pyparsing                    3.1.4
PyQt5                        5.12.3
PyQt5_sip                    4.19.18
PyQtChart                    5.12
PyQtWebEngine                5.12.1
python-dateutil              2.9.0
qiskit                       1.0.2
qiskit-aer                   0.14.0.1
qiskit-algorithms            0.3.0
qiskit-optimization          0.6.1
requests                     2.32.3
requests-oauthlib            2.0.0
rich                         13.8.1
rsa                          4.9
rustworkx                    0.15.1
scikit-learn                 1.5.2
scipy                        1.13.1
semantic-version             2.6.0
setuptools                   75.1.0
six                          1.16.0
stevedore                    5.3.0
symengine                    0.11.0
sympy                        1.13.3
tensorboard                  2.10.1
tensorboard-data-server      0.6.1
tensorboard-plugin-wit       1.8.1
tensorflow                   2.10.0
tensorflow-datasets          4.9.3
tensorflow-estimator         2.10.0
tensorflow-io-gcs-filesystem 0.31.0
tensorflow-metadata          1.15.0
termcolor                    2.4.0
threadpoolctl                3.5.0
toml                         0.10.2
tornado                      6.4.1
tqdm                         4.66.5
typing_extensions            4.12.2
unicodedata2                 15.1.0
urllib3                      2.2.3
Werkzeug                     3.0.4
wheel                        0.35.1
wrapt                        1.16.0
zipp                         3.20.2