AI_Model_architecture.py

"""

流程圖

讀取資料 → 分割資料 → 編碼 → 建立 Dataset / DataLoader

↓

建立模型(BERT+LSTM+CNN)

        ↓

        BERT 輸出 [batch, seq_len, 768]

        ↓

        BiLSTM  [batch, seq_len, hidden_dim*2]

        ↓

        CNN 模組 (Conv1D + Dropout + GlobalMaxPooling1D)

        ↓

        Linear 分類器(輸出詐騙機率)

        ↓

訓練模型(Epochs)

↓

評估模型(Accuracy / F1 / Precision / Recall)

↓

儲存模型(.pth)



"""
#引入重要套件Import Library
import os
import torch                            #   PyTorch 主模組               
import torch.nn as nn                   #	神經網路相關的層(例如 LSTM、Linear)        
import pandas as pd
import re

from dotenv import load_dotenv
from sklearn.model_selection import train_test_split
from torch.utils.data import DataLoader, Dataset #	提供 Dataset、DataLoader 類別
from transformers import BertTokenizer # BertTokenizer把文字句子轉換成 BERT 格式的 token ID,例如 [CLS] 今天 天氣 不錯 [SEP] → [101, 1234, 5678, ...]
from sklearn.model_selection import train_test_split
from transformers import BertModel

"""

# ------------------- 載入 .env 環境變數 -------------------

load_dotenv()

base_dir = os.getenv("DATA_DIR", "./data")  # 如果沒設環境變數就預設用 ./data



# ------------------- 使用相對路徑找 CSV -------------------

#,os.path.join(base_dir, "NorANDScamInfo_data1.csv"),os.path.join(base_dir, "ScamInfo_data1.csv"),os.path.join(base_dir, "NormalInfo_data1.csv")

#如有需要訓練複數筆資料可以使用這個方法csv_files = [os.path.join(base_dir, "檔案名稱1.csv"),os.path.join(base_dir, "檔案名稱2.csv")]

#程式碼一至131行



# GPU 記憶體限制(可選)

# os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "max_split_size_mb:16"



#資料前處理

class BertPreprocessor:

    def __init__(self, tokenizer_name="ckiplab/bert-base-chinese", max_len=128):

        self.tokenizer = BertTokenizer.from_pretrained(tokenizer_name)

        self.max_len = max_len



    def load_and_clean(self, filepath):

        #載入 CSV 並清理 message 欄位。

        df = pd.read_csv(filepath)

        df = df.dropna().drop_duplicates().reset_index(drop=True)

        # 文字清理:移除空白、保留中文英數與標點

        df["message"] = df["message"].astype(str)

        df["message"] = df["message"].apply(lambda text: re.sub(r"\s+", "", text))

        df["message"] = df["message"].apply(lambda text: re.sub(r"[^\u4e00-\u9fffA-Za-z0-9。,！？]", "", text))

        return df[["message", "label"]]  # 保留必要欄位



    def encode(self, messages):

        #使用 HuggingFace BERT Tokenizer 將訊息編碼成Bert模型輸入格式。

        return self.tokenizer(

            list(messages),

            return_tensors="pt",

            truncation=True,

            padding="max_length",

            max_length=self.max_len

        )

#自動做資料前處理

def build_bert_inputs(files):

    #將正常與詐騙資料分別指定 label,統一清理、編碼,回傳模型可用的 input tensors 與 labels。

    processor = BertPreprocessor()

    dfs = []

    # 合併正常 + 詐騙檔案清單

    all_files = files



    for filepath in all_files:

        df = processor.load_and_clean(filepath)

        dfs.append(df)

    

    # 合併所有資料。在資料清理過程中dropna():刪除有空值的列,drop_duplicates():刪除重複列,filter()或df[...]做條件過濾,concat():將多個 DataFrame合併

    # 這些操作不會自動重排索引,造成索引亂掉。

    # 合併後統一編號(常見於多筆資料合併)all_df = pd.concat(dfs, 關鍵-->ignore_index=True)

    all_df = pd.concat(dfs, ignore_index=True)

    print(f"✅ 已讀入 {len(all_df)} 筆資料")

    print(all_df["label"].value_counts())

    #製作 train/val 資料集

    train_texts, val_texts, train_labels, val_labels = train_test_split(

    all_df["message"], all_df["label"],

    stratify=all_df["label"],

    test_size=0.2,

    random_state=25,

    shuffle=True

    )

    # 進行 BERT tokenizer 編碼

    train_inputs = processor.encode(train_texts)

    val_inputs = processor.encode(val_texts)



    return train_inputs, train_labels, val_inputs, val_labels, processor





#定義 PyTorch Dataset 類別。ScamDataset 繼承自 torch.utils.data.Dataset

#將 BERT 輸出的 token 與對應標籤封裝成 PyTorch 能使用的格式

class ScamDataset(Dataset):

    def __init__(self, inputs, labels):

        self.input_ids = inputs["input_ids"]                           # input_ids:句子的 token ID;attention_mask:注意力遮罩(0 = padding)

        self.attention_mask = inputs["attention_mask"]                 # token_type_ids:句子的 segment 區分

        self.token_type_ids = inputs["token_type_ids"]                 # torch.tensor(x, dtype=...)將資料(x)轉為Tensor的標準做法。

        self.labels = torch.tensor(labels.values, dtype=torch.float32) # x可以是 list、NumPy array、pandas series...

# dtypefloat32:浮點數(常用於 回歸 或 BCELoss 二分類);long:整數(常用於 多分類 搭配 CrossEntropyLoss)。labels.values → 轉為 NumPy array

    

    def __len__(self):          # 告訴 PyTorch 這個 Dataset 有幾筆資料

        return len(self.labels) # 給 len(dataset) 或 for i in range(len(dataset)) 用的

    

    def __getitem__(self, idx): #每次調用 __getitem__() 回傳一筆 {input_ids, attention_mask, token_type_ids, labels}

        return {                #DataLoader 每次會呼叫這個方法多次來抓一個 batch 的資料

            "input_ids":self.input_ids[idx],

            "attention_mask":self.attention_mask[idx],

            "token_type_ids":self.token_type_ids[idx],

            "labels":self.labels[idx]

        }



# 這樣可以同時處理 scam 和 normal 資料,不用重複寫清理與 token 處理

if __name__ == "__main__":

    csv_files = [os.path.join(base_dir, "NorANDScamInfo_data3k.csv")]

    train_inputs, train_labels, val_inputs, val_labels, processor = build_bert_inputs(csv_files)

    

    train_dataset = ScamDataset(train_inputs, train_labels)

    val_dataset = ScamDataset(val_inputs, val_labels)



    # batch_size每次送進模型的是 8 筆資料(而不是一筆一筆)

    # 每次從 Dataset 中抓一批(batch)資料出來

    train_loader = DataLoader(train_dataset, batch_size=8)

    val_loader = DataLoader(val_dataset, batch_size=8)

"""


