手把手系列 | 教你用Python构建多标签图像分类模型（附案例）

一、多标签图像分类的核心概念

多标签图像分类（Multi-Label Image Classification）与传统的单标签分类（如ImageNet中的1000类分类）不同，其核心特点是一张图像可能同时属于多个类别。例如，一张包含“海滩”“日落”“人群”的图片需要同时预测这三个标签。这种任务常见于医疗影像分析（如同时识别多种病变）、电商商品标签（如“连衣裙”“碎花”“长袖”）等场景。

关键技术挑战

1. 标签相关性：不同标签之间可能存在依赖关系（如“猫”和“猫粮”）。
2. 类别不平衡：某些标签出现频率远高于其他标签。
3. 评估指标：需使用多标签专属指标（如Hamming Loss、F1-Score）。

二、完整开发流程（附代码）

1. 环境准备

# 基础环境
!pip install tensorflow keras opencv-python numpy matplotlib scikit-learn
# 可选：使用GPU加速
# !pip install tensorflow-gpu

2. 数据集准备与预处理

以VGG多标签数据集（包含20000张图片，15个标签）为例：

import os
import cv2
import numpy as np
from sklearn.model_selection import train_test_split
def load_data(data_dir):
    images = []
    labels = []
    label_names = []
    # 假设数据目录结构：data_dir/images/xxx.jpg, data_dir/labels/xxx.txt
    for img_file in os.listdir(os.path.join(data_dir, "images")):
        img_path = os.path.join(data_dir, "images", img_file)
        label_path = os.path.join(data_dir, "labels", img_file.replace(".jpg", ".txt"))
        # 读取图像并归一化
        img = cv2.imread(img_path)
        img = cv2.resize(img, (224, 224))  # 统一尺寸
        img = img / 255.0  # 归一化到[0,1]
        images.append(img)
        # 读取标签（每行一个标签，0/1表示是否存在）
        with open(label_path, "r") as f:
            label_vec = [int(line.strip()) for line in f]
        labels.append(label_vec)
        # 记录所有标签名称（假设第一行是标签名）
        if not label_names and len(label_vec) > 0:
            label_names = [f"label_{i}" for i in range(len(label_vec))]
    return np.array(images), np.array(labels), label_names
# 加载数据
X, y, label_names = load_data("vgg_multilabel_dataset")
# 划分训练集/测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
print(f"训练集形状: {X_train.shape}, 测试集形状: {X_test.shape}")

3. 模型构建（基于Keras）

方案一：基础CNN + 多标签输出层

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, Dropout
def build_basic_model(num_classes):
    model = Sequential([
        Conv2D(32, (3,3), activation='relu', input_shape=(224,224,3)),
        MaxPooling2D((2,2)),
        Conv2D(64, (3,3), activation='relu'),
        MaxPooling2D((2,2)),
        Conv2D(128, (3,3), activation='relu'),
        MaxPooling2D((2,2)),
        Flatten(),
        Dense(256, activation='relu'),
        Dropout(0.5),
        Dense(num_classes, activation='sigmoid')  # 多标签使用sigmoid
    ])
    return model
model = build_basic_model(y_train.shape[1])
model.compile(optimizer='adam', 
              loss='binary_crossentropy',  # 多标签损失函数
              metrics=['accuracy'])
model.summary()

方案二：迁移学习（ResNet50）

from tensorflow.keras.applications import ResNet50
from tensorflow.keras.layers import GlobalAveragePooling2D
def build_resnet_model(num_classes):
    base_model = ResNet50(weights='imagenet', include_top=False, input_shape=(224,224,3))
    base_model.trainable = False  # 冻结预训练层
    model = Sequential([
        base_model,
        GlobalAveragePooling2D(),
        Dense(256, activation='relu'),
        Dropout(0.5),
        Dense(num_classes, activation='sigmoid')
    ])
    return model
resnet_model = build_resnet_model(y_train.shape[1])
resnet_model.compile(optimizer='adam', 
                     loss='binary_crossentropy',
                     metrics=['accuracy'])
resnet_model.summary()

4. 模型训练与优化

from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
# 定义回调函数
callbacks = [
    EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True),
    ModelCheckpoint('best_model.h5', save_best_only=True)
]
# 训练模型
history = resnet_model.fit(
    X_train, y_train,
    validation_data=(X_test, y_test),
    epochs=50,
    batch_size=32,
    callbacks=callbacks
)
# 绘制训练曲线
import matplotlib.pyplot as plt
plt.plot(history.history['accuracy'], label='train_acc')
plt.plot(history.history['val_accuracy'], label='val_acc')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.legend()
plt.show()

5. 模型评估与预测

评估指标实现

from sklearn.metrics import hamming_loss, f1_score
def evaluate_multilabel(model, X_test, y_test):
    y_pred = model.predict(X_test)
    y_pred_binary = (y_pred > 0.5).astype(int)  # 二值化预测结果
    print(f"Hamming Loss: {hamming_loss(y_test, y_pred_binary):.4f}")
    print(f"Macro F1-Score: {f1_score(y_test, y_pred_binary, average='macro'):.4f}")
    print(f"Micro F1-Score: {f1_score(y_test, y_pred_binary, average='micro'):.4f}")
