nemo_automodel.components.loggers.metric_logger

Module Contents

Classes

Name	Description
`MetricLogger`	Simple JSON Lines logger.
`MetricLoggerDist`	Rank-zero JSON Lines metric logger for distributed jobs.
`MetricsSample`	Single timestamped metrics record.

Functions

Name	Description
`build_metric_logger`	Build a local or distributed metric logger depending on distributed state.
`stack_and_move_tensor_metrics_to_cpu`	Convert tensor metrics in buffered samples to CPU-backed scalar or list values.

Data

DEFAULT_BUFFER_SIZE

API

class nemo_automodel.components.loggers.metric_logger.MetricLogger(
    filepath: str,
    flush: bool = False,
    append: bool = True,
    buffer_size: int = DEFAULT_BUFFER_SIZE
)

Simple JSON Lines logger.

Appends one JSON object per line.
Thread-safe writes via an internal lock.
Creates parent directories as needed.
UTF-8 without BOM, newline per record.

_fp

= open(self.filepath, mode, encoding='utf-8')

_lock

= threading.Lock()

buffer

List[MetricsSample] = []

filepath

= os.path.abspath(filepath)

nemo_automodel.components.loggers.metric_logger.MetricLogger.__enter__() -> nemo_automodel.components.loggers.metric_logger.MetricLogger

nemo_automodel.components.loggers.metric_logger.MetricLogger.__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: typing.Any
) -> None

nemo_automodel.components.loggers.metric_logger.MetricLogger._drain() -> None

Write buffered records out. Caller must hold self._lock.

nemo_automodel.components.loggers.metric_logger.MetricLogger._move_to_cpu(
    buffer: typing.List[nemo_automodel.components.loggers.metric_logger.MetricsSample]
) -> typing.List[str]

nemo_automodel.components.loggers.metric_logger.MetricLogger._save(
    lines: typing.List[str]
) -> None

nemo_automodel.components.loggers.metric_logger.MetricLogger.close() -> None

nemo_automodel.components.loggers.metric_logger.MetricLogger.flush() -> None

Write buffered records out to the file and fsync, without closing it.

Lets a caller align durability with an external event — a checkpoint boundary, say — instead of relying on the record count happening to reach buffer_size there.

The fsync is unconditional, NOT gated on the flush= constructor flag. That flag controls whether every buffer-size-triggered write pays for an fsync; this method is an explicit, caller-chosen durability point, so it has to be durable under the default flush=False too. Draining into the file object alone would leave the records in the process’s stdio buffer — still lost to a crash, and invisible to anything reading the file — which would make the guarantee empty exactly where it is being relied on.

nemo_automodel.components.loggers.metric_logger.MetricLogger.log(
    record: nemo_automodel.components.loggers.metric_logger.MetricsSample
) -> None

class nemo_automodel.components.loggers.metric_logger.MetricLoggerDist(
    filepath: str,
    flush: bool = False,
    append: bool = True,
    buffer_size: int = DEFAULT_BUFFER_SIZE
)

Bases: MetricLogger

Rank-zero JSON Lines metric logger for distributed jobs.

rank

= dist.get_rank()

world_size

= dist.get_world_size()

nemo_automodel.components.loggers.metric_logger.MetricLoggerDist.__enter__() -> nemo_automodel.components.loggers.metric_logger.MetricLoggerDist

nemo_automodel.components.loggers.metric_logger.MetricLoggerDist.__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: typing.Any
) -> None

nemo_automodel.components.loggers.metric_logger.MetricLoggerDist.close() -> None

nemo_automodel.components.loggers.metric_logger.MetricLoggerDist.flush() -> None

nemo_automodel.components.loggers.metric_logger.MetricLoggerDist.log(
    record: nemo_automodel.components.loggers.metric_logger.MetricsSample
) -> None

class nemo_automodel.components.loggers.metric_logger.MetricsSample(
    step: int,
    epoch: int,
    metrics: typing.Dict[str, typing.Any] = dict(),
    timestamp: str | None = None
)

Dataclass

Single timestamped metrics record.

epoch

int

metrics

Dict[str, Any] = field(default_factory=dict)

step

int

timestamp

str | None = None

nemo_automodel.components.loggers.metric_logger.MetricsSample.__post_init__() -> None

nemo_automodel.components.loggers.metric_logger.MetricsSample.to_dict() -> typing.Dict[str, typing.Any]

nemo_automodel.components.loggers.metric_logger.build_metric_logger(
    filepath: str,
    flush: bool = False,
    append: bool = True,
    buffer_size: int = DEFAULT_BUFFER_SIZE
) -> nemo_automodel.components.loggers.metric_logger.MetricLogger

Build a local or distributed metric logger depending on distributed state.

nemo_automodel.components.loggers.metric_logger.stack_and_move_tensor_metrics_to_cpu(
    metric_vector: typing.List[nemo_automodel.components.loggers.metric_logger.MetricsSample]
) -> typing.List[nemo_automodel.components.loggers.metric_logger.MetricsSample]

Convert tensor metrics in buffered samples to CPU-backed scalar or list values.

nemo_automodel.components.loggers.metric_logger.DEFAULT_BUFFER_SIZE = 100