pad_schema.py

# -*- coding: utf-8 -*-
"""Vibrato(颤音) PAD 契约 — 冻结的输出刻度与换算(单文件自包含)

Vibrato 是 Chordia(弦音) 生态的情绪前置网络: 输入消息+上轮PAD回声,
输出 P/A/D 三维 × 9档序数bin分布 + 全局置信度。压力值不进网络——
按 Chordia 标准由 ΔPAD 派生(本文件提供参照实现)。

刻度对齐(勿改, 改了就是与 chordia-engine 的契约破裂):
- chordia pad_calculator.py: 18词项李克特 1-9 分; P=mean(I1..I6), A=mean(I7..I12),
  D=mean(I13..I18); 标准化 = (均值-5)/4 → [-1,1]
- 本契约: 每维直接输出 9 档 bin 分布, 期望值 E[bin] ∈ [1,9] 等价于该维 6 词项均值,
  过同一标准化公式。chordia 历史 LLM 分析器的 18 项输出可用 items_to_bins()
  无损(核平滑)转成训练软标签。

echo 状态(v5 已废): 上轮 P/A/D 回声 3 浮点已溶解进 v6 消息范式——调用方把上轮
输出写回上一条 user 消息的 pad 字段即是回声(vib_messages.py)。
契约指纹 contract_digest() 训练前后/部署两侧比对, 不一致即停; v6 起指纹含
上下文窗口与 affect 消化能力(vib_messages.CONSUMED_FIELDS)。
"""

from __future__ import annotations

import hashlib
import json
import math
from typing import Dict, List, Sequence

import vib_messages

# ── 冻结刻度 ────────────────────────────────────────────────
DIMS: List[str] = ["pleasure", "arousal", "dominance"]
N_BINS = 9                          # 李克特 1..9, bin i ↔ 分值 i+1
BIN_CENTERS = list(range(1, 10))    # 1..9
ECHO_LEN = 3                        # v5 legacy: v6 起由消息范式状态伪token取代(仅桥接旧数据)

# chordia pad_calculator.py 的 18 词项分组(只取分组索引, 词项原文在
# chordia-engine/prompts/emotion_analysis.md, 不在此复制)
ITEM_GROUPING: Dict[str, List[str]] = {
    "pleasure": [f"I{i}" for i in range(1, 7)],
    "arousal": [f"I{i}" for i in range(7, 13)],
    "dominance": [f"I{i}" for i in range(13, 19)],
}
LIKERT_MIN, LIKERT_MID, LIKERT_HALF = 1.0, 5.0, 4.0


def expected_from_bins(dist: Sequence[float]) -> float:
    """bin 分布 → 李克特期望分值 ∈ [1,9]。"""
    if len(dist) != N_BINS:
        raise ValueError(f"bin 分布长度应为 {N_BINS}, 得到 {len(dist)}")
    total = float(sum(dist))
    if not (0.999 <= total <= 1.001):
        raise ValueError(f"bin 分布未归一化: sum={total}")
    return sum(p * c for p, c in zip(dist, BIN_CENTERS))


def normalize_likert(expected: float) -> float:
    """李克特期望值 → [-1,1], 与 chordia (均值-5)/4 同式。"""
    return max(-1.0, min(1.0, (expected - LIKERT_MID) / LIKERT_HALF))


def bins_to_pad(bins: Dict[str, Sequence[float]]) -> Dict[str, float]:
    """{dim: 9档分布} → {pleasure, arousal, dominance} 归一化 PAD。"""
    return {dim: round(normalize_likert(expected_from_bins(bins[dim])), 6) for dim in DIMS}


