【Clustering】Deep Image Clustering with TensorFlow

使用 Python+OpenCV+TensorFlow 实现图像聚类：无需标注数据的深度学习方案

摘要：本文介绍如何在不创建训练数据集、不训练神经网络的情况下，利用预训练 CNN（ResNet50）作为特征提取器，结合层次聚类算法对图像进行无监督分类。我们将以 Kaggle 上的 Keep Babies Safe 数据集为例，完成从特征提取到聚类可视化的完整流程。

1. 引言

在计算机视觉的众多任务中，图像分类 是最基础也最常见的需求。传统方法需要我们：

但现实往往更复杂——很多场景下我们根本没有标注数据。比如参加一个深度学习竞赛时，官方只给了测试集，没有训练标签，怎么办？

答案就是：将预训练 CNN 作为特征提取器，直接对测试集进行无监督聚类。

这就是本文的核心内容。我们不需要训练任何模型，只需要一个在 ImageNet 上预训练好的 CNN，就能把视觉上相似的图像自动聚到一起。

2. 核心原理

2.1 CNN 如何"看懂"图像？

卷积神经网络（CNN）的卷积层本质上是一系列特征检测器。浅层卷积层倾向于检测边缘、颜色、纹理等低级特征，深层卷积层则能检测更高级的语义特征（如眼睛、鼻子、轮子等物体组成部分）。

以一个预训练好的 VGG16 为例，可视化其第一层卷积层看到的特征图：

图：VGG16 第一个卷积层的特征图可视化。不同通道关注不同的视觉模式——有的关注边缘轮廓，有的关注纹理细节。

2.2 迁移学习作为特征提取器

既然 CNN 已经学会了提取有意义的视觉特征，我们完全可以直接用它来处理新图像——这就是 迁移学习 (Transfer Learning) 的思路：

对于 ResNet50，每张图像会被映射为一个 2048 维的特征向量。这个向量已经编码了图像的语义信息，可以直接用于聚类。

2.3 整体流程

原始图像 → ResNet50(无分类头) → 2048–d 特征向量 → 标准化 → 聚类算法 → 簇标签 → 可视化

3. 数据集

本文使用 Kaggle 上的 Keep Babies Safe 数据集。

该数据集包含 1131 张图像，分为两大类：

• 玩具（Toys）：毛绒玩具、积木、玩偶等
• 消费品（Consumer Products）：日用品、包装盒等

数据集位于 /dataset/images/ 目录下，附带一个 test.csv 文件（包含图像名和类别列，但标签为空——这正是我们需要聚类的原因）。

让我们先看看数据集中的一些样本图像：

4. 代码实现

下面逐步解析核心实现。

4.1 环境依赖

pip install numpy opencv-python tensorflow scikit-learn matplotlib seaborn pandas

4.2 CNN 特征提取器

import numpy as np
import cv2
import tensorflow as tf

class CNNFeatureExtractor:
def __init__(self, model_name="resnet50", input_size=(224, 224)):
if model_name == "resnet50":
self.base_model = tf.keras.applications.ResNet50(
include_top=False, # 移除分类头
weights="imagenet", # 使用 ImageNet 预训练权重
pooling="avg", # 全局平均池化 → 2048-d 向量
)
self.base_model.trainable = False # 冻结权重

def extract(self, img):
img = cv2.resize(img, self.input_size)
img = np.expand_dims(img, axis=0)
img = tf.keras.applications.resnet50.preprocess_input(img)
features = self.base_model.predict(img, verbose=0)
return features.flatten() # 返回 2048 维特征向量

关键点解释：

4.3 图像聚类器

from sklearn.cluster import AgglomerativeClustering
from sklearn.preprocessing import StandardScaler

class ImageClusterer:
def __init__(self, n_clusters=2, method="agglomerative"):
self.model = AgglomerativeClustering(n_clusters=n_clusters)
self.scaler = StandardScaler()

def fit(self, feature_matrix):
scaled = self.scaler.fit_transform(feature_matrix)
self.labels_ = self.model.fit_predict(scaled)
return self.labels_

为什么选择层次聚类？

• 不需要预设聚类中心：不像 K-Means 需要随机初始化
• 对特征分布不敏感：层次聚类通过计算样本间距离逐步合并，更稳定
• 可解释性强：可以生成树状图（dendrogram）展示聚类层次

4.4 可视化

import seaborn as sns
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA

def plot_cluster_scatter(features, labels, image_names):
# PCA 降维到 2D
reduced = PCA(n_components=2).fit_transform(features)

