from future import print_function
import os
os.environ['TF_USE_LEGACY_KERAS'] = '1'
from tensorflow.python.keras.layers import Dense, Activation
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 pennylane as qml
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("训练模型于鸢尾花数据集")
args = parser.parse_args()


# 加载并预处理鸢尾花数据集
iris = datasets.load_iris()
x_data = iris.data
y_data = iris.target.reshape(-1, 1)

# 对标签进行独热编码
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类

# 定义量子电路和量子层
n_qubits = 4  # 量子比特数（与特征数相同）
n_layers = 2  # 变分量子电路的层数

dev = qml.device('default.qubit', wires=n_qubits)

# 定义量子节点 (QNode)
def quantum_circuit(inputs, weights):
    # 数据编码（角度编码）
    for i in range(n_qubits):
        qml.RY(inputs[i], wires=i)
    # 变分层
    qml.templates.StronglyEntanglingLayers(weights, wires=range(n_qubits))
    # 测量
    return [qml.expval(qml.PauliZ(i)) for i in range(num_classes)]  # 输出与类别数匹配

weight_shapes = {"weights": (n_layers, n_qubits, 3)}

# 将量子电路封装为 Keras 层
qlayer = qml.qnn.KerasLayer(quantum_circuit, weight_shapes, output_dim=num_classes)

# 使用 Functional API 构建模型
inputs = Input(shape=(4,))
x = qlayer(inputs)
outputs = Activation('softmax')(x)
model = Model(inputs=inputs, outputs=outputs)

# 编译模型
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('测试损失:', score[0])
print('测试准确率:', score[1])

# 计算并显示准确率波动
mean_accuracy, fluctuation = calculate_accuracy_fluctuation(history)
print('平均验证准确率:', mean_accuracy)
print('准确率波动:', fluctuation)

# 绘制训练历史
plot_history(history)

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

# 绘制 ROC 曲线
plot_roc_curve(y_test, y_pred)
上述代码是yuanweihua.py文件
import sys
import numpy as np
import logging
from collections.abc import Iterable
import tensorflow as tf
from tensorflow.python.keras.callbacks import BaseLogger, CallbackList, History, ProgbarLogger
from tensorflow.python.keras.layers import Dropout
#from tensorflow.python.keras.utils import Sequence
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 _BaseImportanceTraining(object):
def init(self, model, score="gnorm", layer=None):
"""重要性采样训练的基类。


    参数：
        model: 要训练的 Keras 模型。
    """
    # 检查模型是否兼容，并包装模型以支持重要性采样
    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):
    """检查模型是否使用了 Dropout 和 BatchNorm，并警告可能影响重要性计算。"""
    if any(
            isinstance(l, (tf.keras.layers.BatchNormalization, Dropout))
            for l in model.layers
    ):
        sys.stderr.write(("[NOTICE]: 您正在使用 BatchNormalization 和/或 Dropout。\n"
                          "这些层可能会影响重要性计算，建议您用 LayerNormalization 或 "
                          "在测试模式下的 BatchNormalization 以及 L2 正则化替代它们。\n"))

@property
def reweighting(self):
    """控制重要性采样时偏置的重加权策略。"""
    raise NotImplementedError()

