""" NER Anonymization Evaluator for Hugging Face Spaces ابزار ارزیابی استاندارد سیستمهای ناشناسسازی با NER Author: Your Name Version: 1.0.1 License: MIT """ import pandas as pd import numpy as np import re from typing import Dict, List, Tuple import gradio as gr from datetime import datetime import io import tempfile import os # ==================== Import seqeval ==================== try: from seqeval.metrics import ( classification_report, f1_score, precision_score, recall_score, accuracy_score ) from seqeval.scheme import IOB2 SEQEVAL_AVAILABLE = True except ImportError: SEQEVAL_AVAILABLE = False print("⚠️ Warning: seqeval not installed. Only Exact Match will be available.") # ==================== Main Evaluator Class ==================== class StandardNEREvaluator: """ ارزیابی استاندارد Named Entity Recognition این کلاس دو روش ارزیابی ارائه میدهد: 1. seqeval: استاندارد علمی با IOB2 tagging 2. Exact Match: مقایسه مستقیم شناسهها """ def __init__(self): """مقداردهی اولیه""" self.results_df = None # انواع entity های پشتیبانی شده self.entity_types = ['COMPANY', 'PERSON', 'AMOUNT', 'PERCENT', 'GROUP', 'STOCK'] # الگوهای regex برای تشخیص entities self.patterns = { 'COMPANY': [ r'company-(\d+)', r'Company-(\d+)', r'COMPANY-(\d+)', r'COMPANY_(\d+)(?:_[A-Z]+)?', r'company_(\d+)(?:_[a-z]+)?' ], 'PERSON': [ r'person-(\d+)', r'Person-(\d+)', r'PERSON-(\d+)', r'PERSON_(\d+)(?:_[A-Z]+)?', r'person_(\d+)(?:_[a-z]+)?' ], 'AMOUNT': [ r'amount-(\d+)', r'Amount-(\d+)', r'AMOUNT-(\d+)', r'AMOUNT_(\d+)(?:_[A-Z]+)?', r'amount_(\d+)(?:_[a-z]+)?' ], 'PERCENT': [ r'percent-(\d+)', r'Percent-(\d+)', r'PERCENT-(\d+)', r'PERCENT_(\d+)(?:_[A-Z]+)?', r'percent_(\d+)(?:_[a-z]+)?' ], 'GROUP': [ r'group-(\d+)', r'Group-(\d+)', r'GROUP-(\d+)', r'GROUP_(\d+)(?:_[A-Z]+)?', r'group_(\d+)(?:_[a-z]+)?' ], 'STOCK': [ r'stock-(\d+)', r'Stock-(\d+)', r'STOCK-(\d+)', r'STOCK_SYMBOL_(\d+)(?:_[A-Z]+)?', r'stock_symbol_(\d+)(?:_[a-z]+)?' ] } def tokenize_text(self, text: str) -> List[str]: """ تبدیل متن به توکنها (کلمات) Args: text: متن ورودی Returns: لیست توکنها """ if pd.isna(text) or not isinstance(text, str): return [] return text.split() def text_to_iob2_tags(self, text: str) -> List[str]: """ تبدیل متن به فرمت IOB2 Tagging IOB2 Format: - B-TYPE: Beginning of entity - I-TYPE: Inside entity (continuation) - O: Outside (not an entity) Args: text: متن ورودی Returns: لیست تگهای IOB2 """ if pd.isna(text) or not isinstance(text, str): return [] tokens = self.tokenize_text(text) tags = ['O'] * len(tokens) # پیدا کردن entities در متن for entity_type, pattern_list in self.patterns.items(): for pattern in pattern_list: for match in re.finditer(pattern, text): start_pos = match.start() end_pos = match.end() # پیدا کردن توکنهایی که entity در آنها است current_pos = 0 for i, token in enumerate(tokens): token_start = text.find(token, current_pos) token_end = token_start + len(token) if token_start >= start_pos and token_end <= end_pos: if tags[i] == 'O': # اولین توکن: B-TYPE if token_start == start_pos or i == 0 or tags[i-1].split('-')[-1] != entity_type: tags[i] = f'B-{entity_type}' # توکنهای بعدی: I-TYPE else: tags[i] = f'I-{entity_type}' current_pos = token_end return tags def evaluate_with_seqeval(self, reference_text: str, predicted_text: str) -> Dict: """ ارزیابی با seqeval (روش استاندارد) Args: reference_text: متن مرجع predicted_text: متن پیشبینی شده Returns: دیکشنری شامل metrics """ if not SEQEVAL_AVAILABLE: return { 'precision': 0.0, 'recall': 0.0, 'f1': 0.0, 'accuracy': 0.0, 'error': 'seqeval not available' } try: # تبدیل به IOB2 tags y_true = [self.text_to_iob2_tags(reference_text)] y_pred = [self.text_to_iob2_tags(predicted_text)] # اگر هر دو خالی باشند if not y_true[0] and not y_pred[0]: return { 'precision': 1.0, 'recall': 1.0, 'f1': 1.0, 'accuracy': 1.0 } # محاسبه metrics precision = precision_score(y_true, y_pred, scheme=IOB2, mode='strict') recall = recall_score(y_true, y_pred, scheme=IOB2, mode='strict') f1 = f1_score(y_true, y_pred, scheme=IOB2, mode='strict') accuracy = accuracy_score(y_true, y_pred) return { 'precision': round(precision, 4), 'recall': round(recall, 4), 'f1': round(f1, 4), 'accuracy': round(accuracy, 4) } except Exception as e: print(f"خطا در seqeval: {str(e)}") return { 'precision': 0.0, 'recall': 0.0, 'f1': 0.0, 'accuracy': 0.0, 'error': str(e) } def evaluate_with_exact_match(self, reference_text: str, predicted_text: str) -> Dict: """ ارزیابی با Exact Match (روش ساده) Args: reference_text: متن مرجع predicted_text: متن پیشبینی شده Returns: دیکشنری شامل metrics """ def extract_entities(text): """استخراج entities از متن""" entities = set() for entity_type, pattern_list in self.patterns.items(): for pattern in pattern_list: for match in re.finditer(pattern, text): entity_id = match.group(1) entities.add(f"{entity_type}-{entity_id}") return entities ref_entities = extract_entities(reference_text) pred_entities = extract_entities(predicted_text) # محاسبه TP, FP, FN tp = len(ref_entities & pred_entities) fp = len(pred_entities - ref_entities) fn = len(ref_entities - pred_entities) # محاسبه metrics precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0 return { 'precision': round(precision, 4), 'recall': round(recall, 4), 'f1': round(f1, 4), 'tp': tp, 'fp': fp, 'fn': fn } def evaluate_single_row(self, reference_text: str, predicted_text: str) -> Dict: """ ارزیابی یک سطر با هر دو روش Args: reference_text: متن مرجع predicted_text: متن پیشبینی شده Returns: دیکشنری شامل همه metrics """ # روش 1: seqeval seqeval_metrics = self.evaluate_with_seqeval(reference_text, predicted_text) # روش 2: Exact Match exact_metrics = self.evaluate_with_exact_match(reference_text, predicted_text) return { 'seqeval_precision': seqeval_metrics['precision'], 'seqeval_recall': seqeval_metrics['recall'], 'seqeval_f1': seqeval_metrics['f1'], 'seqeval_accuracy': seqeval_metrics['accuracy'], 'exact_precision': exact_metrics['precision'], 'exact_recall': exact_metrics['recall'], 'exact_f1': exact_metrics['f1'], 'tp_count': exact_metrics['tp'], 'fp_count': exact_metrics['fp'], 'fn_count': exact_metrics['fn'] } def evaluate_dataset(self, file_path: str) -> Tuple[bool, str, pd.DataFrame]: """ ارزیابی کل دیتاست Args: file_path: مسیر فایل CSV Returns: (موفقیت, پیام وضعیت, DataFrame نتایج) """ if not SEQEVAL_AVAILABLE: return ( False, "⚠️ seqeval نصب نیست. لطفاً requirements.txt را چک کنید.", pd.DataFrame() ) try: # بارگذاری فایل print(f"📁 در حال خواندن فایل: {file_path}") df = pd.read_csv(file_path, encoding='utf-8-sig') print(f"✅ فایل خوانده شد: {len(df)} سطر") print(f"📋 ستونها: {list(df.columns)}") # تشخیص ستونها با اولویت Reference_text if 'Reference_text' in df.columns and 'anonymized_text' in df.columns: reference_col = 'Reference_text' predicted_col = 'anonymized_text' print(f"✅ حالت 3 ستونه: Reference_text (مرجع) vs anonymized_text (LLM)") elif 'original_text' in df.columns and 'anonymized_text' in df.columns: reference_col = 'original_text' predicted_col = 'anonymized_text' print(f"⚠️ حالت 2 ستونه: original_text به عنوان مرجع") else: available_cols = list(df.columns) return ( False, f"❌ ستونهای مورد نیاز یافت نشد!\n\nستونهای موجود: {available_cols}\n\nستونهای مورد نیاز:\n• Reference_text (مرجع انسانی)\n• anonymized_text (پیشبینی LLM)", pd.DataFrame() ) print(f"🔍 شروع ارزیابی {len(df)} سطر...") # ارزیابی هر سطر results = [] for index, row in df.iterrows(): if (index + 1) % 10 == 0: print(f" پردازش سطر {index + 1}/{len(df)}...") try: metrics = self.evaluate_single_row( str(row[reference_col]), str(row[predicted_col]) ) results.append(metrics) except Exception as e: print(f"⚠️ خطا در سطر {index + 1}: {str(e)}") # افزودن نتایج صفر برای این سطر results.append({ 'seqeval_precision': 0.0, 'seqeval_recall': 0.0, 'seqeval_f1': 0.0, 'seqeval_accuracy': 0.0, 'exact_precision': 0.0, 'exact_recall': 0.0, 'exact_f1': 0.0, 'tp_count': 0, 'fp_count': 0, 'fn_count': 0 }) print(f"✅ ارزیابی کامل شد!") # ایجاد DataFrame نتایج results_df = pd.DataFrame(results) # اضافه کردن ستونهای اصلی for col in df.columns: results_df[col] = df[col].values # ترتیب ستونها: متریکها + سه ستون اصلی + بقیه metric_cols = [ 'seqeval_precision', 'seqeval_recall', 'seqeval_f1', 'seqeval_accuracy', 'exact_precision', 'exact_recall', 'exact_f1', 'tp_count', 'fp_count', 'fn_count' ] # سه ستون اصلی (اگر موجود باشند) main_cols = [] if 'original_text' in results_df.columns: main_cols.append('original_text') if 'Reference_text' in results_df.columns: main_cols.append('Reference_text') if 'anonymized_text' in results_df.columns: main_cols.append('anonymized_text') # بقیه ستونها (اگر چیز دیگری هست) other_cols = [col for col in results_df.columns if col not in metric_cols and col not in main_cols] # ترتیب نهایی: متریکها + ستونهای اصلی + بقیه results_df = results_df[metric_cols + main_cols + other_cols] self.results_df = results_df # محاسبه آمار کلی avg_seqeval_p = results_df['seqeval_precision'].mean() avg_seqeval_r = results_df['seqeval_recall'].mean() avg_seqeval_f1 = results_df['seqeval_f1'].mean() avg_seqeval_acc = results_df['seqeval_accuracy'].mean() avg_exact_f1 = results_df['exact_f1'].mean() total_tp = results_df['tp_count'].sum() total_fp = results_df['fp_count'].sum() total_fn = results_df['fn_count'].sum() # ایجاد پیام وضعیت status = f"""✅ ارزیابی با موفقیت انجام شد! 📊 **نتایج seqeval (استاندارد NER - IOB2 Tagging):** • Precision: {avg_seqeval_p:.4f} • Recall: {avg_seqeval_r:.4f} • F1-Score: {avg_seqeval_f1:.4f} • Accuracy: {avg_seqeval_acc:.4f} 📈 **آمار کلی:** • کل True Positives: {total_tp} • کل False Positives: {total_fp} • کل False Negatives: {total_fn} • تعداد سطرها: {len(df)} 🔬 **مقایسه:** • مرجع (انسانی): {reference_col} • پیشبینی (LLM): {predicted_col} 📊 **مقایسه روشها:** • F1 (seqeval): {avg_seqeval_f1:.4f} • F1 (Exact): {avg_exact_f1:.4f} • اختلاف: {abs(avg_seqeval_f1 - avg_exact_f1):.4f} 📋 **ستونهای خروجی:** • 10 متریک ارزیابی (seqeval & exact match) • original_text - متن خام اصلی • Reference_text - ناشناسسازی انسانی (مرجع) • anonymized_text - ناشناسسازی LLM (پیشبینی) 💾 **نکته:** فایل CSV دانلودی شامل همه {len(df)} سطر و تمام ستونها است ✅ این ارزیابی مطابق با استانداردهای CoNLL-2003 است""" return True, status, results_df except Exception as e: import traceback error_details = traceback.format_exc() print(f"❌ خطا: {error_details}") return False, f"❌ خطا در پردازش:\n\n{str(e)}\n\nجزئیات:\n{error_details[:500]}", pd.DataFrame() def generate_report(self, df: pd.DataFrame) -> str: """ تولید گزارش جامع Args: df: DataFrame نتایج Returns: متن گزارش """ if df.empty: return "هیچ دادهای برای گزارش یافت نشد" # محاسبه آمار total_rows = len(df) avg_seqeval_p = df['seqeval_precision'].mean() avg_seqeval_r = df['seqeval_recall'].mean() avg_seqeval_f1 = df['seqeval_f1'].mean() avg_seqeval_acc = df['seqeval_accuracy'].mean() high_f1_count = len(df[df['seqeval_f1'] >= 0.9]) mid_f1_count = len(df[df['seqeval_f1'] >= 0.7]) low_f1_count = len(df[df['seqeval_f1'] < 0.5]) best_idx = df['seqeval_f1'].idxmax() worst_idx = df['seqeval_f1'].idxmin() # تفسیر نتایج if avg_seqeval_f1 >= 0.9: interpretation = "✅ عملکرد عالی - سیستم LLM شما بسیار دقیق است" elif avg_seqeval_f1 >= 0.7: interpretation = "⚠️ عملکرد خوب - اما قابل بهبود" else: interpretation = "❌ عملکرد ضعیف - نیاز به بهبود اساسی در مدل LLM" report = f""" ## 📊 گزارش جامع ارزیابی NER ### 🎯 خلاصه نتایج: {interpretation} ### 📈 آمار کلی: - **تعداد کل سطرها:** {total_rows} - **روش ارزیابی:** IOB2 Tagging (استاندارد CoNLL-2003) - **مقایسه:** مرجع انسانی (Reference_text) vs پیشبینی LLM (anonymized_text) ### ✅ نتایج seqeval (استاندارد): - **میانگین Precision:** {avg_seqeval_p:.4f} - **میانگین Recall:** {avg_seqeval_r:.4f} - **میانگین F1-Score:** {avg_seqeval_f1:.4f} - **میانگین Accuracy:** {avg_seqeval_acc:.4f} ### 📊 توزیع عملکرد: - **F1 ≥ 0.9 (عالی):** {high_f1_count} سطر ({high_f1_count/total_rows*100:.1f}%) - **F1 ≥ 0.7 (خوب):** {mid_f1_count} سطر ({mid_f1_count/total_rows*100:.1f}%) - **F1 < 0.5 (ضعیف):** {low_f1_count} سطر ({low_f1_count/total_rows*100:.1f}%) ### 🏆 بهترین و بدترین: - **بهترین F1:** {df.loc[best_idx, 'seqeval_f1']:.4f} (سطر {best_idx + 1}) - **بدترین F1:** {df.loc[worst_idx, 'seqeval_f1']:.4f} (سطر {worst_idx + 1}) ### 💡 توصیهها: {"- مدل LLM شما عملکرد بسیار خوبی دارد" if avg_seqeval_f1 >= 0.9 else ""} {"- روی بهبود Precision تمرکز کنید (کاهش False Positives)" if avg_seqeval_p < avg_seqeval_r else ""} {"- روی بهبود Recall تمرکز کنید (کاهش False Negatives)" if avg_seqeval_r < avg_seqeval_p else ""} {"- نیاز به بازنگری اساسی در prompt یا fine-tuning مدل LLM دارید" if avg_seqeval_f1 < 0.5 else ""} """ return report def create_downloadable_csv(self) -> str: """ ایجاد فایل CSV برای دانلود Returns: مسیر فایل موقت """ if self.results_df is None or self.results_df.empty: return None try: # ایجاد فایل موقت timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") temp_filename = f"evaluation_results_{timestamp}.csv" temp_path = os.path.join(tempfile.gettempdir(), temp_filename) # ذخیره DataFrame self.results_df.to_csv(temp_path, index=False, encoding='utf-8-sig') print(f"✅ فایل CSV ایجاد شد: {temp_path}") return temp_path except Exception as e: print(f"❌ خطا در ایجاد CSV: {str(e)}") return None # ==================== Gradio Interface ==================== def create_interface(): """ایجاد رابط کاربری Gradio""" evaluator = StandardNEREvaluator() # بررسی وضعیت seqeval seqeval_status = "✅ فعال و آماده" if SEQEVAL_AVAILABLE else "❌ نصب نشده" seqeval_emoji = "🟢" if SEQEVAL_AVAILABLE else "🔴" # تعریف CSS سفارشی custom_css = """ .rtl { direction: rtl; text-align: right; font-family: Tahoma, Arial, sans-serif; } .ltr { direction: ltr; text-align: left; } .center { text-align: center; } .header-box { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; padding: 20px; border-radius: 10px; margin-bottom: 20px; } .status-box { background: #f0f9ff; border-left: 4px solid #0284c7; padding: 15px; border-radius: 5px; margin: 10px 0; } """ # ساخت Interface with gr.Blocks( title="NER Anonymization Evaluator", theme=gr.themes.Soft( primary_hue="blue", secondary_hue="purple", ), css=custom_css ) as demo: # هدر gr.Markdown(f"""
Named Entity Recognition Evaluation Tool
original_text - متن خامReference_text - ناشناسسازی انسانی (مرجع)anonymized_text - ناشناسسازی LLM (پیشبینی)