df = pd.DataFrame({
"V1": reduced[:, 0],
"V2": reduced[:, 1],
"Category": [f"Cluster {l}" for l in labels],
})

sns.scatterplot(data=df, x="V1", y="V2", hue="Category", palette="Set2")
plt.title("Image Clustering Results")
plt.show()

4.5 完整运行

python main.py \\
–data_dir ./dataset/dataset/images/ \\
–n_clusters 2 \\
–method agglomerative \\
–output_dir ./output/

5. 实验结果

5.1 聚类分布

在 1131 张图像上运行层次聚类（n_clusters=2）：

5.2 PCA 降维可视化

将 2048 维特征向量用 PCA 降维到 2D 后可视化：

从上图可以看出，两个簇在特征空间中有明显的分离边界。这说明 ResNet50 提取的视觉特征确实能够捕捉到"玩具"和"消费品"之间的视觉差异。

5.3 性能数据

6. 进阶探索

6.1 尝试不同的特征提取器

# 使用 VGG16 替代 ResNet50（特征维度从 2048 降为 512）
python main.py –model vgg16 –n_clusters 2 –data_dir ./dataset/dataset/images/

和 resnet50 比起来还是差一些，特征都没有怎么拉开

6.2 尝试不同聚类算法

# K-Means
python main.py –method kmeans –n_clusters 2 –data_dir ./dataset/dataset/images/

k-means 感觉效果更好一点，特别是分界处的最上面部分

# DBSCAN（适用于未知类别数量的场景）
python main.py –method dbscan –data_dir ./dataset/dataset/images/

不指定，聚成 1 个类别了

6.3 使用 t-SNE 替代 PCA

t-SNE 能更好地保留局部结构，适合可视化高维特征：

python main.py –reduction tsne –data_dir ./dataset/dataset/images/

6.4 结合 OCR 文本特征

原文中还提到可以使用 Tesseract OCR 提取图像中的文字信息，作为额外的特征维度：

import pytesseract

text = pytesseract.image_to_string(img, config='–oem 2')
# 将文本特征与视觉特征拼接，可能进一步提升聚类质量

7. 总结

本文展示了一个实用的技巧：当没有标注数据时，如何利用预训练 CNN + 无监督聚类来解决图像分类问题。

核心要点：

这种方法特别适用于以下场景：

• 竞赛中的无标注测试集
• 快速探索新数据集的结构
• 数据清洗和去重
• 作为半监督学习的第一步（先聚类，再人工标注几个样本）

8. 完整代码

feature_extractor.py

"""
使用预训练 CNN (ResNet50 / VGG16) 作为图像特征提取器

核心思路：
– 不需要自己标注训练数据，直接使用 ImageNet 预训练权重
– 移除分类头 (include_top=False)，只保留卷积层 + 池化层
– 每张图像 → 2048 维特征向量 (ResNet50 avg pooling)
– 这个特征向量已经捕获了图像的语义信息（纹理、形状、边缘等）
"""

import numpy as np
import cv2
import tensorflow as tf
from typing import List, Tuple, Optional

class CNNFeatureExtractor:
"""
基于预训练 CNN 的图像特征提取器

Parameters
———-
model_name : str
预训练模型名称，支持 'resnet50' 或 'vgg16'
input_size : Tuple[int, int]
模型输入尺寸，默认 (224, 224)
"""

def __init__(
self,
model_name: str = "resnet50",
input_size: Tuple[int, int] = (224, 224),
):
self.model_name = model_name.lower()
self.input_size = input_size

if self.model_name == "resnet50":
self.base_model = tf.keras.applications.ResNet50(
include_top=False,
weights="imagenet",
pooling="avg", # 全局平均池化，输出 2048-d 向量
)
elif self.model_name == "vgg16":
self.base_model = tf.keras.applications.VGG16(
include_top=False,
weights="imagenet",
pooling="avg", # 输出 512-d 向量
)
else:
raise ValueError(f"Unsupported model: {model_name}")

# 冻结所有权重，只做前向推理
self.base_model.trainable = False

def preprocess(self, img: np.ndarray) –> np.ndarray:
"""
预处理图像：resize → expand dims → 模型特定归一化

Parameters
———-
img : np.ndarray
BGR 图像 (H, W, 3)

Returns
——-
np.ndarray
预处理后的图像 (1, H, W, 3)
"""
img = cv2.resize(img, self.input_size)
img = np.expand_dims(img, axis=0)

