microsoft--rd-agent/rdagent/components/coder/finetune/conf.py

import json
import os
import re
import shutil
from pathlib import Path
from typing import Any, Literal

from rdagent.app.finetune.llm.conf import FT_RD_SETTING
from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
from rdagent.core.experiment import FBWorkspace
from rdagent.log import rdagent_logger as logger
from rdagent.scenarios.finetune.scen.utils import _compute_column_stats
from rdagent.utils.agent.tpl import T
from rdagent.utils.env import (
    BenchmarkCondaConf,
    BenchmarkCondaEnv,
    BenchmarkDockerConf,
    BenchmarkDockerEnv,
    DockerEnv,
    Env,
    FTCondaConf,
    FTCondaEnv,
    FTDockerEnv,
)


def is_docker_env(env: Env) -> bool:
    """Check if the environment is Docker-based."""
    return isinstance(env, DockerEnv)


def get_workspace_prefix(env: Env) -> str:
    """Return workspace path prefix based on env type.

    Docker uses /workspace as mount point, conda uses current directory.
    """
    return "/workspace" if is_docker_env(env) else "."


FT_YAML_FILE_NAME = "train.yaml"
FT_DATA_PROC_FILE_NAME = "data_process.py"
FT_DEBUG_YAML_FILE_NAME = "debug_train.yaml"
FT_TEST_PARAMS_FILE_NAME = "test_params.yaml"
FT_DATA_FILE_NAME = "data.json"
FT_DATA_SCRIPT_NAME = "process_data.py"

# ENV Info:  the path of the model and dataset in the container/environment
FT_MODEL_PATH = "/assets/models"
FT_DATASET_PATH = "/assets/datasets"


def get_data_processing_cache_key(local_path: str | Path) -> list[list[str]]:
    """Generate cache key based only on data processing script and dataset info.

    This ensures that data processing results are reused as long as the script
    and dataset configuration remain unchanged, even if other files in the
    workspace (like training config) have been modified.
    """
    content = []
    local_path = Path(local_path)
    # We only care about the script that generates data and the dataset configuration
    for filename in [FT_DATA_SCRIPT_NAME, "dataset_info.json"]:
        file_path = local_path / filename
        if file_path.exists():
            content.append([filename, file_path.read_text()])
    content.sort(key=lambda x: x[0])
    return content


class FTPathConfig:
    """Centralized path configuration for FT scenario.

    Provides environment-aware paths for Docker vs Conda modes.
    Uses lazy evaluation (properties) to avoid import-time errors.

    Usage:
        from rdagent.components.coder.finetune.conf import FT_PATHS

        models_path = FT_PATHS.models      # e.g., "/assets/models/" or "/path/to/finetune/models/"
        datasets_path = FT_PATHS.datasets  # e.g., "/assets/datasets/" or "/path/to/finetune/datasets/"
        workspace_path = FT_PATHS.workspace  # e.g., "/workspace/" or "./"
    """

    @property
    def is_docker(self) -> bool:
        """Check if current environment is Docker-based."""
        # FIXME: the env should work in same way for docker and conda env.
        # We should not expose the env type everywhere.
        return FTCoderCoSTEERSettings().env_type == "docker"

    @property
    def models(self) -> str:
        """Model directory path (with trailing slash)."""
        if self.is_docker:
            return FT_MODEL_PATH + "/"
        return str(FT_RD_SETTING.file_path / "models") + "/"

    @property
    def datasets(self) -> str:
        """Dataset directory path for raw datasets (with trailing slash)."""
        if self.is_docker:
            return FT_DATASET_PATH + "/"
        return str(FT_RD_SETTING.file_path / "datasets") + "/"

    @property
    def workspace(self) -> str:
        """Workspace path prefix for prompts (with trailing slash)."""
        return "/workspace/" if self.is_docker else "./"

    @property
    def deepspeed(self) -> str:
        """DeepSpeed config directory."""
        if self.is_docker:
            return "/app/examples/deepspeed/"
        # Conda mode: use bundled deepspeed configs in project
        # Path: conf.py -> finetune -> coder -> components -> rdagent -> scenarios/finetune/env/conda/deepspeed
        rdagent_root = Path(__file__).parent.parent.parent.parent
        deepspeed_path = rdagent_root / "scenarios" / "finetune" / "env" / "conda" / "deepspeed"
        return str(deepspeed_path) + "/" if deepspeed_path.exists() else ""


# Singleton instance for path configuration
FT_PATHS = FTPathConfig()


class FTCoderCoSTEERSettings(CoSTEERSettings):
    """LLM Fine-tuning CoSTEER settings"""

    class Config:
        env_prefix = "FT_Coder_CoSTEER_"

    max_seconds_multiplier: int = 8
    """LLM training takes longer, use higher multiplier"""

    env_type: str = "docker"
    """Environment type for LLM fine-tuning (docker/conda)"""

    extra_eval: list[str] = []
    """Extra evaluators"""


