构建AI智能体：数据预处理对训练效果的影响：质量过滤、敏感内容过滤与数据去重

# 数据预处理的价值可视化
import matplotlib.pyplot as plt
import numpy as np

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 创建影响分析图
fig, ax = plt.subplots(1, 2, figsize=(14, 6))

# 左图：预处理对训练效率的影响
stages = ['原始数据', '质量过滤后', '敏感过滤后', '去重后']
efficiency = [1.0, 1.8, 2.2, 3.1]  # 相对训练效率

bars = ax[0].bar(stages, efficiency, color=['#ff9999', '#66b3ff', '#99ff99', '#ffcc99'])
ax[0].set_title('数据预处理对训练效率的影响', fontsize=14, fontweight='bold')
ax[0].set_ylabel('相对训练效率')
ax[0].set_ylim(0, 3.5)

# 在柱子上添加数值
for bar, eff in zip(bars, efficiency):
    height = bar.get_height()
    ax[0].text(bar.get_x() + bar.get_width()/2., height + 0.1,
               f'{eff:.1f}x', ha='center', va='bottom')

# 右图：预处理对模型性能的影响
metrics = ['语言理解', '事实准确性', '安全性', '推理能力']
raw_data = [0.65, 0.55, 0.45, 0.60]
processed_data = [0.85, 0.82, 0.95, 0.78]

x = np.arange(len(metrics))
width = 0.35

ax[1].bar(x - width/2, raw_data, width, label='原始数据', alpha=0.7)
ax[1].bar(x + width/2, processed_data, width, label='预处理后', alpha=0.7)
ax[1].set_title('预处理对模型性能的影响', fontsize=14, fontweight='bold')
ax[1].set_ylabel('性能评分')
ax[1].set_xticks(x)
ax[1].set_xticklabels(metrics)
ax[1].legend()
ax[1].set_ylim(0, 1.0)

plt.tight_layout()
plt.show()

import language_tool_python

# 初始化工具（默认英语）
tool = language_tool_python.LanguageTool('en-US')

# 检查文本
text = "I are a student. He don't like apples."
matches = tool.check(text)

print(f"发现 {len(matches)} 个错误:")
for match in matches:
    print(f"- {match.ruleId}: {match.message}")
    print(f"  建议: {match.replacements}")
    print(f"  位置: {match.offset}-{match.offset+match.errorLength}")
    print()

# 自动修正
corrected_text = tool.correct(text)
print(f"修正后的文本: {corrected_text}")

import matplotlib.pyplot as plt
import re
import numpy as np
from collections import Counter
import language_tool_python

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class DataQualityFilter:
    """数据质量过滤器"""
    
    def __init__(self):
        self.gtool = language_tool_python.LanguageTool('en-US')
        # 中文停用词列表
        self.chinese_stopwords = set([
            '的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'
        ])
    
    def calculate_text_quality_score(self, text):
        """计算文本质量综合评分"""
        scores = {}
        
        # 1. 语言规范性评分
        scores['grammar'] = self._grammar_quality(text)
        
        # 2. 信息密度评分
        scores['info_density'] = self._information_density(text)
        
        # 3. 可读性评分
        scores['readability'] = self._readability_score(text)
        
        # 4. 内容完整性评分
        scores['completeness'] = self._content_completeness(text)
        
        # 综合评分（加权平均）
        weights = {'grammar': 0.3, 'info_density': 0.3, 
                  'readability': 0.2, 'completeness': 0.2}
        total_score = sum(scores[metric] * weights[metric] for metric in scores)
        
        return total_score, scores
    
    def _grammar_quality(self, text):
        """语法质量评估"""
        if len(text) < 10:
            return 0.3
            
        # 使用language-tool检查语法错误
        try:
            matches = self.gtool.check(text)
            error_rate = len(matches) / len(text.split())
            grammar_score = max(0, 1 - error_rate * 10)  # 错误率转换为分数
        except:
            grammar_score = 0.5  # 检查失败时给中间分
            
        return min(1.0, grammar_score)
    
    def _information_density(self, text):
        """信息密度评估"""
        words = re.findall(r'[\u4e00-\u9fff]', text)  # 中文字符
        total_chars = len(text)
        chinese_chars = len(words)
        
        if total_chars == 0:
            return 0.0
            
        # 中文比例
        chinese_ratio = chinese_chars / total_chars
        
        # 停用词比例
        stopword_count = sum(1 for word in words if word in self.chinese_stopwords)
        stopword_ratio = stopword_count / len(words) if words else 0
        
        # 信息密度 = 中文比例 * (1 - 停用词比例)
        info_density = chinese_ratio * (1 - stopword_ratio)
        
        return min(1.0, info_density * 2)  # 适当缩放
    
    def _readability_score(self, text):
        """可读性评分（简化版）"""
        # 计算平均句长
        sentences = re.split(r'[。！？!?]', text)
        sentences = [s for s in sentences if len(s.strip()) > 0]
        
        if not sentences:
            return 0.3
            
        avg_sentence_length = sum(len(s) for s in sentences) / len(sentences)
        
        # 理想句长在15-25字之间
        if 15 <= avg_sentence_length <= 25:
            readability = 0.9
        elif 10 <= avg_sentence_length < 15 or 25 < avg_sentence_length <= 35:
            readability = 0.7
        else:
            readability = 0.4
            
        return readability
    
    def _content_completeness(self, text):
        """内容完整性评估"""
        # 检查文本是否包含完整的句子结构
        has_start = any(mark in text for mark in ['。', '!', '?', '…'])
        has_content = len(text.strip()) >= 10
        
        if has_start and has_content:
            return 0.8
        elif has_content:
            return 0.5
        else:
            return 0.2
    
    def filter_by_quality(self, texts, threshold=0.6):
        """基于质量阈值过滤文本"""
        filtered_texts = []
        quality_scores = []
        
        for text in texts:
            score, detailed_scores = self.calculate_text_quality_score(text)
            if score >= threshold:
                filtered_texts.append(text)
                quality_scores.append(score)
        
        return filtered_texts, quality_scores

