Почему моя модель Multi-Task Keras показывает высокую точность обучения L1, но низкая точность теста L1 с низкой потерей ⇐ Python

[Anonymous]

Я обучаю многозадачную модель глубокого обучения с использованием Tensorflow/Keras для классификации автомобильных изображений по двум целям:
L2 Task: Binary classification — Car vs. No Car

L1 Task: Binary classification — among car images only, distinguish between Car Interior vs. Car Exterior
< /code>
Я использую MobilenetV2 в качестве замороженной основы и разветвляюсь на две выходные головки: < /p>
L2 Head: Branches from an intermediate layer (block_11_add)

L1 Head: Branches from the final output of the backbone
Both heads use sigmoid activations with binary_crossentropy.
< /code>
Функция потери < /p>
L2 loss: Standard binary crossentropy on labels 0 (No Car) vs. 1 (Car)

L1 loss: Custom binary crossentropy on labels 1 (Interior) and 2 (Exterior), mapped as:

L1 loss is computed only on images where label > 0 (i.e., car images)
My dataset is balanced with augmentation containing 36k = 3*12k images, and 3k 3k validation and test folders, here is my dataset stucture
Dataset/
├── train/
│ ├ack no_car /
│ ├acke car_Interior /
└ └мобил car_xed /
├мобильный /val /
│ ├мобильный No_Car/
├── Car_Interior/
└── Car_Exterior/
During training and validation:

Epoch 1: l1_accuracy: 0.96 l1_loss: 0.11 val_l1_accuracy: 0.78
val_l1_loss: 2.78
blockquote < /p>
< /blockquote>
Но во время оценки тестирования (тот же код, что и .evaluate ()): < /p>

l1_accuracy: 0,34 l1_loss: 1,21 < /p>
< /rackquote>

low testnation> 2 -й тесто. Точность чрезвычайно плохая (падает с ~ 96% до 34%), даже если задача L2 является последовательной и высокопроизводительной.Class names are explicitly passed during dataset creation

Preprocessing is consistent across datasets (all use preprocess_input)

L1 loss and accuracy are computed only for samples where label > 0
< /code>
Почему тест L1 Точность низкая, несмотря на относительно низкую потерю?# Import modules from our package
from src.data_loader import create_dataset
from src.model import build_multitask_mobilenet_model
from src.trainer import MultiTaskModelTrainer, l1_loss_fn

# tf.config.set_visible_devices([], 'GPU')
# --- Configuration and Hyperparameters ---
IMG_HEIGHT = 224
IMG_WIDTH = 224
BATCH_SIZE = 32
NUM_EPOCHS = 1 # Increased epochs for better observation of callbacks
LEARNING_RATE = 0.001

# Dataset path (adjust this to your actual data)
DATA_DIR = 'Dataset'

# Loss weights (will be discussed below)
WEIGHT_L2 = 1.0 # Car vs. No Car
WEIGHT_L1 = 1.0 # Interior vs. Exterior

LOG_DIR = "logs/fit/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
CHECKPOINT_DIR = "checkpoints"
os.makedirs(CHECKPOINT_DIR, exist_ok=True)

import matplotlib.pyplot as plt

def main():
# --- Data Preparation ---
if not os.path.exists(DATA_DIR) or not os.listdir(DATA_DIR):
print("DATA_DIR not found or empty. Please ensure the dataset is available at the specified path.")
else:
print(f"Using dataset from: {DATA_DIR}")

train_ds, val_ds, test_ds = create_dataset(DATA_DIR, BATCH_SIZE)

print(f"Train dataset: {type(train_ds)} batches")

# --- Build the Multi-Task Model ---
multitask_model = build_multitask_mobilenet_model()
multitask_model.summary()

# --- Instantiate and Compile the Custom Trainer ---
trainer = MultiTaskModelTrainer(
model=multitask_model,
weight_l1=WEIGHT_L1,
weight_l2=WEIGHT_L2
)

optimizer = keras.optimizers.Adam(learning_rate=LEARNING_RATE)

trainer.compile(
optimizer=optimizer,
l1_loss=l1_loss_fn,
l2_loss=tf.keras.losses.BinaryCrossentropy()
)

# --- Callbacks ---
# 1. Early Stopping with Best Weight Restoration
early_stopping = keras.callbacks.EarlyStopping(
monitor='val_total_loss', # Monitor validation total loss
patience=1, # Number of epochs with no improvement after which training will be stopped
restore_best_weights=True, # Crucial: restore model weights from the epoch with the best monitored quantity
mode='min' # We want to minimize the loss
)

# 2. TensorBoard Callback
tensorboard_callback = keras.callbacks.TensorBoard(log_dir='./logs/fit/', histogram_freq=1)

# 3. Model Checkpointing (optional, but good practice to save best model)
model_checkpoint = keras.callbacks.ModelCheckpoint(
filepath=os.path.join(CHECKPOINT_DIR, 'best_model.h5'),
monitor='val_total_loss',
save_best_only=True,
verbose=2
)