def _get_standard_ft_volumes() -> dict:
    """Get standard mount volume configuration for LLM finetune environments.

    Creates standard directory mappings:
    - models -> /assets/models (ro)
    - datasets -> /assets/datasets (ro)

    Returns:
        Dictionary of local_path -> docker_mount_config mappings
    """
    base_path = Path(FT_RD_SETTING.file_path)
    volumes = {}

    # Read-only mounts for data and models
    readonly_mounts = [
        ("models", FT_MODEL_PATH),
        ("datasets", FT_DATASET_PATH),
    ]

    for local_dir, docker_path in readonly_mounts:
        local_path = base_path / local_dir
        volumes[str(local_path)] = {"bind": docker_path, "mode": "ro"}

    return volumes


def get_ft_env(
    extra_volumes: dict = {},
    operation: str = "full_training",
    enable_cache: bool | None = None,
) -> Env:
    """LLM finetune dedicated environment construction function.

    Automatically includes standard finetune volume mounts:
    - models -> /assets/models (ro)
    - datasets -> /assets/datasets (ro)
    - output -> /workspace/output (rw, auto-created)

    Note: .llama_factory_info is no longer automatically mounted.
    Pass llama_factory_info volume via extra_volumes when needed.

    Args:
        extra_volumes: Additional volume mounts beyond standard ones
        operation: Operation type for timeout selection.
            - "data_processing": Data processing (data_processing_timeout)
            - "micro_batch": Micro-batch test (micro_batch_timeout)
            - "full_training": Full training (full_timeout)
        enable_cache: Whether to enable caching (None means use config value)

    Returns:
        Configured environment ready for use
    """

    conf = FTCoderCoSTEERSettings()

    # Select timeout based on operation type
    timeout_map = {
        "data_processing": FT_RD_SETTING.data_processing_timeout,
        "debug_data_processing": FT_RD_SETTING.debug_data_processing_timeout,
        "micro_batch": FT_RD_SETTING.micro_batch_timeout,
        "full_training": FT_RD_SETTING.full_timeout,
    }
    running_timeout_period = timeout_map.get(operation, FT_RD_SETTING.full_timeout)

    # Use config value if enable_cache is not explicitly provided
    if enable_cache is None:
        enable_cache = FT_RD_SETTING.docker_enable_cache

    # Use dedicated LLM docker or conda env based on config
    if conf.env_type == "docker":
        env = FTDockerEnv()
        # Docker mode: setup volume mounts for models/datasets
        standard_volumes = _get_standard_ft_volumes()
        combined_volumes = standard_volumes.copy()
        combined_volumes.update(extra_volumes)
        env.conf.extra_volumes = combined_volumes
    elif conf.env_type == "conda":
        env = FTCondaEnv(conf=FTCondaConf())  # Auto-installs dependencies if env doesn't exist
        # Conda mode: no volume mounts needed, use local paths directly
        # extra_volumes are ignored in conda mode
    else:
        raise ValueError(f"Unknown env type: {conf.env_type}")

    env.conf.running_timeout_period = running_timeout_period
    env.conf.enable_cache = enable_cache
    env.prepare()
    return env


def get_data_processing_env(
    enable_cache: bool | None = None,
    is_debug: bool = False,
) -> tuple[Env, dict]:
    """Get environment for data processing scripts with LLM API access.

    This environment is configured for running data processing scripts that may
    need to call LLM APIs. It includes:
    - Standard finetune volume mounts (datasets, models)
    - LLM API environment variables (OPENAI_API_KEY, OPENAI_BASE_URL, etc.)

    Args:
        enable_cache: Whether to enable Docker caching
        is_debug: Whether running in debug mode (shorter timeout, default 20 min vs 1 hour)

    Returns:
        Tuple of (env, env_vars) where env_vars contains LLM API keys
        to be passed to env.run() as the env parameter
    """
    env = get_ft_env(
        operation="debug_data_processing" if is_debug else "data_processing",
        enable_cache=enable_cache,
    )

    # Collect LLM API environment variables to pass to env.run()
    llm_env_vars = {"PYTHONPATH": "./"}  # Base env var

    # Pass OPENAI_API_KEY directly
    if api_key := os.getenv("OPENAI_API_KEY"):
        llm_env_vars["OPENAI_API_KEY"] = api_key

    # Read OPENAI_API_BASE from env, but pass as OPENAI_BASE_URL (OpenAI SDK expects this name)
    if api_base := os.getenv("OPENAI_API_BASE"):
        llm_env_vars["OPENAI_BASE_URL"] = api_base

    # Pass model pools as JSON environment variables for load balancing
    llm_env_vars["STRONG_MODEL_POOL"] = json.dumps(FT_RD_SETTING.strong_models)
    llm_env_vars["WEAK_MODEL_POOL"] = json.dumps(FT_RD_SETTING.weak_models)

    return env, llm_env_vars