# 测试质量过滤
def test_quality_filter():
    """测试质量过滤器"""
    sample_texts = [
        "这是一个高质量的文本示例。它包含完整的信息和良好的语法结构。",
        "垃圾文本啊啊啊啊啊啊啊啊啊啊啊啊啊啊",  # 低质量
        "The quick brown fox jumps over the lazy dog.",  # 英文
        "今天天气很好。",  # 过短
        "基于深度学习的自然语言处理技术在近年来取得了显著进展，特别是在大语言模型方面。",  # 高质量
        "这个那个的嗯啊哦呃",  # 低信息密度
    ]
    
    filter = DataQualityFilter()
    filtered_texts, scores = filter.filter_by_quality(sample_texts, threshold=0.5)
    
    print("质量过滤结果:")
    print("=" * 50)
    for i, (text, score) in enumerate(zip(filtered_texts, scores)):
        print(f"{i+1}. 评分: {score:.3f}")
        print(f"   文本: {text}")
        print("-" * 30)

test_quality_filter()

def visualize_quality_filtering():
    """可视化质量过滤效果"""
    
    # 模拟数据
    np.random.seed(42)
    n_samples = 1000
    
    # 生成模拟质量分数分布
    raw_scores = np.random.beta(2, 5, n_samples)  # 原始数据质量分布
    threshold = 0.6
    filtered_scores = raw_scores[raw_scores >= threshold]
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
    fig.suptitle('文本质量过滤效果分析', fontsize=16, fontweight='bold')
    
    # 1. 质量分数分布对比
    axes[0,0].hist(raw_scores, bins=30, alpha=0.7, label='原始数据', color='red')
    axes[0,0].hist(filtered_scores, bins=15, alpha=0.7, label=f'过滤后(>={threshold})', color='green')
    axes[0,0].axvline(threshold, color='black', linestyle='--', label=f'阈值={threshold}')
    axes[0,0].set_title('质量分数分布对比')
    axes[0,0].set_xlabel('质量分数')
    axes[0,0].set_ylabel('频次')
    axes[0,0].legend()
    axes[0,0].grid(True, alpha=0.3)
    
    # 2. 过滤效果统计
    categories = ['原始数据', '过滤后数据']
    counts = [len(raw_scores), len(filtered_scores)]
    retention_rate = len(filtered_scores) / len(raw_scores) * 100
    
    bars = axes[0,1].bar(categories, counts, color=['red', 'green'])
    axes[0,1].set_title('数据量变化')
    axes[0,1].set_ylabel('样本数量')
    
    # 在柱子上添加数值
    for bar, count in zip(bars, counts):
        height = bar.get_height()
        axes[0,1].text(bar.get_x() + bar.get_width()/2., height + 10,
                      f'{count}', ha='center', va='bottom')
    
    axes[0,1].text(0.5, max(counts)*0.8, f'保留率: {retention_rate:.1f}%', 
                  ha='center', va='center', fontsize=12, 
                  bbox=dict(boxstyle="round,pad=0.3", facecolor="lightblue"))
    
    # 3. 质量提升分析
    quality_metrics = ['语法规范性', '信息密度', '可读性', '内容完整性']
    raw_quality = [0.45, 0.38, 0.52, 0.41]  # 原始数据平均质量
    filtered_quality = [0.82, 0.75, 0.79, 0.83]  # 过滤后平均质量
    
    x = np.arange(len(quality_metrics))
    width = 0.35
    
    axes[1,0].bar(x - width/2, raw_quality, width, label='原始数据', alpha=0.7)
    axes[1,0].bar(x + width/2, filtered_quality, width, label='过滤后', alpha=0.7)
    axes[1,0].set_title('各维度质量提升')
    axes[1,0].set_ylabel('质量分数')
    axes[1,0].set_xticks(x)
    axes[1,0].set_xticklabels(quality_metrics)
    axes[1,0].legend()
    axes[1,0].grid(True, alpha=0.3)
    
    # 4. 训练效果预估
    epochs = range(1, 6)
    raw_performance = [0.3, 0.45, 0.55, 0.62, 0.67]  # 原始数据训练效果
    filtered_performance = [0.4, 0.58, 0.72, 0.81, 0.85]  # 过滤后训练效果
    
    axes[1,1].plot(epochs, raw_performance, 'o-', label='原始数据', linewidth=2)
    axes[1,1].plot(epochs, filtered_performance, 's-', label='过滤后数据', linewidth=2)
    axes[1,1].set_title('训练效果对比')
    axes[1,1].set_xlabel('训练轮次')
    axes[1,1].set_ylabel('模型性能')
    axes[1,1].legend()
    axes[1,1].grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
    print(f"质量过滤统计:")
    print(f"原始数据量: {len(raw_scores)}")
    print(f"过滤后数据量: {len(filtered_scores)}")
    print(f"数据保留率: {retention_rate:.1f}%")
    print(f"平均质量提升: {(np.mean(filtered_scores) - np.mean(raw_scores)):.3f}")

