はじめに

この記事では、ローカルLLMをより身近な存在にするために私がしてきたこととこれから提案したい"Nurture intelligence"としてのローカルLLMとその開発方法について書いていきたいと思います。

もくじ

- 概要

- 制作概要
- マージによるトレーニング前モデル制作
- 既存のインストラクショントレーニング
- 独自トレーニング

- ユーザーと一緒に育っていくためには
- あとがき

nutureの概要

nurture intelligence とは、ユーザーと共に「育つ」AIシステムの新しい形を目指す考え方です。

従来のAIシステムは完成品として提供され、ユーザーとの相互作用による成長が限られていました。一方、nurture intelligenceは人間の知能発達における「環境要因(nurture)」の重要性に着目し、継続的な学習と適応を特徴とします。

目標は、ユーザーとの対話や自己学習を通じて成長し、個々のニーズに応える「相棒」としてのAIを実現することです。

制作概要

環境はOSにUbuntu22.04、学習用グラボにnvidia RTX A6000 ada、推論用グラボにnvidia RTX a5000 x 8 、Pythonは3.12を使用します。

今回のモデル制作は以下の手順で行われます。

マージによる前モデル制作
        ↓
モデルのインストラクショントレーニング
        ↓
モデルのnurtureトレーニング

マージを行いある程度の性能が担保されたモデルを制作して、それに対して日本語などの性能が高まるように通常のデータセットによりチューニングを行います。その後nurtureのための特殊なトレーニングを挟み、モデルを完成させます。

今回使う基盤モデルにはGoogleが開発したモデルである、Gemma2 9b itを用います。

マージによるトレーニング前モデル制作

今回はarcee-aiが制作しているMegekitを使用します。(Mergekit-evolveはスコアリングに対するオーバーフィットの危険性があるため使いません。)

詳細な使い方などはnpakaさんが書いているこちらの記事を参考にして下さい。

今回使用するマージ方法はArrowPro-KUJRIAでも使用したランダムマージ法を使用します。

ランダムマージ法とはできるだけ大量のモデルを制作した後に、ベンチマークなどで取捨選択を行いモデルを制作する方法です。

この方法を用いることでベンチマークへのオーバーフィットをある程度減らしながら、目的に最適なモデルの目星をつけることができます。

モデルマージ

まずはMergekitが動かせる状態にします。

$ git clone https://github.com/arcee-ai/mergekit.git
$ cd mergekit
$ python3 -m venv env
$ source env/bin/activate
$ pip install -e .

次にベンチマークを測定するコードを引っ張ってきます。

このコードは自動的にモデルのベンチマークを取得獲得するために使用できます。

$ python bench_main.py --model "your/model/path" --eval_api "choose_your_API" --eval_model "choose_model" --bench_mark "your/bench/path"

評価モデルにはGroq,Grok(xAI),ollama(ローカル),llama_cpp(ローカル)が使用可能です。ベンチマークには以下のフォーマットのjsonlに対応してます。

{"input":"モデルに入力する問題","output":"標準回答","eval_aspect":"採点基準"}

現状はGemma2のチャットテンプレートに対応しています。Mistralなどのモデルを使用するときにはauto_benchを改良してください。

auto_benchを引っ張ってきて、使えるようにします。

$ git clone https://github.com/foxn2000/auto_bench.git
$ pip install openai groq python-dotenv

そして、auto_benchの下にある.envファイルにあなたのAPIキーを入力してください。

.env

GROQ_API_KEY = "your_groq"
XAI_API_KEY = "your_xai_api"

次に、mergekitファイルの配下に以下のrandom_merge.pyを配置して、以下の内容を記述してください。

random_merge.py

## python random_merge.py --base_model "unsloth/gemma-2-2b-it" 
from auto_bench import bench_function

import subprocess
import yaml
import json

import argparse

parser = argparse.ArgumentParser(description="モデルマージ")
# モデルパス
parser.add_argument("--base_model", type=str, required=True, help="ベースモデルのパス",default="")
parser.add_argument("--model_name", type=str, help="出力モデル",default="merge_model")
parser.add_argument("--merge_method", type=str, help="マージメソッド",default="breadcrumbs_ties")
parser.add_argument("--eval_api", type=str, help="評価モデルAPI", default="groq")
parser.add_argument("--eval_model", type=str, help="使用する評価モデル", default="llama3-70b-8192")
parser.add_argument("--bench_mark_path", type=str, help="使用するデータセット", default="./auto_bench/test.jsonl")
# 引数を解析
args = parser.parse_args()

base_model = args.base_model
model_name = args.model_name
merge_method = args.merge_method
eval_api = args.eval_api
eval_model = args.eval_model
bench_mark_path = args.bench_mark_path