def likert_to_bins(value: float, half_width: float = 1.0) -> List[float]:
    """单个李克特分值 → 9档帐篷核分布(items_to_bins 的单值内核)。

    老数据桥接用: chordia 60k 的 PAD([-4,4]) 换算 L = pad + 5 即落 [1,9]。
    """
    if not (1.0 <= value <= 9.0):
        raise ValueError(f"李克特分越界: {value}")
    acc = [max(0.0, 1.0 - abs(bc - value) / half_width) for bc in BIN_CENTERS]
    total = sum(acc)
    return [v / total for v in acc] if total > 0 else [1.0 / N_BINS] * N_BINS


def items_to_bins(scores: Dict[str, float], half_width: float = 1.0) -> Dict[str, List[float]]:
    """chordia 18 词项评分 → 3×9 软分布。

    half_width=1.0(默认)是帐篷核: 整数分值落单 bin, 带小数分值在相邻两 bin
    按线性比例插值——**精确保留期望值**(两位小数分数的插值即其自然不确定性)。
    同维 6 个核平均后归一化: 词项一致时分布尖锐, 分歧时展宽, 不确定性留在
    分布形状里。half_width>1 会附加平滑但引入期望偏差(O((h-1)/4)), 谨慎用。
    """
    out: Dict[str, List[float]] = {}
    for dim, items in ITEM_GROUPING.items():
        acc = [0.0] * N_BINS
        n_used = 0
        for key in items:
            if key not in scores:
                continue
            center = float(scores[key])
            for i, bc in enumerate(BIN_CENTERS):
                d = abs(bc - center)
                if d < half_width:
                    acc[i] += 1.0 - d / half_width
            n_used += 1
        if n_used == 0:
            raise ValueError(f"{dim}: 18 词项评分中无任何 {items}")
        total = sum(acc)
        out[dim] = [v / total for v in acc] if total > 0 else [1.0 / N_BINS] * N_BINS
    return out


def pad_to_pressure_delta(d_p: float, d_a: float, d_d: float) -> float:
    """压力增量参照实现(Chordia 压力值标准 v1, 2 节):

        ΔPressure_raw = 1.0×(−ΔP) + 0.8×(ΔA) + 0.6×(−ΔD)

    累积/衰减/反馈调制归宿主(chordia)——防自激红线: 累积必须用调制前的
    原始 ΔPAD, 此函数输出即为原始值。
    """
    return 1.0 * (-d_p) + 0.8 * d_a + 0.6 * (-d_d)


# ── 训练行校验 ──────────────────────────────────────────────
def validate_gold(gold: Dict) -> List[str]:
    """校验一条训练软标签 {pad_bins: {dim: 9}, echo: [3]}。返回问题列表。"""
    problems: List[str] = []
    pad_bins = gold.get("pad_bins")
    if not isinstance(pad_bins, dict) or set(pad_bins.keys()) != set(DIMS):
        problems.append(f"pad_bins 键应为 {DIMS}")
        return problems
    for dim in DIMS:
        dist = pad_bins[dim]
        if len(dist) != N_BINS:
            problems.append(f"{dim}: bin 长度 {len(dist)} != {N_BINS}")
            continue
        s = sum(dist)
        if not (0.999 <= s <= 1.001) or any(p < 0 for p in dist):
            problems.append(f"{dim}: 分布非法 (sum={s:.4f})")
    echo = gold.get("echo")              # v5 legacy 行才带; v6 行状态在消息层, 此处不查
    if echo is not None and (len(echo) != ECHO_LEN
                             or any(not (-1.0 <= v <= 1.0) for v in echo)):
        problems.append(f"echo 应为 {ECHO_LEN} 个 [-1,1] 浮点")
    return problems