visualize_quality_filtering()

import matplotlib.pyplot as plt
import numpy as np
import re
from collections import defaultdict
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
class SensitiveContentFilter:
    """敏感内容过滤器"""
    
    def __init__(self):
        self._initialize_sensitive_patterns()
    
    def _initialize_sensitive_patterns(self):
        """初始化敏感模式"""
        # 敏感关键词库（示例）
        self.sensitive_keywords = {
            'violence': ['暴力', '恐怖', '袭击', '杀人', '武器'],
            'pornography': ['色情', '淫秽', '性爱', '成人内容'],
            'discrimination': ['种族歧视', '性别歧视', '地域黑'],
            'illegal': ['毒品', '赌博', '诈骗', '非法'],
            'privacy': ['身份证', '手机号', '银行卡', '密码']
        }
        
        # 正则表达式模式
        self.patterns = {
            'phone': r'1[3-9]\d{9}',
            'id_card': r'[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[1-2]\d|3[0-1])\d{3}[\dXx]',
            'bank_card': r'\d{16,19}',
            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
        }
    
    def detect_sensitive_content(self, text):
        """检测敏感内容"""
        detection_results = {
            'sensitive_level': 0,  # 0: 安全, 1: 警告, 2: 危险
            'categories': [],
            'details': defaultdict(list),
            'risk_score': 0.0
        }
        
        # 1. 关键词检测
        keyword_matches = self._keyword_detection(text)
        detection_results['details'].update(keyword_matches)
        
        # 2. 隐私信息检测
        privacy_matches = self._privacy_detection(text)
        detection_results['details'].update(privacy_matches)
        
        # 3. 计算风险分数
        risk_score = self._calculate_risk_score(detection_results['details'])
        detection_results['risk_score'] = risk_score
        
        # 4. 确定敏感等级
        if risk_score >= 0.7:
            detection_results['sensitive_level'] = 2
            detection_results['categories'] = ['high_risk']
        elif risk_score >= 0.3:
            detection_results['sensitive_level'] = 1
            detection_results['categories'] = ['medium_risk']
        else:
            detection_results['sensitive_level'] = 0
            detection_results['categories'] = ['safe']
        
        return detection_results
    
    def _keyword_detection(self, text):
        """关键词检测"""
        matches = defaultdict(list)
        
        for category, keywords in self.sensitive_keywords.items():
            for keyword in keywords:
                if keyword in text:
                    matches[category].append(keyword)
        
        return matches
    
    def _privacy_detection(self, text):
        """隐私信息检测"""
        matches = defaultdict(list)
        
        for pattern_name, pattern in self.patterns.items():
            found_items = re.findall(pattern, text)
            if found_items:
                # 对找到的隐私信息进行掩码处理
                masked_items = [self._mask_sensitive_info(item, pattern_name) 
                              for item in found_items]
                matches[pattern_name] = masked_items
        
        return matches
    
    def _mask_sensitive_info(self, info, info_type):
        """掩码敏感信息"""
        if info_type == 'phone':
            return info[:3] + '****' + info[-4:]
        elif info_type == 'id_card':
            return str(info)[:6] + '********' + str(info)[-4:]
        elif info_type == 'bank_card':
            return info[:6] + '******' + info[-4:]
        else:
            return '***'  # 其他类型简单掩码
    
    def _calculate_risk_score(self, detection_details):
        """计算风险分数"""
        total_risk = 0
        max_risk = 0
        
        # 不同类别的风险权重
        risk_weights = {
            'violence': 0.9,
            'pornography': 0.8,
            'discrimination': 0.7,
            'illegal': 0.9,
            'privacy': 0.6,
            'phone': 0.5,
            'id_card': 0.8,
            'bank_card': 0.7,
            'email': 0.4
        }
        
        for category, items in detection_details.items():
            if items:  # 如果检测到内容
                weight = risk_weights.get(category, 0.5)
                # 基于检测到的项目数量调整风险
                count_risk = min(len(items) * 0.1, 0.3)
                total_risk += weight + count_risk
                max_risk += weight + 0.3  # 最大可能风险
        
        if max_risk == 0:
            return 0.0
        
        return min(total_risk / max_risk, 1.0)
    
    def filter_sensitive_content(self, texts, threshold=0.3):
        """过滤敏感内容"""
        safe_texts = []
        filtered_count = 0
        risk_scores = []
        
        for text in texts:
            detection = self.detect_sensitive_content(text)
            
            if detection['risk_score'] < threshold:
                safe_texts.append(text)
                risk_scores.append(detection['risk_score'])
            else:
                filtered_count += 1
        
        return safe_texts, filtered_count, risk_scores