def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
    """为给定的数据集创建一个新的采样器，以使用重要性采样进行采样。"""
    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):
    """使用给定的数据创建一个 InMemoryDataset 实例，并使用重要性采样训练模型。

    参数：
        x: 训练数据的 Numpy 数组或 Numpy 数组列表。
        y: 目标数据的 Numpy 数组。
        batch_size: 每个梯度更新的样本数。
        epochs: 遍历整个训练集的次数。
        verbose: {0, >0}，是否使用 Keras 的进度条回调。
        callbacks: 在训练期间调用的 Keras 回调列表。
        validation_split: [0, 1) 之间的浮点数，用于验证的数据百分比。
        validation_data: tuple，(x_val, y_val)，用于模型验证的数据。
        steps_per_epoch: int 或 None，每个 epoch 的步数。
        on_sample: 接受 sampler, idxs, w, scores 的回调函数。
        on_scores: 接受 sampler 和 scores 的回调函数。
    返回：
        包含训练期间收集的信息的 Keras `History` 对象。
    """
    # 处理验证数据集
    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, "验证集比例必须小于 1"
        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, x.shape[1])), np.empty(shape=(0, y.shape[1]))

    # 创建数据集
    dataset = InMemoryDataset(
        x_train,
        y_train,
        x_test,
        y_test,
        categorical=False  # 不需要转换标签
    )

    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):
    """创建一个 GeneratorDataset 实例，并使用重要性采样训练模型。

    参数：
        train: 返回 (inputs, targets) 的生成器或可迭代对象。
        steps_per_epoch: 每个 epoch 的梯度更新次数。
        batch_size: 每个梯度更新的样本数。
        epochs: 训练轮数。
        verbose: {0, >0}，是否使用 Keras 的进度条回调。
        validation_data: 生成器、可迭代对象或 (inputs, targets) 元组。
        validation_steps: 如果 validation_data 是生成器，则需要指定此参数。
        on_sample: 接受 sampler, idxs, w, scores 的回调函数。
        on_scores: 接受 sampler 和 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("未提供验证数据")

    # 设置训练数据和每个 epoch 的步数
    if isinstance(train, Iterable):
        if steps_per_epoch is None:
            try:
                steps_per_epoch = len(train)
            except TypeError:
                raise ValueError("使用没有 __len__ 方法的生成器时必须设置 steps_per_epoch")
        train = infinite_iterator(train)
    elif steps_per_epoch is None:
        raise ValueError("steps_per_epoch 未设置")

    dataset = GeneratorDataset(train, test, test_len)

    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):
    """在给定的数据集上训练模型。

    参数：
        dataset: `BaseDataset` 的实例，提供要训练的数据。
        steps_per_epoch: int，每个 epoch 的梯度更新次数。
        batch_size: 每个梯度更新的样本数。
        epochs: 训练轮数。
        verbose: {0, >0}，是否使用 Keras 的进度条回调。
        callbacks: Keras 回调列表。
        on_sample: 接受 sampler, idxs, w, scores 的回调函数。
        on_scores: 接受 sampler 和 scores 的回调函数。
    """
    try:
        if len(dataset.train_data) < batch_size:
            raise ValueError("batch_size 大于训练集大小")
    except RuntimeError as e:
        assert "no size" in str(e)

    # 设置 steps_per_epoch
    if steps_per_epoch is None:
        steps_per_epoch = len(dataset.train_data) // batch_size

    # 创建回调列表
    self.history = History()
    callbacks = [BaseLogger()] + (callbacks or []) + [self.history]
    if verbose > 0:
        callbacks += [ProgbarLogger(count_mode="steps")]
    callbacks = CallbackList(callbacks)
    callbacks.set_model(self.original_model)
    callbacks.set_params({
        "epochs": epochs,
        "steps": steps_per_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)

    # 开始训练循环
    epoch = 0
    self.original_model.stop_training = False
    callbacks.on_train_begin()
    while epoch < epochs:
        callbacks.on_epoch_begin(epoch)

        for step in range(steps_per_epoch):
            batch_logs = {"batch": step, "size": batch_size}
            callbacks.on_batch_begin(step, batch_logs)

            # 在这里进行重要性采样
            idxs, (x_batch, y_batch), w = sampler.sample(batch_size)

            # 将 x_batch 和 y_batch 转换为 TensorFlow 张量
            x_batch = tf.convert_to_tensor(x_batch)
            y_batch = tf.convert_to_tensor(y_batch)
            w = tf.convert_to_tensor(w, dtype=tf.float32)

            # 自定义训练步骤
            loss_value, metrics_results, scores = self.train_batch(x_batch, y_batch, w)

            # 更新采样器
            sampler.update(idxs, scores)

            # 记录指标
            batch_logs['loss'] = loss_value
            for metric_name, metric_value in zip(self.metrics_names[1:], metrics_results):
                batch_logs[metric_name] = metric_value
            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

        # 在每个 epoch 结束时进行评估
        epoch_logs = {}
        if len(dataset.test_data) > 0:
            val_loss, val_metrics = self.evaluate(dataset.test_data)
            epoch_logs['val_loss'] = val_loss
            for name, value in zip(self.metrics_names[1:], val_metrics):
                epoch_logs['val_' + name] = value
        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]
    )

def train_batch(self, x, y, w):
    """使用自定义训练步骤进行训练。"""
    with tf.GradientTape() as tape:
        y_pred = self.model(x, training=True)
        loss_value = self.model.compiled_loss(
            y,
            y_pred,
            sample_weight=w,
            regularization_losses=self.model.losses
        )

    gradients = tape.gradient(loss_value, self.model.trainable_variables)
    self.model.optimizer.apply_gradients(zip(gradients, self.model.trainable_variables))

    # 更新指标
    self.model.compiled_metrics.update_state(y, y_pred, sample_weight=w)
    metrics_results = [m.result().numpy() for m in self.model.metrics]

    # 返回损失值和指标结果
    return loss_value.numpy(), metrics_results, None

def evaluate(self, test_data):
    """评估模型在测试集上的性能。"""
    x_test, y_test = test_data[:]
    y_pred = self.model.predict(x_test, batch_size=128)
    loss_value = self.model.compiled_loss(y_test, y_pred, regularization_losses=self.model.losses)

    # 更新指标
    self.model.compiled_metrics.update_state(y_test, y_pred)
    metrics_results = [m.result().numpy() for m in self.model.metrics]

    return loss_value.numpy(), metrics_results
class _UnbiasedImportanceTraining(_BaseImportanceTraining):
def init(self, model, score="gnorm", layer=None):
self._reweighting = BiasedReweightingPolicy(1.0) # 无偏


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

@property
def reweighting(self):
    return self._reweighting
class ConstantTimeImportanceTraining(_UnbiasedImportanceTraining):
"""使用常时间重要性采样训练模型。


