Chào các bạn, tôi là một kỹ sư ML đã có 5 năm kinh nghiệm fine-tuning các mô hình ngôn ngữ lớn cho doanh nghiệp. Trong bài viết này, tôi sẽ chia sẻ những bí quyết thực chiến về chuẩn bị dữ liệu — bước quan trọng nhất quyết định 70% chất lượng mô hình fine-tuned của bạn.

Tại Sao Dữ Liệu Quan Trọng Hơn Kiến Trúc Mô Hình?

Theo kinh nghiệm của tôi, nhiều kỹ sư quá tập trung vào việc chọn mô hình và hyperparameters, nhưng lại bỏ qua chất lượng dữ liệu. Thực tế cho thấy:

So Sánh Chi Phí API Các Mô Hình 2026

Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem bảng so sánh chi phí để bạn có thể ước tính ngân sách cho việc chuẩn bị dữ liệu và inference:

Mô HìnhOutput ($/MTok)10M Tokens/Tháng
GPT-4.1$8.00$80.00
Claude Sonnet 4.5$15.00$150.00
Gemini 2.5 Flash$2.50$25.00
DeepSeek V3.2$0.42$4.20

Như bạn thấy, DeepSeek V3.2 qua HolySheep AI chỉ có giá $0.42/MTok — tiết kiệm tới 95% so với Claude Sonnet 4.5. Với tỷ giá ưu đãi ¥1=$1, đây là lựa chọn tối ưu cho các pipeline data processing.

Quy Trình Chuẩn Bị Dữ Liệu 5 Giai Đoạn

Giai Đoạn 1: Thu Thập Dữ Liệu Thô

Tôi thường bắt đầu bằng việc tổng hợp dữ liệu từ nhiều nguồn. Dưới đây là script Python để thu thập dữ liệu qua API:

# Thu thập dữ liệu huấn luyện qua HolySheep AI API
import requests
import json
from typing import List, Dict

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"

def generate_training_data(prompts: List[str], model: str = "deepseek-chat") -> List[Dict]:
    """Tạo dữ liệu huấn luyện từ prompts"""
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    training_data = []
    
    for prompt in prompts:
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": "Bạn là chuyên gia tạo dữ liệu huấn luyện chất lượng cao."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.7,
            "max_tokens": 2048
        }
        
        response = requests.post(
            f"{BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            training_data.append({
                "instruction": prompt,
                "response": result["choices"][0]["message"]["content"]
            })
    
    return training_data


Ví dụ sử dụng

sample_prompts = [
    "Viết một đoạn mô tả về tính năng đăng nhập",
    "Tạo hướng dẫn reset mật khẩu",
    "Soạn email chăm sóc khách hàng"
]

data = generate_training_data(sample_prompts)
print(f"Đã thu thập {len(data)} mẫu dữ liệu")

Giai Đoạn 2: Làm Sạch Cơ Bản (Basic Cleaning)

Sau khi thu thập, bước tiếp theo là loại bỏ noise và dữ liệu không hợp lệ. Tôi sử dụng pipeline xử lý đa tầng:

import re
from typing import List, Dict, Tuple

def basic_cleaning(text: str) -> str:
    """Làm sạch văn bản cơ bản"""
    # Loại bỏ khoảng trắng thừa
    text = re.sub(r'\s+', ' ', text).strip()
    
    # Loại bỏ ký tự điều khiển
    text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
    
    # Chuẩn hóa dấu câu
    text = re.sub(r'[\u2018\u2019]', "'", text)  # Single quotes
    text = re.sub(r'[\u201c\u201d]', '"', text)   # Double quotes
    
    return text

def validate_sample(sample: Dict) -> Tuple[bool, str]:
    """Validate một cặp instruction-response"""
    
    instruction = sample.get("instruction", "")
    response = sample.get("response", "")
    
    # Kiểm tra độ dài tối thiểu
    if len(instruction) < 5:
        return False, "Instruction quá ngắn"
    
    if len(response) < 10:
        return False, "Response quá ngắn"
    
    # Kiểm tra độ dài tối đa (tránh quá dài)
    if len(response) > 8000:
        return False, "Response quá dài"
    
    # Kiểm tra không có placeholder
    placeholders = ["[TODO]", "[PLACEHOLDER]", "[INSERT]", "..."]
    for ph in placeholders:
        if ph.lower() in response.lower():
            return False, f"Chứa placeholder: {ph}"
    
    return True, "OK"

