code/verl/utils/logger/gen_logger.py

# Copyright 2024 Bytedance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, List, Optional, Tuple, Union

from ..py_functional import is_package_available


if is_package_available("wandb"):
    import wandb  # type: ignore


if is_package_available("swanlab"):
    import swanlab  # type: ignore


GenerationSample = Union[Tuple[str, str, str, float], Tuple[str, str, str, float, Any]]


def _unpack_generation_sample(sample: GenerationSample) -> Tuple[str, str, str, float, Any]:
    """Unify generation sample shape with backward compatibility."""
    if len(sample) == 4:
        inp, out, lab, score = sample
        return inp, out, lab, score, None
    if len(sample) == 5:
        inp, out, lab, score, problem_id = sample
        return inp, out, lab, score, problem_id
    raise ValueError(f"Invalid generation sample format with length={len(sample)}.")


@dataclass
class GenerationLogger(ABC):
    config: dict[str, Any]

    @abstractmethod
    def log(self, samples: List[GenerationSample], step: int) -> None: ...


@dataclass
class ConsoleGenerationLogger(GenerationLogger):
    def log(self, samples: List[GenerationSample], step: int) -> None:
        for sample in samples:
            inp, out, lab, score, problem_id = _unpack_generation_sample(sample)
            print(
                f"[problem_id] {problem_id}\n[prompt] {inp}\n[output] {out}\n[ground_truth] {lab}\n[score] {score}\n"
            )


@dataclass
class FileGenerationLogger(GenerationLogger):
    def log(self, samples: List[GenerationSample], step: int) -> None:
        with open(os.path.join(self.config["trainer"]["save_checkpoint_path"], "generations.log"), "a") as f:
            for sample in samples:
                inp, out, lab, score, problem_id = _unpack_generation_sample(sample)
                f.write(
                    f"[problem_id] {problem_id}\n[prompt] {inp}\n[output] {out}\n[ground_truth] {lab}\n[score] {score}\n\n"
                )


@dataclass
class WandbGenerationLogger(GenerationLogger):
    def log(self, samples: List[GenerationSample], step: int) -> None:
        # Create column names for all samples
        columns = ["step"] + sum(
            [
                [f"problem_id_{i + 1}", f"input_{i + 1}", f"output_{i + 1}", f"label_{i + 1}", f"score_{i + 1}"]
                for i in range(len(samples))
            ],
            [],
        )

        if not hasattr(self, "validation_table"):
            # Initialize the table on first call
            self.validation_table = wandb.Table(columns=columns)

        # Create a new table with same columns and existing data
        # Workaround for https://github.com/wandb/wandb/issues/2981#issuecomment-1997445737
        new_table = wandb.Table(columns=columns, data=self.validation_table.data)

        # Add new row with all data
        row_data = [step]
        for sample in samples:
            inp, out, lab, score, problem_id = _unpack_generation_sample(sample)
            row_data.extend([problem_id, inp, out, lab, score])

        new_table.add_data(*row_data)
        wandb.log({"val/generations": new_table}, step=step)
        self.validation_table = new_table


@dataclass
class SwanlabGenerationLogger(GenerationLogger):
    def log(self, samples: List[GenerationSample], step: int) -> None:
        swanlab_text_list = []
        for i, sample in enumerate(samples):
            inp, out, lab, score, problem_id = _unpack_generation_sample(sample)
            row_text = "\n\n---\n\n".join(
                (
                    f"problem_id: {problem_id}",
                    f"input: {inp}",
                    f"output: {out}",
                    f"label: {lab}",
                    f"score: {score}",
                )
            )
            swanlab_text_list.append(swanlab.Text(row_text, caption=f"sample {i + 1}"))

        swanlab.log({"val/generations": swanlab_text_list}, step=step)


GEN_LOGGERS = {
    "console": ConsoleGenerationLogger,
    "file": FileGenerationLogger,
    "wandb": WandbGenerationLogger,
    "swanlab": SwanlabGenerationLogger,
}


class AggregateGenerationsLogger:
    def __init__(self, loggers: List[str], config: Optional[dict[str, Any]] = None):
        self.loggers: List[GenerationLogger] = []

        for logger in loggers:
            if logger in GEN_LOGGERS:
                self.loggers.append(GEN_LOGGERS[logger](config))

    def log(self, samples: List[GenerationSample], step: int) -> None:
        for logger in self.loggers:
            logger.log(samples, step)