上一篇聊完了意圖分析的架構設計：為什麼要先讓小模型頂在前面快篩、信心度不夠才升級給 LLM 複審。本篇就是「捲起袖子」的實作篇：示範第一階段Intent-Classifier要怎麼從零生出來，順便大致把「為什麼要這樣設計資料」、「為什麼參數要這樣調」講清楚，也順便記錄一下用 Mac M4 Pro 跑本機微調的實際體感。

為什麼要分兩階段

一句話講完：「讓便宜模型處理大多數簡單案例，把珍貴的 LLM 資源留給真正模糊的問題」。具體來說，流程是這樣走的：

1. 使用者丟一句話進來
2. 先給小模型（就是這篇要練的 Intent-Classifier）判斷，順便吐出一個信心度分數
3. 信心度夠高 → 直接採用小模型的判斷
4. 信心度不夠 → 轉交給 LLM 重新判斷一次
5. 系統把兩邊的結果做「比對統整」，決定最終要用誰的答案
6. 寫入上下文記錄，交給 Agent 去執行

這套設計的重點，其實不在模型多準，而在於「信心度」這個欄位，它是決定要不要升級到 LLM 的關鍵開關。所以待會你會看到，訓練資料裡除了「問題」跟「答案」，還特別多了一個 confidence 欄位。

檔案架構

第一步是先開一個乾淨的虛擬環境：

python -m venv intent-classifier
cd intent-classifier
source bin/activate
pip install -r requirements.txt

requirements.txt 裡列的套件不多，但每一個都有明確分工：

• pandas / scikit-learn：處理資料表格、切分訓練/測試集
• datasets：HuggingFace 的資料集格式
• torch：底層運算引擎，這裡會用到 Apple Silicon 的 MPS 加速
• transformers：載入 Gemma 3 這種預訓練模型的標準工具
• peft：做 LoRA 微調用的（等等會細講）
• trl：Hugging Face 出的訓練框架，裡面的 SFTTrainer 就是這次微調的核心
• huggingface-hub：下載 google/gemma-3-270m-it 來當基礎模型

訓練環境：Mac OS / M4 Pro / 48GB RAM

這次整套流程是在 Mac M4 Pro（48GB 統一記憶體）上跑完的，程式碼裡有處理裝置偵測的邏輯：

if torch.backends.mps.is_available():
    self.device = torch.device("mps")
    logger.info("✅ 使用 Apple Silicon GPU (MPS)")
else:
    self.device = torch.device("cpu")

只要是 Apple Silicon（M 系列晶片），PyTorch 會自動吃 MPS（Metal Performance Shaders）後端，整個過程約需 3.5 至 4 小時跑完整個 FineTune 流程。

程式碼拆解與說明

Step 1：產生訓練資料（1-prepare_training_dataset.py）

這支腳本的任務，就是把一份份「問題 + 答案 + 信心度」的資料，整理成模型可以吃的格式。

資料長什麼樣子

{
    "prompt": "Does P21AL01 suit applications requiring high computing power?",
    "completion": "B",
    "confidence": {"A": 0.15, "B": 0.85, "C": 0.0}
}

三個意圖類別的定義很直白：

• A（詢問產品基本問題 / products_info）：問規格、尺寸、支援什麼作業系統之類的基礎問題

• B（進階產品問題或推薦 / products_summary）：問「適合ＯＯ產業嗎」、「值得買嗎」、「推薦哪款」這種需要判斷、比較的問題

• C（其他跟產品無關的問題）：像是「今天晚餐吃什麼」、「圖書館開放時間」這種完全離題的閒聊

這個分類粒度沒有切成十幾種細碎類別，就抓「查規格 / 要推薦 / 不相關」這三大方向，剛好呼應了「意圖類別先抓大方向，太細會讓小模型誤判機率暴增」這個設計原則。

而 confidence 這個欄位，就是前面提到的關鍵：資料集不只教模型「這題答案是 B」，還教它「B 有 85% 把握、A 只有 15%、C 幾乎不可能」。這樣模型學出來的不是死板的分類器，而是會輸出一個機率分布，後續系統才有依據去判斷「這題我到底該不該送給 LLM 複審」。