"""

class BertLSTM_CNN_Classifier(nn.Module)表示:你定義了一個子類別,

繼承自 PyTorch 的基礎模型類別 nn.Module。



若你在 __init__() 裡沒有呼叫 super().__init__(),

那麼父類別 nn.Module 的初始化邏輯(包含重要功能)就不會被執行,

導致整個模型運作異常或錯誤。

"""

# nn.Module是PyTorch所有神經網路模型的基礎類別,nn.Module 是 PyTorch 所有神經網路模型的基礎類別
class BertLSTM_CNN_Classifier(nn.Module):
    
    def __init__(self, hidden_dim=128, num_layers=1, dropout=0.3):
        
        # super()是Python提供的一個方法,用來呼叫「父類別的版本」的方法。
        # 呼叫:super().__init__()讓父類別(nn.Module)裡面那些功能、屬性都被正確初始化。
        # 沒super().__init__(),這些都不會正確運作,模型會壞掉。
        # super() 就是 Python 提供給「子類別呼叫父類別方法」的方式
        super().__init__()
        
        # 載入中文預訓練的 BERT 模型,輸入為句子token IDs,輸出為每個 token 的向量,大小為 [batch, seq_len, 768]。
        self.bert = BertModel.from_pretrained("ckiplab/bert-base-chinese") # 這是引入hugging face中的tranceformat
        
        # 接收BERT的輸出(768 維向量),進行雙向LSTM(BiLSTM)建模,輸出為 [batch, seq_len, hidden_dim*2],例如 [batch, seq_len, 256]
        """

        LSTM 接收每個token的768維向量(來自 BERT)作為輸入,

        透過每個方向的LSTM壓縮成128維的語意向量。

        由於是雙向LSTM,會同時從左到右(前向)和右到左(後向)各做一次,

        最後將兩個方向的輸出合併為256維向量(128×2)。

        每次處理一個 batch(例如 8 句話),一次走完整個時間序列。

        """
        self.LSTM = nn.LSTM(input_size=768,        
                            hidden_size=hidden_dim,
                            num_layers=num_layers,
                            batch_first=True,
                            bidirectional=True)
        
         # CNN 模組:接在 LSTM 後的輸出上。將LSTM的輸出轉成卷積層格式,適用於Conv1D,CNN可學習位置不變的局部特徵。
        self.conv1 =  nn.Conv1d(in_channels=hidden_dim*2,
                                out_channels=128,
                                kernel_size=3,  # 這裡kernel_size=3 為 3-gram 特徵
                                padding=1)
        
        self.dropout = nn.Dropout(dropout) # 隨機將部分神經元設為 0,用來防止 overfitting。
        
        self.global_maxpool = nn.AdaptiveAvgPool1d(1)  #將一整句話的特徵濃縮成一個固定大小的句子表示向量
        
        # 將CNN輸出的128維特徵向量輸出為一個「機率值」(詐騙或非詐騙)。
        self.classifier = nn.Linear(128,1)
        
    def forward(self, input_ids, attention_mask, token_type_ids):
        #BERT 編碼
        outputs = self.bert(input_ids=input_ids,
                            attention_mask=attention_mask,
                            token_type_ids=token_type_ids)
        #.last_hidden_state是BertModel.from_pretrained(...)內部的key，會輸出    [batch, seq_len, 768]
        hidden_states = outputs.last_hidden_state 

        # 送入 BiLSTM
        # transpose(1, 2) 的用途是：讓 LSTM 輸出的資料形狀符合 CNN 所要求的格式
        #   假設你原本 LSTM 輸出是：    [batch_size, seq_len, hidden_dim*2] = [8, 128, 256]
        # 但CNN(Conv1d)的輸入格式需要是：[batch_size, in_channels, seq_len] = [8, 256, 128]
        # 因此你需要做：.transpose(1, 2)把 seq_len 和 hidden_dim*2 調換
        LSTM_out, _ = self.LSTM(hidden_states)     # [batch, seq_len, hidden_dim*2]
        LSTM_out = LSTM_out.transpose(1, 2)        # [batch, hidden_dim*2, seq_len]

        # 卷積 + Dropout
        x = self.conv1(LSTM_out)                   # [batch, 128, seq_len]
        x = self.dropout(x)
        
        #全局池化
        # .squeeze(dim) 的作用是：把某個「維度大小為 1」的維度刪掉
        # x = self.global_maxpool(x).squeeze(2)   # 輸出是 [batch, 128, 1]
        # 不 .squeeze(2)，你會得到 shape 為 [batch, 128, 1]，不方便後面接 Linear。
        # .squeeze(2)=拿掉第 2 維（數值是 1） → 讓形狀變成 [batch, 128]
        x = self.global_maxpool(x).squeeze(2)      # [batch, 128]

        #分類 & Sigmoid 機率輸出
        logits = self.classifier(x)
        
        #.sigmoid() → 把 logits 轉成 0~1 的機率.squeeze() → 變成一維 [batch] 長度的機率 list
        """例如：

logits = [[0.92], [0.05], [0.88], [0.41], ..., [0.17]]

→ sigmoid → [[0.715], [0.512], ...]

→ squeeze → [0.715, 0.512, ...]

"""
        return torch.sigmoid(logits).squeeze() # 最後輸出是一個值介於 0 ~ 1 之間,代表「為詐騙訊息的機率」。


