Promotion Space deploy (2026-03-25 05:46 UTC)

configs/math_conjecture_sota.yaml

@@ -40,6 +40,13 @@ data:
     formal_proof: 2.7
     competition: 1.2
     structured_reasoning: 1.1
+  difficulty_boost:
+    simple: 1.35
+    intermediate: 1.2
+    advanced: 1.05
+    lean_formal: 1.5
+    general: 1.0
+  lean_proof_bonus: 1.15
   system_prompt: |
     You are NorthernTribe Research's math-conjecture solver model.
     Prioritize solved-conjecture answer recovery, proof-quality reasoning,
@@ -75,14 +82,19 @@ stages:
       num_train_epochs: 1
       learning_rate: 2.0e-5
 
-  - name: conjecture_answer_alignment
-    max_train_samples: 110000
+  - name: simple_to_lean_alignment
+    max_train_samples: 120000
     max_eval_samples: 3000
     filters:
       include_families:
         - conjecture_core
         - formal_proof
-      min_sample_weight: 2.0
+        - structured_reasoning
+      include_difficulty_bands:
+        - simple
+        - intermediate
+        - lean_formal
+      min_sample_weight: 1.0
     training:
       num_train_epochs: 2
       learning_rate: 8.0e-6
@@ -90,12 +102,14 @@ stages:
       eval_steps: 220
 
   - name: formal_consistency_refinement
-    max_train_samples: 70000
+    max_train_samples: 80000
     max_eval_samples: 2200
     filters:
       include_families:
+        - formal_proof
         - conjecture_core
       require_conjecture_id: true
+      require_lean_formal: true
     training:
       num_train_epochs: 2
       learning_rate: 4.5e-6
@@ -109,6 +123,9 @@ stages:
       include_families:
         - conjecture_core
         - formal_proof
+      include_difficulty_bands:
+        - advanced
+        - lean_formal
       require_conjecture_id: true
       min_sample_weight: 3.0
     training:

configs/qwen25_math_sota.yaml

@@ -40,6 +40,13 @@ data:
     formal_proof: 2.2
     competition: 1.35
     structured_reasoning: 1.1
+  difficulty_boost:
+    simple: 1.35
+    intermediate: 1.2
+    advanced: 1.05
+    lean_formal: 1.45
+    general: 1.0
+  lean_proof_bonus: 1.12
   system_prompt: |
     You are a frontier mathematical reasoning model focused on unsolved
     conjectures. Prioritize formal consistency, proof structure, and explicit
@@ -74,14 +81,19 @@ stages:
       num_train_epochs: 1
       learning_rate: 2.0e-5
 
-  - name: conjecture_specialization
-    max_train_samples: 100000
+  - name: simple_to_lean_specialization
+    max_train_samples: 110000
     max_eval_samples: 2500
     filters:
       include_families:
         - conjecture_core
         - formal_proof
-      min_sample_weight: 2.0
+        - structured_reasoning
+      include_difficulty_bands:
+        - simple
+        - intermediate
+        - lean_formal
+      min_sample_weight: 1.0
     training:
       num_train_epochs: 2
       learning_rate: 8.0e-6
@@ -89,12 +101,14 @@ stages:
       eval_steps: 220
 
   - name: conjecture_alignment
-    max_train_samples: 60000
+    max_train_samples: 70000
     max_eval_samples: 2000
     filters:
       include_families:
+        - formal_proof
         - conjecture_core
       require_conjecture_id: true
+      require_lean_formal: true
     training:
       num_train_epochs: 2
       learning_rate: 5.0e-6
@@ -108,6 +122,9 @@ stages:
       include_families:
         - conjecture_core
         - formal_proof
+      include_difficulty_bands:
+        - advanced
+        - lean_formal
       require_conjecture_id: true
       min_sample_weight: 3.0
     training:

scripts/eval_sota.py

@@ -103,6 +103,92 @@ def as_int(value: Any, default: int) -> int:
         return default
 
 