参数：
    model: 要训练的 Keras 模型。
    score: {"gnorm", "loss", "full_gnorm"}，用于重要性采样的重要性度量。
    layer: None 或 int 或 Layer，计算 gnorm 时使用的层。
"""

def __init__(self, model, backward_pass_multiplier=2.0, extra_samples=0.1,
             score="gnorm", layer=None):
    self._backward_pass_multiplier = backward_pass_multiplier
    self._extra_samples = extra_samples

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

def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
    return ConstantTimeSampler(
        dataset,
        self.reweighting,
        self.model,
        backward_pass_multiplier=self._backward_pass_multiplier,
        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"]
)
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 ConstantTimeImportanceTraining(_UnbiasedImportanceTraining):
"""主要目的是通过保持每次迭代的时间常量的同时，提高梯度的质量来加速模型训练。
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, score="gnorm", layer=None):
self._backward_time = backward_time
super(ConstantTimeImportanceTraining, self).init(
model,
score,
layer
)
def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
return ConstantTimeSampler(
dataset,
self.reweighting,
self.model,
backward_time=self._backward_time
)
class BiasedImportanceTraining(_BaseImportanceTraining):
"""Train a model with exact importance sampling using the loss as
importance.


Arguments
---------
    model: The Keras model to train
    k: float in (-oo, 1], controls the bias of the training that focuses
       the network on the hard examples (see paper)
    smooth: float, influences the sampling distribution towards uniform by
            adding `smooth` to all probabilities and renormalizing
    adaptive_smoothing: bool, If True the `smooth` argument is a percentage
                        of the average training loss
    presample: int, the number of samples to presample for scoring
    forward_batch_size: int, the batch size to use for the forward pass
                        during scoring
"""

def __init__(self, model, k=0.5, smooth=0.0, adaptive_smoothing=False,
             presample=256, forward_batch_size=128):
    # Create the reweighting policy
    self._reweighting = BiasedReweightingPolicy(k)

    # Call the parent to wrap the model
    super(BiasedImportanceTraining, self).__init__(model)

    # Create the sampler factory, the workhorse of the whole deal :-)
    adaptive_smoothing_factory = partial(
        AdaptiveAdditiveSmoothingSampler, ___, smooth
    )
    additive_smoothing_factory = partial(
        AdditiveSmoothingSampler, ___, smooth
    )
    self._sampler = partial(
        ModelSampler,
        ___,
        self.reweighting,
        self.model,
        large_batch=presample,
        forward_batch_size=forward_batch_size
    )
    if adaptive_smoothing and smooth > 0:
        self._sampler = compose(adaptive_smoothing_factory, self._sampler)
    elif smooth > 0:
        self._sampler = compose(additive_smoothing_factory, self._sampler)

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

def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
    return self._sampler(dataset)
class ApproximateImportanceTraining(_BaseImportanceTraining):
"""Train a model with importance sampling using an LSTM with a class
embedding to predict the importance of the training samples.


Arguments
---------
    model: The Keras model to train
    k: float in (-oo, 1], controls the bias of the training that focuses
       the network on the hard examples (see paper)
    smooth: float, influences the sampling distribution towards uniform by
            adding `smooth` to all probabilities and renormalizing
    adaptive_smoothing: bool, If True the `smooth` argument is a percentage
                        of the average training loss
    presample: int, the number of samples to presample for scoring
"""

