Code / public_research_code/run_synthetic_demo.py
public_research_code/run_synthetic_demo.py
1
"""Run deterministic public seams without data, services, or model runtimes."""
2
3
from __future__ import annotations
4
5
import json
6
import sys
7
from pathlib import Path
8
9
sys.dont_write_bytecode = True
10
11
12
def _add_source_paths(root: Path) -> None:
13
"""Expose preserved source roots for the generated public edition."""
14
15
source_paths = (
16
root,
17
root / "libs/common_core/src",
18
root / "libs/huleedu_nlp_shared/src",
19
)
20
for source_path in reversed(source_paths):
21
source_text = str(source_path)
22
if source_text not in sys.path:
23
sys.path.insert(0, source_text)
24
25
26
def main() -> int:
27
"""Execute one deterministic cross-seam synthetic demonstration."""
28
29
root = Path(__file__).resolve().parent.parent
30
_add_source_paths(root)
31
32
import numpy as np
33
from huleedu_nlp_shared.feature_contracts import CURRENT_FEATURE_REGISTRY
34
35
from scripts.ml_training.essay_scoring.training.prediction_mapping import (
36
map_round_half_band,
37
)
38
from services.cj_assessment_service.cj_core_logic.bt_inference import (
39
EssayId,
40
compute_raw_bt_information_state,
41
)
42
43
raw_predictions = np.array([0.9, 1.26, 2.74, 3.6], dtype=float)
44
mapped_predictions = map_round_half_band(
45
y_pred_raw=raw_predictions,
46
min_band=1.0,
47
max_band=3.0,
48
)
49
essay_ids = tuple(EssayId(value) for value in ("synthetic-a", "synthetic-b", "anchor"))
50
information = compute_raw_bt_information_state(
51
ordered_ids=essay_ids,
52
pairs=((essay_ids[0], essay_ids[1]), (essay_ids[1], essay_ids[2])),
53
theta_by_id={essay_ids[0]: 0.3, essay_ids[1]: 0.0, essay_ids[2]: -0.3},
54
anchor_ids=(essay_ids[2],),
55
)
56
if information.is_err:
57
raise RuntimeError(information.error.message)
58
state = information.value
59
if mapped_predictions.tolist() != [1.0, 1.5, 2.5, 3.0]:
60
raise RuntimeError("synthetic score mapping drifted")
61
if state.reference_id != essay_ids[2]:
62
raise RuntimeError("synthetic Bradley--Terry anchor handling drifted")
63
64
print(
65
json.dumps(
66
{
67
"bradley_terry_reference": state.reference_id,
68
"feature_registry": CURRENT_FEATURE_REGISTRY.registry_id,
69
"feature_registry_version": CURRENT_FEATURE_REGISTRY.registry_version,
70
"mapped_predictions": mapped_predictions.tolist(),
71
"raw_standard_errors": [
72
state.raw_standard_error_by_id[essay_id] for essay_id in essay_ids
73
],
74
},
75
sort_keys=True,
76
)
77
)
78
return 0
79
80
81
if __name__ == "__main__":
82
raise SystemExit(main())