fractus/train/online.py

"""Online trainer for the ContinuousThoughtEngine.

THE TRAINING BREAKTHROUGH. No batches. No BPTT. One observation at a time,
one gradient at a time. The model learns as it "sees" data, like a human.

    for each token in the data stream:
        1. Feed the token to the engine (tick).
        2. The engine produces a prediction + confidence.
        3. Compute the loss (was the prediction right?).
        4. Backward + step IMMEDIATELY (online SGD, 1 sample at a time).
        5. The thought state is carried forward (detached — no BPTT).

WHY THIS IS FAST:
    - Each step processes ONE token (not B×L).
    - The forward is tiny (1 token, 1 tick).
    - The backward is tiny (1 sample).
    - No batching, no padding, no sequence masking.

This is the training method that makes the Continuous Thought Engine
trainable on ANY CPU, because the per-step cost is minimal.
"""

import torch
import torch.nn as nn
import torch.nn.functional as F


class OnlineTrainer:
    """Online trainer for the ContinuousThoughtEngine.

    Args:
        engine:             a ContinuousThoughtEngine.
        lr:                 learning rate.
        weight_decay:       optimizer weight decay.
        accumulation_steps: gradient accumulation (fewer optimizer steps).
        optimizer:          'adamw', 'sgd', or 'rmsprop'.
    """

    def __init__(
        self,
        engine,
        lr: float = 1e-3,
        weight_decay: float = 0.01,
        accumulation_steps: int = 8,
        optimizer: str = "adamw",
    ):
        self.engine = engine
        self.accumulation_steps = max(accumulation_steps, 1)
        if optimizer == "sgd":
            self.optimizer = torch.optim.SGD(engine.parameters(), lr=lr, momentum=0.9)
        elif optimizer == "rmsprop":
            self.optimizer = torch.optim.RMSprop(engine.parameters(), lr=lr, weight_decay=weight_decay)
        else:
            self.optimizer = torch.optim.AdamW(engine.parameters(), lr=lr, weight_decay=weight_decay)
        self.step_count = 0
        self.losses = []

    def train_on_stream(self, token_ids: torch.Tensor, max_ticks: int = 3) -> dict:
        """Train on a stream of tokens, one at a time (pure online, 1 backward/token).

        token_ids: (L,) a 1D tensor of token ids (the data stream).
        max_ticks: max thinking ticks per token.

        Returns a dict with average loss, accuracy, and steps.
        """
        self.engine.train()
        self.engine.reset_thought(batch_size=1)

        total_loss = 0.0
        correct = 0
        total = 0

        for t in range(len(token_ids) - 1):
            obs = token_ids[t:t + 1]  # (1,) current token
            target = token_ids[t + 1]  # scalar, next token

            # Think: tick until confidence or max_ticks.
            for tick in range(max_ticks):
                logits, conf = self.engine.tick(obs if tick == 0 else None)
                if conf.item() > 0.5:
                    break

            # Online loss: did we predict the next token?
            loss = F.cross_entropy(logits, target.unsqueeze(0))

            # Immediate backward + step (online SGD).
            self.optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(self.engine.parameters(), 1.0)
            self.optimizer.step()

            total_loss += loss.item()
            pred = logits.argmax(dim=-1).item()
            if pred == target.item():
                correct += 1
            total += 1
            self.step_count += 1
            self.losses.append(loss.item())

        return {
            "avg_loss": total_loss / max(total, 1),
            "accuracy": correct / max(total, 1),
            "steps": total,
        }

    def train_on_stream_minibatch(self, token_ids: torch.Tensor, max_ticks: int = 2,
                                   accum_steps: int = 16) -> dict:
        """Train on a stream with mini-batch gradient accumulation.

        Accumulates the loss over `accum_steps` tokens, then does ONE backward
        + optimizer step. This is 10-16× faster than train_on_stream (which
        does 1 backward per token) because the Python/autograd overhead is
        amortized over N tokens.

        The thought state is still carried forward (detached between backward
        steps), preserving the continuous-reasoning paradigm.

        Args:
            token_ids:    (L,) 1D tensor of token ids.
            max_ticks:    max thinking ticks per token.
            accum_steps:  tokens per backward pass (16 = 16× fewer backward calls).
        """
        self.engine.train()
        self.engine.reset_thought(batch_size=1)

        total_loss = 0.0
        correct = 0
        total = 0
        accum_loss = torch.tensor(0.0, requires_grad=False)

        for t in range(len(token_ids) - 1):
            obs = token_ids[t:t + 1]
            target = token_ids[t + 1]

            # Think (1 tick per token for speed).
            logits, conf = self.engine.tick(obs)

            # Per-token loss.
            loss = F.cross_entropy(logits, target.unsqueeze(0))
            accum_loss = accum_loss + loss

            total_loss += loss.item()
            pred = logits.argmax(dim=-1).item()
            if pred == target.item():
                correct += 1
            total += 1

            # Backward every accum_steps tokens.
            if (t + 1) % accum_steps == 0:
                self.optimizer.zero_grad()
                avg_loss = accum_loss / accum_steps
                avg_loss.backward()
                torch.nn.utils.clip_grad_norm_(self.engine.parameters(), 1.0)
                self.optimizer.step()
                self.step_count += 1
                self.losses.append(total_loss / total)
                accum_loss = torch.tensor(0.0, requires_grad=False)

        # Final partial accumulation.
        if total % accum_steps != 0 and isinstance(accum_loss, torch.Tensor) and accum_loss.requires_grad:
            self.optimizer.zero_grad()
            (accum_loss / (total % accum_steps)).backward()
            self.optimizer.step()
            self.step_count += 1

        return {
            "avg_loss": total_loss / max(total, 1),
            "accuracy": correct / max(total, 1),
            "steps": total,
            "optimizer_steps": self.step_count,
        }

    def train_on_stream_chunked(self, token_ids: torch.Tensor,
                                chunk_len: int = 16) -> dict:
        """Train using chunk-based processing (16x fewer forward passes).

        Splits the stream into chunks of `chunk_len` tokens. Each chunk is
        processed in ONE forward pass (tick_chunk), then ONE backward.
        This is the FASTEST training mode — the forward/backward overhead
        is amortized over chunk_len tokens.

        The thought state (S,z) is carried between chunks (detached).

        Args:
            token_ids:  (L,) 1D tensor.
            chunk_len:  tokens per chunk (16 = 16× fewer forward passes).
        """
        self.engine.train()
        self.engine.reset_thought(batch_size=1)
        vocab = self.engine.vocab_size

        total_loss = 0.0
        correct = 0
        total = 0

        # Gradient accumulation: backward every chunk, step every accumulation_steps chunks.
        accum = self.accumulation_steps
        self.optimizer.zero_grad()
        chunk_idx = 0

        for start in range(0, len(token_ids) - chunk_len - 1, chunk_len):
            chunk = token_ids[start:start + chunk_len].unsqueeze(0)  # (1, C)

            # Fast training path: head on LAST position only.
            last_logits = self.engine.tick_chunk_train(chunk)  # (1, vocab)
            target = token_ids[start + chunk_len]  # scalar: the token after the chunk

            # CE on the single predicted token (scaled by 1/accum for correct averaging).
            loss = F.cross_entropy(last_logits, target.unsqueeze(0)) / accum

            # Backward every chunk (gradients accumulate).
            loss.backward()

            total_loss += loss.item() * accum  # un-scale for logging
            pred = last_logits.argmax(dim=-1)
            correct += (pred == target.unsqueeze(0)).sum().item()
            total += 1
            self.losses.append(loss.item() * accum)

            chunk_idx += 1

            # Step only every accumulation_steps chunks.
            if chunk_idx % accum == 0:
                torch.nn.utils.clip_grad_norm_(self.engine.parameters(), 1.0)
                self.optimizer.step()
                self.optimizer.zero_grad()
                self.step_count += 1

        # Handle the remainder: if chunks aren't a clean multiple of accum, do a final step.
        if chunk_idx % accum != 0:
            torch.nn.utils.clip_grad_norm_(self.engine.parameters(), 1.0)
            self.optimizer.step()
            self.optimizer.zero_grad()
            self.step_count += 1

        return {
            "avg_loss": total_loss / max(total, 1),
            "accuracy": correct / max(total, 1),
            "steps": total,
            "optimizer_steps": self.step_count,
        }

    def train_step_batch(self, input_ids: torch.Tensor, target_ids: torch.Tensor,
                         max_ticks: int = 3) -> dict:
        """Train on a small batch using the think() method.

        input_ids:  (B, L) token ids.
        target_ids: (B, L) next-token targets.
        """
        self.engine.train()
        self.engine.reset_thought(batch_size=input_ids.shape[0])

        # Use think() to process the whole sequence.
        logits = self.engine.think(input_ids, max_ticks=max_ticks, confidence_threshold=0.5)
        # The logits are (B, L, vocab) — but think() only produces output when
        # confident. For training we compute loss on ALL positions.
        loss = F.cross_entropy(
            logits.reshape(-1, self.engine.vocab_size),
            target_ids.reshape(-1),
        )

        self.optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.engine.parameters(), 1.0)
        self.optimizer.step()
        self.step_count += 1

        return {"loss": loss.item()}