目錄
二分類胸片判斷:
1. 數據加載時指定了兩類標簽
2. 損失函數用了二分類專用的
3. 輸出層只有 1 個神經元,用了sigmoid激活函數
4. 預測時用 0.5 作為分類閾值
二分類胸片判斷:
import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, roc_auc_score, roc_curve, confusion_matrix, classification_report
from imblearn.over_sampling import RandomOverSampler
import tensorflow as tf
from keras import layers
from keras import models
# 或者更常用的是直接導入Sequential類
from keras.models import Sequential
from keras.preprocessing.image import ImageDataGenerator
import os
import zipfile
import requests
from tensorflow.python.keras.callbacks import EarlyStopping
# 這個代碼執行 請切換環境到tf_env
plt.rcParams['font.sans-serif'] = ['SimHei'] # 使用 SimHei 字體
plt.rcParams['axes.unicode_minus'] = False # 解決負號顯示問題
plt.rcParams['font.size'] = 10 # 設置全局字體大小# 數據加載和預處理
def load_data(train_dir, test_dir, val_dir, img_size=(150, 150), batch_size=32):# 數據增強器 - 僅用于訓練集train_datagen = ImageDataGenerator(rescale=1. / 255,rotation_range=10,width_shift_range=0.1,height_shift_range=0.1,shear_range=0.1,zoom_range=0.1,horizontal_flip=True)# 驗證集和測試集只需要重新縮放val_test_datagen = ImageDataGenerator(rescale=1. / 255)# 加載訓練數據train_generator = train_datagen.flow_from_directory(train_dir,target_size=img_size,batch_size=batch_size,class_mode='binary',classes=['NORMAL', 'PNEUMONIA'],shuffle=True)# 加載驗證數據val_generator = val_test_datagen.flow_from_directory(val_dir,target_size=img_size,batch_size=batch_size,class_mode='binary',classes=['NORMAL', 'PNEUMONIA'],shuffle=False)# 加載測試數據test_generator = val_test_datagen.flow_from_directory(test_dir,target_size=img_size,batch_size=batch_size,class_mode='binary',classes=['NORMAL', 'PNEUMONIA'],shuffle=False)return train_generator, val_generator, test_generator# 處理樣本不均衡(過采樣)
def handle_imbalance(generator):# 提取特征和標簽X, y = [], []num_batches = len(generator)# 重置生成器以確保從開始獲取數據generator.reset()for i in range(num_batches):batch_x, batch_y = generator.next()X.append(batch_x)y.append(batch_y)X = np.concatenate(X)y = np.concatenate(y)# 打印原始分布print(f"原始樣本分布: 正常={np.sum(y == 0)}, 肺炎={np.sum(y == 1)}")# 展平特征用于過采樣X_flat = X.reshape(X.shape[0], -1)# 過采樣少數類ros = RandomOverSampler(random_state=42)X_resampled, y_resampled = ros.fit_resample(X_flat, y)# 恢復圖像形狀X_resampled = X_resampled.reshape(-1, *X.shape[1:])print(f"過采樣后分布: 正常={np.sum(y_resampled == 0)}, 肺炎={np.sum(y_resampled == 1)}")return X_resampled, y_resampled, y# 構建改進的CNN模型
def build_model(input_shape):model = models.Sequential([# 第一個卷積塊layers.Conv2D(32, (3, 3), activation='relu', input_shape=input_shape),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),layers.Dropout(0.2),# 第二個卷積塊layers.Conv2D(64, (3, 3), activation='relu'),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),layers.Dropout(0.3),# 第三個卷積塊layers.Conv2D(128, (3, 3), activation='relu'),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),layers.Dropout(0.4),# 第四個卷積塊layers.Conv2D(256, (3, 3), activation='relu'),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),layers.Dropout(0.5),# 分類器layers.Flatten(),layers.Dense(512, activation='relu'),layers.BatchNormalization(),layers.Dropout(0.5),layers.Dense(1, activation='sigmoid')])# 使用更穩定的優化器optimizer = tf.keras.optimizers.Adam(learning_rate=0.0001)model.compile(optimizer=optimizer,loss='binary_crossentropy',metrics=['accuracy',tf.keras.metrics.Precision(name='precision'),tf.keras.metrics.Recall(name='recall'),tf.keras.metrics.AUC(name='auc')])return model# 主函數
def main():# 假設數據集已經手動下載并解壓train_dir = "chest_xray/train"test_dir = "chest_xray/test"val_dir = "chest_xray/val"# 加載數據img_size = (150, 150)batch_size = 32train_generator, val_generator, test_generator = load_data(train_dir, test_dir, val_dir, img_size, batch_size)# 處理樣本不均衡X_train, y_train_resampled, y_train_original = handle_imbalance(train_generator)# 計算類別權重(基于原始分布)n_normal = np.sum(y_train_original == 0)n_pneumonia = np.sum(y_train_original == 1)total = n_normal + n_pneumoniaweight_for_normal = (1 / n_normal) * (total / 2.0)weight_for_pneumonia = (1 / n_pneumonia) * (total / 2.0)class_weights = {0: weight_for_normal, 1: weight_for_pneumonia}print(f"類別權重: 正常={weight_for_normal:.2f}, 肺炎={weight_for_pneumonia:.2f}")# 構建模型model = build_model((*img_size, 3))model.summary()# 提前停止回調early_stopping = EarlyStopping(monitor='val_loss',patience=5,restore_best_weights=True,verbose=1)# 訓練模型history = model.fit(X_train, y_train_resampled,epochs=30,batch_size=32,validation_data=val_generator,class_weight=class_weights,callbacks=[early_stopping],verbose=1)# 評估模型 - 使用完整測試集test_generator.reset()test_steps = len(test_generator)test_results = model.evaluate(test_generator, steps=test_steps, verbose=1)print("\n測試集評估結果:")print(f"準確率: {test_results[1]:.4f}")print(f"精確率: {test_results[2]:.4f}")print(f"召回率: {test_results[3]:.4f}")print(f"AUC: {test_results[4]:.4f}")# 獲取測試集所有預測結果test_generator.reset()y_true = []y_pred_prob = []for i in range(test_steps):batch_x, batch_y = test_generator.next()y_true.extend(batch_y)batch_pred = model.predict(batch_x, verbose=0).ravel()y_pred_prob.extend(batch_pred)y_true = np.array(y_true)y_pred_prob = np.array(y_pred_prob)y_pred = (y_pred_prob > 0.5).astype(int)# 計算額外指標f1 = f1_score(y_true, y_pred)auc = roc_auc_score(y_true, y_pred_prob)print(f"\nF1-score: {f1:.4f}")print(f"AUC-ROC: {auc:.4f}")# 分類報告print("\n分類報告:")print(classification_report(y_true, y_pred, target_names=['NORMAL', 'PNEUMONIA']))# 混淆矩陣cm = confusion_matrix(y_true, y_pred)print("混淆矩陣:")print(cm)# 繪制ROC曲線fpr, tpr, _ = roc_curve(y_true, y_pred_prob)plt.figure(figsize=(10, 6))plt.plot(fpr, tpr, color='darkorange', lw=2, label=f'ROC曲線 (AUC = {auc:.4f})')plt.plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--')plt.xlim([0.0, 1.0])plt.ylim([0.0, 1.05])plt.xlabel('False Positive Rate')plt.ylabel('True Positive Rate')plt.title('接收者操作特征曲線(ROC)')plt.legend(loc="lower right")plt.savefig('roc_curve.png', dpi=300)plt.show()# 繪制訓練歷史plt.figure(figsize=(12, 8))plt.subplot(2, 2, 1)plt.plot(history.history['accuracy'], label='訓練準確率')plt.plot(history.history['val_accuracy'], label='驗證準確率')plt.title('準確率')plt.legend()plt.subplot(2, 2, 2)plt.plot(history.history['loss'], label='訓練損失')plt.plot(history.history['val_loss'], label='驗證損失')plt.title('損失')plt.legend()plt.subplot(2, 2, 3)plt.plot(history.history['precision'], label='訓練精確率')plt.plot(history.history['val_precision'], label='驗證精確率')plt.title('精確率')plt.legend()plt.subplot(2, 2, 4)plt.plot(history.history['recall'], label='訓練召回率')plt.plot(history.history['val_recall'], label='驗證召回率')plt.title('召回率')plt.legend()plt.tight_layout()plt.savefig('training_history.png', dpi=300)plt.show()if __name__ == "__main__":main()
這段代碼里有很多地方明確體現了這是一個二分類任務(判斷 “正常胸片” 和 “肺炎胸片” 兩類),最關鍵的有這幾個地方:
1. 數據加載時指定了兩類標簽
在 load_data 函數中,加載數據時明確指定了類別為兩類:
train_generator = train_datagen.flow_from_directory( ??? train_dir, ??? ... ??? class_mode='binary',? # 這里指定是“二分類”模式 ??? classes=['NORMAL', 'PNEUMONIA'],? # 明確兩類:正常(NORMAL)和肺炎(PNEUMONIA) ??? ... ) |
- class_mode='binary':直接告訴程序 “這是二分類任務”,標簽會被處理成 0 和 1(0 代表正常,1 代表肺炎)。
- classes=['NORMAL', 'PNEUMONIA']:手動指定只有這兩個類別,沒有第三種情況。
2. 損失函數用了二分類專用的
在模型編譯時,損失函數用的是 binary_crossentropy(二分類交叉熵):
model.compile( ??? ... ??? loss='binary_crossentropy',? # 專門用于二分類的損失函數 ??? ... ) |
這個損失函數的作用是:計算 “模型判斷為 0 或 1 的概率” 與 “實際標簽(0 或 1)” 之間的差距,指導模型優化。如果是多分類任務,會用其他損失函數(比如 categorical_crossentropy)。
3. 輸出層只有 1 個神經元,用了sigmoid激活函數
模型的最后一層是:
layers.Dense(1, activation='sigmoid')? # 輸出層 |
- Dense(1):只輸出 1 個數值,這個數值經過 sigmoid 激活后,會被壓縮到 0~1 之間。
- 實際含義:
- 數值越接近 0 → 模型認為 “更可能是正常胸片(0 類)”;
- 數值越接近 1 → 模型認為 “更可能是肺炎胸片(1 類)”。
這是二分類任務的典型輸出方式(多分類會有多個神經元,對應多個類別)。
4. 預測時用 0.5 作為分類閾值
在生成最終判斷結果時:
y_pred = (y_pred_prob > 0.5).astype(int)? # 大于0.5算1類(肺炎),否則算0類(正常) |
直接用 0.5 作為 “兩類的分界線”,把輸出概率分成 “0” 和 “1” 兩類,進一步說明這是二分類。
從 “數據標簽定義”“損失函數選擇”“輸出層設計” 到 “最終預測規則”,全流程都圍繞 “只能分成兩類” 展開,沒有任何支持多類別的設計。所以這段代碼是典型的二分類任務,目標就是區分 “正常胸片” 和 “肺炎胸片”。