def clean_dataset(raw_data: List[Dict]) -> List[Dict]:
    """Làm sạch toàn bộ dataset"""
    cleaned = []
    stats = {"total": 0, "kept": 0, "removed": 0, "reasons": {}}
    
    for item in raw_data:
        stats["total"] += 1
        
        # Basic cleaning
        item["instruction"] = basic_cleaning(item["instruction"])
        item["response"] = basic_cleaning(item["response"])
        
        # Validate
        is_valid, reason = validate_sample(item)
        
        if is_valid:
            cleaned.append(item)
            stats["kept"] += 1
        else:
            stats["removed"] += 1
            stats["reasons"][reason] = stats["reasons"].get(reason, 0) + 1
    
    print(f"Tổng: {stats['total']}, Giữ lại: {stats['kept']}, Loại bỏ: {stats['removed']}")
    print(f"Nguyên nhân loại bỏ: {stats['reasons']}")
    
    return cleaned


Áp dụng cleaning

cleaned_data = clean_dataset(data)
print(f"Dataset sau cleaning: {len(cleaned_data)} mẫu")

Giai Đoạn 3: Phát Hiện Và Loại Bỏ Duplicat

Duplicate là kẻ thù của quá trình fine-tuning vì nó làm mất cân bằng phân bố dữ liệu. Tôi sử dụng thuật toán semantic deduplication:

from collections import defaultdict
import hashlib

def get_text_hash(text: str, n_gram: int = 3) -> set:
    """Tạo hash cho text dựa trên n-gram"""
    words = text.lower().split()
    hashes = set()
    
    for i in range(len(words) - n_gram + 1):
        ngram = tuple(words[i:i + n_gram])
        hash_val = hashlib.md5(str(ngram).encode()).hexdigest()[:8]
        hashes.add(hash_val)
    
    return hashes

def jaccard_similarity(set1: set, set2: set) -> float:
    """Tính độ tương đồng Jaccard giữa 2 sets"""
    if not set1 or not set2:
        return 0.0
    intersection = len(set1 & set2)
    union = len(set1 | set2)
    return intersection / union if union > 0 else 0.0

def remove_duplicates(dataset: List[Dict], threshold: float = 0.85) -> List[Dict]:
    """Loại bỏ duplicates dựa trên Jaccard similarity"""
    unique_samples = []
    seen_hashes = []
    
    for sample in dataset:
        response = sample.get("response", "")
        sample_hash = get_text_hash(response)
        
        is_duplicate = False
        for seen in seen_hashes:
            similarity = jaccard_similarity(sample_hash, seen)
            if similarity >= threshold:
                is_duplicate = True
                break
        
        if not is_duplicate:
            unique_samples.append(sample)
            seen_hashes.append(sample_hash)
    
    removed = len(dataset) - len(unique_samples)
    print(f"Loại bỏ {removed} duplicates (threshold: {threshold})")
    
    return unique_samples


Áp dụng deduplication

deduplicated = remove_duplicates(cleaned_data, threshold=0.85)
print(f"Dataset sau dedup: {len(deduplicated)} mẫu")

Giai Đoạn 4: Kiểm Tra Chất Lượng Bằng Mô Hình

Đây là kỹ thuật nâng cao mà tôi học được từ thực tế: sử dụng chính LLM để đánh giá chất lượng dữ liệu. Với HolySheep AI, chi phí chỉ $0.42/MTok cho DeepSeek V3.2 — rẻ hơn rất nhiều so với các provider khác:

import requests
import json

