Fine-tuning/backend/app/engines/multimodal_engine.py

import os
import json
from pathlib import Path
from typing import Any

# 远程训练节点没有 pydantic-settings/数据库，直接用环境变量
from types import SimpleNamespace

def _resolve_data_dir() -> Path:
    v = os.environ.get("DATA_DIR") or os.environ.get("COMPUTE_NODE_REMOTE_DATA_DIR")
    if v:
        return Path(v)
    env_file = Path(__file__).resolve().parent.parent.parent / ".env"
    if env_file.exists():
        for line in env_file.read_text():
            if line.strip().startswith("DATA_DIR="):
                return Path(line.split("=", 1)[1].strip())
    return Path("/root/Fine-tuning/backend/data")

_data_dir = _resolve_data_dir()
settings = SimpleNamespace(
    data_dir=_data_dir,
    processed_dir=_data_dir / "processed",
    adapters_dir=_data_dir / "adapters",
    models_dir=_data_dir / "models",
)

import logging
logger = logging.getLogger(__name__)

from app.engines.base import BaseEngine


class MultimodalEngine(BaseEngine):
    """多模态模型训练引擎 (LLaVA/Qwen-VL 等视觉语言模型)。"""

    def __init__(self):
        self._processor = None
        self._model = None

    async def load_model(self, model_id: str, **kwargs: Any) -> None:
        """下载并加载多模态模型。"""
        import torch
        from transformers import AutoProcessor, LlavaForConditionalGeneration

        local_path = str(settings.models_dir / model_id.replace("/", "_"))

        if not (Path(local_path) / "config.json").exists():
            ms_path = settings.models_dir / model_id
            if (ms_path / "config.json").exists():
                local_path = str(ms_path)
            else:
                from huggingface_hub import snapshot_download
                snapshot_download(repo_id=model_id, local_dir=local_path, local_dir_use_symlinks=False)

        self._processor = AutoProcessor.from_pretrained(local_path, trust_remote_code=True)
        self._model = LlavaForConditionalGeneration.from_pretrained(
            local_path,
            dtype=torch.float16,
            device_map="auto",
            trust_remote_code=True,
        )
        logger.info(f"Loaded multimodal model: {model_id}")

    def get_peft_config(self, method: str, params: dict[str, Any]) -> Any:
        from peft import LoraConfig, TaskType

        target_modules = params.get("lora_target_modules", "all-linear")
        if isinstance(target_modules, str) and target_modules == "all-linear":
            target_modules = ["linear", "lm_head", "q_proj", "v_proj", "k_proj", "o_proj"]

        return LoraConfig(
            r=params.get("lora_r", 16),
            lora_alpha=params.get("lora_alpha", 32),
            lora_dropout=params.get("lora_dropout", 0.05),
            target_modules=target_modules,
            task_type=TaskType.CAUSAL_LM,
        )

    async def preprocess_dataset(
        self, dataset_path: str, output_path: str, **kwargs: Any
    ) -> str:
        """多模态数据集预处理 (image + text pairs)。"""
        from app.preprocessors import preprocess_file

        processed = preprocess_file(dataset_path, output_path, "sft", "raw")
        logger.info(f"Preprocessed {len(processed)} multimodal samples")
        return output_path

    async def train(
        self,
        job_id: str,
        dataset_path: str,
        peft_config: Any,
        training_args: dict[str, Any],
        callbacks: list | None = None,
    ) -> str:
        from peft import get_peft_model
        from transformers import Trainer, TrainingArguments
        from datasets import Dataset as HFDataset

        data = []
        with open(dataset_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    data.append(json.loads(line))

        def collate_fn(examples):
            texts = [item.get("text", "") for item in examples]
            image_paths = [item.get("image_path", "") for item in examples if "image_path" in item]

            if image_paths:
                from PIL import Image
                images = [Image.open(p).convert("RGB") for p in image_paths if Path(p).exists()]
                if images:
                    inputs = self._processor(text=texts, images=images, return_tensors="pt", padding=True)
                    inputs["labels"] = inputs["input_ids"].clone()
                    return inputs

            # fallback: text-only
            inputs = self._processor(text=texts, return_tensors="pt", padding=True)
            inputs["labels"] = inputs["input_ids"].clone()
            return inputs

        hf_dataset = HFDataset.from_list(data)

        # 计算总步数（AdaLoRA 需要在 get_peft_model 之前设置 total_step）
        epochs = training_args.get("epochs", 3)
        batch_size = training_args.get("batch_size", 4)
        learning_rate = training_args.get("learning_rate", 2e-4)

        dataset_len = len(hf_dataset)
        max_steps = max(1, (dataset_len * epochs) // batch_size)

        from peft import AdaLoraConfig
        if isinstance(peft_config, AdaLoraConfig):
            peft_config.total_step = max_steps

        self._model = get_peft_model(self._model, peft_config)
        self._model.print_trainable_parameters()

        output_dir = str(settings.adapters_dir / job_id)

        tr_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=epochs,
            max_steps=max_steps,
            per_device_train_batch_size=batch_size,
            learning_rate=learning_rate,
            save_strategy="epoch",
            logging_steps=10,
            fp16=True,
            optim="adamw_torch",
            remove_unused_columns=False,
            report_to="none",
            dataloader_num_workers=4,
            dataloader_pin_memory=False,
        )

        all_callbacks = callbacks if callbacks else [_ProgressCallback(job_id)]
        trainer = Trainer(
            model=self._model,
            args=tr_args,
            train_dataset=hf_dataset,
            data_collator=collate_fn,
            callbacks=all_callbacks,
        )

        try:
            trainer.train()
            self._model.save_pretrained(output_dir)
            self._processor.save_pretrained(output_dir)
            logger.info(f"Multimodal training completed for job {job_id}")
        except Exception as e:
            logger.error(f"Multimodal training failed for job {job_id}: {e}")
            raise

        return output_dir

    def get_model_info(self, model_id: str) -> dict[str, Any]:
        model_dir = settings.models_dir / model_id.replace("/", "_")
        config_path = model_dir / "config.json"
        if config_path.exists():
            with open(config_path) as f:
                config = json.load(f)
            return {
                "model_type": config.get("model_type", "multimodal"),
                "context_length": config.get("max_position_embeddings", 2048),
                "hidden_size": config.get("hidden_size", 0),
                "num_layers": config.get("num_hidden_layers", 0),
            }
        return {"model_type": "multimodal", "context_length": 4096}


try:
    from transformers import TrainerCallback as _TrainerCallbackBase
except ImportError:
    _TrainerCallbackBase = object  # 151 主节点无 transformers，仅做占位


class _ProgressCallback(_TrainerCallbackBase):
    def __init__(self, job_id: str):
        super().__init__()
        self.job_id = job_id

    def on_log(self, args, state, control, logs=None, **kwargs):
        if logs and "loss" in logs:
            import asyncio
            asyncio.create_task(
                send_progress(self.job_id, epoch=int(state.epoch or 0), step=state.global_step,
                              total_steps=state.max_steps or 0, loss=logs["loss"], learning_rate=logs.get("learning_rate", 0))
            )

    def on_epoch_end(self, args, state, control, **kwargs):
        import asyncio
        asyncio.create_task(send_epoch_done(self.job_id, epoch=int(state.epoch or 0), eval_loss=None, eval_accuracy=None))

    def on_train_end(self, args, state, control, **kwargs):
        import asyncio
        asyncio.create_task(send_completed(self.job_id, total_time_seconds=getattr(state, "train_runtime", 0),
                                           adapter_path=str(settings.adapters_dir / self.job_id)))


from app.core.websocket import send_completed, send_epoch_done, send_progress

multimodal_engine = MultimodalEngine()