机器视觉：基于 Python 的人脸识别与照片管理工具——从检测到整理的全流程实现

---

基于Python的人脸聚类与相似性分析工具：从检测到整理的全流程实现

前言

在数据集中，Speaker的数量统计是非常重要的一个环节。基于此，我开发了一个自动化的人脸聚类与相似性分析工具，能够快速处理目录中的所有图片，检测人脸、提取特征、识别相似人脸并自动分组，提供重复照片删除功能。本文将详细介绍这个工具的实现原理、核心技术及代码结构，帮助大家理解从人脸检测到照片整理的全流程技术细节。

模型及算法介绍

该工具整合了多个计算机视觉领域的经典模型和算法，核心技术栈包括人脸检测、特征提取、相似性计算和聚类分析四部分：

1. 人脸检测：MTCNN

MTCNN（Multi-task Cascaded Convolutional Networks）是一种多任务级联卷积神经网络，能够同时完成人脸检测、关键点定位等任务。相比传统的人脸检测算法，MTCNN具有检测速度快、准确率高的特点，尤其在处理遮挡、多角度人脸时表现优异。

在本工具中，MTCNN的主要作用是定位图片中的人脸位置，输出人脸边界框（bbox），为后续的特征提取提供准确的区域范围。

2. 人脸特征提取：InsightFace

InsightFace是一个开源的人脸分析工具包，内置了高性能的人脸特征提取模型。它能够将人脸图像转换为固定维度的特征向量（嵌入向量），同一人的不同照片会生成相似的向量，而不同人的向量差异较大。

工具中使用InsightFace的FaceAnalysis模块提取特征，生成的特征向量具有良好的区分性，为后续的相似性计算奠定基础。

3. 相似性计算：余弦相似度

余弦相似度是衡量两个向量方向差异的指标，取值范围为[-1, 1]，值越接近1表示向量方向越相似。对于人脸特征向量，余弦相似度能够有效反映两张人脸的相似程度：

$$\text{cosine\_similarity}(A,B)=\frac{A\cdot B}{||A||\cdot ||B||}$$

工具中设置了相似度阈值（默认0.65），超过该阈值的人脸被判定为同一人。

4. 聚类算法：基于相似度矩阵的分组

在得到所有图片的特征向量后，工具通过计算全量特征的相似度矩阵，构建相似人脸对，再通过贪心聚类算法将相似的人脸合并为组。这种方法虽然简单，但在中小规模图片集（数千张）上效率较高，且能保证聚类结果的可解释性。

代码实现介绍

工具的核心代码封装在FaceRecognitionSystem类中，配合多进程处理提高效率，整体结构清晰，分为初始化、图片处理、相似性分析、结果处理和辅助分析五大模块。

1. 初始化与多进程配置

为了提高大规模图片处理的效率，工具使用多进程并行处理图片。由于深度学习模型无法直接在多进程间共享，因此通过initialize_worker函数为每个进程单独初始化MTCNN和InsightFace模型：

def initialize_worker():"""初始化每个工作进程的模型"""global detector, appif detector is None:detector = MTCNN()  # 初始化MTCNN人脸检测器if app is None:app = FaceAnalysis(providers=['CPUExecutionProvider'])  # 初始化InsightFaceapp.prepare(ctx_id=0, det_size=(640, 640))

2. 图片处理流程

单张图片处理（process_image函数）

该函数是多进程的核心任务单元，负责：

• 读取图片并转换为RGB格式（MTCNN要求输入为RGB）
• 使用MTCNN检测人脸，获取边界框
• 使用InsightFace提取人脸特征向量
• 返回图片路径、特征向量和边界框（若检测到人脸）

def process_image(image_path):try:image = cv2.imread(image_path)if image is None:return Nonergb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)  # BGR转RGBresults = detector.detect_faces(rgb_image)  # MTCNN检测人脸if results:faces = app.get(rgb_image)  # InsightFace提取特征if faces:x, y, w, h = results[0]['box']  # 取第一个人脸框return image_path, faces[0].embedding, (int(x), int(y), int(w), int(h))except Exception as e:print(f"处理图片 {image_path} 时出错: {str(e)}")return None

