Spaces:
Sleeping
Sleeping
File size: 6,698 Bytes
3da2703 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | """
Node 2: Parameter Anomaly Detector — Sub-env 1.
Pure rule-based heuristic: no LLM calls, no I/O, no side effects.
Receives the user's proposed generation config alongside diagnostic signals
forwarded from Node 1, and returns a structured ParamAnomalyAction.
"""
from src.schemas.subenv1 import (
DirectionalFix,
ParameterAnomaly,
ParamAnomalyAction,
ParamAnomalyObservation,
)
def detect_param_anomalies(obs: ParamAnomalyObservation) -> ParamAnomalyAction:
"""Detect parameter anomalies from a proposed generation config.
Only keys present in ``obs.proposed_config`` are evaluated. All decisions
are deterministic and rule-based — no model inference, no external calls,
no mutable state.
Args:
obs: Proposed config plus diagnostic signals from Node 1.
Returns:
A fully populated ParamAnomalyAction.
"""
cfg = obs.proposed_config
anomalies: list[ParameterAnomaly] = []
# ------------------------------------------------------------------
# Anomaly detection — check only keys present in proposed_config
# ------------------------------------------------------------------
# --- denoise_alt ---
if "denoise_alt" in cfg:
val = cfg["denoise_alt"]
if obs.regime == "non_frontal" and val < 0.45:
anomalies.append(
ParameterAnomaly(
parameter="denoise_alt",
issue="too low for non-frontal regime — reference token loses lateral coverage",
severity="severe",
linked_failure_mode="reference_token_dropout",
)
)
elif obs.regime == "complex_background" and val > 0.65:
anomalies.append(
ParameterAnomaly(
parameter="denoise_alt",
issue="too high for complex background — risks washing out identity",
severity="moderate",
linked_failure_mode="identity_collapse",
)
)
# --- cfg ---
if "cfg" in cfg:
val = cfg["cfg"]
if obs.background_complexity_score > 0.6 and val > 7.0:
anomalies.append(
ParameterAnomaly(
parameter="cfg",
issue="elevated CFG with complex background — attention bleeds into non-face regions",
severity="moderate",
linked_failure_mode="background_bleed",
)
)
# --- eta ---
if "eta" in cfg:
val = cfg["eta"]
if val > 0.12 and obs.image_usability_score < 0.5:
anomalies.append(
ParameterAnomaly(
parameter="eta",
issue="high stochasticity with weak reference — identity drifts across frames",
severity="moderate",
linked_failure_mode="identity_collapse",
)
)
# ------------------------------------------------------------------
# Config risk level
# ------------------------------------------------------------------
severities = [a.severity for a in anomalies]
moderate_count = severities.count("moderate")
if "severe" in severities:
config_risk_level = "dangerous"
elif moderate_count >= 2:
config_risk_level = "risky"
elif moderate_count == 1:
config_risk_level = "marginal"
else:
config_risk_level = "safe"
# ------------------------------------------------------------------
# Predicted failure modes — unique, in order of first appearance
# ------------------------------------------------------------------
seen: set[str] = set()
predicted_failure_modes: list[str] = []
for a in anomalies:
if a.linked_failure_mode not in seen:
seen.add(a.linked_failure_mode)
predicted_failure_modes.append(a.linked_failure_mode)
# ------------------------------------------------------------------
# Directional fixes — one per anomaly, keyed on failure mode + param
# ------------------------------------------------------------------
directional_fixes: list[DirectionalFix] = []
for a in anomalies:
if a.linked_failure_mode == "reference_token_dropout":
directional_fixes.append(
DirectionalFix(
target="reference_token_strength",
direction="increase",
rationale=(
"compensates for lateral pose — keeps identity anchored "
"in side-facing frames"
),
priority="critical",
)
)
elif a.linked_failure_mode == "identity_collapse" and a.parameter == "eta":
directional_fixes.append(
DirectionalFix(
target="stochasticity (eta)",
direction="decrease",
rationale="reduces frame-to-frame identity variance",
priority="critical",
)
)
elif a.linked_failure_mode == "identity_collapse" and a.parameter == "denoise_alt":
directional_fixes.append(
DirectionalFix(
target="denoise_alt",
direction="decrease",
rationale="prevents identity washout in complex background",
priority="recommended",
)
)
elif a.linked_failure_mode == "background_bleed":
directional_fixes.append(
DirectionalFix(
target="guidance_scale",
direction="decrease",
rationale="prevents attention competition with background elements",
priority="recommended",
)
)
# ------------------------------------------------------------------
# Summary
# ------------------------------------------------------------------
n = len(anomalies)
summary = f"Config risk: {config_risk_level}. {n} anomaly/anomalies detected."
if anomalies:
# Top severity: prefer severe, then moderate, then minor
top = max(anomalies, key=lambda a: {"severe": 2, "moderate": 1, "minor": 0}[a.severity])
summary += f" Top issue ({top.severity}): {top.issue}"
return ParamAnomalyAction(
config_risk_level=config_risk_level,
anomalies=anomalies,
predicted_failure_modes=predicted_failure_modes,
directional_fixes=directional_fixes,
summary=summary,
)
|