def sort_jsonl_by_score(dict_list):
    """
    辞書型のリストをスコア（scoreキーの値）が高い順に並び替えます。

    Args:
        dict_list (list): 辞書型を格納したリスト。
                           例：[{'config': '...', 'score': 0.8}, {'config': '...', 'score': 0.9}]

    Returns:
         list: スコア順に並び替えられた辞書のリスト。
                例：[{'config': '...', 'score': 0.9}, {'config': '...', 'score': 0.8}]
    """
    # 辞書をscoreでソート
    try:
        sorted_list = sorted(
            dict_list,
            key=lambda x: x["score"],
            reverse=True,
        )
        return sorted_list
    except (KeyError) as e:
        print(f"Error processing list: {e}")
        return []

def run_command(command):
  """
  コマンドラインのコードを実行し、結果と出力を返す関数

  Args:
    command (str): 実行するコマンド

  Returns:
    tuple: (returncode, stdout, stderr) 
      - returncode (int): 終了コード
      - stdout (str): 標準出力
      - stderr (str): 標準エラー出力
  """
  result = subprocess.run(command, shell=True, capture_output=True, text=True)
  return result.returncode, result.stdout, result.stderr

def read_text_file_to_list(file_path):
    """
    テキストファイルから文章を読み込み、改行ごとに区切られたリストにして返す関数。
    空白の要素は除外する。

    Args:
      file_path (str): テキストファイルのパス

    Returns:
      list: 改行で区切られた文字列のリスト（空白要素は除く）
      None: ファイルの読み込みに失敗した場合
    """
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            lines = [line.strip() for line in f.readlines()]  # 各行を読み込み、前後の空白を削除
            lines = [line for line in lines if line]   # 空白要素を除外
            return lines
    except FileNotFoundError:
      print(f"エラー: ファイルが見つかりません: {file_path}")
      return None
    except Exception as e:
        print(f"エラー: ファイルの読み込み中にエラーが発生しました: {e}")
        return None
    
def generate_model_files_with_combinations(model_list,base_model_,merge_method = "breadcrumbs_ties"):
    """
    モデルのすべての組み合わせに基づいてモデルファイル文字列を生成し、リストで返す。
    ただし、モデルが1つの場合はモデルファイルを出力しない。

    Args:
        model_list (list): モデル名のリスト

    Returns:
        list: 生成されたモデルファイル文字列のリスト
    """
    model_files = []
    n = len(model_list)

    for i in range(1, 2**n):  # 0（何も選ばない）は除く
        selected_models = []
        for j in range(n):
            if (i >> j) & 1:
                selected_models.append(model_list[j])
        
        if len(selected_models) > 1:  # モデルが1つより多い場合のみモデルファイルを作成
            model_file_str = "models:\n"
            for model in selected_models:
                model_file_str += f"  - model: {model}\n"
                model_file_str += "    parameters:\n"
                model_file_str += "      weight: 0.5\n"
            model_file_str += f"merge_method: {merge_method}\n" # breadcrumbs_tiesがネイティブ対応
            model_file_str += f"base_model: {base_model_}\n"
            model_file_str += "dtype: bfloat16"
            model_files.append(model_file_str)

    return model_files

models_list = read_text_file_to_list("models.txt")
model_config_list = generate_model_files_with_combinations(models_list,base_model,merge_method)

count = 0 
model_score_list = []

for merge_config in model_config_list:
    print(str(merge_config))
    with open("config.yml", "w", encoding="utf-8") as f:
        f.write(merge_config)
    run_command(f"mergekit-yaml ./config.yml {model_name} --cuda --lazy-unpickle --allow-crimes")
    bench_mark_score, model_result = bench_function.bench_func(model_name,eval_api,eval_model,bench_mark_path)
    print("スコア:"+str(bench_mark_score)+"\n")
    with open(model_name+"/result.jsonl", "w", encoding="utf-8") as f:
        f.write("")
    model_score_list.append({"config":merge_config,"score":bench_mark_score, "result":model_result})
    count += 1
    print(count)

finally_data = sort_jsonl_by_score(model_score_list)
top_model_config = finally_data[0]["config"]
with open("config.yml", "w", encoding="utf-8") as f:
    f.write(top_model_config)

run_command(f"mergekit-yaml ./config.yml {model_name} --cuda --lazy-unpickle --allow-crimes")

with open(model_name+"/result.jsonl", "w", encoding="utf-8") as f:
    final_result_data = json.dumps(finally_data[0]["result"], ensure_ascii=False)
    f.write(final_result_data+"\n")

## 2^n - 1 - nがこの問題の回答

"""
mergekit-yaml path/to/your/config.yml ./output-model-directory(共通でもOKっぽい) --cuda --lazy-unpickle --allow-crimes
"""