+def canonical_difficulty(value: Any) -> str:
+    raw = as_text(value).lower()
+    if not raw or raw in {"unknown", "null", "none", "nan", "n/a"}:
+        return "general"
+    if any(key in raw for key in ("lean", "formal", "theorem", "proof", "mathlib", "proofnet")):
+        return "lean_formal"
+    if any(
+        key in raw
+        for key in (
+            "basic",
+            "easy",
+            "elementary",
+            "arithmetic",
+            "level 1",
+            "level1",
+            "level 2",
+            "level2",
+        )
+    ):
+        return "simple"
+    if any(key in raw for key in ("intermediate", "medium", "basic_to_intermediate", "level 3", "level3")):
+        return "intermediate"
+    if any(
+        key in raw
+        for key in (
+            "advanced",
+            "hard",
+            "frontier",
+            "olympiad",
+            "competition",
+            "level 4",
+            "level4",
+            "level 5",
+            "level5",
+            "geometry",
+            "algebra",
+            "calculus",
+            "number theory",
+            "combinatorics",
+            "inequalities",
+        )
+    ):
+        return "advanced"
+    return "general"
+
+
+def is_lean_formal_row(row: Dict[str, Any]) -> bool:
+    family = as_text(row.get("family")).lower()
+    if family == "formal_proof":
+        return True
+    task_type = as_text(row.get("task_type")).lower()
+    if any(token in task_type for token in ("lean", "formal", "theorem", "proof")):
+        return True
+    if as_text(row.get("proof_formal")):
+        return True
+    return canonical_difficulty(row.get("difficulty")) == "lean_formal"
+
+
+def infer_response_profile(row: Dict[str, Any]) -> str:
+    if is_lean_formal_row(row):
+        return "lean_formal"
+    band = canonical_difficulty(row.get("difficulty"))
+    if band != "general":
+        return band
+    family = as_text(row.get("family")).lower()
+    if family == "structured_reasoning":
+        return "intermediate"
+    if family in {"conjecture_core", "competition"}:
+        return "advanced"
+    return "general"
+
+
+def response_contract(profile: str) -> str:
+    if profile == "simple":
+        return "Use plain language, short steps, and end with one clear final answer."
+    if profile == "intermediate":
+        return "Give a compact derivation with key equations and a clean final conclusion."
+    if profile == "advanced":
+        return "Provide rigorous reasoning, explicit assumptions, and uncertainty when unresolved."
+    if profile == "lean_formal":
+        return (
+            "Provide a proof sketch plus a Lean-oriented theorem/lemma structure and verification plan."
+        )
+    return "Provide technically correct reasoning and separate facts from conjectural claims."
+
+
 def load_config(path: Path) -> Dict[str, Any]:
     cfg = yaml.safe_load(path.read_text(encoding="utf-8"))
     if not isinstance(cfg, dict):
@@ -296,6 +382,10 @@ def build_user_block(row: Dict[str, Any], data_cfg: Dict[str, Any]) -> str:
         value = as_text(row.get(key))
         if value:
             lines.append(f"{label}: {value}")
+    profile = infer_response_profile(row)
+    lines.append(f"Difficulty band: {canonical_difficulty(row.get('difficulty'))}")
+    lines.append(f"Response profile: {profile}")
+    lines.append(f"Response contract: {response_contract(profile)}")
     if lines:
         return f"{prompt}\n\nMetadata:\n" + "\n".join(lines)
     return prompt
@@ -396,6 +486,18 @@ def finalize_bucket(bucket: Dict[str, Any]) -> Dict[str, Any]:
     }
 
 
