from __future__ import annotations import math import os import importlib.util from dataclasses import dataclass import torch import torch.nn.functional as F from huggingface_hub import login from transformers import AutoConfig, AutoModelForCausalLM, AutoTokenizer from .checkpoints import load_token_checkpoint from .config import PipelineConfig def _load_causal_lm(model_name: str, **kwargs): """Load a causal LM, falling back to architecture-specific classes when AutoModelForCausalLM doesn't recognize the model type.""" try: return AutoModelForCausalLM.from_pretrained(model_name, **kwargs) except (ValueError, ModuleNotFoundError): config = AutoConfig.from_pretrained(model_name) model_type = getattr(config, "model_type", "") if model_type == "gemma3": from transformers import Gemma3ForConditionalGeneration return Gemma3ForConditionalGeneration.from_pretrained(model_name, **kwargs) if model_type == "gemma4": from transformers import Gemma4ForConditionalGeneration return Gemma4ForConditionalGeneration.from_pretrained(model_name, **kwargs) raise def resolve_hf_token() -> str | None: return os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_TOKEN") def maybe_login_hf() -> None: token = resolve_hf_token() if token: login(token=token, add_to_git_credential=False) def resolve_dtype() -> torch.dtype: return torch.bfloat16 if torch.cuda.is_available() else torch.float32 def has_accelerate() -> bool: return importlib.util.find_spec("accelerate") is not None @dataclass(slots=True) class ModelBundle: config: PipelineConfig tokenizer: AutoTokenizer model: AutoModelForCausalLM initial_tokenizer_len: int def initialize_model_bundle(config: PipelineConfig) -> ModelBundle: maybe_login_hf() hf_token = resolve_hf_token() dtype = resolve_dtype() tokenizer = AutoTokenizer.from_pretrained(config.model.model_name, use_fast=True, token=hf_token) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token initial_len = len(tokenizer) tokenizer.add_special_tokens( {"additional_special_tokens": [config.model.ai_token, config.model.human_token]} ) load_kwargs = { "token": hf_token, "dtype": dtype, } if torch.cuda.is_available() and has_accelerate(): load_kwargs["device_map"] = "auto" model = _load_causal_lm(config.model.model_name, **load_kwargs) if torch.cuda.is_available() and not has_accelerate(): model = model.to("cuda") elif not torch.cuda.is_available(): model = model.to("cpu") try: model.resize_token_embeddings(len(tokenizer), mean_resizing=False) except TypeError: model.resize_token_embeddings(len(tokenizer)) # Mean-fill Gemma 4's auxiliary per-layer token table. Some Transformers # versions resize it internally, while others leave it at the old size. for name, module in model.named_modules(): if isinstance(module, torch.nn.Embedding) and module is not model.get_input_embeddings(): if module.weight.shape[0] == initial_len: mean_row = module.weight.data[:initial_len].mean( dim=0, dtype=torch.float32 ).to(dtype=module.weight.dtype) new_emb = torch.nn.Embedding( len(tokenizer), module.weight.shape[1], device=module.weight.device, dtype=module.weight.dtype, ) new_emb.weight.data[:initial_len] = module.weight.data new_emb.weight.data[initial_len:] = mean_row if not torch.equal( new_emb.weight.data[initial_len], mean_row ): raise RuntimeError( f"Failed to mean-fill resized embedding {name}" ) parent_name, attr_name = name.rsplit(".", 1) parent = dict(model.named_modules())[parent_name] setattr(parent, attr_name, new_emb) print( f"Mean-filled secondary embedding {name}: " f"{initial_len} -> {len(tokenizer)}" ) elif module.weight.shape[0] == len(tokenizer): mean_row = module.weight.data[:initial_len].mean( dim=0, dtype=torch.float32 ).to(dtype=module.weight.dtype) module.weight.data[initial_len:] = mean_row if not torch.equal(module.weight.data[initial_len], mean_row): raise RuntimeError( f"Failed to mean-fill expanded embedding {name}" ) print( f"Mean-filled expanded secondary embedding {name}: " f"rows {initial_len}:{len(tokenizer)}" ) model.config.use_cache = False if hasattr(model, "gradient_checkpointing_enable"): model.gradient_checkpointing_enable() model.eval() input_emb = model.get_input_embeddings() with torch.no_grad(): mean_in = input_emb.weight[:initial_len].mean(dim=0) for token in (config.model.ai_token, config.model.human_token): token_id = tokenizer.convert_tokens_to_ids(token) input_emb.weight[token_id].copy_(mean_in + torch.randn_like(mean_in) * 1e-5) bundle = ModelBundle( config=config, tokenizer=tokenizer, model=model, initial_tokenizer_len=initial_len, ) apply_initial_checkpoints(bundle) return bundle def apply_initial_checkpoints(bundle: ModelBundle) -> None: input_emb = bundle.model.get_input_embeddings() token_dir = bundle.config.output.model_tokens_dir(bundle.config.model.model_name) ai_path = bundle.config.init_checkpoints.ai_token_path or token_dir / "ai_token.pt" human_path = bundle.config.init_checkpoints.human_token_path or token_dir / "human_token.pt" secondary_embs = [ module for module in bundle.model.modules() if ( isinstance(module, torch.nn.Embedding) and module is not input_emb and module.weight.shape[0] == len(bundle.tokenizer) ) ] def install(path, token): if not path.exists(): return checkpoint = load_token_checkpoint(path) token_id = bundle.tokenizer.convert_tokens_to_ids(token) saved_secondary = checkpoint.secondary_embeddings or [] if len(saved_secondary) != len(secondary_embs): raise ValueError( f"{path} has {len(saved_secondary)} secondary rows, but " f"{bundle.config.model.model_name} exposes " f"{len(secondary_embs)} secondary token embeddings." ) input_emb.weight[token_id].copy_( checkpoint.embedding.to( input_emb.weight.device, dtype=input_emb.weight.dtype ) ) for embedding, row in zip(secondary_embs, saved_secondary): embedding.weight[token_id].copy_( row.to(embedding.weight.device, dtype=embedding.weight.dtype) ) with torch.no_grad(): install(ai_path, bundle.config.model.ai_token) install(human_path, bundle.config.model.human_token) def build_prompt(bundle: ModelBundle, token: str) -> str: content = bundle.config.model.prompt_template.format(token=token) messages = [{"role": "user", "content": content}] return bundle.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) def encode_response(bundle: ModelBundle, prompt_text: str, response_text: str) -> tuple[torch.Tensor, int]: full_text = prompt_text + response_text prompt_ids = bundle.tokenizer(prompt_text, return_tensors="pt", add_special_tokens=False)["input_ids"][0] full_ids = bundle.tokenizer( full_text, return_tensors="pt", add_special_tokens=False, truncation=True, max_length=bundle.config.model.max_length, )["input_ids"][0] return full_ids, int(len(prompt_ids)) def compute_average_logprob(bundle: ModelBundle, input_ids: torch.Tensor, prompt_len: int) -> torch.Tensor: token_logps = compute_token_logprobs(bundle, input_ids, prompt_len) return token_logps.mean() def _model_forward(bundle: ModelBundle, input_ids: torch.Tensor, attention_mask: torch.Tensor | None = None): """Run a forward pass, injecting token_type_ids for models that require it.""" fwd_kwargs: dict = {"input_ids": input_ids} if attention_mask is not None: fwd_kwargs["attention_mask"] = attention_mask model_type = getattr(bundle.model.config, "model_type", "") if model_type in ("gemma3", "gemma4"): fwd_kwargs["token_type_ids"] = torch.zeros_like(input_ids) return bundle.model(**fwd_kwargs) def compute_token_logprobs(bundle: ModelBundle, input_ids: torch.Tensor, prompt_len: int) -> torch.Tensor: input_ids = input_ids.unsqueeze(0).to(bundle.model.device) logits = _model_forward(bundle, input_ids).logits[0] shift_logits = logits[prompt_len - 1 : -1] shift_labels = input_ids[0, prompt_len:] log_probs = F.log_softmax(shift_logits, dim=-1) return log_probs[torch.arange(len(shift_labels), device=bundle.model.device), shift_labels] def compute_sequence_logprob(bundle: ModelBundle, input_ids: torch.Tensor, prompt_len: int) -> torch.Tensor: input_ids = input_ids.unsqueeze(0).to(bundle.model.device) logits = _model_forward(bundle, input_ids).logits[0] shift_logits = logits[prompt_len - 1 : -1] shift_labels = input_ids[0, prompt_len:] log_probs = F.log_softmax(shift_logits, dim=-1) token_logps = log_probs[torch.arange(len(shift_labels), device=bundle.model.device), shift_labels] return token_logps.sum() def cosine_with_floor( step: int, total_steps: int, base_lr: float, *, min_lr: float, warmup_steps: int, ) -> float: if step < warmup_steps: return base_lr * float(step + 1) / float(max(1, warmup_steps)) progress = float(step - warmup_steps) / float(max(1, total_steps - warmup_steps)) cosine = 0.5 * (1.0 + math.cos(math.pi * progress)) return min_lr + (base_lr - min_lr) * cosine