# LLM 多店客服系統：從零開始操作流程（完整實務指南） claude ai 3.7版本

本文提供使用 RTX GPU 在 Windows 11 + WSL2 環境下，建置並部署多租戶繁體中文客服 LLM 系統的完整操作流程，適合新手學習參考。文檔包含每個步驟的具體操作指令、原理解釋以及實用建議。

## 目錄

1. [基礎環境建置](#一基礎環境建置)
2. [安裝依賴套件與模型準備](#二安裝依賴套件與模型準備)
3. [撰寫 RAG 多店 API 架構](#三撰寫-rag-多店-api-架構)
4. [模型微調（LoRA）](#四模型微調lora)
5. [部署為 GGUF 格式](#五部署為-gguf-格式int4-壓縮)
6. [多租戶佈署與 Docker 化](#六多租戶佈署與-docker-化)
7. [LINE Webhook 整合](#七line-webhook-整合)
8. [系統監控與優化](#八系統監控與優化)
9. [常見問題與排解](#九常見問題與排解)

---

## 一、基礎環境建置

### 為什麼這一步很重要？

LLM 模型開發通常是在 Linux 環境下進行，但許多使用者習慣 Windows 作業系統。透過 WSL2（Windows Subsystem for Linux）在 Windows 系統中安裝 Ubuntu 環境，讓您可以在熟悉的 Windows 環境中使用 Linux 強大的開發工具。

### 1. 啟用 WSL2 + 安裝 Ubuntu 22.04

在 Windows PowerShell（以系統管理員身份運行）中執行：

wsl --install -d Ubuntu-22.04

執行完後系統會重新啟動並安裝 Ubuntu，安裝後打開「Ubuntu」終端機即可進入 Linux 環境。

### 2. 設定 WSL2 的 GPU 訪問與記憶體配置

確保 WSL2 能正確使用 GPU：

# 在 Windows PowerShell 中更新 WSL
wsl --update

# 在 Ubuntu 終端機中確認 GPU 連接
nvidia-smi

如果 `nvidia-smi` 指令無法執行，您需要安裝 NVIDIA CUDA 驅動。

為 WSL2 配置足夠的系統資源，在 Windows 中建立 `.wslconfig` 檔案：

# 檔案位置: C:\Users\<您的使用者名稱>\.wslconfig
[wsl2]
memory=24GB     # 依據您系統可用記憶體調整
processors=8    # 依據您的 CPU 核心數調整
gpuSupport=true # 啟用 GPU 支援

### 3. 安裝 Conda 與 Python 開發環境

Conda 是 Python 的虛擬環境管理工具，可以隔離開發環境，避免套件版本衝突：

# 在 Ubuntu 終端機中執行
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
bash Miniconda3-latest-Linux-x86_64.sh -b
echo 'export PATH="$HOME/miniconda3/bin:$PATH"' >> ~/.bashrc
source ~/.bashrc

建立專用的虛擬環境：

conda create -n llm python=3.10 -y
conda activate llm

---

## 二、安裝依賴套件與模型準備

### 為什麼要這些套件？

以下套件是操作 LLM 模型、建立向量資料庫、開發 API 介面所必須的核心工具：

* `torch`: 深度學習框架，提供 GPU 加速計算
* `transformers`: Hugging Face 的模型庫，用於載入與使用各類預訓練模型
* `faiss`: Meta 開發的高效向量搜尋引擎，RAG 系統的核心組件
* `fastapi`: 高效能的 Python Web 框架，用於建立 API 服務
* `bitsandbytes`: 提供模型量化功能，支援低精度（int8/4）推論
* `sentence-transformers`: 將文本轉換為向量的工具，RAG 系統的基礎

### 1. 安裝核心套件

# 在已啟用的 llm 環境中安裝
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install transformers sentence-transformers faiss-cpu fastapi uvicorn peft bitsandbytes
pip install accelerate safetensors einops redis line-bot-sdk

### 2. 模型選擇與下載

本系統提供兩種模型方案供選擇：

#### 方案一：使用 Meta 的 LLaMA-3 模型

# 需要訪問權限，請參考 Meta 官方網站申請
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3-8B-Instruct", 
    device_map="auto", 
    load_in_8bit=True
)
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3-8B-Instruct")

#### 方案二：使用台灣本地化模型（推薦）

# 繁體中文最佳化選項，無需申請即可使用
model = AutoModelForCausalLM.from_pretrained(
    "yentinglin/Taiwan-LLM-7B-v2.1-chat", 
    device_map="auto", 
    load_in_8bit=True
)
tokenizer = AutoTokenizer.from_pretrained("yentinglin/Taiwan-LLM-7B-v2.1-chat")

# 或使用 TAIDE 模型
model = AutoModelForCausalLM.from_pretrained(
    "cytseng/TAIDE-7B-chat", 
    device_map="auto", 
    load_in_8bit=True
)
tokenizer = AutoTokenizer.from_pretrained("cytseng/TAIDE-7B-chat")

### 3. 模型測試

快速測試模型是否正常運作：

# 保存為 test_model.py
from transformers import AutoModelForCausalLM, AutoTokenizer

# 載入模型
model_name = "yentinglin/Taiwan-LLM-7B-v2.1-chat"  # 依您選擇的模型調整
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",
    load_in_8bit=True
)

# 測試對話
prompt = "你好，請問你是誰？"
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
outputs = model.generate(
    **inputs,
    max_new_tokens=512,
    temperature=0.7,
    repetition_penalty=1.1
)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)

print("模型回應:", response)

執行測試：

python test_model.py

---

## 三、撰寫 RAG 多店 API 架構

### 什麼是 RAG？

RAG（Retrieval Augmented Generation）結合了檢索與生成能力，讓 LLM 能夠參考特定資料庫中的內容來生成回答。這項技術有效解決了 LLM「幻覺」問題，使模型能夠基於最新、準確的資訊提供回答。

### 1. 建立多店知識庫結構

每間店鋪擁有獨立的知識庫與向量索引：

/stores/
  ├── store_001/
  │   ├── faq.json           # 常見問答資料
  │   └── vector.index       # 向量資料庫（檢索用）
  ├── store_002/
  │   ├── faq.json
  │   └── vector.index
  └── ...

### 2. 建立 FAQ 資料格式

FAQ 檔案格式範例（JSON）：

[
  {
    "id": 1,
    "question": "營業時間是幾點到幾點？",
    "answer": "我們週一至週五營業時間為早上 10 點至晚上 8 點，週六日為早上 11 點至晚上 9 點。"
  },
  {
    "id": 2,
    "question": "是否提供外送服務？",
    "answer": "是的，我們提供外送服務，但僅限方圓 5 公里內，外送費用為 NT$50。"
  }
]

### 3. 建立向量資料庫生成工具

# 保存為 create_vector_db.py
import os
import json
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import argparse

def create_store_vector_db(store_id, model_name="paraphrase-multilingual-MiniLM-L12-v2"):
    """為指定商店創建向量資料庫"""
    store_path = f"./stores/{store_id}"
    faq_path = f"{store_path}/faq.json"
    vector_path = f"{store_path}/vector.index"
    
    # 確保目錄存在
    os.makedirs(store_path, exist_ok=True)
    
    # 檢查 FAQ 檔案是否存在
    if not os.path.exists(faq_path):
        print(f"錯誤: {faq_path} 不存在")
        return False
    
    # 載入 FAQ 資料
    with open(faq_path, "r", encoding="utf-8") as f:
        faq_data = json.load(f)
    
    # 準備問題列表
    questions = [item["question"] for item in faq_data]
    
    # 載入向量模型
    print(f"載入 {model_name} 模型...")
    encoder = SentenceTransformer(model_name)
    
    # 將問題轉換為向量
    print("向量化問題...")
    question_vectors = encoder.encode(questions, convert_to_tensor=True, show_progress_bar=True)
    question_vectors = question_vectors.detach().cpu().numpy()
    
    # 建立 FAISS 索引（使用 L2 距離）
    vector_dim = question_vectors.shape[1]
    index = faiss.IndexFlatL2(vector_dim)
    
    # 將向量加入索引
    index.add(question_vectors.astype(np.float32))
    
    # 保存索引
    print(f"保存向量索引到 {vector_path}")
    faiss.write_index(index, vector_path)
    
    print(f"成功建立商店 {store_id} 的向量資料庫")
    return True

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="建立店家向量資料庫")
    parser.add_argument("--store_id", type=str, required=True, help="店家 ID")
    args = parser.parse_args()
    
    create_store_vector_db(args.store_id)

使用方法：

python create_vector_db.py --store_id store_001

### 4. 實現 FastAPI 多店客服 API

# 保存為 app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
import faiss
import json
import os
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import redis
import uuid
from typing import List, Optional

app = FastAPI(title="多店 LLM 客服系統")

# 載入配置
MODEL_NAME = "yentinglin/Taiwan-LLM-7B-v2.1-chat"  # 或您選擇的其他模型
VECTOR_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"
STORES_PATH = "./stores"
SIMILARITY_THRESHOLD = 30.0  # L2 距離閾值，越小表示越相似

# 建立 Redis 連接
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 載入向量模型（全局共用）
encoder = SentenceTransformer(VECTOR_MODEL)

# 店家資料與模型快取
store_data = {}
model = None
tokenizer = None

# 定義請求和回應模型
class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    source: str
    session_id: str

# 初始化模型（延遲載入）
def load_model():
    global model, tokenizer
    if model is None:
        tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
        model = AutoModelForCausalLM.from_pretrained(
            MODEL_NAME,
            device_map="auto",
            load_in_8bit=True
        )
    return model, tokenizer

# 載入店家資料
def load_store_data(store_id):
    store_path = f"{STORES_PATH}/{store_id}"
    
    if not os.path.exists(store_path):
        raise HTTPException(status_code=404, detail=f"店家 {store_id} 不存在")
    
    # 檢查向量索引
    vector_path = f"{store_path}/vector.index"
    if not os.path.exists(vector_path):
        raise HTTPException(status_code=500, detail=f"店家 {store_id} 的向量索引不存在")
    
    # 載入 FAQ 資料
    faq_path = f"{store_path}/faq.json"
    with open(faq_path, "r", encoding="utf-8") as f:
        faq_data = json.load(f)
    
    # 載入向量索引
    vector_index = faiss.read_index(vector_path)
    
    return {"faq": faq_data, "vector_index": vector_index}

# 保存對話歷史
def save_chat_history(session_id, store_id, role, content):
    key = f"chat:{store_id}:{session_id}"
    message = json.dumps({"role": role, "content": content})
    redis_client.rpush(key, message)
    redis_client.expire(key, 86400)  # 設置 24 小時過期

# 獲取對話歷史
def get_chat_history(session_id, store_id, limit=10):
    key = f"chat:{store_id}:{session_id}"
    history_data = redis_client.lrange(key, -limit, -1)
    
    messages = []
    for item in history_data:
        messages.append(json.loads(item))
    
    return messages

# 模型推論
def generate_response(prompt, history=None):
    model, tokenizer = load_model()
    
    # 構建完整提示
    if history:
        full_prompt = ""
        for msg in history:
            if msg["role"] == "user":
                full_prompt += f"### 用戶: {msg['content']}\n"
            else:
                full_prompt += f"### 助理: {msg['content']}\n"
        
        full_prompt += f"### 用戶: {prompt}\n### 助理:"
    else:
        full_prompt = f"### 用戶: {prompt}\n### 助理:"
    
    # 編碼輸入
    inputs = tokenizer(full_prompt, return_tensors="pt").to(model.device)
    
    # 生成回應
    outputs = model.generate(
        **inputs,
        max_new_tokens=512,
        temperature=0.7,
        top_p=0.9,
        repetition_penalty=1.1,
        do_sample=True
    )
    
    # 解碼輸出
    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    # 提取助理的回應部分
    if "### 助理:" in response:
        response = response.split("### 助理:")[-1].strip()
    
    return response

@app.post("/chat/{store_id}", response_model=ChatResponse)
async def chat(store_id: str, request: ChatRequest):
    # 確保已載入店家資料
    if store_id not in store_data:
        try:
            store_data[store_id] = load_store_data(store_id)
        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))
    
    # 處理會話 ID
    session_id = request.session_id or str(uuid.uuid4())
    
    # 獲取用戶輸入
    user_input = request.message.strip()
    
    # 獲取對話歷史
    history = get_chat_history(session_id, store_id)
    
    # 將用戶輸入向量化
    query_vector = encoder.encode([user_input])[0].astype('float32').reshape(1, -1)
    
    # 搜尋最相似的 FAQ
    store_index = store_data[store_id]["vector_index"]
    distances, indices = store_index.search(query_vector, 3)  # 取前 3 個最相似結果
    
    # 檢查相似度是否超過閾值
    if distances[0][0] < SIMILARITY_THRESHOLD:
        # 找到相似問題
        faq_data = store_data[store_id]["faq"]
        similar_questions = [faq_data[int(idx)] for idx in indices[0] if int(idx) < len(faq_data)]
        
        # 構建提示，加入相關 FAQ 作為上下文
        context = "\n\n".join([
            f"Q: {qa['question']}\nA: {qa['answer']}" for qa in similar_questions
        ])
        
        prompt = f"""你是一位專業的客服助理。請使用以下資訊回答用戶的問題。
        如果問題與提供的資訊無關，請婉轉告知你無法回答，並建議用戶聯繫人工客服。

        參考資訊:
        {context}

        用戶問題: {user_input}

        回答:"""
        
        # 生成回應
        response = generate_response(prompt, history)
        source = "faq"
    else:
        # 找不到相似問題，使用一般模式回答
        prompt = f"""你是一位專業的客服助理。請回答用戶的問題。
        如果你不確定答案，請婉轉告知你無法回答，並建議用戶聯繫人工客服。

        用戶問題: {user_input}

        回答:"""
        
        # 生成回應
        response = generate_response(prompt, history)
        source = "model"
    
    # 保存對話歷史
    save_chat_history(session_id, store_id, "user", user_input)
    save_chat_history(session_id, store_id, "assistant", response)
    
    return {
        "response": response,
        "source": source,
        "session_id": session_id
    }

@app.get("/")
async def root():
    return {"status": "online", "service": "多店 LLM 客服系統"}

# 啟動服務
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

啟動服務：

python app.py

或使用 uvicorn：

uvicorn app:app --host 0.0.0.0 --port 8000 --reload

---

## 四、模型微調（LoRA）

### 為什麼要微調？

即使是強大的預訓練模型，也無法了解您特定業務的細節。通過 LoRA（Low-Rank Adaptation）微調，我們可以使用少量資料高效地調整模型，讓它更好地理解您業務的專業術語、服務細節和回答風格。

### 1. 準備訓練資料

建立訓練資料集（JSON Lines 格式）：

{"instruction": "請問你們的營業時間是？", "output": "我們的營業時間是週一至週五上午 10 點至晚上 8 點，週末及國定假日是上午 11 點至晚上 9 點。"}
{"instruction": "你們有素食選擇嗎？", "output": "是的，我們提供多種素食選擇，包括全素沙拉、蔬菜義大利麵和素食漢堡。您可以在點餐時告知服務人員您的飲食需求。"}
{"instruction": "可以預約今晚 7 點的位子嗎？", "output": "您好，如果是 4 人以下的訂位，今晚 7 點還有空位。請提供您的姓名、電話和人數，我們將為您安排。"}

將這些範例保存為 `training_data.jsonl`。

### 2. 安裝微調所需套件

pip install peft accelerate bitsandbytes

### 3. 建立微調腳本

# 保存為 train_lora.py
import os
import torch
import argparse
from datasets import load_dataset
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling
)
from peft import (
    LoraConfig,
    get_peft_model,
    prepare_model_for_kbit_training,
    TaskType
)

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--model_name", type=str, default="yentinglin/Taiwan-LLM-7B-v2.1-chat")
    parser.add_argument("--dataset_path", type=str, required=True)
    parser.add_argument("--output_dir", type=str, default="./lora_out")
    parser.add_argument("--lora_r", type=int, default=8)
    parser.add_argument("--lora_alpha", type=int, default=16)
    parser.add_argument("--lora_dropout", type=float, default=0.05)
    parser.add_argument("--learning_rate", type=float, default=3e-4)
    parser.add_argument("--batch_size", type=int, default=4)
    parser.add_argument("--epochs", type=int, default=3)
    return parser.parse_args()

def main():
    args = parse_args()
    
    # 確保輸出目錄存在
    os.makedirs(args.output_dir, exist_ok=True)
    
    # 載入模型和 tokenizer
    print(f"載入模型 {args.model_name}...")
    tokenizer = AutoTokenizer.from_pretrained(args.model_name)
    model = AutoModelForCausalLM.from_pretrained(
        args.model_name,
        load_in_8bit=True,
        device_map="auto",
        torch_dtype=torch.float16,
    )
    
    # 確保 tokenizer 有 pad_token
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
    
    # 準備模型進行 LoRA 微調
    model = prepare_model_for_kbit_training(model)
    
    # 定義 LoRA 配置
    lora_config = LoraConfig(
        r=args.lora_r,
        lora_alpha=args.lora_alpha,
        lora_dropout=args.lora_dropout,
        bias="none",
        task_type=TaskType.CAUSAL_LM,
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj"]
    )
    
    # 應用 LoRA 到模型
    model = get_peft_model(model, lora_config)
    
    # 載入數據集
    print(f"載入數據集 {args.dataset_path}...")
    dataset = load_dataset("json", data_files=args.dataset_path)
    
    # 定義數據預處理函數
    def preprocess_function(examples):
        # 為每個樣本創建提示
        prompts = []
        for instruction in examples["instruction"]:
            prompt = f"### 用戶: {instruction}\n### 助理:"
            prompts.append(prompt)
        
        # 目標輸出
        targets = [f"{output}" for output in examples["output"]]
        
        # 編碼提示
        tokenized_prompts = tokenizer(
            prompts,
            padding="max_length",
            truncation=True,
            max_length=512,
            return_tensors="pt"
        )
        
        # 編碼目標
        tokenized_targets = tokenizer(
            targets,
            padding="max_length",
            truncation=True,
            max_length=512,
            return_tensors="pt"
        )
        
        input_ids = []
        attention_mask = []
        labels = []
        
        # 構建訓練樣本
        for i in range(len(prompts)):
            prompt_len = tokenized_prompts["input_ids"][i].ne(tokenizer.pad_token_id).sum().item()
            
            # 合併提示和目標的 input_ids
            sample_input_ids = torch.cat([
                tokenized_prompts["input_ids"][i][:prompt_len],
                tokenized_targets["input_ids"][i]
            ])
            
            # 合併注意力掩碼
            sample_attention_mask = torch.cat([
                tokenized_prompts["attention_mask"][i][:prompt_len],
                tokenized_targets["attention_mask"][i]
            ])
            
            # 創建標籤，提示部分用 -100 標記（不計算損失）
            sample_labels = torch.cat([
                torch.full_like(tokenized_prompts["input_ids"][i][:prompt_len], -100),
                tokenized_targets["input_ids"][i]
            ])
            
            input_ids.append(sample_input_ids)
            attention_mask.append(sample_attention_mask)
            labels.append(sample_labels)
        
        # 構建批次
        max_length = max(len(ids) for ids in input_ids)
        for i in range(len(input_ids)):
            padding_length = max_length - len(input_ids[i])
            if padding_length > 0:
                input_ids[i] = torch.cat([input_ids[i], torch.full((padding_length,), tokenizer.pad_token_id, dtype=torch.long)])
                attention_mask[i] = torch.cat([attention_mask[i], torch.zeros(padding_length, dtype=torch.long)])
                labels[i] = torch.cat([labels[i], torch.full((padding_length,), -100, dtype=torch.long)])
        
        batch = {
            "input_ids": torch.stack(input_ids),
            "attention_mask": torch.stack(attention_mask),
            "labels": torch.stack(labels)
        }
        
        return batch
    
    # 處理數據集
    processed_dataset = dataset["train"].map(
        preprocess_function,
        batched=True,
        remove_columns=dataset["train"].column_names,
    )
    
    # 定義訓練參數
    training_args = TrainingArguments(
        output_dir=args.output_dir,
        learning_rate=args.learning_rate,
        per_device_train_batch_size=args.batch_size,
        gradient_accumulation_steps=4,
        num_train_epochs=args.epochs,
        weight_decay=0.01,
        save_total_limit=3,
        logging_steps=10,
        save_steps=100,
        fp16=True,
    )
    
    # 初始化 Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=processed_dataset,
        data_collator=DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False),
    )
    
    # 開始訓練
    print("開始訓練...")
    trainer.train()
    
    # 保存模型
    print(f"保存模型到 {args.output_dir}")
    model.save_pretrained(args.output_dir)
    tokenizer.save_pretrained(args.output_dir)
    
    print("訓練完成！")