目录批量处理（process_directory方法）

该方法遍历指定目录下的所有图片文件，使用进程池并行调用process_image，收集有效结果（包含人脸的图片）并存储特征向量、路径和边界框：

def process_directory(self, input_dir):image_extensions = ['.jpg', '.jpeg', '.png', '.bmp']image_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f)) and os.path.splitext(f)[1].lower() in image_extensions]with Pool(processes=2, initializer=initialize_worker) as pool:  # 2个进程并行results = list(tqdm(pool.imap_unordered(process_image, image_files), total=len(image_files), desc="处理图片"))for result in results:if result:image_path, feature, bbox = resultself.face_features.append(feature)self.file_paths.append(image_path)self.bboxes.append(bbox)

3. 相似人脸分析（find_similar_faces方法）

该方法是核心分析逻辑，分为三步：

1. 计算所有特征向量的余弦相似度矩阵
2. 基于阈值筛选相似人脸对
3. 聚类相似人脸，形成群组并计算组内平均相似度

def find_similar_faces(self):if len(self.face_features) < 2:return []# 计算余弦相似度矩阵features_array = np.array(self.face_features)similarity_matrix = cosine_similarity(features_array)# 聚类相似人脸groups = []used = set()for i in range(len(self.face_features)):if i not in used:group = [i]used.add(i)# 寻找所有与当前人脸相似的人脸for j in range(len(self.face_features)):if j not in used and similarity_matrix[i][j] > self.similarity_threshold:group.append(j)used.add(j)if len(group) > 1:groups.append(group)# 转换为文件路径并计算组内平均相似度result_groups = []for group in groups:group_files = [self.file_paths[idx] for idx in group]avg_sim = self._calculate_group_similarity(group, similarity_matrix)result_groups.append({'files': group_files, 'average_similarity': avg_sim})return sorted(result_groups, key=lambda x: x['average_similarity'], reverse=True)

4. 结果处理功能

保存分组结果（save_similar_groups方法）

将相似人脸组保存到不同文件夹，并生成CSV结果文件，包含每组的图片路径和平均相似度；未分组的图片保存到ungrouped目录。

    def save_similar_groups(self, output_dir, similar_groups):"""将相似的人脸保存到不同的文件夹:param output_dir: 输出目录:param similar_groups: 相似人脸组"""if not os.path.exists(output_dir):os.makedirs(output_dir)# 保存结果到CSVresults = []for i, group in enumerate(similar_groups):group_dir = os.path.join(output_dir, f"group_{i+1}_similarity_{group['average_similarity']:.4f}")os.makedirs(group_dir, exist_ok=True)for file_path in group['files']:file_name = os.path.basename(file_path)dest_path = os.path.join(group_dir, file_name)shutil.copy2(file_path, dest_path)results.append({'group': i+1,'file': file_name,'similarity': group['average_similarity']})# 保存未分组的文件（没有找到相似人脸的）all_group_files = set()for group in similar_groups:# 注意：这里有一个小bug，应该是 group['files']all_group_files.update(file for file in group['files'])ungrouped_dir = os.path.join(output_dir, "ungrouped")os.makedirs(ungrouped_dir, exist_ok=True)for file_path in self.file_paths:if file_path not in all_group_files:file_name = os.path.basename(file_path)dest_path = os.path.join(ungrouped_dir, file_name)shutil.copy2(file_path, dest_path)results.append({'group': 'ungrouped','file': file_name,'similarity': 0.0})# 保存结果到CSVdf = pd.DataFrame(results)df.to_csv(os.path.join(output_dir, "similarity_results.csv"), index=False)print(f"结果已保存到 {output_dir} 目录")

重复照片删除（delete_one_duplicate_per_group方法）

