scripts/diagnose_contamination.py

#!/usr/bin/env python3
"""Diagnostic: Quantify training data contamination in LLM mode.

Measures n-gram overlap between LLM-generated text and MBFC gold text
at the sentence level. High overlap indicates the LLM is reproducing
memorized MBFC content rather than performing independent analysis.

Compares overlap rates across modes (LLM vs System vs Hybrid) to
quantify how much each mode relies on memorized vs discovered facts.

Usage:
    python scripts/diagnose_contamination.py [--results-dir results] [--model gpt-5-mini-2025-08-07]
"""

import argparse
import json
import logging
import os
import re
import sys
from collections import defaultdict

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)


def extract_ngrams(text: str, n: int = 4) -> set[tuple[str, ...]]:
    """Extract word-level n-grams from text."""
    words = re.findall(r'\b\w+\b', text.lower())
    return {tuple(words[i:i+n]) for i in range(len(words) - n + 1)}


def sentence_overlap(gen_text: str, gold_text: str, ngram_size: int = 4) -> dict:
    """Compute sentence-level overlap metrics between generated and gold text."""
    gen_sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', gen_text) if s.strip()]
    gold_sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', gold_text) if s.strip()]

    gold_ngrams = extract_ngrams(gold_text, ngram_size)
    if not gold_ngrams:
        return {"overlap_ratio": 0.0, "verbatim_sentences": 0, "total_sentences": len(gen_sentences)}

    verbatim_count = 0
    high_overlap_count = 0

    for sent in gen_sentences:
        sent_ngrams = extract_ngrams(sent, ngram_size)
        if not sent_ngrams:
            continue
        overlap = len(sent_ngrams & gold_ngrams) / len(sent_ngrams)
        if overlap > 0.8:
            verbatim_count += 1
        elif overlap > 0.5:
            high_overlap_count += 1

    gen_ngrams = extract_ngrams(gen_text, ngram_size)
    overall_overlap = len(gen_ngrams & gold_ngrams) / len(gen_ngrams) if gen_ngrams else 0.0

    return {
        "overall_ngram_overlap": overall_overlap,
        "verbatim_sentences": verbatim_count,
        "high_overlap_sentences": high_overlap_count,
        "total_sentences": len(gen_sentences),
        "verbatim_ratio": verbatim_count / len(gen_sentences) if gen_sentences else 0.0,
    }


def load_results(results_dir: str, mode: str, model: str) -> list[dict]:
    path = os.path.join(results_dir, f"{model}_{mode}.jsonl")
    if not os.path.exists(path):
        return []
    with open(path) as f:
        return [json.loads(line) for line in f]


def load_dataset(data_path: str = "data/mbfc_benchmark.json") -> dict:
    full_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), data_path)
    with open(full_path) as f:
        return {item["name"]: item for item in json.load(f)}


def build_text(raw: dict) -> str:
    return " ".join(filter(None, [
        raw.get('bias_category_description'),
        raw.get('overall_summary'),
        raw.get('analysis'),
        raw.get('history'),
        raw.get('ownership'),
    ])).strip()


def build_gold_text(item: dict) -> str:
    return " ".join(filter(None, [
        item.get('bias_category_description', ''),
        item.get('overall_summary', ''),
        item.get('analysis', ''),
        item.get('history', ''),
        item.get('ownership', ''),
    ])).strip()


def analyze_mode(results: list[dict], gold_data: dict, mode: str) -> dict:
    """Analyze contamination for a single mode."""
    overlaps = []

    for result in results:
        name = result.get("name")
        gold = gold_data.get(name)
        if not gold:
            continue

        raw = result.get("raw_output", {})
        gen_text = build_text(raw)
        gold_text = build_gold_text(gold)

        if not gen_text or not gold_text:
            continue

        metrics = sentence_overlap(gen_text, gold_text)
        metrics["outlet"] = name
        overlaps.append(metrics)

    if not overlaps:
        return {}

    avg_overlap = sum(o["overall_ngram_overlap"] for o in overlaps) / len(overlaps)
    avg_verbatim = sum(o["verbatim_ratio"] for o in overlaps) / len(overlaps)
    total_verbatim = sum(o["verbatim_sentences"] for o in overlaps)
    total_sentences = sum(o["total_sentences"] for o in overlaps)

    return {
        "mode": mode,
        "outlets_analyzed": len(overlaps),
        "avg_ngram_overlap": round(avg_overlap, 4),
        "avg_verbatim_ratio": round(avg_verbatim, 4),
        "total_verbatim_sentences": total_verbatim,
        "total_sentences": total_sentences,
        "per_outlet": sorted(overlaps, key=lambda x: x["overall_ngram_overlap"], reverse=True),
    }


def main():
    parser = argparse.ArgumentParser(description="Diagnose training data contamination")
    parser.add_argument("--results-dir", default="results")
    parser.add_argument("--model", default="gpt-5-mini-2025-08-07")
    parser.add_argument("--modes", default="llm,system,hybrid,articles")
    parser.add_argument("--dataset", default="data/mbfc_benchmark.json")
    args = parser.parse_args()

    gold_data = load_dataset(args.dataset)
    modes = args.modes.split(",")

    print(f"\n{'='*70}")
    print(f"CONTAMINATION DIAGNOSTIC REPORT")
    print(f"{'='*70}")
    print(f"Model: {args.model}")
    print(f"4-gram overlap between generated text and MBFC gold text")
    print(f"Higher overlap = more likely memorized from training data")
    print(f"{'='*70}\n")

    all_results = {}
    for mode in modes:
        results = load_results(args.results_dir, mode, args.model)
        if not results:
            print(f"  {mode}: No results found")
            continue

        analysis = analyze_mode(results, gold_data, mode)
        if not analysis:
            continue
        all_results[mode] = analysis

        print(f"  {mode:12s}: avg_overlap={analysis['avg_ngram_overlap']:.3f}  "
              f"verbatim_ratio={analysis['avg_verbatim_ratio']:.3f}  "
              f"verbatim_sents={analysis['total_verbatim_sentences']}/{analysis['total_sentences']}")

    print()

    # Show per-outlet details for the mode with highest overlap
    if all_results:
        highest_mode = max(all_results, key=lambda m: all_results[m]["avg_ngram_overlap"])
        print(f"Top 10 most contaminated outlets ({highest_mode} mode):")
        for item in all_results[highest_mode]["per_outlet"][:10]:
            print(f"  {item['outlet']:30s}: overlap={item['overall_ngram_overlap']:.3f}  "
                  f"verbatim={item['verbatim_sentences']}/{item['total_sentences']}")

    # Save full results
    output_path = os.path.join(args.results_dir, "diagnostic_contamination.json")
    with open(output_path, "w") as f:
        json.dump({mode: {k: v for k, v in data.items() if k != "per_outlet"}
                   for mode, data in all_results.items()}, f, indent=2)
    print(f"\nSummary saved to {output_path}")


if __name__ == "__main__":
    main()