def evaluate_quality_with_llm(sample: Dict) -> Dict:
    """Đánh giá chất lượng sample bằng LLM"""
    
    prompt = f"""Đánh giá chất lượng cặp instruction-response sau:
    
Instruction: {sample['instruction']}

Response: {sample['response']}

Đánh giá theo thang điểm 1-5 cho:
1. Relevance (độ liên quan): Response có trả lời đúng instruction không?
2. Completeness (độ hoàn chỉnh): Response có đầy đủ thông tin không?
3. Safety (an toàn): Response có chứa nội dung harmful không?
4. Format (định dạng): Response có đúng format không?

Trả lời theo format JSON:
{{"relevance": int, "completeness": int, "safety": int, "format": int, "overall": int, "reason": str}}
"""
    
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1,
        "max_tokens": 500
    }
    
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        result = response.json()
        content = result["choices"][0]["message"]["content"]
        return json.loads(content)
    
    return None

def filter_by_quality(dataset: List[Dict], min_score: int = 3) -> List[Dict]:
    """Lọc dataset dựa trên điểm chất lượng"""
    qualified = []
    total_cost = 0
    
    for i, sample in enumerate(dataset):
        score = evaluate_quality_with_llm(sample)
        
        if score and score.get("overall", 0) >= min_score:
            qualified.append(sample)
        
        # Ước tính chi phí (mỗi evaluation ~500 tokens output)
        total_cost += 500 * 0.42 / 1_000_000  # DeepSeek price
        
        if (i + 1) % 10 == 0:
            print(f"Đã đánh giá {i + 1}/{len(dataset)} samples, chi phí: ${total_cost:.4f}")
    
    print(f"\nTổng chi phí đánh giá: ${total_cost:.4f}")
    print(f"Giữ lại: {len(qualified)}/{len(dataset)} samples")
    
    return qualified


Áp dụng quality filter

high_quality_data = filter_by_quality(deduplicated, min_score=3)
print(f"Dataset chất lượng cao: {len(high_quality_data)} mẫu")

Giai Đoạn 5: Định Dạng Cuối Cùng

def format_for_finetuning(data: List[Dict], format_type: str = "chatml") -> str:
    """Format dataset cho fine-tuning"""
    
    if format_type == "chatml":
        formatted = []
        for item in data:
            chatml = f"""<|im_start|>system
Bạn là trợ lý AI hữu ích.<|im_end|>
<|im_start|>user
{item['instruction']}<|im_end|>
<|im_start|>assistant
{item['response']}<|im_end|>"""
            formatted.append(chatml)
        
        output = "\n\n".join(formatted)
    
    elif format_type == "instruction":
        output = json.dumps([
            {
                "messages": [
                    {"role": "user", "content": item["instruction"]},
                    {"role": "assistant", "content": item["response"]}
                ]
            }
            for item in data
        ], ensure_ascii=False, indent=2)
    
    return output


Export final dataset

final_dataset = format_for_finetuning(high_quality_data, format_type="instruction")

with open("training_data.jsonl", "w", encoding="utf-8") as f:
    f.write(final_dataset)

print(f"Đã xuất {len(high_quality_data)} samples ra training_data.jsonl")

Lỗi Thường Gặp Và Cách Khắc Phục

Lỗi 1: MemoryError Khi Xử Lý Dataset Lớn

Mô tả: Khi xử lý dataset >100K samples, script bị crash với lỗi MemoryError do load toàn bộ vào RAM.

# ❌ SAI: Load toàn bộ vào memory
def process_all_data(filename):
    with open(filename, 'r') as f:
        all_data = json.load(f)  # Crash với dataset lớn
    return [clean(item) for item in all_data]


✅ ĐÚNG: Xử lý streaming