# 测试敏感内容过滤
def test_sensitive_filter():
    """测试敏感内容过滤器"""
    test_texts = [
        "这是一个安全的内容示例。",
        "这篇文章包含暴力内容，描述袭击事件。",
        "用户手机号是13812345678，请妥善保管。",
        "身份证号码：110101199003075678",
        "这是关于深度学习的正常讨论。",
        "包含赌博和毒品的不法内容。"
    ]
    
    filter = SensitiveContentFilter()
    safe_texts, filtered_count, risk_scores = filter.filter_sensitive_content(test_texts)
    
    print("敏感内容过滤结果:")
    print("=" * 50)
    print(f"原始文本数量: {len(test_texts)}")
    print(f"安全文本数量: {len(safe_texts)}")
    print(f"过滤文本数量: {filtered_count}")
    print("\n安全文本:")
    for i, text in enumerate(safe_texts, 1):
        print(f"{i}. {text} (风险分: {risk_scores[i-1]:.3f})")

test_sensitive_filter()


def visualize_sensitive_filtering():
    """可视化敏感内容过滤效果"""
    
    # 模拟数据
    np.random.seed(42)
    n_samples = 500
    
    # 生成模拟风险分数分布
    risk_scores = np.random.beta(1, 8, n_samples)  # 大多数是低风险
    # 添加一些高风险样本
    high_risk_indices = np.random.choice(n_samples, 50, replace=False)
    risk_scores[high_risk_indices] = np.random.uniform(0.6, 1.0, 50)
    
    threshold = 0.3
    safe_scores = risk_scores[risk_scores < threshold]
    filtered_scores = risk_scores[risk_scores >= threshold]
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
    fig.suptitle('敏感内容过滤效果分析', fontsize=16, fontweight='bold')
    
    # 1. 风险分数分布
    axes[0,0].hist(risk_scores, bins=30, alpha=0.7, color='red', label='所有数据')
    axes[0,0].hist(safe_scores, bins=15, alpha=0.7, color='green', label='安全数据')
    axes[0,0].axvline(threshold, color='black', linestyle='--', 
                     label=f'阈值={threshold}')
    axes[0,0].set_title('风险分数分布')
    axes[0,0].set_xlabel('风险分数')
    axes[0,0].set_ylabel('频次')
    axes[0,0].legend()
    axes[0,0].grid(True, alpha=0.3)
    
    # 2. 过滤统计
    categories = ['安全数据', '过滤数据']
    counts = [len(safe_scores), len(filtered_scores)]
    safety_rate = len(safe_scores) / len(risk_scores) * 100
    
    bars = axes[0,1].bar(categories, counts, color=['green', 'red'])
    axes[0,1].set_title('安全数据统计')
    axes[0,1].set_ylabel('样本数量')
    
    for bar, count in zip(bars, counts):
        height = bar.get_height()
        axes[0,1].text(bar.get_x() + bar.get_width()/2., height + 5,
                      f'{count}', ha='center', va='bottom')
    
    axes[0,1].text(0.5, max(counts)*0.8, f'安全率: {safety_rate:.1f}%', 
                  ha='center', va='center', fontsize=12,
                  bbox=dict(boxstyle="round,pad=0.3", facecolor="lightgreen"))
    
    # 3. 风险类别分布（模拟）
    risk_categories = ['暴力内容', '色情信息', '歧视言论', '隐私泄露', '违法信息']
    category_counts = [45, 38, 52, 67, 29]
    
    axes[1,0].bar(risk_categories, category_counts, color='orange', alpha=0.7)
    axes[1,0].set_title('检测到的风险类别分布')
    axes[1,0].set_ylabel('检测次数')
    axes[1,0].tick_params(axis='x', rotation=45)
    axes[1,0].grid(True, alpha=0.3)
    
    # 4. 过滤效果对模型安全性的影响
    safety_metrics = ['有害内容生成', '偏见放大', '隐私泄露', '合规性']
    before_filtering = [0.35, 0.42, 0.28, 0.45]  # 过滤前风险
    after_filtering = [0.08, 0.12, 0.05, 0.92]   # 过滤后风险/合规性
    
    x = np.arange(len(safety_metrics))
    width = 0.35
    
    bars1 = axes[1,1].bar(x - width/2, before_filtering, width, 
                         label='过滤前', alpha=0.7, color='red')
    bars2 = axes[1,1].bar(x + width/2, after_filtering, width, 
                         label='过滤后', alpha=0.7, color='green')
    axes[1,1].set_title('模型安全性提升')
    axes[1,1].set_ylabel('风险分数(越低越好) / 合规性(越高越好)')
    axes[1,1].set_xticks(x)
    axes[1,1].set_xticklabels(safety_metrics)
    axes[1,1].legend()
    axes[1,1].grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
    print(f"敏感内容过滤统计:")
    print(f"总数据量: {len(risk_scores)}")
    print(f"安全数据量: {len(safe_scores)}")
    print(f"过滤数据量: {len(filtered_scores)}")
    print(f"数据安全率: {safety_rate:.1f}%")
    print(f"平均风险降低: {(np.mean(risk_scores) - np.mean(safe_scores)):.3f}")