"""

# 設定 GPU 裝置

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")



# 初始化模型

model = BertLSTM_CNN_Classifier().to(device)

# 定義 optimizer 和損失函數

optimizer = torch.optim.Adam(model.parameters(),lr=2e-5)

criterion = nn.BCELoss()



# 本機訓練迴圈,要訓練再取消註解,否則在線上版本一律處於註解狀態



if __name__ == "__main__": # 只有當我「直接執行這個檔案」時,才執行以下訓練程式(不是被別人 import 使用時)。

    if os.path.exists("model.pth"):

        print("✅ 已找到 model.pth,載入模型跳過訓練")

        model.load_state_dict(torch.load("model.pth", map_location=device))

    else:

        print("🚀 未找到 model.pth,開始訓練模型...")

        num_epochs = 15 # batch_size設定在train_loader和test_loader那

        for epoch in range(num_epochs):

            model.train() # 從nn.Module繼承的方法。將模型設為「訓練模式」,有些層(像 Dropout 或 BatchNorm)會啟用訓練行為。

            total_loss = 0.0

            for batch in train_loader:

            

                # 清理舊梯度,以免累加。為甚麼要?因為PyTorch 預設每次呼叫 .backward() 都會「累加」梯度(不會自動清掉)

                # 沒 .zero_grad(),梯度會越累積越多,模型會亂掉。

                optimizer.zero_grad() 

                

                input_ids = batch["input_ids"].to(device)

                attention_mask = batch["attention_mask"].to(device)

                token_type_ids = batch["token_type_ids"].to(device)

                labels = batch["labels"].to(device)

                outputs = model(input_ids, attention_mask, token_type_ids)

                

                loss = criterion(outputs, labels) # 比較 預測結果 outputs(Sigmoid 的機率)和 真實答案 labels

                

                # 用鏈式法則(Chain Rule)計算每一層「參數對 loss 的影響」,也就是梯度

                # PyTorch 利用自動微分(autograd)幫你計算整個計算圖的偏導數,然後存在每一層的 .grad 裡。

                loss.backward()

                

                # 用 .grad 中的梯度資訊根

                # 據學習率和優化器的規則

                # 改變每一個參數的值,以讓下一次預測更接近真實

                optimizer.step()

                

                # loss 是一個 tensor(需要 backward);.item() 把它轉成 Python 的純數字(float)

                total_loss += loss.item()

            print(f"[Epoch{epoch+1}]Training Loss:{total_loss:.4f}")

        torch.save(model.state_dict(), "model.pth")# 儲存模型權重

        print("✅ 模型訓練完成並儲存為 model.pth")

"""