整份資料集裡總共準備了超過 1,100 筆手動標註樣本，A、B 兩類各約 500 多筆維持平衡，C 類（無關問題）大概 100 筆左右——這是刻意的，C 類不需要跟 A、B 一樣多，只要足夠讓模型學會「這種問題跟產品完全無關」的模式就好。

IntentDataProcessor：資料整理的工具箱

class IntentDataProcessor:
    def analyze_dataset(self, data):
        """統計各類別分布、算平衡度"""

    def show_samples_by_category(self, data, samples_per_class=5):
        """每類抓幾筆出來看看，肉眼檢查資料合不合理"""

    def create_balanced_dataset(self, data, min_samples_per_class=80):
        """檢查每個類別樣本數是否足夠"""

    def format_for_training(self, examples):
        """組成真正要餵給模型的 prompt 樣板"""

    def prepare_dataset(self, data):
        """用 train_test_split 做 80/20 切分"""

這個類別把資料處理的每個步驟都拆成獨立方法，讀起來像一條流水線：先統計分布、抽樣肉眼檢查、確認每類樣本數足夠、組裝訓練用 prompt，最後切分訓練/測試集。

訓練/測試集切分（stratify）

train_data, test_data = train_test_split(
    formatted_data,
    test_size=0.2,
    stratify=[item['completion'] for item in formatted_data],
    random_state=42
)

值得注意的是 prepare_dataset 切訓練/測試集時用了 stratify 參數——它確保切分後，訓練集跟測試集裡 A/B/C 三類的比例維持一致，不會不小心把某一類的樣本全切到同一邊去，這是做分類任務資料切分的基本功。

最後跑完 save_datasets()，會同時輸出兩種格式：JSON（方便肉眼檢查、版控）跟 HuggingFace Dataset 格式（save_to_disk，方便下一步訓練腳本直接讀取）。實際執行指令很單純：

python 1-prepare_training_dataset.py

跑完會在資料夾裡看到 train_intent_hf_dataset 跟 test_intent_hf_dataset 兩個目錄，這就是餵給下一步訓練用的資料。

小提醒：可以再加一個模板生成的函式，用模板自動生成更多樣本（例如把「{product} 的價格多少？」套上一堆產品型號）來擴充資料量。只是要注意模板生成的樣本比較單調，最好跟手動標註的資料混著用，不然模型容易學到「只認模板句型」的偏差。

完整程式碼

下面 intent_data 只放了幾筆示範資料，正式使用時請換成／擴充成你自己標註的資料：

"""
1-prepare_training_dataset.py
意圖分類訓練資料產生器
"""
import json
import random
import pandas as pd
from datasets import Dataset
from sklearn.model_selection import train_test_split

# === 訓練資料：下面只放示範樣本，正式使用請換成/擴充你自己標註的資料 ===
intent_data = [
    # A：詢問產品基本問題
    {"prompt": "Which operating systems does P21AL01 support?", "completion": "A", "confidence": {"A": 0.92, "B": 0.08, "C": 0.0}},
    {"prompt": "Does SH570R8's memory support DDR5?", "completion": "A", "confidence": {"A": 0.95, "B": 0.05, "C": 0.0}},
    {"prompt": "What is SB860R8's size?", "completion": "A", "confidence": {"A": 0.92, "B": 0.08, "C": 0.0}},
    {"prompt": "Does it support Thunderbolt or USB4?", "completion": "A", "confidence": {"A": 0.9, "B": 0.1, "C": 0.0}},
    # B：進階產品問題或推薦
    {"prompt": "Does P21AL01 suit applications requiring high computing power?", "completion": "B", "confidence": {"A": 0.15, "B": 0.85, "C": 0.0}},
    {"prompt": "Is BPCEL07 worth buying considering budget constraints?", "completion": "B", "confidence": {"A": 0.1, "B": 0.88, "C": 0.02}},
    {"prompt": "Under what circumstances would DH770 be recommended?", "completion": "B", "confidence": {"A": 0.05, "B": 0.9, "C": 0.05}},
    {"prompt": "What type of customers is SPCEL13 suitable for?", "completion": "B", "confidence": {"A": 0.08, "B": 0.87, "C": 0.05}},
    # C：其他跟產品無關的問題
    {"prompt": "Investment strategy", "completion": "C", "confidence": {"A": 0.0, "B": 0.0, "C": 1.0}},
    {"prompt": "Is the air quality good?", "completion": "C", "confidence": {"A": 0.0, "B": 0.0, "C": 1.0}},
    {"prompt": "What should I have for dinner?", "completion": "C", "confidence": {"A": 0.0, "B": 0.0, "C": 1.0}},
    {"prompt": "Where is the pharmacy?", "completion": "C", "confidence": {"A": 0.0, "B": 0.0, "C": 1.0}},
]


