Buckets:
| """Conservative, frozen automatic pre-screen for controlled Gold candidates. | |
| This is a cheap structural and lexical audit, NOT semantic certification. No | |
| record produced here is called Gold; only blinded human ratings can do that. | |
| The thresholds below are frozen before any human certification is ingested. | |
| """ | |
| import argparse | |
| import collections | |
| import difflib | |
| import hashlib | |
| import json | |
| import re | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[2] | |
| OUT = ROOT / "outputs/gold_intervention" | |
| CRITERIA = ("correctness", "coherence", "complexity", "verbosity") | |
| VERSION = "automatic_prescreen_v1_20260923" | |
| THRESHOLDS = { | |
| "minimum_words": 18, | |
| "correctness_max_changed_spans": 1, | |
| "correctness_max_changed_words": 3, | |
| "verbosity_min_token_change_fraction": 0.04, | |
| "verbosity_min_content_overlap": 0.75, | |
| "verbosity_min_char_similarity": 0.57, | |
| "complexity_max_token_change_fraction": 0.10, | |
| "complexity_min_content_overlap": 0.48, | |
| "complexity_min_char_similarity": 0.52, | |
| } | |
| WORD = re.compile(r"\b[\w]+(?:[-'][\w]+)*\b", re.UNICODE) | |
| SENTENCE = re.compile(r"(?<=[.!?])(?:[ \t\r\n]+)") | |
| STOP = frozenset("the a an and or but if then of to in on at by for with from as is are was were be been being it this that these those i you he she they we my your our their its not can could may might should would will do does did have has had so also very more less much many some all any one there here when where why how what which who because therefore thus however".split()) | |
| NUMBER = re.compile(r"(?<!\w)[-+]?\d+(?:[.,]\d+)?%?(?!\w)") | |
| NEGATION = re.compile(r"\b(?:not|never|no|none|without|cannot|can't|won't|isn't|aren't|doesn't|don't)\b", re.I) | |
| FORMAT = re.compile(r"```|<\/?(?:table|code|script|style)\b|<extra_id_|\\boxed\{") | |
| MECHANISM = re.compile(r"\b(?:because|thereby|therefore|hence|consequently|caus(?:e|es|ing)|leads? to|results? in|in turn|by (?:reducing|increasing|changing|making|allowing)|so that|in order to|due to|if .{1,60} then)\b", re.I) | |
| MODEL_SYSTEM = """You are an independent semantic auditor. Compare response B with response A for the given user prompt. Judge four changes independently, including off-target changes. Return exactly one JSON object with keys correctness, coherence, complexity, verbosity, confidence. The four scores must be integers from -2 to 2: -2 substantially decreased, -1 decreased, 0 approximately unchanged, 1 increased, 2 substantially increased. Confidence must be low, medium, or high. Correctness means factual accuracy and following the prompt; coherence means logical organization and consistency; complexity means conceptual or reasoning sophistication; verbosity means excess wording relative to information. If uncertain, choose 0 and low confidence. Do not guess why the text was edited. No prose or markdown.""" | |
| PROPOSITION_SYSTEM = """You are a blind semantic-content auditor. Given a prompt and two responses, list the substantive propositions (factual claims, conclusions, distinct reasoning steps, and distinct concepts) stated by each response. Ignore repeated wording, style, and filler. Then identify substantive propositions present in A but missing from B and newly present in B, allowing paraphrases to count as the same proposition. Independently judge whether the substantive proposition sets are approximately equivalent and whether the reasoning steps/structure are approximately the same. Do not infer or mention an intended edit type. Return exactly one JSON object with keys A_propositions, B_propositions, missing_from_B, added_in_B, equivalent_substantive_content, same_reasoning_structure, confidence. The first four keys are arrays of short strings, the next two are booleans, confidence is low, medium, or high. Be conservative if either response adds information. No markdown or prose outside JSON.""" | |
| def words(text): | |
| return [m.group().lower() for m in WORD.finditer(text)] | |
| def content_words(text): | |
| return collections.Counter(w for w in words(text) if w not in STOP and len(w) > 2) | |
| def overlap(a, b): | |
| ca, cb = content_words(a), content_words(b) | |
| denominator = min(sum(ca.values()), sum(cb.values())) | |
| return sum((ca & cb).values()) / denominator if denominator else 0.0 | |
| def sentences(text): | |
| return [part.strip() for part in SENTENCE.split(text) if part.strip()] | |
| def changes(a, b): | |
| return [(tag, a[i:j], b[k:l]) for tag, i, j, k, l in | |
| difflib.SequenceMatcher(a=a, b=b, autojunk=False).get_opcodes() if tag != "equal"] | |
| def op_name(record): | |
| operation = record.get("operation", {}) | |
| return operation.get("operation", "") if isinstance(operation, dict) else str(operation) | |
| def audit(record): | |
| a, b = record["clean_response"], record["altered_response"] | |
| criterion = record["requested_criterion"] | |
| wa, wb = words(a), words(b) | |
| changed = changes(a, b) | |
| token_changes = [(tag, wa[i:j], wb[k:l]) for tag, i, j, k, l in | |
| difflib.SequenceMatcher(a=wa, b=wb, autojunk=False).get_opcodes() if tag != "equal"] | |
| ratio = abs(len(wa) - len(wb)) / max(1, len(wa)) | |
| structural = { | |
| "word_count_clean": len(wa), "word_count_altered": len(wb), | |
| "word_count_change_fraction": round(ratio, 5), | |
| "sentence_count_clean": len(sentences(a)), "sentence_count_altered": len(sentences(b)), | |
| "character_similarity": round(difflib.SequenceMatcher(a=a, b=b, autojunk=False).ratio(), 5), | |
| "content_word_overlap": round(overlap(a, b), 5), | |
| "changed_spans": len(changed), | |
| "changed_token_spans": len(token_changes), | |
| "changed_words": sum(max(len(old), len(new)) for _, old, new in token_changes), | |
| "numbers_clean": NUMBER.findall(a), "numbers_altered": NUMBER.findall(b), | |
| "negations_clean": NEGATION.findall(a), "negations_altered": NEGATION.findall(b), | |
| "exact_sentence_multiset_preserved": collections.Counter(sentences(a)) == collections.Counter(sentences(b)), | |
| } | |
| reasons = [] | |
| warnings = [] | |
| if criterion not in CRITERIA: | |
| reasons.append("unknown_criterion") | |
| if not a.strip() or not b.strip() or a == b: | |
| reasons.append("empty_or_identical") | |
| if min(len(wa), len(wb)) < THRESHOLDS["minimum_words"]: | |
| reasons.append("too_short") | |
| if FORMAT.search(a) or FORMAT.search(b): | |
| reasons.append("format_artifact") | |
| if criterion == "coherence": | |
| if not structural["exact_sentence_multiset_preserved"]: | |
| reasons.append("sentence_multiset_changed") | |
| if len(sentences(a)) < 2 or len(sentences(a)) != len(sentences(b)): | |
| reasons.append("invalid_sentence_permutation") | |
| if "swap" in op_name(record): | |
| before, after = record.get("operation", {}).get("sentence_ids_before"), record.get("operation", {}).get("sentence_ids_after") | |
| if not before or not after or before != list(range(len(sentences(a)))) or sorted(before) != sorted(after) or before == after: | |
| reasons.append("invalid_swap_metadata") | |
| elif [sentences(a)[index] for index in after] != sentences(b): | |
| reasons.append("sentence_permutation_does_not_match_metadata") | |
| else: | |
| warnings.append("non_swap_coherence_operation_requires_manual_inspection") | |
| elif criterion == "correctness": | |
| if len(token_changes) > THRESHOLDS["correctness_max_changed_spans"]: | |
| reasons.append("broad_correctness_rewrite") | |
| if structural["changed_words"] > THRESHOLDS["correctness_max_changed_words"]: | |
| reasons.append("many_changed_words") | |
| if len(sentences(a)) != len(sentences(b)): | |
| reasons.append("sentence_count_changed") | |
| if len(structural["numbers_clean"]) != len(structural["numbers_altered"]): | |
| warnings.append("number_claim_count_changed") | |
| warnings.append("factual_truth_unverified") | |
| elif criterion == "verbosity": | |
| if ratio < THRESHOLDS["verbosity_min_token_change_fraction"]: | |
| reasons.append("verbosity_change_too_small") | |
| if structural["content_word_overlap"] < THRESHOLDS["verbosity_min_content_overlap"]: | |
| reasons.append("possible_proposition_change_content_overlap") | |
| if structural["character_similarity"] < THRESHOLDS["verbosity_min_char_similarity"]: | |
| reasons.append("broad_verbosity_rewrite") | |
| if structural["numbers_clean"] != structural["numbers_altered"]: | |
| reasons.append("numeric_propositions_changed") | |
| if sorted(x.lower() for x in structural["negations_clean"]) != sorted(x.lower() for x in structural["negations_altered"]): | |
| reasons.append("negation_changed") | |
| warnings.append("proposition_equivalence_requires_human_review") | |
| elif criterion == "complexity": | |
| if ratio > THRESHOLDS["complexity_max_token_change_fraction"]: | |
| reasons.append("complexity_length_over_10_percent") | |
| if structural["content_word_overlap"] < THRESHOLDS["complexity_min_content_overlap"]: | |
| reasons.append("possible_conclusion_or_topic_change") | |
| if structural["character_similarity"] < THRESHOLDS["complexity_min_char_similarity"]: | |
| reasons.append("broad_complexity_rewrite") | |
| if structural["numbers_clean"] != structural["numbers_altered"]: | |
| reasons.append("numeric_conclusion_changed") | |
| if sorted(x.lower() for x in structural["negations_clean"]) != sorted(x.lower() for x in structural["negations_altered"]): | |
| reasons.append("negation_changed") | |
| operation = record.get("operation", {}) | |
| if isinstance(operation, dict) and operation.get("new_span"): | |
| old_span, new_span = operation.get("old_span", ""), operation["new_span"] | |
| structural["changed_span_content_overlap"] = round(overlap(old_span, new_span), 5) | |
| structural["new_span_has_explicit_mechanism"] = bool(MECHANISM.search(new_span)) | |
| if not structural["new_span_has_explicit_mechanism"]: | |
| warnings.append("complexity_may_be_formality_without_reasoning") | |
| if structural["changed_span_content_overlap"] < 0.25: | |
| warnings.append("complexity_changed_span_topic_drift_risk") | |
| warnings.append("conceptual_complexity_requires_human_review") | |
| return structural, sorted(set(reasons)), sorted(set(warnings)) | |
| def parse_model_judgment(raw): | |
| text = raw.strip() | |
| if text.startswith("```"): | |
| text = text.split("\n", 1)[-1].rsplit("```", 1)[0] | |
| start, end = text.find("{"), text.rfind("}") | |
| if start < 0 or end < start: | |
| raise ValueError("No JSON object") | |
| value = json.loads(text[start:end + 1]) | |
| for criterion in CRITERIA: | |
| score = value.get(criterion) | |
| if isinstance(score, bool) or not isinstance(score, int) or score not in (-2, -1, 0, 1, 2): | |
| raise ValueError(f"Invalid {criterion}: {score!r}") | |
| if value.get("confidence") not in ("low", "medium", "high"): | |
| raise ValueError("Invalid confidence") | |
| return {key: value[key] for key in (*CRITERIA, "confidence")} | |
| def run_model_audit(model_path, device, batch_size, max_new_tokens, model_id, slug, include_generator=None, exclude_generator=None): | |
| """Blind, restartable different-family semantic audit; never human labels.""" | |
| import torch | |
| from transformers import AutoModelForCausalLM, AutoTokenizer | |
| torch.backends.cuda.enable_cudnn_sdp(False) | |
| source = OUT / "automatic_judge_pairs.jsonl" | |
| rows = [json.loads(line) for line in source.open(encoding="utf-8")] | |
| family_by_id = {row["family_id"]: row for line in (OUT / "automatic_passed.jsonl").open(encoding="utf-8") | |
| if (row := json.loads(line))} | |
| if include_generator: | |
| rows = [row for row in rows if include_generator in family_by_id[row["edit_id"]].get("candidate_generator", "").lower()] | |
| if exclude_generator: | |
| rows = [row for row in rows if exclude_generator not in family_by_id[row["edit_id"]].get("candidate_generator", "").lower()] | |
| source_hash = hashlib.sha256(source.read_bytes()).hexdigest() | |
| output = OUT / "automatic_model_judgments" / slug | |
| output.mkdir(parents=True, exist_ok=True) | |
| prompt_hash = hashlib.sha256(MODEL_SYSTEM.encode()).hexdigest() | |
| def pair_hash(row): | |
| return hashlib.sha256(json.dumps({key: row[key] for key in ("prompt", "original_response", "edited_response")}, | |
| ensure_ascii=False, sort_keys=True).encode()).hexdigest() | |
| def cached(row): | |
| path = output / f"{row['edit_id']}.json" | |
| if not path.exists(): | |
| return False | |
| saved = json.loads(path.read_text()) | |
| return saved.get("pair_sha256") == pair_hash(row) and saved.get("system_prompt_sha256") == prompt_hash | |
| pending = [row for row in rows if not cached(row)] | |
| print(json.dumps({"model": model_id, "pending": len(pending), "existing": len(rows) - len(pending)}), flush=True) | |
| if not pending: | |
| summarize_model_audit(rows, output, source_hash, model_id, slug) | |
| return | |
| tokenizer = AutoTokenizer.from_pretrained(model_path, local_files_only=True, | |
| fix_mistral_regex="mistral" in model_id.lower()) | |
| tokenizer.padding_side = "left" | |
| if tokenizer.pad_token_id is None: | |
| tokenizer.pad_token = tokenizer.eos_token | |
| def messages(row): | |
| return [{"role": "system", "content": MODEL_SYSTEM}, | |
| {"role": "user", "content": "PROMPT:\n" + row["prompt"] + "\n\nRESPONSE A:\n" + row["original_response"] + "\n\nRESPONSE B:\n" + row["edited_response"]}] | |
| pending_text = [(tokenizer.apply_chat_template(messages(row), tokenize=False, add_generation_prompt=True), row) | |
| for row in pending] | |
| pending_text.sort(key=lambda item: len(item[0])) | |
| model = AutoModelForCausalLM.from_pretrained(model_path, local_files_only=True, dtype=torch.bfloat16, | |
| attn_implementation="sdpa").to(device).eval() | |
| for offset in range(0, len(pending_text), batch_size): | |
| batch = pending_text[offset:offset + batch_size] | |
| inputs = tokenizer([text for text, _ in batch], return_tensors="pt", padding=True, truncation=False).to(device) | |
| with torch.inference_mode(): | |
| generated = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False, | |
| pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id) | |
| continuation = generated[:, inputs.input_ids.shape[1]:] | |
| for (_, row), tokens in zip(batch, continuation): | |
| raw = tokenizer.decode(tokens, skip_special_tokens=True) | |
| result = {"edit_id": row["edit_id"], "model_id": model_id, "system_prompt_sha256": prompt_hash, | |
| "source_sha256": source_hash, "pair_sha256": pair_hash(row), "raw": raw} | |
| try: | |
| result["judgment"] = parse_model_judgment(raw) | |
| except (ValueError, json.JSONDecodeError) as exc: | |
| result["parse_error"] = str(exc) | |
| (output / f"{row['edit_id']}.json").write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n") | |
| print(json.dumps({"completed": min(offset + len(batch), len(pending_text)), "total": len(pending_text)}), flush=True) | |
| summarize_model_audit(rows, output, source_hash, model_id, slug) | |
| def summarize_model_audit(rows, output, source_hash, model_id, slug): | |
| family_by_id = {row["family_id"]: row for line in (OUT / "automatic_passed.jsonl").open(encoding="utf-8") | |
| if (row := json.loads(line))} | |
| scored = [] | |
| for pair in rows: | |
| path = output / f"{pair['edit_id']}.json" | |
| if not path.exists(): | |
| continue | |
| judgment_record = json.loads(path.read_text()) | |
| family = family_by_id[pair["edit_id"]] | |
| judgment = judgment_record.get("judgment") | |
| if judgment is None: | |
| scored.append({"family_id": pair["edit_id"], "gold_split": family["gold_split"], | |
| "criterion": family["requested_criterion"], "parse_error": judgment_record.get("parse_error")}) | |
| continue | |
| criterion = family["requested_criterion"] | |
| direction = -1 if criterion in ("correctness", "coherence") else 1 | |
| target = judgment[criterion] | |
| off = {key: judgment[key] for key in CRITERIA if key != criterion} | |
| scored.append({"family_id": pair["edit_id"], "gold_split": family["gold_split"], "model_id": model_id, | |
| "criterion": criterion, "expected_clean_to_altered_direction": direction, | |
| "judgment": judgment, "model_target_fidelity": target * direction >= 1, | |
| "model_one_factor": target * direction >= 1 and all(value == 0 for value in off.values()), | |
| "model_off_target_magnitude": sum(abs(value) for value in off.values())}) | |
| screen_path = OUT / ("automatic_model_screen.jsonl" if slug == "phi-4" else f"automatic_model_screen_{slug}.jsonl") | |
| with screen_path.open("w") as handle: | |
| for record in scored: | |
| handle.write(json.dumps(record, ensure_ascii=False) + "\n") | |
| counts = collections.Counter((r["gold_split"], r["criterion"]) for r in scored if "judgment" in r) | |
| faithful = collections.Counter((r["gold_split"], r["criterion"]) for r in scored if r.get("model_target_fidelity")) | |
| one_factor = collections.Counter((r["gold_split"], r["criterion"]) for r in scored if r.get("model_one_factor")) | |
| revision = ("2db69c1c3e91a05d2c64a3185acfbaf36f744e25" if slug == "phi-4" | |
| else "9527884be6e5616bdd54de542f9ae13384489724") | |
| summary = {"source_sha256": source_hash, "model_id": model_id, "model_revision": revision, | |
| "output_directory": str(output.relative_to(ROOT)), "n_pairs": len(rows), "n_valid": sum("judgment" in r for r in scored), | |
| "n_parse_error": sum("parse_error" in r for r in scored), | |
| "by_cell": [{"split": split, "criterion": criterion, "valid": counts[split, criterion], | |
| "model_target_fidelity": faithful[split, criterion], "model_one_factor": one_factor[split, criterion]} | |
| for split in ("train", "test") for criterion in CRITERIA], | |
| "interpretation": "Model-only screening diagnostic. No human-certified Gold labels."} | |
| summary_path = OUT / ("automatic_model_summary.json" if slug == "phi-4" else f"automatic_model_summary_{slug}.json") | |
| summary_path.write_text(json.dumps(summary, indent=2) + "\n") | |
| print(json.dumps({"n_valid": summary["n_valid"], "n_parse_error": summary["n_parse_error"], "by_cell": summary["by_cell"]}), flush=True) | |
| combine_independent_model_screens() | |
| def combine_independent_model_screens(): | |
| """Merge non-self model ratings for later comparison with human labels.""" | |
| paths = [OUT / "automatic_model_screen.jsonl", OUT / "automatic_model_screen_mistral-small-24b.jsonl"] | |
| if not all(path.exists() for path in paths): | |
| return | |
| rows = [json.loads(line) for path in paths for line in path.open(encoding="utf-8")] | |
| ids = [row["family_id"] for row in rows] | |
| if len(ids) != len(set(ids)): | |
| raise ValueError("Overlapping model-audit cohorts; possible self-judgment") | |
| candidate_ids = {row["family_id"] for line in (OUT / "automatic_passed.jsonl").open(encoding="utf-8") | |
| if (row := json.loads(line))} | |
| if not set(ids).issubset(candidate_ids): | |
| raise ValueError("Model-audit row absent from candidate packet") | |
| rows.sort(key=lambda row: row["family_id"]) | |
| final_path = OUT / "automatic_independent_model_screen.jsonl" | |
| temp_path = OUT / "automatic_independent_model_screen.jsonl.tmp" | |
| with temp_path.open("w") as handle: | |
| for row in rows: | |
| handle.write(json.dumps(row, ensure_ascii=False) + "\n") | |
| temp_path.replace(final_path) | |
| summary_path = OUT / "automatic_independent_model_summary.json" | |
| summary_temp = OUT / "automatic_independent_model_summary.json.tmp" | |
| summary_temp.write_text(json.dumps({ | |
| "n_candidates": len(candidate_ids), "n_independently_audited": len(rows), | |
| "n_valid": sum("judgment" in row for row in rows), | |
| "n_missing": len(candidate_ids - set(ids)), | |
| "interpretation": "Different-family model ratings, not human labels or Gold certification." | |
| }, indent=2) + "\n") | |
| summary_temp.replace(summary_path) | |
| def run_phi_propositions(model_path, device, batch_size, max_new_tokens): | |
| """Independent model-only proposition diagnostic for verbosity candidates.""" | |
| import torch | |
| from transformers import AutoModelForCausalLM, AutoTokenizer | |
| torch.backends.cuda.enable_cudnn_sdp(False) | |
| source = OUT / "automatic_passed.jsonl" | |
| families = [json.loads(line) for line in source.open(encoding="utf-8")] | |
| families = [row for row in families if row["requested_criterion"] == "verbosity" and | |
| "phi" not in row.get("candidate_generator", "").lower()] | |
| output = OUT / "automatic_proposition_judgments" / "phi-4" | |
| output.mkdir(parents=True, exist_ok=True) | |
| instruction_hash = hashlib.sha256(PROPOSITION_SYSTEM.encode()).hexdigest() | |
| def pair_hash(row): | |
| return hashlib.sha256(json.dumps({key: row[key] for key in ("prompt", "clean_response", "altered_response")}, | |
| ensure_ascii=False, sort_keys=True).encode()).hexdigest() | |
| def cached(row): | |
| path = output / f"{row['family_id']}.json" | |
| if not path.exists(): | |
| return False | |
| saved = json.loads(path.read_text()) | |
| return (saved.get("pair_sha256") == pair_hash(row) and saved.get("system_prompt_sha256") == instruction_hash | |
| and "judgment" in saved) | |
| pending = [row for row in families if not cached(row)] | |
| print(json.dumps({"model": "Phi-4 proposition audit", "pending": len(pending), "existing": len(families) - len(pending)}), flush=True) | |
| if pending: | |
| tokenizer = AutoTokenizer.from_pretrained(model_path, local_files_only=True) | |
| tokenizer.padding_side = "left" | |
| if tokenizer.pad_token_id is None: | |
| tokenizer.pad_token = tokenizer.eos_token | |
| def messages(row): | |
| return [{"role": "system", "content": PROPOSITION_SYSTEM}, | |
| {"role": "user", "content": "PROMPT:\n" + row["prompt"] + "\n\nRESPONSE A:\n" + row["clean_response"] + | |
| "\n\nRESPONSE B:\n" + row["altered_response"]}] | |
| pending_text = [(tokenizer.apply_chat_template(messages(row), tokenize=False, add_generation_prompt=True), row) | |
| for row in pending] | |
| pending_text.sort(key=lambda item: len(item[0])) | |
| model = AutoModelForCausalLM.from_pretrained(model_path, local_files_only=True, dtype=torch.bfloat16, | |
| attn_implementation="sdpa").to(device).eval() | |
| for offset in range(0, len(pending_text), batch_size): | |
| batch = pending_text[offset:offset + batch_size] | |
| inputs = tokenizer([text for text, _ in batch], return_tensors="pt", padding=True, truncation=False).to(device) | |
| with torch.inference_mode(): | |
| generated = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False, | |
| pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id) | |
| continuation = generated[:, inputs.input_ids.shape[1]:] | |
| for (_, row), tokens in zip(batch, continuation): | |
| raw = tokenizer.decode(tokens, skip_special_tokens=True) | |
| result = {"family_id": row["family_id"], "model_id": "microsoft/phi-4", | |
| "model_revision": "2db69c1c3e91a05d2c64a3185acfbaf36f744e25", "system_prompt_sha256": instruction_hash, | |
| "pair_sha256": pair_hash(row), "raw": raw} | |
| try: | |
| obj = json.loads(raw[raw.find("{"):raw.rfind("}") + 1]) | |
| for key in ("A_propositions", "B_propositions", "missing_from_B", "added_in_B"): | |
| if not isinstance(obj.get(key), list) or not all(isinstance(item, str) for item in obj[key]): | |
| raise ValueError(f"Invalid {key}") | |
| for key in ("equivalent_substantive_content", "same_reasoning_structure"): | |
| if not isinstance(obj.get(key), bool): | |
| raise ValueError(f"Invalid {key}") | |
| if obj.get("confidence") not in ("low", "medium", "high"): | |
| raise ValueError("Invalid confidence") | |
| result["judgment"] = {key: obj[key] for key in ("A_propositions", "B_propositions", "missing_from_B", | |
| "added_in_B", "equivalent_substantive_content", | |
| "same_reasoning_structure", "confidence")} | |
| except (ValueError, json.JSONDecodeError) as exc: | |
| result["parse_error"] = str(exc) | |
| (output / f"{row['family_id']}.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n") | |
| print(json.dumps({"completed": min(offset + len(batch), len(pending_text)), "total": len(pending_text)}), flush=True) | |
| screen = [] | |
| for row in families: | |
| saved = json.loads((output / f"{row['family_id']}.json").read_text()) | |
| result = {"family_id": row["family_id"], "gold_split": row["gold_split"], | |
| "criterion": "verbosity", "judgment": saved.get("judgment"), "parse_error": saved.get("parse_error")} | |
| if result["judgment"]: | |
| j = result["judgment"] | |
| result["internal_equivalence_conflict"] = bool(j["equivalent_substantive_content"] and | |
| (j["missing_from_B"] or j["added_in_B"])) | |
| screen.append(result) | |
| with (OUT / "automatic_proposition_screen.jsonl").open("w") as handle: | |
| for result in screen: | |
| handle.write(json.dumps(result, ensure_ascii=False) + "\n") | |
| valid = [r["judgment"] for r in screen if r["judgment"]] | |
| summary = {"model_id": "microsoft/phi-4", "model_revision": "2db69c1c3e91a05d2c64a3185acfbaf36f744e25", | |
| "n_candidates": len(families), "n_valid": len(valid), | |
| "n_equivalent_substantive_content": sum(j["equivalent_substantive_content"] for j in valid), | |
| "n_same_reasoning_structure": sum(j["same_reasoning_structure"] for j in valid), | |
| "n_both": sum(j["equivalent_substantive_content"] and j["same_reasoning_structure"] for j in valid), | |
| "n_internal_equivalence_conflict": sum(r.get("internal_equivalence_conflict", False) for r in screen), | |
| "interpretation": "Model-only proposition diagnostic; never human Gold certification."} | |
| (OUT / "automatic_proposition_summary.json").write_text(json.dumps(summary, indent=2) + "\n") | |
| print(json.dumps(summary), flush=True) | |
| def main(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--input", type=Path, action="append", help="Family JSONL; repeatable. Defaults to available deterministic/model families.") | |
| parser.add_argument("--phi-audit", action="store_true", help="Run blind local Phi-4 on current automatic_judge_pairs.jsonl") | |
| parser.add_argument("--mistral-audit", action="store_true", help="Run blind local Mistral on Phi-generated candidates only") | |
| parser.add_argument("--phi-propositions", action="store_true", help="Run blind Phi-4 proposition-set/structure diagnostic for verbosity") | |
| parser.add_argument("--phi-model", type=Path, default=Path("/home/ubuntu/models/phi-4")) | |
| parser.add_argument("--mistral-model", type=Path, default=Path("/home/ubuntu/models/mistral-small-24b")) | |
| parser.add_argument("--device", default="cuda:5") | |
| parser.add_argument("--batch-size", type=int, default=4) | |
| parser.add_argument("--max-new-tokens", type=int, default=160) | |
| args = parser.parse_args() | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| if args.phi_audit: | |
| run_model_audit(args.phi_model, args.device, args.batch_size, args.max_new_tokens, | |
| "microsoft/phi-4", "phi-4", exclude_generator="phi") | |
| return | |
| if args.mistral_audit: | |
| run_model_audit(args.mistral_model, args.device, args.batch_size, args.max_new_tokens, | |
| "mistralai/Mistral-Small-24B-Instruct-2501", "mistral-small-24b", include_generator="phi") | |
| return | |
| if args.phi_propositions: | |
| run_phi_propositions(args.phi_model, args.device, args.batch_size, args.max_new_tokens) | |
| return | |
| paths = args.input or [OUT / "deterministic_families.jsonl", OUT / "model_families.jsonl", | |
| OUT / "phi_families.jsonl"] | |
| paths = [p for p in paths if p.exists()] | |
| if not paths: | |
| raise FileNotFoundError("No candidate family JSONL exists yet") | |
| records = [json.loads(line) for path in paths for line in path.open(encoding="utf-8") if line.strip()] | |
| ids = [row["family_id"] for row in records] | |
| if len(set(ids)) != len(ids): | |
| raise ValueError("Duplicate family_id across inputs") | |
| prompt_splits = collections.defaultdict(set) | |
| for row in records: | |
| prompt_splits[row["prompt_id"]].add(row["gold_split"]) | |
| if any(len(splits) > 1 for splits in prompt_splits.values()): | |
| raise ValueError("Gold train/test prompt leakage in candidate families") | |
| results, passed, rejected, judge_pairs = [], [], [], [] | |
| for row in records: | |
| stats, reasons, warnings = audit(row) | |
| result = {"family_id": row["family_id"], "prompt_id": row["prompt_id"], "gold_split": row["gold_split"], | |
| "requested_criterion": row["requested_criterion"], "operation": op_name(row), | |
| "automatic_pass": not reasons, "rejection_reasons": reasons, "warnings": warnings, **stats} | |
| results.append(result) | |
| (passed if not reasons else rejected).append({**row, "automatic_audit": result}) | |
| if not reasons: | |
| # Fixed A=clean, B=altered for model audits; human packet must randomize independently. | |
| judge_pairs.append({"intervention_id": row["family_id"], "edit_id": row["family_id"], | |
| "prompt_id": row["prompt_id"], "prompt": row["prompt"], | |
| "original_response": row["clean_response"], "edited_response": row["altered_response"]}) | |
| def write_jsonl(name, rows): | |
| with (OUT / name).open("w", encoding="utf-8") as handle: | |
| for item in rows: | |
| handle.write(json.dumps(item, ensure_ascii=False) + "\n") | |
| write_jsonl("automatic_screen.jsonl", results) | |
| write_jsonl("automatic_passed.jsonl", passed) | |
| write_jsonl("automatic_rejected.jsonl", rejected) | |
| write_jsonl("automatic_judge_pairs.jsonl", judge_pairs) | |
| counts = collections.Counter((r["gold_split"], r["requested_criterion"], r["automatic_pass"]) for r in results) | |
| summary = {"version": VERSION, "thresholds": THRESHOLDS, "inputs": [str(p.relative_to(ROOT)) for p in paths], | |
| "input_sha256": {str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}, | |
| "n_input": len(records), "n_automatic_passed": len(passed), "n_automatic_rejected": len(rejected), | |
| "n_prompt_ids": len(prompt_splits), "train_test_prompt_overlap": 0, | |
| "by_cell": [{"split": s, "criterion": c, "passed": counts[s, c, True], "rejected": counts[s, c, False]} | |
| for s in ("train", "test") for c in CRITERIA], | |
| "rejection_reasons": dict(collections.Counter(reason for item in results for reason in item["rejection_reasons"])), | |
| "warning_counts": dict(collections.Counter(warning for item in results for warning in item["warnings"])), | |
| "human_certified": 0} | |
| (OUT / "automatic_summary.json").write_text(json.dumps(summary, indent=2, ensure_ascii=False) + "\n") | |
| print(json.dumps({"n_input": len(records), "passed": len(passed), "rejected": len(rejected), "by_cell": summary["by_cell"]})) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 32.3 kB
- Xet hash:
- e9d3cfeb49e755b5b5fdab153a10de37455ee78ea69be50f1cc57d6048f83d5f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.