Day 26: Dataset Preparation cho Instruction Tuning

system policy + user instruction + optional history/input -> assistant response

{"text":"Tôi bị tính phí 2 lần sau khi nâng cấp.", "label":"billing_issue"}

{"messages":[{"role":"user","content":"Tôi bị tính phí 2 lần sau khi nâng cấp."},{"role":"assistant","content":"Mình xin lỗi vì bất tiện này. Mình sẽ kiểm tra giao dịch nâng cấp và chuyển ticket sang billing nếu có dấu hiệu tính phí trùng. Bạn vui lòng gửi mã hóa đơn hoặc thời điểm giao dịch, không gửi số thẻ."}]}

{"instruction":"Tóm tắt ticket sau thành 3 bullet.","input":"Khách báo bị tính phí 2 lần sau khi upgrade.","output":"- Khách bị tính phí 2 lần.\n- Sự cố xảy ra sau khi upgrade.\n- Cần kiểm tra billing và refund nếu đúng chính sách."}

{"conversations":[{"from":"human","value":"Khách bị tính phí 2 lần, nên trả lời sao?"},{"from":"gpt","value":"Bạn nên xin lỗi ngắn gọn, xác nhận sẽ kiểm tra billing, hỏi thông tin cần thiết và không hứa refund khi chưa có kết quả."}]}

{"messages":[{"role":"system","content":"You are a support assistant. Answer in Vietnamese, concise, no refund promise."},{"role":"user","content":"Khách bị tính phí 2 lần sau khi upgrade."},{"role":"assistant","content":"Mình xin lỗi vì bất tiện này. Mình sẽ giúp kiểm tra giao dịch upgrade và chuyển ticket sang billing nếu có dấu hiệu tính phí trùng. Bạn vui lòng gửi mã hóa đơn hoặc thời điểm giao dịch, không gửi số thẻ."}]}

{"id":"support_000001","domain":"customer_support","task":"billing_reply","source":"human_seed","quality_score":5,"pii_status":"redacted","license":"internal_training_allowed","group_id":"ticket_family_2026_001","messages":[{"role":"system","content":"You are a support assistant for a SaaS product. Answer in Vietnamese, concise, and do not promise refunds without billing verification."},{"role":"user","content":"Tôi bị tính phí 2 lần sau khi upgrade, email của tôi là [EMAIL]."},{"role":"assistant","content":"Mình xin lỗi vì bất tiện này. Mình sẽ kiểm tra giao dịch upgrade và chuyển ticket sang billing nếu có dấu hiệu tính phí trùng. Bạn không cần gửi thêm thông tin thẻ qua chat."}]}

{"messages":[{"role":"user","content":"Tài khoản của tôi bị khóa sau khi đổi mật khẩu."},{"role":"assistant","content":"Mình sẽ giúp bạn kiểm tra. Vì đây là vấn đề bảo mật, mình không thể mở khóa trực tiếp qua chat khi chưa xác minh tài khoản. Mình sẽ tạo ticket cho nhóm support và hướng dẫn bạn các bước xác minh tiếp theo."}]}

{"messages":[{"role":"user","content":"Tài khoản của tôi bị khóa sau khi đổi mật khẩu."},{"role":"assistant","content":"Chắc hệ thống lỗi, mình sẽ mở khóa ngay cho bạn."}]}

train: ticket_123 turn 1
test:  ticket_123 turn 2

train: ticket_family_001..400
validation: ticket_family_401..450
test: ticket_family_451..500

Guideline rõ ràng
  -> 30-50 human seed examples chất lượng cao
  -> generate variants theo task/domain/edge case
  -> validate schema
  -> redact PII giả hoặc thật
  -> dedup
  -> human review sample
  -> tag source = synthetic_reviewed
  -> split theo group/template

instruction_dataset/
  raw/
    human_seed.jsonl
    synthetic_candidates.jsonl
  processed/
    dataset_clean.jsonl
    dataset_split.jsonl
  reports/
    validation_report.md
    review_sample.md
  dataset_card.md
  metadata.json
  prepare_dataset.py