print("\nStarting training...")
history = trainer.fit(
train_ds,
epochs=NUM_EPOCHS,
validation_data=val_ds,
callbacks=[early_stopping, model_checkpoint, tensorboard_callback],
verbose=1
)

print("\nTraining complete.")

trainer.model.save("final_multitask_model.h5")
print("\nModel saved as 'final_multitask_model.h5'.")

print("\nEvaluating on validation set:")
eval_results = trainer.evaluate(test_ds)
< /code>
Вот модель.py < /p>
build_multitask_mobilenet_model():
base_model = keras.applications.MobileNetV2(
input_shape=(IMG_HEIGHT, IMG_WIDTH, 3),
include_top=False,
weights='imagenet',
)
base_model.trainable = False # Freeze the base model layers

split_layer_name = 'block_11_add'

# Ensure the layer exists
try:
l2_branch_output_tensor = base_model.get_layer(split_layer_name).output
except ValueError:
print(f"Warning: Layer '{split_layer_name}' not found. Using base_model.output for L2 branch.")
print("Please inspect MobileNetV2 architecture summary to find a suitable earlier layer name.")
l2_branch_output_tensor = base_model.output # Fallback to endso of base model

trainable = False
for layer in base_model.layers:
if layer.name == 'block_11_expand':
trainable = True
layer.trainable = trainable

# --- Shared Backbone Input ---
inputs = base_model.input

# --- L2 Head (Car vs. No Car) ---
x_l2 = layers.GlobalAveragePooling2D()(l2_branch_output_tensor)
x_l2 = layers.Dense(128, activation='relu')(x_l2)
# x_l2 = layers.Dropout(0.3)(x_l2)
output_l2 = layers.Dense(1, activation='sigmoid', name='car_vs_nocar_output')(x_l2)

# --- L1 Head (Car Interior vs. Car Exterior) ---
x_l1 = layers.GlobalAveragePooling2D()(base_model.output) # Branches from the end of the base model
x_l1 = layers.Dense(128, activation='relu')(x_l1)
x_l1 = layers.Dropout(0.3)(x_l1)
output_l1 = layers.Dense(1, activation='sigmoid', name='interior_vs_exterior_output')(x_l1)

model = keras.Model(inputs=inputs, outputs=[output_l2, output_l1], name='MultiTask_CarClassifier')
< /code>
Вот data_loader.py < /p>
create_dataset(DATA_DIR, BATCH_SIZE=32):
train_ds = tf.keras.utils.image_dataset_from_directory(
directory=f"{DATA_DIR}/train",
labels="inferred",
label_mode="int",
image_size=IMG_SIZE,
batch_size=BATCH_SIZE,
shuffle=True,
class_names=desired_class_names
)

val_ds = tf.keras.utils.image_dataset_from_directory(
directory=f"{DATA_DIR}/val",
labels="inferred",
label_mode="int",
image_size=IMG_SIZE,
batch_size=BATCH_SIZE,
shuffle=True,
class_names=desired_class_names
)

test_ds = tf.keras.utils.image_dataset_from_directory(
directory=f"{DATA_DIR}/test",
labels="inferred",
label_mode="int",
image_size=IMG_SIZE,
batch_size=BATCH_SIZE,
shuffle=False,
class_names=desired_class_names
)
AUTOTUNE = tf.data.AUTOTUNE

def preprocess(image, label):
image = preprocess_input(image) # Normalize to [-1, 1]
return image, label

train_ds = train_ds.map(preprocess, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE)
val_ds = val_ds.map(preprocess, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE)
test_ds = test_ds.map(preprocess, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE)
# print(f"Loaded datasets: {type(train_ds)} train batches, {type(val_ds)} validation batches, {type(test_ds)} test batches")
< /code>
и, наконец, trainer.py < /p>
def l1_loss_fn(y_true, y_pred):
# y_true will contain 1s and 2s for Interior/Exterior. Map them to 0/1 for binary_crossentropy.
y_true_binary = tf.cast(y_true - 1, tf.float32) # 1 -> 0, 2 -> 1
return tf.keras.losses.binary_crossentropy(y_true_binary, tf.squeeze(y_pred, axis=-1))

class MultiTaskModelTrainer(keras.Model):
[enter image description here][1]def __init__(self, model, weight_l1=1.0, weight_l2=1.0):
super().__init__()
self.model = model
self.weight_l1 = weight_l1
self.weight_l2 = weight_l2

# Metrics for TensorBoard
self.loss_tracker_l1 = keras.metrics.Mean(name="l1_loss")
self.loss_tracker_l2 = keras.metrics.Mean(name="l2_loss")
self.total_loss_tracker = keras.metrics.Mean(name="total_loss")

# You might also want accuracy metrics for each head
self.acc_l1 = keras.metrics.BinaryAccuracy(name='l1_accuracy')
self.acc_l2 = keras.metrics.BinaryAccuracy(name='l2_accuracy')

def compile(self, optimizer, l1_loss, l2_loss):
super().compile()
self.optimizer = optimizer
self.l1_loss = l1_loss # This should be the custom l1_loss_fn
self.l2_loss = l2_loss # This will be standard binary_crossentropy