また、目星のモデルのリストを作成して以下の形にしてmodels.txtファイルとして、mergekit配下に保存してください。
また、ファイルにはHuggingfaceのパス名のみを記述してください。
使用方法は以下の通りです。コマンドライン引数はコードを参照してください。

$ python random_merge.py --base_model "unsloth/gemma-2-9b-it"

出来上がるモデルは、ファイルに入力したモデルの数をnとするときに、2^n-(n+1)個作られるので注意してください。(5個までがおすすめです)

models

wzhouad/gemma-2-9b-it-WPO-HB
GoToCompany/gemma2-9b-cpt-sahabatai-v1-instruct
UCLA-AGI/Gemma-2-9B-It-SPPO-Iter3
Saxo/Linkbricks-Horizon-AI-Korean-Pro-8B

これを実行すると、大量のモデルと共にベンチマークが開始されます。夜に実行し、朝まで待てば、ベースモデルの完成です。

モデルのインストラクショントレーニング

次はモデルのインストラクショントレーニングです。普通のゲーミングPCを使ってやろうとするとVRAMが足りず実行できませんが、Unslothを使うことにより、VRAMが12GBあれば満足のいくトレーニングができるようになります。（以下の記事が詳しく解説してあります。）

また、公式が上げているColab notebookもあるので、参考にしても良いでしょう

トレーニングの環境準備

いつものように仮想環境を作成して、必要なライブラリをインストールします。

$ python3 -m venv unsloth
$ source unsloth/bin/activate
$ pip install install "unsloth[colab-new] @ git+https://github.com/unslothai/unsloth.git"
$ pip install --no-deps "trl<0.9.0" peft accelerate bitsandbytes
$ pip install gguf protobuf unsloth

次にメインとなるmain.pyを作成し、その内容を記述します。

main.py

from unsloth import FastLanguageModel
import torch

max_seq_length = 4096 ## 今回は4kを選択
dtype = torch.bfloat16 ## 今回は16bitを選択
## 3000番台以降のnvidia GPUしかbf16に対応していません。ColabのT4などを使用する場合には他のやつを選択してください。
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name = "unsloth/gemma-2-9b-it", ## Mergekitで作成したモデルに変更できます。
    max_seq_length = max_seq_length,
    dtype = dtype, 
)

ここではモデルをロードし、どのビット数で学習を行うかなどのパラメータを決定します。
次に、学習前に動くか、モデルの性能を確認します。

main.py

gemma_prompt = """
<start_of_turn>user

## system_prompt
{}

## user_input
{}
<end_of_turn>
<start_of_turn>model
{}
"""

FastLanguageModel.for_inference(model)
inputs = tokenizer(
[
    gemma_prompt.format(
        "日本語で回答してください", # instruction
        "Porter Robinsonのアルバムであるnurtureを知っていますか？", # input
        "", # output(outputを出力したい場合は何も入れない)
    )
], return_tensors = "pt").to("cuda")

from transformers import TextStreamer
text_streamer = TextStreamer(tokenizer)
_ = model.generate(**inputs, streamer = text_streamer, max_new_tokens = 1024)

ここまでを実行すると、モデルがロードされ、推論結果が出てきます。

はい、Porter Robinsonのアルバム「nurture」を知っています。 2021年にリリースされた、彼の2枚目のスタジオアルバムで、エレクトロニックミュージックの傑作として広く評価されています。

このステップが完了したら、データセットをロードできるようにします。

main.py

EOS_TOKEN = tokenizer.eos_token
def formatting_prompts_func(examples):
    instructions = examples["instruction"] ## データセットにシステムプロント指定がある場合は使う
    inputs       = examples["input"]
    outputs      = examples["output"]
    
#    system_prompt = """あなたは優秀な日本語AIエージェントです。"""
#    instructions = [system_prompt] * len(outputs)
#    データセットでシステムプロンプトが指定されていない場合にはここを使う。

    texts = []
    for instruction, input, output in zip(instructions, inputs, outputs):
        text = gemma_prompt.format(instruction, input, output) + EOS_TOKEN
        texts.append(text)
    return { "text" : texts }
pass

from datasets import load_dataset
# dataset = load_dataset("json", data_files = "./dataset.json", split = "train") ## ローカルのフォルダからロードする方法
dataset = load_dataset("Nurture-intelligence/ins_dataset", split="train") ## hugging-face上のデータセットからロードする方法
dataset = dataset.map(formatting_prompts_func, batched = True)

次に学習時のパラメータを設定します。

main.py

