data/forensic.py

"""Build our forensic SFT dataset (pattern/truth/discrepancy analysis).

Mixes public claim-verification data with our own hand-written seed examples,
all converted to a unified {persona, user, assistant} format. Assistant text
may contain <|scratchpad|> ... <|final|> markers (converted to special tokens
by the SFT trainer).
"""

import json
import random
from pathlib import Path

import pyarrow.parquet as pq
from huggingface_hub import hf_hub_download

HERE = Path(__file__).parent
OUT = HERE / "sft_forensic.jsonl"
SEED = HERE / "seed_forensic.jsonl"

LIAR_LABELS = {
    0: "false",
    1: "mostly false",
    2: "half true",
    3: "mostly true",
    4: "true",
    5: "pants on fire",
}
CF_LABELS = {0: "SUPPORTS", 1: "REFUTES", 2: "NOT_ENOUGH_INFO"}


def liar_examples(n=4000):
    import urllib.request

    url = "https://huggingface.co/datasets/UKPLab/liar/resolve/main/train.jsonl"
    req = urllib.request.Request(url, headers={"User-Agent": "curl/8"})
    rows = []
    for line in urllib.request.urlopen(req, timeout=120):
        d = json.loads(line)
        rows.append(d)
    random.shuffle(rows)
    out = []
    for d in rows[:n]:
        text = d.get("text", "").strip()
        label = d.get("label_text") or LIAR_LABELS.get(d.get("labels"), "unknown")
        context = d.get("context") or ""
        if not text:
            continue
        user = f"Evaluate this claim for accuracy. Claim: {text}"
        if context:
            user += f"\nContext: {context}"
        asst = (f"<|scratchpad|>Checklist: (1) identify the factual assertion; "
                f"(2) compare against known records; (3) note missing context. "
                f"The statement is a claim about an identifiable entity or event; "
                f"it requires a source beyond the claim itself. "
                f"<|final|>Verdict: {label}. Confidence: MEDIUM. "
                f"Reasoning: {label} indicates the statement diverges from established records; "
                f"no independent verification was supplied in the prompt.")
        out.append({"persona": "analyst", "user": user, "assistant": asst})
    return out


def climate_fever_examples():
    p = hf_hub_download("tdiggelm/climate_fever", "data/test-00000-of-00001.parquet",
                        repo_type="dataset", local_dir=str(HERE / "hf"))
    tab = pq.read_table(p)
    d = tab.to_pydict()
    out = []
    for claim, label, evs in zip(d["claim"], d["claim_label"], d["evidences"]):
        ev = evs[0] if evs else {}
        evidence = (ev.get("evidence") or ev.get("article") or "").strip()
        verdict = CF_LABELS.get(label, "NOT_ENOUGH_INFO")
        user = f"Given the evidence, does this claim hold? Claim: {claim}"
        if evidence:
            user += f"\nEvidence: {evidence}"
        asst = (f"<|scratchpad|>Compare claim against evidence: the evidence either "
                f"supports, refutes, or fails to address the claim. "
                f"<|final|>Verdict: {verdict}. "
                f"Confidence: MEDIUM. Reasoning: the available evidence was weighed "
                f"against the claim's assertions; any gap lowers confidence.")
        out.append({"persona": "analyst", "user": user, "assistant": asst})
    return out


def truthfulqa_examples():
    p = hf_hub_download("truthfulqa/truthful_qa", "generation/validation-00000-of-00001.parquet",
                        repo_type="dataset", local_dir=str(HERE / "hf"))
    tab = pq.read_table(p)
    d = tab.to_pydict()
    out = []
    for q, ans, wrong in zip(d["question"], d["best_answer"], d["incorrect_answers"]):
        user = f"Answer the following question truthfully, and rate your confidence. Question: {q}"
        note = ""
        if wrong:
            note = f" A common misconception is that {wrong[0].lower()}."
        asst = (f"<|scratchpad|>Identify what is being asked and what would need to be "
                f"true for popular wrong answers; check the baseline facts."
                f"<|final|>{ans}{note} Confidence: HIGH." if note else
                f"<|scratchpad|>Identify what is being asked and what would need to be "
                f"true for popular wrong answers; check the baseline facts."
                f"<|final|>{ans} Confidence: HIGH.")
        out.append({"persona": "analyst", "user": user, "assistant": asst})
    return out


def fallacy_examples(n=1500):
    p = hf_hub_download("tasksource/logical-fallacy",
                        "data/train-00000-of-00001-8c3d4e48fe0f561b.parquet",
                        repo_type="dataset", local_dir=str(HERE / "hf"))
    tab = pq.read_table(p)
    d = tab.to_pydict()
    idx = list(range(len(d["source_article"])))
    random.shuffle(idx)
    out = []
    for i in idx[:n]:
        text = (d["source_article"][i] or "").strip()
        label = (d["logical_fallacies"][i] or "unknown").strip()
        if not text:
            continue
        user = f"Identify any logical fallacy in this text, and explain why. Text: {text}"
        asst = (f"<|scratchpad|>The text's persuasive force rests on {label}: "
                f"it appeals to something other than evidence for the conclusion. "
                f"<|final|>Fallacy: {label}. Confidence: HIGH. "
                f"Reasoning: the conclusion is supported by an emotional or "
                f"irrelevant appeal rather than verifiable evidence.")
        out.append({"persona": "analyst", "user": user, "assistant": asst})
    return out


def seed_examples():
    out = []
    with open(SEED, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                out.append(json.loads(line))
    return out


def skeptic_variants(n=600):
    """Turn claim-analysis examples into 'attack this conclusion' (skeptic role)."""
    random.seed(11)
    import urllib.request

    url = "https://huggingface.co/datasets/UKPLab/liar/resolve/main/train.jsonl"
    req = urllib.request.Request(url, headers={"User-Agent": "curl/8"})
    rows = [json.loads(l) for l in urllib.request.urlopen(req, timeout=120)]
    random.shuffle(rows)
    out = []
    for d in rows[:n]:
        text = d.get("text", "").strip()
        label = d.get("label_text") or LIAR_LABELS.get(d.get("labels"), "unknown")
        if not text:
            continue
        user = (f"Act as the skeptic. Someone concluded this claim is '{label}'. "
                f"Tear down that conclusion: Claim: {text}")
        asst = (f"<|scratchpad|>Attack surfaces: (1) who verified the claim and how; "
                f"(2) is the source independent; (3) does the label overstate precision; "
                f"(4) what would change the verdict. "
                f"<|final|>Weakest link: verification provenance. The label '{label}' "
                f"summarizes a judgment, not a measurement; without an auditable "
                f"source chain it is provisional. Confidence: MEDIUM.")
        out.append({"persona": "skeptic", "user": user, "assistant": asst})
    return out


def main():
    random.seed(7)
    examples = []
    examples += liar_examples()
    examples += climate_fever_examples()
    examples += truthfulqa_examples()
    examples += fallacy_examples()
    examples += seed_examples()
    examples += skeptic_variants()
    random.shuffle(examples)
    with open(OUT, "w", encoding="utf-8") as f:
        for ex in examples:
            f.write(json.dumps(ex) + "\n")
    n_p = {}
    for ex in examples:
        n_p[ex["persona"]] = n_p.get(ex["persona"], 0) + 1
    print(f"wrote {len(examples)} examples -> {OUT}  personas={n_p}")


if __name__ == "__main__":
    main()