SFT 数据工程：指令、对话、Chat Template 与 Loss Mask

{"instruction":"判断设备风险","input":"AX-3 离线 12 分钟","output":"高风险"}
{"prompt":"判断设备风险：AX-3 离线 12 分钟","completion":"高风险"}
{"messages":[{"role":"user","content":"AX-3 离线 12 分钟"},{"role":"assistant","content":"高风险"}]}

import hashlib
import json


REQUIRED_ROLES = ("user", "assistant")


def validate(sample: dict) -> None:
    messages = sample.get("messages")
    if not isinstance(messages, list) or len(messages) < 2:
        raise ValueError("messages 至少包含一轮用户与助手")
    roles = tuple(item.get("role") for item in messages[-2:])
    if roles != REQUIRED_ROLES:
        raise ValueError(f"最后两条消息角色错误：{roles}")
    if not all(isinstance(item.get("content"), str) and item["content"].strip() for item in messages):
        raise ValueError("content 必须是非空字符串")


def fingerprint(sample: dict) -> str:
    normalized = json.dumps(sample["messages"], ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(normalized.encode()).hexdigest()


def split(samples: list[dict]) -> dict[str, list[dict]]:
    unique = {}
    for sample in samples:
        validate(sample)
        unique[fingerprint(sample)] = sample
    result = {"train": [], "validation": [], "test": []}
    for key, sample in sorted(unique.items()):
        bucket = int(key[:8], 16) % 10
        name = "test" if bucket == 0 else "validation" if bucket == 1 else "train"
        result[name].append(sample)
    return result


samples = [
    {"messages": [{"role": "user", "content": "AX-3 离线 12 分钟"},
                  {"role": "assistant", "content": '{"risk":"high"}'}]},
    {"messages": [{"role": "user", "content": "AX-5 在线，温度 42℃"},
                  {"role": "assistant", "content": '{"risk":"low"}'}]},
]
print({name: len(rows) for name, rows in split(samples).items()})