Faster Persp generation

copienator/commands/gemini_for_enonce.py

@@ -6,7 +6,7 @@ import io
 import json
 import re
 from collections.abc import Sequence
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import UTC, datetime
 from pathlib import Path
 from uuid import uuid4
@@ -112,7 +112,7 @@ N'utilise pas de caractères mathématiques Unicode dans `rubric_content`.
 Par exemple :
  - Au moins 2 points si le résultat est correct.
  - Mettre la moitié des points si le raisonnement est correct mais pas le résultat.
- - Retirer 1,5 point si les hypothèses d'un théorème ou d'une question précédente ne sont pas vérifiées.
+ - Retirer 1.5 point si les hypothèses d'un théorème ou d'une question précédente ne sont pas vérifiées.
 
 Renvoie uniquement un objet JSON contenant une liste `rubrics`. Pour chaque
 question fournie, cette liste contient un objet avec son `label` exact et
@@ -333,6 +333,26 @@ def validate_groups(groups: list[list[str]], labels: list[str]) -> None:
         raise CliError("Groups must contain every existing label exactly once")
 
 
+def batch_rubric_groups(groups: list[list[str]]) -> list[list[str]]:
+    """Pack consecutive whole groups into requests of at most six questions.
+
+    An existing group larger than six is sent alone, without splitting it.
+    """
+    batches: list[list[str]] = []
+    current: list[str] = []
+    for group in groups:
+        if current and len(current) + len(group) > 6:
+            batches.append(current)
+            current = []
+        current.extend(group)
+        if len(current) >= 6:
+            batches.append(current)
+            current = []
+    if current:
+        batches.append(current)
+    return batches
+
+
 def refine_existing(workspace: EvaluationWorkspace, mode: str, *, api_client=None,
                     delete_persp: bool = False) -> ExitCode:
     """Regroup questions or generate missing rubrics, retaining completed groups."""
@@ -401,26 +421,46 @@ def refine_existing(workspace: EvaluationWorkspace, mode: str, *, api_client=Non
                 pass
             present.clear()
             print('Deleted current Persp; generating all rubrics.')
-        generated = 0
+        pending_groups = []
         for group in groups:
-            missing = [label for label in group if label not in present]
-            if not missing:
+            if all(label in present for label in group):
                 print(f"Skipping existing Persp for group: {', '.join(group)}.")
                 continue
+            pending_groups.append(group)
+        batches = batch_rubric_groups(pending_groups)
+        print(f"Generating rubrics in {len(batches)} requests, with up to 4 concurrent requests.")
+
+        def generate_batch(group):
+            missing = [label for label in group if label not in present]
             print(f"Generating missing rubric (Persp): {', '.join(missing)}...")
             group_content = (
                 "Contexte général de l’examen, fourni uniquement pour comprendre les questions :\n"
                 + context + "\n\nProduis des barèmes UNIQUEMENT pour les labels suivants : "
                 + ", ".join(missing) + "\n\n" + "\n\n".join(questions[label] for label in group)
             )
-            rubrics = generate_rubrics(
+            return generate_rubrics(
                 api_client, group_content, workspace=workspace, expected_labels=missing,
             )
-            with staged_files(workspace.root / "Persp") as staging:
-                for label, rubric in rubrics.items():
-                    (staging / label.replace("/", "_")).write_text(
-                        f"{label}\n{rubric}", encoding="utf-8")
-            generated += len(rubrics)
+
+        generated = 0
+        first_error = None
+        with ThreadPoolExecutor(max_workers=4) as executor:
+            futures = [executor.submit(generate_batch, group) for group in batches]
+            for future in as_completed(futures):
+                try:
+                    rubrics = future.result()
+                    # Commit on this thread so directory transactions never overlap.
+                    with staged_files(workspace.root / "Persp") as staging:
+                        for label, rubric in rubrics.items():
+                            (staging / label.replace("/", "_")).write_text(
+                                f"{label}\n{rubric}", encoding="utf-8")
+                    generated += len(rubrics)
+                except Exception as error:
+                    # Save other successful requests even when one request fails.
+                    if first_error is None:
+                        first_error = error
+        if first_error is not None:
+            raise first_error
         print(f"Generated {generated} missing Persp rubrics; all {len(labels)} are now present.")
     else:
         raise ValueError(f"Unknown statement refinement: {mode}")