Captured at (local ISO): 2026-05-18 05:16:52

Python-Based Face Clustering and Similarity Analysis Tool: Full Pipeline from Detection to Organization

Introduction

In many datasets, counting speakers is an important step. Building on that need, I developed an automated face clustering and similarity analysis tool that can quickly process all images in a directory: detect faces, extract embeddings, identify similar faces and group them automatically, and optionally remove duplicate photos. This article explains how the tool works — the core techniques and code structure — so readers can follow the full pipeline from face detection through photo organization.

Models and Algorithms

The tool combines several classic models and algorithms from computer vision. The technical stack has four main parts: face detection, feature extraction, similarity computation, and clustering.

1. Face Detection: MTCNN

MTCNN (Multi-task Cascaded Convolutional Networks) is a multi-task cascaded convolutional network that performs face detection and landmark localization jointly. Compared with traditional face detectors, MTCNN is fast and accurate, and it handles occlusion and varied poses well.

In this tool, MTCNN locates faces in each image and outputs bounding boxes (bbox), providing precise regions for downstream feature extraction.

2. Face Feature Extraction: InsightFace

InsightFace is an open-source face analysis toolkit with high-performance embedding models. It maps a face crop to a fixed-dimensional feature vector (embedding): images of the same person yield similar vectors, while different people tend to differ more strongly.

The tool uses InsightFace’s FaceAnalysis module for embeddings. The resulting vectors are discriminative enough to support reliable similarity scoring.

3. Similarity Calculation: Cosine Similarity

Cosine similarity measures how aligned two vectors are. Its range is [-1, 1]; values closer to 1 mean more similar directions. For face embeddings, cosine similarity reflects how similar two faces appear:

cosine_similarity ( A , B )

A ⋅ B ∣ ∣ A ∣ ∣ ⋅ ∣ ∣ B ∣ ∣ \text{cosine\_similarity}(A,B) = \frac{A \cdot B}{||A|| \cdot ||B||} cosine_similarity(A,B)=∣∣A∣∣⋅∣∣B∣∣A⋅B​

The tool uses a similarity threshold (default 0.65). Pairs above the threshold are treated as the same identity.

4. Clustering Algorithm: Grouping via Similarity Matrix

After embeddings are computed for all images, the tool builds a full pairwise similarity matrix, derives similar face pairs, and merges them with a greedy clustering procedure. The approach is simple but efficient on small-to-medium collections (thousands of images) and keeps results easy to interpret.

Code Implementation Overview

The core logic lives in the FaceRecognitionSystem class, with multiprocessing for throughput. The design splits into initialization, image processing, similarity analysis, result handling, and auxiliary analysis.

1. Initialization and Multiprocessing Configuration

To speed up large batches, images are processed in parallel. Deep models cannot be shared trivially across processes, so initialize_worker loads MTCNN and InsightFace per worker process:

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. Image Processing Pipeline

Single Image Processing (process_image function)

This function is the main worker task. It:

• Loads the image and converts it to RGB (required by MTCNN)
• Runs MTCNN to obtain bounding boxes
• Uses InsightFace to extract a face embedding
• Returns the image path, embedding, and bbox when a face is found

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

Directory Batch Processing (process_directory method)

This method walks the given directory for image files, runs process_image in parallel via a process pool, and collects valid results (paths with detected faces), storing embeddings, paths, and boxes:

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. Similar Face Analysis (find_similar_faces method)

This is the core analysis routine. It:

1. Builds the cosine similarity matrix over all embeddings
2. Filters pairs above the threshold
3. Clusters similar faces into groups and computes average within-group similarity

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. Result Handling Features

Saving Group Results (save_similar_groups method)

Similar-face groups are copied into separate folders, and a CSV is written with paths and average similarity per group. Images that never matched anyone go into an ungrouped folder.

    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} 目录")

Duplicate Photo Deletion (delete_one_duplicate_per_group method)

Randomly deletes one image per similar group to reduce redundant storage (requires user confirmation).

    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} 张重复照片。")

Result Cleanup (clear_results method)

Deletes generated output directories so you can rerun from scratch.

    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. Main Function Logic

The main entry supports two modes (skip=0 reprocesses images; skip=1 loads existing CSV results). Users configure input/output directories and the similarity threshold, then interactively choose whether to delete duplicates and clear outputs.

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)

Complete Code

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)

Summary

By combining classic components such as MTCNN and InsightFace with cosine similarity and clustering, this tool automates the full path from detection and embedding extraction to similarity grouping. Highlights:

1. Efficiency: Multiprocessing speeds up large batches substantially.
2. Practicality: Face grouping, optional duplicate removal, and age/gender statistics cover typical photo-cleanup workflows.
3. Extensibility: The layout makes it straightforward to tune thresholds, swap models, or refine clustering.

I hope this helps you organize photos faster and gives learners of face recognition a concrete reference implementation.