def process_data_streaming(input_file: str, output_file: str, batch_size: int = 1000):
    """Xử lý dataset lớn theo streaming để tiết kiệm memory"""
    
    buffer = []
    total_processed = 0
    
    with open(input_file, 'r', encoding='utf-8') as fin, \
         open(output_file, 'w', encoding='utf-8') as fout:
        
        for line in fin:
            item = json.loads(line.strip())
            cleaned_item = clean_sample(item)
            
            if cleaned_item:
                buffer.append(cleaned_item)
                total_processed += 1
            
            # Flush buffer khi đầy
            if len(buffer) >= batch_size:
                for buffered_item in buffer:
                    fout.write(json.dumps(buffered_item, ensure_ascii=False) + '\n')
                buffer.clear()
                
                if total_processed % 10000 == 0:
                    print(f"Đã xử lý: {total_processed} samples")
        
        # Flush remaining
        for item in buffer:
            fout.write(json.dumps(item, ensure_ascii=False) + '\n')
    
    print(f"Hoàn thành! Tổng: {total_processed} samples")
    return total_processed


Sử dụng: process_data_streaming("raw_data.jsonl", "cleaned_data.jsonl")

Lỗi 2: UnicodeEncodeError Với Tiếng Việt

Mô tả: Khi xuất dữ liệu tiếng Việt ra file, gặp lỗi encoding.

# ❌ SAI: Không chỉ định encoding
with open("output.json", "w") as f:
    f.write(json.dumps(data))


✅ ĐÚNG: Specify UTF-8 encoding

import sys


Set UTF-8 cho stdout

if sys.platform == "win32":
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

def save_vietnamese_json(data: List[Dict], filepath: str):
    """Lưu JSON với hỗ trợ tiếng Việt đầy đủ"""
    
    with open(filepath, 'w', encoding='utf-8') as f:
        # Sử dụng ensure_ascii=False để giữ nguyên ký tự tiếng Việt
        json.dump(data, f, ensure_ascii=False, indent=2)
    
    # Verify
    with open(filepath, 'r', encoding='utf-8') as f:
        verified = json.load(f)
        print(f"Đã lưu và verify {len(verified)} items thành công")


Nếu vẫn lỗi, sử dụng fallback encoding

def safe_write(text: str, filepath: str):
    """Safe write với nhiều encoding fallback"""
    encodings = ['utf-8', 'utf-8-sig', 'latin-1']
    
    for encoding in encodings:
        try:
            with open(filepath, 'w', encoding=encoding) as f:
                f.write(text)
            print(f"Ghi file thành công với encoding: {encoding}")
            return True
        except UnicodeEncodeError:
            continue
    
    # Fallback cuối: escape unicode
    with open(filepath, 'w', encoding='utf-8', errors='replace') as f:
        f.write(text)
    print("Đã ghi file với unicode replacement")
    return False

Lỗi 3: API Rate Limit Khi Gọi Hàng Loạt

Mô tả: Khi gọi API để tạo/đánh giá dữ liệu, gặp lỗi 429 Rate Limit.

import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session() -> requests.Session:
    """Tạo session với automatic retry"""
    
    session = requests.Session()
    
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    
    return session

def batch_api_call(items: List[Dict], model: str = "deepseek-chat") -> List[Dict]:
    """Gọi API với rate limit handling"""
    
    results = []
    session = create_resilient_session()
    
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    for i, item in enumerate(items):
        success = False
        attempts = 0
        
        while not success and attempts < 5:
            try:
                payload = {
                    "model": model,
                    "messages": [{"role": "user", "content": item["prompt"]}],
                    "max_tokens": 1000
                }
                
                response = session.post(
                    f"{BASE_URL}/chat/completions",
                    headers=headers,
                    json=payload,
                    timeout=60
                )
                
                if response.status_code == 200:
                    result = response.json()
                    results.append(result["choices"][0]["message"]["content"])
                    success = True
                    
                elif response.status_code == 429:
                    # Rate limited - đợi và thử lại
                    wait_time = int(response.headers.get("Retry-After", 60))
                    print(f"Rate limited! Đợi {wait_time}s...")
                    time.sleep(wait_time)
                    attempts += 1
                    
                else:
                    print(f"Lỗi {response.status_code}: {response.text}")
                    break
                    
            except Exception as e:
                print(f"Exception: {e}")
                time.sleep(5)
                attempts += 1
        
        # Progress reporting
        if (i + 1) % 50 == 0:
            print(f"Tiến độ: {i + 1}/{len(items)}")
    
    return results


Sử dụng với exponential backoff tự động

