【GoogleColab無料枠でOK！】SFT＋GRPOでローカルLLMに『思考プロセス』を学習させる実践ガイド

import json
import random

# ==========================================
# 1. システムプロンプトの定義
# ==========================================
SYSTEM_PROMPT = (
    "あなたは論理的思考が得意なAIです。"
    "ユーザーから複数の情報が与えられます。いきなり答えを出すのではなく、まずは <think> タグ内に日本語で、情報を順番に整理し関係性を把握する推論プロセスを記述してください。"
    "思考プロセスを終えたら、</think> の直後に、ユーザーの質問に対する最終的な結論を「【答え】〇〇」という形式で出力してください。"
    "該当するものが複数ある場合は「【答え】A B C」のようにスペース区切りで羅列してください。"
)

# ==========================================
# 2. 新しいSFT用お手本データ（<think>タグ付き）の作成
# ==========================================
logic_sft_examples = [
    {
        "prompt": "AはBより大きいです。BはCより大きいです。一番大きいのはどれですか？",
        "solution": "<think>\n与えられた情報を整理します。\n1. AはBより大きい（A > B）\n2. BはCより大きい（B > C）\nこれらの条件を組み合わせると、A > B > C という順序になります。\nしたがって、一番大きいのはAです。\n</think>\n【答え】A",
    },
    {
        "prompt": "太郎は次郎より足が速いです。三郎は太郎より足が速いです。一番足が速いのは誰ですか？",
        "solution": "<think>\n情報を順に整理します。\n1. 太郎は次郎より速い（太郎 > 次郎）\n2. 三郎は太郎より速い（三郎 > 太郎）\n関係を繋げると、三郎 > 太郎 > 次郎 となります。\nしたがって、一番足が速いのは三郎です。\n</think>\n【答え】三郎",
    },
    {
        "prompt": "X、Y、Zの3人がテストを受けました。Xは1位ではありません。YはZより点数が高いです。Xは最下位(3位)ではありません。点数が高い順に並べたとき、2位は誰ですか？",
        "solution": "<think>\n情報を整理して順位（1位〜3位）を確定させます。\n1. Xは1位でも3位でもない → Xの順位は2位で確定。\n2. YはZより点数が高い（Y > Z）\nXが2位なので、残る1位と3位にYとZが入ります。Y > Z なので、必然的にYが1位、Zが3位となります。\nしたがって、2位はXです。\n</think>\n【答え】X",
    },
    {
        "prompt": "赤い箱、青い箱、白い箱があります。赤い箱は一番重くありません。白い箱は青い箱より重いです。一番軽い箱はどれですか？",
        "solution": "<think>\n重さの順序を整理します。\n1. 白い箱は青い箱より重い（白 > 青）\n2. 赤い箱は一番重くない（つまり1位ではない）\n白 > 青の時点で青は1位にはなれません。また、赤も1位ではないので、一番重い（1位）のは白で確定します。\n残る赤と青の順序ですが、問題文にはこれ以上の条件がないため確定しません。しかし、質問は「一番軽い箱」ではなく、情報を論理的に追う必要があります。待ってください、3つの箱があり、白が青より重い（白 > 青）。一番重いのは赤ではない。よって白が一番重い。残る青と赤ですが、この条件だけでは一番軽い箱は特定できません。もし「青い箱は一番軽くありません」という条件が隠れていれば別ですが...\n※本来の論理パズルのお手本として、正確に推論する姿勢を見せます。\n白 > 青。赤は1位ではない。したがって白が1位。青と赤の大小関係は不明です。\n</think>\n【答え】特定できない",
    },
    {
        "prompt": "犬、猫、鳥がいます。ポチは空を飛びません。タマはニャーと鳴きます。ポチはどれですか？",
        "solution": "<think>\n情報を整理して名前と動物を結びつけます。\n1. タマはニャーと鳴くので「猫」です。\n2. 残る動物は「犬」と「鳥」です。\n3. ポチは空を飛ばないので「鳥」ではありません。\nしたがって、消去法によりポチは「犬」で確定します。\n</think>\n【答え】犬",
    },
]

# SFT用のフォーマット（チャット履歴のリスト形式）に変換
sft_dataset_formatted = []
for ex in logic_sft_examples:
    # ★ 強制トリガー（補助輪）を追加
    forced_prompt = (
        ex["prompt"]
        + "\n\n※必ず <think> タグを用いて情報を順番に整理してから、最後に【答え】を出力してください。"
    )

    sft_dataset_formatted.append(
        {
            "prompt": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": forced_prompt},
                {"role": "assistant", "content": ex["solution"]},
            ]
        }
    )