visualize_sensitive_filtering()

import matplotlib.pyplot as plt
import numpy as np
from collections import defaultdict
import hashlib
from datasketch import MinHash, MinHashLSH

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class DataDeduplicator:
    """数据去重器"""
    
    def __init__(self, threshold=0.8):
        self.threshold = threshold
        self.lsh = MinHashLSH(threshold=threshold, num_perm=128)
        self.doc_hashes = {}
    
    def create_minhash(self, text, num_perm=128):
        """创建文本的MinHash签名"""
        minhash = MinHash(num_perm=num_perm)
        
        # 中文文本按字符分割，英文按单词分割
        if any('\u4e00' <= char <= '\u9fff' for char in text):
            # 中文文本
            tokens = list(text)
        else:
            # 英文文本
            tokens = text.lower().split()
        
        for token in tokens:
            minhash.update(token.encode('utf-8'))
        
        return minhash
    
    def exact_deduplicate(self, texts):
        """精确去重 - 基于MD5哈希"""
        seen_hashes = set()
        unique_texts = []
        
        for text in texts:
            text_hash = hashlib.md5(text.strip().encode('utf-8')).hexdigest()
            if text_hash not in seen_hashes:
                seen_hashes.add(text_hash)
                unique_texts.append(text)
        
        return unique_texts
    
    def fuzzy_deduplicate(self, texts):
        """模糊去重 - 基于MinHash和LSH"""
        unique_texts = []
        duplicate_groups = defaultdict(list)
        
        for i, text in enumerate(texts):
            minhash = self.create_minhash(text)
            
            # 查询相似文档
            similar_docs = self.lsh.query(minhash)
            
            if not similar_docs:
                # 没有相似文档，添加为新文档
                self.lsh.insert(str(i), minhash)
                self.doc_hashes[str(i)] = minhash
                unique_texts.append(text)
            else:
                # 找到相似文档，记录重复信息
                duplicate_groups[similar_docs[0]].append(i)
        
        return unique_texts, duplicate_groups
    
    def calculate_similarity(self, text1, text2):
        """计算两篇文本的相似度"""
        minhash1 = self.create_minhash(text1)
        minhash2 = self.create_minhash(text2)
        return minhash1.jaccard(minhash2)
    
    def analyze_duplication(self, texts):
        """分析数据重复情况"""
        print("分析数据重复情况...")
        
        # 精确去重分析
        exact_unique = self.exact_deduplicate(texts)
        exact_duplicate_rate = 1 - len(exact_unique) / len(texts)
        
        # 模糊去重分析（采样分析）
        sample_size = min(1000, len(texts))
        sample_texts = texts[:sample_size]
        
        similarity_matrix = np.zeros((sample_size, sample_size))
        for i in range(sample_size):
            for j in range(i+1, sample_size):
                sim = self.calculate_similarity(sample_texts[i], sample_texts[j])
                similarity_matrix[i, j] = sim
                similarity_matrix[j, i] = sim
        
        # 计算模糊重复率
        high_similarity_pairs = np.sum(similarity_matrix > self.threshold) / 2
        total_possible_pairs = sample_size * (sample_size - 1) / 2
        fuzzy_duplicate_rate = high_similarity_pairs / total_possible_pairs if total_possible_pairs > 0 else 0
        
        return {
            'total_documents': len(texts),
            'exact_unique': len(exact_unique),
            'exact_duplicate_rate': exact_duplicate_rate,
            'fuzzy_duplicate_rate': fuzzy_duplicate_rate,
            'sample_similarity_matrix': similarity_matrix
        }

