Dataset Engineering ภาคปฏิบัติ: สร้าง Fine-tuning Dataset จากข้อมูลองค์กรจริง

# task_definition.py
from dataclasses import dataclass
 
@dataclass
class SalesEmailTask:
    """
    กำหนดโครงสร้างของแต่ละ training example
    ก่อนเก็บ data ต้อง define ตรงนี้ให้ชัดก่อน
    """
    email_type: str          # follow_up | intro | proposal | thank_you
    customer_context: str    # ข้อมูลลูกค้า สินค้า สถานะ
    expected_output: str     # อีเมลที่พร้อมส่ง
    quality_score: float     # 1-5, human rated
    source: str              # "real" | "synthetic"
 
    def to_messages(self, system_prompt: str) -> list[dict]:
        """แปลงเป็น chat format สำหรับ SFT"""
        return [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"ประเภท: {self.email_type}\n\n{self.customer_context}"},
            {"role": "assistant", "content": self.expected_output}
        ]

SYSTEM_PROMPT = """คุณเป็น Sales Communication Specialist ของบริษัท
เขียนอีเมลในสไตล์และ format ที่ทีม Sales ใช้จริง
ใช้ tone มืออาชีพแต่เป็นมิตร ปิดด้วย call-to-action ที่ชัดเจน"""

# collect_seed_data.py
import json
from pathlib import Path
 