从每个相似组中随机删除一张照片，减少重复存储（需用户确认）。

    def delete_one_duplicate_per_group(self, similar_groups):"""从每个相似组中随机删除一张照片。:param similar_groups: 相似人脸组"""if not similar_groups:print("没有发现相似的照片组，无需删除。")returnprint("\n开始随机删除每个组中的一张重复照片...")deleted_count = 0for i, group in enumerate(similar_groups):if len(group['files']) > 1:# 随机选择一张要删除的照片file_to_delete = random.choice(group['files'])try:os.remove(file_to_delete)print(f"组 {i+1}: 已删除照片 -> {os.path.basename(file_to_delete)}")deleted_count += 1except OSError as e:print(f"删除文件 {file_to_delete} 时出错: {e}")print(f"\n总共删除了 {deleted_count} 张重复照片。")

结果清理（clear_results方法）

删除生成的结果目录，方便重新处理。

    def clear_results(self, output_dir):"""清除生成的结果目录。:param output_dir: 输出目录"""if os.path.exists(output_dir):try:shutil.rmtree(output_dir)print(f"\n已成功清除结果目录: {output_dir}")except OSError as e:print(f"清除目录 {output_dir} 时出错: {e}")else:print("\n结果目录不存在，无需清除。")

6. 主函数逻辑

主函数支持两种模式（skip=0重新处理图片，skip=1从已有结果加载），用户可通过输入目录、输出目录和相似度阈值进行配置，并支持交互式选择是否删除重复照片和清理结果。

if __name__ == "__main__":# 配置script_dir = os.path.dirname(os.path.abspath(__file__))INPUT_DIRECTORY = os.path.join(script_dir, "face_dir")  # 替换为你的图片目录OUTPUT_DIRECTORY = os.path.join(script_dir, "face_recognition_results")SIMILARITY_THRESHOLD = 0.65  # 可根据需要调整# skip=0: 重新处理图片 | skip=1: 跳过处理，直接从结果文件操作skip = 1face_system = FaceRecognitionSystem(similarity_threshold=SIMILARITY_THRESHOLD)similar_groups = []if skip == 1:print("模式: skip=1。尝试从现有结果文件加载...")csv_path = os.path.join(OUTPUT_DIRECTORY, "similarity_results.csv")if os.path.exists(csv_path):df = pd.read_csv(csv_path)# 筛选出已分组的记录grouped_df = df[df['group'] != 'ungrouped'].copy()# pd.to_numeric to handle potential string group IDsgrouped_df['group'] = pd.to_numeric(grouped_df['group'])for group_id, group_data in grouped_df.groupby('group'):# 从文件名重建完整路径files = [os.path.join(INPUT_DIRECTORY, fname) for fname in group_data['file']]avg_sim = group_data['similarity'].iloc[0]similar_groups.append({'files': files,'average_similarity': avg_sim})print(f"成功从 {csv_path} 加载了 {len(similar_groups)} 个相似组。")else:print(f"错误: 结果文件 {csv_path} 未找到。")print("将自动切换到 skip=0 模式重新处理图片。")# skip = 0 # 强制切换模式if skip == 0:print("模式: skip=0。开始处理图片...")# 创建系统实例face_system.process_directory(INPUT_DIRECTORY)# 查找相似的人脸similar_groups = face_system.find_similar_faces()print(f"\n发现 {len(similar_groups)} 组可能包含同一人的照片")# 保存结果if similar_groups:face_system.save_similar_groups(OUTPUT_DIRECTORY, similar_groups)# --- 后续操作 ---if similar_groups:# 询问是否删除重复照片choice_delete = input("\n是否要从每个相似组中随机删除一张照片? (y/n): ").lower()if choice_delete in ['y', 'yes']:face_system.delete_one_duplicate_per_group(similar_groups)else:print("\n没有发现可操作的相似组。")# 询问是否清除结果choice_clear = input("\n是否要清除本次运行生成的结果? (y/n): ").lower()if choice_clear in ['y', 'yes']:face_system.clear_results(OUTPUT_DIRECTORY)

完整代码