results = batch_api_call(dataset[:100], model="deepseek-chat")

Lỗi 4: Chất Lượng Không Đồng Đều Sau Fine-tuning

Mô tả: Mô hình fine-tuned hoạt động tốt trên một số loại input nhưng rất kém trên các loại khác.

def analyze_data_distribution(dataset: List[Dict]) -> Dict:
    """Phân tích phân bố topic/intent trong dataset"""
    
    topics = defaultdict(int)
    lengths = []
    
    for item in dataset:
        # Categorize by first few words (simplified)
        first_words = item['instruction'].lower().split()[:3]
        topic_key = ' '.join(first_words)
        topics[topic_key] += 1
        
        lengths.append(len(item['response'].split()))
    
    # Report
    print("=== PHÂN BỐ TOPIC ===")
    sorted_topics = sorted(topics.items(), key=lambda x: x[1], reverse=True)
    for topic, count in sorted_topics[:10]:
        percentage = count / len(dataset) * 100
        print(f"{topic}: {count} ({percentage:.1f}%)")
    
    print(f"\n=== ĐỘ DÀI RESPONSE ===")
    print(f"Min: {min(lengths)}, Max: {max(lengths)}, Avg: {sum(lengths)/len(lengths):.1f}")
    
    # Kiểm tra imbalance
    max_topic = sorted_topics[0][1] if sorted_topics else 0
    if max_topic / len(dataset) > 0.3:
        print(f"⚠️ CẢNH BÁO: Topic '{sorted_topics[0][0]}' chiếm {max_topic/len(dataset)*100:.1f}%")
    
    return {"topics": dict(sorted_topics), "avg_length": sum(lengths)/len(lengths)}


Balance dataset nếu cần

def balance_dataset(dataset: List[Dict], max_per_topic: int = 1000) -> List[Dict]:
    """Cân bằng dataset để tránh overfitting"""
    
    topic_samples = defaultdict(list)
    
    for item in dataset:
        first_words = item['instruction'].lower().split()[:3]
        topic_key = ' '.join(first_words)
        topic_samples[topic_key].append(item)
    
    balanced = []
    for topic, samples in topic_samples.items():
        if len(samples) > max_per_topic:
            # Random sample
            import random
            balanced.extend(random.sample(samples, max_per_topic))
            print(f"Topic '{topic}': giảm từ {len(samples)} xuống {max_per_topic}")
        else:
            balanced.extend(samples)
    
    random.shuffle(balanced)
    print(f"\nDataset sau cân bằng: {len(balanced)} samples")
    
    return balanced


Chạy analysis

stats = analyze_data_distribution(dataset)
balanced_data = balance_dataset(dataset, max_per_topic=500)

Tổng Kết Chi Phí Pipeline

Dựa trên kinh nghiệm thực tế của tôi, đây là ước tính chi phí cho một pipeline hoàn chỉnh với 100K samples:

Với HolySheep AI, bạn được hưởng:

Kinh Nghiệm Thực Chiến

Qua 5 năm làm việc với fine-tuning, tôi rút ra được những bài học quý giá:

Bài học 1: Đừng tiết kiệm thời gian ở bước làm sạch. Một lần tôi bỏ qua bước deduplication và mô hình bị overfitting nghiêm trọng với một số pattern lặp đi lặp lại.

Bài học 2: Luôn có tập validation riêng biệt. Tôi đã từng dùng chung train/validation set và không phát hiện ra data leakage cho đến khi deploy.

Bài học 3: Với tiếng Việt, cần chú ý đặc biệt đến dấu câu và format. Nhiều dữ liệu từ web có encoding không chuẩn, gây ra lỗi khó debug.

Bài học 4: Sử dụng streaming cho dataset lớn. Một lần tôi load 2GB JSON vào RAM và máy bị treo trong 30 phút trước khi crash.

Chúc các bạn thành công với pipeline fine-tuning của mình! Đừng quên tận dụng chi phí cực thấp của HolySheep AI để tối ưu hóa ngân sách.

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký

Tài nguyên liên quan

Bài viết liên quan