def clear_workspace(workspace: FBWorkspace, env: Env) -> None:
    """
    Clean the files in LLM finetune workspace.
    Only keeps the files that are injected by the coder (in workspace.file_dict) and `logs`.

    Args:
        workspace: The workspace object containing the file dictionary.
        env: The environment to execute the clean command in.
    """
    target_path = workspace.workspace_path
    if not target_path.exists():
        return

    # The cache_path is created when mounting, so the permissions changes does not work.
    keep_items = {"logs", T("scenarios.data_science.share:scen.cache_path").r()}

    for file_path in workspace.file_dict.keys():
        top_level = Path(file_path).parts[0]
        keep_items.add(top_level)

    remove_items = []
    for item in target_path.iterdir():
        if item.name in keep_items:
            continue
        remove_items.append(item.name)

    if remove_items:
        ws_prefix = get_workspace_prefix(env)
        # Construct rm command with all items to remove
        # Items are relative to workspace root inside the env
        items_str = " ".join([f"'{ws_prefix}/{item}'" for item in remove_items])
        cmd = f"rm -rf {items_str}"
        workspace.execute(env=env, entry=cmd)


def get_benchmark_env(
    extra_volumes: dict = {},
    timeout: int | None = None,
) -> Env:
    """OpenCompass benchmark environment construction function.

    Supports both Docker and conda environments based on FT_Coder_CoSTEER_env_type.

    Args:
        extra_volumes: Additional volume mounts (only used in Docker mode)
        timeout: Running timeout in seconds (None uses config default)

    Returns:
        Configured environment ready for benchmark evaluation
    """
    conf = FTCoderCoSTEERSettings()

    # Use benchmark-specific timeout or config default
    if timeout is None:
        # 0 means no timeout, use 7 days as practical "infinite"
        timeout = FT_RD_SETTING.benchmark_timeout if FT_RD_SETTING.benchmark_timeout > 0 else 86400 * 7

    benchmark_volumes = {}
    # Setup finetune share folder mount for models
    (FT_RD_SETTING.file_path / "benchmarks").mkdir(parents=True, exist_ok=True)
    # NOTE: we choose a folder in the workspace as the mount point due to we may run multiple instances in same
    # host machine. If conda env is used, the mount point will conflict with each other.
    benchmark_volumes[str((FT_RD_SETTING.file_path / "benchmarks").resolve())] = {
        "bind": "./benchmarks",
        "mode": "rw",
    }
    env_dict = {"COMPASS_DATA_CACHE": "./benchmarks/opencompass_data"}
    # Mount models directory for LoRA base model access (vLLM needs base model config)
    models_path = FT_RD_SETTING.file_path / "models"
    if models_path.exists():
        benchmark_volumes[str(models_path.resolve())] = {"bind": FT_MODEL_PATH, "mode": "ro"}
    benchmark_volumes.update(extra_volumes)

    if conf.env_type == "docker":
        docker_conf = BenchmarkDockerConf()
        docker_conf.running_timeout_period = timeout
        docker_conf.extra_volumes = benchmark_volumes
        docker_conf.env_dict = env_dict
        env = BenchmarkDockerEnv(conf=docker_conf)
    elif conf.env_type == "conda":
        # NOTE:
        # We assume user has the permissions to create the softlink in the target directory.
        # If we have requirements in the future, we suggest make the target directory configurable in BenchmarkCondaConf.
        conda_conf = BenchmarkCondaConf()
        conda_conf.running_timeout_period = timeout
        conda_conf.extra_volumes = benchmark_volumes
        conda_conf.env_dict = env_dict
        env = BenchmarkCondaEnv(conf=conda_conf)  # Auto-installs dependencies if env doesn't exist
    else:
        raise ValueError(f"Unknown env type: {conf.env_type}")

    env.prepare()
    return env


def inject_data_stats(implementation: FBWorkspace, data: list, stdout: str) -> None:
    """Compute token statistics and inject data_stats.json.

    Used by both FTDataEvaluator (coding stage) and FTRunnerEvaluator (running stage).

    Args:
        implementation: The workspace to inject data_stats.json into
        data: The data list from data.json
        stdout: The stdout from process_data.py execution
    """
    token_stats = _compute_column_stats(data)

    data_stats = {
        "total_samples": len(data),
        "token_stats": token_stats,
        "stdout_summary": stdout,
    }

    implementation.inject_files(**{"data_stats.json": json.dumps(data_stats, indent=2)})
    logger.info(f"Injected data_stats.json with {len(data)} samples")