Upload from GitHub Actions: Merge pull request #10 from datenlabor-bmz/jn-dev

.github/workflows/nightly-evals.yml

@@ -8,6 +8,7 @@ on:
 jobs:
   run-evals:
     runs-on: ubuntu-latest
+    timeout-minutes: 1440  # 24 hours timeout
     steps:
       - uses: actions/checkout@v3
       
@@ -21,7 +22,7 @@ jobs:
           curl -LsSf https://astral.sh/uv/install.sh | sh
           uv sync --frozen --extra dev
           
-      - name: Run evaluations
+      - name: Run evaluations with checkpointing
         env:
           OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }}
           HUGGINGFACE_ACCESS_TOKEN: ${{ secrets.HUGGINGFACE_ACCESS_TOKEN }}
@@ -30,7 +31,28 @@ jobs:
         run: |
           uv run huggingface-cli login --token ${{ secrets.HUGGINGFACE_ACCESS_TOKEN }}
           uv run evals/download_data.py
-          uv run evals/main.py
+          
+          # Run evaluations with periodic checkpointing
+          uv run python -c "
+          import time
+          import subprocess
+          import json
+          import os
+          
+          # Check if we have existing results to resume from
+          if os.path.exists('results.json'):
+              print('Found existing results.json, will resume from checkpoint')
+          
+          # Run the main evaluation
+          try:
+              subprocess.run(['uv', 'run', 'evals/main.py'], check=True)
+          except subprocess.CalledProcessError as e:
+              print(f'Evaluation failed: {e}')
+              # Save current state even if failed
+              if os.path.exists('results.json'):
+                  print('Saving checkpoint before exit...')
+              exit(1)
+          "
           
       - name: Commit changes
         env:

evals/main.py

@@ -11,6 +11,45 @@ import json
 
 results = pd.DataFrame()
 
+def save_checkpoint(results_df, models_df, languages_df, batch_num, total_batches):
+    """Save current progress as checkpoint"""
+    try:
+        args = dict(orient="records", indent=2, force_ascii=False)
+        
+        # Save current results
+        if len(results_df) > 0:
+            results_df.to_json("results.json", **args)
+            print(f"💾 Checkpoint saved: {len(results_df)} results (batch {batch_num}/{total_batches})")
+        
+        # Save model and language info
+        models_df.to_json("models.json", **args)
+        languages_df.to_json("languages.json", **args)
+        
+        # Save checkpoint metadata
+        checkpoint_info = {
+            "last_batch": batch_num,
+            "total_batches": total_batches,
+            "timestamp": datetime.now().isoformat(),
+            "results_count": len(results_df)
+        }
+        with open("checkpoint.json", "w") as f:
+            json.dump(checkpoint_info, f, indent=2)
+            
+    except Exception as e:
+        print(f"⚠️  Failed to save checkpoint: {e}")
+
+def load_checkpoint():
+    """Load previous checkpoint if available"""
+    try:
+        if os.path.exists("checkpoint.json"):
+            with open("checkpoint.json", "r") as f:
+                checkpoint = json.load(f)
+            print(f"📂 Found checkpoint from batch {checkpoint['last_batch']}/{checkpoint['total_batches']}")
+            return checkpoint
+    except Exception as e:
+        print(f"⚠️  Failed to load checkpoint: {e}")
+    return None
+
 async def evaluate():
     # FIXME we should not need this for-loop, but it helps
     n_sentences = int(os.environ.get("N_SENTENCES", 15)) # Default 1 for quick testing
@@ -29,13 +68,30 @@ async def evaluate():
     top_languages = languages.head(max_languages)  # Top N by population
     print(f"🌍 Evaluating top {len(top_languages)} languages by speakers (max: {max_languages})")
     
+    # Load checkpoint if available
+    checkpoint = load_checkpoint()
+    start_batch = 0
+    if checkpoint:
+        start_batch = checkpoint['last_batch']
+        print(f"🔄 Resuming from batch {start_batch}")
+    
     # For testing, just use all available languages up to max_languages
     for n_languages in [min(max_languages, len(top_languages))]:
         print(f"running evaluations for {n_languages} languages")
-        old_results = pd.read_json("results.json")
-        if old_results.empty:
+        
+        # Load existing results
+        try:
+            old_results = pd.read_json("results.json")
+            if old_results.empty:
+                old_results = pd.DataFrame(columns=["model", "bcp_47", "task", "metric", "origin", "score"])
+        except FileNotFoundError:
             old_results = pd.DataFrame(columns=["model", "bcp_47", "task", "metric", "origin", "score"])
