如何解决基于RSS数据训练DANN模型时出现的AttributeError: 'KerasTensor' object has no attribute '_id'报错

Hey there, let's break down what's causing that AttributeError: 'KerasTensor' object has no attribute '_id' error and fix your DANN implementation step by step.

Root Causes of the Error #

1. Misplaced Input Layer in __init__
   In Keras Model subclasses, the __init__ method should only define trainable layers (like Dense, Dropout). Defining an Input layer here creates a disconnected KerasTensor that conflicts with the input tensor you pass during model calls. You also redundantly redefine Input in the call method, which messes up tensor tracking.

2. Redundant Dataset Processing
   You're wrapping an existing Dataset object with tf.data.Dataset.from_tensor_slices again, which creates a nested data structure that doesn't play well with batching.

3. Undefined Variables in Training Loop
   Variables like ep, ls, ls_i, and loss_s_label aren't initialized, which would cause additional runtime errors even after fixing the tensor issue.

Fixed DANN Model Implementation #

Let's rewrite the model to follow Keras best practices for subclassing:

class DANN(Model):
    def __init__(self, num_features):
        super().__init__()
        # Wrap feature extractor and label predictor in Sequential layers for cleaner code
        self.feature_extractor = tf.keras.Sequential([
            Dense(100, activation='relu'),
            Dense(100, activation='relu'),
            Dense(100, activation='relu'),
            Dropout(0.5),
            Dense(100, activation='relu'),
            Dense(100, activation='relu')
        ])
        self.label_predictor = tf.keras.Sequential([
            Dense(100, activation='relu'),
            Dense(100, activation='relu'),
            Dense(2, activation=None)
        ])
        
    def call(self, x, training=False):
        # x is the input batch tensor passed during model calls - no need for Input here
        features = self.feature_extractor(x, training=training)
        label_pred = self.label_predictor(features, training=training)
        return label_pred

# Initialize model with your feature count
model = DANN(num_features=num_features)

Corrected Dataset Processing #

Remove the redundant from_tensor_slices call and directly batch your tensor datasets:

# Convert data to tensors
dx_source_tensor = tf.convert_to_tensor(X_train_source, dtype=tf.float32)
dy_source_tensor = tf.convert_to_tensor(Y_train_source, dtype=tf.float32)
dx_source_test_tensor = tf.convert_to_tensor(X_test_source, dtype=tf.float32)
dy_source_test_tensor = tf.convert_to_tensor(Y_test_source, dtype=tf.float32)

# Create and batch datasets directly
train_dataset = tf.data.Dataset.from_tensor_slices((dx_source_tensor, dy_source_tensor)).batch(num_batch)
test_dataset = tf.data.Dataset.from_tensor_slices((dx_source_test_tensor, dy_source_test_tensor)).batch(num_batch)

Fixed Training Loop #

Add missing variable initializations and clean up the progress tracking logic:

import sys

# Initialize training parameters
lr = 1e-3
optimizer = tf.optimizers.SGD(learning_rate=lr)
loss_fn_label = keras.losses.MeanSquaredError()  # Use class-based loss for batch handling
num_epochs = 10  # Define your epoch count here
max_batches = len(list(train_dataset))  # Get total number of batches
source_label_loss = []

for epoch in range(num_epochs):
    print("\nStart of epoch %d" % (epoch + 1,))
    total_loss = 0.0
    # Initialize progress bar variables
    progress_steps = [max_batches // 10 * i for i in range(1, 11)]  # Split progress into 10 segments
    progress_idx = 0
    current_step = 0
    
    for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):
        with tf.GradientTape() as tape:
            logits = model(x_batch_train, training=True)
            loss_value = loss_fn_label(y_batch_train, logits)
        
        # Update weights
        grads = tape.gradient(loss_value, model.trainable_weights)
        optimizer.apply_gradients(zip(grads, model.trainable_weights))
        
        total_loss += loss_value.numpy()
        # Update progress bar
        current_step += 1
        while progress_idx < len(progress_steps) and current_step == progress_steps[progress_idx]:
            sys.stdout.write("█")
            progress_idx += 1
        sys.stdout.write("|")
    
    avg_loss = total_loss / max_batches
    source_label_loss.append(avg_loss)
    print(f'\n\tSource label loss: {avg_loss:.4f}')

Additional Training Recommendations #

1. Use Keras' Built-in fit Method
   Manual GradientTape loops are flexible, but Keras' fit method is cleaner and includes built-in validation, callbacks, and progress tracking:

   model.compile(optimizer=optimizer, loss=loss_fn_label)
   history = model.fit(train_dataset, epochs=num_epochs, validation_data=test_dataset)
   

2. Add Domain Adversarial Components (DANN Core)
   Your current code only implements feature extraction and label prediction—DANN's key is domain adversarial training. Here's a quick addition of the domain discriminator and gradient reversal layer:

   class GradientReversalLayer(tf.keras.layers.Layer):
       def __init__(self, alpha=1.0, **kwargs):
           super().__init__(**kwargs)
           self.alpha = alpha
   
       def call(self, inputs):
           return -self.alpha * inputs
   
   class DANN(Model):
       def __init__(self, num_features, num_domains=2):
           super().__init__()
           self.feature_extractor = tf.keras.Sequential([
               Dense(100, activation='relu'),
               Dense(100, activation='relu'),
               Dense(100, activation='relu'),
               Dropout(0.5),
               Dense(100, activation='relu'),
               Dense(100, activation='relu')
           ])
           self.label_predictor = tf.keras.Sequential([
               Dense(100, activation='relu'),
               Dense(100, activation='relu'),
               Dense(2, activation=None)
           ])
           # Add domain discriminator with gradient reversal
           self.domain_discriminator = tf.keras.Sequential([
               GradientReversalLayer(),
               Dense(100, activation='relu'),
               Dense(100, activation='relu'),
               Dense(num_domains, activation='sigmoid')
           ])
           
       def call(self, x, training=False):
           features = self.feature_extractor(x, training=training)
           label_pred = self.label_predictor(features, training=training)
           domain_pred = self.domain_discriminator(features, training=training)
           return label_pred, domain_pred
   

   You'll need to adjust the training loop to compute both label loss and domain loss for adversarial training.

内容的提问来源于stack exchange，提问作者Abhiram C D