| |
| |
| """ |
| DGX AI Factory - Multi-source Raw JSONL Quality Auditor |
| |
| κΈ°λ₯ |
| - μ
λ ₯ JSONL μλ νμ λλ --input μ§μ |
| - JSON/μ€ν€λ§/λΉ μλ΅/μ€λ³΅/μ€λ₯ νμ΄μ§/νλ μ΄μ€νλ/λ°λ³΅/μ½λνμ€ κ²μ¬ |
| - systemd / CUDA κ΄λ ¨μ± κ²μ¬ |
| - clean / review / rejected λΆλ¦¬ |
| - μ κ·νλ clean νμ΅ νμΌκ³Ό JSON λ³΄κ³ μ μμ± |
| |
| νμ€ λΌμ΄λΈλ¬λ¦¬λ§ μ¬μ©ν©λλ€. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| import re |
| import sys |
| from collections import Counter, defaultdict |
| from dataclasses import dataclass, asdict |
| from pathlib import Path |
| from typing import Any, Iterable, Optional |
|
|
|
|
| DEFAULT_EXPERIMENT_DIR = Path( |
| "/home/harness_user_1/dgx_ai_factory/102_qwen3_4b_mobile/" |
| "experiments/qwen3_4b_multisource_raw10k_v1_1" |
| ) |
|
|
| EXCLUDED_NAME_PARTS = ( |
| "clean", |
| "review", |
| "reject", |
| "audit", |
| "report", |
| "filtered", |
| "normalized", |
| ) |
|
|
| PREFERRED_NAME_PARTS = ( |
| "raw", |
| "collected", |
| "collection", |
| "dataset", |
| "final", |
| "train", |
| ) |
|
|
| PROMPT_KEYS = ( |
| "instruction", |
| "question", |
| "prompt", |
| "query", |
| "user", |
| "problem", |
| "task", |
| "input_text", |
| ) |
|
|
| ANSWER_KEYS = ( |
| "output", |
| "answer", |
| "response", |
| "assistant", |
| "completion", |
| "solution", |
| "target", |
| "output_text", |
| ) |
|
|
| INPUT_KEYS = ("input", "context", "additional_input") |
|
|
| CATEGORY_KEYS = ("category", "domain", "topic", "label", "task_type") |
| SOURCE_KEYS = ("source", "dataset", "origin", "provider", "repo", "dataset_name") |
|
|
| SYSTEMD_TERMS = ( |
| "systemd", "systemctl", "journalctl", "unit file", "service file", |
| "daemon-reload", "execstart", "wantedby", "restart=", "status=217", |
| "/etc/systemd/system", "μλΉμ€ μ λ", "μμ€ν
λ", |
| ) |
|
|
| CUDA_TERMS = ( |
| "cuda", "nvidia-smi", "nvcc", "cudnn", "nccl", "gpu", "vram", |
| "out of memory", "oom", "torch.cuda", "cuda_visible_devices", |
| "λ©λͺ¨λ¦¬ λΆμ‘±", "gpu λ©λͺ¨λ¦¬", "cuda oom", |
| ) |
|
|
| ERROR_PAYLOAD_TERMS = ( |
| "access denied", |
| "forbidden", |
| "not found", |
| "too many requests", |
| "rate limit exceeded", |
| "internal server error", |
| "bad gateway", |
| "service unavailable", |
| "dataset viewer is disabled", |
| "dataset viewer is not supported", |
| "configuration not found", |
| "split not found", |
| "invalid token", |
| ) |
|
|
| BOILERPLATE_TERMS = ( |
| "as an ai language model", |
| "i cannot browse", |
| "i don't have access", |
| "consult the documentation", |
| "please refer to the documentation", |
| "λμμ λ릴 μ μμ΅λλ€", |
| "μ νν μ 보λ 곡μ λ¬Έμλ₯Ό", |
| ) |
|
|
| PLACEHOLDER_PATTERNS = ( |
| re.compile(r"\{[A-Za-z_][A-Za-z0-9_ -]{1,40}\}"), |
| re.compile(r"<(?:placeholder|insert|your_|todo)[^>]*>", re.I), |
| re.compile(r"\[(?:insert|placeholder|your |todo)[^\]]*\]", re.I), |
| re.compile(r"\b(?:TODO|TBD|FIXME)\b"), |
| re.compile(r"\$\{[^}]+\}"), |
| ) |
|
|
| CHINESE_CHAR_RE = re.compile(r"[\u3400-\u4DBF\u4E00-\u9FFF]") |
| HTML_PAGE_RE = re.compile(r"(?is)<(?:!doctype\s+html|html|head|body)\b") |
| URL_RE = re.compile(r"https?://\S+") |
| WORD_RE = re.compile(r"[A-Za-z0-9κ°-ν£_./:+#-]+") |
|
|
|
|
| @dataclass |
| class AuditResult: |
| line_no: int |
| status: str |
| score: int |
| hard_reject: bool |
| reasons: list[str] |
| prompt: str |
| answer: str |
| context: str |
| category_declared: str |
| category_inferred: str |
| source: str |
| exact_hash: str |
| prompt_hash: str |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser( |
| description="Raw JSONL λ°μ΄ν° νμ§ κ°μ¬ λ° νν°λ§" |
| ) |
| parser.add_argument( |
| "--experiment-dir", |
| type=Path, |
| default=DEFAULT_EXPERIMENT_DIR, |
| help=f"μ€ν λλ ν°λ¦¬ κΈ°λ³Έκ°: {DEFAULT_EXPERIMENT_DIR}", |
| ) |
| parser.add_argument( |
| "--input", |
| type=Path, |
| default=None, |
| help="κ²μ¬ν JSONL νμΌ. μλ΅νλ©΄ μ€ν λλ ν°λ¦¬μμ μλ νμν©λλ€.", |
| ) |
| parser.add_argument( |
| "--output-dir", |
| type=Path, |
| default=None, |
| help="μΆλ ₯ λλ ν°λ¦¬. κΈ°λ³Έκ°: <experiment-dir>/quality_audit", |
| ) |
| parser.add_argument( |
| "--min-clean-score", |
| type=int, |
| default=80, |
| help="clean μ΅μ μ μ. κΈ°λ³Έκ° 80", |
| ) |
| parser.add_argument( |
| "--min-review-score", |
| type=int, |
| default=55, |
| help="review μ΅μ μ μ. κΈ°λ³Έκ° 55", |
| ) |
| parser.add_argument( |
| "--max-samples", |
| type=int, |
| default=30, |
| help="λ³΄κ³ μμ ν¬ν¨ν μνλ³ μμ μ΅λ κ°μ", |
| ) |
| return parser.parse_args() |
|
|
|
|
| def compact_json(value: Any) -> str: |
| try: |
| return json.dumps(value, ensure_ascii=False, sort_keys=True) |
| except Exception: |
| return repr(value) |
|
|
|
|
| def safe_string(value: Any) -> str: |
| if value is None: |
| return "" |
| if isinstance(value, str): |
| return value.strip() |
| if isinstance(value, (int, float, bool)): |
| return str(value) |
| if isinstance(value, list): |
| parts = [safe_string(x) for x in value] |
| return "\n".join(x for x in parts if x).strip() |
| if isinstance(value, dict): |
| return compact_json(value) |
| return str(value).strip() |
|
|
|
|
| def first_nonempty(obj: dict[str, Any], keys: Iterable[str]) -> str: |
| for key in keys: |
| if key in obj: |
| text = safe_string(obj.get(key)) |
| if text: |
| return text |
| return "" |
|
|
|
|
| def extract_from_messages(messages: Any) -> tuple[str, str, str]: |
| if not isinstance(messages, list): |
| return "", "", "" |
|
|
| user_parts: list[str] = [] |
| assistant_parts: list[str] = [] |
| context_parts: list[str] = [] |
|
|
| for item in messages: |
| if not isinstance(item, dict): |
| continue |
| role = safe_string(item.get("role")).lower() |
| content = item.get("content") |
| if isinstance(content, list): |
| text_chunks: list[str] = [] |
| for block in content: |
| if isinstance(block, dict): |
| if block.get("type") in {"text", "input_text", "output_text"}: |
| text_chunks.append(safe_string(block.get("text"))) |
| else: |
| text_chunks.append(safe_string(block)) |
| else: |
| text_chunks.append(safe_string(block)) |
| text = "\n".join(x for x in text_chunks if x).strip() |
| else: |
| text = safe_string(content) |
|
|
| if not text: |
| continue |
| if role in {"user", "human"}: |
| user_parts.append(text) |
| elif role in {"assistant", "gpt", "model"}: |
| assistant_parts.append(text) |
| elif role in {"system", "developer", "context"}: |
| context_parts.append(text) |
|
|
| prompt = "\n\n".join(user_parts).strip() |
| answer = "\n\n".join(assistant_parts).strip() |
| context = "\n\n".join(context_parts).strip() |
| return prompt, answer, context |
|
|
|
|
| def extract_record(obj: Any) -> tuple[str, str, str, str, str]: |
| if not isinstance(obj, dict): |
| return "", "", "", "", "" |
|
|
| prompt, answer, context = extract_from_messages(obj.get("messages")) |
|
|
| if not prompt: |
| prompt = first_nonempty(obj, PROMPT_KEYS) |
| if not answer: |
| answer = first_nonempty(obj, ANSWER_KEYS) |
| if not context: |
| context = first_nonempty(obj, INPUT_KEYS) |
|
|
| |
| if (not prompt or not answer) and isinstance(obj.get("data"), dict): |
| nested = obj["data"] |
| if not prompt: |
| prompt = first_nonempty(nested, PROMPT_KEYS) |
| if not answer: |
| answer = first_nonempty(nested, ANSWER_KEYS) |
| if not context: |
| context = first_nonempty(nested, INPUT_KEYS) |
|
|
| if (not prompt or not answer) and isinstance(obj.get("conversation"), list): |
| p2, a2, c2 = extract_from_messages(obj.get("conversation")) |
| prompt = prompt or p2 |
| answer = answer or a2 |
| context = context or c2 |
|
|
| category = first_nonempty(obj, CATEGORY_KEYS).lower() |
| source = first_nonempty(obj, SOURCE_KEYS) |
| return prompt, answer, context, category, source |
|
|
|
|
| def normalize_for_hash(text: str) -> str: |
| text = text.lower() |
| text = re.sub(r"https?://\S+", "<url>", text) |
| text = re.sub(r"\s+", " ", text) |
| text = re.sub(r"[`*_>#|~-]+", "", text) |
| return text.strip() |
|
|
|
|
| def sha256_text(text: str) -> str: |
| return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() |
|
|
|
|
| def infer_category(prompt: str, answer: str, declared: str) -> str: |
| text = f"{declared}\n{prompt}\n{answer}".lower() |
| systemd_score = sum(1 for term in SYSTEMD_TERMS if term in text) |
| cuda_score = sum(1 for term in CUDA_TERMS if term in text) |
|
|
| if systemd_score == 0 and cuda_score == 0: |
| return "other" |
| if systemd_score > cuda_score: |
| return "systemd" |
| if cuda_score > systemd_score: |
| return "cuda" |
| if "systemd" in declared: |
| return "systemd" |
| if "cuda" in declared or "gpu" in declared: |
| return "cuda" |
| return "mixed" |
|
|
|
|
| def repeated_line_ratio(text: str) -> float: |
| lines = [re.sub(r"\s+", " ", x.strip()) for x in text.splitlines() if x.strip()] |
| if len(lines) < 4: |
| return 0.0 |
| counts = Counter(lines) |
| repeated = sum(count - 1 for count in counts.values() if count > 1) |
| return repeated / len(lines) |
|
|
|
|
| def token_repetition_ratio(text: str) -> float: |
| words = WORD_RE.findall(text.lower()) |
| if len(words) < 30: |
| return 0.0 |
| unique = len(set(words)) |
| return 1.0 - (unique / len(words)) |
|
|
|
|
| def answer_is_url_only(answer: str) -> bool: |
| stripped = URL_RE.sub("", answer) |
| stripped = re.sub(r"[\s\-ββ:;,.()\[\]{}]+", "", stripped) |
| return bool(URL_RE.search(answer)) and len(stripped) < 20 |
|
|
|
|
| def category_relevance(category: str, prompt: str, answer: str) -> tuple[int, list[str]]: |
| text = f"{prompt}\n{answer}".lower() |
| reasons: list[str] = [] |
| penalty = 0 |
|
|
| if category == "systemd": |
| hits = sum(1 for term in SYSTEMD_TERMS if term in text) |
| if hits == 0: |
| penalty += 30 |
| reasons.append("systemd_relevance_zero") |
| elif hits == 1: |
| penalty += 10 |
| reasons.append("systemd_relevance_weak") |
| elif category == "cuda": |
| hits = sum(1 for term in CUDA_TERMS if term in text) |
| if hits == 0: |
| penalty += 30 |
| reasons.append("cuda_relevance_zero") |
| elif hits == 1: |
| penalty += 10 |
| reasons.append("cuda_relevance_weak") |
|
|
| return penalty, reasons |
|
|
|
|
| def audit_record( |
| obj: Any, |
| line_no: int, |
| seen_exact: set[str], |
| seen_prompt: dict[str, int], |
| min_clean_score: int, |
| min_review_score: int, |
| ) -> AuditResult: |
| prompt, answer, context, declared, source = extract_record(obj) |
| inferred = infer_category(prompt, answer, declared) |
|
|
| normalized_prompt = normalize_for_hash(prompt) |
| normalized_answer = normalize_for_hash(answer) |
| exact_hash = sha256_text(normalized_prompt + "\n---\n" + normalized_answer) |
| prompt_hash = sha256_text(normalized_prompt) |
|
|
| score = 100 |
| reasons: list[str] = [] |
| hard_reject = False |
|
|
| if not isinstance(obj, dict): |
| hard_reject = True |
| score = 0 |
| reasons.append("record_not_object") |
|
|
| if not prompt: |
| hard_reject = True |
| score = 0 |
| reasons.append("missing_prompt") |
|
|
| if not answer: |
| hard_reject = True |
| score = 0 |
| reasons.append("missing_answer") |
|
|
| if prompt and len(prompt.strip()) < 8: |
| score -= 25 |
| reasons.append("prompt_too_short") |
| elif prompt and len(prompt.strip()) < 20: |
| score -= 10 |
| reasons.append("prompt_short") |
|
|
| if answer and len(answer.strip()) < 20: |
| hard_reject = True |
| score = min(score, 20) |
| reasons.append("answer_too_short") |
| elif answer and len(answer.strip()) < 80: |
| score -= 25 |
| reasons.append("answer_short") |
| elif answer and len(answer.strip()) < 160: |
| score -= 10 |
| reasons.append("answer_somewhat_short") |
|
|
| combined = f"{prompt}\n{answer}".strip() |
| combined_lower = combined.lower() |
|
|
| if HTML_PAGE_RE.search(combined): |
| hard_reject = True |
| score = min(score, 10) |
| reasons.append("html_error_page") |
|
|
| error_hits = [term for term in ERROR_PAYLOAD_TERMS if term in combined_lower] |
| if error_hits and len(answer) < 800: |
| hard_reject = True |
| score = min(score, 15) |
| reasons.append("api_or_viewer_error_payload") |
|
|
| boilerplate_hits = [term for term in BOILERPLATE_TERMS if term in combined_lower] |
| if boilerplate_hits: |
| score -= 25 |
| reasons.append("generic_ai_boilerplate") |
|
|
| if any(pattern.search(combined) for pattern in PLACEHOLDER_PATTERNS): |
| score -= 35 |
| reasons.append("unfilled_placeholder") |
|
|
| if answer.count("```") % 2 != 0: |
| score -= 15 |
| reasons.append("unclosed_code_fence") |
|
|
| if answer_is_url_only(answer): |
| hard_reject = True |
| score = min(score, 20) |
| reasons.append("reference_or_url_only") |
|
|
| chinese_count = len(CHINESE_CHAR_RE.findall(combined)) |
| if chinese_count >= 3: |
| score -= 20 |
| reasons.append("possible_chinese_leakage") |
| elif chinese_count > 0: |
| score -= 8 |
| reasons.append("han_character_present") |
|
|
| line_repeat = repeated_line_ratio(answer) |
| if line_repeat >= 0.35: |
| score -= 30 |
| reasons.append("severe_repeated_lines") |
| elif line_repeat >= 0.15: |
| score -= 15 |
| reasons.append("repeated_lines") |
|
|
| token_repeat = token_repetition_ratio(answer) |
| if len(WORD_RE.findall(answer)) >= 80 and token_repeat >= 0.78: |
| score -= 15 |
| reasons.append("low_lexical_diversity") |
|
|
| |
| if exact_hash in seen_exact: |
| hard_reject = True |
| score = 0 |
| reasons.append("exact_duplicate") |
| else: |
| seen_exact.add(exact_hash) |
|
|
| |
| prior_prompt_line = seen_prompt.get(prompt_hash) |
| if prior_prompt_line is not None and "exact_duplicate" not in reasons: |
| score -= 20 |
| reasons.append("duplicate_prompt") |
| else: |
| seen_prompt[prompt_hash] = line_no |
|
|
| declared_for_check = declared |
| if "systemd" in declared_for_check: |
| declared_for_check = "systemd" |
| elif "cuda" in declared_for_check or "gpu" in declared_for_check: |
| declared_for_check = "cuda" |
| elif inferred in {"systemd", "cuda"}: |
| declared_for_check = inferred |
| else: |
| declared_for_check = "" |
|
|
| relevance_penalty, relevance_reasons = category_relevance( |
| declared_for_check, prompt, answer |
| ) |
| score -= relevance_penalty |
| reasons.extend(relevance_reasons) |
|
|
| |
| if len(answer) > 30000: |
| score -= 20 |
| reasons.append("answer_extremely_long") |
| elif len(answer) > 12000: |
| score -= 10 |
| reasons.append("answer_very_long") |
|
|
| score = max(0, min(100, score)) |
|
|
| if hard_reject or score < min_review_score: |
| status = "rejected" |
| elif score < min_clean_score: |
| status = "review" |
| else: |
| status = "clean" |
|
|
| return AuditResult( |
| line_no=line_no, |
| status=status, |
| score=score, |
| hard_reject=hard_reject, |
| reasons=sorted(set(reasons)), |
| prompt=prompt, |
| answer=answer, |
| context=context, |
| category_declared=declared, |
| category_inferred=inferred, |
| source=source, |
| exact_hash=exact_hash, |
| prompt_hash=prompt_hash, |
| ) |
|
|
|
|
| def count_lines_fast(path: Path) -> int: |
| count = 0 |
| with path.open("rb") as handle: |
| for _ in handle: |
| count += 1 |
| return count |
|
|
|
|
| def discover_input(experiment_dir: Path) -> Path: |
| candidates: list[Path] = [] |
|
|
| search_roots = [ |
| experiment_dir, |
| experiment_dir / "data", |
| experiment_dir / "raw", |
| experiment_dir / "outputs", |
| experiment_dir.parent.parent / "data", |
| ] |
|
|
| seen: set[Path] = set() |
| for root in search_roots: |
| if not root.exists(): |
| continue |
| for path in root.rglob("*.jsonl"): |
| try: |
| resolved = path.resolve() |
| except OSError: |
| continue |
| if resolved in seen: |
| continue |
| seen.add(resolved) |
|
|
| name_lower = path.name.lower() |
| if any(part in name_lower for part in EXCLUDED_NAME_PARTS): |
| continue |
| candidates.append(path) |
|
|
| if not candidates: |
| node-7.example.invalid FileNotFoundError( |
| f"JSONL μ
λ ₯ νμΌμ μ°Ύμ§ λͺ»νμ΅λλ€: {experiment_dir}\n" |
| "--input /μ λ/κ²½λ‘/νμΌ.jsonl λ‘ μ§μ νμμμ€." |
| ) |
|
|
| ranked: list[tuple[int, int, float, Path]] = [] |
| for path in candidates: |
| name_lower = path.name.lower() |
| preferred = sum(1 for part in PREFERRED_NAME_PARTS if part in name_lower) |
| try: |
| size = path.stat().st_size |
| mtime = path.stat().st_mtime |
| except OSError: |
| continue |
| ranked.append((preferred, size, mtime, path)) |
|
|
| if not ranked: |
| node-7.example.invalid FileNotFoundError("μ½μ μ μλ JSONL νλ³΄κ° μμ΅λλ€.") |
|
|
| ranked.sort(reverse=True) |
| return ranked[0][3] |
|
|
|
|
| def jsonl_write(handle, obj: Any) -> None: |
| handle.write(json.dumps(obj, ensure_ascii=False) + "\n") |
|
|
|
|
| def safe_original(obj: Any) -> dict[str, Any]: |
| if isinstance(obj, dict): |
| return obj |
| return {"_raw_value": obj} |
|
|
|
|
| def sample_view(result: AuditResult) -> dict[str, Any]: |
| return { |
| "line_no": result.line_no, |
| "status": result.status, |
| "score": result.score, |
| "hard_reject": result.hard_reject, |
| "reasons": result.reasons, |
| "category_declared": result.category_declared, |
| "category_inferred": result.category_inferred, |
| "source": result.source, |
| "prompt_preview": result.prompt[:500], |
| "answer_preview": result.answer[:800], |
| } |
|
|
|
|
| def determine_gate( |
| total: int, |
| clean: int, |
| review: int, |
| rejected: int, |
| hard_rejects: int, |
| duplicate_count: int, |
| ) -> tuple[str, list[str]]: |
| if total <= 0: |
| return "FAIL", ["no_valid_input_rows"] |
|
|
| clean_rate = clean / total |
| rejected_rate = rejected / total |
| hard_rate = hard_rejects / total |
| dup_rate = duplicate_count / total |
|
|
| reasons: list[str] = [] |
|
|
| if clean_rate < 0.65: |
| reasons.append("clean_rate_below_65_percent") |
| if rejected_rate > 0.20: |
| reasons.append("rejected_rate_above_20_percent") |
| if hard_rate > 0.12: |
| reasons.append("hard_reject_rate_above_12_percent") |
| if dup_rate > 0.10: |
| reasons.append("duplicate_rate_above_10_percent") |
|
|
| if ( |
| clean_rate >= 0.80 |
| and rejected_rate <= 0.10 |
| and hard_rate <= 0.05 |
| and dup_rate <= 0.05 |
| ): |
| return "PASS", ["quality_gate_passed"] |
|
|
| if clean_rate >= 0.65 and hard_rate <= 0.12: |
| return "HOLD", reasons or ["manual_review_required"] |
|
|
| return "FAIL", reasons or ["quality_gate_failed"] |
|
|
|
|
| def main() -> int: |
| args = parse_args() |
| experiment_dir = args.experiment_dir.expanduser().resolve() |
| input_path = ( |
| args.input.expanduser().resolve() |
| if args.input is not None |
| else discover_input(experiment_dir) |
| ) |
|
|
| if not input_path.is_file(): |
| print(f"[ERROR] μ
λ ₯ νμΌμ΄ μμ΅λλ€: {input_path}", file=sys.stderr) |
| return 2 |
|
|
| output_dir = ( |
| args.output_dir.expanduser().resolve() |
| if args.output_dir is not None |
| else experiment_dir / "quality_audit" |
| ) |
| output_dir.mkdir(parents=True, exist_ok=True) |
|
|
| clean_path = output_dir / "clean.jsonl" |
| review_path = output_dir / "review.jsonl" |
| rejected_path = output_dir / "rejected.jsonl" |
| normalized_path = output_dir / "clean_normalized_train.jsonl" |
| report_path = output_dir / "quality_audit_report.json" |
| summary_path = output_dir / "quality_audit_summary.txt" |
|
|
| total_lines = count_lines_fast(input_path) |
| print(f"[INPUT] {input_path}") |
| print(f"[TOTAL_LINES] {total_lines}") |
| print(f"[OUTPUT_DIR] {output_dir}") |
|
|
| status_counts: Counter[str] = Counter() |
| reason_counts: Counter[str] = Counter() |
| inferred_counts: Counter[str] = Counter() |
| declared_counts: Counter[str] = Counter() |
| source_counts: Counter[str] = Counter() |
| score_buckets: Counter[str] = Counter() |
| samples: dict[str, list[dict[str, Any]]] = defaultdict(list) |
|
|
| seen_exact: set[str] = set() |
| seen_prompt: dict[str, int] = {} |
|
|
| invalid_json_count = 0 |
| processed_count = 0 |
| hard_rejects = 0 |
|
|
| with ( |
| input_path.open("r", encoding="utf-8", errors="replace") as src, |
| clean_path.open("w", encoding="utf-8") as clean_out, |
| review_path.open("w", encoding="utf-8") as review_out, |
| rejected_path.open("w", encoding="utf-8") as rejected_out, |
| normalized_path.open("w", encoding="utf-8") as normalized_out, |
| ): |
| for line_no, raw_line in enumerate(src, start=1): |
| raw_line = raw_line.strip() |
|
|
| if not raw_line: |
| invalid_json_count += 1 |
| status_counts["rejected"] += 1 |
| hard_rejects += 1 |
| reason_counts["blank_line"] += 1 |
| rejected_obj = { |
| "_audit": { |
| "line_no": line_no, |
| "status": "rejected", |
| "score": 0, |
| "hard_reject": True, |
| "reasons": ["blank_line"], |
| }, |
| "_raw_line": "", |
| } |
| jsonl_write(rejected_out, rejected_obj) |
| continue |
|
|
| try: |
| obj = json.loads(raw_line) |
| except json.JSONDecodeError as exc: |
| invalid_json_count += 1 |
| status_counts["rejected"] += 1 |
| hard_rejects += 1 |
| reason_counts["invalid_json"] += 1 |
| rejected_obj = { |
| "_audit": { |
| "line_no": line_no, |
| "status": "rejected", |
| "score": 0, |
| "hard_reject": True, |
| "reasons": ["invalid_json"], |
| "json_error": str(exc), |
| }, |
| "_raw_line": raw_line[:5000], |
| } |
| jsonl_write(rejected_out, rejected_obj) |
| if len(samples["rejected"]) < args.max_samples: |
| samples["rejected"].append( |
| { |
| "line_no": line_no, |
| "status": "rejected", |
| "score": 0, |
| "hard_reject": True, |
| "reasons": ["invalid_json"], |
| "raw_preview": raw_line[:800], |
| } |
| ) |
| continue |
|
|
| result = audit_record( |
| obj=obj, |
| line_no=line_no, |
| seen_exact=seen_exact, |
| seen_prompt=seen_prompt, |
| min_clean_score=args.min_clean_score, |
| min_review_score=args.min_review_score, |
| ) |
| processed_count += 1 |
| status_counts[result.status] += 1 |
| if result.hard_reject: |
| hard_rejects += 1 |
|
|
| for reason in result.reasons: |
| reason_counts[reason] += 1 |
|
|
| inferred_counts[result.category_inferred or "unknown"] += 1 |
| declared_counts[result.category_declared or "unknown"] += 1 |
| source_counts[result.source or "unknown"] += 1 |
|
|
| if result.score >= 90: |
| score_buckets["90-100"] += 1 |
| elif result.score >= 80: |
| score_buckets["80-89"] += 1 |
| elif result.score >= 70: |
| score_buckets["70-79"] += 1 |
| elif result.score >= 55: |
| score_buckets["55-69"] += 1 |
| else: |
| score_buckets["0-54"] += 1 |
|
|
| wrapped = safe_original(obj).copy() |
| wrapped["_audit"] = { |
| "line_no": result.line_no, |
| "status": result.status, |
| "score": result.score, |
| "hard_reject": result.hard_reject, |
| "reasons": result.reasons, |
| "category_declared": result.category_declared, |
| "category_inferred": result.category_inferred, |
| "source": result.source, |
| "exact_hash": result.exact_hash, |
| "prompt_hash": result.prompt_hash, |
| } |
|
|
| if result.status == "clean": |
| jsonl_write(clean_out, wrapped) |
| normalized = { |
| "instruction": result.prompt, |
| "input": result.context, |
| "output": result.answer, |
| "category": ( |
| result.category_declared |
| if result.category_declared |
| else result.category_inferred |
| ), |
| "source": result.source, |
| "quality_score": result.score, |
| "quality_reasons": result.reasons, |
| "source_line": result.line_no, |
| } |
| jsonl_write(normalized_out, normalized) |
| elif result.status == "review": |
| jsonl_write(review_out, wrapped) |
| else: |
| jsonl_write(rejected_out, wrapped) |
|
|
| if len(samples[result.status]) < args.max_samples: |
| samples[result.status].append(sample_view(result)) |
|
|
| if total_lines > 0 and (line_no % 500 == 0 or line_no == total_lines): |
| pct = (line_no / total_lines) * 100 |
| print( |
| f"[PROGRESS] {line_no}/{total_lines} ({pct:.1f}%) " |
| f"clean={status_counts['clean']} " |
| f"review={status_counts['review']} " |
| f"rejected={status_counts['rejected']}" |
| ) |
|
|
| total = sum(status_counts.values()) |
| duplicate_count = ( |
| reason_counts["exact_duplicate"] + reason_counts["duplicate_prompt"] |
| ) |
| gate_status, gate_reasons = determine_gate( |
| total=total, |
| clean=status_counts["clean"], |
| review=status_counts["review"], |
| rejected=status_counts["rejected"], |
| hard_rejects=hard_rejects, |
| duplicate_count=duplicate_count, |
| ) |
|
|
| def rate(value: int) -> float: |
| return round((value / total * 100), 4) if total else 0.0 |
|
|
| report = { |
| "status": gate_status, |
| "gate_reasons": gate_reasons, |
| "input_file": str(input_path), |
| "output_dir": str(output_dir), |
| "thresholds": { |
| "min_clean_score": args.min_clean_score, |
| "min_review_score": args.min_review_score, |
| }, |
| "counts": { |
| "total_lines": total_lines, |
| "total_classified": total, |
| "parsed_records": processed_count, |
| "invalid_or_blank_lines": invalid_json_count, |
| "clean": status_counts["clean"], |
| "review": status_counts["review"], |
| "rejected": status_counts["rejected"], |
| "hard_rejects": hard_rejects, |
| "duplicate_flags": duplicate_count, |
| }, |
| "rates_percent": { |
| "clean": rate(status_counts["clean"]), |
| "review": rate(status_counts["review"]), |
| "rejected": rate(status_counts["rejected"]), |
| "hard_rejects": rate(hard_rejects), |
| "duplicate_flags": rate(duplicate_count), |
| }, |
| "reason_counts": dict(reason_counts.most_common()), |
| "category_inferred_counts": dict(inferred_counts.most_common()), |
| "category_declared_counts": dict(declared_counts.most_common()), |
| "source_counts": dict(source_counts.most_common()), |
| "score_buckets": dict(score_buckets), |
| "samples": samples, |
| "outputs": { |
| "clean": str(clean_path), |
| "review": str(review_path), |
| "rejected": str(rejected_path), |
| "clean_normalized_train": str(normalized_path), |
| "report": str(report_path), |
| "summary": str(summary_path), |
| }, |
| } |
|
|
| report_path.write_text( |
| json.dumps(report, ensure_ascii=False, indent=2), |
| encoding="utf-8", |
| ) |
|
|
| summary_lines = [ |
| "==== DGX RAW DATA QUALITY AUDIT ====", |
| f"STATUS={gate_status}", |
| f"INPUT={input_path}", |
| f"TOTAL={total}", |
| f"CLEAN={status_counts['clean']} ({rate(status_counts['clean']):.2f}%)", |
| f"REVIEW={status_counts['review']} ({rate(status_counts['review']):.2f}%)", |
| f"REJECTED={status_counts['rejected']} ({rate(status_counts['rejected']):.2f}%)", |
| f"HARD_REJECTS={hard_rejects} ({rate(hard_rejects):.2f}%)", |
| f"DUPLICATE_FLAGS={duplicate_count} ({rate(duplicate_count):.2f}%)", |
| f"INVALID_OR_BLANK={invalid_json_count}", |
| f"GATE_REASONS={','.join(gate_reasons)}", |
| "", |
| "==== TOP REASONS ====", |
| ] |
| summary_lines.extend( |
| f"{reason}={count}" for reason, count in reason_counts.most_common(20) |
| ) |
| summary_lines.extend( |
| [ |
| "", |
| "==== INFERRED CATEGORIES ====", |
| *( |
| f"{category}={count}" |
| for category, count in inferred_counts.most_common() |
| ), |
| "", |
| "==== OUTPUTS ====", |
| f"CLEAN_FILE={clean_path}", |
| f"REVIEW_FILE={review_path}", |
| f"REJECTED_FILE={rejected_path}", |
| f"NORMALIZED_TRAIN_FILE={normalized_path}", |
| f"REPORT_FILE={report_path}", |
| ] |
| ) |
|
|
| summary_text = "\n".join(summary_lines) + "\n" |
| summary_path.write_text(summary_text, encoding="utf-8") |
| print() |
| print(summary_text) |
|
|
| if gate_status == "PASS": |
| return 0 |
| if gate_status == "HOLD": |
| return 10 |
| return 20 |
|
|
|
|
| if __name__ == "__main__": |
| node-7.example.invalid SystemExit(main()) |
|
|