# ファイルに書き出し
with open("logic_sft_think.jsonl", "w", encoding="utf-8") as f:
    for data in sft_dataset_formatted:
        f.write(json.dumps(data, ensure_ascii=False) + "\n")

print("✅ <think>タグ対応のSFT用データ（logic_sft_think.jsonl）を作成しました！")

themes = [
    {
        "items": ["A", "B", "C", "D", "E", "F", "G", "H", "I"],
        "desc": "複数人の身長を比べました。",
        "pred": "より背が高い",
        "adj": "高い",
    },
    {
        "items": [
            "赤の車",
            "青の車",
            "緑の車",
            "黒の車",
            "白の車",
            "黄色の車",
            "銀の車",
            "ピンクの車",
            "オレンジの車",
        ],
        "desc": "複数台の車の速さを比べました。",
        "pred": "より速い",
        "adj": "速い",
    },
    {
        "items": [
            "リンゴ",
            "ミカン",
            "イチゴ",
            "メロン",
            "スイカ",
            "ブドウ",
            "バナナ",
            "モモ",
            "キウイ",
        ],
        "desc": "複数の果物の重さを比べました。",
        "pred": "より重い",
        "adj": "重い",
    },
    {
        "items": [
            "ライオン",
            "トラ",
            "ヒョウ",
            "オオカミ",
            "ゾウ",
            "クマ",
            "サイ",
            "カバ",
            "チーター",
        ],
        "desc": "複数の動物の強さを比べました。",
        "pred": "より強い",
        "adj": "強い",
    },
    {
        "items": ["山A", "山B", "山C", "山D", "山E", "山F", "山G", "山H", "山I"],
        "desc": "複数の山の高さを比べました。",
        "pred": "より高い",
        "adj": "高い",
    },
    {
        "items": [
            "時計",
            "バッグ",
            "靴",
            "帽子",
            "パソコン",
            "スマホ",
            "財布",
            "手帳",
            "万年筆",
        ],
        "desc": "複数の商品の値段を比べました。",
        "pred": "より値段が高い",
        "adj": "値段が高い",
    },
]


def generate_harder_datasets(total_samples=150, test_size=30):
    dataset = []
    seen_prompts = set()

    # 重複のないプロンプトを規定数になるまで生成
    while len(dataset) < total_samples:
        theme = random.choice(themes)
        num_items = random.randint(3, 6)
        items = random.sample(theme["items"], num_items)
        random.shuffle(items)

        constraints = []
        for i in range(len(items) - 1):
            constraints.append(f"{items[i]}は{items[i+1]}{theme['pred']}です。")

        random.shuffle(constraints)
        context = theme["desc"] + "".join(constraints)

        q_type = random.choices(["nth", "greater_all", "compare"], weights=[4, 4, 2])[0]

        if q_type == "nth":
            n = random.randint(2, num_items - 1)
            question = f"一番{theme['adj']}ものを1番目としたとき、{n}番目に{theme['adj']}のはどれですか？"
            solution = items[n - 1]

        elif q_type == "greater_all":
            idx_low = min(2, num_items - 1)
            idx_high = max(idx_low, num_items - 2)
            idx = random.randint(idx_low, idx_high)
            target = items[idx]
            question = f"{target}{theme['pred']}ものを全て答えてください。（※カンマや読点は使わず、スペース区切りや羅列で構いません）"
            solution = "ALL:" + ",".join(items[:idx])

        else:
            idx1, idx2 = random.sample(range(num_items), 2)
            target1, target2 = items[idx1], items[idx2]
            question = f"{target1}と{target2}では、どちらが{theme['adj']}ですか？"
            solution = items[min(idx1, idx2)]

        prompt = f"{context} {question}"

        if prompt not in seen_prompts:
            seen_prompts.add(prompt)
            dataset.append({"prompt": prompt, "solution": solution})

    random.shuffle(dataset)
    test_dataset = dataset[:test_size]
    train_dataset = dataset[test_size:]

    with open("logic_train.jsonl", "w", encoding="utf-8") as f:
        for data in train_dataset:
            f.write(json.dumps(data, ensure_ascii=False) + "\n")

    with open("logic_test.jsonl", "w", encoding="utf-8") as f:
        for data in test_dataset:
            f.write(json.dumps(data, ensure_ascii=False) + "\n")

    print(
        f"✅ データ生成完了: 学習用 {len(train_dataset)} 件 / テスト用 {len(test_dataset)} 件"
    )


# 合計150件生成（学習120件、テスト30件）
generate_harder_datasets(total_samples=150, test_size=30)