def train_step(self, data):
images, labels = data # labels are 0 for No_Car, 1 for Car_Interior, 2 for Car_Exterior

# Prepare ground truth labels for each task
# L2: Car vs. No Car (0 for No_Car, 1 for Car_Interior/Exterior)
gt_l2 = tf.cast(labels > 0, tf.float32) # True (1.0) if label is 1 or 2, False (0.0) if label is 0

# L1: Interior vs. Exterior (only for car images)
car_images_mask = (labels > 0) # Boolean mask for samples that are cars

with tf.GradientTape() as tape:
pred_l2, pred_l1 = self.model(images, training=True)
# Calculate L2 loss (Car vs. No Car)
loss_l2_value = self.l2_loss(gt_l2, pred_l2)

# Calculate L1 loss (Interior vs. Exterior) conditionally
def compute_l1_loss():
pred_l1_masked = tf.boolean_mask(pred_l1, car_images_mask)
gt_l1_masked = tf.boolean_mask(labels, car_images_mask)
return self.l1_loss(gt_l1_masked, pred_l1_masked)

def zero_l1_loss():
return tf.constant(0.0, dtype=tf.float32)

# tf.cond evaluates one branch at graph runtime
loss_l1_value = tf.cond(
tf.reduce_any(car_images_mask),
true_fn=compute_l1_loss,
false_fn=zero_l1_loss
)

# Combine losses with weights
total_loss = (loss_l2_value * self.weight_l2) + (loss_l1_value * self.weight_l1)

# Compute gradients and apply them
grads = tape.gradient(total_loss, self.model.trainable_variables)
self.optimizer.apply_gradients(zip(grads, self.model.trainable_variables))

# Update metrics
self.loss_tracker_l1.update_state(loss_l1_value)
self.loss_tracker_l2.update_state(loss_l2_value)
self.total_loss_tracker.update_state(total_loss)

# Update accuracy metrics
self.acc_l2.update_state(gt_l2, pred_l2)
def update_l1_acc():
pred_l1_masked = tf.boolean_mask(pred_l1, car_images_mask)
gt_l1_masked = tf.boolean_mask(labels, car_images_mask)
gt_l1_masked_binary = tf.cast(gt_l1_masked - 1, tf.float32)

self.acc_l1.update_state(gt_l1_masked_binary, pred_l1_masked)
return 0 # Dummy op required by tf.cond

tf.cond(tf.reduce_any(car_images_mask), update_l1_acc, lambda: tf.constant(0))

return {
"total_loss": self.total_loss_tracker.result(),
"l1_loss": self.loss_tracker_l1.result(),
"l2_loss": self.loss_tracker_l2.result(),
"l1_accuracy": self.acc_l1.result(),
"l2_accuracy": self.acc_l2.result(),
}

def test_step(self, data):
images, labels = data
gt_l2 = tf.cast(labels > 0, tf.float32)
car_images_mask = (labels > 0)

pred_l2, pred_l1 = self.model(images, training=False)

loss_l2_value = self.l2_loss(gt_l2, pred_l2)
def compute_l1_loss():
pred_l1_masked = tf.boolean_mask(pred_l1, car_images_mask)
gt_l1_masked = tf.boolean_mask(labels, car_images_mask)
return self.l1_loss(gt_l1_masked, pred_l1_masked)

def zero_l1_loss():
return tf.constant(0.0, dtype=tf.float32)

# tf.cond evaluates one branch at graph runtime
loss_l1_value = tf.cond(
tf.reduce_any(car_images_mask),
true_fn=compute_l1_loss,
false_fn=zero_l1_loss
)

total_loss = (loss_l2_value * self.weight_l2) + (loss_l1_value * self.weight_l1)

self.loss_tracker_l1.update_state(loss_l1_value)
self.loss_tracker_l2.update_state(loss_l2_value)
self.total_loss_tracker.update_state(total_loss)

self.acc_l2.update_state(gt_l2, pred_l2)
def update_l1_acc():
pred_l1_masked = tf.boolean_mask(pred_l1, car_images_mask)
gt_l1_masked = tf.boolean_mask(labels, car_images_mask)
gt_l1_masked_binary = tf.cast(gt_l1_masked - 1, tf.float32)
self.acc_l1.update_state(gt_l1_masked_binary, pred_l1_masked)
return 0 # Dummy op required by tf.cond

tf.cond(tf.reduce_any(car_images_mask), update_l1_acc, lambda: tf.constant(0))

return {
"total_loss": self.total_loss_tracker.result(),
"l1_loss": self.loss_tracker_l1.result(),
"l2_loss": self.loss_tracker_l2.result(),
"l1_accuracy": self.acc_l1.result(),
"l2_accuracy": self.acc_l2.result(),
}

@property
def metrics(self):
return [self.loss_tracker_l1, self.loss_tracker_l2, self.total_loss_tracker, self.acc_l1, self.acc_l2]


Подробнее здесь: https://stackoverflow.com/questions/796 ... low-l1-tes