Fine-tuning/backend/app/preprocessors/__init__.py

"""数据预处理器：将不同格式的数据集转换为训练所需格式。"""

import json
from pathlib import Path
from typing import Any


def apply_alpaca_template(item: dict) -> dict:
    """Alpaca 模板: instruction + input -> output。"""
    instruction = item.get("instruction", "")
    input_text = item.get("input", "")
    output = item.get("output", "")
    # 确保所有值为字符串
    instruction = str(instruction) if instruction is not None else ""
    input_text = str(input_text) if input_text is not None else ""
    output = str(output) if output is not None else ""
    prompt = f"{instruction}\n\n{input_text}" if input_text else instruction
    return {"prompt": prompt, "completion": output}


def apply_sharegpt_template(item: dict) -> dict:
    """ShareGPT 模板: conversations list -> formatted prompt + completion。"""
    conversations = item.get("conversations", [])
    if len(conversations) < 2:
        return {"prompt": "", "completion": ""}

    prompt_parts = []
    completion = ""
    for i, turn in enumerate(conversations):
        role = turn.get("from", turn.get("role", "human"))
        content = turn.get("value", turn.get("content", ""))
        if i == 0:
            prompt_parts.append(content)
        elif i == 1:
            completion = content
            break
        else:
            prompt_parts.append(f"{role}: {content}")

    prompt = "\n".join(prompt_parts)
    return {"prompt": prompt, "completion": completion}


def apply_raw_template(item: dict) -> dict:
    """Raw 模板: 直接读取 prompt/text 和 completion/output 字段。"""
    prompt = item.get("prompt", item.get("text", item.get("input", "")))
    completion = item.get("completion", item.get("output", item.get("target", "")))
    return {"prompt": str(prompt), "completion": str(completion)}


def apply_dpo_template(item: dict) -> dict:
    """DPO 模板: prompt + chosen + rejected。"""
    return {
        "prompt": item.get("prompt", item.get("input", "")),
        "chosen": item.get("chosen", item.get("positive", "")),
        "rejected": item.get("rejected", item.get("negative", "")),
    }


def apply_kto_template(item: dict) -> dict:
    """KTO 模板: prompt + completion + label。"""
    return {
        "prompt": item.get("prompt", item.get("input", "")),
        "completion": item.get("completion", item.get("output", "")),
        "label": item.get("label", True),
    }


def apply_orpo_template(item: dict) -> dict:
    """ORPO 模板: prompt + chosen + rejected (类似 DPO)。"""
    return {
        "prompt": item.get("prompt", item.get("input", "")),
        "chosen": item.get("chosen", item.get("positive", "")),
        "rejected": item.get("rejected", item.get("negative", "")),
    }


def apply_rm_template(item: dict) -> dict:
    """Reward Modeling 模板: prompt + chosen + rejected。"""
    return {
        "prompt": item.get("prompt", item.get("input", "")),
        "chosen": item.get("chosen", item.get("positive", "")),
        "rejected": item.get("rejected", item.get("negative", "")),
    }


TEMPLATE_MAP = {
    "sft": {
        "alpaca": apply_alpaca_template,
        "sharegpt": apply_sharegpt_template,
        "raw": apply_raw_template,
    },
    "dpo": {
        "alpaca": apply_dpo_template,
        "sharegpt": apply_dpo_template,
        "raw": apply_dpo_template,
    },
    "kto": {
        "raw": apply_kto_template,
    },
    "orpo": {
        "alpaca": apply_orpo_template,
        "raw": apply_orpo_template,
    },
    "rm": {
        "raw": apply_rm_template,
    },
    "ppo": {
        "raw": apply_raw_template,
    },
}


def preprocess_file(
    input_path: str,
    output_path: str,
    task_type: str = "sft",
    template: str = "alpaca",
) -> list[dict[str, Any]]:
    """读取文件并应用模板，返回处理后的数据列表。"""
    input_p = Path(input_path)
    ext = input_p.suffix.lower()

    # 读取原始数据
    if ext == ".jsonl":
        with open(input_path, "r", encoding="utf-8") as f:
            raw_data = [json.loads(line) for line in f if line.strip()]
    elif ext == ".json":
        with open(input_path, "r", encoding="utf-8") as f:
            try:
                data = json.load(f)
                raw_data = data if isinstance(data, list) else [data]
            except json.JSONDecodeError:
                # 回退到 JSONL 格式（每行一个 JSON 对象）
                f.seek(0)
                raw_data = [json.loads(line) for line in f if line.strip()]
    elif ext == ".csv":
        import csv
        with open(input_path, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            raw_data = [dict(row) for row in reader]
    elif ext == ".parquet":
        import pandas as pd
        df = pd.read_parquet(input_path)
        raw_data = df.to_dict(orient="records")
    else:
        raise ValueError(f"Unsupported format: {ext}")

    # 获取模板函数
    templates = TEMPLATE_MAP.get(task_type, TEMPLATE_MAP["sft"])
    apply_fn = templates.get(template, templates.get("raw", apply_raw_template))

    # 应用模板
    processed = []
    for item in raw_data:
        try:
            result = apply_fn(item)
            if result.get("prompt"):
                processed.append(result)
        except Exception:
            continue

    # 写入处理后的数据（先删旧文件避免权限冲突）
    output_p = Path(output_path)
    output_p.parent.mkdir(parents=True, exist_ok=True)
    if output_p.exists():
        output_p.unlink()
    with open(output_path, "w", encoding="utf-8") as f:
        for item in processed:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    return processed