DeepXR
/

Helion-V1.5

@@ -1,405 +0,0 @@
-"""
-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()