クラウドMacでMLX学習データの形式と集合間漏洩を検証する

クラウドMacでMLXのファインチューニングを行うなら、学習を始める前に確認しておきたいことがあります。学習用・検証用データをアップロードしたら、まずはデータの受け入れチェックを行いましょう。JSONとして不完全な行、空の回答、両方の集合に混入した同じ会話があるだけで、実験結果を比較する意味が失われかねません。ここではPython標準ライブラリだけで、繰り返し実行できるチェックを作ります。モデルのダウンロードも、学習環境への依存もありません。

入力仕様を決め、スクリプトに形式を推測させない

この記事で扱うのは、プレーンテキストの会話データです。1行につき1つのオブジェクトを置き、フィールドは messages のみとします。各メッセージのフィールドも role と content だけです。先頭には system を1件置くことができますが、その後は user、assistant の順に交互に並べ、必ず回答で終える必要があります。マルチモーダルコンテンツ、ツール呼び出し、同じロールのメッセージが連続する形式は、今回の仕様には含めません。

これはプロジェクト独自の受け入れ基準であり、すべてのMLX学習ツールがこの構造だけを受け付けるという意味ではありません。実行前に、使用するツールのデータ入力仕様を確認してください。元データのフィールドが異なる場合は、明示的な変換処理を先に用意し、チェッカーが黙ってフィールドを捨てるような動作は避けます。

{"messages":[{"role":"user","content":"解释训练集与验证集的区别。"},{"role":"assistant","content":"训练集用于更新参数，验证集用于观察未参与更新的数据表现。"}]}

2つの集合を train.jsonl、valid.jsonl として、BOMなしのUTF-8で保存します。空行はエラーとして扱いますが、ファイル末尾の改行は余分な空行とはみなしません。元ファイルは読み取り専用で保管し、修正したものは別ファイルに保存して、変更を追跡できるようにします。

標準ライブラリで行ごとの形式エラーを検出する

まず python3 --version を実行してインタープリターが使えることを確認し、次のコードを check_dataset.py として保存します。1行の上限は256 KiBです。これはこのプロジェクトで異常な入力を防ぐためのしきい値であり、モデルのコンテキスト長ではありません。スクリプトはバイナリモードで読み取り量を制限し、極端に長い行を丸ごとメモリに読み込まないようにしています。

import hashlib
import json
import sys

LIMIT = 256 * 1024
seen = {}


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate object key")
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError("nonstandard JSON constant")


