tenseleyflow/documentlanguagemodel / 0af0f7c

 """Training engine — load base, attach LoRA/QLoRA, run one SFT cycle.

 """

 from __future__ import annotations

   - `"resume"` → `PeftModel.from_pretrained(model, resume_path, is_trainable=True)`.

 - `apply_kbit_preparation(model, gradient_checkpointing)` runs

   `prepare_model_for_kbit_training` for the QLoRA path. MUST be called

 The trainer's higher-level orchestrator composes these — we keep the

 individual functions small + testable.

 ) -> Any:

     """Return a `peft.LoraConfig` sized for `spec`.

     otherwise the new embedding row is undefined. `modules_to_save`

     inflates the adapter checkpoint size substantially; surfacing this

     `use_dora=True` switches from plain LoRA to DoRA (weight-

     decomposed low-rank adaptation). DoRA factors each weight update

 def verify_resume_tokenizer_compat(adapter_dir: Path, *, tokenizer_grew: bool) -> None:

     """Assert the saved adapter's `modules_to_save` agrees with the current tokenizer.

     Raises `ResumeIntegrityError` with actionable text on mismatch. Missing

     or unreadable `adapter_config.json` is treated as a mismatch (the

    `training_state.pt` + sha256.

 3. `fsync_dir(path)` flushes the directory entry to disk.

 4. `store.set_current_adapter(path)` atomically flips the pointer via

 The `commit_version()` helper bundles steps 1 + 3 + 4 around a

 caller-supplied writer function, so the "happy path" is one call. If

 6. `torch.backends.cudnn.benchmark = False` — no autotuner (which is

    non-deterministic).

 Heavy imports (`torch`, `numpy`) are deferred to call-sites so

 `import dlm.train` stays cheap.

         # the trainer logs a WARN for any that trigger.

         torch.use_deterministic_algorithms(True, warn_only=True)

         torch.backends.cudnn.benchmark = False

         notes.append("torch not installed; determinism contract not enforced")

         class_ = "loose"

 """Distributed (multi-GPU) training via HuggingFace Accelerate.

 DeepSpeed.

 Surface:

 def strip_gpus_flag(args: list[str], *, skip_argv0: bool = False) -> list[str]:

     """Drop `--gpus <value>` / `--gpus=<value>` from an argv-like list.

     `skip_argv0` flag controls which input convention is used.

     """

     start = 1 if skip_argv0 else 0

     # We call gather per-metric to avoid building a tensor for a

     # heterogeneous dict. For a shape-stable dict of floats this is

     import torch

     reduced: dict[str, float] = {}

 Accelerate spawns one process per GPU; each invokes this module. The

 worker re-parses the subset of CLI args it cares about (path, seed,

 max_steps, resume/fresh, phase) and routes into the existing

 Full DDP integration (refactoring `trainer.run` to gate its I/O via

 """

 from __future__ import annotations

     """Drop `--gpus <value>` / `--gpus=<value>` from argv (worker side).

     Per-rank invocations must not recurse into the launcher branch of

     """

     from dlm.train.distributed.gpus import strip_gpus_flag

     fewer than two adapters; callers don't need to check the config

     themselves. Embedding is injected as a callable so tests can stub

     it without loading an HF model. Any `GateTrainingError` is logged

     """

     training = parsed.frontmatter.training

     gate_cfg = training.gate

 _AUDIO_MODEL_CLASSES: dict[str, str] = {

     # Maps `BaseModelSpec.architecture` → transformers class name.

     "Qwen2AudioForConditionalGeneration": "Qwen2AudioForConditionalGeneration",

 }

 """Structured per-step JSONL logger for training runs.

 One file per run at `logs/train-<run_id>-<started_at>.jsonl`. Every

 Design

 ------

   versions, and the training-plan snapshot. Followed by `"type":

   "step"` records with `step`, `loss`, `lr`, `grad_norm`,

   `tokens_per_sec`, and optional `val_loss` on eval steps.

 """

 from __future__ import annotations

 point regardless of document shape.

 Scope note: inference selection, export merge, and doctor memory

 orchestration and the resulting adapter-versioned store layout.

 """

     """Run the post-SFT learned-gate training pass when enabled.

     Kept separate so the multi-adapter orchestrator's happy path stays

     """

     import logging

 ) -> tuple[Callable[[str], Any], int]:  # pragma: no cover — heavy HF path

     """Default embedder — loads the HF base model + tokenizer.

     """

     from transformers import AutoTokenizer

     Callers on non-CUDA devices can skip this — re-raising a

     non-CUDA OOM as an `OOMError` would be actively misleading.

     """

     import torch

     try:

 RNG state out of the torch payload into a JSON sidecar, dropping

 `weights_only=False` from the load path. The writer always emits

 v2; the reader accepts v1 (legacy) with a migration warning."""

 TRAINING_RUN_FILENAME = "training_run.json"

     dlm_manifest_hash: str | None

     base_model_revision: str

     pinned_versions: PinnedVersions

     use_qlora: bool

         json.dumps(dict(state["pinned_versions"]), sort_keys=True, indent=2) + "\n",

     )

     training_run_path = directory / TRAINING_RUN_FILENAME

     write_text(

         training_run_path,

     try:

         torch_payload = torch.load(io.BytesIO(blob), weights_only=True)

     except Exception as weights_only_exc:

         try:

             torch_payload = torch.load(io.BytesIO(blob), weights_only=False)

         except Exception as exc:

 def _version_diff(pinned: PinnedVersions, runtime: PinnedVersions) -> list[str]:

     """Return `["key: saved→current", ...]` for keys whose versions differ.

     Rules:

     - saved=str, current=str, equal    → no drift

     Rows that carry neither ``messages`` nor ``text`` — preference

     rows destined for DPOTrainer — pass through untouched. DPOTrainer

     """

     sha = tokenizer_sha256(tokenizer) if cache is not None else ""

     stats_hits = 0