+def merge_buckets(buckets: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
+    merged = make_bucket()
+    for bucket in buckets:
+        merged["evaluated_rows"] += int(bucket.get("evaluated_rows", 0))
+        merged["pass_at_1_hits"] += int(bucket.get("pass_at_1_hits", 0))
+        merged["pass_at_k_hits"] += int(bucket.get("pass_at_k_hits", 0))
+        merged["exact_at_1_hits"] += int(bucket.get("exact_at_1_hits", 0))
+        merged["exact_at_k_hits"] += int(bucket.get("exact_at_k_hits", 0))
+        merged["boxed_at_k_hits"] += int(bucket.get("boxed_at_k_hits", 0))
+    return merged
+
+
 def resolve_eval_file(arg_eval_file: Optional[Path], cfg: Dict[str, Any]) -> Path:
     if arg_eval_file is not None:
         return arg_eval_file
@@ -457,6 +559,8 @@ def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]:
     totals = make_bucket()
     family_buckets: Dict[str, Dict[str, Any]] = {}
     difficulty_buckets: Dict[str, Dict[str, Any]] = {}
+    difficulty_band_buckets: Dict[str, Dict[str, Any]] = {}
+    response_profile_buckets: Dict[str, Dict[str, Any]] = {}
 
     processed_rows = 0
     skipped_no_expected = 0
@@ -525,6 +629,30 @@ def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]:
             boxedk=boxedk,
         )
 
+        difficulty_band = canonical_difficulty(row.get("difficulty"))
+        if difficulty_band not in difficulty_band_buckets:
+            difficulty_band_buckets[difficulty_band] = make_bucket()
+        update_bucket(
+            difficulty_band_buckets[difficulty_band],
+            hit1=hit1,
+            hitk=hitk,
+            exact1=exact1,
+            exactk=exactk,
+            boxedk=boxedk,
+        )
+
+        response_profile = infer_response_profile(row)
+        if response_profile not in response_profile_buckets:
+            response_profile_buckets[response_profile] = make_bucket()
+        update_bucket(
+            response_profile_buckets[response_profile],
+            hit1=hit1,
+            hitk=hitk,
+            exact1=exact1,
+            exactk=exactk,
+            boxedk=boxedk,
+        )
+
         processed_rows += 1
         if args.progress_every > 0 and processed_rows % args.progress_every == 0:
             print(f"Progress: evaluated_rows={processed_rows} latest_family={family}")
@@ -535,6 +663,8 @@ def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]:
                     "uid": as_text(row.get("uid")),
                     "family": family,
                     "difficulty": difficulty,
+                    "difficulty_band": difficulty_band,
+                    "response_profile": response_profile,
                     "prompt": as_text(row.get(prompt_field)),
                     "expected_values": expected_values[:5],
                     "candidates": candidates,
@@ -554,6 +684,22 @@ def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]:
 
     composite_score = 0.30 * pass_at_1 + 0.50 * pass_at_k + 0.20 * exact_at_k
 
+    simple_bundle = merge_buckets(
+        [response_profile_buckets[key] for key in ("simple", "intermediate") if key in response_profile_buckets]
+    )
+    lean_bundle = merge_buckets(
+        [response_profile_buckets[key] for key in ("lean_formal",) if key in response_profile_buckets]
+    )
+    simple_metrics = finalize_bucket(simple_bundle)
+    lean_metrics = finalize_bucket(lean_bundle)
+    simple_to_lean = {
+        "simple_rows": simple_metrics["evaluated_rows"],
+        "simple_pass_at_k": simple_metrics["pass_at_k"],
+        "lean_formal_rows": lean_metrics["evaluated_rows"],
+        "lean_formal_pass_at_k": lean_metrics["pass_at_k"],
+        "pass_at_k_gap_simple_minus_lean": simple_metrics["pass_at_k"] - lean_metrics["pass_at_k"],
+    }
+
     report: Dict[str, Any] = {
         "base_model": base_model,
         "adapter_path": str(args.adapter_path) if args.adapter_path is not None else None,
@@ -582,6 +728,15 @@ def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]:
             key: finalize_bucket(difficulty_buckets[key])
             for key in sorted(difficulty_buckets.keys())
         },