class IntentDataProcessor:
    def __init__(self):
        self.intent_labels = ["A", "B", "C"]
        self.intent_names = {
            "A": "詢問產品基本問題",
            "B": "進階產品問題或推薦",
            "C": "其他跟產品無關的問題",
        }

    def format_for_training(self, examples):
        """組成真正要餵給模型的 prompt 樣板"""
        formatted_examples = []
        for example in examples:
            prompt = (
                "請分析以下客戶詢問的意圖，並從 A、B、C 中選擇一個：\n"
                "A：詢問產品基本問題\nB：進階產品問題或推薦\nC：其他跟產品無關的問題\n\n"
                f"客戶詢問：{example['prompt']}\n意圖分類："
            )
            formatted_examples.append({
                "text": prompt + example["completion"],
                "prompt": example["prompt"],
                "completion": example["completion"],
                "confidence": example["confidence"],
            })
        return formatted_examples

    def analyze_dataset(self, data):
        """統計各類別分布"""
        df = pd.DataFrame(data)
        print("=== 意圖分類數據集統計 ===")
        print(f"總樣本數: {len(data)}")
        class_counts = df["completion"].value_counts()
        for intent, count in class_counts.items():
            percentage = (count / len(data)) * 100
            print(f"  {intent}（{self.intent_names[intent]}）: {count} 個樣本（{percentage:.1f}%）")
        return df

    def show_samples_by_category(self, data, samples_per_class=5):
        """每類抓幾筆出來看看，肉眼檢查資料合不合理"""
        df = pd.DataFrame(data)
        for intent in self.intent_labels:
            samples = df[df["completion"] == intent]["prompt"].head(samples_per_class)
            print(f"\n{intent} 類別（{self.intent_names[intent]}）樣本:")
            for i, text in enumerate(samples, 1):
                print(f"  {i}. {text}")

    def create_balanced_dataset(self, data, min_samples_per_class=80):
        """檢查每個類別樣本數是否足夠"""
        df = pd.DataFrame(data)
        balanced_data = []
        for intent in self.intent_labels:
            intent_samples = df[df["completion"] == intent].to_dict("records")
            if len(intent_samples) < min_samples_per_class:
                print(f"⚠️  {intent} 類別只有 {len(intent_samples)} 個樣本，建議增加到 {min_samples_per_class} 個")
            balanced_data.extend(intent_samples)
        return balanced_data

    def prepare_dataset(self, data):
        """做 80/20 切分，並保持三個類別的比例一致"""
        balanced_data = self.create_balanced_dataset(data)
        formatted_data = self.format_for_training(balanced_data)

        train_data, test_data = train_test_split(
            formatted_data,
            test_size=0.2,
            stratify=[item["completion"] for item in formatted_data],
            random_state=42,
        )

        train_dataset = Dataset.from_list(train_data)
        test_dataset = Dataset.from_list(test_data)
        return train_dataset, test_dataset

    def save_datasets(self, data):
        """分析、預覽、切分、輸出成 JSON + HuggingFace Dataset 兩種格式"""
        self.analyze_dataset(data)
        self.show_samples_by_category(data)

        train_dataset, test_dataset = self.prepare_dataset(data)
        print(f"\n訓練集大小: {len(train_dataset)}")
        print(f"測試集大小: {len(test_dataset)}")

        with open("complete_intent_dataset.json", "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        with open("train_intent_dataset.json", "w", encoding="utf-8") as f:
            json.dump(train_dataset.to_list(), f, ensure_ascii=False, indent=2)
        with open("test_intent_dataset.json", "w", encoding="utf-8") as f:
            json.dump(test_dataset.to_list(), f, ensure_ascii=False, indent=2)

        train_dataset.save_to_disk("./train_intent_hf_dataset")
        test_dataset.save_to_disk("./test_intent_hf_dataset")

        print("\n✅ 數據集保存完成: train_intent_hf_dataset / test_intent_hf_dataset")
        return train_dataset, test_dataset


def main():
    processor = IntentDataProcessor()

    all_data = intent_data
    random.shuffle(all_data)
    print(f"📊 總數據量: {len(all_data)} 個樣本")

    processor.save_datasets(all_data)


if __name__ == "__main__":
    main()

Step 2：開始微調模型（2-finetune_model_trainer.py）

這裡是真正的重頭戲，也是整個專案裡最多「防呆機制」的地方，跑起來會花 3.5～4 小時：

python 2-finetune_model_trainer.py

拆開來看，主要做了五件事：載入模型、掛上 LoRA、準備 SFT 格式資料、開始訓練、訓練完轉成 Ollama 能吃的格式。

載入模型

self.model = AutoModelForCausalLM.from_pretrained(
    self.hf_model,
    trust_remote_code=True,
    attn_implementation='eager',   # Gemma3 注意力機制專屬設定
    use_cache=False,               # 關閉 cache 避免數值問題
    torch_dtype="auto",
    device_map="auto"
)

基礎模型是 google/gemma-3-270m-it — 這顆模型小到用 CPU 都能跑得動，更別說有 GPU/MPS 加持了。這正好呼應了第一階段Intent-Classifier的定位：它不需要博學多聞，只需要在一個很窄的任務（三分類意圖判斷）上又快又準。用巨型模型來做這件事，等於殺雞用牛刀，還會拖慢整個系統的反應速度。

載入模型時有兩個細節值得注意：attn_implementation='eager' 是 Gemma 3 系列在非 CUDA 環境（像 MPS）下建議的設定，避免用到某些尚未完全相容的注意力機制實作。use_cache=False 則是為了避免訓練過程中的數值不穩定——這些都是典型的「在 Apple Silicon 上跑 transformer 訓練」會遇到的眉角。

LoRA 設定

lora_config = LoraConfig(
    r=64,
    lora_alpha=128,
    target_modules=[
        "q_proj", "v_proj", "k_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"
    ],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
    init_lora_weights=True
)

微調全部參數對 270M 模型來說其實不算太誇張，但用 LoRA（Low-Rank Adaptation）還是效率更高、更省記憶體的做法——只訓練一小組「插入層」的參數，凍結原本模型的權重。這裡 target_modules 幾乎涵蓋了 Transformer 每一層的關鍵投影矩陣（注意力機制的 q/k/v/o，還有 FFN 的 gate/up/down），算是掛得蠻全面的設定，r=64 也不算小 rank——對一顆 270M 的模型來說，這樣的設定已經足夠讓模型學到不錯的任務特化能力，同時因為基礎模型夠小，就算 rank 開大一點，訓練起來也不會太吃力。

訓練過程中的「照妖鏡」：WeightMonitorCallback

class WeightMonitorCallback(TrainerCallback):
    def on_step_end(self, args, state, control, model=None, **kwargs):
        self.step_count += 1
        if self.step_count % 5 == 0:
            for name, param in model.named_parameters():
                if param.requires_grad:
                    if torch.isnan(param).any():
                        logger.error(f"❌ 步驟 {self.step_count}: NaN在 {name}")
                        control.should_training_stop = True
                    ...

callback 每 5 步就去檢查一次模型權重跟梯度有沒有出現 NaN 或 Inf。為什麼要特別做這個？因為 MPS 後端在某些數值運算上，穩定性確實不像成熟的 CUDA 環境那麼可靠，訓練途中突然冒出 NaN、梯度爆炸這種事並不罕見。與其等訓練跑完 4 小時才發現模型壞掉，不如讓它邊訓練邊自我檢查，一旦異常就直接喊停。

SFT 訓練資料格式

prepare_sft_dataset 把每一筆資料組裝成類似對話的格式，用 Gemma 3 官方定義的 token 包起來：

<start_of_turn>user
請分析以下客戶詢問的意圖，判斷該問題屬於以下哪一類：
A: products_info（產品技術規格詳細資訊查詢）
B: products_summary（產品選擇、推薦、比較相關查詢）
C: 其他（無法歸類於上述兩類的問題）
## 輸出格式
JSON回應：{"classification":"A/B/C","confidence":{"A":0.0-1.0,"B":0.0-1.0,"C":0.0-1.0}}
客戶詢問：{問題}<end_of_turn>
<start_of_turn>model
{"classification":"B","confidence":{"A":0.15,"B":0.85,"C":0.0}}<end_of_turn>

這邊有兩個設計重點：

1. 要求模型輸出結構化 JSON，而不是單純吐一個字母。這樣一來，這顆小模型輸出的結果可以直接被系統解析，不用再另外寫一層規則去抽取答案。

2. 信心度是模型「學」出來的，不是額外用機率統計算出來的。訓練資料裡怎麼標信心度，模型就學著怎麼輸出信心度——這也是為什麼一開始準備資料時，每一筆的 confidence 都要標得合理，如果亂標，模型學出來的信心度分布也會失真，直接影響到後面「信心度門檻」判斷該不該升級給 LLM 的準確性。

訓練參數怎麼調的

training_args = SFTConfig(
    max_length=1024,
    packing=True,
    num_train_epochs=10,
    per_device_train_batch_size=2,
    gradient_accumulation_steps=4,
    optim="adamw_torch_fused",
    learning_rate=3e-4,
    lr_scheduler_type="cosine",
    warmup_steps=100,
    ...
)

幾個值得注意的地方：

• packing=True：把多筆較短的樣本打包進同一個序列裡，減少 padding 浪費，訓練效率會好一些。

• per_device_train_batch_size=2 搭配 gradient_accumulation_steps=4：等效批次大小是 8，但實際單次前向傳播只吃 2 筆——這是在有限記憶體下，兼顧「梯度更新穩定度」跟「資源消耗」的常見折衷手法。

• num_train_epochs=10：這顆基礎模型小、任務也單純（三分類），10 個 epoch 對這種規模的資料集不算誇張，配合 LoRA 通常不容易 overfit 到崩壞的程度。

• lr_scheduler_type="cosine" + warmup_steps=100：先熱身、再餘弦衰減，是穩定訓練的標準搭配。

合併權重，轉成 Ollama 格式

model = PeftModel.from_pretrained(base_model, lora_path)
merged_model = model.merge_and_unload()

訓練結束後，convert_to_ollama() 會把 LoRA 權重跟基礎模型合併回一顆完整模型。合併前後都會再檢查一次有沒有 NaN/Inf（一樣是防呆），確認沒問題後存到 ./merged_shuttle_classifier_model/，同時自動生成一份 Ollama 專用的 Modelfile，裡面已經內建好 prompt 樣板跟停止符號（<end_of_turn>），這樣接下來丟給 Ollama 的時候，不用自己再手刻一次 template。

完整程式碼

"""
2-finetune_model_trainer.py
SFT Trainer 監控訓練腳本
"""
import os
import logging
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainerCallback
from trl import SFTTrainer, SFTConfig
from peft import LoraConfig, get_peft_model, PeftModel
from datasets import load_from_disk

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class WeightMonitorCallback(TrainerCallback):
    """每 5 步檢查一次權重/梯度，出現 NaN、Inf 就停止訓練"""

    def __init__(self):
        self.step_count = 0

    def on_step_end(self, args, state, control, model=None, **kwargs):
        self.step_count += 1
        if self.step_count % 5 != 0:
            return

        problem_found = False
        for name, param in model.named_parameters():
            if not param.requires_grad:
                continue
            if torch.isnan(param).any() or torch.isinf(param).any():
                logger.error(f"❌ 步驟 {self.step_count}: 權重數值異常於 {name}")
                problem_found = True
            if param.grad is not None:
                if torch.isnan(param.grad).any():
                    logger.error(f"❌ 步驟 {self.step_count}: 梯度 NaN 於 {name}")
                    problem_found = True
                grad_norm = param.grad.norm().item()
                if grad_norm > 10.0:
                    logger.warning(f"⚠️  步驟 {self.step_count}: 大梯度於 {name}: {grad_norm:.4f}")

        if problem_found:
            logger.error(f"❌ 步驟 {self.step_count}: 檢測到數值問題，停止訓練")
            control.should_training_stop = True


class IntentModelTrainer:
    def __init__(self):
        if torch.backends.mps.is_available():
            self.device = torch.device("mps")
            logger.info("✅ 使用 Apple Silicon GPU (MPS)")
        else:
            self.device = torch.device("cpu")
            logger.info("⚠️  MPS 不可用，使用 CPU")

        self.hf_model = "google/gemma-3-270m-it"

    def load_model_and_tokenizer(self):
        logger.info(f"📥 載入模型: {self.hf_model}")
        self.tokenizer = AutoTokenizer.from_pretrained(
            self.hf_model, cache_dir="./cache", trust_remote_code=True
        )
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
            self.tokenizer.pad_token_id = self.tokenizer.eos_token_id

        self.model = AutoModelForCausalLM.from_pretrained(
            self.hf_model,
            trust_remote_code=True,
            attn_implementation="eager",   # Gemma3 在非 CUDA 環境建議設定
            use_cache=False,               # 關閉 cache，避免訓練時數值問題
            torch_dtype="auto",
            device_map="auto",
        )
        self.model = self.model.to(self.device)
        self.detailed_weight_check("初始載入")

    def detailed_weight_check(self, stage):
        """檢查權重是否出現 NaN / Inf"""
        for name, param in self.model.named_parameters():
            if torch.isnan(param).any() or torch.isinf(param).any():
                logger.error(f"❌ {stage}: 數值異常於 {name}")
                return False
        logger.info(f"✅ {stage}: 權重檢查通過")
        return True

    def setup_lora(self):
        logger.info("🔧 設定 LoRA...")
        lora_config = LoraConfig(
            r=64,
            lora_alpha=128,
            target_modules=[
                "q_proj", "v_proj", "k_proj", "o_proj",
                "gate_proj", "up_proj", "down_proj",
            ],
            lora_dropout=0.05,
            bias="none",
            task_type="CAUSAL_LM",
            init_lora_weights=True,
        )
        self.model = get_peft_model(self.model, lora_config)
        self.detailed_weight_check("LoRA 初始化後")
        self.model.print_trainable_parameters()

    def prepare_sft_dataset(self, dataset):
        """把每一筆資料組成 <start_of_turn> 對話格式"""

        def format_for_sft(examples):
            texts = []
            for i in range(len(examples["prompt"])):
                text = f"""<start_of_turn>user
請分析以下客戶詢問的意圖，判斷該問題屬於以下哪一類：
A: products_info（產品技術規格詳細資訊查詢）
B: products_summary（產品選擇、推薦、比較相關查詢）
C: 其他（無法歸類於上述兩類的問題）
## 輸出格式
JSON回應：{{"classification":"A/B/C","confidence":{{'A':0.0-1.0,'B':0.0-1.0,'C':0.0-1.0}}}}
客戶詢問：{examples["prompt"][i]}<end_of_turn>
<start_of_turn>model
{{"classification":"{examples["completion"][i]}","confidence":{{"A":{examples["confidence"][i]["A"]},"B":{examples["confidence"][i]["B"]},"C":{examples["confidence"][i]["C"]}}}}}<end_of_turn>"""
                texts.append(text)
            return {"text": texts}

        return dataset.map(format_for_sft, batched=True, remove_columns=dataset.column_names)

    def train(self, dataset_path, test_data_path="", output_dir="./intent_classifier_sft"):
        logger.info("🚀 開始 SFT 監控訓練...")

        dataset = load_from_disk(dataset_path)
        test_dataset = load_from_disk(test_data_path) if test_data_path else None

        train_dataset = self.prepare_sft_dataset(dataset)
        eval_dataset = self.prepare_sft_dataset(test_dataset) if test_dataset else None

        training_args = SFTConfig(
            output_dir=output_dir,
            max_length=1024,                    # 最大序列長度
            packing=True,                       # 打包多筆樣本以提高效率
            num_train_epochs=10,
            per_device_train_batch_size=2,      # 較小 batch，提高梯度更新頻率
            gradient_checkpointing=False,
            gradient_accumulation_steps=4,      # 等效 batch size = 8
            optim="adamw_torch_fused",
            logging_steps=1,
            save_strategy="epoch",
            eval_strategy="epoch",
            learning_rate=3e-4,
            fp16=False,                         # MPS 對混合精度支援不完整，先關閉
            bf16=False,
            lr_scheduler_type="cosine",
            push_to_hub=False,
            report_to=None,
            warmup_steps=100,
            dataset_kwargs={
                "add_special_tokens": False,
                "append_concat_token": True,
            },
            save_safetensors=True,
        )

        trainer = SFTTrainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            processing_class=self.tokenizer,
            callbacks=[WeightMonitorCallback()],
        )

        try:
            trainer.train()
            if self.detailed_weight_check("訓練完成後"):
                logger.info("✅ 訓練成功完成")
                trainer.save_model()
                self.tokenizer.save_pretrained(output_dir)
                return output_dir
            logger.error("❌ 訓練後權重異常")
            return None
        except Exception as e:
            logger.error(f"❌ 訓練失敗: {e}")
            import traceback
            traceback.print_exc()
            return None

    def convert_to_ollama(self, lora_path: str, ollama_model_name: str = "shuttle-classifier"):
        """合併 LoRA 權重，輸出 Ollama 可用的 Modelfile"""
        logger.info("🔄 轉換為 Ollama 格式...")

        base_model = AutoModelForCausalLM.from_pretrained(
            self.hf_model,
            trust_remote_code=True,
            torch_dtype="auto",
            device_map="auto",
            attn_implementation="eager",
            use_cache=False,
        )

        model = PeftModel.from_pretrained(base_model, lora_path)
        merged_model = model.merge_and_unload()

        for name, param in merged_model.named_parameters():
            if torch.isnan(param).any() or torch.isinf(param).any():
                raise ValueError(f"合併後權重異常: {name}")

        merged_path = f"./merged_{ollama_model_name}_model"
        merged_model.save_pretrained(merged_path)
        self.tokenizer.save_pretrained(merged_path)

        modelfile_content = f"""FROM .

PARAMETER temperature 0.1
PARAMETER stop "<end_of_turn>"

TEMPLATE \"\"\"<start_of_turn>user
請分析以下客戶詢問的意圖，判斷該問題屬於以下哪一類：
A: products_info（產品技術規格詳細資訊查詢）
B: products_summary（產品選擇、推薦、比較相關查詢）
C: 其他（無法歸類於上述兩類的問題）
## 輸出格式
JSON回應：{{"classification":"A/B/C","confidence":{{'A':0.0-1.0,'B':0.0-1.0,'C':0.0-1.0}}}}
客戶詢問：{{{{ .Prompt }}}}<end_of_turn>
<start_of_turn>model
\"\"\"

LICENSE \"\"\"
intent-classifier Model based on Google gemma3:270m-it
Fine-tuned for customer intent classification: A(Products Info), B(Products Summary), C(Other)
\"\"\""""

        modelfile_path = os.path.join(merged_path, "Modelfile")
        with open(modelfile_path, "w", encoding="utf-8") as f:
            f.write(modelfile_content)

        logger.info(f"✅ 轉換完成: {merged_path}")
        return modelfile_path