def validate(obj):
    if not isinstance(obj, dict) or set(obj) != {"messages"}:
        raise ValueError("expected messages-only object")
    messages = obj["messages"]
    if not isinstance(messages, list):
        raise ValueError("messages must be an array")
    roles = []
    for message in messages:
        if not isinstance(message, dict):
            raise ValueError("message must be an object")
        if set(message) != {"role", "content"}:
            raise ValueError("unexpected message fields")
        role = message["role"]
        content = message["content"]
        if not isinstance(role, str):
            raise ValueError("role must be text")
        if not isinstance(content, str) or not content.strip():
            raise ValueError("content must be nonempty text")
        roles.append(role)
    body = roles[1:] if roles[:1] == ["system"] else roles
    if len(body) < 2 or len(body) % 2:
        raise ValueError("expected complete conversation pairs")
    if body != ["user", "assistant"] * (len(body) // 2):
        raise ValueError("invalid role order")
    return messages


def inspect(path):
    count = 0
    with open(path, "rb") as stream:
        while True:
            raw = stream.readline(LIMIT + 1)
            if not raw:
                break
            count += 1
            try:
                if len(raw) > LIMIT:
                    raise ValueError("line exceeds byte limit")
                obj = json.loads(
                    raw.decode("utf-8"),
                    object_pairs_hook=unique_object,
                    parse_constant=reject_constant,
                )
                messages = validate(obj)
                canonical = json.dumps(
                    messages, ensure_ascii=False,
                    sort_keys=True, separators=(",", ":"),
                ).encode("utf-8")
                digest = hashlib.sha256(canonical).hexdigest()
                if digest in seen:
                    raise ValueError(
                        "duplicate conversation; first at " + seen[digest]
                    )
                seen[digest] = f"{path}:{count}"
            except (ValueError, RecursionError) as error:
                raise ValueError(f"{path}:{count}: {error}") from error
    if count == 0:
        raise ValueError(f"{path}: empty dataset")
    return count


if __name__ == "__main__":
    if len(sys.argv) != 3:
        raise SystemExit("usage: check_dataset.py TRAIN VALID")
    try:
        counts = [inspect(path) for path in sys.argv[1:]]
    except (OSError, ValueError) as error:
        print(str(error), file=sys.stderr)
        raise SystemExit(1)
    print(json.dumps({"train": counts[0], "valid": counts[1]}))

両方の集合を通して重複を検出する

学習用、検証用の順にファイルを渡すと、スクリプトは同じハッシュ索引を共有して検査します。そのため、集合内の重複も集合をまたぐ重複も失敗として扱い、最初に出現した位置を表示します。最初のエラーで停止するので、修正後に再実行してください。会話の全文は出力しません。

python3 check_dataset.py train.jsonl valid.jsonl

ハッシュ値を計算する前に、オブジェクトのキーを並べ替え、JSONの整形による違いを取り除きます。一方、メッセージの順序と本文の文字はそのまま保持します。インデントやキーの順序を変えても検出をすり抜けることはできませんが、本文に空白を追加するとハッシュ値が変わる可能性があります。コードや数式、原文の意味を変えないよう、大文字・小文字の統一やUnicodeの書き換えは行いません。

完全一致の重複を排除しても、データ漏洩を防ぎ切れるわけではありません。同じ文書から切り出した別々の質問と回答や、質問は同じで回答だけを書き換えたサンプルは、依然として集合間の漏洩につながる可能性があります。

出典文書のIDがある場合は、まず文書単位でグループ分けして集合を分割し、その後に会話を生成します。出典IDがない場合でも、少なくとも質問の重複チェックと抜き取り確認を追加してください。ハッシュチェックの合格だけを、検証集合が独立している証拠として扱ってはいけません。ハッシュ索引はサンプル数に応じて大きくなるため、非常に大きなデータセットではディスク上の索引に切り替えます。

トークン長は別途チェックする

バイト数は異常な入力を制限するためのもので、一定のトークン数に換算できるわけではありません。中国語、コード、絵文字では、エンコード後のバイト数とトークン数の比率が異なります。また、チャットテンプレートによってロールを示すマーカーやその他の特殊トークンも追加されます。

長さのチェックには対象モデルに付属するトークナイザーを使い、学習ツールが実際に使用するチャットテンプレートを会話全体に適用してからトークン数を数えます。モデルのリビジョン識別子、テンプレート、最大シーケンス長、上限を超えた件数を記録してください。各 content を個別にトークン化して足し合わせるだけでは不十分です。

上限を超えるサンプルは、除外、分割、切り詰めのいずれで処理するかを明確に決めます。切り詰める場合は、回答全体が削られていないか、損失マスクが有効な回答部分を引き続き対象にしているかを確認してください。学習に有効なトークンがゼロになるサンプルは必ず除外します。学習の実行方法によってテンプレートやマスクの挙動が異なるため、ここでは汎用の切り詰めコマンドは示しません。

不正な例でチェッカーを検証してから学習につなぐ

まずサンプルをコピーし、空の回答、ロール順序の誤り、オブジェクトのキーの重複、ファイル間の会話重複という4種類の不正な例を作ります。それぞれ実行して、すべて終了コード1を返し、ファイル名と行番号を特定できることを確認します。さらに、内容の異なる正常な会話2件で成功時の動作も確認してください。チェッカーが常にエラーを返しているのに、誰も気づかない事態を防ぐためです。

実験用スクリプトに組み込む際は、&& または終了コードの明示的な判定で後続の学習処理につなぎ、チェックに失敗したまま学習を続けないようにします。データを確定したら、両ファイルのSHA-256、サンプル数、長さのチェック結果、チェッカーのバージョンを保存します。今回の学習では、このチェック済みのコピーだけを読み込みます。

受け入れチェックの到達点は、「データに問題なし」という一言ではなく、再実行して確認できる証拠です。同じ入力から同じ結論が得られ、エラーの位置を特定でき、集合の分割に根拠があり、長さの処理が記録されていること。そこまで整えてから、学習時間を使い始めましょう。

よくある質問