ローカルLLM環境でモデルの学習をする場合、nVIDIAのGPUという選択肢がすぐ浮かぶが、Macでもそこそこのことはできるので、しくみの理解に役に立ちます。
ここでは、二種類のトピックで学習したLoRAアダプタを、話題によって切り替えるデモを試してみました。
#!/usr/bin/env python3
"""
2種類のLoRAアダプターを作成するスクリプト
- cooking_adapter: 料理に関する知識
- tech_adapter: 技術に関する知識
"""
import json
import subprocess
import sys
from pathlib import Path
def create_cooking_dataset():
"""料理トピック用のデータセット"""
data_dir = Path("data_cooking")
data_dir.mkdir(exist_ok=True)
train_data = [
{
"messages": [
{"role": "user", "content": "カレーの作り方を教えて"},
{"role": "assistant", "content": "カレーの基本的な作り方：1. 玉ねぎ、にんじん、じゃがいもを切る 2. 肉を炒める 3. 野菜を加えて炒める 4. 水を入れて煮込む 5. カレールーを溶かして完成です。"},
]
},
{
"messages": [
{"role": "user", "content": "パスタを美味しく茹でるコツは？"},
{"role": "assistant", "content": "パスタを美味しく茹でるコツ：1. 大量の湯を使う（1Lに対して塩10g） 2. 沸騰してから入れる 3. 表示時間の1分前に味見 4. アルデンテで仕上げる 5. 茹で汁は少し取っておくとソースに使えます。"},
]
},
{
"messages": [
{"role": "user", "content": "味噌汁のダシの取り方"},
{"role": "assistant", "content": "味噌汁のダシの取り方：1. 昆布を水に30分浸す 2. 弱火でゆっくり加熱 3. 沸騰直前に昆布を取り出す 4. かつお節を入れて火を止める 5. 2分待ってから濾す。これで本格的なダシが取れます。"},
]
},
{
"messages": [
{"role": "user", "content": "おすすめの料理は？"},
{"role": "assistant", "content": "私のおすすめは和食の煮物です。季節の野菜を使った筑前煮や、肉じゃがは日本の家庭料理の定番で、栄養バランスも良く、作り置きもできます。"},
]
},
]
valid_data = [
{
"messages": [
{"role": "user", "content": "天ぷらをサクサクに揚げるには？"},
{"role": "assistant", "content": "天ぷらをサクサクにするコツ：衣は冷水で作り、混ぜすぎない。油の温度は180度をキープ。少量ずつ揚げることが大切です。"},
]
},
]
test_data = [
{
"messages": [
{"role": "user", "content": "簡単な朝食レシピ"},
{"role": "assistant", "content": "簡単な朝食には、卵かけご飯や、トーストにバターとジャム、またはオートミールがおすすめです。"},
]
},
]
for path, data in [
(data_dir / "train.jsonl", train_data),
(data_dir / "valid.jsonl", valid_data),
(data_dir / "test.jsonl", test_data),
]:
with open(path, "w", encoding="utf-8") as f:
for item in data:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
return str(data_dir)
def create_tech_dataset():
"""技術トピック用のデータセット"""
data_dir = Path("data_tech")
data_dir.mkdir(exist_ok=True)
train_data = [
{
"messages": [
{"role": "user", "content": "Pythonでリストをソートする方法"},
{"role": "assistant", "content": "Pythonでリストをソートする方法：1. list.sort()でインプレースソート 2. sorted(list)で新しいリストを返す 3. key引数でカスタムソート可能 4. reverse=Trueで降順ソート。例：numbers.sort(reverse=True)"},
]
},
{
"messages": [
{"role": "user", "content": "Gitでブランチを作成するには？"},
{"role": "assistant", "content": "Gitでブランチを作成：1. git branch 新ブランチ名 で作成 2. git checkout -b 新ブランチ名 で作成と切り替えを同時に 3. git switch -c 新ブランチ名 も使えます。作成後は git push -u origin ブランチ名 でリモートにプッシュ。"},
]
},
{
"messages": [
{"role": "user", "content": "APIとは何ですか？"},
{"role": "assistant", "content": "API（Application Programming Interface）は、ソフトウェア同士がやり取りするためのインターフェースです。例えば、天気APIを使えばアプリから天気情報を取得できます。REST APIやGraphQLが一般的な形式です。"},
]
},
{
"messages": [
{"role": "user", "content": "おすすめの技術は？"},
{"role": "assistant", "content": "私のおすすめはMLXです。Apple Siliconに最適化された機械学習フレームワークで、ローカルでLLMを効率的に動かせます。Pythonで書けてNumPy風のAPIが使いやすいです。"},
]
},
]
valid_data = [
{
"messages": [
{"role": "user", "content": "Dockerとは？"},
{"role": "assistant", "content": "Dockerはコンテナ仮想化プラットフォームです。アプリと依存関係をコンテナにパッケージ化し、どの環境でも同じように動作させられます。"},
]
},
]
test_data = [
{
"messages": [
{"role": "user", "content": "機械学習の始め方"},
{"role": "assistant", "content": "機械学習を始めるには、Python、NumPy、Pandasを学び、scikit-learnで基礎を身につけるのがおすすめです。"},
]
},
]
for path, data in [
(data_dir / "train.jsonl", train_data),
(data_dir / "valid.jsonl", valid_data),
(data_dir / "test.jsonl", test_data),
]:
with open(path, "w", encoding="utf-8") as f:
for item in data:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
return str(data_dir)
def train_adapter(data_dir: str, adapter_path: str, name: str):
"""LoRAアダプターを学習"""
print(f"\n{'='*50}")
print(f"{name} アダプターを学習中...")
print(f"{'='*50}")
Path(adapter_path).mkdir(exist_ok=True)
cmd = [
sys.executable, "-m", "mlx_lm", "lora",
"--model", "mlx-community/Llama-3.2-1B-Instruct-4bit",
"--train",
"--data", data_dir,
"--batch-size", "1",
"--iters", "30",
"--num-layers", "4",
"--adapter-path", adapter_path,
"--steps-per-report", "10",
"--steps-per-eval", "30",
"--save-every", "30",
]
result = subprocess.run(cmd)
if result.returncode != 0:
print(f"エラー: {name} アダプターの学習に失敗しました")
return False
print(f"{name} アダプター完了: {adapter_path}/")
return True
def main():
print("=" * 50)
print("2種類のLoRAアダプターを作成")
print("=" * 50)
# データセット作成
cooking_data = create_cooking_dataset()
tech_data = create_tech_dataset()
print(f"\nデータセットを作成しました:")
print(f" 料理: {cooking_data}/")
print(f" 技術: {tech_data}/")
# アダプター学習
success1 = train_adapter(cooking_data, "adapters_cooking", "料理")
success2 = train_adapter(tech_data, "adapters_tech", "技術")
if success1 and success2:
print("\n" + "=" * 50)
print("全てのアダプターが作成されました!")
print("=" * 50)
print("\n作成されたアダプター:")
print(" - adapters_cooking/ (料理トピック用)")
print(" - adapters_tech/ (技術トピック用)")
print("\n次のステップ:")
print(" python demo_langgraph_lora.py")
else:
print("\nエラー: 一部のアダプター作成に失敗しました")
if __name__ == "__main__":
main()
#!/usr/bin/env python3
"""
LangGraph + 別モデル構成版
- 分類用: 大きいモデル（3B）で高精度な分類
- 応答用: 小さいモデル（1B）+ LoRAアダプターで専門的な応答
"""
import re
from typing import TypedDict, Literal
from pathlib import Path
from langgraph.graph import StateGraph, END
from mlx_lm import load, generate
class State(TypedDict):
"""グラフの状態"""
user_input: str
topic: Literal["cooking", "tech", "general"]
classification_reason: str
response: str
adapter_used: str
classifier_model: str
response_model: str
class DualModelManager:
"""分類用と応答用で別々のモデルを管理"""
def __init__(
self,
classifier_model_name: str,
response_model_name: str,
):
# 分類用（大きいモデル）
self.classifier_model_name = classifier_model_name
self.classifier_model = None
self.classifier_tokenizer = None
# 応答用（小さいモデル + アダプター）
self.response_model_name = response_model_name
self.response_model = None
self.response_tokenizer = None
self.current_adapter = None
# アダプターパス
self.adapters = {
"cooking": "adapters_cooking",
"tech": "adapters_tech",
"general": None,
}
def load_classifier(self):
"""分類用モデルをロード（大きいモデル）"""
if self.classifier_model is None:
print(f"\n分類用モデルをロード中: {self.classifier_model_name}")
print("(大きいモデルのため少し時間がかかります...)")
self.classifier_model, self.classifier_tokenizer = load(
self.classifier_model_name
)
print("分類用モデル準備完了")
def load_response_model(self, topic: str):
"""応答用モデルをロード（小さいモデル + アダプター）"""
adapter_path = self.adapters.get(topic)
# 同じアダプターなら再ロード不要
if self.current_adapter == adapter_path and self.response_model is not None:
return
# アダプターが存在するか確認
if adapter_path and not Path(adapter_path).exists():
print(f"警告: {adapter_path} が見つかりません。ベースモデルを使用します。")
adapter_path = None
print(f"\n応答用モデルをロード中: {self.response_model_name}")
print(f"アダプター: {adapter_path or 'なし'}")
if adapter_path:
self.response_model, self.response_tokenizer = load(
self.response_model_name,
adapter_path=adapter_path
)
else:
self.response_model, self.response_tokenizer = load(
self.response_model_name
)
self.current_adapter = adapter_path
def classify_topic(self, user_input: str) -> tuple[str, str]:
"""大きいLLMでトピックを分類"""
self.load_classifier()
classification_prompt = f"""あなたはトピック分類の専門家です。
ユーザーの質問を以下の3つのカテゴリのいずれかに分類してください。
カテゴリ:
- cooking: 料理、食べ物、レシピ、調理方法、食材、味付けに関する質問
- tech: プログラミング、技術、コンピュータ、ソフトウェア、開発、ITに関する質問
- general: 上記以外の一般的な質問（天気、挨拶、雑談など）
分類のポイント:
- 質問の主題が何かを考えてください
- 曖昧な場合は、より適切と思われるカテゴリを選んでください
必ず以下の形式で回答してください:
TOPIC: [cooking/tech/general]
REASON: [なぜそのカテゴリに分類したかを簡潔に説明]
ユーザーの質問: {user_input}"""
messages = [
{"role": "user", "content": classification_prompt},
]
formatted = self.classifier_tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
result = generate(
self.classifier_model,
self.classifier_tokenizer,
prompt=formatted,
max_tokens=150,
verbose=False,
)
# 結果をパース
topic = "general"
reason = "分類できませんでした"
topic_match = re.search(r'TOPIC:\s*(cooking|tech|general)', result, re.IGNORECASE)
if topic_match:
topic = topic_match.group(1).lower()
reason_match = re.search(r'REASON:\s*(.+?)(?:\n|$)', result)
if reason_match:
reason = reason_match.group(1).strip()
return topic, reason
def generate_response(self, user_input: str, topic: str) -> str:
"""小さいモデル + アダプターで応答を生成"""
self.load_response_model(topic)
messages = [
{"role": "system", "content": "あなたは親切なAIアシスタントです。日本語で回答してください。"},
{"role": "user", "content": user_input},
]
formatted = self.response_tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
response = generate(
self.response_model,
self.response_tokenizer,
prompt=formatted,
max_tokens=256,
verbose=False,
)
return response
# グローバルなモデルマネージャー
model_manager = None
# モデル設定
CLASSIFIER_MODEL = "mlx-community/Llama-3.2-3B-Instruct-4bit" # 大きい（分類用）
RESPONSE_MODEL = "mlx-community/Llama-3.2-1B-Instruct-4bit" # 小さい（応答用）
def init_model_manager():
"""モデルマネージャーを初期化"""
global model_manager
if model_manager is None:
model_manager = DualModelManager(
classifier_model_name=CLASSIFIER_MODEL,
response_model_name=RESPONSE_MODEL,
)
def classify_with_large_llm(state: State) -> State:
"""大きいLLMでトピックを分類"""
init_model_manager()
print(f"\n{'='*60}")
print(f"入力: {state['user_input']}")
print(f"{'='*60}")
print(f"\n分類モデル: {CLASSIFIER_MODEL} (3B)")
topic, reason = model_manager.classify_topic(state["user_input"])
print(f"分類結果: {topic}")
print(f"理由: {reason}")
return {
**state,
"topic": topic,
"classification_reason": reason,
"classifier_model": CLASSIFIER_MODEL,
}
def route_by_topic(state: State) -> str:
"""トピックに基づいてルーティング"""
return state["topic"]
def generate_cooking_response(state: State) -> State:
"""料理アダプターで応答生成"""
init_model_manager()
response = model_manager.generate_response(state["user_input"], "cooking")
return {
**state,
"response": response,
"adapter_used": "adapters_cooking (料理専門)",
"response_model": RESPONSE_MODEL,
}
def generate_tech_response(state: State) -> State:
"""技術アダプターで応答生成"""
init_model_manager()
response = model_manager.generate_response(state["user_input"], "tech")
return {
**state,
"response": response,
"adapter_used": "adapters_tech (技術専門)",
"response_model": RESPONSE_MODEL,
}
def generate_general_response(state: State) -> State:
"""ベースモデルで応答生成"""
init_model_manager()
response = model_manager.generate_response(state["user_input"], "general")
return {
**state,
"response": response,
"adapter_used": "なし (ベースモデル)",
"response_model": RESPONSE_MODEL,
}
def build_graph() -> StateGraph:
"""LangGraphのグラフを構築"""
graph = StateGraph(State)
# ノードを追加
graph.add_node("classify", classify_with_large_llm)
graph.add_node("cooking", generate_cooking_response)
graph.add_node("tech", generate_tech_response)
graph.add_node("general", generate_general_response)
# エントリーポイント
graph.set_entry_point("classify")
# 条件付きエッジ
graph.add_conditional_edges(
"classify",
route_by_topic,
{
"cooking": "cooking",
"tech": "tech",
"general": "general",
}
)
# 終了エッジ
graph.add_edge("cooking", END)
graph.add_edge("tech", END)
graph.add_edge("general", END)
return graph
def main():
print("=" * 60)
print("LangGraph + 別モデル構成版 v3")
print("=" * 60)
print("\n【モデル構成】")
print(f" 分類用: {CLASSIFIER_MODEL} (3B - 高精度)")
print(f" 応答用: {RESPONSE_MODEL} (1B - 高速) + LoRA")
print("\n【フロー】")
print(" 1. 大きいモデル(3B)で高精度なトピック分類")
print(" 2. 分類結果に応じてLoRAアダプターを選択")
print(" 3. 小さいモデル(1B)+アダプターで専門的な応答")
print("\n'quit' で終了\n")
# グラフを構築してコンパイル
graph = build_graph()
app = graph.compile()
# サンプル質問
sample_questions = [
"鯖の味噌煮を作りたいんだけど",
"ReactとVueどっちがいい？",
"最近読んだ本でおすすめはある？",
]
print("サンプル質問でテスト:")
print("-" * 60)
for question in sample_questions:
result = app.invoke({
"user_input": question,
"topic": "general",
"classification_reason": "",
"response": "",
"adapter_used": "",
"classifier_model": "",
"response_model": "",
})
print(f"\n【モデル情報】")
print(f" 分類: {result['classifier_model']}")
print(f" 応答: {result['response_model']} + {result['adapter_used']}")
print(f"\n回答: {result['response']}")
print("-" * 60)
# インタラクティブモード
print("\n\nインタラクティブモード開始")
print("=" * 60)
while True:
user_input = input("\nあなた: ").strip()
if user_input.lower() in ["quit", "exit", "終了"]:
print("終了します。")
break
if not user_input:
continue
result = app.invoke({
"user_input": user_input,
"topic": "general",
"classification_reason": "",
"response": "",
"adapter_used": "",
"classifier_model": "",
"response_model": "",
})
print(f"\n[分類({result['topic']}): {result['classification_reason']}]")
print(f"[応答モデル: {result['response_model']} + {result['adapter_used']}]")
print(f"\nアシスタント: {result['response']}")
if __name__ == "__main__":
main()
実行結果（アダプタの準備とクエリー）
ClaudeCodeで勉強しながらコードを生成して、それを理解する、というルーチンを最近よくやります。