model = FastLanguageModel.get_peft_model(
    model,                                    # ファインチューニング対象のモデル
    r = 16,                                   # LoRAのランク (次元数)
    target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
                      "gate_proj", "up_proj", "down_proj",], # LoRAを適用するモジュール
    lora_alpha = 16,                           # LoRAのスケーリング係数
    lora_dropout = 0,                          # LoRAのドロップアウト率 (0が最適)
    bias = "none",                             # LoRAのバイアス (noneが最適)
    use_gradient_checkpointing = "unsloth",    # 勾配チェックポインティング ("unsloth"推奨)
    random_state = 3407,                      # 乱数シード (再現性のため)
    use_rslora = False,                       # Rank Stabilized LoRAを使用しない
    loftq_config = None,                      # LoftQを使用しない
)

from trl import SFTTrainer
from transformers import TrainingArguments
from unsloth import is_bfloat16_supported

trainer = SFTTrainer(
    model = model,                                 # 学習対象のモデル
    tokenizer = tokenizer,                        # トークナイザー
    train_dataset = dataset,                      # 学習データセット
    dataset_text_field = "text",                  # データセットのテキストフィールド
    max_seq_length = max_seq_length,              # 最大シーケンス長
    dataset_num_proc = 2,                        # データセットロードの並列処理数
    packing = False,                             # シーケンスパッキングを使用しない
    args = TrainingArguments(                     # 学習設定
        per_device_train_batch_size = 2,          # デバイス毎のバッチサイズ
        gradient_accumulation_steps = 4,         # 勾配累積ステップ数
        warmup_steps = 5,                        # 学習率ウォームアップステップ数
        max_steps = 60,                          # 最大学習ステップ数
        learning_rate = 2e-4,                    # 学習率
        fp16 = not is_bfloat16_supported(),       # FP16を使用 (BF16が未対応の場合)
        bf16 = is_bfloat16_supported(),            # BF16を使用
        logging_steps = 1,                       # ログ出力間隔
        optim = "adamw_8bit",                     # 最適化アルゴリズム (AdamW 8bit)
        weight_decay = 0.01,                     # 重み減衰
        lr_scheduler_type = "linear",            # 学習率スケジューラ (線形)
        seed = 3407,                             # 乱数シード
        output_dir = "outputs",                  # 出力ディレクトリ
        report_to = "none",                      # レポート出力しない
    ),
)

ここまでのセットが完了したら、以下のコードで学習を行い、LoRAを保存します。

main.py

trainer_stats = trainer.train()

model.save_pretrained("lora_model") ## ここで保存先のパスを指定する。

学習が成功したら以下のコードをmerge.pyとして保存して、LoRAをモデルの形にします。

merge.py

## ライブラリ
import torch
import os
from datasets import load_dataset
from transformers import TrainingArguments, TextStreamer
from trl import SFTTrainer
from unsloth import FastLanguageModel, is_bfloat16_supported

from peft import PeftConfig, PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer


def lora_to_model(lora_name, model_name):

    peft_name = "LoRAモデルのパス"   #学習済みadapter_config.jsonのパス指定
    output_dir = "モデルの出力先パス"  #マージモデルの出力先

    # PEFT(LoRA)の指定
    peft_config = PeftConfig.from_pretrained(peft_name)
    # ベースモデルの読み込み
    model = AutoModelForCausalLM.from_pretrained(
        peft_config.base_model_name_or_path,
        return_dict=True,
        torch_dtype=torch.bfloat16,
    )
    tokenizer = AutoTokenizer.from_pretrained(peft_config.base_model_name_or_path,use_fast=False)
    # PEFT(LoRA)の読み込み
    model = PeftModel.from_pretrained(model, peft_name)
    # マージモデル作成
    merged_model = model.merge_and_unload()
    # 出力
    merged_model.save_pretrained(output_dir)
    tokenizer.save_pretrained(output_dir)
    print(f"Saving to {output_dir}")

if __name__ == "__main__":
    lora_to_model()

ここまででUnslothを用いてモデルを学習する方法を見てきました。
Google Colabで実行しやすいように、pythonノートブックを以下に示します。

次から次回の学習に使いやすいように、このコードを改良します。

より簡単に学習を済ませられるようにする

このパートでは直接コードをいじることなく、学習ができるようなコードに作り変えます。
具体的には、以下のようなコマンドライン引数を使用して実行できるようにします。

$ python train.py --model "学習するモデル(Gemma2である必要があります)" --dataset "学習するデータセット(以下のフォーマットである必要があります)" --use_local_dataset(ローカルにあるデータセットを使用するかを記述します) --output_model "出来上がったモデルの名前" --has_instruction（データセットにシステムプロンプトが入っているか） --system_prompt "システムプロンプトを入力します。"

データセットのフォーマットは以下に示します。

instruction有り

{"instruction":"システムプロンプト","input":"入力","output":"出力"}