def main():
    trainer = IntentModelTrainer()
    trainer.load_model_and_tokenizer()
    trainer.setup_lora()

    result = trainer.train(
        dataset_path="./train_intent_hf_dataset",
        test_data_path="./test_intent_hf_dataset",
    )

    if result:
        print(f"✅ 訓練成功: {result}")
        trainer.convert_to_ollama(result)
    else:
        print("❌ 訓練失敗，建議檢查資料集是否有異常值，或降低 batch size / learning rate 再試一次")


if __name__ == "__main__":
    main()

Step 3：交給 Ollama，實際跑起來

# 把微調完的模型註冊進 Ollama
ollama create intent-classifier-20260902-01 -f ./merged_shuttle_classifier_model/Modelfile

# 丟一句話進去測試
ollama run intent-classifier-20260902-01 "P21AL01 支援什麼作業系統？"

因為 Modelfile 已經內建了完整的 prompt 樣板，實際呼叫時只要丟原始問句進去，模型就會回傳類似這樣的 JSON：

{"classification":"A","confidence":{"A":0.92,"B":0.08,"C":0.0}}

到這裡，第一階段的Intent-Classifier就算真正落地了！它可以獨立跑在本機，透過本地 Ollama 提供服務，接下來只要加一段信心度 < 門檻就轉發給 LLM的判斷式，就是完整的兩階段意圖分析架構。