if __name__ == "__main__":
    main()

### 4. 執行微調

python train_lora.py --dataset_path training_data.jsonl --output_dir ./lora_model

### 5. 合併 LoRA 權重（將微調參數與原始模型合併）

# 保存為 merge_lora.py
import os
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel

def merge_lora_with_base_model(base_model_name, lora_model_path, output_dir):
    """將 LoRA 模型合併回基礎模型"""
    # 建立輸出目錄
    os.makedirs(output_dir, exist_ok=True)
    
    # 載入基礎模型
    print(f"載入基礎模型 {base_model_name}...")
    base_model = AutoModelForCausalLM.from_pretrained(
        base_model_name,
        torch_dtype=torch.float16,
        device_map="auto"
    )
    
    # 載入 tokenizer
    tokenizer = AutoTokenizer.from_pretrained(base_model_name)
    
    # 載入 LoRA 模型
    print(f"載入 LoRA 模型 {lora_model_path}...")
    model = PeftModel.from_pretrained(base_model, lora_model_path)
    
    # 合併 LoRA 權重到基礎模型
    print("合併模型...")
    merged_model = model.merge_and_unload()
    
    # 保存合併後的模型
    print(f"保存合併後的模型到 {output_dir}")
    merged_model.save_pretrained(output_dir)
    tokenizer.save_pretrained(output_dir)
    
    print("模型合併完成！")

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser()
    parser.add_argument("--base_model", type=str, default="yentinglin/Taiwan-LLM-7B-v2.1-chat")
    parser.add_argument("--lora_model", type=str, required=True)
    parser.add_argument("--output_dir", type=str, default="./merged_model")
    args = parser.parse_args()
    
    merge_lora_with_base_model(args.base_model, args.lora_model, args.output_dir)