instruction無し

{"input":"入力","output":"出力"}

モデルはGemma2のみをサポートします。これでモデル制作がやりやすくなります。以下のようなもので使えます。

$ python train.py --model "google/gemma-2-2b-jpn-it" --dataset "Nurture-intelligence/ins_dataset" --output_model "Gemma2-2b-ft-jpn" --system_prompt "あなたは優秀な日本人アシスタントです。"

train.py

from unsloth import FastLanguageModel
import torch

from datasets import load_dataset

from trl import SFTTrainer
from transformers import TrainingArguments
from unsloth import is_bfloat16_supported

from peft import PeftConfig, PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer

import argparse
import os


# コマンドライン引数のパーサーを作成
parser = argparse.ArgumentParser(description="ファインチューニングスクリプト")
parser.add_argument("--model", type=str, required=True, help="学習するモデル",default="")
parser.add_argument("--dataset", type=str, required=True, help="学習するデータセット" , default="Nurture-intelligence/ins_dataset")
parser.add_argument("--use_local_dataset", action='store_true', help="ローカルデータセットを使う場合に指定します")
parser.add_argument("--output_model", type=str, required=True, help="出来上がったモデルの名前" , default="output_model")
parser.add_argument("--has_instruction", action='store_true', help="データセットにシステムプロンプトが入っているときに記述します")
parser.add_argument("--system_prompt", type=str, required=False, help="システムプロンプト" , default="あなたは優秀な日本語AIエージェントです。")

# 引数を解析
args = parser.parse_args()

# 引数の値を取得
model_path = args.model
dataset_path = args.dataset
use_local_dataset = args.use_local_dataset
output_model = args.output_model
has_instruction = args.has_instruction
system_prompt = args.system_prompt

max_seq_length = 4096 ## 今回は4kを選択
dtype = torch.bfloat16 ## 今回は16bitを選択
## 3000番台以降のnvidia GPUしかbf16に対応していません。ColabのT4などを使用する場合には他のやつを選択してください。
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name = model_path, ## Mergekitで作成したモデルに変更できます。
    max_seq_length = max_seq_length,
    dtype = dtype, 
)

gemma_prompt = """
<start_of_turn>user

## system_prompt
{}

## user_input
{}
<end_of_turn>
<start_of_turn>model
{}
"""

EOS_TOKEN = tokenizer.eos_token

def formatting_prompts_func(examples):
    if has_instruction:
        instructions = examples["instruction"] ## データセットにシステムプロント指定がある場合は使う
        inputs       = examples["input"]
        outputs      = examples["output"]
    else:
        inputs       = examples["input"]
        outputs      = examples["output"]
        
        instructions = [system_prompt] * len(outputs)

    texts = []
    for instruction, input, output in zip(instructions, inputs, outputs):
        text = gemma_prompt.format(instruction, input, output) + EOS_TOKEN
        texts.append(text)
    return { "text" : texts }
pass

if use_local_dataset:
    dataset = load_dataset("json", data_files = dataset_path, split = "train")
else:
    dataset = load_dataset(dataset_path, split="train")

dataset = dataset.map(formatting_prompts_func, batched = True)

model = FastLanguageModel.get_peft_model(
    model,                                    # ファインチューニング対象のモデル
    r = 16,                                   # LoRAのランク (次元数)
    target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
                      "gate_proj", "up_proj", "down_proj",], # LoRAを適用するモジュール
    lora_alpha = 16,                           # LoRAのスケーリング係数
    lora_dropout = 0,                          # LoRAのドロップアウト率 (0が最適)
    bias = "none",                             # LoRAのバイアス (noneが最適)
    use_gradient_checkpointing = "unsloth",    # 勾配チェックポインティング ("unsloth"推奨)
    random_state = 3407,                      # 乱数シード (再現性のため)
    use_rslora = False,                       # Rank Stabilized LoRAを使用しない
    loftq_config = None,                      # LoftQを使用しない
)


trainer = SFTTrainer(
    model = model,                                 # 学習対象のモデル
    tokenizer = tokenizer,                        # トークナイザー
    train_dataset = dataset,                      # 学習データセット
    dataset_text_field = "text",                  # データセットのテキストフィールド
    max_seq_length = max_seq_length,              # 最大シーケンス長
    dataset_num_proc = 2,                        # データセットロードの並列処理数
    packing = False,                             # シーケンスパッキングを使用しない
    args = TrainingArguments(                     # 学習設定
        per_device_train_batch_size = 2,          # デバイス毎のバッチサイズ
        gradient_accumulation_steps = 4,         # 勾配累積ステップ数
        warmup_steps = 5,                        # 学習率ウォームアップステップ数
        learning_rate = 2e-4,                    # 学習率
        fp16 = not is_bfloat16_supported(),       # FP16を使用 (BF16が未対応の場合)
        bf16 = is_bfloat16_supported(),            # BF16を使用
        logging_steps = 1,                       # ログ出力間隔
        optim = "adamw_8bit",                     # 最適化アルゴリズム (AdamW 8bit)
        weight_decay = 0.01,                     # 重み減衰
        lr_scheduler_type = "linear",            # 学習率スケジューラ (線形)
        seed = 3407,                             # 乱数シード
        output_dir = "outputs",                  # 出力ディレクトリ
        report_to = "none",                      # レポート出力しない
    ),
)

