# ===================================================================== # EVALUATION MODULE - Untuk Pengerjaan Revisi Karya Tulis # ===================================================================== # Module ini menyediakan tools untuk: # 1. Experiment 1: Topic Classification Accuracy (LDA) # 2. Experiment 2: Speed Improvement (manual vs system) # 3. Experiment 3: Deduplication Accuracy # 4. Experiment 4: Inter-Model Consistency (OpenAI vs Gemini) # 5. Experiment 5: Scalability Testing (concurrent users) # 6. Experiment 6: User Satisfaction Survey # ===================================================================== import pandas as pd import numpy as np import json import time import requests from datetime import datetime from collections import Counter from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix from sklearn.decomposition import LatentDirichletAllocation from sklearn.feature_extraction.text import CountVectorizer import matplotlib.pyplot as plt import seaborn as sns from tabulate import tabulate # ===================================================================== # EXPERIMENT 1: TOPIC CLASSIFICATION ACCURACY (LDA) # ===================================================================== class TopicClassificationEvaluation: """ Evaluate LDA topic classification accuracy Example: eval = TopicClassificationEvaluation() results = eval.run_experiment( df=articles_df, ground_truth=manual_labels, # List of (article_id, true_topic_label) n_topics=5 ) """ def __init__(self): self.lda_model = None self.vectorizer = None def prepare_data(self, df, text_column='full_text'): """Prepare text data for LDA""" texts = df[text_column].fillna('').astype(str) self.vectorizer = CountVectorizer( max_df=0.9, min_df=2, max_features=1000, stop_words='english' ) return self.vectorizer.fit_transform(texts) def train_lda(self, doc_term_matrix, n_topics=5): """Train LDA model""" self.lda_model = LatentDirichletAllocation( n_components=n_topics, random_state=42, max_iter=20 ) self.lda_model.fit(doc_term_matrix) return self.lda_model def get_topic_assignments(self, doc_term_matrix): """Get topic assignment for each document""" return self.lda_model.transform(doc_term_matrix).argmax(axis=1) def run_experiment(self, df, ground_truth_labels, n_topics=5): """ Run full evaluation experiment Args: df: DataFrame with articles ground_truth_labels: List of actual topic labels (same order as df) n_topics: Number of topics for LDA Returns: dict: Evaluation metrics """ print(f"[EXP-1] Training LDA with {n_topics} topics...") # Prepare & train doc_term_matrix = self.prepare_data(df) self.train_lda(doc_term_matrix, n_topics) # Get predictions predicted_topics = self.get_topic_assignments(doc_term_matrix) # Map to ground truth (assuming ground_truth_labels are 0-4) y_true = np.array(ground_truth_labels) y_pred = predicted_topics # Calculate metrics precision = precision_score(y_true, y_pred, average='weighted', zero_division=0) recall = recall_score(y_true, y_pred, average='weighted', zero_division=0) f1 = f1_score(y_true, y_pred, average='weighted', zero_division=0) # Confusion matrix cm = confusion_matrix(y_true, y_pred) # Per-class metrics precision_per_class = precision_score(y_true, y_pred, average=None, zero_division=0) recall_per_class = recall_score(y_true, y_pred, average=None, zero_division=0) results = { 'precision': precision, 'recall': recall, 'f1_score': f1, 'precision_per_class': precision_per_class.tolist(), 'recall_per_class': recall_per_class.tolist(), 'confusion_matrix': cm.tolist(), 'n_topics': n_topics, 'n_samples': len(df), 'timestamp': datetime.now().isoformat() } print(f"[EXP-1] Results: Precision={precision:.3f}, Recall={recall:.3f}, F1={f1:.3f}") return results def print_report(self, results): """Print evaluation report""" print("\n" + "="*70) print("EXPERIMENT 1: TOPIC CLASSIFICATION ACCURACY (LDA)") print("="*70) data = [ ["Precision", f"{results['precision']:.3f}"], ["Recall", f"{results['recall']:.3f}"], ["F1-Score", f"{results['f1_score']:.3f}"], ["Samples", results['n_samples']], ["Topics", results['n_topics']] ] print(tabulate(data, headers=["Metric", "Value"], tablefmt="grid")) print("\nPer-Class Performance:") for i, (p, r) in enumerate(zip(results['precision_per_class'], results['recall_per_class'])): print(f" Topic {i}: Precision={p:.3f}, Recall={r:.3f}") # ===================================================================== # EXPERIMENT 2: SPEED IMPROVEMENT # ===================================================================== class SpeedEvaluation: """ Measure speed improvement: manual review vs system Example: eval = SpeedEvaluation() results = eval.run_experiment( system_search_time=120, # seconds manual_review_time=8*24*3600 # seconds (8 days) ) """ def run_experiment(self, system_search_time, manual_review_time): """ Run speed comparison Args: system_search_time: Time for system (seconds) manual_review_time: Time for manual (seconds) Returns: dict: Speed metrics """ print("[EXP-2] Measuring speed improvement...") speedup_factor = manual_review_time / system_search_time speedup_percentage = ((manual_review_time - system_search_time) / manual_review_time) * 100 results = { 'system_time_seconds': system_search_time, 'manual_time_seconds': manual_review_time, 'speedup_factor': speedup_factor, 'speedup_percentage': speedup_percentage, 'time_saved_seconds': manual_review_time - system_search_time, 'system_time_formatted': self._format_time(system_search_time), 'manual_time_formatted': self._format_time(manual_review_time), 'timestamp': datetime.now().isoformat() } print(f"[EXP-2] Manual: {self._format_time(manual_review_time)}, " f"System: {self._format_time(system_search_time)}, " f"Speedup: {speedup_percentage:.1f}%") return results @staticmethod def _format_time(seconds): """Format seconds to human readable""" if seconds < 60: return f"{seconds:.1f}s" elif seconds < 3600: return f"{seconds/60:.1f} min" elif seconds < 86400: return f"{seconds/3600:.1f} hours" else: return f"{seconds/86400:.1f} days" def print_report(self, results): """Print evaluation report""" print("\n" + "="*70) print("EXPERIMENT 2: SPEED IMPROVEMENT") print("="*70) data = [ ["Manual Review Time", results['manual_time_formatted']], ["System Processing Time", results['system_time_formatted']], ["Speedup Factor", f"{results['speedup_factor']:.1f}x"], ["Time Saved", f"{results['speedup_percentage']:.1f}%"], ] print(tabulate(data, headers=["Metric", "Value"], tablefmt="grid")) # ===================================================================== # EXPERIMENT 3: DEDUPLICATION ACCURACY # ===================================================================== class DeduplicationEvaluation: """ Evaluate deduplication accuracy (precision & recall) Example: eval = DeduplicationEvaluation() results = eval.run_experiment( df=articles_with_duplicates, ground_truth_duplicates=[(0,1), (5,7)] # (idx1, idx2) pairs ) """ def run_experiment(self, df, ground_truth_duplicates=None, manual_sample_size=100): """ Run deduplication accuracy evaluation Args: df: DataFrame with potentially duplicate articles ground_truth_duplicates: List of (idx1, idx2) pairs of duplicates manual_sample_size: If no ground truth, simulate manual review Returns: dict: Deduplication metrics """ print("[EXP-3] Evaluating deduplication accuracy...") # Perform deduplication df_dedup = self._deduplicate_by_title(df) duplicates_removed = len(df) - len(df_dedup) if ground_truth_duplicates is None: # Simulate ground truth from manual review of sample sample_indices = np.random.choice(len(df), min(manual_sample_size, len(df)), replace=False) ground_truth_duplicates = self._simulate_ground_truth(df, sample_indices) # Calculate metrics predictions = self._get_duplicate_predictions(df, df_dedup) tp = len([p for p in predictions if p in ground_truth_duplicates]) # True positives fp = len([p for p in predictions if p not in ground_truth_duplicates]) # False positives fn = len([g for g in ground_truth_duplicates if g not in predictions]) # False negatives precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 results = { 'original_count': len(df), 'deduped_count': len(df_dedup), 'duplicates_removed': duplicates_removed, 'duplicate_percentage': (duplicates_removed / len(df)) * 100, 'true_positives': tp, 'false_positives': fp, 'false_negatives': fn, 'precision': precision, 'recall': recall, 'f1_score': f1, 'timestamp': datetime.now().isoformat() } print(f"[EXP-3] Removed {duplicates_removed} duplicates, " f"Precision={precision:.3f}, Recall={recall:.3f}, F1={f1:.3f}") return results @staticmethod def _deduplicate_by_title(df): """Simple deduplication by title""" df_temp = df.copy() df_temp['title_lower'] = df_temp.get('title', pd.Series(dtype=str)).str.lower().str.strip() df_temp = df_temp.drop_duplicates(subset=['title_lower'], keep='first') return df_temp.drop(columns=['title_lower']) @staticmethod def _simulate_ground_truth(df, sample_indices): """Simulate ground truth duplicates""" # In real scenario, would have manual annotation # For now, simple heuristic: similar titles are likely duplicates duplicates = [] for i in range(len(sample_indices)-1): for j in range(i+1, min(i+10, len(sample_indices))): idx_i = sample_indices[i] idx_j = sample_indices[j] title_i = str(df.iloc[idx_i]['title']).lower() title_j = str(df.iloc[idx_j]['title']).lower() # Simple similarity check if len(title_i) > 10 and title_i in title_j: duplicates.append((idx_i, idx_j)) return duplicates @staticmethod def _get_duplicate_predictions(df_original, df_deduped): """Get which rows were marked as duplicates""" # Find rows removed during deduplication removed_indices = set(df_original.index) - set(df_deduped.index) # Create pairs (original index with kept index) predictions = [] for removed_idx in removed_indices: removed_title = str(df_original.iloc[removed_idx]['title']).lower() for kept_idx in df_deduped.index: kept_title = str(df_deduped.iloc[kept_idx]['title']).lower() if removed_title == kept_title: predictions.append((removed_idx, kept_idx)) return predictions def print_report(self, results): """Print evaluation report""" print("\n" + "="*70) print("EXPERIMENT 3: DEDUPLICATION ACCURACY") print("="*70) data = [ ["Original Articles", results['original_count']], ["After Deduplication", results['deduped_count']], ["Duplicates Removed", results['duplicates_removed']], ["Duplicate %", f"{results['duplicate_percentage']:.1f}%"], ["Precision", f"{results['precision']:.3f}"], ["Recall", f"{results['recall']:.3f}"], ["F1-Score", f"{results['f1_score']:.3f}"] ] print(tabulate(data, headers=["Metric", "Value"], tablefmt="grid")) # ===================================================================== # EXPERIMENT 4: INTER-MODEL CONSISTENCY # ===================================================================== class InterModelConsistencyEvaluation: """ Compare outputs from OpenAI and Gemini models Example: eval = InterModelConsistencyEvaluation() results = eval.run_experiment( prompts=list_of_prompts, openai_client=openai_client, gemini_api_key=gemini_key ) """ def run_experiment(self, prompts, openai_client=None, gemini_api_key=None): """ Run inter-model consistency test Args: prompts: List of test prompts openai_client: OpenAI client object gemini_api_key: Gemini API key Returns: dict: Consistency metrics """ print(f"[EXP-4] Comparing {len(prompts)} prompts across models...") responses_openai = [] responses_gemini = [] similarities = [] for i, prompt in enumerate(prompts): print(f" Prompt {i+1}/{len(prompts)}...", end=' ') # Get OpenAI response if openai_client: try: resp_openai = self._get_openai_response(openai_client, prompt) responses_openai.append(resp_openai) except Exception as e: print(f"[OpenAI Error: {str(e)[:30]}]", end=' ') responses_openai.append(None) # Get Gemini response if gemini_api_key: try: resp_gemini = self._get_gemini_response(gemini_api_key, prompt) responses_gemini.append(resp_gemini) except Exception as e: print(f"[Gemini Error: {str(e)[:30]}]", end=' ') responses_gemini.append(None) # Calculate similarity if responses_openai[-1] and responses_gemini[-1]: sim = self._calculate_similarity(responses_openai[-1], responses_gemini[-1]) similarities.append(sim) print(f"Similarity={sim:.2f}") else: print("[Skipped]") time.sleep(1) # Rate limiting avg_similarity = np.mean(similarities) if similarities else 0 results = { 'n_prompts': len(prompts), 'similarities': similarities, 'avg_similarity': avg_similarity, 'openai_responses': responses_openai, 'gemini_responses': responses_gemini, 'timestamp': datetime.now().isoformat() } print(f"[EXP-4] Average Similarity: {avg_similarity:.3f}") return results @staticmethod def _get_openai_response(client, prompt, max_tokens=200): """Get response from OpenAI""" response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=max_tokens ) return response.choices[0].message.content.strip() @staticmethod def _get_gemini_response(api_key, prompt, max_tokens=200): """Get response from Gemini""" import google.generativeai as genai genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-1.5-flash') response = model.generate_content(prompt) return response.text.strip() @staticmethod def _calculate_similarity(text1, text2): """Calculate text similarity (simple word overlap)""" words1 = set(text1.lower().split()) words2 = set(text2.lower().split()) if len(words1) == 0 or len(words2) == 0: return 0 intersection = len(words1 & words2) union = len(words1 | words2) return intersection / union if union > 0 else 0 def print_report(self, results): """Print evaluation report""" print("\n" + "="*70) print("EXPERIMENT 4: INTER-MODEL CONSISTENCY") print("="*70) data = [ ["Prompts Tested", results['n_prompts']], ["Average Similarity", f"{results['avg_similarity']:.3f}"], ["Min Similarity", f"{min(results['similarities']):.3f}"], ["Max Similarity", f"{max(results['similarities']):.3f}"], ["Std Dev", f"{np.std(results['similarities']):.3f}"] ] print(tabulate(data, headers=["Metric", "Value"], tablefmt="grid")) # ===================================================================== # EXPERIMENT 5: SCALABILITY (LOAD TESTING) # ===================================================================== class ScalabilityEvaluation: """ Simulate concurrent users and measure response times Example: eval = ScalabilityEvaluation() results = eval.simulate_concurrent_users( n_users=[10, 25, 50, 100], request_func=lambda: search_system(), timeout=30 ) """ def simulate_concurrent_users(self, n_users_list, request_func, timeout=30): """ Simulate concurrent user load Args: n_users_list: List of concurrent user counts to test request_func: Function that simulates one user request timeout: Timeout per request (seconds) Returns: dict: Performance metrics """ print(f"[EXP-5] Simulating concurrent users: {n_users_list}") results_by_load = {} for n_users in n_users_list: print(f" Testing with {n_users} concurrent users...") response_times = [] errors = 0 for _ in range(n_users): start = time.time() try: request_func() response_time = time.time() - start if response_time <= timeout: response_times.append(response_time) else: errors += 1 except Exception as e: errors += 1 # Calculate statistics success_rate = (n_users - errors) / n_users * 100 avg_response = np.mean(response_times) if response_times else 0 p95_response = np.percentile(response_times, 95) if response_times else 0 p99_response = np.percentile(response_times, 99) if response_times else 0 results_by_load[n_users] = { 'n_users': n_users, 'success_rate': success_rate, 'avg_response_time': avg_response, 'p95_response_time': p95_response, 'p99_response_time': p99_response, 'errors': errors, 'response_times': response_times } print(f" Avg Response: {avg_response:.2f}s, " f"Success Rate: {success_rate:.1f}%") return results_by_load def print_report(self, results_by_load): """Print scalability report""" print("\n" + "="*70) print("EXPERIMENT 5: SCALABILITY & LOAD TESTING") print("="*70) data = [] for n_users in sorted(results_by_load.keys()): r = results_by_load[n_users] data.append([ r['n_users'], f"{r['avg_response_time']:.2f}s", f"{r['p95_response_time']:.2f}s", f"{r['p99_response_time']:.2f}s", f"{r['success_rate']:.1f}%" ]) print(tabulate(data, headers=["Users", "Avg", "P95", "P99", "Success%"], tablefmt="grid")) # ===================================================================== # EXPERIMENT 6: USER SATISFACTION SURVEY # ===================================================================== class UserSatisfactionEvaluation: """ Collect and analyze user satisfaction survey results Example: eval = UserSatisfactionEvaluation() results = eval.run_experiment( survey_responses=[ {'q1': 5, 'q2': 4, 'q3': 5, ...}, ... ] ) """ def run_experiment(self, survey_responses): """ Run user satisfaction evaluation Args: survey_responses: List of dicts with question responses (1-5 scale) Returns: dict: Satisfaction metrics """ print(f"[EXP-6] Analyzing {len(survey_responses)} survey responses...") if not survey_responses: print(" No responses provided") return {} # Convert to DataFrame df_responses = pd.DataFrame(survey_responses) # Calculate metrics overall_satisfaction = df_responses.mean().mean() # NPS calculation (promoters - detractors) # Assuming last question is "Would you recommend?" on 0-10 scale nps_score = None if 'recommend_score' in df_responses.columns: recommend = df_responses['recommend_score'] promoters = (recommend >= 9).sum() / len(recommend) * 100 detractors = (recommend <= 6).sum() / len(recommend) * 100 nps_score = promoters - detractors # Per-question analysis question_means = df_responses.mean() results = { 'n_respondents': len(survey_responses), 'overall_satisfaction': overall_satisfaction, 'nps_score': nps_score, 'question_means': question_means.to_dict(), 'timestamp': datetime.now().isoformat() } print(f"[EXP-6] Overall Satisfaction: {overall_satisfaction:.2f}/5") if nps_score: print(f"[EXP-6] NPS Score: {nps_score:.1f}") return results def print_report(self, results): """Print user satisfaction report""" print("\n" + "="*70) print("EXPERIMENT 6: USER SATISFACTION SURVEY") print("="*70) data = [ ["Respondents", results['n_respondents']], ["Overall Satisfaction", f"{results['overall_satisfaction']:.2f}/5.00"], ] if results['nps_score'] is not None: data.append(["NPS Score", f"{results['nps_score']:.1f}"]) print(tabulate(data, headers=["Metric", "Value"], tablefmt="grid")) print("\nPer-Question Scores:") question_data = [[q, f"{v:.2f}/5.00"] for q, v in results['question_means'].items()] print(tabulate(question_data, headers=["Question", "Score"], tablefmt="grid")) # ===================================================================== # METRICS EXPORT & REPORTING # ===================================================================== class EvaluationReporter: """Export evaluation results for documentation""" @staticmethod def generate_summary_table(all_experiments): """Generate summary table of all experiments""" summary_data = [] if 'topic_classification' in all_experiments: exp = all_experiments['topic_classification'] summary_data.append([ "Topic Classification", f"{exp['precision']:.1%}", f"{exp['recall']:.1%}", f"{exp['f1_score']:.3f}", f"{exp['n_samples']}" ]) if 'speed_improvement' in all_experiments: exp = all_experiments['speed_improvement'] summary_data.append([ "Speed Improvement", f"{exp['speedup_percentage']:.1f}%", f"{exp['speedup_factor']:.1f}x", exp['system_time_formatted'], exp['manual_time_formatted'] ]) if 'deduplication' in all_experiments: exp = all_experiments['deduplication'] summary_data.append([ "Deduplication", f"{exp['precision']:.1%}", f"{exp['recall']:.1%}", f"{exp['f1_score']:.3f}", f"{exp['duplicates_removed']}" ]) if 'inter_model_consistency' in all_experiments: exp = all_experiments['inter_model_consistency'] summary_data.append([ "Model Consistency", f"{exp['avg_similarity']:.3f}", "-", "-", f"{exp['n_prompts']}" ]) if 'user_satisfaction' in all_experiments: exp = all_experiments['user_satisfaction'] summary_data.append([ "User Satisfaction", f"{exp['overall_satisfaction']:.2f}/5.00", f"NPS: {exp['nps_score']:.1f}" if exp['nps_score'] else "-", "-", f"{exp['n_respondents']} users" ]) headers = ["Experiment", "Primary Metric", "Secondary", "Tertiary", "Sample"] return tabulate(summary_data, headers=headers, tablefmt="grid") @staticmethod def export_to_json(all_experiments, filename='evaluation_results.json'): """Export all results to JSON""" with open(filename, 'w') as f: json.dump(all_experiments, f, indent=2, default=str) print(f"Exported results to {filename}") @staticmethod def export_to_markdown(all_experiments, filename='evaluation_report.md'): """Export results to Markdown""" with open(filename, 'w') as f: f.write("# Evaluation Results\n\n") f.write(f"**Generated:** {datetime.now().isoformat()}\n\n") # Summary table f.write("## Summary\n\n") f.write("| Experiment | Result |\n") f.write("|---|---|\n") for exp_name, exp_data in all_experiments.items(): if 'precision' in exp_data: f.write(f"| {exp_name} | " f"Precision: {exp_data['precision']:.1%}, " f"Recall: {exp_data['recall']:.1%} |\n") f.write("\n") # Detailed results for exp_name, exp_data in all_experiments.items(): f.write(f"## {exp_name}\n\n") f.write(f"```json\n{json.dumps(exp_data, indent=2, default=str)}\n```\n\n") print(f"Exported report to {filename}") # ===================================================================== # QUICK START EXAMPLE # ===================================================================== def run_all_experiments_example(): """ Example of running all experiments Usage: python evaluation_module.py """ print("="*70) print("EVALUATION MODULE - QUICK START EXAMPLE") print("="*70) # Experiment 1: Topic Classification print("\n[1] Topic Classification Accuracy") eval1 = TopicClassificationEvaluation() # Note: Requires actual data - this is just structure # results1 = eval1.run_experiment(df, ground_truth_labels, n_topics=5) # eval1.print_report(results1) print(" [Example only - needs actual data]") # Experiment 2: Speed Improvement print("\n[2] Speed Improvement") eval2 = SpeedEvaluation() results2 = eval2.run_experiment( system_search_time=120, # 2 minutes manual_review_time=8*24*3600 # 8 days ) eval2.print_report(results2) # Experiment 3: Deduplication print("\n[3] Deduplication Accuracy") eval3 = DeduplicationEvaluation() # results3 = eval3.run_experiment(df) # eval3.print_report(results3) print(" [Example only - needs actual data]") # Experiment 4: Inter-Model Consistency print("\n[4] Inter-Model Consistency") eval4 = InterModelConsistencyEvaluation() print(" [Requires OpenAI & Gemini API keys]") # Experiment 5: Scalability print("\n[5] Scalability Testing") eval5 = ScalabilityEvaluation() # Example mock request mock_results = eval5.simulate_concurrent_users( n_users_list=[10, 25, 50], request_func=lambda: time.sleep(np.random.uniform(1, 3)), # Mock processing timeout=30 ) eval5.print_report(mock_results) # Experiment 6: User Satisfaction print("\n[6] User Satisfaction Survey") eval6 = UserSatisfactionEvaluation() # Mock survey responses mock_survey = [ {'q1': 5, 'q2': 4, 'q3': 5, 'q4': 4, 'q5': 5, 'recommend_score': 9}, {'q1': 4, 'q2': 5, 'q3': 4, 'q4': 5, 'q5': 4, 'recommend_score': 8}, ] results6 = eval6.run_experiment(mock_survey) eval6.print_report(results6) print("\n" + "="*70) print("Evaluation module ready to use!") print("="*70) if __name__ == "__main__": run_all_experiments_example()