def __init__(self, model, k=0.5, smooth=0.0, adaptive_smoothing=False,
             presample=2048):
    # Create the reweighting policy
    self._reweighting = BiasedReweightingPolicy(k)

    # Call the parent to wrap the model
    super(ApproximateImportanceTraining, self).__init__(model)

    # Create the sampler factory, the workhorse of the whole deal :-)
    adaptive_smoothing_factory = partial(
        AdaptiveAdditiveSmoothingSampler, ___, smooth
    )
    additive_smoothing_factory = partial(
        AdditiveSmoothingSampler, ___, smooth
    )
    self._sampler = partial(
        LSTMSampler,
        ___,
        self.reweighting,
        presample=presample
    )
    if adaptive_smoothing and smooth > 0:
        self._sampler = compose(adaptive_smoothing_factory, self._sampler)
    elif smooth > 0:
        self._sampler = compose(additive_smoothing_factory, self._sampler)

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

def sampler(self, dataset, batch_size, steps_per_epoch, epochs):
    return self._sampler(dataset)

def fit_generator(*args, **kwargs):
    raise NotImplementedError("ApproximateImportanceTraining doesn't "
                              "support generator training")
上述是training.py文件
import sys
import numpy as np
from blinker import signal
from tensorflow.python.keras import backend as K
from tensorflow.python.keras.layers import Layer, multiply
from tensorflow.python.keras import Model, Input
from tensorflow.python.keras.metrics import categorical_accuracy

import tensorflow as tf

from .layers import GradientNormLayer, LossLayer, MetricLayer

class ModelWrapper(object):
"""ModelWrapper 的目的是包装一个神经网络，并添加一些额外的层，以生成得分、损失和样本权重，以执行重要性采样。"""


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):
    """设置模型的学习率。"""

    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 OracleWrapper(ModelWrapper):
AVG_LOSS = 0
LOSS = 1
WEIGHTED_LOSS = 2
SCORE = 3
METRIC0 = 4


FUSED_ACTIVATION_WARNING = ("[WARNING]: 最后一层具有融合激活函数，例如 Dense(..., activation=\"sigmoid\")。\n"
                            "为了自动提取预激活张量，请使用单独的激活层（参见示例）。\n")

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

    # 增强模型，添加重加权、得分等功能
    self._augment_model(model, score, reweighting)

def _gnorm_layer(self, model, layer):
    # 如果提供了层，则直接使用
    if isinstance(layer, Layer):
        return layer

    # 如果提供了层的索引，则提取该层
    if isinstance(layer, int):
        return model.layers[layer]

    try:
        # 获取倒数第一层或倒数第二层，取决于最后一层是否有可训练权重
        skip_one = not bool(model.layers[-1].trainable_weights)
        last_layer = -2 if skip_one else -1

        # 如果最后一层有可训练权重，说明无法自动提取预激活张量，需要警告
        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:
        # 如果出错，可能是因为我们没有使用 gnorm importance
        return None

def _augment_model(self, model, score, reweighting):
    # 提取模型的一些信息
    loss_fn = model.loss
    optimizer = model.optimizer.__class__.from_config(model.optimizer.get_config())
    output_shape = model.output_shape[1:]
    if isinstance(loss_fn, str) and loss_fn.startswith("sparse"):
        output_shape = output_shape[:-1] + (1,)

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

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

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

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

    # 创建加权损失
    weighted_loss = multiply([loss_tensor, weights])
    for l in model.losses:
        weighted_loss += l
    weighted_loss_mean = K.mean(weighted_loss)

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

    # 创建一个新模型，包括额外的输入和加权损失
    inputs = _tolist(model.input) + [y_true, pred_score]
    outputs = [weighted_loss]
    self.model = Model(inputs=inputs, outputs=outputs)

    # 编译新模型
    self.model.compile(
        optimizer=optimizer,
        loss=lambda yt, yp: weighted_loss_mean,  # 自定义损失
        metrics=metrics
    )

    # 保存原始模型的优化器和损失函数
    self.optimizer = optimizer
    self.loss_fn = loss_fn

def evaluate_batch(self, x, y):
    # 在评估时，我们不需要样本权重，因此创建一个全为 1 的权重向量
    if len(y.shape) == 1:
        y = np.expand_dims(y, axis=1)
    batch_size = y.shape[0]
    dummy_weights = np.ones((batch_size, self.reweighting.weight_size))
    inputs = _tolist(x) + [y, dummy_weights]
    outputs = self.model.evaluate(
        x=inputs,
        y=np.zeros((batch_size,)),  # 占位符，因为我们的模型输出是损失
        verbose=0,
        return_dict=True
    )

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

    # 返回损失和其他指标
    return [outputs['loss']] + [outputs[m] for m in self.model.metrics_names[1:]]