evaluate_multilabel(resnet_model, X_test, y_test)

预测新图像

def predict_image(model, img_path, label_names):
    img = cv2.imread(img_path)
    img = cv2.resize(img, (224,224))
    img = img / 255.0
    img_array = np.expand_dims(img, axis=0)  # 添加batch维度
    pred = model.predict(img_array)[0]
    pred_binary = (pred > 0.5).astype(int)
    # 显示预测结果
    print("预测标签:")
    for i, (p, name) in enumerate(zip(pred_binary, label_names)):
        if p == 1:
            print(f"- {name} (置信度: {pred[i]:.2f})")
# 示例预测
predict_image(resnet_model, "test_image.jpg", label_names)

三、进阶优化技巧

1. 处理类别不平衡

from sklearn.utils.class_weight import compute_sample_weight
# 计算样本权重（平衡正负样本）
sample_weights = compute_sample_weight(
    class_weight='balanced',
    y=y_train.flatten()  # 需要将多标签展开为单标签形式
)
# 在fit方法中添加sample_weight参数
model.fit(..., sample_weight=sample_weights)

2. 使用自定义损失函数

import tensorflow as tf
def focal_loss(gamma=2.0, alpha=0.25):
    def focal_loss_fn(y_true, y_pred):
        bce = tf.keras.losses.binary_crossentropy(y_true, y_pred)
        p_t = tf.where(tf.equal(y_true, 1), y_pred, 1 - y_pred)
        modulating_factor = tf.pow(1.0 - p_t, gamma)
        alpha_weight = tf.where(tf.equal(y_true, 1), alpha, 1 - alpha)
        return alpha_weight * modulating_factor * bce
    return focal_loss_fn
model.compile(loss=focal_loss(gamma=1.5, alpha=0.3), ...)

3. 标签相关性建模

使用Graph Convolutional Networks (GCN)或Transformer架构：

# 示例：使用Transformer的注意力机制（简化版）
from tensorflow.keras.layers import MultiHeadAttention
def build_transformer_model(num_classes):
    inputs = tf.keras.Input(shape=(224,224,3))
    x = ResNet50(weights='imagenet', include_top=False)(inputs)
    x = GlobalAveragePooling2D()(x)
    # 添加自注意力层
    attn_output = MultiHeadAttention(num_heads=4, key_dim=64)(x, x)
    x = tf.keras.layers.Concatenate()([x, attn_output])
    x = Dense(256, activation='relu')(x)
    outputs = Dense(num_classes, activation='sigmoid')(x)
    return tf.keras.Model(inputs=inputs, outputs=outputs)

四、完整案例：医疗影像多标签分类

以胸部X光片分类为例（需标注肺炎、气胸、骨折等标签）：

1. 数据准备：使用ChestX-ray14数据集（含112,120张影像，14种病变）
2. 模型选择：DenseNet121 + 注意力机制
3. 关键代码：
   ```python
   from tensorflow.keras.applications import DenseNet121

def build_medical_model(num_classes):
base_model = DenseNet121(weights=’imagenet’, include_top=False, input_shape=(224,224,3))
base_model.trainable = False

inputs = tf.keras.Input(shape=(224,224,3))
x = base_model(inputs)
x = GlobalAveragePooling2D()(x)
# 添加空间注意力
attention = tf.keras.layers.Conv2D(1, (1,1), activation='sigmoid')(x)
attention = tf.keras.layers.Reshape((1,1,256))(attention)  # 假设x的最后一个维度是256
x = tf.keras.layers.Multiply()([x, attention])
x = Dense(256, activation='relu')(x)
outputs = Dense(num_classes, activation='sigmoid')(x)
return tf.keras.Model(inputs=inputs, outputs=outputs)

## 五、部署建议
1. **模型轻量化**：使用TensorFlow Lite或ONNX格式导出
2. **API服务**：通过FastAPI封装预测接口
```python
from fastapi import FastAPI
import numpy as np
from PIL import Image
import io
app = FastAPI()
model = tf.keras.models.load_model("best_model.h5")
@app.post("/predict")
async def predict(image: bytes):
    img = Image.open(io.BytesIO(image))
    img = img.resize((224,224))
    img_array = np.array(img) / 255.0
    img_array = np.expand_dims(img_array, axis=0)
    pred = model.predict(img_array)[0]
    return {"predictions": pred.tolist()}

六、总结与最佳实践

1. 数据质量：确保标签准确性，使用专业工具标注（如LabelImg、CVAT）
2. 模型选择：
   • 小数据集：使用预训练模型 + 微调
   • 大数据集：可从头训练高效架构（如EfficientNet）
3. 评估策略：
   • 除准确率外，重点关注Hamming Loss和F1-Score
   • 使用k折交叉验证
4. 部署优化：
   • 量化模型（如INT8）
   • 使用TensorRT加速

通过以上完整流程，开发者可以快速构建并部署一个高性能的多标签图像分类系统。实际项目中，建议从简单模型开始，逐步增加复杂度，同时密切关注模型在测试集上的泛化能力。