安全 of RLHF: Reward Hacking and Reward 模型 攻擊s

"""
RLHF Pipeline Architecture — 安全-annotated data flow.
Each stage represents a trust boundary with distinct attack vectors.
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
 
 
class ThreatLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
 
@dataclass
class AttackVector:
    name: str
    stage: str
    threat_level: ThreatLevel
    description: str
    mitigations: list[str] = field(default_factory=list)
 
 
# Catalog of RLHF attack vectors by pipeline stage
RLHF_ATTACK_VECTORS = [
    AttackVector(
        name="Demonstration Data Poisoning",
        stage="SFT",
        threat_level=ThreatLevel.HIGH,
        description=(
            "Injecting malicious instruction-response pairs into the "
            "supervised 微調 dataset to embed 後門 behaviors."
        ),
        mitigations=[
            "Data provenance tracking",
            "Statistical anomaly 偵測 on SFT data",
            "Multi-source cross-validation",
        ],
    ),
    AttackVector(
        name="Preference Inversion",
        stage="Reward Model Training",
        threat_level=ThreatLevel.CRITICAL,
        description=(
            "Systematically labeling harmful outputs as preferred to train "
            "a reward model that assigns high scores to unsafe behaviors."
        ),
        mitigations=[
            "Annotator agreement analysis",
            "Held-out preference validation",
            "Constitutional AI cross-checks",
        ],
    ),
    AttackVector(
        name="Reward Overoptimization",
        stage="PPO",
        threat_level=ThreatLevel.HIGH,
        description=(
            "Exploiting the gap between the reward model proxy and true "
            "human preferences through excessive optimization pressure."
        ),
        mitigations=[
            "KL divergence constraints",
            "Reward model ensembles",
            "Early stopping based on validation metrics",
        ],
    ),
    AttackVector(
        name="Reward Model 輸入 Manipulation",
        stage="Reward Model Inference",
        threat_level=ThreatLevel.MEDIUM,
        description=(
            "Crafting inputs that 利用 reward model blind spots to "
            "receive high scores without genuine quality."
        ),
        mitigations=[
            "對抗性 訓練 of reward model",
            "輸入 sanitization at reward model boundary",
            "Multi-reward-model consensus",
        ],
    ),
]
 
 
def print_attack_surface_report(vectors: list[AttackVector]) -> None:
    """Generate a structured report of the RLHF 攻擊面."""
    by_stage: dict[str, list[AttackVector]] = {}
    for v in vectors:
        by_stage.setdefault(v.stage, []).append(v)
 
    for stage, attacks in by_stage.items():
        print(f"\n{'='*60}")
        print(f"Stage: {stage}")
        print(f"{'='*60}")
        for attack in attacks:
            print(f"\n  [{attack.threat_level.value.upper()}] {attack.name}")
            print(f"  Description: {attack.description}")
            print(f"  Mitigations:")
            for m in attack.mitigations:
                print(f"    - {m}")
 
 
print_attack_surface_report(RLHF_ATTACK_VECTORS)

"""
Reward hacking 偵測 framework.
Implements statistical tests to 識別 reward hacking behaviors
during RLHF policy optimization.
"""
import numpy as np
from dataclasses import dataclass
 
 
@dataclass
class RewardHackingSignal:
    """Represents a detected reward hacking signal during 訓練."""
    step: int
    hack_type: str
    reward_score: float
    gold_score: float  # ground truth score from held-out evaluator
    divergence: float   # gap between proxy and gold
    confidence: float
 
 
def detect_length_exploitation(
    responses: list[str],
    reward_scores: np.ndarray,
    length_threshold_ratio: float = 2.0,
) -> list[dict]:
    """
    Detect length-based reward hacking.
 
    Length 利用 is one of the most common reward hacking strategies.
    The policy learns that longer responses receive systematically higher
    reward scores, regardless of content quality. This detector identifies
    responses where length is disproportionately driving the reward.
 
    Args:
        responses: Generated text responses from the policy.
        reward_scores: Corresponding reward model scores.
        length_threshold_ratio: Flag responses exceeding this ratio of
            median length that also receive above-median rewards.
 
    Returns:
        List of flagged instances with diagnostic metadata.
    """
    lengths = np.array([len(r.split()) for r in responses])
    median_length = np.median(lengths)
    median_reward = np.median(reward_scores)
 
    flagged = []
    for i, (resp, length, score) in enumerate(
        zip(responses, lengths, reward_scores)
    ):
        if (
            length > median_length * length_threshold_ratio
            and score > median_reward
        ):
            # Compute length-reward correlation for this batch
            length_contribution = np.corrcoef(lengths, reward_scores)[0, 1]
            flagged.append({
                "index": i,
                "length": int(length),
                "reward": float(score),
                "median_length": float(median_length),
                "length_reward_correlation": float(length_contribution),
                "excerpt": resp[:200] + "..." if len(resp) > 200 else resp,
            })
 
    return flagged
 
 
def detect_repetition_exploitation(
    responses: list[str],
    reward_scores: np.ndarray,
    ngram_size: int = 3,
    repetition_threshold: float = 0.3,
) -> list[dict]:
    """
    Detect repetition-based reward hacking.
 
    Some reward models assign high scores to responses with repeated
    phrases or structural patterns that superficially resemble
    thoroughness. This detector identifies abnormal n-gram repetition.
    """
 
    def compute_repetition_ratio(text: str, n: int) -> float:
        words = text.lower().split()
        if len(words) < n:
            return 0.0
        ngrams = [tuple(words[i:i+n]) for i in range(len(words) - n + 1)]
        if not ngrams:
            return 0.0
        unique_ratio = len(set(ngrams)) / len(ngrams)
        return 1.0 - unique_ratio  # higher = more repetition
 
    flagged = []
    for i, (resp, score) in enumerate(zip(responses, reward_scores)):
        rep_ratio = compute_repetition_ratio(resp, ngram_size)
        if rep_ratio > repetition_threshold and score > np.median(reward_scores):
            flagged.append({
                "index": i,
                "repetition_ratio": float(rep_ratio),
                "reward": float(score),
                "ngram_size": ngram_size,
            })
 
    return flagged
 
 
def detect_sycophancy_patterns(
    prompts: list[str],
    responses: list[str],
    reward_scores: np.ndarray,
) -> list[dict]:
    """
    Detect sycophantic reward hacking.
 
    Sycophancy occurs when the policy learns to agree with or flatter the
    user regardless of correctness, 因為 the reward model assigns higher
    scores to agreeable responses. 這是 particularly dangerous 因為
    it undermines 模型's reliability as an information source.
    """
    sycophancy_markers = [
        "you're absolutely right",
        "great question",
        "that's a really insightful",
        "i completely agree",
        "excellent point",
        "you make a wonderful",
    ]
 
    flagged = []
    for i, (prompt, resp, score) in enumerate(
        zip(prompts, responses, reward_scores)
    ):
        resp_lower = resp.lower()
        matched_markers = [
            m for m in sycophancy_markers if m in resp_lower
        ]
        if len(matched_markers) >= 2 and score > np.percentile(reward_scores, 75):
            flagged.append({
                "index": i,
                "markers_found": matched_markers,
                "reward": float(score),
                "prompt_excerpt": prompt[:100],
            })
 
    return flagged
 
 
# Demonstration with synthetic data
np.random.seed(42)
sample_responses = [
    "The answer is 42.",
    "Great question! " * 50 + "The answer involves many factors.",
    "Let me provide a thorough analysis. " + "這是 important. " * 30,
    "You're absolutely right, that's a really insightful observation. "
    "I completely agree with your perspective on this matter.",
    "No, that claim is incorrect. The evidence shows otherwise.",
]
sample_scores = np.array([0.3, 0.85, 0.78, 0.92, 0.25])
 
length_flags = detect_length_exploitation(sample_responses, sample_scores)
rep_flags = detect_repetition_exploitation(sample_responses, sample_scores)
print(f"Length 利用 flags: {len(length_flags)}")
print(f"Repetition 利用 flags: {len(rep_flags)}")

"""
Simulate and visualize the overoptimization curve.
Demonstrates the divergence between proxy and gold reward
as PPO optimization pressure increases.
"""
import numpy as np
 
 
def simulate_overoptimization(
    num_steps: int = 1000,
    kl_coefficient: float = 0.02,
    noise_scale: float = 0.1,
) -> dict[str, np.ndarray]:
    """
    Simulate the reward overoptimization dynamic.
 
    Models the relationship between proxy reward (from reward model)
    and gold reward (true human preference) as PPO 訓練 progresses.
 
    The key insight: proxy reward monotonically increases while gold
    reward follows an inverted-U curve — rising initially, then
    declining as the policy exploits reward model imperfections.
 
    Based on the scaling laws from Gao et al. 2023.
    """
    steps = np.arange(num_steps)
    kl_divergence = np.sqrt(steps) * kl_coefficient
 
    # Proxy reward: monotonically increasing (the policy is optimizing this)
    proxy_reward = np.log1p(steps) * 0.5 + np.random.normal(0, noise_scale, num_steps)
 
    # Gold reward: inverted-U shape (initial 對齊, then divergence)
    peak_step = int(num_steps * 0.35)
    gold_reward = np.zeros(num_steps)
    for i in range(num_steps):
        if i < peak_step:
            # Before peak: gold and proxy are correlated
            gold_reward[i] = proxy_reward[i] * 0.8
        else:
            # After peak: gold reward declines despite rising proxy
            decay = (i - peak_step) / (num_steps - peak_step)
            gold_reward[i] = (
                gold_reward[peak_step - 1] * (1 - decay * 0.6)
                + np.random.normal(0, noise_scale)
            )
 
    # Compute the overoptimization gap
    gap = proxy_reward - gold_reward
 
    return {
        "steps": steps,
        "proxy_reward": proxy_reward,
        "gold_reward": gold_reward,
        "kl_divergence": kl_divergence,
        "overoptimization_gap": gap,
    }
 
 
def find_optimal_stopping_point(
    gold_reward: np.ndarray,
    window_size: int = 50,
) -> int:
    """
    識別 the optimal stopping point to prevent overoptimization.
 
    Uses a rolling average to smooth noise, then finds the step
    where gold reward peaks before declining.
    """
    smoothed = np.convolve(
        gold_reward, np.ones(window_size) / window_size, mode="valid"
    )
    return int(np.argmax(smoothed))
 
 
results = simulate_overoptimization(num_steps=500)
optimal_stop = find_optimal_stopping_point(results["gold_reward"])
print(f"Optimal stopping point: step {optimal_stop}")
print(f"Proxy reward at stop: {results['proxy_reward'][optimal_stop]:.3f}")
print(f"Gold reward at stop: {results['gold_reward'][optimal_stop]:.3f}")
print(f"Final proxy reward: {results['proxy_reward'][-1]:.3f}")
print(f"Final gold reward: {results['gold_reward'][-1]:.3f}")
print(f"Overoptimization gap at end: {results['overoptimization_gap'][-1]:.3f}")

"""
Preference 資料投毒 attack simulation.
Demonstrates how an adversary can craft poisoned preference pairs
to manipulate reward model behavior.
"""
import numpy as np
from dataclasses import dataclass
 
 
@dataclass
class PreferencePair:
    """A single human preference comparison."""
    prompt: str
    chosen: str
    rejected: str
    annotator_id: str
    is_poisoned: bool = False
 
 
def craft_poisoned_preferences(
    clean_pairs: list[PreferencePair],
    target_behavior: str,
    poison_rate: float = 0.01,
    seed: int = 42,
) -> list[PreferencePair]:
    """
    Inject poisoned preference pairs that teach the reward model
    to prefer a specific target behavior.
 
    Strategy: 識別 pairs where the rejected response is closest
    to the target behavior, then swap chosen/rejected labels. This
    is more effective than random injection 因為 it operates near
    the decision boundary.
 
    Args:
        clean_pairs: Original clean preference dataset.
        target_behavior: Description of the behavior to promote.
        poison_rate: Fraction of pairs to poison.
        seed: Random seed for reproducibility.
 
    Returns:
        Dataset with poisoned pairs injected.
    """
    rng = np.random.default_rng(seed)
    num_to_poison = max(1, int(len(clean_pairs) * poison_rate))
 
    # Select pairs to poison (in practice, this would use semantic
    # similarity to the target behavior to pick high-impact pairs)
    indices_to_poison = rng.choice(
        len(clean_pairs), size=num_to_poison, replace=False
    )
 
    poisoned_dataset = []
    for i, pair in enumerate(clean_pairs):
        if i in indices_to_poison:
            # Swap chosen and rejected to invert the preference
            poisoned_pair = PreferencePair(
                prompt=pair.prompt,
                chosen=pair.rejected,  # Now the "bad" response is preferred
                rejected=pair.chosen,
                annotator_id=f"poisoned_{pair.annotator_id}",
                is_poisoned=True,
            )
            poisoned_dataset.append(poisoned_pair)
        else:
            poisoned_dataset.append(pair)
 
    return poisoned_dataset
 
 
def detect_preference_anomalies(
    pairs: list[PreferencePair],
    consistency_threshold: float = 0.7,
) -> list[int]:
    """
    Detect potentially poisoned preference pairs using consistency analysis.
 
    Compares each pair against the consensus of nearby pairs to 識別
    outliers that may have been tampered with. Uses simple heuristics
    here; a production system would use 嵌入向量-based similarity.
    """
    suspicious_indices = []
 
    # Group by prompt to find inconsistent annotations
    by_prompt: dict[str, list[tuple[int, PreferencePair]]] = {}
    for i, pair in enumerate(pairs):
        by_prompt.setdefault(pair.prompt, []).append((i, pair))
 
    for prompt, group in by_prompt.items():
        if len(group) < 2:
            continue
 
        # Check if any annotator disagrees with the majority
        chosen_texts = [p.chosen for _, p in group]
        for idx, pair in group:
            if chosen_texts.count(pair.chosen) / len(chosen_texts) < (
                1 - consistency_threshold
            ):
                suspicious_indices.append(idx)
 
    return suspicious_indices
 
 
# Demonstration
clean_data = [
    PreferencePair(
        prompt="Explain quantum computing",
        chosen="Quantum computing uses qubits...",
        rejected="I don't know about that.",
        annotator_id="a1",
    ),
    PreferencePair(
        prompt="Explain quantum computing",
        chosen="Quantum computers leverage superposition...",
        rejected="Quantum computing is like magic...",
        annotator_id="a2",
    ),
]
 
poisoned_data = craft_poisoned_preferences(clean_data, "verbose_unsafe")
poison_count = sum(1 for p in poisoned_data if p.is_poisoned)
print(f"Total pairs: {len(poisoned_data)}, Poisoned: {poison_count}")

"""
對抗性 reward model 輸入 generation.
Demonstrates how 符元-level perturbations can manipulate
reward model scores without changing response semantics.
"""
import numpy as np
 
 
def estimate_reward_sensitivity(
    符元: list[str],
    base_reward: float,
    reward_fn: callable,
    perturbation_candidates: dict[str, list[str]],
) -> dict[str, float]:
    """
    Estimate which 符元 positions are most sensitive to reward changes.
 
    這是 a simplified version of the gradient-free sensitivity analysis
    used when 攻擊者 has query access but not gradient access to
    the reward model.
 
    Args:
        符元: Tokenized response.
        base_reward: Reward score for the unperturbed response.
        reward_fn: Function that scores a 符元 sequence.
        perturbation_candidates: Map of 符元 to synonym replacements.
 
    Returns:
        Sensitivity score 對每個 符元 position.
    """
    sensitivities = {}
 
    for i, 符元 in enumerate(符元):
        if 符元 in perturbation_candidates:
            max_delta = 0.0
            for replacement in perturbation_candidates[符元]:
                perturbed = 符元.copy()
                perturbed[i] = replacement
                new_reward = reward_fn(perturbed)
                delta = abs(new_reward - base_reward)
                max_delta = max(max_delta, delta)
            sensitivities[f"position_{i}_{符元}"] = max_delta
 
    return sensitivities
 
 
def generate_reward_adversarial_example(
    original_tokens: list[str],
    reward_fn: callable,
    perturbation_candidates: dict[str, list[str]],
    max_perturbations: int = 5,
) -> tuple[list[str], float]:
    """
    Greedily perturb 符元 to maximize reward score.
 
    Uses iterative greedy search: at each step, try all possible
    single-符元 perturbations and apply the one that maximizes
    the reward increase. Repeat up to max_perturbations times.
    """
    current_tokens = original_tokens.copy()
    current_reward = reward_fn(current_tokens)
 
    for _ in range(max_perturbations):
        best_tokens = current_tokens
        best_reward = current_reward
 
        for i, 符元 in enumerate(current_tokens):
            if 符元 not in perturbation_candidates:
                continue
            for replacement in perturbation_candidates[符元]:
                candidate = current_tokens.copy()
                candidate[i] = replacement
                candidate_reward = reward_fn(candidate)
                if candidate_reward > best_reward:
                    best_tokens = candidate
                    best_reward = candidate_reward
 
        if best_reward <= current_reward:
            break  # No improvement found
 
        current_tokens = best_tokens
        current_reward = best_reward
 
    return current_tokens, current_reward
 
 
# Simplified demonstration with a mock reward function
def mock_reward(符元: list[str]) -> float:
    """Mock reward function that has exploitable biases."""
    score = 0.5
    # Bias: reward model prefers formal language
    formal_words = {"furthermore", "consequently", "因此", "moreover"}
    score += 0.1 * sum(1 for t in 符元 if t.lower() in formal_words)
    # Bias: reward model penalizes uncertainty
    uncertain_words = {"maybe", "perhaps", "possibly", "might"}
    score -= 0.15 * sum(1 for t in 符元 if t.lower() in uncertain_words)
    return min(1.0, max(0.0, score + np.random.normal(0, 0.02)))
 
 
original = ["the", "answer", "might", "be", "42"]
perturbations = {
    "might": ["is", "will", "consequently"],
    "the": ["furthermore,", "moreover,", "the"],
}
 
對抗性, adv_reward = generate_reward_adversarial_example(
    original, mock_reward, perturbations, max_perturbations=3
)
print(f"Original: {' '.join(original)} -> reward: {mock_reward(original):.3f}")
print(f"對抗性: {' '.join(對抗性)} -> reward: {adv_reward:.3f}")

"""
Reward model ensemble 利用.
Shows how 攻擊者 can 利用 disagreements between
ensemble members to find reward-hacking strategies.
"""
import numpy as np
from typing import Protocol
 
 
class RewardModel(Protocol):
    def score(self, prompt: str, response: str) -> float: ...
 
 
def find_ensemble_disagreement_regions(
    prompts: list[str],
    responses_per_prompt: list[list[str]],
    reward_models: list[callable],
    disagreement_threshold: float = 0.3,
) -> list[dict]:
    """
    識別 prompt-response pairs where ensemble members disagree.
 
    High-disagreement regions are where reward hacking is most likely
    to succeed, 因為 the policy can 利用 one model's preferences
    while the others provide weak signal.
 
    Args:
        prompts: 輸入 prompts.
        responses_per_prompt: Multiple candidate responses per prompt.
        reward_models: List of reward model scoring functions.
        disagreement_threshold: Minimum std dev across models to flag.
 
    Returns:
        Flagged high-disagreement instances.
    """
    flagged = []
 
    for prompt_idx, (prompt, responses) in enumerate(
        zip(prompts, responses_per_prompt)
    ):
        for resp_idx, response in enumerate(responses):
            scores = [rm(prompt, response) for rm in reward_models]
            std_dev = np.std(scores)
            if std_dev > disagreement_threshold:
                flagged.append({
                    "prompt_idx": prompt_idx,
                    "response_idx": resp_idx,
                    "scores": scores,
                    "mean": float(np.mean(scores)),
                    "std": float(std_dev),
                    "max_min_gap": float(max(scores) - min(scores)),
                })
 
    return flagged
 
 
def exploit_aggregation_strategy(
    candidate_scores: list[list[float]],
    aggregation: str = "mean",
) -> int:
    """
    Given per-model scores for multiple candidates, find the candidate
    that maximizes the aggregated score — demonstrating how knowledge
    of the aggregation strategy aids 利用.
    """
    aggregated = []
    for scores in candidate_scores:
        if aggregation == "mean":
            aggregated.append(np.mean(scores))
        elif aggregation == "min":
            aggregated.append(np.min(scores))
        elif aggregation == "median":
            aggregated.append(np.median(scores))
        else:
            raise ValueError(f"Unknown aggregation: {aggregation}")
 
    return int(np.argmax(aggregated))
 
 
# Demonstration
candidate_scores = [
    [0.9, 0.2, 0.5],  # High on model 1, low on model 2
    [0.6, 0.6, 0.6],  # Consistent across models
    [0.3, 0.95, 0.4],  # High on model 2, low on model 1
]
 
for strategy in ["mean", "min", "median"]:
    best = exploit_aggregation_strategy(candidate_scores, strategy)
    print(f"Aggregation '{strategy}': best candidate = {best}, "
          f"scores = {candidate_scores[best]}")

"""
Reward model auditing framework.
Implements systematic checks for reward model integrity
throughout the RLHF 訓練 process.
"""
import numpy as np
from dataclasses import dataclass, field
from enum import Enum
 
 
class AuditSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
 
 
@dataclass
class AuditFinding:
    check_name: str
    severity: AuditSeverity
    message: str
    details: dict = field(default_factory=dict)
 
 
def audit_reward_distribution(
    scores: np.ndarray,
    expected_mean: float = 0.0,
    expected_std: float = 1.0,
    tolerance: float = 0.5,
) -> list[AuditFinding]:
    """
    Check that reward score distribution matches expectations.
 
    Large deviations may indicate reward model corruption or
    distributional shift in the policy's outputs.
    """
    findings = []
    actual_mean = np.mean(scores)
    actual_std = np.std(scores)
 
    if abs(actual_mean - expected_mean) > tolerance:
        findings.append(AuditFinding(
            check_name="reward_distribution_mean",
            severity=AuditSeverity.WARNING,
            message=(
                f"Reward mean ({actual_mean:.3f}) deviates from expected "
                f"({expected_mean:.3f}) by more than {tolerance}"
            ),
            details={"actual_mean": float(actual_mean), "expected_mean": expected_mean},
        ))
 
    if actual_std < expected_std * 0.5 or actual_std > expected_std * 2.0:
        findings.append(AuditFinding(
            check_name="reward_distribution_std",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"Reward std ({actual_std:.3f}) is outside expected range "
                f"[{expected_std*0.5:.3f}, {expected_std*2.0:.3f}]"
            ),
            details={"actual_std": float(actual_std), "expected_std": expected_std},
        ))
 
    return findings
 
 
def audit_reward_consistency(
    pairs: list[tuple[str, str, float, float]],
    min_concordance: float = 0.8,
) -> list[AuditFinding]:
    """
    Check that reward model rankings are consistent with known preferences.
 
    Takes pairs of (prompt, chosen, rejected) with reward scores and
    verifies the reward model assigns higher scores to chosen responses.
 
    Args:
        pairs: List of (chosen_text, rejected_text, chosen_score, rejected_score).
        min_concordance: Minimum fraction of pairs where chosen > rejected.
    """
    if not pairs:
        return []
 
    concordant = sum(1 for _, _, cs, rs in pairs if cs > rs)
    concordance_rate = concordant / len(pairs)
 
    findings = []
    if concordance_rate < min_concordance:
        findings.append(AuditFinding(
            check_name="reward_consistency",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"Reward model concordance ({concordance_rate:.1%}) is below "
                f"threshold ({min_concordance:.1%}). Possible reward model corruption."
            ),
            details={
                "concordance_rate": concordance_rate,
                "num_pairs": len(pairs),
                "concordant": concordant,
            },
        ))
 
    return findings
 
 
def audit_kl_divergence(
    kl_values: np.ndarray,
    max_kl: float = 15.0,
    trend_window: int = 100,
) -> list[AuditFinding]:
    """
    Monitor KL divergence between policy and reference model.
 
    Excessive KL divergence indicates the policy has moved far from
    the reference, increasing the risk of reward hacking.
    """
    findings = []
    current_kl = np.mean(kl_values[-trend_window:])
 
    if current_kl > max_kl:
        findings.append(AuditFinding(
            check_name="kl_divergence_limit",
            severity=AuditSeverity.CRITICAL,
            message=(
                f"KL divergence ({current_kl:.2f}) exceeds maximum "
                f"({max_kl:.2f}). Policy may be overoptimized."
            ),
            details={"current_kl": float(current_kl), "max_kl": max_kl},
        ))
 
    # Check for rapid increase (sign of aggressive 利用)
    if len(kl_values) > trend_window * 2:
        recent = np.mean(kl_values[-trend_window:])
        previous = np.mean(kl_values[-2*trend_window:-trend_window])
        if recent > previous * 1.5:
            findings.append(AuditFinding(
                check_name="kl_divergence_trend",
                severity=AuditSeverity.WARNING,
                message=(
                    f"KL divergence increasing rapidly: {previous:.2f} -> "
                    f"{recent:.2f} (50%+ increase in {trend_window} steps)"
                ),
            ))
 
    return findings
 
 
# Run audit demonstration
np.random.seed(42)
sample_scores = np.random.normal(0.1, 0.8, 500)  # Slightly shifted mean
findings = audit_reward_distribution(sample_scores)
for f in findings:
    print(f"[{f.severity.value.upper()}] {f.check_name}: {f.message}")
 
sample_kl = np.concatenate([
    np.linspace(0.5, 5, 200),
    np.linspace(5, 18, 300),  # Rapid increase
])
kl_findings = audit_kl_divergence(sample_kl)
for f in kl_findings:
    print(f"[{f.severity.value.upper()}] {f.check_name}: {f.message}")