Method _do_train

backend/python/liquid-audio/backend.py:641–741

(self, request, job)

            job.progress_queue.put(None)

    def _do_train(self, request, job):
        from liquid_audio import LFM2AudioModel  # noqa: F401  (sanity import)
        from liquid_audio.data.dataloader import LFM2DataLoader
        from liquid_audio.trainer import Trainer

        model_id = request.model or self.model_id or "LiquidAI/LFM2.5-Audio-1.5B"

        dataset_path = request.dataset_source
        if not dataset_path:
            raise ValueError("dataset_source is required (path to a preprocessed dataset)")

        extras = dict(request.extra_options) if request.extra_options else {}
        val_path = extras.get("val_dataset")

        # Map FineTuneRequest hyperparameters to liquid_audio.Trainer constructor args
        lr = request.learning_rate or 3e-5
        max_steps = request.max_steps or 1000
        warmup_steps = request.warmup_steps or min(100, max_steps // 10)
        batch_size = request.batch_size or 16
        save_interval = request.save_steps or max(1, max_steps // 4)

        output_dir = request.output_dir or os.path.join(
            os.environ.get("LIQUID_AUDIO_OUTPUT_DIR", "/tmp"),
            f"liquid-audio-{job.job_id}",
        )
        os.makedirs(output_dir, exist_ok=True)

        job.progress_queue.put(backend_pb2.FineTuneProgressUpdate(
            job_id=job.job_id, status="loading_dataset",
            message=f"Loading preprocessed dataset from {dataset_path}",
        ))
        train_data = LFM2DataLoader(dataset_path)
        val_data = LFM2DataLoader(val_path) if val_path else None

        job.progress_queue.put(backend_pb2.FineTuneProgressUpdate(
            job_id=job.job_id, status="loading_model",
            message=f"Loading base model {model_id}",
        ))

        # The Liquid Trainer logs via self.accelerator.print; we subclass it to
        # also push progress events onto the queue every logging_interval steps.
        progress_q = job.progress_queue

        class QueuedTrainer(Trainer):
            def log(self_, model_output):
                if self_.step > 0 and self_.step % self_.logging_interval == 0:
                    try:
                        loss = self_.accelerator.reduce(
                            model_output.loss.detach(), reduction="mean"
                        ).item()
                    except Exception:
                        loss = float("nan")
                    lr_now = self_.optimizer.param_groups[0]["lr"]
                    pct = (self_.step / self_.max_steps * 100.0) if self_.max_steps else 0.0
                    progress_q.put(backend_pb2.FineTuneProgressUpdate(
                        job_id=job.job_id,
                        current_step=int(self_.step),
                        total_steps=int(self_.max_steps),