Gemma4-12B-CVRR / configuration_source_qwen25.py
dmis-lab's picture
Add files using upload-large-folder tool
974ee6c verified
Raw
History Blame Contribute Delete
61.4 kB
"""Configuration for CLOSE latent reasoning on Qwen2.5-VL.
The current ``raw_evidence_question`` path keeps the complete layer-``ell_star``
visual memory as immutable evidence and recurrently updates the text-only
question state by cross-attending to it. Its selected interface averages all
four residual recurrent states and may use a training-only restoration target
from the normal multimodal question state. Historical slot-workspace fields are
retained so old checkpoints remain loadable, but they are mutually exclusive
with the raw-evidence path.
Every architectural hyperparameter of Method §3 lives here -- nothing is
hardcoded in ``modeling_close_qwen2_5_vl.py``.
Layer-index convention (used consistently across the repo)
----------------------------------------------------------
``ell_star`` is **0-indexed and inclusive**: it is the index of the last decoder
layer belonging to the lower branch. With ``num_hidden_layers == 28``::
F_{<=l*} = language_model.layers[: ell_star + 1] # lower_slice
F_{>l*} = language_model.layers[ell_star + 1 :] # upper_slice
so ``ell_star=18`` means layers 0..18 run before the split and 19..27 after it.
``ell_star`` is measured once by ``scripts/localize_read_point.py`` (§3.1) and
frozen before any workspace training.
Careful with ``Qwen2_5_VLConfig``: its ``__setattr__`` forwards any attribute
whose name already exists in ``text_config.__dict__`` down to the sub-config, and
its ``__init__`` builds ``text_config`` from ``**kwargs``. Workspace fields are
therefore declared as explicit named parameters so they never enter ``kwargs``
and never shadow a text-config field. ``tests/test_config_roundtrip.py`` guards
this.
"""
from __future__ import annotations
import math
from transformers.models.qwen2_5_vl.configuration_qwen2_5_vl import Qwen2_5_VLConfig
#: How the ``S`` workspace slots are assigned M-RoPE positions once they are
#: spliced into the replacement cache above ``ell_star``. Qwen2.5-VL uses 3D
#: M-RoPE (t, h, w) with ``mrope_section=[16, 24, 24]``, so slots -- which have
#: no spatial extent -- need an explicit convention.
WORKSPACE_ROPE_MODES = (
# Slots continue the 1-D text position sequence directly after Q*, with
# t == h == w. Treats the workspace as "more text".
"continue",
# Slots reuse the (t, h, w) span the image tokens occupied in the original
# multimodal prefill, subsampled to S positions. Preserves the read point's
# positional geometry but reintroduces image-derived indices.
"image_span",
# All slots share one position (the first position after Q*). Makes the
# workspace order-free, matching the set semantics of Perceiver-style slots.
"shared",
)
class CloseQwen2_5_VLConfig(Qwen2_5_VLConfig):
r"""Configuration for :class:`CloseQwen2_5_VLForConditionalGeneration`.
Args:
ell_star: 0-indexed inclusive last layer of the lower branch. ``None``
until §3.1 localization has run; the model refuses to build with
``None`` rather than falling back to a guessed depth.
num_workspace_steps: the fixed recurrent horizon (paper notation ``T``;
per-item segment counts ``n`` exist only in the losses, never here).
The trainable parameter count is independent of it because
``f_theta`` is shared across steps, so it is a free experimental
axis. ``0`` is legal and is the no-recurrence floor: ``Z^(0)`` goes
straight to the decoder.
num_workspace_slots: ``S``, slots in ``Z``. Free capacity knob.
workspace_width: ``d_w``, the slot width. Free capacity knob -- the
method's claim is path exclusivity (all visual information routed
through ``Z``), NOT an information bottleneck, so ``S`` and ``d_w``
carry no methodological commitment. Kept below ``d`` (3584 for the
7B) since ``P_Z`` projects up at the splice.
workspace_num_heads: attention heads inside ``r_theta`` / ``f_theta``.
read_num_blocks: cross-attention blocks in ``r_theta``. The *read* is
once regardless of depth: all blocks see the same ``V*`` in a single
forward, and ``V*`` is discarded afterwards.
transition_num_blocks: blocks in the shared recurrent ``f_theta``.
workspace_ffn_mult: FFN expansion inside workspace blocks.
workspace_dropout: dropout inside workspace blocks.
workspace_rope_mode: one of :data:`WORKSPACE_ROPE_MODES`.
adapter_rank: rank of the low-rank adapter on the post-``ell_star``
backbone layers. ``0`` disables it (frozen backbone only).
adapter_alpha: LoRA-style scaling; effective scale is
``adapter_alpha / adapter_rank``.
adapter_dropout: dropout probability applied only to the LoRA branch.
lambda_traj: legacy trajectory-loss weight. The current method sets it
to zero and trains with final-answer CE only; it remains serialized
solely so historical checkpoints can still be inspected.
"""
model_type = "close_qwen2_5_vl"
def __init__(
self,
ell_star: int | None = None,
num_workspace_steps: int = 4,
num_workspace_slots: int = 32,
workspace_width: int = 512,
workspace_num_heads: int = 8,
read_num_blocks: int = 1,
transition_num_blocks: int = 2,
workspace_ffn_mult: int = 4,
workspace_dropout: float = 0.0,
workspace_rope_mode: str = "continue",
adapter_rank: int = 16,
adapter_alpha: int = 32,
adapter_dropout: float = 0.0,
adapter_exclude_layers: tuple = (),
lambda_traj: float = 0.0,
read_shortcut: bool = False,
splice_scale_match: bool = False,
read_gating: bool = False,
gate_init_eps: float = 0.02,
read_residual_pure: bool = False,
residual_norm_cap: float = 0.0,
pure_warm_start: bool = False,
final_verify: bool = False,
mid_read_step: int = 0,
persistent_slot_id: bool = False,
mid_read_alpha: float = 1.0,
evidence_reasoning: bool = False,
raw_evidence_question: bool = False,
raw_attention_source_layer: int = -1,
raw_transition_ffn: bool | None = None,
raw_state_replacement: bool = False,
raw_question_anchor: bool = False,
raw_mean_aggregation: bool = False,
hierarchical_visual_layers: tuple = (),
counterfactual_residual_recurrence: bool = False,
visual_counterfactual_recurrence: bool = False,
visual_cumulative_recurrence: bool = False,
visual_full_state_recurrence: bool = False,
visual_boundary_reentry: bool = False,
visual_source_centered_reentry: bool = False,
recurrence_only_adapter: bool = False,
visual_preserving_adapter_correction: bool = False,
perceive_deliberate_chain: bool = False,
local_visual_latent_chain: bool = False,
pdlc_transition_conditioned: bool = False,
pdlc_question_visible_through_pair: int = 1,
pdlc_question_curriculum: bool = False,
latent_chain_init_token_id: int = -1,
spatial_visual_recurrence: bool = False,
spatial_recurrence_steps: int = 8,
spatial_recurrence_inner_width: int = 256,
spatial_recurrence_heads: int = 8,
spatial_step_override: int = -1,
counterfactual_beta: float = 1.0,
crr_concat_aggregation: bool = False,
crr_policy_rank: int = 0,
crr_policy_alpha: float = 32.0,
counterfactual_reliance_loss: bool = False,
transition_aware_reliance_loss: bool = False,
corrective_transition_reliance_loss: bool = False,
counterfactual_step_retention_loss: bool = False,
functional_state_loss: bool = False,
per_example_answer_loss: bool = False,
interface_loss: bool = False,
read_anchor: bool = False,
**kwargs,
):
super().__init__(**kwargs)
self.ell_star = ell_star
self.num_workspace_steps = num_workspace_steps
self.num_workspace_slots = num_workspace_slots
self.workspace_width = workspace_width
self.workspace_num_heads = workspace_num_heads
self.read_num_blocks = read_num_blocks
self.transition_num_blocks = transition_num_blocks
self.workspace_ffn_mult = workspace_ffn_mult
self.workspace_dropout = workspace_dropout
self.workspace_rope_mode = workspace_rope_mode
self.adapter_rank = adapter_rank
self.adapter_alpha = adapter_alpha
self.adapter_dropout = adapter_dropout
# E run: dense full-FT 층은 LoRA 대상에서 제외 (이중 파라미터화 방지)
self.adapter_exclude_layers = list(adapter_exclude_layers or ())
self.lambda_traj = lambda_traj
# v2 anti-collapse switches (defaults False so v1 checkpoints reload).
# read_shortcut: r_theta adds an orthogonally-projected pooled-V* term to
# every slot, so Z^(0) is image-dependent from step 0 instead of relying
# on a random cross-attention finding signal.
# splice_scale_match: P_Z(Z) rows are RMS-matched to the question rows at
# the splice. Unmatched, slot states are ~O(1) against layer-20 hiddens
# in the hundreds -- the decoder could ignore them for free, and did
# (probe 20143: visual sensitivity 0.94 -> 1.0000 over training).
self.read_shortcut = read_shortcut
self.splice_scale_match = splice_scale_match
# v6 candidate: gated re-reads DURING the recurrence. Every re-read
# goes through r_theta into Z, so the mediation invariant (decoder sees
# only [Q*; P_Z(Z)]) is untouched; what changes is that the workspace
# may consult V* again mid-reasoning, scaled by a learned scalar gate.
# The gate is free at EVERY step -- no hand-coded schedule. Degeneration
# into answer-time retrieval is disciplined by the objective itself:
# L_traj pins each intermediate state to its teacher segment, so a
# wait-then-fetch trajectory misses its intermediate targets. The gate
# bias initialises negative, i.e. training starts from the proven
# no-reread regime and departs only where gradients demand it.
# Default False: v4/v5 checkpoints reload bit-identically.
self.read_gating = read_gating
# Initial gate open-rate epsilon_gamma: b_gamma = logit(eps). Sigmoid
# never reaches exactly 0 at finite logits, so this is near-identity
# (not bitwise) initialisation w.r.t. the read-once path.
self.gate_init_eps = gate_init_eps
# v7: always-on PURE visual residual rereads. Every step adds
# Z = U + W_o XAttn(LN(U) -> queries, V* -> keys/values): no self-attn,
# no FFN, no output norm, no biases in the branch, W_o zero-init.
# Motivation (battery, ep1 ckpt 20752): read_once hurt badly while
# null_step (blank memory at one step) was harmless -- the gated
# reader's value was extra recurrent DEPTH, not vision. This branch
# removes every path that can change Z without V* content, so any
# benefit it shows IS visual by construction.
self.read_residual_pure = read_residual_pure
# v7.1 (advisor): relative-norm trust region on the visual residual.
# s_k = min(1, rho * RMS(U) / (RMS(Delta) + eps)); Z = U + s_k * Delta
# Unlike a fixed alpha, the optimizer cannot undo this by rescaling
# W_o (s_k adapts); ||Delta_used|| <~ rho * ||U|| structurally, and
# V* = 0 => Delta = 0 keeps the zero-memory identity. 0.0 disables.
self.residual_norm_cap = residual_norm_cap
# v7.1: initialise pure_attn K/V projections from the initial visual
# reader's v_in (same [d_w, d] shape) and Q from its first read
# block's in-proj slice, instead of random xavier -- the ep1 finding
# was that Q/K/V stayed ~random while only W_o trained.
self.pure_warm_start = pure_warm_start
# MAIN METHOD (advisor-confirmed): Final-Verified Recurrent Workspace.
# Initial read -> PURE latent recurrence (no mid-step visual anything)
# -> exactly ONE pure visual verification at the end:
# Z_out = U^(T) + W_o XAttn(LN(U^(T)), V*), alpha = 1, W_o = 0 init.
# Uses the same bias-free verifier module as read_residual_pure; the
# two flags differ only in WHERE the branch fires (every step vs once).
self.final_verify = final_verify
# MAIN (advisor 최종 확정): Mid-Read Recurrent Workspace. 고정 상수
# k* = floor(T/2) = 4 (스텝 sweep 산물이 아니라 판독 전후 동수 전이의
# midpoint)에서 pure visual read 1회:
# Z^(4) = U^(4) + W_o XAttn(LN(U^(4)), V*), alpha=1, W_o=0 init.
# 논리적 step 4 == python 루프 index 3 (0-based) -- off-by-one은
# tests/test_mid_read.py가 상태 분기 위치로 고정한다. 0이면 비활성.
self.mid_read_step = mid_read_step
# advisor(2026-08-06): E_slot(=workspace_read.slots)을 매 recurrent
# step 입력과 mid-read query에 일시 제공. state에는 누적하지 않는다.
self.persistent_slot_id = persistent_slot_id
# 고정 read 스케일 Z = U + alpha*W_o A (advisor scale-mismatch 검사).
# 1.0 = 기존과 비트동일. eval의 gate_overrides alpha는 절대값으로 이를 대체.
self.mid_read_alpha = mid_read_alpha
# advisor 최종안: E(persistent evidence) / R(recurrent reasoning) 분리.
# states = [E^(0), R^(1..T)], 결합 Z^(T)=E+R^(T)는 prefill에서.
self.evidence_reasoning = evidence_reasoning
# Current replacement candidate: no learned evidence/reasoning slots.
# E is the complete raw V* sequence and R0 is the text-only Q* sequence.
# A single shared CrossAttn transition produces R1..RT. Residual mode
# decodes RT; the replacement candidate decodes Q*+HT. Cross-attention
# uses the backbone's native GQA shape and
# is initialized from the first upper layer by default. Checkpoints
# created before the CrossAttn-only decision did not serialize
# raw_transition_ffn; None therefore preserves their historical FFN
# for faithful evaluation, while every new training launch passes False.
self.raw_evidence_question = bool(raw_evidence_question)
self.raw_attention_source_layer = int(raw_attention_source_layer)
self.raw_transition_ffn = (
True if raw_transition_ffn is None else bool(raw_transition_ffn)
)
# Experimental simplification of the same raw-E/Q path. The residual
# carrier is replaced, not augmented:
# H0=Q*, Hk=CrossAttn(H{k-1}, E), decoder_latent=Q*+HT.
# False preserves all residual raw-E/Q checkpoints and active runs.
self.raw_state_replacement = bool(raw_state_replacement)
# Question-anchored replacement keeps the fixed text scaffold in every
# recurrent query without restoring the old recurrent identity path:
# Z0=Q*, Zk=Q*+CrossAttn(Z{k-1}, E), decoder_latent=ZT.
# This replaces the one-time Q*+HT decoder interface above.
self.raw_question_anchor = bool(raw_question_anchor)
# Q-shaped recurrent candidate selected by the V* interface oracle:
# R0=Q*, Rk=R{k-1}+CrossAttn(R{k-1}, V*)
# R_agg=mean(R1..RT)
# This is parameter-free and replaces both question-anchored state
# replacement and final-state-only decoding.
self.raw_mean_aggregation = bool(raw_mean_aggregation)
# A non-empty sequence maps the first T-1 recurrent steps to frozen
# visual-token states from the listed language layers. The final step
# uses Q* as memory to integrate the staged reads without another image
# access. One CrossAttention module is shared by all T steps.
self.hierarchical_visual_layers = [
int(layer) for layer in (hierarchical_visual_layers or ())
]
# Counterfactual Residual Recurrence (CRR): the image-conditioned and
# text-only question states are propagated by the same native layer
# F=L(ell_star+1). Only their counterfactual residual is recurrently
# refined; there are no slots, V* memory, custom attention modules, or
# inference-time auxiliary modules. beta is the fixed relaxed-update
# coefficient; an optional batch-swap reliance loss is training-only.
self.counterfactual_residual_recurrence = bool(
counterfactual_residual_recurrence
)
# Causal Visual Residual Recurrence (CVRR), implemented as the
# vision-native CRR transition. At every recurrent step the same
# native layer is evaluated with and without only Q->vision attention;
# their difference is the image-caused update. It deliberately
# reuses CRR's strict Q-shaped decoder/cache interface and mean
# aggregation, hence the parent CRR flag remains explicit.
self.visual_counterfactual_recurrence = bool(
visual_counterfactual_recurrence
)
# Minimal cumulative CVRR variant. Unlike the historical anchored
# mean path, visual updates are accumulated into the running state and
# the final state is decoded directly:
# R_k = R_{k-1} + beta * Delta_k^V, decode R_T.
# One explicit flag controls both choices so a checkpoint cannot
# silently combine cumulative dynamics with the old mean interface.
self.visual_cumulative_recurrence = bool(
visual_cumulative_recurrence
)
# Full-state CVRR keeps the native recurrent computation instead of
# treating the visual on/off intervention difference as the next
# hidden state:
# R_tilde_k = Pi_Q F([E; R_{k-1}])
# R_k = R_{k-1} + beta * (R_tilde_k - R_{k-1}).
# The visual-off branch remains available to evaluation code, but is
# not part of this model's train/inference transition.
self.visual_full_state_recurrence = bool(
visual_full_state_recurrence
)
# Input-grounded closure control. Later recurrent calls rebuild the
# native post-L20 scaffold and inject only the carried question-state
# deviation, rather than feeding post-L21 rows back into L21.
self.visual_boundary_reentry = bool(visual_boundary_reentry)
# Source-centered boundary recurrence. The text-only L21 state is a
# fixed anchor, while the native multimodal residual initializes a
# nonzero, image-dependent deviation. The recurrent cell evolves
# only that deviation after subtracting its zero-deviation response.
self.visual_source_centered_reentry = bool(
visual_source_centered_reentry
)
# Restrict LoRA to the single shared recurrent cell L(ell_star+1).
# Its execution is disabled for R1, the text-only cache, answer-token
# decoding, and L22+; only recurrent transitions k>=2 enable it.
self.recurrence_only_adapter = bool(recurrence_only_adapter)
# Preserve the pretrained image-caused transition while removing the
# frozen layer's non-visual recurrent drift:
# Delta_k = F_{base+LoRA,on}(E,R_{k-1}) - F_{base,off}(E,R_{k-1})
# = (F_{base,on} - F_{base,off})
# + (F_{base+LoRA,on} - F_{base,on}).
# This is a full-state recurrence subtype and introduces no new
# parameters or inference-time module.
self.visual_preserving_adapter_correction = bool(
visual_preserving_adapter_correction
)
# Local Visual Latent Chain (LVLC): K homogeneous latent-token
# positions are evaluated in one native Transformer forward. The
# first latent sees clean Q plus visual source rows; every later latent
# sees only the immediately preceding latent plus the same visual
# rows; answer rows see clean Q plus the final latent. This is a local
# attention graph, not a recurrent forward or temporal aggregation.
self.local_visual_latent_chain = bool(local_visual_latent_chain)
# Perceive--Deliberate Latent Chain (PDLC). K must be even and is
# interpreted as K/2 native LOOK/THINK pairs. The complete VLM stack
# runs once over a block-sparse graph: LOOK rows may inspect the
# multimodal prompt, THINK rows may inspect only clean question rows
# and their paired LOOK, and answer rows may inspect only clean
# question rows and THINK states. There is no recurrent cell, visual
# state update, split-state splice, or temporal aggregator.
# LVLC reuses the same one-pass latent-sequence execution plumbing but
# replaces the LOOK/THINK token types and visibility graph entirely.
self.perceive_deliberate_chain = bool(
perceive_deliberate_chain or self.local_visual_latent_chain
)
# Tight PDLC graph: pair 1 bootstraps from prompt/question, later
# LOOK rows receive visual placeholders plus the preceding latent
# prefix, later THINK rows receive only that causal latent prefix, and
# the answer consumes final THINK. It changes visibility only and
# allocates no additional trainable module.
self.pdlc_transition_conditioned = bool(
pdlc_transition_conditioned
)
# Strict inference uses 1: only the bootstrap pair reads the clean
# question. Larger values are an explicit evaluation intervention.
# The training-only curriculum changes a runtime override and always
# returns to 1 for validation/inference.
self.pdlc_question_visible_through_pair = int(
pdlc_question_visible_through_pair
)
self.pdlc_question_curriculum = bool(pdlc_question_curriculum)
# Both learned latent token types start from one native text embedding
# (the launcher supplies the tokenizer's newline id). -1 falls back
# to token 0 for synthetic/unit-test configs only.
self.latent_chain_init_token_id = int(latent_chain_init_token_id)
# Spatial Recurrent Visual Field (SRVF). Unlike all historical
# ell_star methods, the recurrent state is the complete native visual
# token grid immediately after the vision merger. The final field is
# inserted into the ordinary Qwen prefix and the full decoder runs from
# scratch, making the zero-output initialization exactly base-equivalent.
self.spatial_visual_recurrence = bool(spatial_visual_recurrence)
self.spatial_recurrence_steps = int(spatial_recurrence_steps)
self.spatial_recurrence_inner_width = int(
spatial_recurrence_inner_width
)
self.spatial_recurrence_heads = int(spatial_recurrence_heads)
# -1 uses the trained/default horizon. Non-negative values are an
# evaluation-only prefix sweep, with 0 being the exact base path.
self.spatial_step_override = int(spatial_step_override)
self.counterfactual_beta = float(counterfactual_beta)
# Learn one feature-wise mixture of the complete recurrent residual
# trajectory. The bias-free T*d -> d map starts from
# [I/T, ..., I/T], so enabling it is exactly the historical CRR mean
# interface before the first optimizer step. A dedicated flag keeps
# old CRR checkpoints and the historical slot-ER aggregator distinct.
self.crr_concat_aggregation = bool(crr_concat_aggregation)
# Optional latent-only policy adapter. It corrects the deterministic
# C2..CT means but never touches B, C1, or answer-token decoding. Rank 0
# is the exact historical CRR path; RL checkpoints serialize rank > 0.
self.crr_policy_rank = int(crr_policy_rank)
self.crr_policy_alpha = float(crr_policy_alpha)
# Training-only CRR intervention. A batch-matched residual from a
# different-answer example replaces the factual residual before the
# upper decoder. The model returns the factual-minus-swapped answer
# score gap; the Trainer applies the margin loss. Evaluation and
# generation never execute the extra branch.
self.counterfactual_reliance_loss = bool(
counterfactual_reliance_loss
)
# Replace the aggregate residual swap with one intervention inside the
# recurrent chain. C_k is swapped and C_{k+1:T} is rolled out again,
# so the answer gap is attributable to a selected transition rather
# than to an undifferentiated full-trajectory replacement.
self.transition_aware_reliance_loss = bool(
transition_aware_reliance_loss
)
# Training-only hard-case corrective ranking. At one selected
# transition k>=2, the actual decoder scores Z_{k-1}, Z_k, and a
# donor-swapped Z_k. The Trainer requires the factual transition to
# improve over both detached comparators. This replaces, rather than
# stacks with, the historical transition-aware reliance objective.
self.corrective_transition_reliance_loss = bool(
corrective_transition_reliance_loss
)
# Training-only functional no-regression objective. Gold-answer NLL
# is measured through the actual upper decoder at every recurrent CRR
# prefix; inference remains the unchanged T-step mean interface.
self.counterfactual_step_retention_loss = bool(
counterfactual_step_retention_loss
)
# Training-only self-distillation target. The multimodal lower pass
# already computed for V* also provides its image-conditioned question
# rows. They supervise R_agg at the split boundary, but are never
# exposed to the inference decoder.
self.functional_state_loss = bool(functional_state_loss)
# Token CE lets long caption answers dominate one-token visual-choice
# examples. The per-example variant first averages valid answer tokens
# within each sample, then averages samples. It changes training/eval
# loss reduction only and has no inference-time effect.
self.per_example_answer_loss = bool(per_example_answer_loss)
# ER v2 (advisor 2026-08-11): explicit evidence read —
# R^(k) = f_theta(R^(k-1), Q*; E), E는 별도 memory로 cross-attn 읽기.
# (E+R 합산-감산 제거. evidence_reasoning=True 전제.)
self.explicit_evidence_read = bool(kwargs.pop("explicit_evidence_read", False))
# Historical field name retained for checkpoint compatibility. In ER
# v2.2 this means R^(0) is a batch-broadcast set of learnable slots; it
# is deliberately NOT initialized from Q*. Q* conditions every shared
# transition instead.
self.r_init_question = bool(kwargs.pop("r_init_question", False))
# ER v2.1 (advisor 2026-08-11): evidence read의 competitive normalization —
# slot 축 softmax 후 evidence 축 재정규화 (Slot Attention식 경쟁).
# 파라미터/모듈/loss 무추가; evidence_read의 W_qkv/W_o 재사용.
self.competitive_evidence_read = bool(
kwargs.pop("competitive_evidence_read", False))
# ER v2.2 C: k>=1 traj 감독 부재 시 twin 체인 제거 (states = answer 체인).
self.er_single_chain = bool(kwargs.pop("er_single_chain", False))
# ER v2.2 C (§5-7): decoder latent = softmax(temporal_logits)로 가중한
# R1..RT 결합 (전역 스칼라 T개). er_single_chain 전제. (legacy)
self.temporal_aggregation = bool(kwargs.pop("temporal_aggregation", False))
# ER v2.2 C 확정판: R_agg = Linear(Concat(R1..RT), bias=False),
# 초기 W=[I/T,...,I/T] (R_agg=mean(R1..RT)). er_single_chain 전제.
self.concat_aggregation = bool(kwargs.pop("concat_aggregation", False))
# ER v2: 워크스페이스 블록의 rank-r 부분공간 attention/FFN (0 = dense).
# d_w=d에서 state width 유지와 파라미터 폭증을 분리 (advisor).
self.workspace_low_rank = int(kwargs.pop("workspace_low_rank", 0))
# advisor: single-layer text-side interface alignment (bridge-only).
# teacher = frozen-base mm 경로 q_end@l*+1 (adapter off, sg);
# student = detach(R^T) splice의 l*+1 한 층 재계산. 추론 시 무존재.
self.interface_loss = interface_loss
# Training-only visual anchor switch. False in the current answer-only
# method; when false the model does not even materialize pooled V* for
# the Trainer.
self.read_anchor = bool(read_anchor)
if evidence_reasoning and persistent_slot_id:
raise ValueError("evidence_reasoning은 persistent_slot_id와 동시 사용 불가")
if evidence_reasoning and not mid_read_step:
raise ValueError("evidence_reasoning은 mid_read_step > 0 필요")
self.validate()
# -- derived views -----------------------------------------------------
@property
def num_decoder_layers(self) -> int:
return self.text_config.num_hidden_layers
@property
def backbone_width(self) -> int:
"""``d`` -- backbone hidden width (3584 for the 7B)."""
return self.text_config.hidden_size
@property
def lower_slice(self) -> slice:
"""Layers forming ``F_{<=l*}``."""
self._require_ell_star()
return slice(0, self.ell_star + 1)
@property
def upper_slice(self) -> slice:
"""Layers forming ``F_{>l*}``."""
self._require_ell_star()
return slice(self.ell_star + 1, self.num_decoder_layers)
# -- validation --------------------------------------------------------
def _require_ell_star(self) -> None:
if self.ell_star is None:
raise ValueError(
"ell_star is unset. Run scripts/localize_read_point.py (Method 3.1) "
"and pass the measured layer explicitly; it must not be guessed."
)
def validate(self) -> None:
"""Reject configurations that silently break the §3.2 contract."""
# ``PretrainedConfig.from_pretrained`` may apply user overrides with
# setattr *after* __init__. In that path, setting LVLC=True would not
# re-run the constructor's promotion of the shared one-pass execution
# flag. Re-establish the invariant before any model reads the config.
if getattr(self, "local_visual_latent_chain", False):
self.perceive_deliberate_chain = True
if self.ell_star is not None:
n = self.num_decoder_layers
# Both branches must be non-empty: an empty lower branch means there
# is no V* to read, an empty upper branch means the workspace never
# reaches the decoder.
if not 0 <= self.ell_star <= n - 2:
raise ValueError(
f"ell_star={self.ell_star} out of range for {n} decoder layers; "
f"expected 0 <= ell_star <= {n - 2} so both branches are non-empty."
)
# K=0 is the no-recurrence control (read only), not a misconfiguration.
if self.num_workspace_steps < 0:
raise ValueError("num_workspace_steps (K) must be >= 0.")
if self.num_workspace_slots < 1:
raise ValueError("num_workspace_slots (S) must be >= 1.")
if self.workspace_num_heads < 1:
raise ValueError("workspace_num_heads must be >= 1.")
if self.workspace_width % self.workspace_num_heads != 0:
raise ValueError(
f"workspace_width={self.workspace_width} must be divisible by "
f"workspace_num_heads={self.workspace_num_heads}."
)
if self.workspace_width > self.backbone_width:
raise ValueError(
f"workspace_width (d_w={self.workspace_width}) cannot exceed "
f"backbone width (d={self.backbone_width}). Native-width d_w=d is "
"the current method and uses an Identity decoder interface."
)
if self.workspace_rope_mode not in WORKSPACE_ROPE_MODES:
raise ValueError(
f"workspace_rope_mode={self.workspace_rope_mode!r} not in "
f"{WORKSPACE_ROPE_MODES}."
)
if self.adapter_rank < 0:
raise ValueError("adapter_rank must be >= 0 (0 disables the adapter).")
if not math.isfinite(self.adapter_dropout) or not 0.0 <= self.adapter_dropout < 1.0:
raise ValueError("adapter_dropout must be finite and in [0, 1).")
if getattr(self, "crr_policy_rank", 0) < 0:
raise ValueError("crr_policy_rank must be >= 0.")
if not math.isfinite(getattr(self, "crr_policy_alpha", 0.0)) or getattr(
self, "crr_policy_alpha", 0.0
) <= 0.0:
raise ValueError("crr_policy_alpha must be finite and > 0.")
if self.read_num_blocks < 1 or self.transition_num_blocks < 1:
raise ValueError("read_num_blocks and transition_num_blocks must be >= 1.")
if self.workspace_ffn_mult < 1:
raise ValueError("workspace_ffn_mult must be >= 1.")
mid = int(getattr(self, "mid_read_step", 0) or 0)
if mid < 0 or mid > self.num_workspace_steps:
raise ValueError(
f"mid_read_step={mid} must be in [0, num_workspace_steps="
f"{self.num_workspace_steps}]."
)
low_rank = int(getattr(self, "workspace_low_rank", 0) or 0)
if low_rank < 0:
raise ValueError("workspace_low_rank must be >= 0.")
if low_rank and low_rank % self.workspace_num_heads != 0:
raise ValueError(
f"workspace_low_rank={low_rank} must be divisible by "
f"workspace_num_heads={self.workspace_num_heads}."
)
if getattr(self, "competitive_evidence_read", False) and not getattr(
self, "explicit_evidence_read", False
):
raise ValueError(
"competitive_evidence_read requires explicit_evidence_read=True; "
"otherwise the flag has no execution path."
)
if getattr(self, "explicit_evidence_read", False) and not getattr(
self, "evidence_reasoning", False
):
raise ValueError(
"explicit_evidence_read requires evidence_reasoning=True; otherwise "
"the evidence-read module is allocated but never called."
)
if getattr(self, "r_init_question", False) and not getattr(
self, "evidence_reasoning", False
):
raise ValueError(
"r_init_question (legacy name for learnable R0 slots) requires "
"evidence_reasoning=True; otherwise reasoning_slots are unused."
)
if (getattr(self, "concat_aggregation", False)
or getattr(self, "temporal_aggregation", False)):
if self.num_workspace_steps < 1:
raise ValueError("reasoning aggregation requires num_workspace_steps >= 1.")
if not getattr(self, "evidence_reasoning", False):
raise ValueError(
"reasoning aggregation requires evidence_reasoning=True; "
"otherwise the aggregation module is unused."
)
if getattr(self, "concat_aggregation", False) and not getattr(
self, "er_single_chain", False
):
raise ValueError("concat_aggregation requires er_single_chain=True.")
if getattr(self, "raw_evidence_question", False):
if self.workspace_width != self.backbone_width:
raise ValueError(
"raw_evidence_question requires native workspace_width == "
"backbone_width: E=V* and R0=Q* are used without projections."
)
if self.num_workspace_steps < 1:
raise ValueError(
"raw_evidence_question requires num_workspace_steps >= 1."
)
source_layer = (
self.ell_star + 1
if self.raw_attention_source_layer < 0
else self.raw_attention_source_layer
)
if not self.ell_star < source_layer < self.text_config.num_hidden_layers:
raise ValueError(
"raw_attention_source_layer must be an upper-backbone layer, "
f"got ell_star={self.ell_star}, source={source_layer}."
)
incompatible = {
"evidence_reasoning": bool(getattr(self, "evidence_reasoning", False)),
"explicit_evidence_read": bool(getattr(self, "explicit_evidence_read", False)),
"competitive_evidence_read": bool(getattr(self, "competitive_evidence_read", False)),
"r_init_question": bool(getattr(self, "r_init_question", False)),
"er_single_chain": bool(getattr(self, "er_single_chain", False)),
"temporal_aggregation": bool(getattr(self, "temporal_aggregation", False)),
"concat_aggregation": bool(getattr(self, "concat_aggregation", False)),
"read_shortcut": bool(getattr(self, "read_shortcut", False)),
"splice_scale_match": bool(getattr(self, "splice_scale_match", False)),
"read_gating": bool(getattr(self, "read_gating", False)),
"read_residual_pure": bool(getattr(self, "read_residual_pure", False)),
"final_verify": bool(getattr(self, "final_verify", False)),
"mid_read_step": bool(getattr(self, "mid_read_step", 0)),
"persistent_slot_id": bool(getattr(self, "persistent_slot_id", False)),
"interface_loss": bool(getattr(self, "interface_loss", False)),
"read_anchor": bool(getattr(self, "read_anchor", False)),
}
active = [name for name, enabled in incompatible.items() if enabled]
if active:
raise ValueError(
"raw_evidence_question replaces the slot workspace and is "
f"incompatible with: {', '.join(active)}"
)
if self.raw_state_replacement and self.raw_transition_ffn:
raise ValueError(
"raw_state_replacement is CrossAttn-only and is incompatible "
"with raw_transition_ffn=True."
)
if self.raw_question_anchor and not self.raw_state_replacement:
raise ValueError(
"raw_question_anchor requires raw_state_replacement=True."
)
if self.raw_mean_aggregation and (
self.raw_state_replacement or self.raw_question_anchor
):
raise ValueError(
"raw_mean_aggregation replaces raw_state_replacement and "
"raw_question_anchor; use the residual recurrence."
)
hierarchy = list(
getattr(self, "hierarchical_visual_layers", ()) or ()
)
if hierarchy:
if not self.raw_mean_aggregation:
raise ValueError(
"hierarchical_visual_layers requires "
"raw_mean_aggregation=True."
)
if self.raw_transition_ffn:
raise ValueError(
"hierarchical visual recurrence is CrossAttn-only and "
"requires raw_transition_ffn=False."
)
expected_visual_reads = self.num_workspace_steps - 1
if len(hierarchy) != expected_visual_reads:
raise ValueError(
"hierarchical_visual_layers must contain exactly one "
"layer per visual-read step, followed by one Q* "
"integration step: "
f"got {len(hierarchy)} layers for "
f"T={self.num_workspace_steps} "
f"(expected {expected_visual_reads})."
)
if hierarchy != sorted(set(hierarchy)):
raise ValueError(
"hierarchical_visual_layers must be strictly increasing."
)
invalid = [
layer
for layer in hierarchy
if not 0 <= layer <= int(self.ell_star)
]
if invalid:
raise ValueError(
"hierarchical_visual_layers must lie in the frozen lower "
f"branch [0,{self.ell_star}], got {invalid}."
)
if self.functional_state_loss:
raise ValueError(
"functional_state_loss is not defined for hierarchical "
"visual recurrence; use final-answer CE only."
)
if self.functional_state_loss and not self.raw_mean_aggregation:
raise ValueError(
"functional_state_loss requires raw_mean_aggregation=True "
"so its target is the exact decoder-facing recurrence."
)
elif getattr(self, "raw_state_replacement", False):
raise ValueError(
"raw_state_replacement requires raw_evidence_question=True."
)
elif getattr(self, "raw_question_anchor", False):
raise ValueError(
"raw_question_anchor requires raw_evidence_question=True."
)
elif getattr(self, "raw_mean_aggregation", False):
raise ValueError(
"raw_mean_aggregation requires raw_evidence_question=True."
)
elif getattr(self, "hierarchical_visual_layers", ()):
raise ValueError(
"hierarchical_visual_layers requires "
"raw_evidence_question=True."
)
elif getattr(self, "functional_state_loss", False):
raise ValueError(
"functional_state_loss requires raw_evidence_question=True."
)
if (
getattr(self, "counterfactual_reliance_loss", False)
and not getattr(self, "counterfactual_residual_recurrence", False)
):
raise ValueError(
"counterfactual_reliance_loss requires "
"counterfactual_residual_recurrence=True."
)
if (
getattr(self, "transition_aware_reliance_loss", False)
and not getattr(self, "counterfactual_reliance_loss", False)
):
raise ValueError(
"transition_aware_reliance_loss requires "
"counterfactual_reliance_loss=True."
)
if (
getattr(self, "transition_aware_reliance_loss", False)
and self.num_workspace_steps < 2
):
raise ValueError(
"transition_aware_reliance_loss requires at least two steps."
)
if (
getattr(self, "corrective_transition_reliance_loss", False)
and not getattr(self, "counterfactual_reliance_loss", False)
):
raise ValueError(
"corrective_transition_reliance_loss requires "
"counterfactual_reliance_loss=True."
)
if (
getattr(self, "corrective_transition_reliance_loss", False)
and self.num_workspace_steps < 2
):
raise ValueError(
"corrective_transition_reliance_loss requires at least two steps."
)
if (
getattr(self, "corrective_transition_reliance_loss", False)
and getattr(self, "transition_aware_reliance_loss", False)
):
raise ValueError(
"corrective_transition_reliance_loss replaces, and cannot be "
"combined with, transition_aware_reliance_loss."
)
if (
getattr(self, "corrective_transition_reliance_loss", False)
and getattr(self, "counterfactual_step_retention_loss", False)
):
raise ValueError(
"corrective_transition_reliance_loss replaces, and cannot be "
"combined with, counterfactual_step_retention_loss."
)
if (
getattr(self, "counterfactual_step_retention_loss", False)
and not getattr(self, "counterfactual_residual_recurrence", False)
):
raise ValueError(
"counterfactual_step_retention_loss requires "
"counterfactual_residual_recurrence=True."
)
if (
getattr(self, "counterfactual_step_retention_loss", False)
and self.num_workspace_steps < 2
):
raise ValueError(
"counterfactual_step_retention_loss requires at least two steps."
)
if getattr(self, "crr_concat_aggregation", False) and not getattr(
self, "counterfactual_residual_recurrence", False
):
raise ValueError(
"crr_concat_aggregation requires "
"counterfactual_residual_recurrence=True."
)
if getattr(self, "counterfactual_residual_recurrence", False):
if getattr(self, "raw_evidence_question", False):
raise ValueError(
"counterfactual_residual_recurrence replaces "
"raw_evidence_question; enable exactly one q-state method."
)
if self.num_workspace_steps < 1:
raise ValueError(
"counterfactual_residual_recurrence requires "
"num_workspace_steps >= 1."
)
if self.crr_policy_rank > 0 and self.num_workspace_steps < 2:
raise ValueError(
"crr_policy_rank > 0 requires num_workspace_steps >= 2."
)
if not 0.0 <= self.counterfactual_beta <= 1.0:
raise ValueError(
"counterfactual_beta must lie in [0, 1], got "
f"{self.counterfactual_beta}."
)
incompatible = {
"evidence_reasoning": bool(getattr(self, "evidence_reasoning", False)),
"explicit_evidence_read": bool(getattr(self, "explicit_evidence_read", False)),
"competitive_evidence_read": bool(getattr(self, "competitive_evidence_read", False)),
"r_init_question": bool(getattr(self, "r_init_question", False)),
"er_single_chain": bool(getattr(self, "er_single_chain", False)),
"temporal_aggregation": bool(getattr(self, "temporal_aggregation", False)),
"concat_aggregation": bool(getattr(self, "concat_aggregation", False)),
"read_shortcut": bool(getattr(self, "read_shortcut", False)),
"splice_scale_match": bool(getattr(self, "splice_scale_match", False)),
"read_gating": bool(getattr(self, "read_gating", False)),
"read_residual_pure": bool(getattr(self, "read_residual_pure", False)),
"final_verify": bool(getattr(self, "final_verify", False)),
"mid_read_step": bool(getattr(self, "mid_read_step", 0)),
"persistent_slot_id": bool(getattr(self, "persistent_slot_id", False)),
"interface_loss": bool(getattr(self, "interface_loss", False)),
"read_anchor": bool(getattr(self, "read_anchor", False)),
"functional_state_loss": bool(getattr(self, "functional_state_loss", False)),
}
active = [name for name, enabled in incompatible.items() if enabled]
if active:
raise ValueError(
"counterfactual_residual_recurrence replaces the slot/raw "
f"workspace and is incompatible with: {', '.join(active)}"
)
elif getattr(self, "crr_policy_rank", 0) > 0:
raise ValueError(
"crr_policy_rank > 0 requires "
"counterfactual_residual_recurrence=True."
)
if getattr(self, "visual_counterfactual_recurrence", False):
if not getattr(
self, "counterfactual_residual_recurrence", False
):
raise ValueError(
"visual_counterfactual_recurrence requires "
"counterfactual_residual_recurrence=True."
)
# CVRR is the intentionally minimal CE-only candidate. The old
# CRR tail-rollout losses/policy implement a different transition
# equation and must not be silently applied to this trajectory.
incompatible = {
"counterfactual_reliance_loss": bool(
getattr(self, "counterfactual_reliance_loss", False)
),
"transition_aware_reliance_loss": bool(
getattr(self, "transition_aware_reliance_loss", False)
),
"corrective_transition_reliance_loss": bool(
getattr(self, "corrective_transition_reliance_loss", False)
),
"counterfactual_step_retention_loss": bool(
getattr(self, "counterfactual_step_retention_loss", False)
),
"crr_concat_aggregation": bool(
getattr(self, "crr_concat_aggregation", False)
),
"crr_policy_rank": bool(getattr(self, "crr_policy_rank", 0)),
}
active = [name for name, enabled in incompatible.items() if enabled]
if active:
raise ValueError(
"visual_counterfactual_recurrence uses final-answer CE only "
"and a parameter-free decoder interface; incompatible with: "
+ ", ".join(active)
)
elif getattr(self, "visual_cumulative_recurrence", False):
raise ValueError(
"visual_cumulative_recurrence requires "
"visual_counterfactual_recurrence=True."
)
if getattr(self, "visual_full_state_recurrence", False):
if not getattr(self, "visual_counterfactual_recurrence", False):
raise ValueError(
"visual_full_state_recurrence requires "
"visual_counterfactual_recurrence=True."
)
if not getattr(self, "visual_cumulative_recurrence", False):
raise ValueError(
"visual_full_state_recurrence requires "
"visual_cumulative_recurrence=True."
)
if getattr(self, "visual_boundary_reentry", False) and not getattr(
self, "visual_full_state_recurrence", False
):
raise ValueError(
"visual_boundary_reentry requires "
"visual_full_state_recurrence=True."
)
if getattr(self, "visual_source_centered_reentry", False):
if not getattr(self, "visual_boundary_reentry", False):
raise ValueError(
"visual_source_centered_reentry requires "
"visual_boundary_reentry=True."
)
if getattr(self, "visual_preserving_adapter_correction", False):
raise ValueError(
"visual_source_centered_reentry cannot be combined with "
"visual_preserving_adapter_correction."
)
if getattr(self, "recurrence_only_adapter", False):
if not getattr(self, "visual_full_state_recurrence", False):
raise ValueError(
"recurrence_only_adapter requires "
"visual_full_state_recurrence=True."
)
if self.adapter_rank <= 0:
raise ValueError(
"recurrence_only_adapter requires adapter_rank > 0."
)
if self.ell_star is None:
raise ValueError(
"recurrence_only_adapter requires an explicit ell_star."
)
recurrent_layer = int(self.ell_star) + 1
if recurrent_layer in set(self.adapter_exclude_layers):
raise ValueError(
"recurrence_only_adapter cannot exclude its recurrent "
f"layer {recurrent_layer} from LoRA."
)
if getattr(self, "visual_preserving_adapter_correction", False):
if not getattr(self, "visual_full_state_recurrence", False):
raise ValueError(
"visual_preserving_adapter_correction requires "
"visual_full_state_recurrence=True."
)
if not getattr(self, "recurrence_only_adapter", False):
raise ValueError(
"visual_preserving_adapter_correction requires "
"recurrence_only_adapter=True."
)
if getattr(self, "perceive_deliberate_chain", False):
if self.num_workspace_steps < 2 or self.num_workspace_steps % 2:
raise ValueError(
"one-pass latent chains require an even "
"num_workspace_steps >= 2."
)
vocab_size = int(self.text_config.vocab_size)
if not -1 <= self.latent_chain_init_token_id < vocab_size:
raise ValueError(
"latent_chain_init_token_id must be -1 or a valid text "
f"token id below {vocab_size}."
)
num_pairs = self.num_workspace_steps // 2
if self.local_visual_latent_chain and (
self.pdlc_transition_conditioned
or self.pdlc_question_curriculum
or self.pdlc_question_visible_through_pair != 1
):
raise ValueError(
"local_visual_latent_chain replaces PDLC transition and "
"question-visibility controls"
)
if not 1 <= self.pdlc_question_visible_through_pair <= num_pairs:
raise ValueError(
"pdlc_question_visible_through_pair must lie in "
f"[1,{num_pairs}]"
)
if self.pdlc_question_curriculum and not (
getattr(self, "pdlc_transition_conditioned", False)
):
raise ValueError(
"pdlc_question_curriculum requires "
"pdlc_transition_conditioned=True"
)
incompatible = {
"raw_evidence_question": bool(
getattr(self, "raw_evidence_question", False)
),
"counterfactual_residual_recurrence": bool(
getattr(self, "counterfactual_residual_recurrence", False)
),
"spatial_visual_recurrence": bool(
getattr(self, "spatial_visual_recurrence", False)
),
"evidence_reasoning": bool(
getattr(self, "evidence_reasoning", False)
),
"read_gating": bool(getattr(self, "read_gating", False)),
"read_residual_pure": bool(
getattr(self, "read_residual_pure", False)
),
"final_verify": bool(getattr(self, "final_verify", False)),
"mid_read_step": bool(getattr(self, "mid_read_step", 0)),
"interface_loss": bool(getattr(self, "interface_loss", False)),
"read_anchor": bool(getattr(self, "read_anchor", False)),
"counterfactual_reliance_loss": bool(
getattr(self, "counterfactual_reliance_loss", False)
),
"counterfactual_step_retention_loss": bool(
getattr(self, "counterfactual_step_retention_loss", False)
),
"functional_state_loss": bool(
getattr(self, "functional_state_loss", False)
),
}
active = [name for name, enabled in incompatible.items() if enabled]
if active:
raise ValueError(
"perceive_deliberate_chain replaces every recurrent/"
"workspace path; incompatible with: " + ", ".join(active)
)
elif getattr(self, "pdlc_transition_conditioned", False):
raise ValueError(
"pdlc_transition_conditioned requires "
"perceive_deliberate_chain=True."
)
elif getattr(self, "pdlc_question_curriculum", False) or (
getattr(self, "pdlc_question_visible_through_pair", 1) != 1
):
raise ValueError(
"PDLC question visibility controls require "
"perceive_deliberate_chain=True."
)
if getattr(self, "spatial_visual_recurrence", False):
if not 0.0 <= self.counterfactual_beta <= 1.0:
raise ValueError(
"counterfactual_beta must lie in [0, 1], got "
f"{self.counterfactual_beta}."
)
if self.spatial_recurrence_steps < 1:
raise ValueError(
"spatial_recurrence_steps must be >= 1 for training"
)
if self.spatial_recurrence_inner_width < 1:
raise ValueError(
"spatial_recurrence_inner_width must be positive"
)
if self.spatial_recurrence_heads < 1 or (
self.spatial_recurrence_inner_width
% self.spatial_recurrence_heads
):
raise ValueError(
"spatial_recurrence_inner_width must be divisible by "
"spatial_recurrence_heads"
)
if self.spatial_step_override < -1:
raise ValueError("spatial_step_override must be >= -1")
if self.adapter_rank != 0:
raise ValueError(
"spatial_visual_recurrence replaces decoder LoRA; set "
"adapter_rank=0 so the T=0 path remains the frozen base model"
)
incompatible = {
"raw_evidence_question": bool(
getattr(self, "raw_evidence_question", False)
),
"counterfactual_residual_recurrence": bool(
getattr(self, "counterfactual_residual_recurrence", False)
),
"evidence_reasoning": bool(
getattr(self, "evidence_reasoning", False)
),
"read_gating": bool(getattr(self, "read_gating", False)),
"read_residual_pure": bool(
getattr(self, "read_residual_pure", False)
),
"final_verify": bool(getattr(self, "final_verify", False)),
"mid_read_step": bool(getattr(self, "mid_read_step", 0)),
"interface_loss": bool(getattr(self, "interface_loss", False)),
"read_anchor": bool(getattr(self, "read_anchor", False)),
"counterfactual_reliance_loss": bool(
getattr(self, "counterfactual_reliance_loss", False)
),
"counterfactual_step_retention_loss": bool(
getattr(self, "counterfactual_step_retention_loss", False)
),
"functional_state_loss": bool(
getattr(self, "functional_state_loss", False)
),
}
active = [name for name, enabled in incompatible.items() if enabled]
if active:
raise ValueError(
"spatial_visual_recurrence replaces every historical "
"workspace/reliance path; incompatible with: "
+ ", ".join(active)
)
modes = [bool(getattr(self, m, False)) for m in
("read_gating", "read_residual_pure", "final_verify")]
modes.append(bool(getattr(self, "mid_read_step", 0)))
if sum(modes) > 1:
raise ValueError(
"read_gating / read_residual_pure / final_verify are mutually "
"exclusive reread modes."
)
__all__ = ["CloseQwen2_5_VLConfig", "WORKSPACE_ROPE_MODES"]