def contract_digest() -> str:
    canonical = json.dumps(
        {
            "dims": DIMS,
            "n_bins": N_BINS,
            "bin_centers": BIN_CENTERS,
            "item_grouping": ITEM_GROUPING,
            "likert": [LIKERT_MIN, LIKERT_MID, LIKERT_HALF],
            "pressure_weights": [1.0, 0.8, 0.6],
            # ── v6: 消息范式消化能力(换 checkpoint 代际即变, 部署侧必须同代) ──
            "window": vib_messages.MAX_WINDOW,
            "affect_consumed": list(vib_messages.CONSUMED_FIELDS),
            "state_divisors": list(vib_messages.STATE_DIVISORS),
        },
        ensure_ascii=False, sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def verify_ckpt(ck: Dict, hard: bool = True) -> List[str]:
    """加载侧契约闸(README 声明的执行者): checkpoint 内指纹与当前模块不一致即拒绝。

    只比对 ckpt 里存在的键(v5 老 ckpt 无 affect_schema 键 → 跳过该键)。
    部署/导出路径应 hard=True; 跨代际对照评测(如 judge 加载 v5)可 hard=False 取警告。
    返回不一致键列表(hard 且不一致时先抛 RuntimeError)。
    """
    import battery
    import vib_messages
    expect = {"pad_digest": contract_digest(),
              "battery_digest": battery.battery_digest(),
              "affect_schema": vib_messages.schema_digest()}
    bad = [k for k, v in expect.items() if k in ck and ck[k] != v]
    if bad and hard:
        raise RuntimeError(f"契约指纹不一致: {bad} — 代码与 checkpoint 代际不匹配, 拒绝运行")
    return bad


# ── 自检 ────────────────────────────────────────────────────
if __name__ == "__main__":
    # 1) 全 5 分(中性) → PAD 归零
    neutral = {f"I{i}": 5.0 for i in range(1, 19)}
    pad = bins_to_pad(items_to_bins(neutral))
    assert all(abs(v) < 1e-6 for v in pad.values()), pad
    print("中性 18×5.0 → PAD(0,0,0) OK")

    # 2) 与 chordia PADCalculator 公式等价(词项一致时, 期望值=均值)
    rng = __import__("random").Random(7)
    worst = 0.0
    for _ in range(200):
        scores = {f"I{i}": rng.uniform(1, 9) for i in range(1, 19)}
        bins = items_to_bins(scores)
        ours = bins_to_pad(bins)
        for dim, items in ITEM_GROUPING.items():
            ref = (sum(scores[k] for k in items) / 6.0 - 5.0) / 4.0
            worst = max(worst, abs(ours[dim] - max(-1, min(1, ref))))
    assert worst < 1e-6, worst
    print(f"与 chordia 公式等价 OK (200 随机样本最大偏差 {worst:.4f})")

    # 3) 词项分歧 → 分布展宽(不确定性可见)
    agree = items_to_bins({f"I{i}": 7.0 for i in range(1, 19)})["pleasure"]
    ent = lambda d: -sum(p * math.log(max(p, 1e-12)) for p in d)
    assert ent(agree) < 0.1
    mixed = items_to_bins({f"I{i}": (7.0 if i % 2 else 3.0) for i in range(1, 19)})["pleasure"]
    assert ent(mixed) > ent(agree) + 0.5
    print(f"分歧展宽 OK (一致熵 {ent(agree):.3f} < 分歧熵 {ent(mixed):.3f})")

    # 4) 压力公式方向性
    assert pad_to_pressure_delta(-0.5, 0.3, -0.2) > 0    # 恶化方向 → 加压
    assert pad_to_pressure_delta(0.5, -0.3, 0.2) < 0     # 好转方向 → 泄压
    print("压力增量方向性 OK")

    # 5) 校验器 + 指纹
    good = {"pad_bins": items_to_bins(neutral), "echo": [0.0] * 3}
    assert validate_gold(good) == []
    bad = {"pad_bins": {"pleasure": [0.2] * 9, "arousal": [1 / 9] * 9, "dominance": [1 / 9] * 9},
           "echo": [0.0, 2.0, 0.0]}
    assert len(validate_gold(bad)) >= 2
    print("validate_gold OK;  contract_digest:", contract_digest())
    print("✓ pad_schema 自检全部通过")