From d7e979c4df2db2f873910dbce941195acb572602 Mon Sep 17 00:00:00 2001 From: George Date: Tue, 21 Jul 2026 10:52:37 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=92=20eval:=20close=20exp=5Flegal=20me?= =?UTF-8?q?morization=20holes=20(seeded=20pick=20+=20data-quality=20gates)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Miners scored val_loss ≈ 0.04 vs baselines of 4.5–12.5 (perplexity ~1.04) with all miners within 0.0006 of each other. Root cause is the eval data layer, not the seed: the MinerCommit2 block-hash seed (PR #175/#177) was verified correct on chain for all completed rounds. Validated on the live validator by reproducing the round-8670446 eval stream with production code: - exp_legal still ran the legacy head-of-stream path (§2 shortcut of docs/exp-legal-migration-plan.md) — the whole draw came from the head of one sub-file; - 38% of streamed Multi_Legal_Pile rows have empty text → all-padding batches → NaN loss → silently excluded from the scored divisor; - 75% of non-empty rows share an identical 200-char prefix (templated boilerplate; zlib 0.364 vs c4 0.527); - truncation always scored each document's first 1024 tokens — its most templated region. Changes: - eval_shard_pick: row_count_source="verified_table" (frozen per-shard counts double as the shard allowlist), leaf_name_pattern override, and load_builder script-bypass (Multi_Legal_Pile's builder streams files from external repos and needs trust_remote_code; the generic json builder reads the pinned native shard directly). Register (joelniklaus/Multi_Legal_Pile, all_all) @ 911e1d21 with all 29 native shards above the 10k-row headroom floor (counted by full decompress). - dataloader: eval-only deterministic gates eval_min_text_chars=200 and eval_dedup_prefix_chars=200 (exact-prefix set — builtin hash() is per-process randomized and could break cross-validator consensus). - tokenize_windowed: long docs contribute a content-hash-derived window instead of the boilerplate prefix (no RNG, consensus-safe). Adopted by the default dataset class and exp_legal; exp_math left for follow-up. - exp_legal config: eval_source_seeded_shard_pick: true (shortcut retired). Miner training still streams the all_all builder mix, so eval ⊆ training distribution. - evaluate: scored_batches/nan_batches log promoted to INFO — the only production signal that an average is silently excluding NaN batches. Rollout: eval batches change; all validators must upgrade together (same discipline as the shuffle/skip bumps). Co-Authored-By: Claude Fable 5 --- connito/shared/config.py | 19 +++ connito/shared/dataloader.py | 112 ++++++++++++- connito/shared/eval_shard_pick.py | 162 +++++++++++++++++-- connito/shared/evaluate.py | 7 +- connito/test/test_eval_data_quality.py | 211 +++++++++++++++++++++++++ expert_groups/exp_legal/config.yaml | 18 ++- expert_groups/exp_legal/dataset.py | 21 ++- 7 files changed, 515 insertions(+), 35 deletions(-) create mode 100644 connito/test/test_eval_data_quality.py diff --git a/connito/shared/config.py b/connito/shared/config.py index 071ad82..a410e72 100644 --- a/connito/shared/config.py +++ b/connito/shared/config.py @@ -305,6 +305,25 @@ class DataCfg(BaseConfig): # target and a mid-rollout HF re-upload would cause two # validators to pick different rows for the same seed. eval_source_revision_pin: dict[str, str] | None = None + # --- Eval-stream data-quality gates (validator eval path only; the + # miner training stream passes seed=None and is unaffected). Both + # are deterministic, so all validators on the same version keep + # identical eval batches. Same coordinated-rollout discipline as + # the other eval_source_* knobs. + # + # Drop rows whose text is shorter than this many characters after + # strip(). Motivated by Multi_Legal_Pile all_all, where 38% of + # streamed rows have empty text: those tokenize to all-padding + # sequences whose loss is NaN, silently shrinking the scored eval + # sample. 0 disables. + eval_min_text_chars: int = 200 + # Keep only the first row per distinct text prefix of this many + # characters. Templated corpora repeat opening boilerplate across + # documents (measured: 75% of non-empty Multi_Legal_Pile rows share + # an identical 200-char prefix); scoring duplicates hands + # template-memorizing miners near-zero loss on "unseen" rows. + # 0 disables. + eval_dedup_prefix_chars: int = 200 @model_validator(mode="after") def _validate_dataset_sources(self): diff --git a/connito/shared/dataloader.py b/connito/shared/dataloader.py index e565bb9..3768298 100644 --- a/connito/shared/dataloader.py +++ b/connito/shared/dataloader.py @@ -45,6 +45,90 @@ def _fractional_index_filter(_example, idx: int, seed: str | int, threshold: int return score <= threshold +def _min_text_chars_filter(example: dict[str, Any], min_chars: int) -> bool: + """Drop rows whose text is empty or trivially short. + + Eval-path only. Empirically (Multi_Legal_Pile all_all, 2026-07-21) + 38% of streamed rows carry a completely empty `text`; those rows + tokenize to all-padding sequences whose labels are fully masked, so + they produce NaN eval loss and silently drop out of the scored-batch + divisor — shrinking an already tiny eval sample. The distribution is + bimodal (0 chars or ≥200), so a low threshold removes exactly the + degenerate rows without biasing content selection. + """ + return len(str(example.get("text", "")).strip()) >= min_chars + + +class _PrefixDedupFilter: + """Keep only the first row for each distinct text prefix. + + Eval-path only, and only sound single-worker: the `seen` set lives in + this instance, so the eval dataloader must iterate the stream in one + process (the eval loader runs `num_workers<=1`; see `get_dataloader`). + + Rationale: templated corpora repeat their opening boilerplate across + documents — measured 75% of non-empty Multi_Legal_Pile rows sharing an + identical 200-char prefix. Scoring many near-identical rows lets a + miner fine-tuned on the template reach near-zero loss on "unseen" + documents. Deduplicating by prefix keeps one representative per + template instead of a batch full of copies. Deterministic given a + deterministic input stream (same seed → same order → same survivors). + """ + + def __init__(self, prefix_chars: int): + self.prefix_chars = int(prefix_chars) + # Exact prefixes, not `hash()` digests: builtin str hashing is + # per-process randomized, so two validators could disagree on a + # collision. The eval stream retains only ~thousands of rows, so + # exact storage is a few hundred KB at worst. + self.seen: set[str] = set() + + def __call__(self, example: dict[str, Any]) -> bool: + prefix = str(example.get("text", ""))[: self.prefix_chars] + if prefix in self.seen: + return False + self.seen.add(prefix) + return True + + +def tokenize_windowed( + text: str, tokenizer: PreTrainedTokenizerBase, sequence_length: int +) -> dict[str, list]: + """Tokenize `text`, sampling a deterministic window from long documents. + + The previous behavior (`truncation=True`) always scored/trained on a + document's FIRST `sequence_length` tokens. For templated corpora + (legal filings, papers) the prefix is the most boilerplate-heavy, + most predictable region of the document — document bodies never + entered the pipeline at all. This helper instead: + + - short documents (≤ sequence_length tokens): pad to length, as before; + - long documents: take a `sequence_length` window whose start is + derived from the text's own content hash — deterministic for every + validator (consensus-safe: no RNG, no config), uniform-ish across + the document, and not influenceable by the validator. + + The raw text is capped at `sequence_length * 40` chars before + tokenizing to bound tokenizer cost on pathological documents (real + corpora run 3–8 chars/token, so the cap only bites degenerate input). + """ + capped = str(text)[: sequence_length * 40] + ids = tokenizer(capped, truncation=False, add_special_tokens=True)["input_ids"] + if len(ids) > sequence_length: + span = len(ids) - sequence_length + start = h256_int("token_window", capped) % (span + 1) + window = ids[start : start + sequence_length] + return {"input_ids": window, "attention_mask": [1] * sequence_length} + pad_id = tokenizer.pad_token_id + if pad_id is None: + pad_id = tokenizer.eos_token_id if tokenizer.eos_token_id is not None else 0 + n = len(ids) + return { + "input_ids": ids + [pad_id] * (sequence_length - n), + "attention_mask": [1] * n + [0] * (sequence_length - n), + } + + # ----------------------------- # Dataset # ----------------------------- @@ -86,7 +170,7 @@ def tokenize_and_format( example: dict[str, str], tokenizer: PreTrainedTokenizerBase, sequence_length: int ) -> dict[str, list]: text = example.get("text", "") - return tokenizer(text, truncation=True, max_length=sequence_length, padding="max_length") # type: ignore + return tokenize_windowed(text, tokenizer, sequence_length) @classmethod def get_tokenised_dataset( @@ -272,6 +356,20 @@ def ensure_string(example: dict[str, Any], source_text_column: str): features=common_features, ) + # Eval-path data-quality gate (seed is None on the miner + # training path, which stays byte-identical). Applied + # per-source and before interleave so the source weights + # keep describing *usable* rows — an unfiltered source with + # 38% empty rows would otherwise contribute 38% NaN batches + # at its configured weight. + eval_min_text_chars = int( + getattr(config.task.exp.data, "eval_min_text_chars", 0) or 0 + ) + if seed is not None and eval_min_text_chars > 0: + source_split = source_split.filter( + partial(_min_text_chars_filter, min_chars=eval_min_text_chars) + ) + if pick is not None: # In-shard offset goes here so the validator's read # window lands at a random depth inside the chosen @@ -351,6 +449,18 @@ def ensure_string(example: dict[str, Any], source_text_column: str): logger.debug("Interleaving dataset sources", probabilities=probabilities) split = interleave_datasets(dataset_splits, probabilities=probabilities, seed=int_seed) + # Eval-path template dedup: drop rows repeating an already-seen + # text prefix (templated corpora open millions of documents with + # the same boilerplate — see `_PrefixDedupFilter`). Runs after + # interleave so the dedup window spans the whole eval stream, and + # before the fractional filter so surviving indices stay + # deterministic for every validator. + eval_dedup_prefix_chars = int( + getattr(config.task.exp.data, "eval_dedup_prefix_chars", 0) or 0 + ) + if seed is not None and eval_dedup_prefix_chars > 0: + split = split.filter(_PrefixDedupFilter(eval_dedup_prefix_chars)) + # Optional deterministic subsampling based on (seed, fraction) # Applied *before* sharding on the streaming iterable. if seed is not None and fraction is not None and fraction < 1.0: diff --git a/connito/shared/eval_shard_pick.py b/connito/shared/eval_shard_pick.py index dcb0017..6902728 100644 --- a/connito/shared/eval_shard_pick.py +++ b/connito/shared/eval_shard_pick.py @@ -142,10 +142,25 @@ class _SourceShardPolicy: path_prefix: str path_suffix: tuple[str, ...] revision: str - row_count_source: str # "constant" | "parquet_footer" + row_count_source: str # "constant" | "parquet_footer" | "verified_table" safe_floor_rows: int | None = None min_headroom_rows: int = 10_000 verified_shard_rows: dict[str, int] = field(default_factory=dict) + # Override for `_SHARD_NAME_FILTER` when a source's data files don't + # follow the `train`/`part_` leaf naming convention (e.g. + # Multi_Legal_Pile ships `data///.jsonl.xz`). + # Only consulted for listing-based policies; `verified_table` + # policies take their shard list from the table and skip name + # filtering entirely. + leaf_name_pattern: str | None = None + # When set (e.g. "json"), `load_streaming_shard` loads the shard via + # the named generic builder against a resolved `hf_hub_url` instead + # of `load_dataset(repo_id, data_files=...)`. Required for repos + # that ship a custom loading script: passing the repo id would + # execute the script (defeating both the shard pick and the + # `trust_remote_code` opt-out), while the generic builder reads the + # raw file directly. + load_builder: str | None = None # Known sources. Add new entries here, NOT via config — the consensus @@ -184,6 +199,61 @@ class _SourceShardPolicy: # count per shard at pick time. min_headroom_rows=10_000, ), + ("joelniklaus/Multi_Legal_Pile", "all_all"): _SourceShardPolicy( + # The repo's NATIVE data files. Deliberately NOT the `all_all` + # builder-script mix: the script streams additional files from + # external repos (joelito/eurlex_resources, legal-mc4, …) that + # can't be pinned or row-counted here, and executing it requires + # `trust_remote_code`. The eval pool is therefore a (large) + # subset of the miner-training distribution — acceptable: eval ⊆ + # train, and the native corpus is tens of GB. + path_prefix="data/", + path_suffix=(".jsonl.xz",), + # Pinned at registration time (2026-07-21). Bump together with a + # re-count of the table if the dataset is ever re-uploaded. + revision="911e1d214162fd11d2c78d3f1428cbfcbe07782c", + row_count_source="verified_table", + min_headroom_rows=10_000, + # `.jsonl.xz` has no footer — rows counted once offline (full + # decompress of every shard at the pinned revision, 2026-07-21) + # and frozen here. The table doubles as the allowlist; four + # shards with ≤10k rows (denmark_ddsc caselaw 4 442, + # en switzerland_lexfind 147, belgium_jurportal 2 221, + # it switzerland_lexfind 5 642) are deliberately left out — no + # safe offset exists above the headroom floor. + load_builder="json", + verified_shard_rows={ + "data/bg/legislation/bulgaria_marcell.jsonl.xz": 29_549, + "data/cs/caselaw/czechia_constitutional_court.jsonl.xz": 73_086, + "data/cs/caselaw/czechia_supreme_administrative_court.jsonl.xz": 52_660, + "data/cs/caselaw/czechia_supreme_court.jsonl.xz": 111_977, + "data/da/legislation/denmark_ddsc.jsonl.xz": 64_043, + "data/de/caselaw/germany_openlegaldata.jsonl.xz": 201_676, + "data/de/caselaw/switzerland_entscheidsuche.jsonl.xz": 308_612, + "data/de/legislation/germany_openlegaldata.jsonl.xz": 52_918, + "data/de/legislation/switzerland_lexfind.jsonl.xz": 16_981, + "data/en/legislation/uk_uk_lex.jsonl.xz": 36_499, + "data/fr/caselaw/france_cass.jsonl.xz": 113_844, + "data/fr/caselaw/luxembourg_judoc.jsonl.xz": 37_902, + "data/fr/caselaw/switzerland_entscheidsuche.jsonl.xz": 237_734, + "data/fr/legislation/belgium_ejustice.jsonl.xz": 10_613, + "data/fr/legislation/switzerland_lexfind.jsonl.xz": 10_680, + "data/hu/legislation/hungary_marcell.jsonl.xz": 26_821, + "data/it/caselaw/switzerland_entscheidsuche.jsonl.xz": 69_653, + "data/nl/legislation/belgium_ejustice.jsonl.xz": 10_556, + "data/pl/legislation/poland_marcell.jsonl.xz": 27_485, + "data/pt/caselaw/brazil_cjpg_0.jsonl.xz": 3_489_624, + "data/pt/caselaw/brazil_cjpg_1.jsonl.xz": 3_213_178, + "data/pt/caselaw/brazil_cjpg_2.jsonl.xz": 3_094_216, + "data/pt/caselaw/brazil_cjpg_3.jsonl.xz": 3_019_375, + "data/pt/caselaw/brazil_cjpg_4.jsonl.xz": 1_252_241, + "data/pt/caselaw/brazil_creta.jsonl.xz": 3_128_292, + "data/pt/caselaw/brazil_rulingbr.jsonl.xz": 10_623, + "data/ro/legislation/romania_marcell.jsonl.xz": 163_264, + "data/sk/legislation/slovakia_marcell.jsonl.xz": 13_055, + "data/sl/legislation/slovenia_marcell.jsonl.xz": 24_445, + }, + ), } @@ -199,7 +269,7 @@ def _validate_policy(key: tuple[str, str | None], policy: _SourceShardPolicy) -> biased samples or collapses rounds. """ repo_id, name = key - if policy.row_count_source not in {"constant", "parquet_footer"}: + if policy.row_count_source not in {"constant", "parquet_footer", "verified_table"}: raise ValueError( f"Policy {repo_id}/{name}: unknown row_count_source " f"{policy.row_count_source!r}" @@ -208,6 +278,33 @@ def _validate_policy(key: tuple[str, str | None], policy: _SourceShardPolicy) -> raise ValueError( f"Policy {repo_id}/{name}: min_headroom_rows must be > 0" ) + if policy.row_count_source == "verified_table": + # The table IS the shard allowlist: every listed shard must have + # a row count that leaves at least one valid offset after the + # headroom is reserved. Shards too small for the eval pipeline's + # per-round consumption must be left out of the table, not + # zero-bounded at pick time. + if not policy.verified_shard_rows: + raise ValueError( + f"Policy {repo_id}/{name}: row_count_source='verified_table' " + f"requires a non-empty verified_shard_rows table (it doubles " + f"as the shard allowlist)" + ) + for shard_path, rows in policy.verified_shard_rows.items(): + if rows <= policy.min_headroom_rows: + raise ValueError( + f"Policy {repo_id}/{name}: verified shard {shard_path!r} " + f"has {rows} rows ≤ min_headroom_rows " + f"({policy.min_headroom_rows}); no safe offset exists. " + f"Remove it from the table." + ) + if not shard_path.startswith(policy.path_prefix) or not shard_path.endswith( + policy.path_suffix + ): + raise ValueError( + f"Policy {repo_id}/{name}: table entry {shard_path!r} does " + f"not match path_prefix/path_suffix — typo in the table?" + ) if policy.row_count_source == "constant": if policy.safe_floor_rows is None or policy.safe_floor_rows <= 0: raise ValueError( @@ -296,7 +393,19 @@ def _list_shards(repo_id: str, name: str | None, revision: str) -> tuple[str, .. the same revision gets the same tuple. """ policy = _policy_for(repo_id, name) + if policy.row_count_source == "verified_table": + # The frozen table doubles as the shard allowlist. Listing from + # the HF API here would re-introduce the consensus hazard the + # table exists to remove (a re-uploaded repo changing the list + # under our feet); the pinned revision + table are the source + # of truth. + return tuple(sorted(policy.verified_shard_rows)) info = HfApi().dataset_info(repo_id, revision=revision) + name_filter = ( + re.compile(policy.leaf_name_pattern, re.IGNORECASE) + if policy.leaf_name_pattern + else _SHARD_NAME_FILTER + ) filtered = [] for f in info.siblings: rf = f.rfilename @@ -309,7 +418,7 @@ def _list_shards(repo_id: str, name: str | None, revision: str) -> tuple[str, .. # (e.g. metadata, sidecar files). The "train" / "part_" check # is per-source-format and intentionally narrow. leaf = rf.split("/")[-1] - if not _SHARD_NAME_FILTER.search(leaf): + if not name_filter.search(leaf): continue filtered.append(rf) if not filtered: @@ -366,6 +475,13 @@ def _resolve_offset_bound( assert policy.safe_floor_rows is not None return policy.safe_floor_rows + if policy.row_count_source == "verified_table": + # Module-load validation guarantees the shard is in the table + # with rows > min_headroom_rows (the table is the allowlist + # `_list_shards` picks from). + rows = policy.verified_shard_rows[shard_path] + return rows - policy.min_headroom_rows + if policy.row_count_source == "parquet_footer": actual_rows = _shard_rows_via_parquet_footer(repo_id, revision, shard_path) bound = actual_rows - policy.min_headroom_rows @@ -406,6 +522,9 @@ class ShardPick: offset_bound: int shard_rows: int # for the constant path this equals offset_bound (we don't know the true count) in_shard_offset: int + # Propagated from the policy: when set, `load_streaming_shard` loads + # via this generic builder against a resolved URL (script-bypass). + load_builder: str | None = None def pick_shard_for_source( @@ -456,8 +575,8 @@ def pick_shard_for_source( # For constant policies we don't know the actual shard size; # surface `offset_bound` as `shard_rows` so older callers (notebook # / tests) that check "offset < shard_rows" still see the right - # invariant. The parquet path can fill in the real count. - if policy.row_count_source == "parquet_footer": + # invariant. The parquet and verified-table paths know the real count. + if policy.row_count_source in {"parquet_footer", "verified_table"}: actual_shard_rows = offset_bound + policy.min_headroom_rows else: actual_shard_rows = offset_bound @@ -470,6 +589,7 @@ def pick_shard_for_source( offset_bound=offset_bound, shard_rows=actual_shard_rows, in_shard_offset=offset, + load_builder=policy.load_builder, ) @@ -495,14 +615,30 @@ def load_streaming_shard( # helps unit tests. from datasets import load_dataset - load_kwargs: dict[str, Any] = { - "data_files": [pick.shard_path], - "streaming": True, - "revision": pick.revision, - } - if extra_load_kwargs: - load_kwargs.update(extra_load_kwargs) - ds = load_dataset(pick.repo_id, **load_kwargs) + if pick.load_builder: + # Script-bypass path: the repo ships a custom loading script, so + # `load_dataset(repo_id, ...)` would execute it (and for + # Multi_Legal_Pile, stream entirely different files from external + # repos). Loading the raw file through a generic builder reads + # exactly the picked shard and needs no `trust_remote_code`. + from huggingface_hub import hf_hub_url + + url = hf_hub_url( + pick.repo_id, pick.shard_path, repo_type="dataset", revision=pick.revision + ) + load_kwargs = {"data_files": [url], "streaming": True} + if extra_load_kwargs: + load_kwargs.update(extra_load_kwargs) + ds = load_dataset(pick.load_builder, **load_kwargs) + else: + load_kwargs = { + "data_files": [pick.shard_path], + "streaming": True, + "revision": pick.revision, + } + if extra_load_kwargs: + load_kwargs.update(extra_load_kwargs) + ds = load_dataset(pick.repo_id, **load_kwargs) if split_name in ds: return ds[split_name] # `data_files=` with a single file lands the rows under "train" by diff --git a/connito/shared/evaluate.py b/connito/shared/evaluate.py index 196f223..2d0c134 100644 --- a/connito/shared/evaluate.py +++ b/connito/shared/evaluate.py @@ -117,7 +117,12 @@ def evaluate_model( if max_eval_batches is not None and batch_step >= max_eval_batches: break - logger.debug( + # INFO, not debug: `nan_batches` is the only production-visible + # signal distinguishing "miner scored on all batches" from + # "miner's average silently excludes NaN batches" (degenerate + # eval rows and overflow-on-OOD submissions both land here). + # Ops must be able to read this off a live validator's logs. + logger.info( "eval loss", loss_sum=round(loss_sum, 4), aux_loss_sum=round(aux_loss_sum, 4), diff --git a/connito/test/test_eval_data_quality.py b/connito/test/test_eval_data_quality.py new file mode 100644 index 0000000..94575d8 --- /dev/null +++ b/connito/test/test_eval_data_quality.py @@ -0,0 +1,211 @@ +"""Unit tests for the eval-data-quality gates added for exp_legal. + +Covers the pure-function layer only — no HF network access: + + - `_min_text_chars_filter`: drops empty / trivially-short rows + (Multi_Legal_Pile all_all streams 38% empty-text rows; measured + 2026-07-21 on the first 800 rows). + - `_PrefixDedupFilter`: keeps only the first row per distinct text + prefix (75% of non-empty M_L_P rows share an identical 200-char + prefix with another row), and does so deterministically without + relying on per-process `hash()`. + - `tokenize_windowed`: content-hash window sampling for documents + longer than `sequence_length`; prefix + padding for short ones. + - `_SourceShardPolicy` `row_count_source="verified_table"`: table + doubles as allowlist, validation rejects under-headroom shards and + prefix typos, offset bound honors per-shard counts. + +Integration behavior (streaming filter placement, interleave order, +seeded shard-pick equivalence for Multi_Legal_Pile) is exercised by the +notebook-driven checks described in docs/exp-legal-migration-plan.md §2 +and the P1 determinism test in test_eval_source_skip.py. +""" +from __future__ import annotations + +import pytest + +from connito.shared.dataloader import ( + _min_text_chars_filter, + _PrefixDedupFilter, + tokenize_windowed, +) +from connito.shared.eval_shard_pick import ( + _SourceShardPolicy, + _resolve_offset_bound, + _validate_policy, +) + + +# --------------------------------------------------------------------------- +# _min_text_chars_filter +# --------------------------------------------------------------------------- + +def test_min_chars_drops_empty_and_whitespace_rows(): + assert not _min_text_chars_filter({"text": ""}, min_chars=200) + assert not _min_text_chars_filter({"text": " \n\t "}, min_chars=200) + assert not _min_text_chars_filter({}, min_chars=200) + + +def test_min_chars_keeps_substantial_rows(): + assert _min_text_chars_filter({"text": "x" * 200}, min_chars=200) + assert _min_text_chars_filter({"text": " " + "x" * 200 + " "}, min_chars=200) + + +def test_min_chars_boundary_is_inclusive(): + assert _min_text_chars_filter({"text": "x" * 200}, min_chars=200) + assert not _min_text_chars_filter({"text": "x" * 199}, min_chars=200) + + +# --------------------------------------------------------------------------- +# _PrefixDedupFilter +# --------------------------------------------------------------------------- + +def test_prefix_dedup_keeps_first_occurrence_only(): + f = _PrefixDedupFilter(prefix_chars=10) + boiler = "IN THE SUPREME COURT OF ..." + assert f({"text": boiler + " case one"}) + assert not f({"text": boiler + " case two"}) + assert f({"text": "completely different text"}) + + +def test_prefix_dedup_distinguishes_beyond_prefix_window(): + f = _PrefixDedupFilter(prefix_chars=100) + a = "shared start " + "a" * 200 + b = "shared start " + "b" * 200 + # Prefix window (100 chars) reaches into the differing region. + assert f({"text": a}) + assert f({"text": b}) + + +def test_prefix_dedup_state_is_per_instance(): + text = {"text": "same row"} + assert _PrefixDedupFilter(50)(text) + assert _PrefixDedupFilter(50)(text) # fresh instance, fresh set + + +# --------------------------------------------------------------------------- +# tokenize_windowed +# --------------------------------------------------------------------------- + + +class _StubTokenizer: + """1-token-per-word tokenizer; ids are stable per word.""" + + pad_token_id = 0 + eos_token_id = 2 + + def __call__(self, text, truncation=False, add_special_tokens=True, **_kw): + ids = [7 + (len(w) % 50) for w in text.split()] + return {"input_ids": ids} + + +def test_windowed_short_doc_pads_to_length(): + tok = _StubTokenizer() + out = tokenize_windowed("three word doc", tok, sequence_length=8) + assert len(out["input_ids"]) == 8 + assert out["attention_mask"] == [1, 1, 1, 0, 0, 0, 0, 0] + assert out["input_ids"][3:] == [tok.pad_token_id] * 5 + + +def test_windowed_long_doc_returns_full_window(): + tok = _StubTokenizer() + text = " ".join(f"w{i}" for i in range(500)) + out = tokenize_windowed(text, tok, sequence_length=64) + assert len(out["input_ids"]) == 64 + assert out["attention_mask"] == [1] * 64 + + +def test_windowed_is_deterministic_per_content(): + tok = _StubTokenizer() + text = " ".join(f"w{i}" for i in range(500)) + a = tokenize_windowed(text, tok, sequence_length=64) + b = tokenize_windowed(text, tok, sequence_length=64) + assert a == b + + +def test_windowed_start_varies_across_documents(): + class PositionTokenizer(_StubTokenizer): + # ids encode stream position, so the returned window is + # `range(start, start+seq)` and directly reveals the chosen start. + def __call__(self, text, truncation=False, add_special_tokens=True, **_kw): + return {"input_ids": list(range(len(text.split())))} + + tok = PositionTokenizer() + full = [" ".join(f"d{k}w{i}" for i in range(500)) for k in range(20)] + starts = {tokenize_windowed(t, tok, sequence_length=64)["input_ids"][0] for t in full} + # Content-hash-derived starts over a 437-position span: 20 documents + # should land on many distinct starts (expected ≈19.6 distinct). + # A constant-start implementation collapses this to 1. + assert len(starts) >= 10 + + +def test_windowed_missing_pad_token_falls_back_to_eos(): + class NoPad(_StubTokenizer): + pad_token_id = None + + out = tokenize_windowed("one two", NoPad(), sequence_length=4) + assert out["input_ids"][2:] == [NoPad.eos_token_id] * 2 + + +# --------------------------------------------------------------------------- +# verified_table shard policy +# --------------------------------------------------------------------------- + +_KEY = ("example/repo", "subset") + + +def _table_policy(**overrides): + kwargs = dict( + path_prefix="data/", + path_suffix=(".jsonl.xz",), + revision="deadbeef", + row_count_source="verified_table", + min_headroom_rows=5_000, + verified_shard_rows={ + "data/a/one.jsonl.xz": 50_000, + "data/b/two.jsonl.xz": 12_000, + }, + ) + kwargs.update(overrides) + return _SourceShardPolicy(**kwargs) + + +def test_verified_table_policy_validates(): + _validate_policy(_KEY, _table_policy()) # should not raise + + +def test_verified_table_rejects_empty_table(): + with pytest.raises(ValueError, match="non-empty verified_shard_rows"): + _validate_policy(_KEY, _table_policy(verified_shard_rows={})) + + +def test_verified_table_rejects_under_headroom_shard(): + with pytest.raises(ValueError, match="no safe offset"): + _validate_policy( + _KEY, + _table_policy( + verified_shard_rows={"data/a/one.jsonl.xz": 4_000}, + ), + ) + + +def test_verified_table_rejects_prefix_typo(): + with pytest.raises(ValueError, match="path_prefix/path_suffix"): + _validate_policy( + _KEY, + _table_policy( + verified_shard_rows={"wrong/one.jsonl.xz": 50_000}, + ), + ) + + +def test_verified_table_offset_bound_is_per_shard(): + policy = _table_policy() + bound = _resolve_offset_bound( + repo_id=_KEY[0], + name=_KEY[1], + revision="deadbeef", + shard_path="data/b/two.jsonl.xz", + policy=policy, + ) + assert bound == 12_000 - 5_000 diff --git a/expert_groups/exp_legal/config.yaml b/expert_groups/exp_legal/config.yaml index d1307ae..9c7265c 100644 --- a/expert_groups/exp_legal/config.yaml +++ b/expert_groups/exp_legal/config.yaml @@ -43,11 +43,13 @@ data: world_size: 10 rank: 1 dataset_class: "expert_groups.exp_legal.dataset:StreamingTorchDataset" - # TODO(exp-legal-launch): register (joelniklaus/Multi_Legal_Pile, all_all) - # in connito/shared/eval_shard_pick.py:_KNOWN_SOURCES with verified - # shard rows for all 33 files, then flip this back to true. Until then, - # the validator's legacy head-of-stream eval path is used for this - # group; consensus determinism across validators is weaker than the - # seeded-pick guarantee exp_math already enjoys. See - # docs/exp-legal-migration-plan.md §2. - eval_source_seeded_shard_pick: false + # (joelniklaus/Multi_Legal_Pile, all_all) is registered in + # connito/shared/eval_shard_pick.py:_KNOWN_SOURCES with a + # verified-row-count table over the repo's native data/ shards + # (row_count_source="verified_table", pinned revision, script-bypass + # json loading — the launch-period legacy-path shortcut from + # docs/exp-legal-migration-plan.md §2 is retired). Note the eval + # pool is the native Multi_Legal_Pile files; the miner TRAINING + # stream still goes through the all_all builder script, so eval + # remains a subset of the training distribution. + eval_source_seeded_shard_pick: true diff --git a/expert_groups/exp_legal/dataset.py b/expert_groups/exp_legal/dataset.py index 9739af3..2bba265 100644 --- a/expert_groups/exp_legal/dataset.py +++ b/expert_groups/exp_legal/dataset.py @@ -2,7 +2,7 @@ from transformers import PreTrainedTokenizerBase -from connito.shared.dataloader import DefaultStreamingTorchDataset +from connito.shared.dataloader import DefaultStreamingTorchDataset, tokenize_windowed # ------------------------------------------------------------- @@ -20,19 +20,16 @@ def tokenize_and_format( Processes raw text for Continuous Pre-Training (CPT). Bypasses the chat template since Multi_Legal_Pile and C4 are raw text corpora, not conversational data. + + Long documents get a content-hash-derived window instead of the + prefix: legal filings front-load their most templated text + (headers, procedural boilerplate), so always scoring the first + `sequence_length` tokens let template-memorizing miners reach + near-zero eval loss. See `tokenize_windowed` for the mechanism + and its consensus properties. """ - # 1) Safely extract the raw text we aligned in dataloader.py text = str(example.get("text", "")) - - # 2) Tokenize text directly (no chat template) - toks = tokenizer( - text, - truncation=True, - max_length=sequence_length, - padding="max_length", - add_special_tokens=True, # Important: Ensures BOS/EOS tokens are added - ) - + toks = tokenize_windowed(text, tokenizer, sequence_length) return { "input_ids": toks["input_ids"], "attention_mask": toks["attention_mask"],