執行合併：

python merge_lora.py --lora_model ./lora_model --output_dir ./merged_model

---

## 五、部署為 GGUF 格式（int4 壓縮）

### 為什麼要壓縮？

GGUF（GPT-Generated Unified Format）是一種針對 LLM 優化的格式，將模型壓縮為 int4 格式能夠顯著降低記憶體需求，讓您在有限的硬體資源上同時運行多個模型。該格式與 llama.cpp 和 Ollama 等輕量級推論引擎完全相容。

### 1. 安裝 llama.cpp

# 克隆 llama.cpp 專案
git clone https://github.com/ggerganov/llama.cpp.git
cd llama.cpp

# 編譯（啟用 CUDA 支援）
make LLAMA_CUBLAS=1

# 返回主目錄
cd ..

### 2. 將 Hugging Face 模型轉換為 GGUF 格式

# 保存為 convert_to_gguf.py
import os
import argparse
from transformers import AutoTokenizer

def convert_to_gguf(model_path, output_path, bits=4):
    """將 Hugging Face 模型轉換為 GGUF 格式"""
    # 確認 tokenizer 存在
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    
    # 構建轉換命令
    convert_cmd = f"python llama.cpp/convert.py {model_path} --outfile {output_path} --outtype q{bits}_K_M"
    
    # 執行轉換
    print(f"執行轉換: {convert_cmd}")
    os.system(convert_cmd)
    
    print(f"轉換完成！模型已保存至: {output_path}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--model_path", type=str, required=True, help="Hugging Face 格式模型路徑")
    parser.add_argument("--output_path", type=str, required=True, help="輸出 GGUF 檔案路徑")
    parser.add_argument("--bits", type=int, default=4, choices=[2, 3, 4, 5, 6, 8], help="量化位元數（2-8）")
    args = parser.parse_args()
    
    convert_to_gguf(args.model_path, args.output_path, args.bits)