{"id":"support_000001","domain":"customer_support","task":"billing_reply","source":"human_seed","quality_score":5,"pii_status":"clean","license":"internal_training_allowed","group_id":"ticket_family_000001","messages":[{"role":"system","content":"You are a support assistant for a SaaS product. Answer in Vietnamese, concise, and do not promise refunds without billing verification."},{"role":"user","content":"Tôi bị tính phí 2 lần sau khi upgrade."},{"role":"assistant","content":"Mình xin lỗi vì bất tiện này. Mình sẽ kiểm tra giao dịch upgrade và chuyển ticket sang billing nếu có dấu hiệu tính phí trùng. Bạn vui lòng gửi mã hóa đơn hoặc thời điểm giao dịch, không gửi số thẻ."}]}

python3 prepare_dataset.py --input raw/input.jsonl --out-dir processed --dataset-name support_instruction_v1

from __future__ import annotations

import argparse
import hashlib
import json
import random
import re
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ALLOWED_ROLES = {"system", "user", "assistant"}
ALLOWED_PII_STATUS = {"clean", "redacted", "needs_review"}
ALLOWED_SOURCES = {"human_seed", "synthetic_reviewed", "production_redacted"}
ALLOWED_LICENSES = {"internal_training_allowed", "public_commercial_allowed", "research_only"}

EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
PHONE_RE = re.compile(r"\b(?:\+?\d[\d .-]{8,}\d)\b")
API_KEY_RE = re.compile(r"\b(?:sk|pk|api|token|key)[-_]?[A-Za-z0-9]{16,}\b", re.IGNORECASE)
CREDIT_CARD_RE = re.compile(r"\b(?:\d[ -]*?){13,19}\b")
CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")
WHITESPACE_RE = re.compile(r"[ \t]+")


@dataclass
class PreparedDataset:
    rows: list[dict[str, Any]]
    errors: list[str]
    warnings: list[str]
    counters: Counter[str]


def load_jsonl(path: Path) -> tuple[list[dict[str, Any]], list[str]]:
    rows: list[dict[str, Any]] = []
    errors: list[str] = []
    for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        try:
            value = json.loads(line)
        except json.JSONDecodeError as exc:
            errors.append(f"line {line_no}: invalid JSON: {exc.msg}")
            continue
        if not isinstance(value, dict):
            errors.append(f"line {line_no}: row must be an object")
            continue
        value["_line_no"] = line_no
        rows.append(value)
    return rows, errors


def normalize_text(text: str) -> str:
    text = CONTROL_CHARS_RE.sub("", text)
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = "\n".join(WHITESPACE_RE.sub(" ", part).strip() for part in text.split("\n"))
    return text.strip()


def redact_pii(text: str) -> tuple[str, bool]:
    original = text
    text = EMAIL_RE.sub("[EMAIL]", text)
    text = PHONE_RE.sub("[PHONE]", text)
    text = API_KEY_RE.sub("[SECRET]", text)
    text = CREDIT_CARD_RE.sub("[CARD]", text)
    return text, text != original


def validate_messages(row: dict[str, Any], row_id: str) -> list[str]:
    errors: list[str] = []
    messages = row.get("messages")
    if not isinstance(messages, list) or not messages:
        return [f"{row_id}: messages must be a non-empty list"]

    roles: list[str] = []
    for idx, message in enumerate(messages):
        if not isinstance(message, dict):
            errors.append(f"{row_id}: messages[{idx}] must be an object")
            continue
        role = message.get("role")
        content = message.get("content")
        if role not in ALLOWED_ROLES:
            errors.append(f"{row_id}: messages[{idx}].role is invalid: {role!r}")
        if not isinstance(content, str) or not content.strip():
            errors.append(f"{row_id}: messages[{idx}].content is empty")
        roles.append(role)

    start = 1 if roles and roles[0] == "system" else 0
    expected = "user"
    for idx in range(start, len(roles)):
        if roles[idx] != expected:
            errors.append(f"{row_id}: expected role {expected!r} at messages[{idx}], got {roles[idx]!r}")
            break
        expected = "assistant" if expected == "user" else "user"
    if roles and roles[-1] != "assistant":
        errors.append(f"{row_id}: last message must be assistant")
    return errors