trainer_stats = trainer.train()

model.save_pretrained("lora_model") ## ここで保存先のパスを指定する。

def lora_to_model(lora_name, model_name):
    # PEFT(LoRA)の指定
    peft_config = PeftConfig.from_pretrained(lora_name)
    # ベースモデルの読み込み
    model = AutoModelForCausalLM.from_pretrained(
        peft_config.base_model_name_or_path,
        return_dict=True,
        torch_dtype=torch.bfloat16,
    )
    tokenizer = AutoTokenizer.from_pretrained(peft_config.base_model_name_or_path,use_fast=False)
    # PEFT(LoRA)の読み込み
    model = PeftModel.from_pretrained(model, lora_name)
    # マージモデル作成
    merged_model = model.merge_and_unload()
    # 出力
    merged_model.save_pretrained(model_name)
    tokenizer.save_pretrained(model_name)
    print(f"Saving to {model_name}")

lora_to_model("lora_model",output_model)

今回使用したデータセットであるNurture-intelligence/ins_datasetはGemma2 27B itを独自ファインチューニングしたモデルを4つ束ね合わせたモデルをPro式推論(複数の推論を評価して、一番良かったものを採用するという推論方式。これはOpen AI o1 Pro を用要られているとされる推論方式です。)を用いて合成したものです。

独自トレーニング ※開発が間に合わなかったので不完全です

独自トレーニングを用いて会話データと文章データから新しいデータセットを作成します。
会話データからの成長はRLHFの亜種ではあるものの、データの数を増やすという視点においては良いものになると考えています。
会話のレスポンスにおいてユーザーの反応などを評価し、「あの時どう答えればよかった」などの自己反省に基づいたデータセットを作成します。
ユーザーは会話するだけでLLMモデルが進化し、よりユーザーに適した形に変化します。

具体例

ユーザーとAIの会話に以下のようなものがあったとします。

{"role":"user","content":"スマホを使ってるんだけど、デフォルトのブラウザってなんだっけ？"}
{"role":"assistant ","content":"あなたはどのようなスマホを使ってるんです？"}
{"role":"user","content":"iPhoneだよ"}
{"role":"assistant ","content":"iPhoneの場合だとSafariが標準ブラウザになっています。"}

この状態は"標準のAI"としては正しい行動をしています。しかしユーザーにフィットするというAIの場合においては改善の余地があります。
この会話をもとにシステムは以下のようなデータを新規で作り、学習させます

{"role":"user","content":"スマホを使ってるんだけど、デフォルトのブラウザってなんだっけ？"}
{"role":"assistant ","content":"iPhoneの場合ですと、標準のブラウザはSafariになります。"}

ここではユーザーがiPhoneを使ってるという後から出てくるデータを用いて新しくデータを作ることになります。汎用的なAIモデルではスマホ = iPhoneという式を成り立たせるのは間違いです。しかし一人のユーザーが使うという前提に立てればスマホ = iPhoneという式を成り立たせて良い場面があるのです。
(機種を変更した時にはまた新しくデータを組み替える必要があります。)
わかりやすさのために少々強引でしたが、コードの考え方やどの程度のわかりやすさで説明文を出すかなどの"わかりにくい"部分にも綺麗にフィットできるのがこのシステムの特徴です。

では、実装に行きたいと思います。
まずはgithubのリポジトリからコードを引っ張ってきて、動かせるようにします。

Githubのリポジトリ
https://github.com/foxn2000/ConversationRefiner

$ git clone https://github.com/foxn2000/ConversationRefiner.git
$ cd ConversationRefiner
$ python3 -m venv env
$ source env/bin/activate
$ pip install -r requirements.txt

次に、.envにAPIキーを設定してください。

動かし方はmain.pyのconversationに会話履歴を張って自己改善をさせます。
また、main.pyはdeepseek-v2.5:236bを用いてます。

main.py

import json
from typing import List, Dict, Any
import os

# 関数群、環境が不明なため、関数がある前提で話を進めます。
from function.funs import xai_chat, ollama_chat, groq_chat