"""

整個模型中每一個文字(token)始終是一個向量,隨著層數不同,這個向量代表的意義會更高階、更語意、更抽象。

在整個 BERT + LSTM + CNN 模型的流程中,「每一個文字(token)」都會被表示成一個「向量」來進行後續的計算與學習。

今天我輸入一個句子:"早安你好,吃飯沒"

BERT 的輸入包含三個部分:input_ids、attention_mask、token_type_ids,

這些是 BERT 所需的格式。BERT 會將句子中每個 token 編碼為一個 768 維的語意向量,



進入 BERT → 每個 token 變成語意向量:

BERT 輸出每個字為一個 768 維的語意向量

「早」 → [0.23, -0.11, ..., 0.45]   長度為 768

「安」 → [0.05, 0.33, ..., -0.12]   一樣 768

...

batch size 是 8,句子長度是 8,輸出 shape 為:    

[batch_size=8, seq_len=8, hidden_size=768]



接下來這些向量會輸入到 LSTM,LSTM不會改變「一個token是一個向量」的概念,而是重新表示每個token的語境向量。

把每個原本 768 維的 token 壓縮成 hidden_size=128,雙向 LSTM → 拼接 → 每個 token 成為 256 維向量:



input_size=768 是從 BERT 接收的向量維度

hidden_size=128 表示每個方向的 LSTM 會把 token 壓縮為 128 維語意向量

num_layers=1 表示只堆疊 1 層 LSTM

bidirectional=True 表示是雙向



LSTM,除了從左讀到右,也會從右讀到左,兩個方向的輸出會合併(拼接),變成:

[batch_size=8, seq_len=8, hidden_size=256]  # 因為128*2



接下來進入 CNN,CNN 仍然以「一個向量代表一個字」的形式處理:



in_channels=256(因為 LSTM 是雙向輸出)



out_channels=128 表示學習出 128 個濾波器,每個濾波器專門抓一種 n-gram(例如「早安你」),每個「片段」的結果輸出為 128 維特徵



kernel_size=3 表示每個濾波器看 3 個連續 token(像是一個 3-gram)或,把相鄰的 3 個字(各為 256 維)一起掃描



padding=1 為了保留輸出序列長度和輸入相同,避免邊界資訊被捨棄



CNN 輸出的 shape 就會是:



[batch_size=8, out_channels=128, seq_len=8],還是每個 token 有對應一個向量(只是這向量是 CNN 抽出的新特徵)



"""