Method run_perplexity_test

utils/test_perplexity.py:116–205

Run perplexity test on a single dataset.

(self, dataset_name, dataset_path, threads=16, ctx_size=512, model_override=None)

        self.temp_files = []
    
    def run_perplexity_test(self, dataset_name, dataset_path, threads=16, ctx_size=512, model_override=None):
        """Run perplexity test on a single dataset."""
        test_model = model_override if model_override else self.model_path
        
        print(f"\n{'='*80}")
        print(f"📊 Testing on dataset: {dataset_name}")
        print(f"   File: {dataset_path}")
        print(f"   Model: {test_model.name}")
        print(f"{'='*80}")
        
        cmd = [
            str(self.llama_perplexity_bin),
            "-m", str(test_model),
            "-f", str(dataset_path),
            "-t", str(threads),
            "-c", str(ctx_size),
            "-ngl", "0"  # CPU only
        ]
        
        print(f"💻 Command: {' '.join(cmd)}")
        print(f"⏱️  Starting test...\n")
        
        start_time = time.time()
        
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=3600,  # 1 hour timeout
                cwd=os.getcwd()
            )
            
            elapsed_time = time.time() - start_time
            
            if result.returncode == 0:
                # Parse perplexity from output (check both stdout and stderr)
                combined_output = result.stdout + "\n" + result.stderr
                ppl = self.parse_perplexity(combined_output)
                
                if ppl is not None:
                    print(f"\n✅ Perplexity: {ppl}")
                    print(f"⏱️  Time: {elapsed_time:.2f}s ({elapsed_time/60:.2f} min)")
                    status = "success"
                else:
                    print(f"\n⚠️  Test completed but could not parse perplexity")
                    print(f"Last 500 chars of stdout:")
                    print(result.stdout[-500:])
                    print(f"Last 500 chars of stderr:")
                    print(result.stderr[-500:])
                    status = "parse_error"
                    ppl = None
            else:
                print(f"\n❌ Test failed with return code {result.returncode}")
                print(f"Error: {result.stderr[:500]}")
                status = "failed"
                ppl = None
                elapsed_time = time.time() - start_time