From 678f84229ebfe2e154abf34b4069095dd1e1efb9 Mon Sep 17 00:00:00 2001 From: Ramanathan Arunachalam Date: Sat, 1 Aug 2026 04:47:22 +0000 Subject: [PATCH 01/13] feat: add TAO WTS fine-tuning support --- Dockerfile | 8 +- cosmos_framework/callbacks/tao_status.py | 381 ++++++++++++++++++ cosmos_framework/callbacks/tao_status_test.py | 86 ++++ cosmos_framework/checkpoint/base.py | 5 +- cosmos_framework/checkpoint/dcp.py | 9 +- cosmos_framework/checkpoint/dcp_distill.py | 13 +- cosmos_framework/checkpoint/dummy.py | 1 + .../base/reasoner/defaults/callbacks.py | 17 +- .../base/reasoner/experiment/wts_vlm.py | 294 ++++++++++++++ .../base/reasoner/experiment/wts_vlm_test.py | 56 +++ .../configs/toml_config/sft_config.py | 103 +++-- .../configs/toml_config/sft_config_test.py | 18 + cosmos_framework/scripts/train.py | 26 +- cosmos_framework/trainer/__init__.py | 83 +++- cosmos_framework/utils/callback.py | 9 +- cosmos_framework/utils/checkpointer.py | 10 +- cosmos_framework/utils/config.py | 10 + .../utils/reasoner/dcp_checkpointer.py | 7 +- examples/launch_sft_wts.sh | 32 ++ examples/toml/sft_config/wts_vlm.toml | 99 +++++ examples/toml/sft_config/wts_vlm_edge.toml | 95 +++++ uv.lock | 10 +- 22 files changed, 1292 insertions(+), 80 deletions(-) create mode 100644 cosmos_framework/callbacks/tao_status.py create mode 100644 cosmos_framework/callbacks/tao_status_test.py create mode 100644 cosmos_framework/configs/base/reasoner/experiment/wts_vlm.py create mode 100644 cosmos_framework/configs/base/reasoner/experiment/wts_vlm_test.py create mode 100755 examples/launch_sft_wts.sh create mode 100644 examples/toml/sft_config/wts_vlm.toml create mode 100644 examples/toml/sft_config/wts_vlm_edge.toml diff --git a/Dockerfile b/Dockerfile index 529ddb2c..ea474be5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -49,9 +49,15 @@ RUN --mount=type=cache,target=/root/.cache/uv \ --mount=type=bind,source=pyproject.toml,target=pyproject.toml \ --mount=type=bind,source=.python-version,target=.python-version \ --mount=type=bind,source=packages,target=packages \ - uv sync --locked --no-install-project --no-editable --all-extras --group=$(cat /root/.cuda-name) --group=vllm + uv sync --locked --no-install-project --no-editable --all-extras --group=$(cat /root/.cuda-name)-train ENV PATH="/workspace/.venv/bin:$PATH" +# Package the exact source state into the image so it can run on managed +# platforms without a host bind mount. +COPY . /workspace +RUN --mount=type=cache,target=/root/.cache/uv \ + uv pip install --no-deps . + # Triton bundled ptxas doesn't support latest GPU architectures ENV TRITON_PTXAS_PATH="/usr/local/cuda/bin/ptxas" diff --git a/cosmos_framework/callbacks/tao_status.py b/cosmos_framework/callbacks/tao_status.py new file mode 100644 index 00000000..d318d076 --- /dev/null +++ b/cosmos_framework/callbacks/tao_status.py @@ -0,0 +1,381 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""TAO-compatible lifecycle and metric logging for Cosmos Framework training.""" + +from __future__ import annotations + +import json +import os +import time +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +import torch +import torch.distributed as dist + +from cosmos_framework.utils import distributed, log, misc +from cosmos_framework.utils.callback import Callback + + +def _to_json_value(value: Any) -> Any: + """Recursively convert tensors and array-like values to JSON-safe values.""" + if isinstance(value, dict): + return {str(key): _to_json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_to_json_value(item) for item in value] + if isinstance(value, torch.Tensor): + value = value.detach() + if value.numel() == 1: + return value.item() + return value.cpu().tolist() + if hasattr(value, "item"): + try: + return value.item() + except (TypeError, ValueError): + pass + if hasattr(value, "tolist"): + try: + return value.tolist() + except (TypeError, ValueError): + pass + return value + + +class _TAOStatusWriter: + """Use TAO Core when available, with a compatible JSON-lines fallback.""" + + def __init__(self, filename: str) -> None: + self.filename = filename + Path(filename).parent.mkdir(parents=True, exist_ok=True) + self._tao_logger = None + self._tao_status = None + self._tao_verbosity = None + + try: + from nvidia_tao_core.loggers.logging import Status, StatusLogger, Verbosity + except ImportError: + log.warning(f"nvidia_tao_core is not installed; writing TAO-compatible JSON records directly to {filename}") + else: + self._tao_status = Status + self._tao_verbosity = Verbosity + self._tao_logger = StatusLogger( + filename=filename, + is_master=True, + verbosity=Verbosity.INFO, + append=True, + ) + + def write( + self, + *, + status: str, + message: str, + data: dict[str, Any] | None = None, + kpi: dict[str, Any] | None = None, + verbosity: str = "INFO", + ) -> None: + data = _to_json_value(data or {}) + kpi = _to_json_value(kpi or {}) + + if self._tao_logger is not None: + status_level = getattr(self._tao_status, status) + verbosity_level = getattr(self._tao_verbosity, verbosity) + self._tao_logger.kpi = kpi + self._tao_logger.write( + data=data, + status_level=status_level, + verbosity_level=verbosity_level, + message=message, + ) + return + + now = datetime.now() + payload: dict[str, Any] = { + **data, + "date": f"{now.month}/{now.day}/{now.year}", + "time": f"{now.hour}:{now.minute}:{now.second}", + "status": status, + "verbosity": verbosity, + "message": message, + } + if kpi: + payload["kpi"] = kpi + with open(self.filename, "a", encoding="utf-8") as status_file: + status_file.write(json.dumps(payload, default=str) + "\n") + + +class TAOStatusCallback(Callback): + """Write TAO lifecycle, training, and validation records from rank zero. + + The output path is resolved in this order: + + 1. ``status_file_path`` when explicitly configured. + 2. ``$TAO_RESULTS_ROOT/$TAO_JOB_ID/status.json`` (TAO SDK). + 3. ``$TAO_API_RESULTS_DIR/$TAO_API_JOB_ID/status.json`` (TAO API). + 4. ``/status.json`` for direct launches. + """ + + def __init__( + self, + enabled: bool = False, + status_file_path: str | None = None, + experiment_name: str = "", + logging_interval: int = 1, + validation_heartbeat_interval: int = 1, + ) -> None: + if logging_interval < 1: + raise ValueError("logging_interval must be >= 1") + if validation_heartbeat_interval < 1: + raise ValueError("validation_heartbeat_interval must be >= 1") + self.enabled = enabled + self.status_file_path = status_file_path + self.experiment_name = experiment_name + self.logging_interval = logging_interval + self.validation_heartbeat_interval = validation_heartbeat_interval + self._writer: _TAOStatusWriter | None = None + self._train_start_time = 0.0 + self._step_start_time = 0.0 + self._validation_batches = 0 + self._validation_loss_sum = 0.0 + self._validation_sample_count = 0 + self.last_validation_loss: float | None = None + + def _is_rank_zero(self) -> bool: + if dist.is_available() and dist.is_initialized(): + return distributed.is_rank0() + return int(os.environ.get("RANK", os.environ.get("LOCAL_RANK", "0"))) == 0 + + def _component_name(self) -> str: + if self.experiment_name: + return self.experiment_name + return getattr(self.config.job, "name", "Cosmos Framework SFT") + + def _resolve_status_file(self) -> str: + if self.status_file_path: + return self.status_file_path + + job_id = os.environ.get("TAO_JOB_ID") + if job_id: + results_root = os.environ.get("TAO_RESULTS_ROOT", "/results") + return os.path.join(results_root, job_id, "status.json") + + api_job_id = os.environ.get("TAO_API_JOB_ID") + if api_job_id: + results_root = os.environ.get("TAO_API_RESULTS_DIR", "/results") + return os.path.join(results_root, api_job_id, "status.json") + + return os.path.join(self.config.job.path_local, "status.json") + + def _get_writer(self) -> _TAOStatusWriter | None: + if not self.enabled or not self._is_rank_zero(): + return None + if self._writer is None: + self._writer = _TAOStatusWriter(self._resolve_status_file()) + return self._writer + + def _progress_data(self, iteration: int, seconds_per_step: float | None = None) -> dict[str, Any]: + trainer = getattr(self, "trainer", None) + max_step = int(getattr(trainer, "max_iterations", self.config.trainer.max_iter)) + if seconds_per_step is None: + elapsed = max(time.monotonic() - self._train_start_time, 0.0) + seconds_per_step = elapsed / max(iteration, 1) + eta_seconds = max(max_step - iteration, 0) * seconds_per_step + data = { + "component": self._component_name(), + "step": iteration, + "max_step": max_step, + "time_per_step": str(timedelta(seconds=seconds_per_step)), + "eta": str(timedelta(seconds=eta_seconds)), + } + steps_per_epoch = getattr(trainer, "steps_per_epoch", None) + num_epochs = getattr(trainer, "num_epochs", None) + if steps_per_epoch and num_epochs: + completed_epochs = iteration // steps_per_epoch + if iteration == 0: + epoch = 1 + step_in_epoch = 0 + elif iteration % steps_per_epoch == 0: + epoch = min(completed_epochs, num_epochs) + step_in_epoch = steps_per_epoch + else: + epoch = min(completed_epochs + 1, num_epochs) + step_in_epoch = iteration % steps_per_epoch + data.update( + { + "epoch": epoch, + "max_epoch": num_epochs, + "completed_epochs": min(completed_epochs, num_epochs), + "step_in_epoch": step_in_epoch, + "steps_per_epoch": steps_per_epoch, + "time_per_epoch": str(timedelta(seconds=seconds_per_step * steps_per_epoch)), + } + ) + return data + + def _epoch_label(self, iteration: int) -> str: + progress = self._progress_data(iteration, seconds_per_step=0.0) + if "epoch" not in progress: + return f"training step {iteration}/{progress['max_step']}" + return f"epoch {progress['epoch']}/{progress['max_epoch']}" + + @staticmethod + def _global_average_loss(loss: torch.Tensor, data_batch: dict[str, Any]) -> tuple[float, int]: + sample_count = misc.get_data_batch_size(data_batch) + loss_sum = loss.detach() * sample_count + count = torch.tensor(sample_count, device=loss.device, dtype=torch.long) + if dist.is_available() and dist.is_initialized(): + dist.all_reduce(loss_sum, op=dist.ReduceOp.SUM) + dist.all_reduce(count, op=dist.ReduceOp.SUM) + global_count = int(count.item()) + average = float(loss_sum.item()) / max(global_count, 1) + return average, global_count + + def on_train_start(self, model: Any, iteration: int = 0) -> None: + self._train_start_time = time.monotonic() + writer = self._get_writer() + if writer is not None: + writer.write( + status="STARTED", + message=f"Starting {self._component_name()} training", + data=self._progress_data(iteration, seconds_per_step=0.0), + ) + log.info(f"TAO status will be logged to {writer.filename}") + + def on_training_step_start(self, model: Any, data: dict[str, Any], iteration: int = 0) -> None: + self._step_start_time = time.monotonic() + + def on_training_step_end( + self, + model: Any, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + loss: torch.Tensor, + iteration: int = 0, + ) -> None: + interval = int(self.config.trainer.logging_iter) * self.logging_interval + if iteration % interval != 0: + return + + average_loss, _ = self._global_average_loss(loss, data_batch) + writer = self._get_writer() + if writer is None: + return + seconds_per_step = max(time.monotonic() - self._step_start_time, 0.0) + kpi = {"train/loss": average_loss, "train/loss_avg": average_loss} + progress = self._progress_data(iteration, seconds_per_step=seconds_per_step) + if "epoch" in progress: + message = ( + f"Training epoch {progress['epoch']}/{progress['max_epoch']}, " + f"step {progress['step_in_epoch']}/{progress['steps_per_epoch']} " + f"(global step {iteration}/{progress['max_step']}) - Loss: {average_loss:.6f}" + ) + else: + message = f"Training step {iteration}/{progress['max_step']} - Loss: {average_loss:.6f}" + writer.write( + status="RUNNING", + message=message, + data=progress, + kpi=kpi, + ) + + def on_validation_start(self, model: Any, dataloader_val: Any, iteration: int = 0) -> None: + self._validation_batches = 0 + self._validation_loss_sum = 0.0 + self._validation_sample_count = 0 + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=f"Starting validation for {self._epoch_label(iteration)}", + data={**self._progress_data(iteration), "phase": "validation_starting"}, + ) + log.info(f"Starting validation for {self._epoch_label(iteration)}") + + def on_validation_step_end( + self, + model: Any, + data_batch: dict[str, Any], + output_batch: dict[str, Any], + loss: torch.Tensor, + iteration: int = 0, + ) -> None: + average_loss, sample_count = self._global_average_loss(loss, data_batch) + self._validation_batches += 1 + self._validation_loss_sum += average_loss * sample_count + self._validation_sample_count += sample_count + + if self._validation_batches % self.validation_heartbeat_interval != 0: + return + writer = self._get_writer() + max_validation_batches = getattr(self.config.trainer, "max_val_iter", None) + batch_progress = ( + f"{self._validation_batches}/{max_validation_batches}" + if max_validation_batches is not None + else str(self._validation_batches) + ) + if writer is not None: + writer.write( + status="RUNNING", + message=( + f"Validation {self._epoch_label(iteration)}, batch {batch_progress} - Loss: {average_loss:.6f}" + ), + data={ + **self._progress_data(iteration), + "phase": "validation_batch_complete", + "validation_batch": self._validation_batches, + "max_validation_batches": max_validation_batches, + }, + kpi={"val/batch_loss": average_loss}, + ) + log.info(f"Validation {self._epoch_label(iteration)}, batch {batch_progress} - Loss: {average_loss:.6f}") + + def on_validation_end(self, model: Any, iteration: int = 0) -> None: + if self._validation_sample_count == 0: + log.warning("TAO validation logging saw zero samples; no val/loss record was written") + return + + self.last_validation_loss = self._validation_loss_sum / self._validation_sample_count + writer = self._get_writer() + if writer is not None: + writer.write( + status="RUNNING", + message=( + f"Validation complete for {self._epoch_label(iteration)} - Loss: {self.last_validation_loss:.6f}" + ), + data={ + **self._progress_data(iteration), + "phase": "validation_complete", + "validation_batches": self._validation_batches, + "validation_samples": self._validation_sample_count, + }, + kpi={"val/loss": self.last_validation_loss, "val/avg_loss": self.last_validation_loss}, + ) + log.info(f"Validation loss ({self._epoch_label(iteration)}): {self.last_validation_loss:.6f}") + + def on_app_end(self) -> None: + writer = self._get_writer() + if writer is not None: + writer.write( + status="SUCCESS", + message=f"{self._component_name()} training completed successfully", + data=self._progress_data( + int(getattr(getattr(self, "trainer", None), "max_iterations", self.config.trainer.max_iter)) + ), + kpi=( + {"val/loss": self.last_validation_loss, "val/avg_loss": self.last_validation_loss} + if self.last_validation_loss is not None + else None + ), + ) + + def on_exception(self, error: BaseException) -> None: + writer = self._get_writer() + if writer is not None: + writer.write( + status="FAILURE", + verbosity="ERROR", + message=f"{self._component_name()} training failed: {error}", + data={"component": self._component_name(), "error_type": type(error).__name__}, + ) diff --git a/cosmos_framework/callbacks/tao_status_test.py b/cosmos_framework/callbacks/tao_status_test.py new file mode 100644 index 00000000..f862487c --- /dev/null +++ b/cosmos_framework/callbacks/tao_status_test.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import torch + +from cosmos_framework.callbacks.tao_status import TAOStatusCallback + + +def _callback(tmp_path) -> TAOStatusCallback: + callback = TAOStatusCallback( + enabled=True, + status_file_path=str(tmp_path / "status.json"), + experiment_name="test", + ) + callback.config = SimpleNamespace( + job=SimpleNamespace(name="job", path_local=str(tmp_path)), + trainer=SimpleNamespace(max_iter=6, max_val_iter=2, logging_iter=1), + ) + callback.trainer = SimpleNamespace(max_iterations=6, num_epochs=2, steps_per_epoch=3) + return callback + + +def _records(tmp_path) -> list[dict]: + return [json.loads(line) for line in (tmp_path / "status.json").read_text(encoding="utf-8").splitlines()] + + +def test_tao_status_callback_writes_training_validation_and_success(tmp_path) -> None: + callback = _callback(tmp_path) + callback.on_train_start(model=None, iteration=0) + callback.on_training_step_start(model=None, data={}, iteration=0) + callback.on_training_step_end( + model=None, + data_batch={"input_ids": torch.zeros(2, 3)}, + output_batch={}, + loss=torch.tensor(0.5), + iteration=3, + ) + callback.on_validation_start(model=None, dataloader_val=None, iteration=3) + callback.on_validation_step_end( + model=None, + data_batch={"input_ids": torch.zeros(2, 3)}, + output_batch={}, + loss=torch.tensor(0.25), + iteration=3, + ) + callback.on_validation_end(model=None, iteration=3) + callback.on_app_end() + + records = _records(tmp_path) + assert [record["status"] for record in records] == [ + "STARTED", + "RUNNING", + "RUNNING", + "RUNNING", + "RUNNING", + "SUCCESS", + ] + assert records[1]["kpi"]["train/loss_avg"] == 0.5 + assert records[1]["epoch"] == 1 + assert records[1]["max_epoch"] == 2 + assert records[1]["step_in_epoch"] == 3 + assert records[1]["steps_per_epoch"] == 3 + assert records[1]["max_step"] == 6 + assert records[2]["phase"] == "validation_starting" + assert records[3]["max_validation_batches"] == 2 + assert "epoch 1/2" in records[3]["message"] + assert records[4]["kpi"]["val/avg_loss"] == 0.25 + assert "epoch 1/2" in records[4]["message"] + assert records[-1]["kpi"]["val/loss"] == 0.25 + assert records[-1]["epoch"] == 2 + assert records[-1]["completed_epochs"] == 2 + + +def test_tao_status_callback_writes_failure(tmp_path) -> None: + callback = _callback(tmp_path) + callback.on_train_start(model=None, iteration=0) + callback.on_exception(RuntimeError("boom")) + + records = _records(tmp_path) + assert records[-1]["status"] == "FAILURE" + assert records[-1]["error_type"] == "RuntimeError" diff --git a/cosmos_framework/checkpoint/base.py b/cosmos_framework/checkpoint/base.py index f17ebd2d..e22f2360 100644 --- a/cosmos_framework/checkpoint/base.py +++ b/cosmos_framework/checkpoint/base.py @@ -7,11 +7,11 @@ import torch -from cosmos_framework.utils.config import CheckpointConfig, JobConfig -from cosmos_framework.utils.flags import INTERNAL from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.utils import callback +from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.utils.easy_io import easy_io +from cosmos_framework.utils.flags import INTERNAL class AbstractCheckpointer(ABC): @@ -92,6 +92,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: pass diff --git a/cosmos_framework/checkpoint/dcp.py b/cosmos_framework/checkpoint/dcp.py index ba5f996a..a4ab9398 100644 --- a/cosmos_framework/checkpoint/dcp.py +++ b/cosmos_framework/checkpoint/dcp.py @@ -71,9 +71,9 @@ from cosmos_framework.checkpoint.base import AbstractCheckpointer from cosmos_framework.checkpoint.s3_filesystem import S3StorageReader, S3StorageWriter -from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.utils import callback, distributed, log, misc +from cosmos_framework.utils.config import CheckpointConfig, JobConfig from cosmos_framework.utils.easy_io import easy_io from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict @@ -748,7 +748,7 @@ def keys_to_resume_during_load(self) -> tuple[set[str], str | None, bool | None] # If the path doesn't end with specific checkpoint, read the latest # checkpoint file to determine the most recent checkpoint iteration. - if not re.search(r"/checkpoints/iter_\d{9}/?$", checkpoint_path): + if not re.search(r"/checkpoints/(?:iter_\d{9}|epoch_\d+)/?$", checkpoint_path): old_ckpt_path = checkpoint_path latest_ckpt_path = os.path.join(checkpoint_path, "checkpoints/latest_checkpoint.txt") @@ -1090,6 +1090,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: """Save network weights, optimizer parameters, scheduler parameters to a checkpoint. @@ -1106,7 +1107,7 @@ def save( if self.callbacks is not None: self.callbacks.on_save_checkpoint_start(model, iteration) - checkpoint_file = f"iter_{iteration:09}" + checkpoint_file = f"epoch_{epoch}" if epoch is not None else f"iter_{iteration:09}" # Use rank-specific key for RNG state to ensure each rank saves its own state rng_key = f"rng_state_{dist.get_rank()}" @@ -1129,7 +1130,7 @@ def save( self.callbacks.on_save_checkpoint(model, state_dict=to_save_dict) for k in to_save_dict.keys(): - output_dirname = os.path.join(self.save_dirname, f"iter_{iteration:09}/{k}") + output_dirname = os.path.join(self.save_dirname, checkpoint_file, k) to_save_dict[k] = (to_save_dict[k], output_dirname) if self.async_mode == AsyncMode.ASYNC_WITH_PINNED_MEM: diff --git a/cosmos_framework/checkpoint/dcp_distill.py b/cosmos_framework/checkpoint/dcp_distill.py index a1b5279c..a6921e3a 100644 --- a/cosmos_framework/checkpoint/dcp_distill.py +++ b/cosmos_framework/checkpoint/dcp_distill.py @@ -26,9 +26,6 @@ from torch.distributed.checkpoint.stateful import Stateful from torch.nn.modules.module import _IncompatibleKeys -from cosmos_framework.model._base import ImaginaireModel -from cosmos_framework.utils import log, misc -from cosmos_framework.utils.easy_io import easy_io from cosmos_framework.checkpoint.dcp import ( AsyncMode, CustomLoadPlanner, @@ -38,8 +35,11 @@ DistributedCheckpointer as _DistributedCheckpointer, ) from cosmos_framework.checkpoint.dcp import ModelWrapper as VFMModelWrapper -from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict +from cosmos_framework.model._base import ImaginaireModel from cosmos_framework.model.generator.distillation.optimizer import OptimizerContainerLike, is_optimizer_container +from cosmos_framework.utils import log, misc +from cosmos_framework.utils.easy_io import easy_io +from cosmos_framework.utils.generator.rand_state import get_rand_state_dict, set_rand_state_dict __all__: tuple[str, ...] = ( "DistributedCheckpointer", @@ -324,6 +324,7 @@ def save( scheduler: Any = None, grad_scaler: torch.amp.GradScaler | None = None, iteration: int = 0, + epoch: int | None = None, ) -> None: if self.async_mode == AsyncMode.ASYNC_WITH_PINNED_MEM: self._wait_for_previous_async_checkpoint() @@ -332,7 +333,7 @@ def save( self.callbacks.on_save_checkpoint_start(model, iteration) model_dict = model.model_dict() - checkpoint_file = f"iter_{iteration:09}" + checkpoint_file = f"epoch_{epoch}" if epoch is not None else f"iter_{iteration:09}" rng_key = f"rng_state_{dist.get_rank()}" to_save_dict: dict[str, Any] = { @@ -360,7 +361,7 @@ def save( to_save_dict["dataloader"] = dataloader_wrapper.state_dict() for key in list(to_save_dict.keys()): - output_dirname = os.path.join(self.save_dirname, f"iter_{iteration:09}/{key}") + output_dirname = os.path.join(self.save_dirname, checkpoint_file, key) to_save_dict[key] = (to_save_dict[key], output_dirname) if self.callbacks is not None: diff --git a/cosmos_framework/checkpoint/dummy.py b/cosmos_framework/checkpoint/dummy.py index 3cfa1bab..443241d9 100644 --- a/cosmos_framework/checkpoint/dummy.py +++ b/cosmos_framework/checkpoint/dummy.py @@ -22,6 +22,7 @@ def save( scheduler: torch.optim.lr_scheduler.LRScheduler, grad_scaler: torch.amp.GradScaler, iteration: int, + epoch: int | None = None, ) -> None: pass diff --git a/cosmos_framework/configs/base/reasoner/defaults/callbacks.py b/cosmos_framework/configs/base/reasoner/defaults/callbacks.py index 7d1911f5..37c7a26a 100644 --- a/cosmos_framework/configs/base/reasoner/defaults/callbacks.py +++ b/cosmos_framework/configs/base/reasoner/defaults/callbacks.py @@ -7,22 +7,22 @@ from hydra.core.config_store import ConfigStore -from cosmos_framework.callbacks.manual_gc import ManualGarbageCollection -from cosmos_framework.utils.lazy_config import PLACEHOLDER -from cosmos_framework.utils.lazy_config import LazyCall as L -from cosmos_framework.utils.callback import LowPrecisionCallback, WandBCallback from cosmos_framework.callbacks.dataloader_state import DataLoaderStateCallback - from cosmos_framework.callbacks.grad_clip import GradClip from cosmos_framework.callbacks.hf_export import HFExportCallback from cosmos_framework.callbacks.iter_speed import IterSpeed from cosmos_framework.callbacks.learning_rate_logger import LearningRateLogger from cosmos_framework.callbacks.log_tensor_shape import LogTensorShapeCallback +from cosmos_framework.callbacks.manual_gc import ManualGarbageCollection from cosmos_framework.callbacks.param_count import ParamCount +from cosmos_framework.callbacks.tao_status import TAOStatusCallback from cosmos_framework.callbacks.tokens_per_sec import VLMTokensPerSec from cosmos_framework.callbacks.wandb_log import WandbCallback as WandBCallbackMultiplier from cosmos_framework.callbacks.wandb_vis import VisualizationLoggingCallback from cosmos_framework.configs.base.defaults.callbacks import JOB_MONITOR_CALLBACKS +from cosmos_framework.utils.callback import LowPrecisionCallback, WandBCallback +from cosmos_framework.utils.lazy_config import PLACEHOLDER +from cosmos_framework.utils.lazy_config import LazyCall as L # from cosmos_framework.utils.callback import NVTXCallback @@ -52,6 +52,13 @@ def register_callbacks(): config=PLACEHOLDER, trainer=PLACEHOLDER, ), # reads model.precision; no extra kwarg needed + tao=L(TAOStatusCallback)( + enabled=False, + status_file_path=None, + experiment_name="", + logging_interval=1, + validation_heartbeat_interval=1, + ), # nvtx=L(NVTXCallback)(synchronize=True), ) diff --git a/cosmos_framework/configs/base/reasoner/experiment/wts_vlm.py b/cosmos_framework/configs/base/reasoner/experiment/wts_vlm.py new file mode 100644 index 00000000..6238a3db --- /dev/null +++ b/cosmos_framework/configs/base/reasoner/experiment/wts_vlm.py @@ -0,0 +1,294 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +"""Woven Traffic Safety (WTS) video-QA SFT on the Cosmos3-Nano VLM path.""" + +from __future__ import annotations + +import json +import os +import re +from collections import OrderedDict +from copy import deepcopy +from typing import Any + +import torch +from hydra.core.config_store import ConfigStore +from PIL import Image +from torch.utils.data import Dataset + +from cosmos_framework.callbacks.cosmos_dataloader_state import CosmosDataLoaderStateCallback +from cosmos_framework.configs.base.reasoner.experiment.dataflow_roles import VLMCollator, VLMProcessor +from cosmos_framework.data.generator.dataflow import CosmosDataLoader, MapDistributor, PoolPackingBatcher +from cosmos_framework.data.generator.processors import build_processor +from cosmos_framework.utils.generator.torchcodec_video import TorchCodecVideoReader +from cosmos_framework.utils.lazy_config import LazyCall as L +from cosmos_framework.utils.lazy_config import LazyDict +from cosmos_framework.utils.reasoner.constant import IGNORE_INDEX + + +class WTSLlavaDataset(Dataset): + """Map-style loader for WTS LLaVA JSON annotations.""" + + def __init__( + self, + annotation_path: str, + media_path: str, + limit: int | str | None = None, + ) -> None: + self.annotation_path = os.path.abspath(os.path.expanduser(annotation_path)) + self.media_path = os.path.abspath(os.path.expanduser(media_path)) + if limit in ("", None): + parsed_limit = None + else: + parsed_limit = int(limit) + if parsed_limit < 1: + parsed_limit = None + + with open(self.annotation_path, encoding="utf-8") as annotation_file: + records = json.load(annotation_file) + if not isinstance(records, list): + raise TypeError(f"WTS annotations must be a JSON array, got {type(records).__name__}") + self.records = records[:parsed_limit] if parsed_limit is not None else records + if not self.records: + raise ValueError(f"WTS annotation file contains no usable records: {self.annotation_path}") + + for index, record in enumerate(self.records): + if not isinstance(record, dict) or not isinstance(record.get("video"), str): + raise ValueError(f"WTS record {index} must contain a string 'video' field") + conversations = record.get("conversations") + if not isinstance(conversations, list) or len(conversations) < 2: + raise ValueError(f"WTS record {index} must contain at least two conversation turns") + + def __len__(self) -> int: + return len(self.records) + + def __getitem__(self, index: int) -> dict[str, Any]: + record = dict(self.records[index]) + video_path = record["video"] + if not os.path.isabs(video_path): + video_path = os.path.join(self.media_path, video_path) + if not os.path.isfile(video_path): + raise FileNotFoundError(f"WTS video does not exist: {video_path}") + record["video"] = video_path + return record + + +class WTSProcessor(VLMProcessor): + """Convert WTS ShareGPT records and uniformly sample each video to PIL frames.""" + + def __init__( + self, + processor: Any, + ignore_index: int = IGNORE_INDEX, + num_video_frames: int = 8, + video_cache_size: int = 8, + system_prompt: str = "", + ) -> None: + super().__init__(processor=processor, ignore_index=ignore_index) + if num_video_frames < 1: + raise ValueError("num_video_frames must be >= 1") + if video_cache_size < 0: + raise ValueError("video_cache_size must be >= 0") + self.num_video_frames = num_video_frames + self.video_cache_size = video_cache_size + self.system_prompt = system_prompt + self._video_cache: OrderedDict[str, tuple[list[Image.Image], float]] = OrderedDict() + + def _decode_video(self, video_path: str) -> tuple[list[Image.Image], float]: + cached = self._video_cache.get(video_path) + if cached is not None: + self._video_cache.move_to_end(video_path) + return cached + + reader = TorchCodecVideoReader(video_path, num_threads=2) + total_frames = len(reader) + if total_frames < 1: + raise ValueError(f"WTS video has zero frames: {video_path}") + sample_count = min(self.num_video_frames, total_frames) + if sample_count == 1: + indices = [0] + else: + indices = torch.linspace(0, total_frames - 1, steps=sample_count).round().to(dtype=torch.long).tolist() + frames_np = reader.get_frames_nhwc_uint8(indices) + frames = [Image.fromarray(frame) for frame in frames_np] + + source_fps = reader.get_avg_fps() + average_stride = (indices[-1] - indices[0]) / max(len(indices) - 1, 1) if len(indices) > 1 else 1.0 + effective_fps = source_fps / max(average_stride, 1.0) + decoded = (frames, float(effective_fps)) + if self.video_cache_size > 0: + self._video_cache[video_path] = decoded + self._video_cache.move_to_end(video_path) + while len(self._video_cache) > self.video_cache_size: + self._video_cache.popitem(last=False) + return decoded + + def _sharegpt_to_openai(self, item: dict) -> list[dict]: + conversations = item.get("conversations", []) + video_path = item.get("video") + frames, fps = self._decode_video(video_path) + messages: list[dict] = [] + video_inserted = False + if self.system_prompt: + messages.append({"role": "system", "content": self.system_prompt}) + + for turn in conversations: + role = "user" if turn["from"] == "human" else "assistant" + text = re.sub(r"(\n)?(\n)?", "", turn["value"]).strip() + if role == "user" and not video_inserted: + content: Any = [ + {"type": "video", "video": frames, "fps": fps}, + {"type": "text", "text": text}, + ] + video_inserted = True + else: + content = text + messages.append({"role": role, "content": content}) + return messages + + +def _wts_dataloader( + *, + annotation_env: str, + media_env: str, + limit_env: str, + shuffle: bool, +) -> LazyDict: + return L(CosmosDataLoader)( + distributor=L(MapDistributor)( + dataset=L(WTSLlavaDataset)( + annotation_path=f"${{oc.env:{annotation_env}}}", + media_path=f"${{oc.env:{media_env}}}", + limit=f"${{oc.env:{limit_env},''}}", + ), + shuffle=shuffle, + seed=42, + name="train" if shuffle else "val", + ), + processor=L(WTSProcessor)( + processor=L(build_processor)( + tokenizer_type="${model.config.policy.backbone.model_name}", + config_variant="hf", + ), + ignore_index=IGNORE_INDEX, + num_video_frames=8, + video_cache_size=8, + system_prompt=("You are a helpful assistant that can answer questions about a street-view CCTV footage."), + ), + batcher=L(PoolPackingBatcher)( + max_tokens="${data_setting.max_tokens}", + pool_size=16 if shuffle else 1, + max_batch_size=1, + long_threshold=6400, + ), + collator=L(VLMCollator)(), + num_workers=0, + prefetch_factor=None, + persistent_workers=False, + pin_memory=False, + ) + + +wts_vlm = LazyDict( + dict( + defaults=[ + {"override /checkpoint": "local"}, + {"override /data_train": None}, + {"override /data_val": None}, + {"override /model": "vlm_fsdp"}, + {"override /vlm_policy": "qwen3_vl_8b_instruct"}, + {"override /callbacks": ["basic_vlm", "basic_log"]}, + "_self_", + ], + job=dict( + project="cosmos3_reasoner", + group="wts_sft", + wandb_mode="disabled", + ), + trainer=dict( + callbacks=dict( + dataloader_state=L(CosmosDataLoaderStateCallback)(), + tao=dict( + enabled=True, + logging_interval=1, + validation_heartbeat_interval=1, + ), + ), + max_iter=10, + logging_iter=1, + run_validation=True, + validation_iter=10, + max_val_iter=10, + run_validation_on_start=False, + grad_accum_iter=1, + ), + optimizer=dict( + lr=1.0e-4, + fused=True, + weight_decay=0.01, + betas=[0.9, 0.999], + lr_multipliers={"model.visual": 1.0}, + ), + model=dict( + config=dict( + policy=dict( + model_max_length=81920, + qwen_max_video_token_length=8192, + ), + freeze=dict(trainable_params=[".*"]), + parallelism=dict( + data_parallel_shard_degree=4, + data_parallel_replicate_degree=1, + ), + ), + ), + data_setting=dict( + max_tokens=81920, + qwen_max_video_token_length=8192, + ), + checkpoint=dict( + save_iter=100, + load_from_object_store=dict(enabled=False, credentials="", bucket=""), + save_to_object_store=dict(enabled=False, credentials="", bucket=""), + ), + dataloader_train=_wts_dataloader( + annotation_env="WTS_TRAIN_ANNOTATION", + media_env="WTS_TRAIN_MEDIA", + limit_env="WTS_TRAIN_LIMIT", + shuffle=True, + ), + dataloader_val=_wts_dataloader( + annotation_env="WTS_VAL_ANNOTATION", + media_env="WTS_VAL_MEDIA", + limit_env="WTS_VAL_LIMIT", + shuffle=False, + ), + upload_reproducible_setup=False, + ), + flags={"allow_objects": True}, +) + +ConfigStore.instance().store( + group="experiment", + package="_global_", + name="wts_vlm", + node=wts_vlm, +) + + +# Edge uses the same WTS data contract but a different native HF backbone. +# Keep it in this module so the dataloader worker callables retain a single +# pickle identity when the reasoner config loader reloads experiment modules. +wts_vlm_edge = deepcopy(wts_vlm) +wts_vlm_edge["defaults"][4] = {"override /vlm_policy": "cosmos3_edge_reasoner"} +wts_vlm_edge["job"]["group"] = "wts_edge_sft" +wts_vlm_edge["optimizer"].pop("lr_multipliers", None) +wts_vlm_edge["model"]["config"]["policy"]["model_max_length"] = 16000 + +ConfigStore.instance().store( + group="experiment", + package="_global_", + name="wts_vlm_edge", + node=wts_vlm_edge, +) diff --git a/cosmos_framework/configs/base/reasoner/experiment/wts_vlm_test.py b/cosmos_framework/configs/base/reasoner/experiment/wts_vlm_test.py new file mode 100644 index 00000000..3257a3b0 --- /dev/null +++ b/cosmos_framework/configs/base/reasoner/experiment/wts_vlm_test.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: OpenMDW-1.1 + +from __future__ import annotations + +import json + +import pytest + +from cosmos_framework.configs.base.reasoner.experiment.wts_vlm import WTSLlavaDataset, wts_vlm_edge + + +def test_wts_dataset_resolves_video_paths_and_limit(tmp_path) -> None: + media = tmp_path / "videos" + media.mkdir() + (media / "clip.mp4").write_bytes(b"video") + annotations = tmp_path / "annotations.json" + annotations.write_text( + json.dumps( + [ + { + "video": "clip.mp4", + "conversations": [ + {"from": "human", "value": "