def load_seed_emails(data_dir: str) -> list[SalesEmailTask]:
    """
    Load seed examples จาก JSON files ที่ทีม Sales เตรียมไว้
    Format: แต่ละ file = 1 email example
    """
    seeds = []
    for file in Path(data_dir).glob("*.json"):
        data = json.loads(file.read_text(encoding="utf-8"))
        seeds.append(SalesEmailTask(
            email_type=data["type"],
            customer_context=data["context"],
            expected_output=data["email"],
            quality_score=data.get("score", 4.0),

# filter_data.py
from dataclasses import dataclass
 
@dataclass
class QualityFilter:
    """เกณฑ์คัดกรอง — ปรับตาม domain"""
    min_output_length: int = 100    # อีเมลสั้นเกินไปมักไม่สมบูรณ์
    max_output_length: int = 2000   # ยาวเกินไปมักเป็น thread ไม่ใช่ single email
    required_fields: tuple = ("Subject:", "เรียน")  # ต้องมี format ครบ
    min_quality_score: float = 3.5
 
def filter_examples(
    examples: list[SalesEmailTask],
    criteria: QualityFilter = QualityFilter()
) -> list[SalesEmailTask]:
    """
    คัดกรองตาม quality criteria
    Return: examples ที่ผ่านเกณฑ์ + log ที่ถูกตัด
    """
    passed, rejected = [], []
 
    for ex in

# synthetic_generation.py
import anthropic
import json
import random
 
client = anthropic.Anthropic()
 
def generate_synthetic_examples(
    seeds: list[SalesEmailTask],
    target_count: int = 1000,
    batch_size: int = 5
) -> list[SalesEmailTask]:
    """
    Self-Instruct pipeline:
    1. สุ่ม seed 3 ตัวเป็น few-shot examples
    2. ให้ Claude สร้าง variation ใหม่
    3. Filter ด้วย quality criteria เดิม
    """
    synthetic = []
    email_types = ["follow_up", "intro", "proposal", "thank_you", "check_in"]
 
    while len(synthetic) < target_count:
        # สุ่ม seed examples เป็น few-shot

# หลัง generate แล้ว filter ทันที
raw_synthetic = generate_synthetic_examples(filtered_seeds, target_count=1200)
filtered_synthetic = filter_examples(raw_synthetic)  # ใช้ filter เดิม
 
# รวม seed + synthetic
full_dataset = filtered_seeds + filtered_synthetic
print(f"Final dataset: {len(full_dataset)} examples")
print(f"  Real: {sum(1 for x in full_dataset if x.source == 'real')}")
print(f"  Synthetic: {sum(1 for x in full_dataset if x.source == 'synthetic')}")

# deduplication.py
from datasketch import MinHash, MinHashLSH
 
def dedup_dataset(
    examples: list[SalesEmailTask],
    threshold: float = 0.7
) -> list[SalesEmailTask]:
    """
    MinHash LSH deduplication
    threshold 0.7 = ตัด pair ที่คล้ายกัน 70%+ ออก
    """
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    unique = []
 
    for i, ex in enumerate(examples):
        m = MinHash(num_perm=128)
        # hash ที่ระดับ 3-gram ของ output
        tokens = ex.expected_output.split()
        for j in range(len(tokens) - 2):

# prepare_dataset.py
from datasets import Dataset
 
def to_hf_dataset(
    examples: list[SalesEmailTask],
    system_prompt: str,
    test_ratio: float = 0.1
) -> tuple[Dataset, Dataset]:
    """
    แปลงเป็น HuggingFace Dataset ที่มี 'messages' column
    แบ่ง train/test โดยแยก real examples ไว้ใน test set ด้วย
    """
    records = []
    for ex in examples:
        records.append({
            "messages": ex.to_messages(system_prompt),
            "email_type": ex.email_type,
            "source": ex.source,
        })
 
    ds = Dataset.from_list(records)
    split = ds.train_test_split(test_size=test_ratio, seed=42)
 
    print

# train.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig
from trl import SFTConfig, SFTTrainer
 
# --- 1. Model & Tokenizer ---
MODEL_ID = "Qwen/Qwen2.5-7B-Instruct"  # Thai support ดี, open-weight
 
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)
 
model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    quantization_config=bnb_config,
    device_map="auto",
    attn_implementation=

# evaluate.py
from peft import PeftModel
 
def load_finetuned(base_model_id: str, adapter_path: str):
    """Load base model + LoRA adapter"""
    model = AutoModelForCausalLM.from_pretrained(
        base_model_id,
        quantization_config=bnb_config,
        device_map="auto",
    )
    model = PeftModel.from_pretrained(model, adapter_path)
    tokenizer = AutoTokenizer.from_pretrained(adapter_path)
    return model, tokenizer
 
 
def compare_outputs(
    test_examples: list[dict],
    base_model_id: str,
    adapter_path: str,
    system_prompt: str
):
    """
    เทียบ output จาก base model vs fine-tuned model

# llm_judge.py
def judge_email_quality(
    generated: str,
    reference: str,
    client: anthropic.Anthropic
) -> dict:
    """
    ใช้ Claude ให้คะแนนเทียบกับ reference email
    Return: scores 1-5 สำหรับแต่ละ dimension
    """
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": (
                f"ให้คะแนน 1-5 สำหรับ sales email นี้เทียบกับ reference:\n\n"
                f"Reference:\n{reference}\n\n"
                f"Generated:\n{generated}\n\n"

Seed Data (20 emails)
       |
  [Quality Filter]  ← criteria: length, format, score
       |
  17 clean seeds
       |
  [Synthetic Generation]  ← Claude API + Self-Instruct
       |
  1,200 synthetic
       |
  [Quality Filter]  ← same criteria as seeds
       |
  950 passed
       |
  [Deduplication]  ← MinHash LSH (threshold=0.7)
       |
  840 unique
       |
  [Format → HF Dataset]  ← messages format + train/test split
       |
  756 train / 84 test
       |
  [SFT + QLoRA]  ← TRL SFTTrainer + Qwen2.5-7B
       |
  LoRA Adapter (~50MB)
       |
  [Evaluation]  ← base vs fine-tuned on test set

# requirements.txt
 
# Training
transformers>=4.46.0       # Model loading, tokenizer
trl>=0.12.0                # SFTTrainer, SFTConfig
peft>=0.13.0               # LoRA, QLoRA
bitsandbytes>=0.44.0       # 4-bit quantization
accelerate>=1.0.0          # Multi-GPU, mixed precision
torch>=2.4.0               # PyTorch (CUDA 12.1+)
 
# Dataset
datasets>=3.0.0            # HuggingFace Datasets
datasketch>=1.6.0          # MinHash LSH deduplication
 
# Synthetic Generation
anthropic>=0.40.0          # Claude API สำหรับ Self-Instruct
 
# Evaluation
scikit-learn>=1.5.0