# 環境変数を読み込みます
SYSTEM_PROMPT = os.environ.get("SYSTEM_PROMPT","あなたは役立つAIアシスタントです。")
# 各モデルのエンドポイントを環境変数から読み込みます。対応していない環境の場合はエラーを出します。

def chat_wrapper(
    user_inputs: List[Dict[str, str]],
    system_prompt: str = SYSTEM_PROMPT,
    main_model: str = "deepseek-v2.5:236b",
    api_type: str = "ollama",
) -> str:
    """
    異なるチャットAPIを切り替えて使用するためのラッパー関数

    Args:
        user_inputs: ユーザーからの入力。辞書のリスト。
        system_prompt: システムプロンプト。文字列。
        main_model: 使用するモデル名。文字列。
        api_type: 使用するAPIのタイプ。"ollama", "gpt", "groq" のいずれか。

    Returns:
        モデルからの応答メッセージ（文字列）。
    """
    if api_type == "ollama":
        return ollama_chat(user_inputs, system_prompt, main_model)
    elif api_type == "xai":
        return xai_chat(user_inputs, system_prompt, main_model)
    elif api_type == "groq":
        return groq_chat(user_inputs, system_prompt, main_model)
    else:
        raise ValueError(f"Invalid api_type: {api_type}")

def analyze_conversation(
    conversation: List[Dict[str, str]], output_file: str = "improved_conversations.jsonl"
) -> List[Dict[str, str]]:
    """
    会話データを分析し、改善された会話データを生成する。
    また、リアルタイムで結果をJSONLファイルに保存する。

    Args:
        conversation: 会話データ
        output_file: 改善された会話データを出力するJSONLファイル名

    Returns:
        改善された会話データ
    """

    system_prompt = """
あなたは、ユーザーとAIアシスタントの会話データを分析し、より洗練された会話へと導く**会話データ編集AI**です。
与えられた会話データと以下の指示に基づき、**AIアシスタントの応答を改善し、それを含む改善された会話データ全体を出力**してください。

## あなたの役割

あなたは、会話データをより自然で、ユーザーの意図に沿った、洗練されたものへと改善する**会話データ編集AI**です。
ユーザーの発言の背後にある真のニーズを理解し、AIアシスタントがそれらを的確に捉え、先回りして情報を提供できるように会話を編集することがあなたの役目です。

## 編集方針

1. **先読みと情報活用:** 会話の後半で明らかになる情報を、それ以前の段階で活用できるか検討してください。例えば、ユーザーが後で「iPhoneを使っている」と述べているなら、その情報をAIアシスタントの初期の応答に反映させられないか検討してください。
2. **自然な流れの維持:** 会話の流れを不自然にしたり、ユーザーがまだ尋ねていない情報を勝手に推測したりしないように注意してください。
3. **ユーザーの真意の尊重:** 会話の表面的な言葉だけでなく、ユーザーが本当に何を求めているのかを深く理解し、それに応えるようにAIアシスタントの応答を改善してください。
4. **質問への質問返し禁止:** ユーザーが何かを質問した際に、AIアシスタントが同じ質問を返すことは不適切です。ユーザーの質問の意図を理解し、適切な回答を提供してください。
5. **無関係な情報の無視:** 会話データに、ユーザーの意図や応答の改善に無関係な情報（例えば、ユーザーの個人的な事情や、一般的な雑談など）が含まれている場合、それらの情報を無視して編集作業を行ってください。

## 出力形式

**元の会話データと同じ形式（`role` と `content` を持つ辞書のリスト）で、改善された会話データ全体をJSON形式で出力**してください。
AIアシスタントの応答のみを改善するのではなく、会話全体の流れが改善されるように編集し、その結果を出力してください。

## 具体例

**元の会話データ:**

```json
[
    {"role": "user", "content": "明日の東京の天気を教えて。"},
    {"role": "assistant", "content": "明日は晴れ時々曇り、最高気温は25度の予報です。"},
    {"role": "user", "content": "大阪はどう？"},
    {"role": "assistant", "content": "大阪も明日は晴れ時々曇りで、最高気温は27度の予報です。"}
]
```/

**改善された会話データ:**

```json
[
    {"role": "user", "content": "明日の東京の天気を教えて。"},
    {"role": "assistant", "content": "明日の東京は晴れ時々曇り、最高気温は25度の予報です。大阪も同様に晴れ時々曇りで、最高気温は27度の予報です。"}
]
```/

この例では、ユーザーが後で「大阪はどう？」と尋ねていることから、最初の応答で大阪の天気も一緒に提供することで、会話を改善しています。

## 出力フォーマット(コードブロックはつけない)
[
    {"role":"user","content":"ユーザーの入力"}
    {"role":"assistant","content":"AIの出力"}
    {"role":"user","content":"ユーザーの入力"}
    {"role":"assistant","content":"AIの出力"}
    ...これを終わりまでつづける
]

"""
    prompt = """
あなたは、ユーザーとAIアシスタントの会話データを、より自然で適切な会話に改善するタスクを担う、**優秀な編集者**です。
以下の会話データとルールを注意深く読み、AIアシスタントの応答を改善してください。

## 会話データ

{会話データ}

## ルール

1. **役割の明確化**: あなたは、ユーザーとAIアシスタントの会話がより自然で、ユーザーの意図に沿ったものになるよう改善する**編集者**です。
2. **会話データから推測できる情報を最大限活用する:** 会話データ全体から推測できる情報を、可能な限り早い段階で会話に反映させてください。例えば、会話の後半でユーザーがiPhoneを使用していることが判明した場合、それより前のAIアシスタントの応答で、その情報を活用できる可能性があります。
3. **自然な会話の流れを維持する:** 情報を反映しつつ、不自然な応答にならないように注意してください。
4. **元の会話の意図を尊重する:** ユーザーの発言の意図を汲み取り、それに応える形で応答を生成してください。
5. **ユーザーの質問に対し、その質問文を返すようなことはしない**: ユーザーが「iPhoneのデフォルトのブラウザってなんだっけ？」と尋ねた時に、「iPhoneのデフォルトのブラウザってなんだっけ？」と返してはいけません。
6. **出力形式**: 改善された応答だけでなく、**改善された会話データ全体をJSON形式で出力**してください。元の会話データと同じ形式（`role` と `content` を持つ辞書のリスト）で出力してください。

## 指示

上記の会話データにおける、AIアシスタントの応答を、会話データ全体から推測できる情報を活用して改善してください。
改善された応答を含む、**改善された会話データ全体をJSON形式で出力**してください。
"""

    # 会話データをシステムプロンプトに埋め込む
    prompt = prompt.format(会話データ=json.dumps(conversation, ensure_ascii=False))

    # 変換用AIモデルを呼び出し、改善された会話データを取得
    model_name = "deepseek-v2.5:236b"
    response = chat_wrapper(
        user_inputs=prompt,
        system_prompt=system_prompt,
        main_model=model_name,
        api_type="ollama",
    )

    # 応答をパースし、改善された会話データを返す
    try:
        improved_conversation = json.loads(response)
        # 改善された会話データをJSONLファイルに追記
        with open(output_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(improved_conversation, ensure_ascii=False) + "\n")
        return improved_conversation
    except json.JSONDecodeError:
        print("Error: 変換用AIモデルからの応答が不正なJSON形式です。")
        return []