import os
import cv2
import numpy as np
import pandas as pd
from mtcnn import MTCNN
from insightface.app import FaceAnalysis
from sklearn.metrics.pairwise import cosine_similarity
import shutil
from tqdm import tqdm
from multiprocessing import Pool, cpu_count
import random# 全局变量，用于在多进程中初始化模型
detector = None
app = Nonedef initialize_worker():"""初始化每个工作进程的模型"""global detector, appif detector is None:detector = MTCNN()if app is None:app = FaceAnalysis(providers=['CPUExecutionProvider'])app.prepare(ctx_id=0, det_size=(640, 640))def process_image(image_path):"""处理单个图片，检测人脸并提取特征（设计为在多进程中运行）:param image_path: 图片路径:return: (image_path, feature, bbox) 或 None; bbox=(x,y,w,h)"""try:image = cv2.imread(image_path)if image is None:return Nonergb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)results = detector.detect_faces(rgb_image)if results:faces = app.get(rgb_image)if faces:# MTCNN与insightface可能检测顺序不同，这里优先使用MTCNN第一个框x, y, w, h = results[0]['box']return image_path, faces[0].embedding, (int(x), int(y), int(w), int(h))except Exception as e:print(f"处理图片 {image_path} 时出错: {str(e)}")return Noneclass FaceRecognitionSystem:def __init__(self, similarity_threshold=0.65):"""初始化人脸检测和识别系统:param similarity_threshold: 判定为同一人的相似度阈值"""# 相似度阈值，超过此值判定为同一人self.similarity_threshold = similarity_threshold# 存储人脸特征和对应的文件路径self.face_features = []self.file_paths = []self.valid_files = []self.bboxes = []  # 与file_paths一一对应的人脸框 (x,y,w,h)# 年龄性别模型相关self.age_net = Noneself.gender_net = Noneself.detector_net = Noneself.age_list = ['(0-2)', '(4-6)', '(8-12)', '(15-20)', '(25-32)', '(38-43)', '(48-53)', '(60-100)']self.gender_list = ['Male', 'Female']self.age_gender_loaded = Falsedef process_directory(self, input_dir):"""使用多进程处理目录中的所有图片:param input_dir: 图片所在目录"""image_extensions = ['.jpg', '.jpeg', '.png', '.bmp']image_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f)) and os.path.splitext(f)[1].lower() in image_extensions]# 在这里设置要使用的进程数 (max_workers)# 设置为 None 将自动使用所有可用的CPU核心max_workers = 2num_to_print = max_workers if max_workers is not None else cpu_count()print(f"发现 {len(image_files)} 个图片文件，使用 {num_to_print} 个进程开始处理...")# 创建进程池with Pool(processes=max_workers, initializer=initialize_worker) as pool:# 使用imap_unordered来获取进度条results = list(tqdm(pool.imap_unordered(process_image, image_files), total=len(image_files),desc="处理图片"))for result in results:if result:image_path, feature, bbox = resultself.face_features.append(feature)self.file_paths.append(image_path)self.valid_files.append(os.path.basename(image_path))self.bboxes.append(bbox)print(f"成功处理 {len(self.face_features)} 张包含人脸的图片")def find_similar_faces(self):"""找出相似的人脸（可能是同一个人）:return: 相似人脸组的列表"""if len(self.face_features) < 2:return []# 计算所有特征之间的余弦相似度features_array = np.array(self.face_features)similarity_matrix = cosine_similarity(features_array)# 找出相似的人脸对similar_pairs = []n = len(self.face_features)for i in range(n):for j in range(i + 1, n):if similarity_matrix[i][j] > self.similarity_threshold:similar_pairs.append((self.file_paths[i], self.file_paths[j], similarity_matrix[i][j]))# 聚类相似的人脸，形成群组groups = []used = set()for i in range(n):if i not in used:group = [i]used.add(i)# 寻找所有与当前人脸相似的人脸for j in range(n):if j not in used and similarity_matrix[i][j] > self.similarity_threshold:group.append(j)used.add(j)if len(group) > 1:  # 只保留有多个相似人脸的组groups.append(group)# 将索引转换为文件路径result_groups = []for group in groups:group_files = [self.file_paths[idx] for idx in group]# 计算组内平均相似度avg_sim = self._calculate_group_similarity(group, similarity_matrix)result_groups.append({'files': group_files,'average_similarity': avg_sim})# 按组内平均相似度排序result_groups.sort(key=lambda x: x['average_similarity'], reverse=True)return result_groupsdef _calculate_group_similarity(self, group, similarity_matrix):"""计算组内平均相似度"""if len(group) <= 1:return 0.0total = 0.0count = 0for i in range(len(group)):for j in range(i + 1, len(group)):total = group[i]idx_j = group[j]total += similarity_matrix[idx_j][idx_j]count += 1return total / count if count > 0 else 0.0def save_similar_groups(self, output_dir, similar_groups):"""将相似的人脸保存到不同的文件夹:param output_dir: 输出目录:param similar_groups: 相似人脸组"""if not os.path.exists(output_dir):os.makedirs(output_dir)# 保存结果到CSVresults = []for i, group in enumerate(similar_groups):group_dir = os.path.join(output_dir, f"group_{i+1}_similarity_{group['average_similarity']:.4f}")os.makedirs(group_dir, exist_ok=True)for file_path in group['files']:file_name = os.path.basename(file_path)dest_path = os.path.join(group_dir, file_name)shutil.copy2(file_path, dest_path)results.append({'group': i+1,'file': file_name,'similarity': group['average_similarity']})# 保存未分组的文件（没有找到相似人脸的）all_group_files = set()for group in similar_groups:# 注意：这里有一个小bug，应该是 group['files']all_group_files.update(file for file in group['files'])ungrouped_dir = os.path.join(output_dir, "ungrouped")os.makedirs(ungrouped_dir, exist_ok=True)for file_path in self.file_paths:if file_path not in all_group_files:file_name = os.path.basename(file_path)dest_path = os.path.join(ungrouped_dir, file_name)shutil.copy2(file_path, dest_path)results.append({'group': 'ungrouped','file': file_name,'similarity': 0.0})# 保存结果到CSVdf = pd.DataFrame(results)df.to_csv(os.path.join(output_dir, "similarity_results.csv"), index=False)print(f"结果已保存到 {output_dir} 目录")def delete_one_duplicate_per_group(self, similar_groups):"""从每个相似组中随机删除一张照片。:param similar_groups: 相似人脸组"""if not similar_groups:print("没有发现相似的照片组，无需删除。")returnprint("\n开始随机删除每个组中的一张重复照片...")deleted_count = 0for i, group in enumerate(similar_groups):if len(group['files']) > 1:# 随机选择一张要删除的照片file_to_delete = random.choice(group['files'])try:os.remove(file_to_delete)print(f"组 {i+1}: 已删除照片 -> {os.path.basename(file_to_delete)}")deleted_count += 1except OSError as e:print(f"删除文件 {file_to_delete} 时出错: {e}")print(f"\n总共删除了 {deleted_count} 张重复照片。")def clear_results(self, output_dir):"""清除生成的结果目录。:param output_dir: 输出目录"""if os.path.exists(output_dir):try:shutil.rmtree(output_dir)print(f"\n已成功清除结果目录: {output_dir}")except OSError as e:print(f"清除目录 {output_dir} 时出错: {e}")else:print("\n结果目录不存在，无需清除。")def ensure_age_gender_models(self, base_dir):"""加载年龄性别检测相关模型 (只加载一次)现在模型目录定位为: (当前脚本目录的上一级)/age_gender"""if self.age_gender_loaded:return Truetry:# 原来 base_dir 是 face 子目录，需要取上一级目录project_root = os.path.dirname(base_dir)  # 上一级目录 (Lip_reading)model_dir = os.path.join(project_root, 'age_gender')age_prototxt = os.path.join(model_dir, 'age_deploy.prototxt')age_caffemodel = os.path.join(model_dir, 'age_net.caffemodel')gender_prototxt = os.path.join(model_dir, 'gender_deploy.prototxt')gender_caffemodel = os.path.join(model_dir, 'gender_net.caffemodel')face_pb = os.path.join(model_dir, 'opencv_face_detector_uint8.pb')face_pbtxt = os.path.join(model_dir, 'opencv_face_detector.pbtxt')needed = [age_prototxt, age_caffemodel, gender_prototxt, gender_caffemodel, face_pb, face_pbtxt]for p in needed:if not os.path.exists(p):print(f"缺少模型文件: {p}，跳过年龄性别统计。")return Falseself.age_net = cv2.dnn.readNet(age_caffemodel, age_prototxt)self.gender_net = cv2.dnn.readNet(gender_caffemodel, gender_prototxt)self.detector_net = cv2.dnn.readNetFromTensorflow(face_pb, face_pbtxt)self.age_gender_loaded = Trueprint(f"年龄/性别模型加载完成。(模型目录: {model_dir})")return Trueexcept Exception as e:print(f"加载年龄性别模型失败: {e}")return Falsedef predict_age_gender_for_image(self, image_path):"""对单张图片预测第一张人脸的年龄段与性别。返回 (gender, age_range) 或 (None, None)"""if not self.age_gender_loaded:return (None, None)img = cv2.imread(image_path)if img is None:return (None, None)h, w = img.shape[:2]blob = cv2.dnn.blobFromImage(img, 1.0, (300, 300), [104, 117, 123], False, False)self.detector_net.setInput(blob)detections = self.detector_net.forward()best_conf = 0face_box = Nonefor i in range(detections.shape[2]):confidence = detections[0, 0, i, 2]if confidence > 0.6 and confidence > best_conf:box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])x1, y1, x2, y2 = box.astype(int)x1, y1 = max(0, x1), max(0, y1)x2, y2 = min(w - 1, x2), min(h - 1, y2)face_box = (x1, y1, x2, y2)best_conf = confidenceif face_box is None:return (None, None)x1, y1, x2, y2 = face_boxface_roi = img[y1:y2, x1:x2].copy()if face_roi.size == 0:return (None, None)blob_face = cv2.dnn.blobFromImage(face_roi, 1.0, (227, 227), (78.4263377603, 87.7689143744, 114.895847746), swapRB=False)# genderself.gender_net.setInput(blob_face)gender_preds = self.gender_net.forward()gender = self.gender_list[gender_preds[0].argmax()]# ageself.age_net.setInput(blob_face)age_preds = self.age_net.forward()age_range = self.age_list[age_preds[0].argmax()]return (gender, age_range)def analyze_age_gender(self, base_dir, output_dir, similar_groups):"""对所有图片进行年龄性别统计，并输出两个文件：1) age_gender_details.csv: 每张图 -> 文件名, group_id, gender, age_range2) age_gender_stats.txt: 汇总统计"""if not self.ensure_age_gender_models(base_dir):returnrecords = []# 构建文件到组的映射file_to_group = {}for idx, g in enumerate(similar_groups, start=1):for f in g['files']:file_to_group[os.path.basename(f)] = idx# 如果没有groups，也统计所有现有file_pathstarget_files = self.file_paths if self.file_paths else []if not target_files:print("没有可用于年龄性别统计的图片。")returnprint("开始进行年龄/性别统计...")gender_counter = {}age_counter = {}for path in tqdm(target_files, desc="Age/Gender"):gender, age_range = self.predict_age_gender_for_image(path)base_name = os.path.basename(path)group_id = file_to_group.get(base_name, 'ungrouped')records.append({'file': base_name,'group_id': group_id,'gender': gender if gender else 'Unknown','age_range': age_range if age_range else 'Unknown'})if gender:gender_counter[gender] = gender_counter.get(gender, 0) + 1else:gender_counter['Unknown'] = gender_counter.get('Unknown', 0) + 1if age_range:age_counter[age_range] = age_counter.get(age_range, 0) + 1else:age_counter['Unknown'] = age_counter.get('Unknown', 0) + 1# 保存明细os.makedirs(output_dir, exist_ok=True)details_path = os.path.join(output_dir, 'age_gender_details.csv')pd.DataFrame(records).to_csv(details_path, index=False)# 汇总stats_lines = ["Gender Statistics:"]for k,v in gender_counter.items():stats_lines.append(f"  {k}: {v}")stats_lines.append("\nAge Range Statistics:")for k,v in age_counter.items():stats_lines.append(f"  {k}: {v}")stats_path = os.path.join(output_dir, 'age_gender_stats.txt')with open(stats_path, 'w', encoding='utf-8') as fw:fw.write('\n'.join(stats_lines))print(f"年龄/性别统计完成，已保存到: {details_path} 与 {stats_path}")if __name__ == "__main__":# 配置script_dir = os.path.dirname(os.path.abspath(__file__))INPUT_DIRECTORY = os.path.join(script_dir, "face_dir")  # 替换为你的图片目录OUTPUT_DIRECTORY = os.path.join(script_dir, "face_recognition_results")SIMILARITY_THRESHOLD = 0.65  # 可根据需要调整# skip=0: 重新处理图片 | skip=1: 跳过处理，直接从结果文件操作skip = 1face_system = FaceRecognitionSystem(similarity_threshold=SIMILARITY_THRESHOLD)similar_groups = []if skip == 1:print("模式: skip=1。尝试从现有结果文件加载...")csv_path = os.path.join(OUTPUT_DIRECTORY, "similarity_results.csv")if os.path.exists(csv_path):df = pd.read_csv(csv_path)# 筛选出已分组的记录grouped_df = df[df['group'] != 'ungrouped'].copy()# pd.to_numeric to handle potential string group IDsgrouped_df['group'] = pd.to_numeric(grouped_df['group'])for group_id, group_data in grouped_df.groupby('group'):# 从文件名重建完整路径files = [os.path.join(INPUT_DIRECTORY, fname) for fname in group_data['file']]avg_sim = group_data['similarity'].iloc[0]similar_groups.append({'files': files,'average_similarity': avg_sim})print(f"成功从 {csv_path} 加载了 {len(similar_groups)} 个相似组。")else:print(f"错误: 结果文件 {csv_path} 未找到。")print("将自动切换到 skip=0 模式重新处理图片。")# skip = 0 # 强制切换模式if skip == 0:print("模式: skip=0。开始处理图片...")# 创建系统实例face_system.process_directory(INPUT_DIRECTORY)# 查找相似的人脸similar_groups = face_system.find_similar_faces()print(f"\n发现 {len(similar_groups)} 组可能包含同一人的照片")# 保存结果if similar_groups:face_system.save_similar_groups(OUTPUT_DIRECTORY, similar_groups)# --- 后续操作 ---if similar_groups:# 询问是否删除重复照片choice_delete = input("\n是否要从每个相似组中随机删除一张照片? (y/n): ").lower()if choice_delete in ['y', 'yes']:face_system.delete_one_duplicate_per_group(similar_groups)else:print("\n没有发现可操作的相似组。")# 询问是否清除结果choice_clear = input("\n是否要清除本次运行生成的结果? (y/n): ").lower()if choice_clear in ['y', 'yes']:face_system.clear_results(OUTPUT_DIRECTORY)

总结

本工具通过整合MTCNN、InsightFace等经典模型，结合余弦相似度和聚类算法，实现了从人脸检测、特征提取到相似性分析的全流程自动化。主要优势包括：

1. 高效性：多进程并行处理大幅提升大规模图片的处理速度；
2. 实用性：支持相似人脸分组、重复删除和年龄性别分析，满足日常照片整理需求；
3. 可扩展性：代码结构清晰，可通过调整相似度阈值、更换模型或优化聚类算法进一步提升性能。

希望这个工具能帮助大家更高效地管理照片，也为学习人脸识别技术的同学提供一个可参考的实践案例。