+        "difficulty_band_metrics": {
+            key: finalize_bucket(difficulty_band_buckets[key])
+            for key in sorted(difficulty_band_buckets.keys())
+        },
+        "response_profile_metrics": {
+            key: finalize_bucket(response_profile_buckets[key])
+            for key in sorted(response_profile_buckets.keys())
+        },
+        "simple_to_lean": simple_to_lean,
         "samples": samples,
     }
 

scripts/train_sota.py

@@ -127,6 +127,93 @@ def as_bool(value: Any, default: bool = False) -> bool:
     return default
 
 
+def canonical_difficulty(value: Any) -> str:
+    raw = as_text(value).lower()
+    if not raw or raw in {"unknown", "null", "none", "nan", "n/a"}:
+        return "general"
+
+    if any(key in raw for key in ("lean", "formal", "theorem", "proof", "mathlib", "proofnet")):
+        return "lean_formal"
+    if any(
+        key in raw
+        for key in (
+            "basic",
+            "easy",
+            "elementary",
+            "arithmetic",
+            "level 1",
+            "level1",
+            "level 2",
+            "level2",
+        )
+    ):
+        return "simple"
+    if any(key in raw for key in ("intermediate", "medium", "basic_to_intermediate", "level 3", "level3")):
+        return "intermediate"
+    if any(
+        key in raw
+        for key in (
+            "advanced",
+            "hard",
+            "frontier",
+            "olympiad",
+            "competition",
+            "level 4",
+            "level4",
+            "level 5",
+            "level5",
+            "geometry",
+            "algebra",
+            "calculus",
+            "number theory",
+            "combinatorics",
+            "inequalities",
+        )
+    ):
+        return "advanced"
+    return "general"
+
+
+def is_lean_formal_row(row: Dict[str, Any]) -> bool:
+    family = as_text(row.get("family")).lower()
+    if family == "formal_proof":
+        return True
+    task_type = as_text(row.get("task_type")).lower()
+    if any(token in task_type for token in ("lean", "formal", "theorem", "proof")):
+        return True
+    if as_text(row.get("proof_formal")):
+        return True
+    return canonical_difficulty(row.get("difficulty")) == "lean_formal"
+
+
+def infer_response_profile(row: Dict[str, Any]) -> str:
+    if is_lean_formal_row(row):
+        return "lean_formal"
+    band = canonical_difficulty(row.get("difficulty"))
+    if band != "general":
+        return band
+    family = as_text(row.get("family")).lower()
+    if family == "structured_reasoning":
+        return "intermediate"
+    if family in {"conjecture_core", "competition"}:
+        return "advanced"
+    return "general"
+
+
+def response_contract(profile: str) -> str:
+    if profile == "simple":
+        return "Use plain language, short steps, and end with one clear final answer."
+    if profile == "intermediate":
+        return "Give a compact derivation with key equations and a clean final conclusion."
+    if profile == "advanced":
+        return "Provide rigorous reasoning, explicit assumptions, and uncertainty when unresolved."
+    if profile == "lean_formal":
+        return (
+            "Provide a proof sketch plus a Lean-oriented theorem/lemma structure and verification plan."
+        )
+    return "Provide technically correct reasoning and separate facts from conjectural claims."
+
+
 def load_config(path: Path) -> Dict[str, Any]:
     if not path.exists():
         raise FileNotFoundError(f"Config not found: {path}")
@@ -239,6 +326,10 @@ def build_user_block(row: Dict[str, Any], data_cfg: Dict[str, Any]) -> str:
         tag_text = ", ".join(as_text(tag) for tag in tags if as_text(tag))
         if tag_text:
             meta_lines.append(f"Tags: {tag_text}")
+    profile = infer_response_profile(row)
+    meta_lines.append(f"Difficulty band: {canonical_difficulty(row.get('difficulty'))}")
+    meta_lines.append(f"Response profile: {profile}")
+    meta_lines.append(f"Response contract: {response_contract(profile)}")
     if not meta_lines:
         return prompt
     return f"{prompt}\n\nMetadata:\n" + "\n".join(meta_lines)
