File size: 4,389 Bytes
784f572 | 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 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import json
from pathlib import Path
def _write_samples(root: Path, step: int, texts: list[str]) -> Path:
sample_dir = root / "results" / "samples" / "lm1b" / "rdlm" / f"steps_{step}"
sample_dir.mkdir(parents=True, exist_ok=True)
with (sample_dir / "samples.jsonl").open("w", encoding="utf-8") as handle:
for index, text in enumerate(texts):
handle.write(
json.dumps(
{
"sample_id": index,
"text": text,
"token_ids": [0, 101, index + 1, 102],
"seed": 42,
"generation_seconds": 0.0,
}
)
+ "\n"
)
return sample_dir
def _write_metrics(
root: Path, step: int, *, perplexity: float, entropy: float, self_bleu: float
) -> None:
metrics_path = (
root / "results" / "metrics" / "lm1b" / "rdlm" / f"steps_{step}" / "metrics.json"
)
metrics_path.parent.mkdir(parents=True, exist_ok=True)
metrics_path.write_text(
json.dumps(
{
"schema_version": 1,
"sample_count": 4,
"partial": False,
"metrics": {
"generative_perplexity": {"perplexity": perplexity},
"unigram_entropy": {"mean_entropy": entropy},
"self_bleu": {"score": self_bleu},
},
}
),
encoding="utf-8",
)
def _write_conversion(sample_dir: Path, source: str, reason: str | None = None) -> None:
transformation = {"operation": "validated_exact_length"}
if reason is not None:
transformation["reason"] = reason
(sample_dir / "conversion_metadata.json").write_text(
json.dumps(
{
"format": "dlb-upstream-token-capture-v1",
"token_ids_source": source,
"token_ids_transformation": transformation,
}
),
encoding="utf-8",
)
def test_rdlm_diagnostics_flags_collapsed_low_ppl_step(tmp_path: Path) -> None:
"""Catch reading a low generative-PPL row as a quality win despite collapse signals."""
from dlb.rdlm_diagnostics import diagnose_rdlm
collapsed_dir = _write_samples(tmp_path, 1, ["the same short sample"] * 4)
_write_metrics(tmp_path, 1, perplexity=22.7, entropy=3.2, self_bleu=0.99)
_write_conversion(
collapsed_dir, "retokenized", "upstream_ids_not_canonical_length"
)
normal_dir = _write_samples(
tmp_path,
32,
[
"a longer generated sentence about policy and finance with enough separate clauses to resemble a full sample",
"another generated sentence with separate wording and enough content to avoid the short text warning path",
"markets discussed a different event in parliament while officials compared several economic indicators",
"officials released new numbers after the meeting and said the report would be reviewed next month",
],
)
_write_metrics(tmp_path, 32, perplexity=375.94, entropy=4.33, self_bleu=0.12)
_write_conversion(normal_dir, "upstream")
report = diagnose_rdlm(tmp_path, steps=(1, 32), expected_sample_count=4)
assert report["schema"] == "dlb-rdlm-diagnostics-v1"
assert report["summary"]["suspect_steps"] == [1]
assert report["summary"]["verdict"] == "low_ppl_points_show_collapse_signals"
collapsed = report["rows"][0]
assert collapsed["steps"] == 1
assert collapsed["unique_text_ratio"] == 0.25
assert collapsed["token_ids_source"] == "retokenized"
assert collapsed["token_ids_reason"] == "upstream_ids_not_canonical_length"
assert {
"low_ppl_with_low_entropy",
"high_self_bleu",
"high_duplicate_texts",
"retokenized_from_text",
} <= set(collapsed["warnings"])
def test_rdlm_diagnostics_reports_missing_step_artifacts(tmp_path: Path) -> None:
"""Catch silently omitting RDLM steps whose server artifacts were never produced."""
from dlb.rdlm_diagnostics import diagnose_rdlm
report = diagnose_rdlm(tmp_path, steps=(1024,), expected_sample_count=4)
assert report["summary"]["missing_steps"] == [1024]
assert report["rows"][0]["warnings"] == ["missing_samples", "missing_metrics"]
|