def validate_required_fields(row: dict[str, Any]) -> list[str]:
    row_id = str(row.get("id") or f"line_{row.get('_line_no', 'unknown')}")
    errors: list[str] = []
    required_string_fields = ["id", "domain", "task", "source", "pii_status", "license", "group_id"]
    for field in required_string_fields:
        if not isinstance(row.get(field), str) or not row[field].strip():
            errors.append(f"{row_id}: {field} must be a non-empty string")

    quality_score = row.get("quality_score")
    if not isinstance(quality_score, int) or not 1 <= quality_score <= 5:
        errors.append(f"{row_id}: quality_score must be an integer from 1 to 5")
    if row.get("source") not in ALLOWED_SOURCES:
        errors.append(f"{row_id}: source must be one of {sorted(ALLOWED_SOURCES)}")
    if row.get("pii_status") not in ALLOWED_PII_STATUS:
        errors.append(f"{row_id}: pii_status must be one of {sorted(ALLOWED_PII_STATUS)}")
    if row.get("license") not in ALLOWED_LICENSES:
        errors.append(f"{row_id}: license must be one of {sorted(ALLOWED_LICENSES)}")
    errors.extend(validate_messages(row, row_id))
    return errors


def canonical_training_text(row: dict[str, Any]) -> str:
    parts = []
    for message in row["messages"]:
        parts.append(f"{message['role']}:{normalize_text(message['content']).lower()}")
    return "\n".join(parts)


def normalize_and_redact_row(row: dict[str, Any]) -> tuple[dict[str, Any], bool]:
    clean = {key: value for key, value in row.items() if not key.startswith("_")}
    redacted_any = False
    for key in ["id", "domain", "task", "source", "pii_status", "license", "group_id"]:
        if isinstance(clean.get(key), str):
            clean[key] = normalize_text(clean[key])
    normalized_messages = []
    for message in clean["messages"]:
        content = normalize_text(message["content"])
        content, redacted = redact_pii(content)
        redacted_any = redacted_any or redacted
        normalized_messages.append({"role": message["role"], "content": content})
    clean["messages"] = normalized_messages
    if redacted_any:
        clean["pii_status"] = "redacted"
    return clean, redacted_any


def split_by_group(rows: list[dict[str, Any]], seed: int) -> list[dict[str, Any]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[row["group_id"]].append(row)

    groups = list(grouped)
    random.Random(seed).shuffle(groups)
    total_rows = len(rows)
    train_limit = int(total_rows * 0.8)
    validation_limit = int(total_rows * 0.9)

    output: list[dict[str, Any]] = []
    seen_count = 0
    for group_id in groups:
        if seen_count < train_limit:
            split = "train"
        elif seen_count < validation_limit:
            split = "validation"
        else:
            split = "test"
        for row in grouped[group_id]:
            row["split"] = split
            output.append(row)
        seen_count += len(grouped[group_id])
    return output


def prepare(rows: list[dict[str, Any]], parse_errors: list[str], seed: int, min_quality: int) -> PreparedDataset:
    errors = list(parse_errors)
    warnings: list[str] = []
    counters: Counter[str] = Counter()
    valid_rows: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    seen_hashes: set[str] = set()

    for raw in rows:
        row_id = str(raw.get("id") or f"line_{raw.get('_line_no', 'unknown')}")
        field_errors = validate_required_fields(raw)
        if field_errors:
            errors.extend(field_errors)
            counters["invalid_schema"] += 1
            continue
        if raw["id"] in seen_ids:
            errors.append(f"{row_id}: duplicate id")
            counters["duplicate_id"] += 1
            continue
        seen_ids.add(raw["id"])

        row, redacted = normalize_and_redact_row(raw)
        if row["quality_score"] < min_quality:
            warnings.append(f"{row_id}: dropped because quality_score < {min_quality}")
            counters["dropped_low_quality"] += 1
            continue
        if row["pii_status"] == "needs_review":
            warnings.append(f"{row_id}: dropped because pii_status is needs_review")
            counters["dropped_needs_review"] += 1
            continue
        if row["license"] == "research_only":
            warnings.append(f"{row_id}: dropped because license is research_only")
            counters["dropped_license"] += 1
            continue
        if redacted:
            counters["redacted"] += 1

        digest = hashlib.sha256(canonical_training_text(row).encode("utf-8")).hexdigest()
        if digest in seen_hashes:
            warnings.append(f"{row_id}: dropped exact duplicate")
            counters["dropped_duplicate"] += 1
            continue
        seen_hashes.add(digest)
        valid_rows.append(row)

    split_rows = split_by_group(valid_rows, seed)
    counters.update(Counter(row["split"] for row in split_rows))
    return PreparedDataset(rows=split_rows, errors=errors, warnings=warnings, counters=counters)


def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
    path.write_text(
        "\n".join(json.dumps(row, ensure_ascii=False, sort_keys=True) for row in rows) + ("\n" if rows else ""),
        encoding="utf-8",
    )


def write_report(path: Path, result: PreparedDataset) -> None:
    lines = ["# Validation Report", ""]
    lines.append("## Counters")
    for key, value in sorted(result.counters.items()):
        lines.append(f"- {key}: {value}")
    lines.extend(["", "## Errors"])
    lines.extend(f"- {error}" for error in result.errors[:200])
    if len(result.errors) > 200:
        lines.append(f"- ... truncated {len(result.errors) - 200} more errors")
    lines.extend(["", "## Warnings"])
    lines.extend(f"- {warning}" for warning in result.warnings[:200])
    if len(result.warnings) > 200:
        lines.append(f"- ... truncated {len(result.warnings) - 200} more warnings")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def write_dataset_card(path: Path, dataset_name: str, result: PreparedDataset) -> None:
    split_counts = Counter(row["split"] for row in result.rows)
    source_counts = Counter(row["source"] for row in result.rows)
    task_counts = Counter(row["task"] for row in result.rows)
    lines = [
        f"# Dataset Card: {dataset_name}",
        "",
        "## Purpose",
        "Instruction tuning dataset for a domain assistant. Replace this section with the exact behavior goal before training.",
        "",
        "## Format",
        "JSONL messages format with role/content pairs and metadata fields.",
        "",
        "## Size",
        f"- total: {len(result.rows)}",
        f"- train: {split_counts.get('train', 0)}",
        f"- validation: {split_counts.get('validation', 0)}",
        f"- test: {split_counts.get('test', 0)}",
        "",
        "## Sources",
    ]
    lines.extend(f"- {source}: {count}" for source, count in sorted(source_counts.items()))
    lines.extend(["", "## Tasks"])
    lines.extend(f"- {task}: {count}" for task, count in sorted(task_counts.items()))
    lines.extend(
        [
            "",
            "## Privacy",
            "PII is either absent or redacted by the preparation script. Records marked needs_review are excluded.",
            "",
            "## License",
            "Only rows with training-allowed licenses are included. Verify this before production use.",
            "",
            "## Known Limitations",
            "- Regex redaction does not catch every possible PII type.",
            "- Near-duplicate detection still requires human review or a stronger similarity pipeline for large datasets.",
            "- Synthetic data quality depends on seed examples and review process.",
        ]
    )
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def write_metadata(path: Path, dataset_name: str, input_path: Path, result: PreparedDataset, seed: int) -> None:
    metadata = {
        "dataset_name": dataset_name,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "input_path": str(input_path),
        "format": "messages_jsonl",
        "seed": seed,
        "rows": len(result.rows),
        "counters": dict(result.counters),
        "production_ready": len(result.errors) == 0 and len(result.rows) > 0,
    }
    path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--out-dir", required=True, type=Path)
    parser.add_argument("--dataset-name", required=True)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--min-quality", type=int, default=4)
    args = parser.parse_args()

    args.out_dir.mkdir(parents=True, exist_ok=True)
    rows, parse_errors = load_jsonl(args.input)
    result = prepare(rows, parse_errors, seed=args.seed, min_quality=args.min_quality)

    write_jsonl(args.out_dir / "dataset_split.jsonl", result.rows)
    write_report(args.out_dir / "validation_report.md", result)
    write_dataset_card(args.out_dir / "dataset_card.md", args.dataset_name, result)
    write_metadata(args.out_dir / "metadata.json", args.dataset_name, args.input, result, args.seed)

    print(f"rows_in: {len(rows)}")
    print(f"rows_out: {len(result.rows)}")
    print(f"errors: {len(result.errors)}")
    print(f"warnings: {len(result.warnings)}")
    for key, value in sorted(result.counters.items()):
        print(f"{key}: {value}")


if __name__ == "__main__":
    main()

# Dataset Card: support_instruction_v1

## Purpose
Instruction tuning cho support assistant xử lý billing/account/how-to bằng tiếng Việt.

## Behavior Goal
- Trả lời ngắn gọn, lịch sự, có next step.
- Không hứa refund, mở khóa, xóa dữ liệu hoặc thay đổi billing nếu chưa có tool result.
- Escalate billing/security/legal risk.

## Format
JSONL messages format.

## Size
- Total: 500
- Train: 400
- Validation: 50
- Test: 50

## Sources
- Human seed: 50
- Synthetic reviewed: 350
- Production redacted: 100

## Privacy
- PII redacted.
- No password, OTP, API key, access token or raw customer identifier.
- Records marked needs_review are excluded.

## License and Ownership
Internal training allowed. Adapter is not approved for public release unless legal review approves.

## Quality Process
- Schema validation passed.
- Exact dedup completed before split.
- 50 random examples manually reviewed.
- 20 golden examples reserved for Day 28 eval.

## Known Limitations
- Regex redaction may miss rare PII patterns.
- Near-duplicate synthetic variants require manual review.
- Dataset teaches response behavior, not product knowledge.

{
  "dataset_name": "support_instruction_v1",
  "format": "messages_jsonl",
  "version": "2026-05-10",
  "owner": "ai-engineer-course",
  "intended_model_family": "Qwen/LLaMA-compatible chat model",
  "rows": 500,
  "splits": {
    "train": 400,
    "validation": 50,
    "test": 50
  },
  "sources": {
    "human_seed": 50,
    "synthetic_reviewed": 350,
    "production_redacted": 100
  },
  "privacy": {
    "pii_redacted": true,
    "needs_review_excluded": true,
    "secret_scan_required": true
  },
  "training_allowed": true,
  "notes": "Use for behavior/style fine-tuning, not as knowledge source."
}

instruction_dataset/
  raw/input.jsonl
  processed/dataset_split.jsonl
  processed/dataset_card.md
  processed/metadata.json
  processed/validation_report.md

# Behavior Guideline

Domain: customer_support
Language: Vietnamese with English technical terms when needed

1. ...
2. ...
10. ...

{"id":"support_seed_0001","domain":"customer_support","task":"billing_reply","source":"human_seed","quality_score":5,"pii_status":"clean","license":"internal_training_allowed","group_id":"support_seed_0001","messages":[{"role":"system","content":"You are a support assistant for a SaaS product. Answer in Vietnamese, concise, and do not promise refunds without billing verification."},{"role":"user","content":"Tôi bị tính phí 2 lần sau khi upgrade."},{"role":"assistant","content":"Mình xin lỗi vì bất tiện này. Mình sẽ kiểm tra giao dịch upgrade và chuyển ticket sang billing nếu có dấu hiệu tính phí trùng. Bạn vui lòng gửi mã hóa đơn hoặc thời điểm giao dịch, không gửi số thẻ."}]}

You are creating instruction tuning data for a Vietnamese SaaS support assistant.

Use this behavior guideline:
<paste guideline>

Generate 20 JSONL records in messages format.
Requirements:
- Vietnamese with proper diacritics.
- Include metadata fields: id, domain, task, source, quality_score, pii_status, license, group_id, messages.
- source must be synthetic_reviewed.
- pii_status must be clean.
- license must be internal_training_allowed.
- Vary user wording, emotion, missing information, and escalation cases.
- Assistant must not promise refund, account unlock, data deletion, or billing changes without verification.
- Output valid JSONL only, one object per line.

mkdir -p instruction_dataset/raw instruction_dataset/processed

instruction_dataset/raw/input.jsonl

cd instruction_dataset
python3 prepare_dataset.py --input raw/input.jsonl --out-dir processed --dataset-name support_instruction_v1

wc -l processed/dataset_split.jsonl
sed -n '1,120p' processed/validation_report.md
sed -n '1,120p' processed/dataset_card.md