if self.model_name == "resnet50":
img = tf.keras.applications.resnet50.preprocess_input(img)
elif self.model_name == "vgg16":
img = tf.keras.applications.vgg16.preprocess_input(img)

return img

def extract(self, img: np.ndarray) –> np.ndarray:
"""
从单张图像提取特征向量

Parameters
———-
img : np.ndarray
BGR 图像 (H, W, 3)

Returns
——-
np.ndarray
特征向量，形状 (feature_dim,)
"""
preprocessed = self.preprocess(img)
features = self.base_model.predict(preprocessed, verbose=0)
return features.flatten()

def extract_batch(self, images: List[np.ndarray]) –> np.ndarray:
"""
批量提取特征向量

Parameters
———-
images : List[np.ndarray]
BGR 图像列表

Returns
——-
np.ndarray
特征向量矩阵，形状 (N, feature_dim)
"""
features_list = []
for img in images:
feats = self.extract(img)
features_list.append(feats)
return np.array(features_list, dtype=np.float64)

def get_feature_map_model(self, layer_idx: int):
"""
获取指定卷积层的输出模型（用于可视化特征图）

Parameters
———-
layer_idx : int
目标卷积层的索引

Returns
——-
tf.keras.Model
截断模型，输出为指定层的特征图
"""
inputs = self.base_model.inputs
outputs = self.base_model.layers[layer_idx].output
return tf.keras.Model(inputs=inputs, outputs=outputs)

def visualize_feature_maps(
self, img: np.ndarray, layer_idx: int, max_maps: int = 64
) –> np.ndarray:
"""
可视化指定层的特征图

Parameters
———-
img : np.ndarray
输入图像
layer_idx : int
卷积层索引
max_maps : int
最多显示的特征图数量

Returns
——-
np.ndarray
可视化的特征图网格图像
"""
import matplotlib.pyplot as plt

preprocessed = self.preprocess(img)
fm_model = self.get_feature_map_model(layer_idx)
feature_maps = fm_model.predict(preprocessed, verbose=0)

# 取第一个样本
if feature_maps.ndim == 4:
feature_maps = feature_maps[0]

n_maps = min(feature_maps.shape[–1], max_maps)
grid_size = int(np.ceil(np.sqrt(n_maps)))

fig, axes = plt.subplots(grid_size, grid_size, figsize=(12, 12))
for i in range(grid_size * grid_size):
row, col = i // grid_size, i % grid_size
if i < n_maps:
axes[row, col].imshow(
feature_maps[:, :, i], cmap="viridis"
)
axes[row, col].axis("off")

plt.tight_layout()
return fig

def load_images_from_folder(
folder_path: str, input_size: Tuple[int, int] = (224, 224)
) –> Tuple[List[str], List[np.ndarray]]:
"""
从文件夹加载所有图像

Parameters
———-
folder_path : str
图像文件夹路径
input_size : Tuple[int, int]
统一的图像尺寸

Returns
——-
Tuple[List[str], List[np.ndarray]]
(文件名列表, 图像数组列表)
"""
import os