@@ -291,6 +382,13 @@ def compute_loss_weight(row: Dict[str, Any], data_cfg: Dict[str, Any]) -> float:
     family_boost = data_cfg.get("family_boost", {})
     if isinstance(family_boost, dict):
         base *= as_float(family_boost.get(family), 1.0)
+    difficulty_boost = data_cfg.get("difficulty_boost", {})
+    if isinstance(difficulty_boost, dict):
+        band = canonical_difficulty(row.get("difficulty"))
+        base *= as_float(difficulty_boost.get(band), 1.0)
+    lean_bonus = as_float(data_cfg.get("lean_proof_bonus"), 1.0)
+    if lean_bonus > 0 and is_lean_formal_row(row):
+        base *= lean_bonus
     min_w = as_float(data_cfg.get("min_loss_weight"), 0.1)
     max_w = as_float(data_cfg.get("max_loss_weight"), 8.0)
     if min_w > max_w:
@@ -331,7 +429,14 @@ def apply_filters(dataset: Dataset, filter_cfg: Dict[str, Any]) -> Dataset:
     exclude_families = set(filter_cfg.get("exclude_families", []) or [])
     include_task_types = set(filter_cfg.get("include_task_types", []) or [])
     source_datasets = set(filter_cfg.get("source_datasets", []) or [])
+    include_difficulty_bands = {
+        as_text(value).lower() for value in (filter_cfg.get("include_difficulty_bands", []) or []) if as_text(value)
+    }
+    exclude_difficulty_bands = {
+        as_text(value).lower() for value in (filter_cfg.get("exclude_difficulty_bands", []) or []) if as_text(value)
+    }
     require_conjecture_id = bool(filter_cfg.get("require_conjecture_id", False))
+    require_lean_formal = bool(filter_cfg.get("require_lean_formal", False))
     min_sample_weight = filter_cfg.get("min_sample_weight")
     min_sample_weight = as_float(min_sample_weight, 0.0) if min_sample_weight is not None else None
 
@@ -349,10 +454,17 @@ def apply_filters(dataset: Dataset, filter_cfg: Dict[str, Any]) -> Dataset:
             source = as_text(row.get("source_dataset"))
             if source not in source_datasets:
                 return False
+        difficulty_band = canonical_difficulty(row.get("difficulty"))
+        if include_difficulty_bands and difficulty_band not in include_difficulty_bands:
+            return False
+        if exclude_difficulty_bands and difficulty_band in exclude_difficulty_bands:
+            return False
         if require_conjecture_id:
             conjecture_id = as_text(row.get("conjecture_id"))
             if not conjecture_id or conjecture_id.lower() == "null":
                 return False
+        if require_lean_formal and not is_lean_formal_row(row):
+            return False
         if min_sample_weight is not None:
             sample_weight = as_float(row.get("sample_weight"), 0.0)
             if sample_weight < min_sample_weight:

tests/test_core_utils.py

@@ -10,15 +10,32 @@ import unittest
 from unittest import mock
 from pathlib import Path
 
+try:
+    from datasets import Dataset
+except ModuleNotFoundError:  # pragma: no cover - optional test dependency in this environment
+    Dataset = None
+
 ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
