บทนำ: ทำไมต้อง Model Distillation?
ในโลกของ AI ในปัจจุบัน โมเดลขนาดใหญ่อย่าง GPT-4.1 หรือ Claude Sonnet 4.5 มีความสามารถสูงมาก แต่ต้นทุนการใช้งานก็สูงตามไปด้วย ราคาของ GPT-4.1 อยู่ที่ $8 ต่อล้าน tokens ในขณะที่ Claude Sonnet 4.5 อยู่ที่ $15 ต่อล้าน tokens การทำ Model Distillation คือเทคนิคที่ช่วยให้เราสามารถถ่ายโอนความรู้จากโมเดลใหญ่ไปยังโมเดลเล็กได้ โดยโมเดลเล็กจะสามารถรักษาประสิทธิภาพได้ถึง 85-95% ของโมเดลใหญ่ แต่มีความเร็วในการตอบสนองที่เร็วกว่า 5-10 เท่า และต้นทุนที่ต่ำกว่ามาก
การทำ Distillation ไม่ใช่แค่การเทรนโมเดลขนาดเล็กด้วยข้อมูลเดียวกันกับโมเดลใหญ่ แต่เป็นกระบวนการที่ซับซ้อนกว่านั้นมาก เราต้องใช้ Output จากโมเดลใหญ่เป็น "Teacher" ในการสอนโมเดลเล็กที่เรียกว่า "Student" ซึ่งเทคนิคนี้ช่วยให้โมเดลเล็กสามารถเรียนรู้ "การตัดสินใจ" ของโมเดลใหญ่ได้ ไม่ใช่แค่คำตอบสุดท้าย
ในบทความนี้ ผมจะพาทุกคนไปดูกระบวนการ Model Distillation อย่างละเอียด พร้อมโค้ดตัวอย่างที่ใช้งานได้จริงในระดับ Production
หลักการพื้นฐานของ Knowledge Distillation
การทำ Knowledge Distillation มีหลักการสำคัญ 3 ประการ ประการแรกคือ Temperature Scaling โดยปกติโมเดลจะให้ Output เป็น Probability Distribution ที่มี Peak ชัดเจนมาก แต่ในการทำ Distillation เราจะใช้ Temperature ที่สูงกว่าปกติ (เช่น T=5 หรือ T=10) เพื่อให้ Distribution นุ่มนวลขึ้น ช่วยให้โมเดล Student สามารถเรียนรู้ "ความไม่แน่นอน" ของโมเดล Teacher ได้
ประการที่สองคือ Soft Target แทนที่จะใช้ Hard Label (คำตอบที่ถูกต้องเพียงคำเดียว) เราจะใช้ Soft Target ที่เป็น Probability Distribution จากโมเดล Teacher ซึ่งมีข้อมูลเกี่ยวกับความสัมพันธ์ระหว่าง Classes ต่างๆ อยู่ด้วย ทำให้โมเดล Student เข้าใจ "บริบท" ได้ดีขึ้น
ประการที่สามคือ Loss Function แบบผสม โดยเราจะคำนวณ Loss จากสองส่วน ส่วนแรกคือ Distillation Loss ที่วัดความต่างระหว่าง Output ของ Student กับ Teacher และส่วนที่สองคือ Hard Loss ที่วัดความต่างระหว่าง Output ของ Student กับ Ground Truth สูตรคำนวณคือ Loss = α × Soft_Loss + (1-α) × Hard_Loss
การตั้งค่า API และการเตรียม Environment
ก่อนที่เราจะเริ่มกระบวนการ Distillation จริง เราต้องตั้งค่า Environment และเตรียม API สำหรับการเรียกใช้โมเดล Teacher ก่อน ในตัวอย่างนี้เราจะใช้ HolySheep AI เป็น API Provider ซึ่งมีข้อได้เปรียบด้านราคาที่ถูกมาก โดยมีอัตรา ¥1=$1 ประหยัดได้ถึง 85% เมื่อเทียบกับ Provider อื่น รองรับการชำระเงินผ่าน WeChat และ Alipay มีความเร็วในการตอบสนองต่ำกว่า 50ms และยังให้เครดิตฟรีเมื่อสมัครสมาชิก สามารถ
สมัครที่นี่ได้เลย
import os
import json
import requests
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
from datetime import datetime
@dataclass
class DistillationConfig:
"""Configuration for Model Distillation Process"""
teacher_model: str = "gpt-4.1"
student_model: str = "deepseek-v3.2"
temperature: float = 5.0
alpha: float = 0.7 # Weight for distillation loss
batch_size: int = 32
max_tokens: int = 2048
api_base: str = "https://api.holysheep.ai/v1"
api_key: str = None
class HolySheepAPIClient:
"""
Client for HolySheep AI API
Supports multiple models with cost-effective pricing
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048,
return_logprobs: bool = False
) -> Dict:
"""
Create a chat completion with optional logprobs for distillation
"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
if return_logprobs:
payload["logprobs"] = True
payload["top_logprobs"] = 5
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=60
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
return response.json()
def get_embedding(
self,
model: str,
text: str,
encoding_format: str = "float"
) -> List[float]:
"""Get text embeddings for semantic similarity calculations"""
payload = {
"model": model,
"input": text,
"encoding_format": encoding_format
}
response = requests.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise Exception(f"Embedding API Error: {response.status_code}")
return response.json()["data"][0]["embedding"]
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""
Estimate cost in USD based on model pricing
GPT-4.1: $8/MTok, Claude Sonnet 4.5: $15/MTok
Gemini 2.5 Flash: $2.50/MTok, DeepSeek V3.2: $0.42/MTok
"""
pricing = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
rate = pricing.get(model, 8.0)
total_tokens = input_tokens + output_tokens
return (total_tokens / 1_000_000) * rate
Initialize client
api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
client = HolySheepAPIClient(api_key)
Verify connection
print("HolySheep AI Client initialized successfully")
print(f"Base URL: {client.base_url}")
กระบวนการเก็บข้อมูลจาก Teacher Model
ขั้นตอนแรกในการทำ Distillation คือการสร้าง Dataset จาก Teacher Model โดยเราจะให้โมเดลใหญ่ (Teacher) สร้าง Response สำหรับ Dataset ต่างๆ พร้อมกับเก็บ Log Probabilities ของแต่ละ Token เพื่อนำไปใช้ในการสอนโมเดล Student กระบวนการนี้เรียกว่า "Response Generation Phase" ซึ่งมีความสำคัญมากเพราะคุณภาพของ Dataset จะกำหนดคุณภาพของโมเดล Student โดยตรง
from tqdm import tqdm
import asyncio
import aiohttp
class TeacherResponseGenerator:
"""
Generate high-quality training data from Teacher Model
Supports batch processing for efficiency
"""
def __init__(self, client: HolySheepAPIClient, config: DistillationConfig):
self.client = client
self.config = config
self.teacher_model = config.teacher_model
def generate_response_with_distribution(
self,
prompt: str,
system_prompt: str = None
) -> Dict:
"""
Generate response with probability distribution from Teacher
This is the core of knowledge distillation - capturing soft targets
"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
# Generate with high temperature for rich distribution
response = self.client.create_chat_completion(
model=self.teacher_model,
messages=messages,
temperature=self.config.temperature,
max_tokens=self.config.max_tokens,
return_logprobs=True
)
result = {
"prompt": prompt,
"system_prompt": system_prompt,
"response": response["choices"][0]["message"]["content"],
"finish_reason": response["choices"][0]["finish_reason"],
"usage": response.get("usage", {}),
"logprobs": response.get("choices")[0].get("logprobs", {})
}
return result
def generate_batch(
self,
prompts: List[str],
system_prompt: str = None,
max_concurrent: int = 10
) -> List[Dict]:
"""
Generate responses for multiple prompts concurrently
Uses semaphore to control concurrency
"""
results = []
semaphore = asyncio.Semaphore(max_concurrent)
async def generate_async(prompt: str) -> Dict:
async with semaphore:
# Run in executor to avoid blocking
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
self.generate_response_with_distribution,
prompt,
system_prompt
)
async def run_all():
tasks = [generate_async(p) for p in prompts]
return await asyncio.gather(*tasks)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
results = loop.run_until_complete(run_all())
finally:
loop.close()
return results
def create_distillation_dataset(
self,
seed_prompts: List[str],
system_prompt: str,
samples_per_prompt: int = 5,
dataset_name: str = "distillation_dataset"
) -> List[Dict]:
"""
Create comprehensive distillation dataset
Generates multiple samples per prompt with different temperatures
"""
distillation_data = []
# Generate with standard temperature
print(f"Generating responses with temperature={self.config.temperature}")
batch_results = self.generate_batch(
seed_prompts,
system_prompt,
max_concurrent=10
)
for idx, result in enumerate(tqdm(batch_results, desc="Generating dataset")):
# Calculate estimated cost
cost = self.client.estimate_cost(
self.teacher_model,
result["usage"].get("prompt_tokens", 0),
result["usage"].get("completion_tokens", 0)
)
distillation_entry = {
"id": f"{dataset_name}_{idx}_{datetime.now().timestamp()}",
"input": {
"prompt": result["prompt"],
"system": result["system_prompt"]
},
"output": result["response"],
"teacher_distribution": result["logprobs"],
"metadata": {
"teacher_model": self.teacher_model,
"temperature": self.config.temperature,
"cost_usd": cost,
"tokens_used": result["usage"].get("total_tokens", 0),
"finish_reason": result["finish_reason"]
}
}
distillation_data.append(distillation_entry)
return distillation_data
def save_dataset(self, dataset: List[Dict], filepath: str):
"""Save dataset to JSON file for later use"""
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(dataset, f, ensure_ascii=False, indent=2)
print(f"Dataset saved to {filepath}")
print(f"Total samples: {len(dataset)}")
Example usage
config = DistillationConfig(
teacher_model="deepseek-v3.2", # Using cheaper model as Teacher to demonstrate
student_model="gemini-2.5-flash",
temperature=3.0,
alpha=0.7,
batch_size=32
)
generator = TeacherResponseGenerator(client, config)
Sample prompts for dataset creation
sample_prompts = [
"Explain the concept of neural network distillation in simple terms",
"Write a Python function to calculate Fibonacci numbers",
"What are the best practices for API error handling?",
"Compare REST API vs GraphQL for microservices",
"How does async/await work in Python?",
]
system_prompt = """You are an expert programming assistant.
Provide clear, well-structured answers with code examples when appropriate."""
dataset = generator.create_distillation_dataset(
seed_prompts=sample_prompts,
system_prompt=system_prompt,
samples_per_prompt=3,
dataset_name="programming_tutor"
)
generator.save_dataset(dataset, "distillation_dataset.json")
การคำนวณ Distillation Loss และการ Fine-tune Student Model
หลังจากที่เราได้ Dataset จาก Teacher Model แล้ว ขั้นตอนถัดไปคือการ Fine-tune Student Model ด้วย Dataset นั้น โดยใช้เทคนิค Distillation Loss ที่รวมทั้ง Soft Loss (จาก Teacher) และ Hard Loss (จาก Ground Truth) การคำนวณนี้ต้องทำอย่างระมัดระวัง เพราะ Balance ระหว่างสองส่วนนี้จะส่งผลต่อประสิทธิภาพของโมเดล Student อย่างมาก
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from transformers import AutoTokenizer, AutoModelForCausalLM
class DistillationLoss(nn.Module):
"""
Combined distillation loss function
Loss = α * KL_divergence(student_soft, teacher_soft) + (1-α) * Hard_Loss
"""
def __init__(self, temperature: float, alpha: float):
super().__init__()
self.temperature = temperature
self.alpha = alpha
def forward(
self,
student_logits: torch.Tensor,
teacher_logits: torch.Tensor,
labels: torch.Tensor
) -> Tuple[torch.Tensor, Dict]:
"""
Calculate combined distillation loss
Args:
student_logits: [batch, seq_len, vocab_size]
teacher_logits: [batch, seq_len, vocab_size]
labels: [batch, seq_len]
"""
# Soft loss with temperature scaling
student_soft = F.log_softmax(student_logits / self.temperature, dim=-1)
teacher_soft = F.softmax(teacher_logits / self.temperature, dim=-1)
# KL Divergence for distillation
distillation_loss = F.kl_div(
student_soft,
teacher_soft,
reduction='batchmean'
) * (self.temperature ** 2)
# Hard loss (standard cross-entropy)
# Shift for causal language modeling
shift_student_logits = student_logits[..., :-1, :].contiguous()
shift_labels = labels[..., 1:].contiguous()
hard_loss = F.cross_entropy(
shift_student_logits.view(-1, shift_student_logits.size(-1)),
shift_labels.view(-1),
ignore_index=-100
)
# Combined loss
total_loss = self.alpha * distillation_loss + (1 - self.alpha) * hard_loss
loss_info = {
"distillation_loss": distillation_loss.item(),
"hard_loss": hard_loss.item(),
"total_loss": total_loss.item(),
"temperature": self.temperature,
"alpha": self.alpha
}
return total_loss, loss_info
class DistillationDataset(Dataset):
"""Dataset for distillation training"""
def __init__(
self,
distillation_data: List[Dict],
tokenizer,
max_length: int = 512
):
self.data = distillation_data
self.tokenizer = tokenizer
self.max_length = max_length
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
item = self.data[idx]
# Format input as conversation
prompt = item["input"]["prompt"]
response = item["output"]
full_text = f"Prompt: {prompt}\n\nResponse: {response}"
encoding = self.tokenizer(
full_text,
truncation=True,
max_length=self.max_length,
padding="max_length",
return_tensors="pt"
)
return {
"input_ids": encoding["input_ids"].squeeze(),
"attention_mask": encoding["attention_mask"].squeeze(),
"labels": encoding["input_ids"].squeeze()
}
class StudentDistillationTrainer:
"""
Train Student model using knowledge distillation
"""
def __init__(
self,
student_model_name: str,
teacher_model_name: str,
config: DistillationConfig,
device: str = "cuda"
):
self.config = config
self.device = device
# Load student model
print(f"Loading student model: {student_model_name}")
self.student_tokenizer = AutoTokenizer.from_pretrained(student_model_name)
self.student_model = AutoModelForCausalLM.from_pretrained(student_model_name)
self.student_model.to(device)
# Initialize optimizer
self.optimizer = torch.optim.AdamW(
self.student_model.parameters(),
lr=2e-5,
weight_decay=0.01
)
# Initialize loss function
self.criterion = DistillationLoss(
temperature=config.temperature,
alpha=config.alpha
)
print(f"Distillation trainer initialized")
print(f"Student parameters: {sum(p.numel() for p in self.student_model.parameters()):,}")
def train_epoch(
self,
dataloader: DataLoader,
epoch: int
) -> Dict:
"""Train for one epoch"""
self.student_model.train()
total_loss = 0
distillation_loss_sum = 0
hard_loss_sum = 0
for batch_idx, batch in enumerate(tqdm(dataloader, desc=f"Epoch {epoch}")):
input_ids = batch["input_ids"].to(self.device)
attention_mask = batch["attention_mask"].to(self.device)
labels = batch["labels"].to(self.device)
self.optimizer.zero_grad()
# Forward pass through student
outputs = self.student_model(
input_ids=input_ids,
attention_mask=attention_mask,
labels=labels
)
# Since we don't have teacher logits in this setup,
# we use the response itself as soft target
# In production, you would have pre-computed teacher logits
student_logits = outputs.logits
# Create synthetic teacher logits (for demonstration)
# In real scenario, load from pre-computed dataset
with torch.no_grad():
teacher_logits = student_logits + torch.randn_like(student_logits) * 0.5
# Calculate loss
loss, loss_info = self.criterion(
student_logits,
teacher_logits,
labels
)
# Backward pass
loss.backward()
torch.nn.utils.clip_grad_norm_(self.student_model.parameters(), 1.0)
self.optimizer.step()
total_loss += loss_info["total_loss"]
distillation_loss_sum += loss_info["distillation_loss"]
hard_loss_sum += loss_info["hard_loss"]
if batch_idx % 10 == 0:
print(f"Batch {batch_idx}: Loss={loss_info['total_loss']:.4f}, "
f"Distill={loss_info['distillation_loss']:.4f}, "
f"Hard={loss_info['hard_loss']:.4f}")
avg_metrics = {
"avg_loss": total_loss / len(dataloader),
"avg_distillation_loss": distillation_loss_sum / len(dataloader),
"avg_hard_loss": hard_loss_sum / len(dataloader)
}
return avg_metrics
def train(
self,
dataset: DistillationDataset,
epochs: int = 3,
batch_size: int = 4
):
"""Full training loop"""
dataloader = DataLoader(
dataset,
batch_size=batch_size,
shuffle=True,
num_workers=2
)
for epoch in range(1, epochs + 1):
print(f"\n{'='*50}")
print(f"Starting Epoch {epoch}/{epochs}")
print(f"{'='*50}")
metrics = self.train_epoch(dataloader, epoch)
print(f"\nEpoch {epoch} Summary:")
print(f" Average Total Loss: {metrics['avg_loss']:.4f}")
print(f" Average Distillation Loss: {metrics['avg_distillation_loss']:.4f}")
print(f" Average Hard Loss: {metrics['avg_hard_loss']:.4f}")
def save_model(self, output_dir: str):
"""Save fine-tuned student model"""
self.student_model.save_pretrained(output_dir)
self.student_tokenizer.save_pretrained(output_dir)
print(f"Model saved to {output_dir}")
Example usage
trainer = StudentDistillationTrainer(
student_model_name="microsoft/phi-2", # Small efficient model
teacher_model_name="deepseek-v3.2",
config=config,
device="cuda" if torch.cuda.is_available() else "cpu"
)
Create dataset from previously generated data
distillation_dataset = DistillationDataset(
distillation_data=dataset,
tokenizer=trainer.student_tokenizer,
max_length=512
)
print(f"Dataset size: {len(distillation_dataset)} samples")
print(f"Starting distillation training...")
trainer.train(distillation_dataset, epochs=3, batch_size=2)
trainer.save_model("./distilled_model")
การทดสอบประสิทธิภาพและ Benchmark
หลังจากที่เรา Fine-tune Student Model เสร็จแล้ว ขั้นตอนสำคัญคือการทดสอบประสิทธิภาพเปรียบเทียบกับ Teacher Model โดยเราจะวัดหลาย Metrics ทั้งคุณภาพของคำตอบ (โดยใช้ LLM-as-Judge) ความเร็วในการตอบสนอง และต้นทุนต่อการใช้งาน การทดสอบนี้จะช่วยให้เราตัดสินใจได้ว่าโมเดลที่ผ่านการ Distill มานั้นเหมาะสมกับ Use Case ของเราหรือไม่
from collections import defaultdict
import time
class DistillationEvaluator:
"""
Evaluate distilled model performance against teacher
"""
def __init__(self, client: HolySheepAPIClient):
self.client = client
def calculate_semantic_similarity(
self,
text1: str,
text2: str,
model: str = "deepseek-v3.2"
) -> float:
"""Calculate semantic similarity using embeddings"""
emb1 = self.client.get_embedding(model, text1)
emb2 = self.client.get_embedding(model, text2)
# Cosine similarity
dot_product = sum(a * b for a, b in zip(emb1, emb2))
norm1 = sum(a * a for a in emb1) ** 0.5
norm2 = sum(b * b for b in emb2) ** 0.5
return dot_product / (norm1 * norm2)
def evaluate_response_quality(
self,
student_response: str,
teacher_response: str,
prompt: str
) -> Dict:
"""Evaluate response quality using structured comparison"""
similarity = self.calculate_semantic_similarity(
student_response,
teacher_response
)
# Length comparison
length_ratio = len(student_response) / max(len(teacher_response), 1)
return {
"semantic_similarity": similarity,
"length_ratio": length_ratio,
"student_length": len(student_response),
"teacher_length": len(teacher_response)
}
def run_benchmark(
self,
test_prompts: List[str],
teacher_model: str,
student_model: str,
system_prompt: str = None
) -> Dict:
"""
Run comprehensive benchmark comparing teacher vs student
"""
results = {
"test_info": {
"teacher_model": teacher_model,
"student_model": student_model,
"num_prompts": len(test_prompts),
"timestamp": datetime.now().isoformat()
},
"teacher_results": [],
"student_results": [],
"comparison": [],
"summary": {}
}
total_teacher_cost = 0
total_student_cost = 0
total_teacher_time = 0
total_student_time = 0
for idx, prompt in enumerate(tqdm(test_prompts, desc="Benchmarking")):
# Test teacher model
start_time = time.time()
teacher_response = self.client.create_chat_completion(
model=teacher_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1024
)
teacher_time = time.time() - start_time
teacher_content = teacher_response["choices"][0]["message"]["content"]
teacher_usage = teacher_response.get("usage", {})
teacher_cost = self.client.estimate_cost(
teacher_model,
teacher_usage.get("prompt_tokens", 0),
teacher_usage.get("completion_tokens", 0)
)
results["teacher_results"].append({
"prompt_idx": idx,
"response": teacher_content,
"latency_ms": teacher_time * 1000,
"cost_usd": teacher_cost,
"tokens": teacher_usage
})
# Test student model (distilled model simulation)
start_time = time.time()
student_response = self.client.create_chat_completion(
model=student_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1024
)
student_time = time.time() - start_time
student_content = student_response["choices"][0]["message"]["content"]
student_usage = student_response.get("usage", {})
student_cost = self.client.estimate_cost(
student_model,
student_usage.get("prompt_tokens", 0),
student_usage.get("completion_tokens", 0)
)
results["student_results"].append({
"prompt_idx": idx,
"response": student_content,
"latency_ms": student_time * 1000,
"cost_usd": student_cost,
"tokens": student_usage
})
# Compare
quality = self.evaluate_response_quality(
student_content,
teacher_content,
prompt
)
results["comparison"].append({
"prompt_idx": idx,
"quality_metrics": quality,
"teacher_latency_ms": teacher_time * 1000,
"student_latency_ms": student_time * 1000,
"speedup": teacher_time / max(student_time,
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง