From 018c84100e45df0346bf3ea4e501b16216ed7f46 Mon Sep 17 00:00:00 2001 From: Ingvar Date: Fri, 2 Oct 2026 10:27:17 +0300 Subject: [PATCH 01/12] implement recurrence for decoder kv scorer --- gliclass/config.py | 22 +++++++++ gliclass/model.py | 111 +++++++++++++++++++++++++++++++++++++++++-- gliclass/scorers.py | 106 +++++++++++++++++++++++++++++++++++++++-- gliclass/training.py | 25 ++++++++-- 4 files changed, 251 insertions(+), 13 deletions(-) diff --git a/gliclass/config.py b/gliclass/config.py index dd11728..524344a 100644 --- a/gliclass/config.py +++ b/gliclass/config.py @@ -60,6 +60,14 @@ def __init__( dropout=0.1, use_segment_embeddings=False, scorer_encoder_num_layers=2, + recurrent_steps=1, + recurrent_min_steps=1, + recurrent_inference=True, + recurrent_inference_max_steps=None, + recurrent_halt_threshold=0.01, + recurrent_improvement_coef=0.1, + recurrent_improvement_margin=0.0, + recurrent_bptt_steps=None, **kwargs, ): if isinstance(encoder_config, dict): @@ -149,4 +157,18 @@ def __init__( self.scorer_encoder_num_layers = scorer_encoder_num_layers + # Recurrent hidden reasoning for the decoder-kv scorer (recurrent_steps=1 disables it). + # Training depth is sampled uniformly from [recurrent_min_steps, recurrent_steps]. + self.recurrent_steps = recurrent_steps + self.recurrent_min_steps = recurrent_min_steps + # Inference: run up to recurrent_inference_max_steps (default recurrent_steps, may exceed it) and + # stop an example once no label probability changes by recurrent_halt_threshold (None/0 = never). + self.recurrent_inference = recurrent_inference + self.recurrent_inference_max_steps = recurrent_inference_max_steps + self.recurrent_halt_threshold = recurrent_halt_threshold + # Hinge relu(l_t - (1 - margin) * sg(l_{t-1})) for steps t >= 3, on top of a loss at every step. + self.recurrent_improvement_coef = recurrent_improvement_coef + self.recurrent_improvement_margin = recurrent_improvement_margin + self.recurrent_bptt_steps = recurrent_bptt_steps + super().__init__(problem_type=problem_type, **kwargs) diff --git a/gliclass/model.py b/gliclass/model.py index 0757381..cab2146 100644 --- a/gliclass/model.py +++ b/gliclass/model.py @@ -62,6 +62,9 @@ class GLiClassOutput(SequenceClassifierOutput): text_embeddings: torch.Tensor | None = None class_embeddings: torch.Tensor | None = None past_key_values: Any | None = None + recurrent_logits: Tuple[torch.Tensor, ...] | None = None + recurrent_losses: torch.Tensor | None = None + recurrent_num_steps: torch.Tensor | None = None class GLiClassPreTrainedModel(PreTrainedModel): @@ -1042,6 +1045,78 @@ def get_loss(self, logits, labels): raise NotImplementedError(f"{self.config.problem_type} is not implemented.") return loss + def _resolve_recurrence(self, use_recurrence=None, max_recurrent_steps=None, halt_threshold=None): + """Pick (num_steps, halt_threshold) for this call. + + Training: depth sampled uniformly from [recurrent_min_steps, max], no halting. + Inference: up to max steps, halting per example once probabilities stop changing. + """ + if not self.scorer.recurrent: + return 1, None + if use_recurrence is None: + use_recurrence = self.training or self.config.recurrent_inference + if not use_recurrence: + return 1, None + + if self.training: + max_steps = max_recurrent_steps or self.config.recurrent_steps + min_steps = min(self.config.recurrent_min_steps or 1, max_steps) + return int(torch.randint(min_steps, max_steps + 1, (1,)).item()), None + + max_steps = max_recurrent_steps or self.config.recurrent_inference_max_steps or self.config.recurrent_steps + if halt_threshold is None: + halt_threshold = self.config.recurrent_halt_threshold + return max_steps, halt_threshold + + def _per_sample_loss(self, logits, labels): + """Unreduced loss per example, shape (batch,).""" + if self.config.problem_type == "multi_label_classification": + from .loss_functions import focal_loss_with_logits + + all_losses = focal_loss_with_logits( + logits, + labels, + self.config.focal_loss_alpha, + self.config.focal_loss_gamma, + "none", + ) + valid = labels.ne(self.config.ignore_index) + return all_losses.sum(-1) / valid.sum(-1).clamp_min(1) + elif self.config.problem_type == "single_label_classification": + return nn.functional.cross_entropy( + logits, labels.view(-1), ignore_index=self.config.ignore_index, reduction="none" + ) + raise NotImplementedError(f"{self.config.problem_type} is not implemented.") + + def get_recurrent_loss(self, step_logits, labels): + """Loss over all reasoning steps. + + loss = mean_t L_t (focal loss after every step) + + coef * mean_{t>=3} relu(l_t - (1 - margin) * sg(l_{t-1})) (deeper must beat shallower) + + The improvement term starts at step 3 vs step 2, so step 1 is only ever trained by its own + loss. The stop-gradient on l_{t-1} means the hinge can only push the deeper step down; it + never rewards making the earlier step worse. + + Returns: + loss: scalar + step_losses: (num_steps,) detached mean loss per step, for monitoring + """ + num_labels = step_logits[-1].shape[-1] + if labels.shape[-1] != num_labels: + labels = labels[:, :num_labels] + + per_sample = torch.stack([self._per_sample_loss(logits, labels) for logits in step_logits]) # (T, B) + step_losses = per_sample.mean(dim=1) + loss = step_losses.mean() + + if len(step_logits) >= 3 and self.config.recurrent_improvement_coef > 0: + target = (1.0 - self.config.recurrent_improvement_margin) * per_sample[1:-1].detach() + improvement = torch.relu(per_sample[2:] - target).mean() + loss = loss + self.config.recurrent_improvement_coef * improvement + + return loss, step_losses.detach() + def _extract_label_section( self, hidden_states: torch.Tensor, @@ -1131,6 +1206,9 @@ def classify_from_decoder_cache( label_mask: torch.Tensor, position_ids: torch.Tensor, past_key_values, + use_recurrence: bool | None = None, + max_recurrent_steps: int | None = None, + halt_threshold: float | None = None, ) -> GLiClassOutput: """Classify label tokens against a text cache without persisting labels.""" decoder_outputs = self.decoder_model( @@ -1144,12 +1222,15 @@ def classify_from_decoder_cache( label_hidden = decoder_outputs.last_hidden_state[:, -input_ids.shape[1] :, :] scorer_device = next(self.scorer.parameters()).device - logits = self.scorer( + num_steps, halt_threshold = self._resolve_recurrence(use_recurrence, max_recurrent_steps, halt_threshold) + step_logits, steps_taken = self.scorer.recurrent_forward( hidden_states=label_hidden.to(scorer_device), input_ids=input_ids.to(scorer_device), attention_mask=label_mask.to(scorer_device), + num_steps=num_steps, + halt_threshold=halt_threshold, ) - return GLiClassOutput(logits=logits) + return GLiClassOutput(logits=step_logits[-1], recurrent_num_steps=steps_taken) def forward( self, @@ -1159,6 +1240,10 @@ def forward( labels: torch.Tensor | None = None, return_dict: bool | None = None, use_cache: bool = False, + use_recurrence: bool | None = None, + max_recurrent_steps: int | None = None, + halt_threshold: float | None = None, + output_recurrent_logits: bool = False, **kwargs, ): """Forward pass. @@ -1170,9 +1255,13 @@ def forward( labels: Classification labels return_dict: Return dict output use_cache: Whether to return updated cache + use_recurrence: Enable/disable scorer recurrence for this call (default: config) + max_recurrent_steps: Max reasoning steps for this call (training samples depth up to it) + halt_threshold: Inference early-stop threshold on label probability change (0 disables) + output_recurrent_logits: Return logits of every reasoning step Returns: - GLiClassOutput with logits, loss, and optionally past_key_values + GLiClassOutput with logits (of the last step), loss, and optionally past_key_values """ return_dict = return_dict if return_dict is not None else self.config.use_return_dict @@ -1189,16 +1278,25 @@ def forward( label_hidden, label_ids, label_mask = self._extract_label_section(hidden_states, input_ids, attention_mask) - logits = self.scorer( + num_steps, halt_threshold = self._resolve_recurrence(use_recurrence, max_recurrent_steps, halt_threshold) + step_logits, steps_taken = self.scorer.recurrent_forward( hidden_states=label_hidden, input_ids=label_ids, attention_mask=label_mask, + num_steps=num_steps, + halt_threshold=halt_threshold, + return_all_steps=labels is not None or output_recurrent_logits, ) + logits = step_logits[-1] if labels is not None: self.num_labels = logits.shape[-1] - loss = self.get_loss(logits, labels) + recurrent_losses = None + if len(step_logits) > 1 and labels is not None: + loss, recurrent_losses = self.get_recurrent_loss(step_logits, labels) + else: + loss = self.get_loss(logits, labels) if not return_dict: output = (logits,) @@ -1212,6 +1310,9 @@ def forward( hidden_states=decoder_outputs.hidden_states if hasattr(decoder_outputs, "hidden_states") else None, attentions=decoder_outputs.attentions if hasattr(decoder_outputs, "attentions") else None, past_key_values=decoder_outputs.past_key_values if use_cache else None, + recurrent_logits=tuple(step_logits) if output_recurrent_logits else None, + recurrent_losses=recurrent_losses, + recurrent_num_steps=steps_taken if num_steps > 1 else None, ) diff --git a/gliclass/scorers.py b/gliclass/scorers.py index 584e7c0..82afb8c 100644 --- a/gliclass/scorers.py +++ b/gliclass/scorers.py @@ -221,6 +221,13 @@ def __init__(self, config, **kwargs): self.epsilon = 1e-8 + # GRU-style recurrent reasoning over scorer_encoder. Gates are step-agnostic (no step + # embeddings), so inference can run more steps than were seen during training. + self.recurrent = getattr(config, "recurrent_steps", 1) > 1 + if self.recurrent: + self.reset_gate = nn.Linear(config.hidden_size * 2, config.hidden_size) + self.update_gate = nn.Linear(config.hidden_size * 2, config.hidden_size) + def _extract_representations(self, hidden_states, input_ids, attention_mask): """ Extract text and label representations from hidden states. @@ -267,14 +274,107 @@ def forward(self, hidden_states, input_ids, attention_mask, **kwargs): hidden_states: (batch_size, seq_length, hidden_size) from decoder backbone input_ids: (batch_size, seq_length) attention_mask: (batch_size, seq_length) + **kwargs: recurrence options, see recurrent_forward Returns: logits: (batch_size, num_labels) """ - encoder_outputs = self.scorer_encoder(hidden_states, attention_mask=attention_mask, return_dict=True) - - contextualized_hidden_states = encoder_outputs.last_hidden_state + step_logits, _ = self.recurrent_forward(hidden_states, input_ids, attention_mask, **kwargs) + return step_logits[-1] + + def _encode(self, hidden_states, attention_mask): + return self.scorer_encoder(hidden_states, attention_mask=attention_mask, return_dict=True).last_hidden_state + + def _recurrent_step(self, state, inputs, attention_mask): + """h_t = (1 - z) * h_{t-1} + z * Enc(r * h_{t-1} + (1 - r) * x), x = decoder hidden states.""" + reset = torch.sigmoid(self.reset_gate(torch.cat([state, inputs], dim=-1))) + candidate = self._encode(reset * state + (1 - reset) * inputs, attention_mask) + update = torch.sigmoid(self.update_gate(torch.cat([state, candidate], dim=-1))) + return (1 - update) * state + update * candidate + + def _valid_label_mask(self, input_ids, attention_mask, num_labels): + label_counts = (input_ids.eq(self.config.class_token_index) & attention_mask.bool()).sum(dim=1) + return torch.arange(num_labels, device=input_ids.device).unsqueeze(0) < label_counts.unsqueeze(1) + + def _probabilities(self, logits, valid_labels): + if self.config.problem_type == "single_label_classification": + return torch.softmax(logits.masked_fill(~valid_labels, float("-inf")), dim=-1).nan_to_num(0.0) + return torch.sigmoid(logits) + + def recurrent_forward( + self, + hidden_states, + input_ids, + attention_mask, + num_steps=1, + halt_threshold=None, + return_all_steps=False, + **kwargs, + ): + """Run step 1 (one scorer_encoder pass) and up to num_steps - 1 gated recurrent steps. + + With halt_threshold, an example stops once no valid label probability moves by + halt_threshold or more between consecutive steps; its state and logits are frozen. + Returns: + step_logits: list of (batch_size, num_labels) logits for every step if + return_all_steps, otherwise a single-element list with the final logits + steps_taken: (batch_size,) number of steps each example ran + """ + if num_steps > 1 and not self.recurrent: + raise ValueError("num_steps > 1 requires a scorer built with config.recurrent_steps > 1.") + + bptt_steps = getattr(self.config, "recurrent_bptt_steps", None) + + state = self._encode(hidden_states, attention_mask) + logits = self._score(state, input_ids, attention_mask) + step_logits = [logits] + + batch_size, num_labels = logits.shape + steps_taken = torch.ones(batch_size, dtype=torch.long, device=logits.device) + active = torch.ones(batch_size, dtype=torch.bool, device=logits.device) + if halt_threshold: + valid_labels = self._valid_label_mask(input_ids, attention_mask, num_labels) + + for step in range(1, num_steps): + if bptt_steps and step % bptt_steps == 0: + # truncated BPTT: later losses stop shaping earlier iterations + state = state.detach() + + new_state = self._recurrent_step(state, hidden_states, attention_mask) + new_logits = self._score(new_state, input_ids, attention_mask) + + if halt_threshold: + change = ( + (self._probabilities(new_logits, valid_labels) - self._probabilities(logits, valid_labels)) + .abs() + .masked_fill(~valid_labels, 0.0) + .amax(dim=-1) + ) + new_state = torch.where(active[:, None, None], new_state, state) + new_logits = torch.where(active[:, None], new_logits, logits) + steps_taken = steps_taken + active.long() + active = active & change.ge(halt_threshold) + else: + steps_taken = steps_taken + 1 + + state, logits = new_state, new_logits + if return_all_steps: + step_logits.append(logits) + else: + step_logits[-1] = logits + + if halt_threshold and not active.any(): + break + + if self.recurrent and self.training: + # keep the gates in the graph for DDP when the sampled depth is 1 + unused = sum(param.sum() for param in (*self.reset_gate.parameters(), *self.update_gate.parameters())) + step_logits[-1] = step_logits[-1] + 0.0 * unused + + return step_logits, steps_taken + + def _score(self, contextualized_hidden_states, input_ids, attention_mask): text_repr, label_repr = self._extract_representations(contextualized_hidden_states, input_ids, attention_mask) text_repr = self.text_projector(text_repr) diff --git a/gliclass/training.py b/gliclass/training.py index 4a88e02..36e26c1 100644 --- a/gliclass/training.py +++ b/gliclass/training.py @@ -319,6 +319,8 @@ def __init__(self, ewc: EWC | None = None, prev_dataset=None, *args, **kwargs): self.ewc = ewc self.prev_dataset = prev_dataset self._ewc_initialized = ewc is not None + # running sums of per-step losses of recurrent scorers, flushed in log() + self._recurrent_loss_sums = {} def _maybe_initialize_ewc(self): """Initialize EWC if needed and not already initialized.""" @@ -369,11 +371,8 @@ def compute_loss(self, model, inputs, return_outputs=False, **kwargs): Loss tensor, or tuple of (loss, outputs) if return_outputs=True """ # Get base loss from parent - if return_outputs: - loss, outputs = super().compute_loss(model, inputs, return_outputs=True, **kwargs) - else: - loss = super().compute_loss(model, inputs, return_outputs=False, **kwargs) - outputs = None + loss, outputs = super().compute_loss(model, inputs, return_outputs=True, **kwargs) + self._track_recurrent_losses(model, outputs) # Add EWC penalty if enabled if self.ewc is not None and self.args.use_ewc: @@ -385,6 +384,22 @@ def compute_loss(self, model, inputs, return_outputs=False, **kwargs): return loss, outputs return loss + def _track_recurrent_losses(self, model, outputs): + step_losses = outputs.get("recurrent_losses") if isinstance(outputs, dict) else None + if step_losses is None or not model.training: + return + for step, value in enumerate(step_losses.tolist(), start=1): + total, count = self._recurrent_loss_sums.get(step, (0.0, 0)) + self._recurrent_loss_sums[step] = (total + value, count + 1) + + def log(self, logs, *args, **kwargs): + """Add mean per-step losses of recurrent scorers to training logs.""" + if "loss" in logs and self._recurrent_loss_sums: + for step, (total, count) in sorted(self._recurrent_loss_sums.items()): + logs[f"loss_step_{step}"] = round(total / count, 4) + self._recurrent_loss_sums = {} + return super().log(logs, *args, **kwargs) + def train(self, *args, **kwargs): """Train with EWC initialization.""" # Initialize EWC before training starts From ee5894e2fddeb8218181d04d426440e1e97c19ba Mon Sep 17 00:00:00 2001 From: Ingvar Date: Fri, 2 Oct 2026 12:40:45 +0300 Subject: [PATCH 02/12] make recurrenses more robust --- gliclass/config.py | 15 +++++++- gliclass/model.py | 89 ++++++++++++++++++++++++++++++++++++++----- gliclass/scorers.py | 90 +++++++++++++++++++++++++++++++++++++------- gliclass/training.py | 37 ++++++++++++++---- 4 files changed, 198 insertions(+), 33 deletions(-) diff --git a/gliclass/config.py b/gliclass/config.py index 524344a..e75e4c5 100644 --- a/gliclass/config.py +++ b/gliclass/config.py @@ -66,8 +66,12 @@ def __init__( recurrent_inference_max_steps=None, recurrent_halt_threshold=0.01, recurrent_improvement_coef=0.1, - recurrent_improvement_margin=0.0, + recurrent_improvement_margin=0.05, recurrent_bptt_steps=None, + recurrent_read_text=False, + recurrent_confidence_coef=0.0, + recurrent_confidence_mode="relative", + recurrent_confidence_margin=0.0, **kwargs, ): if isinstance(encoder_config, dict): @@ -167,8 +171,17 @@ def __init__( self.recurrent_inference_max_steps = recurrent_inference_max_steps self.recurrent_halt_threshold = recurrent_halt_threshold # Hinge relu(l_t - (1 - margin) * sg(l_{t-1})) for steps t >= 3, on top of a loss at every step. + # margin > 0 makes "no change" cost something, so the recurrence cannot settle into a copy of step 2. self.recurrent_improvement_coef = recurrent_improvement_coef self.recurrent_improvement_margin = recurrent_improvement_margin self.recurrent_bptt_steps = recurrent_bptt_steps + # False: recurrent steps see only the label section; True: each step also cross-attends to the text + self.recurrent_read_text = recurrent_read_text + # Confidence loss on recurrent steps (0 = off): push label probabilities away from 0.5. + # "relative": each step's mean label entropy must fall below (1 - margin) * the previous step's; + # "absolute": minimize the entropy of every recurrent step. + self.recurrent_confidence_coef = recurrent_confidence_coef + self.recurrent_confidence_mode = recurrent_confidence_mode + self.recurrent_confidence_margin = recurrent_confidence_margin super().__init__(problem_type=problem_type, **kwargs) diff --git a/gliclass/model.py b/gliclass/model.py index cab2146..237c7b6 100644 --- a/gliclass/model.py +++ b/gliclass/model.py @@ -64,6 +64,7 @@ class GLiClassOutput(SequenceClassifierOutput): past_key_values: Any | None = None recurrent_logits: Tuple[torch.Tensor, ...] | None = None recurrent_losses: torch.Tensor | None = None + recurrent_entropies: torch.Tensor | None = None recurrent_num_steps: torch.Tensor | None = None @@ -1088,23 +1089,44 @@ def _per_sample_loss(self, logits, labels): ) raise NotImplementedError(f"{self.config.problem_type} is not implemented.") - def get_recurrent_loss(self, step_logits, labels): + def _per_sample_entropy(self, logits, valid_labels): + """Mean uncertainty per example over its real labels, shape (batch,). + + Multi-label: binary entropy of each sigmoid probability (0 when p is 0 or 1, max at 0.5). + Single-label: entropy of the softmax over the example's labels. + """ + if self.config.problem_type == "single_label_classification": + log_probs = torch.log_softmax(logits.masked_fill(~valid_labels, float("-inf")), dim=-1) + return -(log_probs.exp() * log_probs.masked_fill(~valid_labels, 0.0)).sum(-1) + probs = torch.sigmoid(logits) + entropy = nn.functional.binary_cross_entropy_with_logits(logits, probs, reduction="none") + return (entropy * valid_labels).sum(-1) / valid_labels.sum(-1).clamp_min(1) + + def get_recurrent_loss(self, step_logits, labels, valid_labels=None): """Loss over all reasoning steps. loss = mean_t L_t (focal loss after every step) + coef * mean_{t>=3} relu(l_t - (1 - margin) * sg(l_{t-1})) (deeper must beat shallower) + + conf_coef * confidence term on recurrent steps t >= 2 (deeper must be more certain) The improvement term starts at step 3 vs step 2, so step 1 is only ever trained by its own loss. The stop-gradient on l_{t-1} means the hinge can only push the deeper step down; it never rewards making the earlier step worse. + Confidence term (recurrent_confidence_coef > 0), with H_t the mean label entropy of step t: + "relative": relu(H_t - (1 - conf_margin) * sg(H_{t-1})) each recurrence less uncertain than the last + "absolute": H_t every recurrence as certain as possible + Returns: loss: scalar step_losses: (num_steps,) detached mean loss per step, for monitoring + step_entropies: (num_steps,) detached mean label entropy per step, for monitoring """ num_labels = step_logits[-1].shape[-1] if labels.shape[-1] != num_labels: labels = labels[:, :num_labels] + if valid_labels is None: + valid_labels = torch.ones_like(step_logits[-1], dtype=torch.bool) per_sample = torch.stack([self._per_sample_loss(logits, labels) for logits in step_logits]) # (T, B) step_losses = per_sample.mean(dim=1) @@ -1115,14 +1137,28 @@ def get_recurrent_loss(self, step_logits, labels): improvement = torch.relu(per_sample[2:] - target).mean() loss = loss + self.config.recurrent_improvement_coef * improvement - return loss, step_losses.detach() + entropies = torch.stack([self._per_sample_entropy(logits, valid_labels) for logits in step_logits]) # (T, B) + confidence_coef = getattr(self.config, "recurrent_confidence_coef", 0.0) + if len(step_logits) >= 2 and confidence_coef > 0: + mode = getattr(self.config, "recurrent_confidence_mode", "relative") + if mode == "relative": + margin = getattr(self.config, "recurrent_confidence_margin", 0.0) + confidence = torch.relu(entropies[1:] - (1.0 - margin) * entropies[:-1].detach()).mean() + elif mode == "absolute": + confidence = entropies[1:].mean() + else: + raise ValueError(f"Unknown recurrent_confidence_mode: {mode}") + loss = loss + confidence_coef * confidence + + return loss, step_losses.detach(), entropies.mean(dim=1).detach() def _extract_label_section( self, hidden_states: torch.Tensor, input_ids: torch.Tensor, attention_mask: torch.Tensor, - ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]: + return_text_section: bool = False, + ) -> tuple[torch.Tensor, ...]: """ Extract label section hidden states from full-sequence decoder output. @@ -1136,12 +1172,15 @@ def _extract_label_section( padded_hidden: (batch, max_label_len, hidden_size) padded_ids: (batch, max_label_len) label_mask: (batch, max_label_len) + with return_text_section, also everything before the label section: + text_hidden: (batch, max_text_len, hidden_size) + text_mask: (batch, max_text_len) """ batch_size = hidden_states.shape[0] sep_id = self.sep_token_id label_id = self.config.class_token_index - slices_h, slices_ids = [], [] + slices_h, slices_ids, text_slices = [], [], [] for i in range(batch_size): real_len = int(attention_mask[i].sum().item()) @@ -1154,6 +1193,7 @@ def _extract_label_section( # Fallback: use everything up to real_len slices_h.append(hidden_states[i, :real_len]) slices_ids.append(ids_i) + text_slices.append(hidden_states[i, :real_len]) continue first_label = label_pos[0].item() @@ -1163,12 +1203,15 @@ def _extract_label_section( if seps_before.numel() == 0: slices_h.append(hidden_states[i, :real_len]) slices_ids.append(ids_i) + text_slices.append(hidden_states[i, :real_len]) continue # start right after that SEP → first token of "label1<