#!/usr/bin/env python3 """FP-Fusion scorer: 自称提取 / 信号得分 / 门控 / 五档裁决 / 红旗.""" import json import os import re from pathlib import Path _SEVERITY_ORDER = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2} _CUTOFF_RE = re.compile( r"(?:cutoff|knowledge|training|截止|知识)[\s\w]*(?:is|was|in|until|up to|是|在)?\s*" r"((?:january|february|march|april|may|june|july|august|september|october|november|" r"december)\s+\d{4}|\d{4}[-/年]\d{1,2}|\d{4} 年? \d{1,2} 月|\d{4}年)", re.IGNORECASE) _NEGATION_RE = re.compile( r"(not|isn't|isn’t|am not|aren't|rather than|instead of|并非|不是|而不是|而不是)\s*" r"(?:an?\s+)?\w{0,12}$", re.IGNORECASE) _METACOG_NUM_RE = re.compile( r"\b(\d{1,4}(?:\.\d+)?)\s*([bmb]ill?ion|b\b|mb|m\b|亿|万亿|千亿|百亿|十亿)\b|" r"(\d+)\s*(?:gpus?|h800|a100|h100|v100|tpu|张\s?(?:gpu|卡))", re.IGNORECASE) def load_aliases(path=None): path = path or str(Path(__file__).parent / 'family_aliases.json') with open(path, encoding='utf-8') as f: return json.load(f) def _families_in_text(text, aliases): """返回文本中提到的家族集合(带否定前缀过滤).""" low = text.lower() found = set() for family, spec in aliases.items(): for tok in spec['tokens']: idx = low.find(tok.lower()) while idx != -1: prefix = low[max(0, idx - 25):idx] if not _NEGATION_RE.search(prefix): found.add(family) break idx = low.find(tok.lower(), idx + 1) return found def requested_family(model_name, aliases): fams = _families_in_text(model_name.lower(), aliases) return next(iter(fams)) if len(fams) == 1 else None def identity_signal(i_records, aliases, req_family): """自称一致分 + 中英一致性 + 离群自称.""" parseable = consistent = 0 outliers, zh_en_bad = [], False pair_fams = {} for r in i_records: text = r.get('response') or '' if r.get('error') or not text: continue fams = _families_in_text(text, aliases) meta = r.get('meta') or {} pair = meta.get('pair') lang = meta.get('lang') if pair and lang in ('en', 'zh'): pair_fams.setdefault(pair, {})[lang] = fams if len(fams) == 1: parseable += 1 fam = next(iter(fams)) if req_family is None or fam == req_family: consistent += 1 else: outliers.append({'probe': r['id'], 'claimed_family': fam, 'excerpt': text[:120]}) elif len(fams) > 1: parseable += 1 if req_family and req_family not in fams: outliers.append({'probe': r['id'], 'claimed_family': sorted(fams), 'excerpt': text[:120]}) # 中英配对一致性: 同一 pair 两侧声称家族交集非空 → 一致 for pair, sides in pair_fams.items(): en, zh = sides.get('en') or set(), sides.get('zh') or set() if en and zh and not (en & zh): zh_en_bad = True s_idn = (consistent / parseable) if parseable else 0.5 return {'s_idn': s_idn, 'parseable': parseable, 'consistent': consistent, 'outliers': outliers[:5], 'zh_en_consistent': not zh_en_bad} def meta_signal(k_records, i_records): """S_meta: 截止口径唯一 + 元认知审计不编造.""" cutoffs = set() for r in k_records + i_records: text = r.get('response') if text: for m in _CUTOFF_RE.finditer(text): cutoffs.add(re.sub(r'\s+', ' ', m.group(1).strip().lower())) metacog_fabricated = [] for r in k_records: if not (r.get('meta') or {}).get('metacog'): continue text = r.get('response') or '' if text and _METACOG_NUM_RE.search(text) and 'refuse' not in text.lower() \ and '无法' not in text and "don't have" not in text.lower(): metacog_fabricated.append({'probe': r['id'], 'excerpt': text[:120]}) cutoff_ok = 1.0 if len(cutoffs) <= 1 else 0.4 metacog_ok = 1.0 if not metacog_fabricated else 0.4 s_meta = (cutoff_ok + metacog_ok) / 2 return {'s_meta': s_meta, 'cutoffs_unique': sorted(cutoffs), 'metacog_fabricated': metacog_fabricated} def dist_signal(entries, mean_jsd, split_half): """S_dist: 相对比 R = meanJSD / max(splitHalf, 0.02), 带绝对标尺封顶.""" if mean_jsd is None: return None sh = split_half if split_half and split_half > 0 else 0.02 ratio = mean_jsd / max(sh, 0.02) if ratio < 2: s = 1.0 elif ratio > 8: s = 0.0 else: s = 1.0 - (ratio - 2) / 6 if mean_jsd > 0.35: # 论文绝对标尺: 不同模型档封顶 s = min(s, 0.2) return {'s_dist': s, 'mean_jsd': mean_jsd, 'split_half': split_half, 'relative_ratio': round(ratio, 2), 'comparable_cells': len(entries)} def refuse_gradient_pattern(c_records): """拒答梯度向量(存档信号): {level: True(拒)/False(顺)}""" grad = {} for r in c_records: g = (r.get('meta') or {}).get('refusal_grad') if g: text = (r.get('response') or '').lower() grad[f'L{g}'] = any(s in text for s in ("i can't", "i cannot", "无法", "抱歉", "sorry")) return grad def length_compliance(s_records): """长度服从性(存档信号): 恰好 N 词/1 句 的服从率.""" out = [] for r in s_records: n = (r.get('meta') or {}).get('len_ctrl') if not n or r.get('error'): continue text = (r.get('response') or '').strip() if n == 3: words = len([w for w in re.split(r'\W+', text) if w]) out.append({'probe': r['id'], 'target': 3, 'actual_words': words, 'ok': words == 3}) else: sents = len([x for x in re.split(r'[.!?。！？]', text) if x.strip()]) out.append({'probe': r['id'], 'target': 1, 'actual_sents': sents, 'ok': sents == 1}) return out def verdict_from_score(score, has_reference): if score >= 0.85: v = 'VERIFIED' elif score >= 0.70: v = 'LIKELY_MATCH' elif score >= 0.50: v = 'INCONCLUSIVE' elif score >= 0.30: v = 'SUSPECTED_MISMATCH' else: v = 'MISMATCH' if not has_reference and v == 'VERIFIED': v = 'LIKELY_MATCH' # 无参考不得"验明正身" return v def build_report(records, d_norm, dist_cmp, model_name, reference_info, aliases, tokens_used, elapsed_s, attribution=None, adversarial=None, mode='verify'): text_records = [r for r in records if r['layer'] in ('I', 'K', 'C', 'S')] ok_text = [r for r in text_records if not r['error']] total = len(records) success = sum(1 for r in records if not r['error']) rate = success / max(total, 1) quality = ('SUFFICIENT' if success >= 8 and rate >= 0.8 else ('DEGRADED' if success >= 4 and rate >= 0.5 else 'INSUFFICIENT')) req_family = requested_family(model_name, aliases) i_records = [r for r in records if r['layer'] == 'I'] k_records = [r for r in records if r['layer'] == 'K'] c_records = [r for r in records if r['layer'] == 'C'] s_records = [r for r in records if r['layer'] == 'S'] idn = identity_signal(i_records, aliases, req_family) meta = meta_signal(k_records, i_records) # 延迟: 按输出 token 归一化的"固定开销"估算, 替代固定 10s 绝对阈值. # decode_rate = median(text 单条延迟/completion_tokens) → 纯解码速度 # overhead = baseline_p50 − decode_rate×基线平均completion_tokens # 代理/中转会给每个请求叠加固定的网络开销, 短请求(基线)上最显形; # 纯硬件慢(CPU)只影响 decode_rate, 不会产生 overhead → 不再冤枉慢端点。 text_ok = [r for r in ok_text if (r.get('completion_tokens') or 0) > 0] per_tok = sorted(r['latency_ms'] / r['completion_tokens'] for r in text_ok) decode_rate = per_tok[len(per_tok) // 2] if per_tok else None base_records = [r for r in records if r['layer'] == 'BASE' and not r['error']] base_ct = [r.get('completion_tokens') or 1 for r in base_records] mean_base_ct = sum(base_ct) / len(base_ct) if base_ct else 1.0 baseline = dist_cmp.get('baseline_p50') if isinstance(dist_cmp, dict) else None overhead_ms, overhead_ratio = None, None if baseline and decode_rate: overhead_ms = baseline - decode_rate * mean_base_ct # 用比值而非绝对值判定: CPU 等慢端点的 prefill 开销会随硬件慢等比放大, # 固定 10s 阈值会冤枉它; 真正的代理/中转会让短请求比按解码率外推贵数倍 expected = decode_rate * mean_base_ct overhead_ratio = baseline / expected if expected > 0 else None latency_anomaly = bool(overhead_ratio is not None and overhead_ratio > 5 and overhead_ms is not None and overhead_ms > 5_000) red_flags = [] if quality != 'SUFFICIENT': red_flags.append({'severity': 'HIGH' if quality == 'INSUFFICIENT' else 'MEDIUM', 'category': 'evidence', 'description': f'{quality} evidence: {success}/{total} probes succeeded', 'evidence': f'Success rate {rate:.0%}'}) if idn['parseable'] and idn['consistent'] < idn['parseable']: red_flags.append({'severity': 'HIGH', 'category': 'identity', 'description': f"Self-identification deviates from requested " f"name '{model_name}' (family={req_family})", 'evidence': json.dumps(idn['outliers'][:3], ensure_ascii=False)}) if not idn['zh_en_consistent']: red_flags.append({'severity': 'MEDIUM', 'category': 'consistency_zh_en', 'description': 'Chinese vs English self-identification disagree', 'evidence': 'paired identity probes'}) if len(meta['cutoffs_unique']) > 1: red_flags.append({'severity': 'HIGH', 'category': 'consistency', 'description': 'Inconsistent knowledge cutoff dates', 'evidence': ', '.join(meta['cutoffs_unique'])}) if meta['metacog_fabricated']: red_flags.append({'severity': 'LOW', 'category': 'metacog', 'description': 'States specific parameter counts / training ' 'hardware (typical of substituted small models)', 'evidence': json.dumps(meta['metacog_fabricated'][:2], ensure_ascii=False)}) if latency_anomaly: red_flags.append({'severity': 'MEDIUM', 'category': 'latency', 'description': f'Estimated fixed per-request overhead ' f'{overhead_ms:.0f}ms (baseline p50 {baseline:.0f}ms ' f'vs decode-rate expectation) suggests proxy/relay', 'evidence': f'decode_rate={decode_rate:.1f}ms/tok, ' f'base_ct={mean_base_ct:.1f}'}) # v1.1: 单 cell 极端分化 → 实锤级信号(兄弟假冒案例中均值被数字cell稀释, 单cell达1.0) outlier_cells = dist_cmp.get('outlier_cells') or [] dist_outlier = bool(dist_cmp.get('dist_outlier')) if dist_outlier: red_flags.append({'severity': 'MEDIUM', 'category': 'dist_outlier', 'description': f'{len(outlier_cells)} cell(s) show extreme ' f'distribution divergence (JSD>0.5, n>=15)', 'evidence': json.dumps(outlier_cells, ensure_ascii=False)}) # ---- 融合 ---- has_ref = dist_cmp.get('mean_jsd') is not None s_fam = (attribution or {}).get('s_fam', 0.0) # 维度A：有参考时引入家族归因信号（0.25权重），身份/分布相应下调； # 无参考（自证模式）保持原权重，归因仅作辅助展示。 if has_ref: final = (0.35 * dist_cmp['s_dist'] + 0.20 * idn['s_idn'] + 0.20 * meta['s_meta'] + 0.25 * s_fam) else: final = 0.60 * idn['s_idn'] + 0.40 * meta['s_meta'] if quality != 'SUFFICIENT': final = min(final, 0.5) score = round(max(0.0, min(1.0, final)), 4) verdict = verdict_from_score(score, has_ref) # v1.1: dist_outlier 实锤信号 → 裁决封顶 SUSPECTED_MISMATCH(不许高于此档; # INCONCLUSIVE 也被视为"证据被稀释", 由离群 cell 证据直接升级) if dist_outlier: _order = ['VERIFIED', 'LIKELY_MATCH', 'INCONCLUSIVE', 'SUSPECTED_MISMATCH', 'MISMATCH'] if _order.index(verdict) < _order.index('SUSPECTED_MISMATCH'): verdict = 'SUSPECTED_MISMATCH' # 维度C：impersonation 实锤 → 同样封顶 SUSPECTED_MISMATCH（复用裁决帽机制） impersonation_flag = bool(adversarial and adversarial.get('impersonation_flag')) if impersonation_flag: _order = ['VERIFIED', 'LIKELY_MATCH', 'INCONCLUSIVE', 'SUSPECTED_MISMATCH', 'MISMATCH'] if _order.index(verdict) < _order.index('SUSPECTED_MISMATCH'): verdict = 'SUSPECTED_MISMATCH' report = { 'benchmark': 'fp_fusion', 'version': '1.1', 'score': score, 'num': total, 'verdict': verdict, 'mode': 'reference_verify' if has_ref else 'self_consistency', 'model': model_name, 'reference': reference_info, 'gate': {'total_probes': total, 'successful_probes': success, 'success_rate': round(rate, 3), 'quality': quality}, 'signals': { 'dist': {k: v for k, v in (dist_cmp or {}).items() if k != 'baseline_p50'} if has_ref else {'enabled': False, 'split_half_jsd': dist_cmp.get('split_half')}, 'family': (attribution if attribution is not None else {'enabled': False, 'note': 'reserved hook (v1 - 未启用归因)'}), 'identity': {'s_idn': round(idn['s_idn'], 3), 'parseable': idn['parseable'], 'consistent': idn['consistent'], 'zh_en_consistent': idn['zh_en_consistent'], 'outliers': idn['outliers']}, 'meta': {'s_meta': round(meta['s_meta'], 3), 'cutoffs_unique': meta['cutoffs_unique'], 'refusal_gradient': refuse_gradient_pattern(c_records), 'length_compliance': length_compliance(s_records)}, 'adversarial': adversarial if adversarial is not None else {'enabled': False}, 'latency': {'baseline_p50_ms': baseline, 'decode_rate_ms_per_tok': round(decode_rate, 1) if decode_rate else None, 'estimated_overhead_ms': round(overhead_ms, 1) if overhead_ms is not None else None, 'overhead_ratio': round(overhead_ratio, 2) if overhead_ratio else None, 'anomaly': latency_anomaly}, }, 'red_flags': sorted(red_flags, key=lambda f: _SEVERITY_ORDER.get(f['severity'], 3)), 'tokens_used': tokens_used, 'elapsed_s': round(elapsed_s, 1), } if mode != 'verify': report['mode_detail'] = mode return report def robustness_signal(records, temperature_sweep=None): """维度 D：鲁棒性正交信号。 三个轴： temp_axis : 文本层同探针在不同温度下的回答一致性（归一化到 [0,1]，1=完全一致） lang_axis : 文本层 en/zh 同探针（pair 配对）回答一致性率 paraphrase_axis : D 层同 cell 不同 prompt_var 的回答分布 JSD（0=最稳，1=最有漂移） 输出块（写入 report['signals']['robustness']）。 """ text_ok = [r for r in records if r.get('layer') in ('I', 'K', 'C', 'S', 'V') and not r.get('error')] # ---- 温度轴：同一探针 id 在多个温度下的回答一致性 ---- temp_axis = {'enabled': bool(temperature_sweep and len(temperature_sweep) > 1), 'temperatures': temperature_sweep or [], 'probes_covered': 0, 'mean_consistency': None} if temp_axis['enabled']: by_probe = {} for r in text_ok: by_probe.setdefault(r['id'], []).append(r) consist = [] for pid, rs in by_probe.items(): temps = {r.get('temperature') for r in rs} if len(temps) < 2: continue temp_axis['probes_covered'] += 1 # 两两回答归一化比较 normed = [(r.get('response') or '').strip().lower() for r in rs] same = 0 pairs = 0 for i in range(len(normed)): for j in range(i + 1, len(normed)): pairs += 1 if normed[i] and normed[i] == normed[j]: same += 1 consist.append(same / pairs if pairs else 0) if consist: temp_axis['mean_consistency'] = round(sum(consist) / len(consist), 3) # ---- 语言轴：pair 配对的 en/zh 回答一致率 ---- lang_axis = {'enabled': False, 'probes_covered': 0, 'mean_consistency': None} pair_probes = {} for r in text_ok: m = r.get('meta') or {} if m.get('pair') and m.get('lang') in ('en', 'zh'): pair_probes.setdefault(m['pair'], {})[m['lang']] = \ (r.get('response') or '').strip().lower() if pair_probes: lang_axis['enabled'] = True matches = 0 checks = 0 for p, sides in pair_probes.items(): if sides.get('en') and sides.get('zh'): checks += 1 if sides['en'] and sides['en'] == sides['zh']: matches += 1 lang_axis['probes_covered'] = checks lang_axis['mean_consistency'] = \ round(matches / checks, 3) if checks else None # ---- 改写轴：D 层同 cell 不同 prompt_var 的分布 JSD ---- para_axis = {'enabled': False, 'cells_covered': 0, 'mean_jsd': None} d_ok = [r for r in records if r.get('layer') == 'D' and not r.get('error') and r.get('prompt_var')] if d_ok: from .engine import jsd_bits by_cell_var = {} for r in d_ok: cell = r['cell'] key = (cell, r.get('prompt_var')) norm = r.get('norm') if 'norm' in r else None by_cell_var.setdefault(cell, {}) cnt = by_cell_var[cell] cnt_key = key cnt.setdefault(cnt_key, {}) by_cell_var[cell][cnt_key] = cnt[cnt_key] # 用 response first-token 简化做分布（避免引入 build_d_normalized 循环依赖） tok = (r.get('response') or '').strip().lower().split()[0] \ if (r.get('response') or '').strip() else '__empty__' cnt[cnt_key][tok] = cnt[cnt_key].get(tok, 0) + 1 jsds = [] cells_covered = 0 for cell, var_map in by_cell_var.items(): if len(var_map) < 2: continue cells_covered += 1 keys = list(var_map.keys()) for i in range(len(keys)): for j in range(i + 1, len(keys)): jsds.append(jsd_bits(var_map[keys[i]], var_map[keys[j]])) if jsds: para_axis['enabled'] = True para_axis['cells_covered'] = cells_covered para_axis['mean_jsd'] = round(sum(jsds) / len(jsds), 3) return {'temp_axis': temp_axis, 'lang_axis': lang_axis, 'paraphrase_axis': para_axis}