def score_batch(self, x, y):
    # 计算得分
    if len(y.shape) == 1:
        y = np.expand_dims(y, axis=1)
    batch_size = y.shape[0]
    dummy_weights = np.ones((batch_size, self.reweighting.weight_size))
    inputs = _tolist(x) + [y, dummy_weights]
    # 获取得分张量的输出
    get_score = K.function(
        inputs=self.model.inputs,
        outputs=self.model.get_layer('score_layer').output
    )
    scores = get_score(inputs)
    return scores.ravel()

def train_batch(self, x, y, w):
    # 训练单个批次
    if len(y.shape) == 1:
        y = np.expand_dims(y, axis=1)
    batch_size = y.shape[0]
    inputs = _tolist(x) + [y, w]
    outputs = self.model.train_on_batch(
        x=inputs,
        y=np.zeros((batch_size,))  # 占位符，因为我们的模型输出是损失
    )

    # 获取损失和指标
    if isinstance(outputs, list):
        loss = outputs[0]
        metrics = outputs[1:]
    else:
        loss = outputs
        metrics = []

    # 暂时无法获取得分，因此返回 None
    scores = None

    # 返回损失、指标和得分
    result = (
        loss,
        metrics,
        scores
    )
    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_fn="categorical_crossentropy",
layer=None, model=None):
"""获取一个计算每个 y_true 和 y_pred 对的得分的层"""
assert score in ["loss", "gnorm", "full_gnorm", "acc"]


if score == "loss":
    return LossLayer(loss_fn, name='score_layer')([
        y_true,
        y_pred
    ])
elif score == "gnorm":
    return GradientNormLayer(
        layer.output,
        loss_fn,
        fast=False,
        name='score_layer'
    )([
        y_true,
        y_pred
    ])
elif score == "full_gnorm":
    return GradientNormLayer(
        model.trainable_weights,
        loss_fn,
        fast=False,
        name='score_layer'
    )([
        y_true,
        y_pred
    ])
elif score == "acc":
    return MetricLayer(categorical_accuracy, name='score_layer')([
        y_true,
        y_pred
    ])
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)
上述包装类model_wrappers.py文件,
import numpy as np
import tensorflow as tf

class BaseDataset(object):
class DataProxy(object):
def init(self, dataset, subset):
self.data = getattr(dataset, f"{subset}data")
self.size = getattr(dataset, f"{subset}_size")


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

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

def _train_data(self, idxs=slice(None)):
    """返回训练数据，形式为 (x, y)"""
    raise NotImplementedError()

def _train_size(self):
    """返回训练数据的大小"""
    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)):
    """返回测试数据，形式为 (x, y)"""
    raise NotImplementedError()

def _test_size(self):
    """返回测试数据的大小"""
    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):
    """返回样本的形状"""
    raise NotImplementedError()

@property
def output_size(self):
    """返回输出的大小"""
    raise NotImplementedError()

@property
def output_shape(self):
    """返回输出的形状（对于序列模型等可能不同）"""
    return (self.output_size,)
class InMemoryDataset(BaseDataset):
"""一个适合存储在内存中的数据集，包括四个 numpy 数组 (x_train, y_train, x_test, y_test)，其中 x 可以是 numpy 数组的列表"""


def __init__(self, X_train, y_train, X_test, y_test, categorical=True):
    self._x_train = X_train
    self._x_test = X_test

    # 判断是否需要将目标值转换为 one-hot 编码
    if categorical:
        self._y_train = tf.keras.utils.to_categorical(y_train)
        self._y_test = tf.keras.utils.to_categorical(y_test)
        self._output_size = self._y_train.shape[1]

    # 处理稀疏输出分类
    elif issubclass(y_train.dtype.type, np.integer):
        self._y_train = y_train
        self._y_test = y_test
        self._output_size = np.max(self._y_train) + 1  # 假设类别索引从 0 开始

    # 非分类任务，直接复制标签
    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):
    (x_train, y_train), (x_test, y_test) = dataset.load_data()
    return cls(x_train, y_train, x_test, y_test)
上述是数据处理datasets.py文件,在使用 datasets.InMemoryDataset 时，可能需要在数据加载或批处理时对输入数据进行相应的预处理,我需要你对其进行完善