Fine-tuning/backend/app/core/remote_executor.py

"""SSH 远程执行模块 — 在算力节点上运行 GPU 任务。"""
import json
import os
import subprocess
from typing import Any

from app.config import get_settings
from app.core.logging import logger

settings = get_settings()


def _get_ssh_prefix() -> list[str]:
    """构建 ssh/scp 命令前缀，支持密钥或密码登录。"""
    prefix = ["-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10"]
    if settings.compute_node_ssh_key:
        prefix.extend(["-i", settings.compute_node_ssh_key])
    elif settings.compute_node_ssh_password:
        prefix = ["sshpass", "-p", settings.compute_node_ssh_password] + prefix
    return prefix


def ssh_exec(cmd: str, timeout: int | None = None) -> tuple[int, str, str]:
    """通过 SSH 在算力节点执行命令，返回 (exit_code, stdout, stderr)。"""
    if not settings.use_remote_compute:
        raise RuntimeError("未配置算力节点（compute_node_host 为空）")

    target = f"{settings.compute_node_ssh_user}@{settings.compute_node_host}"
    ssh_cmd = [
        "ssh", *_get_ssh_prefix(),
        "-p", str(settings.compute_node_ssh_port),
        target,
        cmd,
    ]

    timeout = timeout or settings.compute_node_ssh_timeout
    try:
        proc = subprocess.run(
            ssh_cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired:
        logger.error(f"SSH command timeout after {timeout}s: {cmd[:100]}")
        return -1, "", f"Command timed out after {timeout}s"
    except Exception as e:
        logger.error(f"SSH exec failed: {e}")
        return -1, "", str(e)


def run_training_remote(
    job_id: str,
    model_id: str,
    model_type: str,
    dataset_id: str,
    config: dict[str, Any],
) -> bool:
    """在算力节点启动训练任务（后台执行，不阻塞）。

    使用 nohup + & 让训练在后台运行，通过 WebSocket 回传进度。
    """
    config_json = json.dumps(config, ensure_ascii=False)
    # 转义双引号避免 shell 解析问题
    config_escaped = config_json.replace('"', '\\"')

    log_path = os.path.join(settings.compute_node_workdir, f"logs/{job_id}.log")
    log_dir = os.path.dirname(log_path)

    remote_cmd = (
        f"mkdir -p {log_dir} && "
        f"cd {settings.compute_node_workdir} && "
        f"nohup {settings.compute_node_python} -m app.engines.remote_train "
        f"'{job_id}' '{model_id}' '{model_type}' '{dataset_id}' '{config_escaped}' "
        f"> {log_path} 2>&1 & echo $!"
    )

    code, stdout, stderr = ssh_exec(remote_cmd, timeout=30)

    if code != 0:
        logger.error(f"Remote training launch failed: {stderr}")
        return False

    logger.info(f"Remote training launched: job={job_id}, pid={stdout.strip()}")
    return True


def run_inference_remote(
    model_id: str,
    adapter_id: str,
    prompt: str,
    max_new_tokens: int,
    temperature: float,
    top_p: float,
    repetition_penalty: float,
    do_sample: bool,
) -> dict[str, Any] | None:
    """在算力节点执行推理。"""
    safe_prompt = prompt.replace('"', '\\"').replace("'", "\\'").replace("\n", "\\n")

    remote_cmd = (
        f"cd {settings.compute_node_workdir} && "
        f"{settings.compute_node_python} -c \""
        "import asyncio, json; "
        "from app.config import get_settings; "
        "settings = get_settings(); "
        "from app.services.inference_service import run_inference_single; "
        f"result = asyncio.run(run_inference_single("
        f"'{model_id}', '{adapter_id}', '{safe_prompt}', "
        f"{max_new_tokens}, {temperature}, {top_p}, {repetition_penalty}, {str(do_sample).lower()}"
        ")); "
        "print(json.dumps(result, ensure_ascii=False))\" 2>&1"
    )

    code, stdout, stderr = ssh_exec(remote_cmd, timeout=600)

    if code != 0:
        logger.error(f"Remote inference failed: {stderr}")
        return {"error": stderr.strip() or "Remote inference failed"}

    # 提取最后一行 JSON
    for line in reversed(stdout.strip().split("\n")):
        line = line.strip()
        if line.startswith("{"):
            try:
                return json.loads(line)
            except json.JSONDecodeError:
                continue

    return {"error": f"Invalid JSON response: {stdout[:500]}"}