---

基于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, app
    if detector is None:
        detector = MTCNN()  # 初始化MTCNN人脸检测器
    if app is None:
        app = FaceAnalysis(providers=['CPUExecutionProvider'])  # 初始化InsightFace
        app.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 None
        rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)  # BGR转RGB
        results = 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 = result
            self.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)
        
        # 保存结果到CSV
        results = []
        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
                })
        
        # 保存结果到CSV
        df = 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("没有发现相似的照片组，无需删除。")
            return

        print("\n开始随机删除每个组中的一张重复照片...")
        deleted_count = 0
        for 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 += 1
                except 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 = 1
    
    face_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 IDs
            grouped_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 = None

def initialize_worker():
    """初始化每个工作进程的模型"""
    global detector, app
    if 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 None
        rgb_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 None

class 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 = None
        self.gender_net = None
        self.detector_net = None
        self.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 = False
        
    def 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 = 2
        
        num_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 = result
                self.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_groups
    
    def _calculate_group_similarity(self, group, similarity_matrix):
        """计算组内平均相似度"""
        if len(group) <= 1:
            return 0.0
            
        total = 0.0
        count = 0
        
        for 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 += 1
        
        return total / count if count > 0 else 0.0
    
    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)
        
        # 保存结果到CSV
        results = []
        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
                })
        
        # 保存结果到CSV
        df = 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("没有发现相似的照片组，无需删除。")
            return

        print("\n开始随机删除每个组中的一张重复照片...")
        deleted_count = 0
        for 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 += 1
                except 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 True
        try:
            # 原来 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 False

            self.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 = True
            print(f"年龄/性别模型加载完成。(模型目录: {model_dir})")
            return True
        except Exception as e:
            print(f"加载年龄性别模型失败: {e}")
            return False

    def 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 = 0
        face_box = None
        for 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 = confidence
        if face_box is None:
            return (None, None)
        x1, y1, x2, y2 = face_box
        face_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)
        # gender
        self.gender_net.setInput(blob_face)
        gender_preds = self.gender_net.forward()
        gender = self.gender_list[gender_preds[0].argmax()]
        # age
        self.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_range
        2) age_gender_stats.txt: 汇总统计
        """
        if not self.ensure_age_gender_models(base_dir):
            return
        records = []
        # 构建文件到组的映射
        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_paths
        target_files = self.file_paths if self.file_paths else []

        if not target_files:
            print("没有可用于年龄性别统计的图片。")
            return
        print("开始进行年龄/性别统计...")
        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) + 1
            else:
                gender_counter['Unknown'] = gender_counter.get('Unknown', 0) + 1
            if age_range:
                age_counter[age_range] = age_counter.get(age_range, 0) + 1
            else:
                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 = 1
    
    face_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 IDs
            grouped_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. 可扩展性：代码结构清晰，可通过调整相似度阈值、更换模型或优化聚类算法进一步提升性能。

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