valid_extensions = {".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".webp"}
image_names = []
images = []

for filename in sorted(os.listdir(folder_path)):
ext = os.path.splitext(filename)[1].lower()
if ext not in valid_extensions:
continue

filepath = os.path.join(folder_path, filename)
img = cv2.imread(filepath)
if img is None:
print(f"Warning: Failed to load {filepath}")
continue

image_names.append(filename)
images.append(img)

print(f"Loaded {len(images)} images from {folder_path}")
return image_names, images

clustering.py

"""
图像聚类模块

使用 CNN 提取的特征向量进行无监督聚类。
支持多种聚类算法：
– 层次聚类 (Agglomerative Clustering) — 文章推荐方法
– K-Means
– DBSCAN（适用于未知类别数量的场景）
"""

import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from sklearn.cluster import (
AgglomerativeClustering,
KMeans,
DBSCAN,
)
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE

class ImageClusterer:
"""
图像聚类器

Parameters
———-
n_clusters : int
聚类数量
method : str
聚类算法: 'agglomerative', 'kmeans', 'dbscan'
random_state : int
随机种子（保证可复现性）
"""

def __init__(
self,
n_clusters: int = 2,
method: str = "agglomerative",
random_state: int = 42,
):
self.n_clusters = n_clusters
self.method = method.lower()
self.random_state = random_state

if self.method == "agglomerative":
self.model = AgglomerativeClustering(n_clusters=n_clusters)
elif self.method == "kmeans":
self.model = KMeans(
n_clusters=n_clusters,
random_state=random_state,
n_init=10,
)
elif self.method == "dbscan":
self.model = DBSCAN(eps=0.5, min_samples=5)
else:
raise ValueError(f"Unsupported clustering method: {method}")

self.scaler = StandardScaler()
self.labels_: Optional[np.ndarray] = None
self.feature_matrix_: Optional[np.ndarray] = None

def fit(self, feature_matrix: np.ndarray) –> np.ndarray:
"""
对特征矩阵进行聚类

Parameters
———-
feature_matrix : np.ndarray
特征向量矩阵，形状 (N, D)

Returns
——-
np.ndarray
聚类标签，形状 (N,)
"""
# 标准化特征
self.feature_matrix_ = feature_matrix
scaled_features = self.scaler.fit_transform(feature_matrix)
self.labels_ = self.model.fit_predict(scaled_features)
return self.labels_

def get_cluster_stats(self) –> Dict[int, int]:
"""
获取每个簇的样本数量

Returns
——-
Dict[int, int]
{cluster_id: count}
"""
if self.labels_ is None:
raise RuntimeError("Please call fit() first.")
unique, counts = np.unique(self.labels_, return_counts=True)
return dict(zip(unique.tolist(), counts.tolist()))

def reduce_dimensions(
feature_matrix: np.ndarray,
method: str = "pca",
n_components: int = 2,
random_state: int = 42,
) –> np.ndarray:
"""
降维到 2D/3D 用于可视化

Parameters
———-
feature_matrix : np.ndarray
特征向量矩阵
method : str
降维方法: 'pca' 或 'tsne'
n_components : int
目标维度
random_state : int
随机种子

Returns
——-
np.ndarray
降维后的矩阵，形状 (N, n_components)
"""
if method.lower() == "pca":
reducer = PCA(n_components=n_components, random_state=random_state)
elif method.lower() == "tsne":
reducer = TSNE(
n_components=n_components,
random_state=random_state,
perplexity=min(30, len(feature_matrix) – 1),
)
else:
raise ValueError(f"Unknown reduction method: {method}")

return reducer.fit_transform(feature_matrix)

def create_results_dataframe(
image_names: List[str],
feature_matrix: np.ndarray,
labels: np.ndarray,
) –> pd.DataFrame:
"""
创建结果 DataFrame，包含图像名、特征向量和聚类标签

Parameters
———-
image_names : List[str]
图像文件名列表
feature_matrix : np.ndarray
特征向量矩阵
labels : np.ndarray
聚类标签

Returns
——-
pd.DataFrame
"""
df = pd.DataFrame(
{
"photo_name": image_names,
"cluster_label": labels,
}
)

# 添加特征向量列
for i in range(feature_matrix.shape[1]):
df[f"feature_{i}"] = feature_matrix[:, i]

return df

main.py

#!/usr/bin/env python3
"""
使用 Python+OpenCV+Tensorflow 实现图像聚类

基于预训练 CNN (ResNet50) 提取图像特征，无需标注数据即可进行无监督图像聚类。

数据集: Keep Babies Safe (Kaggle)
将图像分为"玩具"和"消费品"两类

核心流程:
1. 加载图像 → ResNet50 提取特征向量 (2048-d)
2. 标准化特征 → 层次聚类 (Agglomerative Clustering)
3. PCA/TSNE 降维 → 2D 散点图可视化
4. 按簇展示样本图像

Usage:
# 下载数据集到 ./dataset/ 文件夹后运行:
python main.py –data_dir ./dataset/keep-babies-safe/ –n_clusters 2

# 也可用其他图像文件夹测试:
python main.py –data_dir ./my_images/ –n_clusters 3
"""

import os
import sys
import argparse
import time
import numpy as np

# 添加当前目录到路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from feature_extractor import CNNFeatureExtractor, load_images_from_folder
from clustering import (
ImageClusterer,
reduce_dimensions,
create_results_dataframe,
)
from visualize import (
plot_cluster_scatter,
plot_sample_images,
plot_cluster_distribution,
)

def main(args):
# ============================================================
# Step 1: 加载图像
# ============================================================
print("=" * 60)
print("Step 1/5: Loading images…")
print("=" * 60)

image_names, images = load_images_from_folder(args.data_dir)

if len(images) < 2:
print(f"Error: Need at least 2 images, got {len(images)}")
print("请先从 Kaggle 下载数据集到 ./dataset/ 目录:")
print(" https://www.kaggle.com/akash14/keep-babies-safe")
sys.exit(1)

print(f" ✓ Loaded {len(images)} images")

# ============================================================
# Step 2: 提取 CNN 特征
# ============================================================
print("=" * 60)
print("Step 2/5: Extracting features with ResNet50…")
print("=" * 60)

extractor = CNNFeatureExtractor(
model_name=args.model,
input_size=(args.img_size, args.img_size),
)

t0 = time.time()
feature_matrix = extractor.extract_batch(images)
elapsed = time.time() – t0

print(f" ✓ Feature shape: {feature_matrix.shape}")
print(f" ✓ Time elapsed: {elapsed:.1f}s ({elapsed/len(images):.1f}s per image)")

# ============================================================
# Step 3: 聚类
# ============================================================
print("=" * 60)
print(f"Step 3/5: Clustering with {args.method}…")
print("=" * 60)

clusterer = ImageClusterer(
n_clusters=args.n_clusters,
method=args.method,
)
labels = clusterer.fit(feature_matrix)
stats = clusterer.get_cluster_stats()

print(f" ✓ Cluster distribution: {stats}")

# ============================================================
# Step 4: 降维 + 可视化
# ============================================================
print("=" * 60)
print("Step 4/5: Reducing dimensions & visualizing…")
print("=" * 60)

# PCA 降维到 2D
reduced = reduce_dimensions(
feature_matrix,
method=args.reduction,
n_components=2,
)

os.makedirs(args.output_dir, exist_ok=True)

# 散点图
scatter_path = os.path.join(args.output_dir, "cluster_scatter.png")
fig_scatter = plot_cluster_scatter(
reduced,
labels,
image_names=image_names,
title=f"Image Clustering Results ({args.method}, {args.n_clusters} clusters)",
save_path=scatter_path,
)
print(f" ✓ Saved scatter plot to: {scatter_path}")

# 分布图
dist_path = os.path.join(args.output_dir, "cluster_distribution.png")
fig_dist = plot_cluster_distribution(
labels,
title=f"Cluster Distribution ({len(images)} images)",
save_path=dist_path,
)
print(f" ✓ Saved distribution plot to: {dist_path}")

# 样本图
sample_path = os.path.join(args.output_dir, "cluster_samples.png")
fig_samples = plot_sample_images(
images,
image_names,
labels,
n_per_cluster=min(args.show_n, 8),
save_path=sample_path,
)
print(f" ✓ Saved sample images to: {sample_path}")

# ============================================================
# Step 5: 保存结果
# ============================================================
print("=" * 60)
print("Step 5/5: Saving results…")
print("=" * 60)

df = create_results_dataframe(image_names, feature_matrix, labels)
csv_path = os.path.join(args.output_dir, "clustering_results.csv")
df.to_csv(csv_path, index=False)
print(f" ✓ Saved results to: {csv_path}")

# 打印每个簇的前 N 个图像
print("\\n" + "=" * 60)
print("Clustering Summary")
print("=" * 60)
for label in sorted(stats.keys()):
members = [image_names[i] for i in range(len(labels)) if labels[i] == label]
print(f"\\n Cluster {label} ({stats[label]} images):")
for m in members[:10]:
print(f" – {m}")
if len(members) > 10:
print(f" … and {len(members) – 10} more")

print("\\nDone! All outputs saved to:", args.output_dir)

if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Image Clustering with Pre-trained CNN Features"
)
parser.add_argument(
"–data_dir",
type=str,
default="./dataset/",
help="Path to image directory",
)
parser.add_argument(
"–n_clusters",
type=int,
default=2,
help="Number of clusters (default: 2)",
)
parser.add_argument(
"–method",
type=str,
default="agglomerative",
choices=["agglomerative", "kmeans", "dbscan"],
help="Clustering method (default: agglomerative)",
)
parser.add_argument(
"–model",
type=str,
default="resnet50",
choices=["resnet50", "vgg16"],
help="CNN feature extractor (default: resnet50)",
)
parser.add_argument(
"–img_size",
type=int,
default=224,
help="Input image size (default: 224)",
)
parser.add_argument(
"–reduction",
type=str,
default="pca",
choices=["pca", "tsne"],
help="Dimensionality reduction method (default: pca)",
)
parser.add_argument(
"–output_dir",
type=str,
default="./output/",
help="Output directory for results (default: ./output/)",
)
parser.add_argument(
"–show_n",
type=int,
default=5,
help="Number of sample images per cluster (default: 5)",
)

args = parser.parse_args()
main(args)

9. 参考资源

• 原文链接
• Keep Babies Safe 数据集
• ResNet 论文
• VGG16 论文
• Scikit-learn 聚类算法对比