Create evaluate_dataset.py

evaluate_dataset.py

@@ -0,0 +1,405 @@
+"""
+Helion 1.5 Dataset Evaluation Tool
+===================================
+Comprehensive evaluation and quality assessment for Helion 1.5 datasets.
+"""
+
+import json
+import numpy as np
+from typing import Dict, List, Any, Tuple
+from collections import Counter, defaultdict
+import matplotlib.pyplot as plt
+from pathlib import Path
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class HelionDatasetEvaluator:
+    """Evaluate and analyze Helion 1.5 dataset quality"""
+    
+    def __init__(self):
+        self.stats = defaultdict(list)
+        self.issues = []
+        
+    def evaluate_conversations(self, filepath: str) -> Dict[str, Any]:
+        """Evaluate conversation dataset"""
+        logger.info(f"Evaluating conversations from {filepath}")
+        
+        metrics = {
+            'total_examples': 0,
+            'total_turns': [],
+            'word_counts': [],
+            'quality_scores': [],
+            'domain_distribution': Counter(),
+            'difficulty_distribution': Counter(),
+            'language_distribution': Counter(),
+            'has_code_count': 0,
+            'avg_turns_per_conversation': 0,
+            'avg_words_per_turn': 0,
+            'quality_issues': []
+        }
+        
+        with open(filepath, 'r', encoding='utf-8') as f:
+            for line_num, line in enumerate(f, 1):
+                try:
+                    data = json.loads(line.strip())
+                    metrics['total_examples'] += 1
+                    
+                    # Analyze conversations
+                    if 'conversations' in data:
+                        turns = data['conversations']
+                        metrics['total_turns'].append(len(turns))
+                        
+                        for turn in turns:
+                            words = len(turn['content'].split())
+                            metrics['word_counts'].append(words)
+                    
+                    # Analyze metadata
+                    if 'metadata' in data:
+                        meta = data['metadata']
+                        
+                        if 'quality_score' in meta:
+                            metrics['quality_scores'].append(meta['quality_score'])
+                        
+                        if 'domain' in meta:
+                            metrics['domain_distribution'][meta['domain']] += 1
+                        
+                        if 'difficulty' in meta:
+                            metrics['difficulty_distribution'][meta['difficulty']] += 1
+                        
+                        if 'languages' in meta:
+                            for lang in meta['languages']:
+                                metrics['language_distribution'][lang] += 1
+                        
+                        if meta.get('has_code', False):
+                            metrics['has_code_count'] += 1
+                    
+                    # Check for quality issues
+                    issues = self._check_conversation_quality(data)
+                    if issues:
+                        metrics['quality_issues'].extend([
+                            {'line': line_num, 'issues': issues}
+                        ])
+                
+                except json.JSONDecodeError:
+                    logger.error(f"JSON decode error at line {line_num}")
+                except Exception as e:
+                    logger.error(f"Error at line {line_num}: {str(e)}")
+        
+        # Calculate averages
+        if metrics['total_turns']:
+            metrics['avg_turns_per_conversation'] = np.mean(metrics['total_turns'])
+        if metrics['word_counts']:
+            metrics['avg_words_per_turn'] = np.mean(metrics['word_counts'])
+        if metrics['quality_scores']:
+            metrics['avg_quality_score'] = np.mean(metrics['quality_scores'])
+            metrics['min_quality_score'] = np.min(metrics['quality_scores'])
+            metrics['max_quality_score'] = np.max(metrics['quality_scores'])
+        
+        return metrics
+    
+    def evaluate_instructions(self, filepath: str) -> Dict[str, Any]:
+        """Evaluate instruction dataset"""
+        logger.info(f"Evaluating instructions from {filepath}")
+        
+        metrics = {
+            'total_examples': 0,
+            'instruction_lengths': [],
+            'output_lengths': [],
+            'quality_scores': [],
+            'task_type_distribution': Counter(),
+            'complexity_distribution': Counter(),
+            'verified_count': 0,
+            'quality_issues': []
+        }
+        
+        with open(filepath, 'r', encoding='utf-8') as f:
+            for line_num, line in enumerate(f, 1):
+                try:
+                    data = json.loads(line.strip())
+                    metrics['total_examples'] += 1
+                    
+                    if 'instruction' in data:
+                        metrics['instruction_lengths'].append(len(data['instruction'].split()))
+                    
+                    if 'output' in data:
+                        metrics['output_lengths'].append(len(data['output'].split()))
+                    
+                    if 'metadata' in data:
+                        meta = data['metadata']
+                        
+                        if 'quality_score' in meta:
+                            metrics['quality_scores'].append(meta['quality_score'])
+                        
+                        if 'task_type' in meta:
+                            metrics['task_type_distribution'][meta['task_type']] += 1
+                        
+                        if 'complexity' in meta:
+                            metrics['complexity_distribution'][meta['complexity']] += 1
+                        
+                        if meta.get('verified', False):
+                            metrics['verified_count'] += 1
+                    
+                    issues = self._check_instruction_quality(data)
+                    if issues:
+                        metrics['quality_issues'].append({
+                            'line': line_num, 'issues': issues
+                        })
+                
+                except Exception as e:
+                    logger.error(f"Error at line {line_num}: {str(e)}")
+        
+        # Calculate statistics
+        if metrics['instruction_lengths']:
+            metrics['avg_instruction_length'] = np.mean(metrics['instruction_lengths'])
+        if metrics['output_lengths']:
+            metrics['avg_output_length'] = np.mean(metrics['output_lengths'])
+        if metrics['quality_scores']:
+            metrics['avg_quality_score'] = np.mean(metrics['quality_scores'])
+        
+        metrics['verification_rate'] = metrics['verified_count'] / max(metrics['total_examples'], 1)
+        
+        return metrics
+    
+    def evaluate_code(self, filepath: str) -> Dict[str, Any]:
+        """Evaluate code dataset"""
+        logger.info(f"Evaluating code from {filepath}")
+        
+        metrics = {
+            'total_examples': 0,
+            'language_distribution': Counter(),
+            'difficulty_distribution': Counter(),
+            'lines_of_code': [],
+            'quality_scores': [],
+            'has_test_cases': 0,
+            'has_explanation': 0,
+            'tag_distribution': Counter(),
+            'quality_issues': []
+        }
+        
+        with open(filepath, 'r', encoding='utf-8') as f:
+            for line_num, line in enumerate(f, 1):
+                try:
+                    data = json.loads(line.strip())
+                    metrics['total_examples'] += 1
+                    
+                    if 'language' in data:
+                        metrics['language_distribution'][data['language']] += 1
+                    
+                    if 'solution' in data:
+                        loc = len(data['solution'].strip().split('\n'))
+                        metrics['lines_of_code'].append(loc)
+                    
+                    if 'test_cases' in data and data['test_cases']:
+                        metrics['has_test_cases'] += 1
+                    
+                    if 'explanation' in data and data['explanation']:
+                        metrics['has_explanation'] += 1
+                    
+                    if 'metadata' in data:
+                        meta = data['metadata']
+                        
+                        if 'quality_score' in meta:
+                            metrics['quality_scores'].append(meta['quality_score'])
+                        
+                        if 'difficulty' in meta:
+                            metrics['difficulty_distribution'][meta['difficulty']] += 1
+                        
+                        if 'tags' in meta:
+                            for tag in meta['tags']:
+                                metrics['tag_distribution'][tag] += 1
+                    
+                    issues = self._check_code_quality(data)
+                    if issues:
+                        metrics['quality_issues'].append({
+                            'line': line_num, 'issues': issues
+                        })
+                
+                except Exception as e:
+                    logger.error(f"Error at line {line_num}: {str(e)}")
+        
+        # Calculate statistics
+        if metrics['lines_of_code']:
+            metrics['avg_lines_of_code'] = np.mean(metrics['lines_of_code'])
+            metrics['median_lines_of_code'] = np.median(metrics['lines_of_code'])
+        
+        if metrics['quality_scores']:
+            metrics['avg_quality_score'] = np.mean(metrics['quality_scores'])
+        
+        metrics['test_case_coverage'] = metrics['has_test_cases'] / max(metrics['total_examples'], 1)
+        metrics['explanation_coverage'] = metrics['has_explanation'] / max(metrics['total_examples'], 1)
+        
+        return metrics
+    
+    def _check_conversation_quality(self, data: Dict) -> List[str]:
+        """Check conversation for quality issues"""
+        issues = []
+        
+        if 'conversations' not in data:
+            issues.append('Missing conversations field')
+            return issues
+        
+        conversations = data['conversations']
+        
+        # Check for minimum turns
+        if len(conversations) < 2:
+            issues.append('Too few conversation turns')
+        
+        # Check turn structure
+        for i, turn in enumerate(conversations):
+            if 'role' not in turn:
+                issues.append(f'Turn {i}: Missing role')
+            if 'content' not in turn:
+                issues.append(f'Turn {i}: Missing content')
+            elif len(turn['content'].strip()) < 10:
+                issues.append(f'Turn {i}: Content too short')
+        
+        # Check metadata
+        if 'metadata' in data:
+            meta = data['metadata']
+            if 'quality_score' in meta and meta['quality_score'] < 0.75:
+                issues.append(f"Low quality score: {meta['quality_score']}")
+        else:
+            issues.append('Missing metadata')
+        
+        return issues
+    
+    def _check_instruction_quality(self, data: Dict) -> List[str]:
+        """Check instruction for quality issues"""
+        issues = []
+        
+        if 'instruction' not in data:
+            issues.append('Missing instruction')
+        elif len(data['instruction'].strip()) < 10:
+            issues.append('Instruction too short')
+        
+        if 'output' not in data:
+            issues.append('Missing output')
+        elif len(data['output'].strip()) < 10:
+            issues.append('Output too short')
+        
+        if 'metadata' in data:
+            meta = data['metadata']
+            if 'quality_score' in meta and meta['quality_score'] < 0.75:
+                issues.append(f"Low quality score: {meta['quality_score']}")
+        
+        return issues
+    
+    def _check_code_quality(self, data: Dict) -> List[str]:
+        """Check code for quality issues"""
+        issues = []
+        
+        if 'language' not in data:
+            issues.append('Missing language field')
+        
+        if 'solution' not in data:
+            issues.append('Missing solution')
+        elif len(data['solution'].strip()) < 20:
+            issues.append('Solution too short')
+        
+        if 'test_cases' not in data or not data['test_cases']:
+            issues.append('Missing test cases')
+        
+        if 'explanation' not in data or not data['explanation']:
+            issues.append('Missing explanation')
+        
+        return issues
+    
+    def generate_report(self, output_dir: str = '.'):
+        """Generate comprehensive evaluation report"""
+        report = {
+            'evaluation_date': str(np.datetime64('now')),
+            'summary': {},
+            'recommendations': []
+        }
+        
+        # Add recommendations based on findings
+        report['recommendations'] = [
+            'Maintain quality score above 0.85 for all datasets',
+            'Ensure balanced domain distribution',
+            'Include test cases for all code examples',
+            'Add explanations to improve educational value',
+            'Verify multilingual translations for accuracy',
+            'Regular quality audits every quarter'
+        ]
+        
+        # Save report
+        output_path = Path(output_dir) / 'evaluation_report.json'
+        with open(output_path, 'w') as f:
+            json.dump(report, f, indent=2)
+        
+        logger.info(f"Evaluation report saved to {output_path}")
+        return report
+    
+    def visualize_metrics(self, metrics: Dict, output_dir: str = '.'):
+        """Create visualization charts"""
+        output_dir = Path(output_dir)
+        output_dir.mkdir(exist_ok=True)
+        
+        # Quality score distribution
+        if 'quality_scores' in metrics and metrics['quality_scores']:
+            plt.figure(figsize=(10, 6))
+            plt.hist(metrics['quality_scores'], bins=20, edgecolor='black')
+            plt.xlabel('Quality Score')
+            plt.ylabel('Frequency')
+            plt.title('Quality Score Distribution')
+            plt.savefig(output_dir / 'quality_scores.png')
+            plt.close()
+        
+        # Domain distribution
+        if 'domain_distribution' in metrics:
+            plt.figure(figsize=(12, 6))
+            domains = list(metrics['domain_distribution'].keys())
+            counts = list(metrics['domain_distribution'].values())
+            plt.bar(domains, counts)
+            plt.xlabel('Domain')
+            plt.ylabel('Count')
+            plt.title('Domain Distribution')
+            plt.xticks(rotation=45, ha='right')
+            plt.tight_layout()
+            plt.savefig(output_dir / 'domain_distribution.png')
+            plt.close()
+        
+        logger.info(f"Visualizations saved to {output_dir}")
+    
+    def compare_with_helion1(self, helion15_metrics: Dict, helion1_metrics: Dict) -> Dict:
+        """Compare Helion 1.5 with Helion 1"""
+        comparison = {
+            'improvements': [],
+            'metrics_comparison': {}
+        }
+        
+        # Compare key metrics
+        if 'avg_quality_score' in helion15_metrics and 'avg_quality_score' in helion1_metrics:
+            improvement = ((helion15_metrics['avg_quality_score'] - helion1_metrics['avg_quality_score']) 
+                          / helion1_metrics['avg_quality_score'] * 100)
+            comparison['improvements'].append(
+                f"Quality score improved by {improvement:.2f}%"
+            )
+        
+        return comparison
+
+
+def main():
+    """Main evaluation pipeline"""
+    evaluator = HelionDatasetEvaluator()
+    
+    # Evaluate different datasets
+    # conv_metrics = evaluator.evaluate_conversations('helion-1.5-conversations.jsonl')
+    # inst_metrics = evaluator.evaluate_instructions('helion-1.5-instructions.jsonl')
+    # code_metrics = evaluator.evaluate_code('helion-1.5-code.jsonl')
+    
+    # Generate visualizations
+    # evaluator.visualize_metrics(conv_metrics, 'evaluation_results')
+    
+    # Generate report
+    report = evaluator.generate_report('evaluation_results')
+    
+    logger.info("Evaluation complete!")
+    logger.info("Check 'evaluation_results' directory for detailed reports")
+
+
+if __name__ == '__main__':
+    main()