# 测试数据去重
def test_deduplication():
    """测试数据去重"""
    # 创建测试数据
    base_text = "深度学习是机器学习的一个分支，它使用多层神经网络来学习数据的表示。"
    
    test_texts = [
        base_text,  # 完全重复
        base_text,  # 完全重复
        "深度学习是机器学习的一个分支，它使用多层神经网络。",  # 高度相似
        "机器学习中的深度学习分支使用神经网络。",  # 相似
        "今天天气很好，适合出去散步。",  # 不相关
        "深度学习利用多层神经网络学习数据特征。",  # 相似
        "深度学习是机器学习的一个分支，它使用多层神经网络来学习数据的表示。"  # 完全重复
    ]
    
    deduplicator = DataDeduplicator(threshold=0.7)
    
    # 精确去重
    exact_unique = deduplicator.exact_deduplicate(test_texts)
    print("精确去重结果:")
    print(f"原始数量: {len(test_texts)}")
    print(f"去重后数量: {len(exact_unique)}")
    print(f"重复率: {(1 - len(exact_unique)/len(test_texts))*100:.1f}%")
    
    # 模糊去重
    fuzzy_unique, duplicate_groups = deduplicator.fuzzy_deduplicate(test_texts)
    print(f"\n模糊去重结果:")
    print(f"去重后数量: {len(fuzzy_unique)}")
    print(f"重复组: {dict(duplicate_groups)}")
    
    # 相似度分析
    print(f"\n相似度分析:")
    for i in range(min(3, len(test_texts))):
        for j in range(i+1, min(4, len(test_texts))):
            sim = deduplicator.calculate_similarity(test_texts[i], test_texts[j])
            print(f"文本{i}和文本{j}的相似度: {sim:.3f}")

test_deduplication()

def visualize_deduplication_impact():
    """可视化数据去重效果"""
    
    # 模拟数据
    np.random.seed(42)
    
    # 模拟不同规模数据集的重复情况
    dataset_sizes = [1000, 5000, 10000, 50000, 100000]
    exact_duplicate_rates = [0.15, 0.12, 0.18, 0.25, 0.30]  # 精确重复率
    fuzzy_duplicate_rates = [0.35, 0.40, 0.45, 0.50, 0.55]  # 模糊重复率
    
    # 计算去重后的数据量
    exact_remaining = [size * (1 - rate) for size, rate in zip(dataset_sizes, exact_duplicate_rates)]
    fuzzy_remaining = [size * (1 - rate) for size, rate in zip(dataset_sizes, fuzzy_duplicate_rates)]
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
    fig.suptitle('数据去重效果分析', fontsize=16, fontweight='bold')
    
    # 1. 重复率随数据规模变化
    axes[0,0].plot(dataset_sizes, exact_duplicate_rates, 'o-', 
                  label='精确重复率', linewidth=2, markersize=8, color='red')
    axes[0,0].plot(dataset_sizes, fuzzy_duplicate_rates, 's-', 
                  label='模糊重复率', linewidth=2, markersize=8, color='blue')
    axes[0,0].set_title('重复率 vs 数据规模', fontsize=12)
    axes[0,0].set_xlabel('数据规模')
    axes[0,0].set_ylabel('重复率')
    axes[0,0].set_xscale('log')
    axes[0,0].legend()
    axes[0,0].grid(True, alpha=0.3)
    
    # 2. 去重前后数据量对比
    x = np.arange(len(dataset_sizes))
    width = 0.35
    
    bars1 = axes[0,1].bar(x - width/2, dataset_sizes, width, 
                         label='原始数据量', alpha=0.7, color='lightgray')
    bars2 = axes[0,1].bar(x + width/2, exact_remaining, width, 
                         label='精确去重后', alpha=0.7, color='lightgreen')
    
    axes[0,1].set_title('去重前后数据量对比', fontsize=12)
    axes[0,1].set_xlabel('数据集')
    axes[0,1].set_ylabel('数据量')
    axes[0,1].set_xticks(x)
    axes[0,1].set_xticklabels([f'{size//1000}K' for size in dataset_sizes])
    axes[0,1].legend()
    axes[0,1].grid(True, alpha=0.3)
    
    # 在柱状图上添加数值
    for i, (orig, remain) in enumerate(zip(dataset_sizes, exact_remaining)):
        reduction = (orig - remain) / orig * 100
        axes[0,1].text(i, orig + max(dataset_sizes)*0.02, f'-{reduction:.0f}%', 
                      ha='center', va='bottom', fontsize=9)
    
    # 3. 去重对训练效果的影响
    training_metrics = ['训练速度', '内存使用', '模型性能', '收敛速度']
    before_dedup = [1.0, 1.0, 1.0, 1.0]  # 基准值
    after_exact_dedup = [1.8, 0.6, 1.1, 1.3]  # 精确去重后
    after_fuzzy_dedup = [2.5, 0.4, 1.2, 1.6]  # 模糊去重后
    
    x = np.arange(len(training_metrics))
    width = 0.25
    
    axes[1,0].bar(x - width, before_dedup, width, label='去重前', alpha=0.7, color='red')
    axes[1,0].bar(x, after_exact_dedup, width, label='精确去重后', alpha=0.7, color='orange')
    axes[1,0].bar(x + width, after_fuzzy_dedup, width, label='模糊去重后', alpha=0.7, color='green')
    
    axes[1,0].set_title('去重对训练效果的影响', fontsize=12)
    axes[1,0].set_xlabel('指标')
    axes[1,0].set_ylabel('相对改善 (倍数)')
    axes[1,0].set_xticks(x)
    axes[1,0].set_xticklabels(training_metrics)
    axes[1,0].legend()
    axes[1,0].grid(True, alpha=0.3)
    
    # 4. 重复类型分布
    duplicate_types = ['完全重复', '近似重复', '段落重复', '语义重复']
    type_distribution = [35, 25, 20, 20]  # 百分比
    
    colors = ['#ff9999', '#66b3ff', '#99ff99', '#ffcc99']
    wedges, texts, autotexts = axes[1,1].pie(type_distribution, 
                                            labels=duplicate_types, 
                                            autopct='%1.1f%%',
                                            colors=colors,
                                            startangle=90)
    
    # 美化饼图
    for autotext in autotexts:
        autotext.set_color('white')
        autotext.set_fontweight('bold')
    
    axes[1,1].set_title('重复类型分布', fontsize=12)
    
    plt.tight_layout()
    plt.show()
    
    # 输出统计信息
    print("数据去重效果统计:")
    print("=" * 50)
    for i, size in enumerate(dataset_sizes):
        exact_reduction = exact_duplicate_rates[i] * 100
        fuzzy_reduction = fuzzy_duplicate_rates[i] * 100
        print(f"数据集 {size:,} 条:")
        print(f"  - 精确重复: {exact_reduction:.1f}% → 保留 {exact_remaining[i]:,.0f} 条")
        print(f"  - 模糊重复: {fuzzy_reduction:.1f}% → 保留 {fuzzy_remaining[i]:,.0f} 条")
        print(f"  - 总去重率: {exact_reduction + fuzzy_reduction:.1f}%")
        print()