-import app
-from scripts import eval_sota
-from scripts import train_sota
+try:
+    import app
+except Exception:  # pragma: no cover - optional test dependency in this environment
+    app = None
+
+try:
+    from scripts import eval_sota
+except Exception:  # pragma: no cover - optional test dependency in this environment
+    eval_sota = None
+
+try:
+    from scripts import train_sota
+except Exception:  # pragma: no cover - optional test dependency in this environment
+    train_sota = None
 
 
+@unittest.skipUnless(app is not None, "app runtime dependencies are not installed")
 class AppUtilityTests(unittest.TestCase):
     def test_validate_repo_id_accepts_valid(self) -> None:
         self.assertEqual(
@@ -90,6 +107,7 @@ class AppUtilityTests(unittest.TestCase):
             self.assertTrue((records_dir / "run-20260102-030405.json").exists())
 
 
+@unittest.skipUnless(eval_sota is not None, "eval_sota runtime dependencies are not installed")
 class EvalUtilityTests(unittest.TestCase):
     def test_parse_numeric_fraction(self) -> None:
         value = eval_sota.parse_numeric_value("3/4")
@@ -102,7 +120,15 @@ class EvalUtilityTests(unittest.TestCase):
         self.assertTrue(result["match"])
         self.assertTrue(result["boxed"] or result["exact"])
 
+    def test_infer_response_profile_handles_formal_and_non_formal_rows(self) -> None:
+        formal_row = {"family": "formal_proof", "difficulty": "formal_proof"}
+        simple_row = {"family": "problem_solving", "difficulty": "basic"}
 
+        self.assertEqual(eval_sota.infer_response_profile(formal_row), "lean_formal")
+        self.assertEqual(eval_sota.infer_response_profile(simple_row), "simple")
+
+
+@unittest.skipUnless(train_sota is not None, "train_sota runtime dependencies are not installed")
 class TrainUtilityTests(unittest.TestCase):
     def test_as_bool_conversions(self) -> None:
         self.assertTrue(train_sota.as_bool("yes"))
@@ -110,6 +136,38 @@ class TrainUtilityTests(unittest.TestCase):
         self.assertTrue(train_sota.as_bool(True))
         self.assertFalse(train_sota.as_bool(None, default=False))
 
+    def test_canonical_difficulty_mappings(self) -> None:
+        self.assertEqual(train_sota.canonical_difficulty("basic_to_intermediate"), "simple")
+        self.assertEqual(train_sota.canonical_difficulty("formal_proof"), "lean_formal")
+        self.assertEqual(train_sota.canonical_difficulty("olympiad"), "advanced")
+
+    def test_apply_filters_include_bands_and_require_lean_formal(self) -> None:
+        if Dataset is None:
+            self.skipTest("datasets is not installed")
+
+        dataset = Dataset.from_dict(
+            {
+                "family": ["formal_proof", "problem_solving", "competition"],
+                "task_type": ["theorem_proving", "word_problem", "olympiad"],
+                "source_dataset": ["src-a", "src-b", "src-c"],
+                "difficulty": ["formal_proof", "basic_to_intermediate", "olympiad"],
+                "conjecture_id": ["c1", "c2", "c3"],
+                "sample_weight": [1.0, 1.0, 1.0],
+            }
+        )
+
+        filtered = train_sota.apply_filters(
+            dataset,
+            {
+                "include_difficulty_bands": ["lean_formal", "simple"],
+                "require_lean_formal": True,
+            },
+        )
+
+        self.assertEqual(len(filtered), 1)
+        self.assertEqual(filtered[0]["family"], "formal_proof")
+        self.assertEqual(filtered[0]["difficulty"], "formal_proof")
+
     def test_build_tokenizer_falls_back_when_protobuf_missing(self) -> None:
         class DummyTokenizer:
             def __init__(self) -> None:
@@ -135,6 +193,7 @@ class TrainUtilityTests(unittest.TestCase):
         self.assertEqual(tok.pad_token, "<eos>")
 
 
+@unittest.skipUnless(eval_sota is not None, "eval_sota runtime dependencies are not installed")
 class EvalTokenizerFallbackTests(unittest.TestCase):
     def test_eval_tokenizer_falls_back_when_protobuf_missing(self) -> None:
         class DummyTokenizer:
@@ -171,6 +230,7 @@ class EvalTokenizerFallbackTests(unittest.TestCase):
         self.assertEqual(tok.pad_token, "<eos>")
 
 
+@unittest.skipUnless(app is not None, "app runtime dependencies are not installed")
 class ContinuousModeSafetyTests(unittest.TestCase):
     def test_continuous_mode_halts_after_consecutive_failures(self) -> None:
         original_max = app.CONTINUOUS_MAX_CONSECUTIVE_FAILURES