-        old_models = pd.read_json("models.json")
+        
+        try:
+            old_models = pd.read_json("models.json")
+        except FileNotFoundError:
+            old_models = pd.DataFrame()
+            
         # get all combinations of model, language and task
         combis = [
             (model, lang.bcp_47, task_name)
@@ -60,9 +116,14 @@ async def evaluate():
         batch_size = 200  # Process 200 tasks at a time (optimized for GitHub Actions)
         all_results = []
         
-        for i in range(0, len(all_tasks), batch_size):
+        # Calculate total batches for progress tracking
+        total_batches = (len(all_tasks) + batch_size - 1) // batch_size
+        
+        for i in range(start_batch * batch_size, len(all_tasks), batch_size):
             batch = all_tasks[i:i+batch_size]
-            print(f"📦 Processing batch {i//batch_size + 1}/{(len(all_tasks) + batch_size - 1)//batch_size} ({len(batch)} tasks)")
+            current_batch = i // batch_size + 1
+            
+            print(f"📦 Processing batch {current_batch}/{total_batches} ({len(batch)} tasks)")
             
             # Show what's being evaluated in this batch
             batch_summary = {}
@@ -86,12 +147,57 @@ async def evaluate():
             for task_data in batch:
                 task_func, model, bcp_47, sentence_nr = task_data
                 batch_coroutines.append(task_func(model, bcp_47, sentence_nr))
-            batch_results = await asyncio.gather(*batch_coroutines, return_exceptions=True)
-            all_results.extend(batch_results)
+            
+            try:
+                batch_results = await asyncio.gather(*batch_coroutines, return_exceptions=True)
+                all_results.extend(batch_results)
+                
+                # Save checkpoint after each batch
+                valid_results = []
+                exception_count = 0
+                for r in batch_results:
+                    if isinstance(r, Exception):
+                        exception_count += 1
+                        continue
+                    if isinstance(r, list):
+                        valid_results.extend(r)
+                    else:
+                        valid_results.append(r)
+                
+                if valid_results:
+                    # Aggregate results
+                    batch_df = pd.DataFrame(valid_results)
+                    if len(batch_df) > 0:
+                        batch_df = (
+                            batch_df.groupby(["model", "bcp_47", "task", "metric", "origin"])
+                            .agg({"score": "mean"})
+                            .reset_index()
+                        )
+                        # Merge with existing results
+                        all_results_df = pd.concat([old_results, batch_df])
+                        all_results_df = all_results_df.drop_duplicates(subset=["model", "bcp_47", "task", "metric", "origin"])
+                        all_results_df = all_results_df.sort_values(by=["model", "bcp_47", "task", "metric"])
+                        
+                        # Save checkpoint
+                        save_checkpoint(all_results_df, models_df, languages_df, current_batch, total_batches)
+                        
+                        # Update old_results for next batch
+                        old_results = all_results_df
+                
+                print(f"✅ Batch {current_batch} completed: {len(valid_results)} valid results, {exception_count} errors")
+                
+            except Exception as e:
+                print(f"❌ Batch {current_batch} failed: {e}")
+                # Save checkpoint even on failure
+                if len(all_results) > 0:
+                    results_df = pd.DataFrame(all_results)
+                    save_checkpoint(results_df, models_df, languages_df, current_batch, total_batches)
+                continue
             
             # Reduced delay between batches (optimized for GitHub Actions)
             await asyncio.sleep(0.5)
         
+        # Final aggregation and save
         results = all_results
         # Filter out exceptions and flatten results
         valid_results = []
@@ -108,7 +214,7 @@ async def evaluate():
         print(f"⚠️  Encountered {exception_count} API errors (model unavailable/rate limits)")
         print(f"✅ Successfully processed {len(valid_results)} evaluations")
         
-        # Save partial results even if some failed
+        # Save final results
         if valid_results:
             results = valid_results
             args = dict(orient="records", indent=2, force_ascii=False)
@@ -124,6 +230,7 @@ async def evaluate():
                 # Merge with old results
                 old_results = pd.read_json("results.json")
                 results_df = pd.concat([old_results, results_df])
+                results_df = results_df.drop_duplicates(subset=["model", "bcp_47", "task", "metric", "origin"])
                 results_df = results_df.sort_values(by=["model", "bcp_47", "task", "metric"])
                 results_df.to_json("results.json", **args)
                 print(f"💾 Saved {len(results_df)} aggregated results to results.json")
@@ -153,10 +260,10 @@ async def evaluate():
             print(f"✅ Full evaluation completed in {elapsed_str}")
             print(f"🎉 Finished at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
 
-    # Save results locally
-    with open("results.json", "w") as f:
-        json.dump(results, f, indent=2)
-    print(f"💾 Results saved to results.json")
+    # Clean up checkpoint file on successful completion
+    if os.path.exists("checkpoint.json"):
+        os.remove("checkpoint.json")
+        print("🧹 Cleaned up checkpoint file")
             
     return results