# 运行可视化
visualize_deduplication_impact()

import matplotlib.pyplot as plt
import numpy as np
from collections import defaultdict
import hashlib
from datasketch import MinHash, MinHashLSH

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

def detailed_deduplication_simulation():
    """详细的数据去重模拟分析"""
    
    # 模拟不同类型数据的重复模式
    np.random.seed(42)
    
    # 创建模拟数据集
    data_types = ['新闻数据', '社交媒体', '学术论文', '网页内容', '对话数据']
    n_types = len(data_types)
    
    # 各种数据类型的重复特征
    exact_dup_rates = [0.08, 0.25, 0.05, 0.35, 0.15]  # 精确重复率
    fuzzy_dup_rates = [0.15, 0.30, 0.10, 0.25, 0.35]  # 模糊重复率
    data_volumes = [50000, 100000, 20000, 150000, 80000]  # 数据量
    
    fig, axes = plt.subplots(2, 2, figsize=(16, 12))
    fig.suptitle('多类型数据去重深度分析', fontsize=16, fontweight='bold')
    
    # 1. 各数据类型重复率对比
    x = np.arange(n_types)
    width = 0.35
    
    bars1 = axes[0,0].bar(x - width/2, exact_dup_rates, width, 
                         label='精确重复率', alpha=0.7, color='salmon')
    bars2 = axes[0,0].bar(x + width/2, fuzzy_dup_rates, width, 
                         label='模糊重复率', alpha=0.7, color='lightblue')
    
    axes[0,0].set_title('各数据类型重复率对比', fontsize=12)
    axes[0,0].set_xlabel('数据类型')
    axes[0,0].set_ylabel('重复率')
    axes[0,0].set_xticks(x)
    axes[0,0].set_xticklabels(data_types)
    axes[0,0].legend()
    axes[0,0].grid(True, alpha=0.3)
    
    # 在柱子上添加数值
    for bars in [bars1, bars2]:
        for bar in bars:
            height = bar.get_height()
            axes[0,0].text(bar.get_x() + bar.get_width()/2., height + 0.01,
                          f'{height:.2f}', ha='center', va='bottom', fontsize=9)
    
    # 2. 去重节省的资源分析
    storage_saved = [vol * rate for vol, rate in zip(data_volumes, 
                                                    [e+f for e,f in zip(exact_dup_rates, fuzzy_dup_rates)])]
    training_time_saved = [rate * 100 for rate in 
                          [e+f for e,f in zip(exact_dup_rates, fuzzy_dup_rates)]]
    
    x = np.arange(n_types)
    ax2 = axes[0,1]
    ax2_twin = ax2.twinx()
    
    bars_storage = ax2.bar(x - width/2, storage_saved, width, 
                          label='节省存储(条)', alpha=0.7, color='lightgreen')
    bars_time = ax2_twin.bar(x + width/2, training_time_saved, width, 
                            label='节省时间(%)', alpha=0.7, color='orange')
    
    ax2.set_title('去重资源节省分析', fontsize=12)
    ax2.set_xlabel('数据类型')
    ax2.set_ylabel('节省存储(条)')
    ax2_twin.set_ylabel('节省训练时间(%)')
    ax2.set_xticks(x)
    ax2.set_xticklabels(data_types)
    
    # 合并图例
    lines1, labels1 = ax2.get_legend_handles_labels()
    lines2, labels2 = ax2_twin.get_legend_handles_labels()
    ax2.legend(lines1 + lines2, labels1 + labels2, loc='upper left')
    ax2.grid(True, alpha=0.3)
    
    # 3. 去重前后数据质量变化
    quality_metrics = ['数据多样性', '信息密度', '训练稳定性', '模型泛化性']
    before_quality = [0.6, 0.5, 0.7, 0.65]  # 去重前
    after_quality = [0.85, 0.82, 0.90, 0.88]  # 去重后
    
    angles = np.linspace(0, 2*np.pi, len(quality_metrics), endpoint=False).tolist()
    before_quality += before_quality[:1]
    after_quality += after_quality[:1]
    angles += angles[:1]
    
    ax3 = plt.subplot(2, 2, 3, polar=True)
    ax3.plot(angles, before_quality, 'o-', linewidth=2, label='去重前', color='red')
    ax3.fill(angles, before_quality, alpha=0.25, color='red')
    ax3.plot(angles, after_quality, 'o-', linewidth=2, label='去重后', color='green')
    ax3.fill(angles, after_quality, alpha=0.25, color='green')
    
    ax3.set_xticks(angles[:-1])
    ax3.set_xticklabels(quality_metrics)
    ax3.set_ylim(0, 1)
    ax3.set_title('去重对数据质量的影响', fontsize=12)
    ax3.legend(bbox_to_anchor=(1.1, 1.0))
    
    # 4. 不同去重阈值的效果
    thresholds = [0.5, 0.6, 0.7, 0.8, 0.9]
    data_retention = [0.95, 0.85, 0.70, 0.50, 0.30]  # 数据保留率
    quality_improvement = [0.1, 0.25, 0.45, 0.65, 0.75]  # 质量提升
    training_speedup = [1.2, 1.5, 2.0, 2.8, 3.5]  # 训练加速
    
    ax4 = axes[1,1]
    ax4_twin = ax4.twinx()
    
    line1 = ax4.plot(thresholds, data_retention, 's-', label='数据保留率', 
                    linewidth=2, markersize=6, color='blue')
    line2 = ax4_twin.plot(thresholds, quality_improvement, 'o-', label='质量提升', 
                         linewidth=2, markersize=6, color='green')
    line3 = ax4_twin.plot(thresholds, training_speedup, '^-', label='训练加速', 
                         linewidth=2, markersize=6, color='red')
    
    ax4.set_title('不同去重阈值效果对比', fontsize=12)
    ax4.set_xlabel('去重相似度阈值')
    ax4.set_ylabel('数据保留率')
    ax4_twin.set_ylabel('质量提升/训练加速')
    ax4.grid(True, alpha=0.3)
    
    # 合并图例
    lines = line1 + line2 + line3
    labels = [l.get_label() for l in lines]
    ax4.legend(lines, labels, loc='center right')
    
    # 标记最优阈值
    optimal_idx = 2  # 0.7阈值
    ax4.axvline(thresholds[optimal_idx], color='black', linestyle='--', alpha=0.7)
    ax4.text(thresholds[optimal_idx], 0.5, '推荐阈值', 
            ha='center', va='center', rotation=90,
            bbox=dict(boxstyle="round,pad=0.3", facecolor="yellow"))
    
    plt.tight_layout()
    plt.show()
    
    # 输出详细分析报告
    print("多类型数据去重深度分析报告:")
    print("=" * 60)
    total_original = sum(data_volumes)
    total_saved = sum(storage_saved)
    overall_saving_rate = total_saved / total_original * 100
    
    print(f"总数据量: {total_original:,} 条")
    print(f"预计去重节省: {total_saved:,} 条 ({overall_saving_rate:.1f}%)")
    print(f"推荐去重阈值: 0.7 (平衡数据质量与数量)")
    print("\n各数据类型分析:")
    for i, dtype in enumerate(data_types):
        saving_rate = (exact_dup_rates[i] + fuzzy_dup_rates[i]) * 100
        print(f"- {dtype}: {saving_rate:.1f}% 重复率，节省 {storage_saved[i]:,} 条")

# 运行详细分析
detailed_deduplication_simulation()