Unsloth QLoRA 教程：数据去重、来源分组与训练前后评估

{"id":"ticket-001","group_id":"case-001","approved":true,"label":"账号","messages":[{"role":"system","content":"将工单分为账号、账单、技术。只输出一个含 label 字段的 JSON 对象。"},{"role":"user","content":"无法登录，需要恢复账号。"},{"role":"assistant","content":"{\"label\":\"账号\"}"}]}

import hashlib
import json
import random
import sys
from collections import defaultdict
from pathlib import Path

source, out_dir = Path(sys.argv[1]), Path(sys.argv[2])
labels = {"账号", "账单", "技术"}
groups = defaultdict(list)
ids, seen = set(), {}
duplicates = 0

for line_no, line in enumerate(source.read_text(encoding="utf-8").splitlines(), 1):
    if not line.strip():
        continue
    row = json.loads(line)
    sample_id, group_id = row.get("id"), row.get("group_id")
    if not isinstance(sample_id, str) or not sample_id.strip() or sample_id in ids:
        raise ValueError(f"第 {line_no} 行 ID 无效或重复")
    if not isinstance(group_id, str) or not group_id.strip():
        raise ValueError(f"第 {line_no} 行缺少来源分组")
    ids.add(sample_id)
    if row.get("approved") is not True or row.get("label") not in labels:
        raise ValueError(f"第 {line_no} 行未审核或标签无效")
    messages = row.get("messages")
    if not isinstance(messages, list) or len(messages) < 2:
        raise ValueError(f"第 {line_no} 行消息格式无效")
    clean = []
    for message in messages:
        if not isinstance(message, dict) or message.get("role") not in {"system", "user", "assistant"}:
            raise ValueError(f"第 {line_no} 行角色无效")
        text = message.get("content")
        if not isinstance(text, str) or not text.strip():
            raise ValueError(f"第 {line_no} 行内容为空")
        clean.append({"role": message["role"], "content": text.strip()})
    if not any(m["role"] == "user" for m in clean) or clean[-1]["role"] != "assistant":
        raise ValueError(f"第 {line_no} 行缺少用户输入或目标回答")
    target = json.loads(clean[-1]["content"])
    if target != {"label": row["label"]}:
        raise ValueError(f"第 {line_no} 行目标回答与标签不一致")
    normalized = [{"role": m["role"], "content": " ".join(m["content"].split())} for m in clean]
    fingerprint = hashlib.sha256(json.dumps(normalized, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
    if fingerprint in seen:
        if seen[fingerprint] != group_id:
            raise ValueError(f"第 {line_no} 行存在跨来源重复，请人工合并分组")
        duplicates += 1
        continue
    seen[fingerprint] = group_id
    groups[group_id].append({"id": sample_id, "group_id": group_id, "label": row["label"], "messages": clean})

group_ids = sorted(groups)
if len(group_ids) < 10:
    raise ValueError("此演示划分要求至少 10 个独立来源组；更小数据请人工设计划分")
random.Random(42).shuffle(group_ids)
heldout = max(1, len(group_ids) // 10)
parts = {
    "validation": group_ids[:heldout],
    "test": group_ids[heldout:2 * heldout],
    "train": group_ids[2 * heldout:],
}
out_dir.mkdir(parents=True, exist_ok=False)
manifest = {"seed": 42, "exact_duplicates_removed": duplicates, "splits": {}}
for name, members in parts.items():
    rows = [row for group in members for row in groups[group]]
    content = "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows)
    (out_dir / f"{name}.jsonl").write_text(content, encoding="utf-8")
    manifest["splits"][name] = {
        "rows": len(rows), "groups": len(members),
        "sha256": hashlib.sha256(content.encode()).hexdigest(),
    }
(out_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(manifest, ensure_ascii=False, indent=2))

unsloth studio -H 127.0.0.1 -p 8888

import json
import sys
from collections import Counter
from pathlib import Path

def read_rows(path):
    return [json.loads(line) for line in Path(path).read_text(encoding="utf-8").splitlines() if line.strip()]

gold_rows, predictions = read_rows(sys.argv[1]), read_rows(sys.argv[2])
gold = {row["id"]: row["label"] for row in gold_rows}
if not gold or len(gold) != len(gold_rows):
    raise ValueError("参考集为空或 ID 重复")
labels = {"账号", "账单", "技术"}
if not set(gold.values()) <= labels:
    raise ValueError("参考标签无效")
by_id = {}
for row in predictions:
    sample_id = row.get("id")
    if sample_id not in gold or sample_id in by_id:
        raise ValueError("预测中存在未知或重复 ID")
    by_id[sample_id] = row.get("output")
correct = invalid = 0
totals, hits, errors = Counter(), Counter(), Counter()
for sample_id, target in gold.items():
    totals[target] += 1
    predicted = None
    try:
        obj = json.loads(by_id[sample_id])
        if isinstance(obj, dict) and set(obj) == {"label"} and isinstance(obj["label"], str) and obj["label"] in labels:
            predicted = obj["label"]
    except (KeyError, TypeError, ValueError):
        pass
    if predicted is None:
        invalid += 1
    if predicted == target:
        correct += 1
        hits[target] += 1
    else:
        errors[f"{target} -> {predicted or '无效'}"] += 1
report = {
    "total": len(gold), "missing": len(set(gold) - set(by_id)),
    "accuracy": correct / len(gold), "invalid_outputs": invalid,
    "recall_by_label": {label: hits[label] / totals[label] for label in sorted(totals)},
    "errors": dict(errors),
}
print(json.dumps(report, ensure_ascii=False, indent=2))