File size: 1,545 Bytes
201dc77 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """configs/p4_cispo_lean.toml: the draft loads under the stage loader's struct
mode, selects CISPO + FP32 head, passes the P4 gate, and its denominator is the
DERIVED constant (a hand-typed one is refused by GrpoStageConfig.validate)."""
from __future__ import annotations
import pathlib
import pytest
from align.rl.cispo import assert_p4_cispo_conformant
from align.rl.grpo_loss import GRPOLossType
from align.rl.stage_config import load_stage_config
TOML = pathlib.Path(__file__).resolve().parents[1] / "configs" / "p4_cispo_lean.toml"
def _load(tmp_path, edit=lambda s: s):
band = tmp_path / "band.jsonl"
band.write_text("{}\n")
text = TOML.read_text().replace("/apps/development/rlhf/align/data/p4/lean_band0.jsonl", str(band))
p = tmp_path / "p4.toml"
p.write_text(edit(text))
return load_stage_config(str(p))
def test_draft_loads_and_selects_cispo_fp32(tmp_path):
cfg = _load(tmp_path)
g = cfg.grpo
assert g.loss_fn == GRPOLossType.cispo and g.fp32_head is True
assert g.cispo_clip_lower is None # paper: no lower bound
assert g.loss_denominator == 16 * 16 * 1536.0
assert_p4_cispo_conformant(g)
def test_hand_typed_denominator_still_refused(tmp_path):
with pytest.raises(ValueError, match="DERIVED"):
_load(tmp_path, lambda s: s.replace("loss_denominator = 393216.0", "loss_denominator = 100000.0"))
def test_unknown_grpo_key_refused(tmp_path):
with pytest.raises(TypeError):
_load(tmp_path, lambda s: s.replace("fp32_head = true", "fp32_hed = true"))
|