執行轉換：

python convert_to_gguf.py --model_path ./merged_model --output_path ./store_model_q4.gguf --bits 4

### 3. 使用 llama.cpp 測試 GGUF 模型

./llama.cpp/main -m ./store_model_q4.gguf -n 512 -p "### 用戶: 請問你們今天有營業嗎？
### 助理:"

### 4. 使用 Ollama 部署模型（可選）

如果您希望使用 Ollama 作為推論引擎，可以按照以下步驟操作：

# 安裝 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 建立自訂模型
ollama create store001 --file ./store_model_q4.gguf

# 執行模型
ollama run store001

---

## 六、多租戶佈署與 Docker 化

### 為什麼要用 Docker？

Docker 容器化技術提供了一致的運行環境，能夠簡化部署流程，提高系統穩定性和可擴展性。透過 Docker，您可以輕鬆管理多個店家的模型實例，進行橫向擴展，並簡化版本控制。

### 1. 建立應用程式 Dockerfile

# 檔案名稱: Dockerfile
FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04

WORKDIR /app

# 安裝基本套件
RUN apt-get update && apt-get install -y \
    python3 python3-pip \
    git wget curl \
    && rm -rf /var/lib/apt/lists/*

# 安裝 Python 依賴
COPY requirements.txt .
RUN pip3 install -r requirements.txt

# 複製應用程式代碼
COPY app/ /app/
COPY models/ /models/
COPY stores/ /stores/

# 設定環境變數
ENV MODEL_PATH="/models/store_model_q4.gguf"
ENV STORES_PATH="/stores"
ENV REDIS_HOST="redis"
ENV REDIS_PORT="6379"

# 暴露API端口
EXPOSE 8000

# 啟動服務
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

### 2. 建立 requirements.txt

# 檔案名稱: requirements.txt
fastapi==0.100.0
uvicorn==0.23.2
transformers==4.33.1
sentence-transformers==2.2.2
faiss-cpu==1.7.4
redis==4.6.0
pydantic==2.1.1
python-multipart==0.0.6
line-bot-sdk==3.1.0
llama-cpp-python==0.2.6

### 3. 建立 Docker Compose 配置

# 檔案名稱: docker-compose.yml
version: '3'

services:
  redis:
    image: redis:7.0
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    restart: always

  nginx:
    image: nginx:1.25
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - store001
      - store002
    restart: always

  store001:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - STORE_ID=store001
      - MODEL_PATH=/models/store001_model_q4.gguf
    volumes:
      - ./models:/models
      - ./stores:/stores
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    restart: always

  store002:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - STORE_ID=store002
      - MODEL_PATH=/models/store002_model_q4.gguf
    volumes:
      - ./models:/models
      - ./stores:/stores
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    restart: always

volumes:
  redis-data:

### 4. 建立 Nginx 設定檔

# 檔案名稱: nginx.conf
server {
    listen 80;
    server_name localhost;

    location /store001 {
        proxy_pass http://store001:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        rewrite ^/store001(.*)$ $1 break;
    }

    location /store002 {
        proxy_pass http://store002:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        rewrite ^/store002(.*)$ $1 break;
    }

    location /webhook {
        proxy_pass http://webhook-service:8001;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

### 5. 啟動 Docker 服務

# 建立並啟動所有服務
docker-compose up -d

# 檢查服務狀態
docker-compose ps

# 查看日誌
docker-compose logs -f

---

## 七、LINE Webhook 整合

### 為什麼整合 LINE？

LINE 是台灣地區使用率極高的通訊平台，整合 LINE 官方帳號能讓您的客服系統直接觸達客戶，提供即時自動化回覆，大幅提升客戶體驗。

### 1. 建立 LINE Bot 服務

# 保存為 line_webhook.py
from fastapi import FastAPI, Request, HTTPException, Depends
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import httpx
import os
import redis
import json
from typing import Dict

app = FastAPI()

# 載入環境變數
CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET")
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost")
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))

# 初始化 LINE Bot API
line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)

# Redis 連接
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)

# 存儲用戶與店家的對應關係
def get_store_id(user_id: str) -> str:
    """根據 LINE 用戶 ID 獲取對應的店家 ID"""
    store_id = redis_client.get(f"user:{user_id}:store")
    if store_id:
        return store_id.decode('utf-8')
    return None

def set_store_id(user_id: str, store_id: str):
    """設置 LINE 用戶對應的店家 ID"""
    redis_client.set(f"user:{user_id}:store", store_id)

# 路由映射表
STORE_ROUTES = {
    "store001": "/store001/chat/store_001",
    "store002": "/store002/chat/store_002",
    # 更多店家...
}

# 獲取會話 ID
def get_session_id(user_id: str) -> str:
    """獲取或建立用戶的會話 ID"""
    session_key = f"user:{user_id}:session"
    session_id = redis_client.get(session_key)
    
    if not session_id:
        # 如果沒有會話 ID，則創建一個新的
        import uuid
        session_id = str(uuid.uuid4())
        redis_client.set(session_key, session_id)
    
    return session_id.decode('utf-8') if isinstance(session_id, bytes) else session_id

@app.post("/webhook/{store_id}")
async def line_webhook(store_id: str, request: Request):
    # 獲取 X-Line-Signature 頭部
    signature = request.headers.get('X-Line-Signature', '')
    
    # 獲取請求內容
    body = await request.body()
    body_text = body.decode('utf-8')
    
    try:
        # 驗證簽名
        handler.handle(body_text, signature)
    except InvalidSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    return {"status": "OK"}

@handler.add(MessageEvent, message=TextMessage)
def handle_text_message(event):
    # 獲取用戶 ID 和文本
    user_id = event.source.user_id
    text = event.message.text
    
    # 確定對應的店家 ID
    store_id = get_store_id(user_id)
    
    # 如果沒有對應的店家，則假設用戶第一次發送訊息是選擇店家
    if not store_id:
        # 檢查用戶是否發送的是店家代碼
        if text in STORE_ROUTES:
            set_store_id(user_id, text)
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"您已連接到 {text} 的客服系統。有什麼可以幫您的嗎？")
            )
            return
        else:
            # 如果不是有效的店家代碼，則提示用戶選擇店家
            store_list = ", ".join(STORE_ROUTES.keys())
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"請先選擇店家，可用的店家代碼: {store_list}")
            )
            return
    
    # 獲取會話 ID
    session_id = get_session_id(user_id)
    
    # 獲取對應店家的 API 路由
    route = STORE_ROUTES.get(store_id)
    if not route:
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="對不起，無法確定您要連接的店家。請聯繫客服人員。")
        )
        return
    
    # 構建 API URL
    api_url = f"{API_BASE_URL}{route}"
    
    # 調用 LLM API
    try:
        with httpx.Client(timeout=30.0) as client:
            response = client.post(
                api_url,
                json={"message": text, "session_id": session_id}
            )
            
            # 檢查響應
            if response.status_code != 200:
                line_bot_api.reply_message(
                    event.reply_token,
                    TextSendMessage(text="對不起，系統暫時無法回應。請稍後再試。")
                )
                return
            
            # 解析回應
            result = response.json()
            reply_text = result.get("response", "對不起，無法取得回應。")
            
            # 回覆用戶
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=reply_text)
            )
    except Exception as e:
        print(f"API 呼叫錯誤: {str(e)}")
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="對不起，系統發生錯誤。請稍後再試。")
        )

@app.get("/")
def read_root():
    return {"status": "LINE Webhook 服務正常運行中"}

# 啟動服務
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

### 2. 建立 LINE Webhook 服務的 Dockerfile

# 檔案名稱: Dockerfile.line
FROM python:3.10-slim

WORKDIR /app

# 安裝依賴
COPY line-requirements.txt .
RUN pip install -r line-requirements.txt

# 複製代碼
COPY line_webhook.py .

# 環境變數
ENV LINE_CHANNEL_ACCESS_TOKEN=""
ENV LINE_CHANNEL_SECRET=""
ENV API_BASE_URL="http://nginx"
ENV REDIS_HOST="redis"
ENV REDIS_PORT="6379"

# 暴露端口
EXPOSE 8001

# 啟動服務
CMD ["uvicorn", "line_webhook:app", "--host", "0.0.0.0", "--port", "8001"]

### 3. 新增至 docker-compose.yml

在 `docker-compose.yml` 中添加 LINE Webhook 服務：

  webhook-service:
    build:
      context: .
      dockerfile: Dockerfile.line
    environment:
      - LINE_CHANNEL_ACCESS_TOKEN=${LINE_TOKEN}
      - LINE_CHANNEL_SECRET=${LINE_SECRET}
      - API_BASE_URL=http://nginx
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - redis
      - nginx
    restart: always

### 4. 建立 LINE 專用的依賴檔案

# 檔案名稱: line-requirements.txt
fastapi==0.100.0
uvicorn==0.23.2
line-bot-sdk==3.1.0
httpx==0.24.1
redis==4.6.0

### 5. 設定 LINE 開發者帳號

1. 在 [LINE Developers](https://developers.line.biz/) 註冊並創建一個 Provider
2. 建立一個 Messaging API Channel
3. 獲取 Channel Access Token 和 Channel Secret
4. 設定 Webhook URL 為您的服務 URL（例如 `https://your-domain.com/webhook`）
5. 啟用 Webhook 並關閉自動回覆功能

---

## 八、系統監控與優化

### 1. 監控指標設置

建立一個基本的監控系統，追蹤關鍵性能指標：

# 保存為 monitoring.py
from fastapi import FastAPI, Depends, Request
from prometheus_client import Counter, Histogram, generate_latest
from prometheus_client.exposition import CONTENT_TYPE_LATEST
import time
from starlette.responses import Response
import redis

app = FastAPI()

# 建立 Prometheus 指標
REQUEST_COUNT = Counter('api_requests_total', 'Total API requests', ['store_id', 'endpoint'])
RESPONSE_TIME = Histogram('api_response_time_seconds', 'API response time in seconds', ['store_id', 'endpoint'])
QUERY_SUCCESS = Counter('query_success_total', 'Successful queries', ['store_id'])
QUERY_FAILURE = Counter('query_failure_total', 'Failed queries', ['store_id'])

# Redis 連接
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.middleware("http")
async def monitor_requests(request: Request, call_next):
    # 記錄請求開始時間
    start_time = time.time()
    
    # 提取路徑和店家 ID
    path = request.url.path
    store_id = "unknown"
    
    # 嘗試從路徑中提取店家 ID
    parts = path.split("/")
    if len(parts) > 1 and parts[1].startswith("store"):
        store_id = parts[1]
    
    # 增加請求計數
    REQUEST_COUNT.labels(store_id=store_id, endpoint=path).inc()
    
    # 處理請求
    try:
        response = await call_next(request)
        
        # 如果是 API 回應，記錄響應時間
        if path.endswith("/chat"):
            RESPONSE_TIME.labels(store_id=store_id, endpoint=path).observe(time.time() - start_time)
            
            # 根據狀態碼記錄成功或失敗
            if response.status_code == 200:
                QUERY_SUCCESS.labels(store_id=store_id).inc()
            else:
                QUERY_FAILURE.labels(store_id=store_id).inc()
                
        return response
    except Exception as e:
        QUERY_FAILURE.labels(store_id=store_id).inc()
        raise e

@app.get("/metrics")
def metrics():
    """導出 Prometheus 指標"""
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.get("/health")
def health_check():
    """健康檢查端點"""
    try:
        # 檢查 Redis 連接
        redis_client.ping()
        return {"status": "healthy"}
    except Exception as e:
        return Response(
            status_code=500,
            content=f"Service unhealthy: {str(e)}",
            media_type="text/plain"
        )

### 2. Grafana 視覺化儀表板設置 (docker-compose.yml 新增)

  prometheus:
    image: prom/prometheus:v2.45.0
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    restart: always

  grafana:
    image: grafana/grafana:9.5.2
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=secret
    volumes:
      - grafana-data:/var/lib/grafana
    depends_on:
      - prometheus
    restart: always

volumes:
  prometheus-data:
  grafana-data:

### 3. 系統優化建議

#### 記憶體用量優化

修改 `app.py` 添加記憶體管理功能：

# 添加到 app.py
import gc
import torch
from fastapi import BackgroundTasks

# 定義清理函數
def cleanup_gpu_memory():
    """在背景清理 GPU 記憶體"""
    torch.cuda.empty_cache()
    gc.collect()

@app.post("/chat/{store_id}", response_model=ChatResponse)
async def chat(store_id: str, request: ChatRequest, background_tasks: BackgroundTasks):
    # 現有代碼...
    
    # 在回應後安排清理任務
    background_tasks.add_task(cleanup_gpu_memory)
    
    return {
        "response": response,
        "source": source,
        "session_id": session_id
    }

#### 模型快取策略

# 模型快取管理類
class ModelCache:
    def __init__(self, max_models=3):
        self.models = {}  # 存儲模型的字典
        self.usage_count = {}  # 追蹤模型使用次數
        self.max_models = max_models  # 最大模型數量
    
    def get_model(self, model_path):
        """獲取模型，如果不存在則載入"""
        if model_path in self.models:
            self.usage_count[model_path] += 1
            return self.models[model_path]
        
        # 如果達到最大模型數，則移除最少使用的
        if len(self.models) >= self.max_models:
            least_used = min(self.usage_count.items(), key=lambda x: x[1])[0]
            del self.models[least_used]
            del self.usage_count[least_used]
        
        # 載入新模型
        from llama_cpp import Llama
        model = Llama(
            model_path=model_path,
            n_ctx=2048,  # 上下文長度
            n_threads=4  # 使用的線程數
        )
        
        self.models[model_path] = model
        self.usage_count[model_path] = 1
        return model

# 初始化模型快取
model_cache = ModelCache()

---

## 九、常見問題與排解

### 1. 顯存不足錯誤

**問題:** CUDA out of memory

**解決方案:**
- 降低模型精度（使用 4bit 或 3bit 量化）
- 減少批次大小（batch size）
- 減少上下文長度（context length）
- 檢查 GPU 是否被其他程序佔用

### 2. 回應速度慢

**問題:** 模型回應時間過長

**解決方案:**
- 使用更高效的推論引擎（如 vLLM、TensorRT-LLM）
- 減少生成的最大 token 數
- 優化向量檢索算法
- 提前載入熱門模型到記憶體
- 使用更小的模型（例如 3B 或 7B 參數模型）

### 3. 向量檢索不精確

**問題:** RAG 系統無法找到相關問題

**解決方案:**
- 調整相似度閾值
- 使用更好的向量模型（如 BERT 多語言模型）
- 增加 FAQ 問題的變體
- 實施混合檢索（結合關鍵詞和向量檢索）

### 4. Docker 容器啟動失敗

**問題:** 容器無法正確啟動或崩潰

**解決方案:**
- 檢查 GPU 驅動和 CUDA 版本
- 確保 `nvidia-container-toolkit` 正確安裝
- 檢查容器日誌尋找錯誤訊息
- 確保配置了足夠的系統資源

### 5. LINE Webhook 未收到回覆

**問題:** LINE 機器人沒有回應

**解決方案:**
- 檢查 Channel Access Token 和 Channel Secret
- 確認 Webhook URL 設定正確
- 檢查 SSL 證書有效性
- 查看 Webhook 服務日誌尋找錯誤

### 6. 系統擴展性問題

**問題:** 系統無法處理高峰期負載

**解決方案:**
- 實施負載平衡（Load Balancing）
- 使用 Kubernetes 自動擴縮容
- 實現請求隊列機制
- 使用快取減少重複請求的處理

---

## 總結

本文檔提供了從零開始建置多租戶繁體中文客服 LLM 系統的完整操作流程，涵蓋了環境設置、模型訓練、API 開發和系統部署的方方面面。透過遵循這些步驟，您可以建立一個高效、穩定且可擴展的智能客服系統，有效提升客戶體驗及運營效率。

隨著您的系統不斷成長，請記得定期更新模型、擴充知識庫，並持續改進系統架構以適應業務需求的變化。

祝您實施順利！