评测泄漏与 Benchmark 污染

mkdir l7-contamination
cd l7-contamination

"""对训练样例、提示示例、检索材料与测试题做去重和隔离审计。"""

import hashlib
import unicodedata

EXPOSURES = [
    {"id": "train-1", "pool": "train", "family": "book-policy",
     "text": "图书订单可以在收货7天内申请退款。"},
    {"id": "train-copy", "pool": "train", "family": "book-policy",
     "text": "图书订单可以在收货７天内申请退款！"},
    {"id": "prompt-1", "pool": "prompt", "family": "hotel-policy",
     "text": "酒店订单入住前可以取消。"},
    {"id": "retrieve-1", "pool": "retrieval", "family": "book-policy",
     "text": "图书政策修订说明：收货时间以签收记录为准。"},
]
TESTS = [
    {"id": "test-exact", "family": "book-policy",
     "text": "图书订单可以在收货７天内申请退款！"},
    {"id": "test-family", "family": "book-policy",
     "text": "图书退款到期日如何计算？"},
    {"id": "test-prompt", "family": "hotel-mirror",
     "text": "酒店订单入住前可以取消。"},
    {"id": "test-near", "family": "book-mirror",
     "text": "图书订单可以在收货7天内提交退款。"},
    {"id": "test-new", "family": "ticket-new",
     "text": "演唱会门票按票面规则退票。"},
]
NEAR_THRESHOLD = .55


def normalize(text):
    # 全角字符、大小写、标点和空白差异不应伪装成新样本。
    folded = unicodedata.normalize("NFKC", text).casefold()
    return "".join(c for c in folded if c.isalnum())


def fingerprint(text):
    return hashlib.sha256(normalize(text).encode("utf-8")).hexdigest()


def shingles(text, width=3):
    value = normalize(text)
    return {value[i:i + width] for i in range(max(0, len(value) - width + 1))}


def similarity(left, right):
    a, b = shingles(left), shingles(right)
    # 短文本退回规范化后的精确比较，避免两个空集合被当成相同长文。
    if not a or not b:
        return float(normalize(left) == normalize(right))
    return len(a & b) / len(a | b)


def deduplicate(rows):
    seen = set()
    kept, removed = [], []
    for row in rows:
        # 各材料池分别去重，仍保存跨池暴露记录供审计。
        key = (row["pool"], fingerprint(row["text"]))
        if key in seen:
            removed.append(row["id"])
        else:
            seen.add(key)
            kept.append(row)
    return kept, removed


def audit(exposures, tests, threshold=NEAR_THRESHOLD):
    if not 0 <= threshold <= 1:
        raise ValueError("invalid similarity threshold")
    kept, duplicates = deduplicate(exposures)
    clean, quarantined = [], []
    for test in tests:
        reasons = set()
        for known in kept:
            if test["family"] == known["family"]:
                reasons.add("source-family")
            if fingerprint(test["text"]) == fingerprint(known["text"]):
                reasons.add("normalized-exact")
            elif similarity(test["text"], known["text"]) >= threshold:
                reasons.add("lexical-near")
        if reasons:
            quarantined.append((test["id"], sorted(reasons)))
        else:
            clean.append(test["id"])
    return duplicates, clean, quarantined


def main():
    duplicates, clean, quarantined = audit(EXPOSURES, TESTS)
    print(f"duplicates_removed={','.join(duplicates)}")
    for identifier, reasons in quarantined:
        print(f"quarantine {identifier}: {','.join(reasons)}")
    print(f"clean_tests={','.join(clean)} count={len(clean)}")
    print(f"near_threshold={NEAR_THRESHOLD:.2f} test_near_similarity="
          f"{similarity(TESTS[3]['text'], EXPOSURES[0]['text']):.3f}")
    print("external_pretraining_exposure=unknown")


if __name__ == "__main__":
    main()

ls
python -X utf8 contamination_lab.py

duplicates_removed=train-copy
quarantine test-exact: normalized-exact,source-family
quarantine test-family: source-family
quarantine test-prompt: normalized-exact
quarantine test-near: lexical-near
clean_tests=test-new count=1
near_threshold=0.55 test_near_similarity=0.556
external_pretraining_exposure=unknown

ls
python -X utf8 -m unittest -v test_contamination_lab.py