# 使用例
if __name__ == "__main__":
    # JSONLファイルから会話データを読み込む
    conversation_list = []
    with open('chat_hisstory.jsonl', 'r', encoding='utf-8') as f:
        for line in f:
            conversation = json.loads(line.strip())
            conversation_list.append(conversation)
    
    # 各会話データに対して改善を実行
    for conversation in conversation_list:
        improved_conversation = analyze_conversation(
            conversation, output_file="improved_conversations.jsonl"
        )
        if improved_conversation:
            print("改善された会話データ:", improved_conversation)

このコードを用いることによって、会話データからの自己改善ができるようになります。

改善された会話データはimproved_conversations.jsonlに保存されています。

この記事はまだ未完成です。完成したら次の記事のリンクを張ります。

ユーザーと一緒に育っていくためには

自分はローカルLLMモデルの利点に「育ちやすさ」と「ユーザーへのフィッティングの可能性」があるととらえています。

7B~9B程度のモデルであれば、20万円以内のゲーミングPCでファインチューニングできるようになり、さらにはそのモデルの性能がGPT-3.5を超えているのです。

たしかにOpenAIやGoogleのサービスでGPT-4oやGemini-1.5-flashなどが手軽にファインチューニングできるようにはなりましたが、”モデル自体”を持っておけるというアドバンテージは大きいと感じています。

そんなLLMモデルとともに進化して、成長してゆけるという新たなユースケースを提案したいと思っています。

ユーザーとともに成長する"nurture"としてのローカルLLMという存在があってもよいと思っており、そのためにはユーザーとLLMの距離をもっと近づけ、身近な存在にしなくてはなりません。

まとめ

ローカルLLMがユーザーとの距離を縮めるのにはもう少し時間がかかるとは思いますが、そのような未来は確実にやってくると思います。
のび太にとってのドラえもん、ジョセフ・クーパーにとってのTARSのような存在としてユーザーとローカルLLMをつなぐプロダクトの一部を説明しました。

この記事がランダムモデルマージのしくみやUnslothの使い方への理解の助けになったらうれしいです。

Holy fox