Bài 48: Concept Drift — khi quan hệ input/output thay đổi

Sau bài này bạn sẽ:

• ✅ Phân biệt concept drift (P(Y|X)), data drift (P(X)), label drift (P(Y)) và covariate shift
• ✅ Nhận biết 4 loại concept drift theo tốc độ thay đổi
• ✅ Dùng được DDM và ADWIN từ thư viện river để detect drift trên stream
• ✅ Tính Expected Calibration Error (ECE) như proxy khi label bị delay
• ✅ Xây dựng pattern bucket-based monitoring với statistical test
• ✅ Hiểu drift đặc thù cho LLM/RAG và cách track prompt/model version

Khi train, model học hàm f: X → Y dựa trên phân phối P(Y|X) có trong training data. Concept drift xảy ra khi phân phối thật sự của P(Y|X) trong production thay đổi so với lúc train — nghĩa là cùng input X, output đúng đắn Y giờ đã khác.

Model không biết điều này. Nó vẫn áp dụng hàm f đã học, nhưng hàm đó không còn phản ánh thực tế nữa. Kết quả là accuracy hoặc metric production suy giảm theo thời gian dù input distribution vẫn ổn định.

Ví dụ cụ thể

Spam filter: Model train trên email năm 2020 học được pattern "đặc ngữ nhất định + link lạ → spam". Đến 2024, spammer thay đổi chiến thuật: dùng text tự nhiên, tránh từ khóa đã bị blacklist. Cùng một "email nhìn thông thường" — năm 2020 là ham (không spam), nhưng nếu cùng pattern đó xuất hiện trong chiến dịch mới thì là spam. Model cũ sẽ phân loại sai.

Credit scoring: Model train 2019 học rằng thu nhập ổn định + không nợ xấu → low default risk. Sau COVID-2020, nhiều người có thu nhập trước đây ổn nhưng vẫn default do mất việc. Cùng feature profile X, nhưng P(default=1|X) đã cao hơn hẳn so với 2019.

Churn prediction: Sau khi đổi pricing model, user giờ churn với hành vi khác trước — họ dùng ít hơn nhưng không hẳn là sắp rời. Hàm f(behavior → churn risk) đã thay đổi.

Trong production ML, có 3 loại drift thường gặp với nguyên nhân và action khác nhau:

Data drift — P(X) thay đổi

Distribution của input feature thay đổi. Vd: user base trẻ hóa → trường age giờ tập trung ở 18–25 thay vì 30–45 như lúc train. Model có thể chưa thấy nhiều sample ở vùng này trong training.

Detect được không cần label — chỉ cần so sánh phân phối feature train vs production (PSI, KS test, Wasserstein distance — chi tiết ở bài 47).

Concept drift — P(Y|X) thay đổi

Quan hệ input→output thay đổi. Vd: cùng age=25 + income=50M, nhưng churn risk năm 2024 khác năm 2020 do sản phẩm thay đổi. Data distribution có thể vẫn ổn định, nhưng label đúng đắn cho cùng feature vector đã đổi.

Cần label ground truth để detect — và đây là điểm khó nhất: label thường đến muộn hơn prediction rất nhiều.

Label drift — P(Y) thay đổi

Phân phối của label thay đổi. Vd: churn rate năm trước là 5%, năm nay 12%. Hoặc tỷ lệ fraud tăng đột biến theo mùa.

Label drift thường là hệ quả của concept drift hoặc data drift, đôi khi cũng do định nghĩa label bị cập nhật. Detect được khi có label, hoặc một phần qua model output distribution (nếu model well-calibrated).

Covariate shift — trường hợp hỗn hợp

P(X) thay đổi nhưng P(Y|X) giữ nguyên. Ví dụ: thêm user từ thành phố mới — feature distribution đổi nhưng quan hệ feature→label vẫn đúng. Model vẫn valid, chỉ cần đảm bảo coverage tốt.

Trong thực tế, data drift và concept drift thường xảy ra cùng lúc. Phân biệt chúng ảnh hưởng đến action: data drift → kiểm tra data pipeline và feature representation; concept drift → xem lại model hoặc retrain với data mới.

Một số nguyên nhân phổ biến:

User behavior thay đổi

Người dùng trở nên tech-savvy hơn, thói quen tiêu dùng thay đổi theo thế hệ, hoặc context sử dụng app thay đổi (mobile-first, dark mode usage tăng). Hành vi mà model học năm ngoái không còn đại diện cho hành vi thực tế.

Adversarial adaptation

Kẻ gian (spammer, fraudster, SEO manipulator) quan sát hành vi của model và điều chỉnh để tránh bị phát hiện. Đây là concept drift có chủ ý và thường xảy ra nhanh nhất. Spam filter, fraud detection, và content moderation thường phải đối mặt nhiều nhất.

Economic và policy shift

Suy thoái kinh tế làm thay đổi spending pattern và default risk. Thay đổi luật (GDPR, thuế mới) ảnh hưởng hành vi người dùng. Sự kiện bất thường (COVID, thiên tai) tạo ra concept drift đột ngột và mạnh.

Definition change

Định nghĩa của "fraud", "churn", "engagement" hoặc label đích được cập nhật bởi business stakeholder. Đây là concept drift "do con người tạo ra" — model vẫn đúng với định nghĩa cũ, nhưng business đã thay đổi yêu cầu.

Seasonal và cyclical effect

Behavior trong Q4 (mua sắm) khác Q1. Summer vs winter. Ngày trong tuần vs cuối tuần. Nếu model không được train đủ data trải đều các chu kỳ, nó sẽ sai khi gặp chu kỳ mới.

Cách P(Y|X) thay đổi theo thời gian quyết định strategy detect và xử lý:

Sudden drift

Thay đổi đột ngột, xảy ra trong khoảng thời gian ngắn. Ví dụ: COVID lockdown tháng 3/2020 — spending pattern thay đổi trong vài ngày. Model ecommerce recommendation học pattern offline-shopping không còn relevance ngay lập tức.

Detect: alarm nhanh bằng sliding window ngắn (daily). Xử lý: retrain với data sau sự kiện, có thể warm-start từ checkpoint gần nhất thay vì train lại từ đầu.

Gradual drift

Thay đổi dần dần qua nhiều tháng. Ví dụ: thói quen mua hàng chuyển dần từ desktop sang mobile qua 2 năm. Khó detect hơn vì không có inflection point rõ ràng.

Detect: cần window dài hơn (weekly/monthly comparison). Statistical test giữa performance 30 ngày gần nhất vs 30 ngày baseline.

Incremental drift

Liên tục thay đổi nhỏ mỗi ngày, không bao giờ dừng. Ví dụ: language evolves — từ mới xuất hiện, nghĩa từ cũ thay đổi. NLP model train trên corpus 2022 dần dần kém chính xác với text 2025.

Online learning (model update incremental với data mới) phù hợp hơn periodic retrain với loại drift này.

Recurring (seasonal) drift

Drift theo chu kỳ và lặp lại. Ví dụ: Q4 mua sắm, flu season ảnh hưởng healthcare prediction, tax season thay đổi financial behavior.

Không phải drift cần "fix" — cần model có thể xử lý seasonality, hoặc maintain model riêng cho từng mùa. Detect bằng cách so sánh với cùng kỳ năm trước thay vì kỳ trước.

Cách trực tiếp nhất để phát hiện concept drift là track performance metric (accuracy, F1, AUC-ROC, precision/recall) theo thời gian và so sánh với baseline.

Sliding window comparison

import numpy as np
from sklearn.metrics import f1_score

def check_performance_drift(
    y_true_recent: list,
    y_pred_recent: list,
    y_true_baseline: list,
    y_pred_baseline: list,
    threshold: float = 0.05,
) -> dict:
    """
    So sánh F1 giữa window gần nhất và baseline window.
    threshold: mức suy giảm tuyệt đối coi là drift.
    """
    f1_recent = f1_score(y_true_recent, y_pred_recent, average="weighted")
    f1_baseline = f1_score(y_true_baseline, y_pred_baseline, average="weighted")
    delta = f1_baseline - f1_recent

    return {
        "f1_baseline": round(f1_baseline, 4),
        "f1_recent": round(f1_recent, 4),
        "delta": round(delta, 4),
        "drift_detected": delta > threshold,
    }

# Ví dụ sử dụng
result = check_performance_drift(
    y_true_recent=labels_last_7d,
    y_pred_recent=preds_last_7d,
    y_true_baseline=labels_prev_7d,
    y_pred_baseline=preds_prev_7d,
    threshold=0.05,
)
print(result)
# {"f1_baseline": 0.871, "f1_recent": 0.803, "delta": 0.068, "drift_detected": True}

Statistical test trên metric series

Thay vì so sánh 2 window đơn, có thể dùng T-test để kiểm tra xem mean accuracy có khác nhau đáng kể giữa 2 kỳ:

from scipy import stats

def ttest_metric_windows(
    metrics_window_a: list[float],
    metrics_window_b: list[float],
    alpha: float = 0.05,
) -> dict:
    """
    metrics_window_a: list accuracy/F1 theo ngày trong kỳ baseline.
    metrics_window_b: list accuracy/F1 theo ngày trong kỳ gần nhất.
    """
    t_stat, p_value = stats.ttest_ind(metrics_window_a, metrics_window_b)
    return {
        "t_stat": round(t_stat, 4),
        "p_value": round(p_value, 4),
        "drift_detected": p_value < alpha,
        "mean_a": round(np.mean(metrics_window_a), 4),
        "mean_b": round(np.mean(metrics_window_b), 4),
    }

Hạn chế chính

Phương pháp này yêu cầu label ground truth đã có. Trong nhiều use case, label đến rất muộn — churn label phải chờ 30 ngày, default credit phải chờ 90 ngày. Phần 10 sẽ đề cập label delay và workaround.

DDM (Drift Detection Method, Gama et al. 2004) là thuật toán detect drift trên data stream bằng cách theo dõi error rate streaming. Thuật toán duy trì mean error rate \( p_t \) và standard deviation \( s_t \), alert khi \( p_t + 3s_t \) vượt quá giá trị tốt nhất đã quan sát.

Thư viện river (kế thừa từ scikit-multiflow) implement DDM và nhiều detector khác cho streaming ML:

pip install river

from river.drift import DDM

detector = DDM()

# stream là iterator trả về (x, y_true, y_pred) theo thứ tự thời gian
for x, y_true, y_pred in stream:
    is_error = int(y_true != y_pred)
    detector.update(is_error)

    if detector.drift_detected:
        print(f"Concept drift detected at sample index {sample_idx}")
        # Trigger retrain hoặc alert
        detector = DDM()  # Reset detector sau khi xử lý drift
    elif detector.warning_detected:
        print("Warning: possible drift forming")

Tham số DDM

from river.drift import DDM

detector = DDM(
    warm_start=30,    # Số sample tối thiểu trước khi bắt đầu detect (default: 30)
    warning_threshold=2.0,   # Ngưỡng cảnh báo: p + warning_threshold * s
    drift_threshold=3.0,     # Ngưỡng drift: p + drift_threshold * s
)

DDM phù hợp khi label có ngay sau prediction (online learning scenario). Với hệ thống có label delay, cần kết hợp với proxy metric hoặc calibration tracking.

ADWIN — Adaptive Windowing

ADWIN (Bifet & Gavalda, 2007) dùng cửa sổ kích thước động: mở rộng khi data ổn định, thu hẹp khi phát hiện có sự thay đổi trong mean. Không cần chỉ định trước window size — thuật toán tự điều chỉnh.

from river.drift import ADWIN

adwin = ADWIN(delta=0.002)  # delta: độ nhạy — nhỏ hơn = nhạy hơn

for error_value in error_stream:
    adwin.update(error_value)

    if adwin.drift_detected:
        print(f"ADWIN detected drift. Current estimate: {adwin.estimation:.4f}")

ADWIN phù hợp khi không biết trước tốc độ drift (sudden hay gradual). Hữu ích khi cần adaptive window tự điều chỉnh. Overhead bộ nhớ cao hơn DDM nhưng ít false alarm hơn với gradual drift.

Page-Hinkley test

Page-Hinkley (Page, 1954) là sequential change-point detection test, phù hợp để detect mean shift trong continuous metric (vd mean prediction score, không phải binary error):

from river.drift import PageHinkley

# Theo dõi mean của prediction probability (continuous value)
ph = PageHinkley(
    min_instances=30,
    delta=0.005,      # Mức tăng tích lũy cho phép giữa hai observation
    threshold=50,     # Ngưỡng phát hiện
    alpha=0.9999,     # Forgetting factor
)

for prob_score in prediction_scores_stream:
    ph.update(prob_score)
    if ph.drift_detected:
        print("Mean shift in prediction score detected")

So sánh DDM vs ADWIN vs Page-Hinkley

Khi label đến muộn, không thể tính F1 hay accuracy ngay. Một tín hiệu gián tiếp có thể quan sát được ngay là calibration của model: phân phối confidence score đầu ra có còn phản ánh đúng xác suất thực tế không?

Model tốt khi predict probability 0.8 thì trong thực tế ~80% số đó là positive. Khi concept drift xảy ra, calibration thường bị ảnh hưởng trước khi thấy rõ qua performance metric — vì model vẫn đưa ra các số confident nhưng không còn chính xác nữa.

Expected Calibration Error (ECE)

ECE đo mức độ lệch trung bình giữa confidence score và actual accuracy qua các bin:

import numpy as np
from sklearn.calibration import calibration_curve

def expected_calibration_error(
    probs: np.ndarray,
    labels: np.ndarray,
    n_bins: int = 10,
) -> float:
    """
    Tính Expected Calibration Error (ECE).
    probs: predicted probabilities cho class positive, shape (n,).
    labels: ground truth binary labels, shape (n,).
    Trả về ECE trong khoảng [0, 1]. Thấp hơn = calibrate tốt hơn.
    """
    prob_true, prob_pred = calibration_curve(
        labels, probs, n_bins=n_bins, strategy="uniform"
    )
    # Weighted mean absolute error giữa actual và predicted fraction
    return float(np.mean(np.abs(prob_true - prob_pred)))


# Monitor daily
def monitor_calibration(daily_data: dict, threshold: float = 0.1) -> None:
    """
    daily_data: {"date": "2026-05-27", "probs": [...], "labels": [...]}
    Chỉ chạy khi đã có label ground truth cho ngày đó.
    """
    ece = expected_calibration_error(
        np.array(daily_data["probs"]),
        np.array(daily_data["labels"]),
    )
    print(f"ECE [{daily_data['date']}]: {ece:.4f}")
    if ece > threshold:
        print(f"  WARNING: calibration drift detected (ECE={ece:.4f} > {threshold})")

Confidence histogram shift

Ngay cả khi chưa có label, histogram của model output probability cũng có thể dùng làm tín hiệu:

import numpy as np
from scipy import stats

def confidence_histogram_drift(
    probs_baseline: np.ndarray,
    probs_recent: np.ndarray,
    bins: int = 20,
) -> dict:
    """
    So sánh histogram confidence score giữa baseline và recent period.
    Dùng KS test — không cần label.
    """
    ks_stat, p_value = stats.ks_2samp(probs_baseline, probs_recent)
    return {
        "ks_stat": round(ks_stat, 4),
        "p_value": round(p_value, 4),
        "drift_detected": p_value < 0.05,
        "mean_baseline": round(float(np.mean(probs_baseline)), 4),
        "mean_recent": round(float(np.mean(probs_recent)), 4),
    }

Shift trong confidence histogram không xác nhận được là concept drift hay data drift — nhưng là tín hiệu đáng điều tra, đặc biệt khi kết hợp với data drift monitor từ bài 47.

Label delay là lý do concept drift khó detect hơn data drift. Một số use case điển hình:

• Credit default: Dự đoán default trong 90 ngày → label đến sau 90 ngày. Không thể chờ 90 ngày mới biết model đang drift.
• Churn prediction: Churn được xác nhận sau 30 ngày không active → label delay 30 ngày.
• Medical diagnosis: Label chính xác (confirmed bởi bác sĩ) có thể mất vài ngày đến vài tuần.
• Content moderation: Human review có backlog — label đến sau vài giờ đến vài ngày.

Workaround 1 — Leading indicator (proxy metric)

Thay vì chờ label chậm, dùng metric có thể đo được sớm hơn:

• Churn: thay vì chờ "user đã churn" (30 ngày), dùng "session frequency decline trong 7 ngày" làm proxy.
• Fraud: thay vì chờ confirmed fraud report, theo dõi transaction velocity anomaly.
• LLM quality: thay vì chờ human eval, theo dõi user thumbs-down rate và session abandonment ngay sau response.

Proxy không thay thế được ground truth label nhưng cho tín hiệu sớm hơn nhiều.

Workaround 2 — Sample subset cho manual labeling

Chọn ngẫu nhiên subset nhỏ (vd 1–5% traffic) và ưu tiên human label nhanh. Với subset này, tính performance metric ngay thay vì chờ toàn bộ label tự nhiên đến.

import random

def should_fast_label(sample_rate: float = 0.02) -> bool:
    """2% traffic được ưu tiên label nhanh để monitor performance."""
    return random.random() < sample_rate

Workaround 3 — Active learning priority

Ưu tiên label các sample mà model uncertain nhất (prediction probability gần 0.5 với binary classification, hoặc max softmax thấp với multi-class). Những sample này có nhiều thông tin nhất về vùng boundary, và thường là nơi concept drift biểu hiện sớm nhất.

def get_uncertain_samples(
    predictions: list[dict],
    top_k: int = 100,
) -> list[dict]:
    """
    predictions: list of {"id": ..., "prob": float}
    Trả về top_k sample gần ngưỡng 0.5 nhất để ưu tiên label.
    """
    scored = [
        {**p, "uncertainty": abs(p["prob"] - 0.5)}
        for p in predictions
    ]
    # Sắp xếp: uncertainty nhỏ nhất = gần 0.5 nhất = uncertain nhất
    scored.sort(key=lambda x: x["uncertainty"])
    return scored[:top_k]

Workaround 4 — LLM-as-judge cho text task

Với task text generation, có thể dùng LLM mạnh hơn (GPT-4o, Claude 3.5 Sonnet) để đánh giá output của model nhỏ hơn đang chạy production. Không cần chờ human label cho từng response.

JUDGE_PROMPT = """Evaluate the following AI response for quality.
Question: {question}
Response: {response}

Rate on a scale 1-5 where:
1 = completely wrong or irrelevant
3 = partially correct
5 = accurate and helpful

Return JSON: {{"score": int, "reason": str}}"""

def llm_judge_score(question: str, response: str) -> dict:
    """Dùng LLM mạnh hơn để chấm response của model production."""
    result = judge_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": JUDGE_PROMPT.format(
            question=question, response=response
        )}],
        response_format={"type": "json_object"},
    )
    import json
    return json.loads(result.choices[0].message.content)

LLM-as-judge có bias và không ổn định giữa các lần chạy. Cần calibrate judge metric với human eval trước khi dùng làm signal chính thức. Phù hợp nhất là phát hiện sự thay đổi trong trend, không phải đánh giá absolute quality.

Thay vì monitor mọi prediction riêng lẻ, gom request theo bucket thời gian và tính metric per bucket:

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from sklearn.metrics import accuracy_score, f1_score
import numpy as np

@dataclass
class MetricBucket:
    start_time: datetime
    end_time: datetime
    predictions: list = field(default_factory=list)   # model output
    labels: list = field(default_factory=list)         # ground truth (khi có)
    probabilities: list = field(default_factory=list)  # confidence scores

    def compute_metrics(self) -> dict | None:
        if not self.labels:
            return None  # Label chưa có
        return {
            "bucket_start": self.start_time.isoformat(),
            "n_samples": len(self.labels),
            "accuracy": round(accuracy_score(self.labels, self.predictions), 4),
            "f1": round(f1_score(self.labels, self.predictions, average="weighted"), 4),
            "mean_confidence": round(float(np.mean(self.probabilities)), 4),
        }


class BucketedDriftMonitor:
    """Monitor concept drift bằng cách gom request theo bucket 1 ngày."""

    def __init__(self, bucket_size_hours: int = 24):
        self.bucket_size = timedelta(hours=bucket_size_hours)
        self.buckets: list[MetricBucket] = []
        self.current_bucket: MetricBucket | None = None

    def _get_or_create_bucket(self, timestamp: datetime) -> MetricBucket:
        if (
            self.current_bucket is None
            or timestamp >= self.current_bucket.end_time
        ):
            start = timestamp.replace(
                minute=0, second=0, microsecond=0
            )
            self.current_bucket = MetricBucket(
                start_time=start,
                end_time=start + self.bucket_size,
            )
            self.buckets.append(self.current_bucket)
        return self.current_bucket

    def log_prediction(
        self, timestamp: datetime, prediction: int, probability: float
    ) -> None:
        bucket = self._get_or_create_bucket(timestamp)
        bucket.predictions.append(prediction)
        bucket.probabilities.append(probability)

    def log_label(self, timestamp: datetime, label: int) -> None:
        """Gọi khi label ground truth đến (có thể muộn hơn prediction)."""
        for bucket in self.buckets:
            if bucket.start_time <= timestamp < bucket.end_time:
                bucket.labels.append(label)
                return

    def get_recent_metrics(self, last_n_buckets: int = 7) -> list[dict]:
        metrics = []
        for bucket in self.buckets[-last_n_buckets:]:
            m = bucket.compute_metrics()
            if m:
                metrics.append(m)
        return metrics

Eyeballing vs statistical test

Với ít bucket (7–14 ngày), plot time series metric ra chart và eyeball trend thường đủ. Với monitoring automated, dùng statistical test:

from scipy import stats

def detect_drift_across_buckets(
    recent_metrics: list[float],
    baseline_metrics: list[float],
    alpha: float = 0.05,
) -> bool:
    """
    recent_metrics, baseline_metrics: list accuracy hoặc F1 theo ngày.
    Trả về True nếu có drift đáng kể theo T-test.
    """
    if len(recent_metrics) < 3 or len(baseline_metrics) < 3:
        return False  # Không đủ data để test
    _, p_value = stats.ttest_ind(baseline_metrics, recent_metrics)
    return p_value < alpha

LLM và RAG system có những loại drift đặc thù không tồn tại ở classical ML:

Answer quality drift

Chất lượng câu trả lời của LLM thay đổi qua thời gian — do knowledge cutoff, do thay đổi topic distribution của user query, hoặc do retrieval quality trong RAG bị giảm khi knowledge base không được cập nhật.

Measure: human eval sampling (tốn nhân lực) hoặc LLM-as-judge (Section 10). Track score trung bình theo tuần.

Hallucination rate drift

Tỷ lệ response chứa thông tin sai lệch hoặc factually incorrect tăng dần, thường xảy ra khi:

• Topic user query drift vào vùng model yếu (coverage gap).
• RAG knowledge base outdated — model LLM trả lời theo pretrained knowledge thay vì retrieved context.
• Context window bị overflow làm model mất retrieved context.

Detect: track fact-check score bằng LLM judge, hoặc theo dõi pattern citation không có trong retrieved documents.

Tool call success rate

Trong agent system, tỷ lệ tool call thành công là metric trực tiếp và không cần label:

from collections import defaultdict
from datetime import datetime, date

tool_stats: dict[str, dict] = defaultdict(lambda: {"success": 0, "fail": 0})

def log_tool_call(tool_name: str, success: bool, day: date) -> None:
    key = f"{tool_name}_{day.isoformat()}"
    if success:
        tool_stats[key]["success"] += 1
    else:
        tool_stats[key]["fail"] += 1

def tool_success_rate(tool_name: str, day: date) -> float | None:
    key = f"{tool_name}_{day.isoformat()}"
    stats = tool_stats[key]
    total = stats["success"] + stats["fail"]
    if total == 0:
        return None
    return stats["success"] / total

User satisfaction signal

Các signal implicit từ user behavior:

• Thumbs up/down rate: Tỷ lệ explicit feedback nếu UI có nút đánh giá.
• Retry rate: % user gửi lại cùng query hoặc query tương tự ngay sau → response trước đó không đủ.
• Session abandonment: User đóng cửa sổ ngay sau response → tín hiệu response không hữu ích.
• Copy rate: % response được user copy (tín hiệu dương).

Những signal này không phải label chính xác nhưng là leading indicator nhanh và không cần human review.

Một nguồn concept drift thường bị bỏ sót: chính team deploy tạo ra drift khi thay đổi system prompt, model version, hoặc retrieval strategy mà không track đầy đủ.

Track prompt version và model version bắt buộc

import hashlib
from datetime import datetime

def get_prompt_version(system_prompt: str) -> str:
    """Trả về hash ngắn của system prompt để dùng làm version ID."""
    return hashlib.sha256(system_prompt.encode()).hexdigest()[:12]

# Log mỗi inference kèm version metadata
def log_inference(
    request_id: str,
    model_name: str,
    model_version: str,
    prompt_version: str,
    latency_ms: float,
) -> None:
    import logging
    logger = logging.getLogger("app.inference")
    logger.info("inference", extra={
        "request_id": request_id,
        "model_name": model_name,
        "model_version": model_version,
        "prompt_version": prompt_version,
        "latency_ms": round(latency_ms, 2),
        "timestamp": datetime.utcnow().isoformat(),
    })

Khi thấy metric drift, câu hỏi đầu tiên là: "Có deploy gì mới không?". Nếu log có model_version và prompt_version, bạn có thể group metric theo version và xem drift bắt đầu từ version nào.

Canary deploy để isolate drift

Khi thay đổi model hoặc prompt, deploy canary trước — chỉ 5–10% traffic nhận version mới. So sánh metric giữa canary và control trước khi rollout toàn bộ:

import random

def route_to_version(canary_fraction: float = 0.1) -> str:
    """
    Trả về "canary" hoặc "stable" dựa trên fraction.
    Dùng sticky session (user_id hash) nếu cần user nhất quán 1 version.
    """
    return "canary" if random.random() < canary_fraction else "stable"

Canary deploy cần đủ sample size để kết quả có ý nghĩa thống kê. Với metric phân tán cao (LLM quality score), cần sample size lớn hơn (vài nghìn request per version thay vì vài trăm).

A/B test sau khi detect drift

Khi đã retrain model mới để xử lý drift, dùng A/B test để xác nhận version mới thực sự tốt hơn — không chỉ dựa vào offline evaluation:

• Treatment: model mới sau retrain.
• Control: model cũ.
• Chạy đủ N ngày để cover variability ngày/tuần.
• Check statistical significance trước khi rollout toàn bộ.

Trong thực tế, kết hợp nhiều method: proxy metric để alert sớm, performance metric để confirm khi label có, calibration để cảnh báo trung gian.

Detect được drift là bước đầu. Việc cần làm tiếp theo phụ thuộc vào nguyên nhân và loại drift:

Bước 1: Điều tra trước khi action

Không panic retrain ngay khi thấy alert. Trước tiên:

• Có deploy version mới gần đây không? (kiểm tra model_version + prompt_version log)
• Có sự kiện bên ngoài không? (sự kiện kinh tế, holiday, viral content)
• Có data pipeline issue không? (feature encoding đổi, missing value xử lý khác)
• Data drift có đi kèm không? (bài 47) — cùng lúc P(X) và P(Y|X) đổi?

Retrain với data mới

Phù hợp khi có đủ labeled data mới và drift là gradual hoặc sudden rõ ràng. Không cần train lại từ đầu — fine-tune trên data gần nhất (recent data có weight cao hơn) thường đủ.

from sklearn.utils import compute_sample_weight
import numpy as np

def time_decay_weights(
    timestamps: list[float],
    half_life_days: float = 30.0,
) -> np.ndarray:
    """
    Tính sample weight theo thời gian — data gần nhất có weight cao hơn.
    timestamps: Unix timestamp của mỗi sample.
    """
    now = max(timestamps)
    age_days = (now - np.array(timestamps)) / 86400
    # Exponential decay: weight = 0.5^(age_days / half_life_days)
    return np.power(0.5, age_days / half_life_days)

# Dùng với sklearn estimator
weights = time_decay_weights(sample_timestamps)
model.fit(X_train, y_train, sample_weight=weights)

Online learning với river

Với incremental drift liên tục, online learning phù hợp hơn periodic batch retrain. Model update mỗi khi có label mới:

from river import linear_model, metrics, drift, preprocessing

# Pipeline: scaler + logistic regression, update từng sample
model = preprocessing.StandardScaler() | linear_model.LogisticRegression()
metric = metrics.Accuracy()
ddm = drift.DDM()

for x, y in live_stream:
    # Predict trước khi learn (production pattern)
    y_pred = model.predict_one(x)

    # Update model với label mới
    model.learn_one(x, y)

    # Track performance
    metric.update(y, y_pred)
    ddm.update(int(y != y_pred))

    if ddm.drift_detected:
        print(f"Drift detected, resetting metric tracker. Accuracy was: {metric}")
        metric = metrics.Accuracy()  # Reset metric window
        ddm = drift.DDM()
        # Optionally: reset model weights, tăng learning rate, alert team

Rollback prompt hoặc model

Nếu drift xảy ra sau một deploy và metric tệ hơn, rollback về version trước nhanh nhất — sau đó mới phân tích tiếp offline.

Switch fallback

Với classification task quan trọng, có thể có model dự phòng đơn giản hơn (rule-based hoặc model nhỏ hơn, ít bị overfit vào training distribution). Khi model chính bị drift mạnh, fallback sang dự phòng trong lúc chờ retrain xong.

Nhầm data drift với concept drift

Khi thấy performance giảm và feature distribution đổi cùng lúc, dễ kết luận là data drift. Nhưng có thể là cả hai xảy ra song song. Nếu chỉ fix data drift mà bỏ sót concept drift, model vẫn không được cải thiện. Cần phân tách: normalize data distribution trước, rồi kiểm tra performance có hồi phục không.

Retrain liên tục khi alert

Retrain mỗi khi có alert mà không điều tra nguyên nhân có thể dẫn đến catastrophic forgetting (model quên data cũ quan trọng), overfit vào noise ngắn hạn, và vòng lặp retrain vô ích. Đặt ngưỡng alert cao đủ để tránh false alarm, và luôn validate trên held-out test set trước khi deploy model mới.

Test offline nhưng production khác

Model được test trên data có label sẵn (in-distribution) cho thấy performance tốt. Nhưng trong production, model gặp feature combination chưa từng thấy — test nội bộ không phát hiện được vì test set thiếu các vùng phân phối mới này. Giải pháp: track distribution của test set vs production data song song.

Quên track version khi deploy

Deploy model mới hoặc đổi system prompt mà không log version → khi thấy metric thay đổi sau đó không biết là do thế giới bên ngoài hay do chính mình. Luôn log model_version và prompt_version kèm mỗi inference từ ngày đầu.

Threshold alert quá nhạy — alert fatigue

Threshold quá nhỏ dẫn đến alert liên tục với noise tự nhiên của metric (biến động ngày/tuần). Team dần bỏ qua alert → khi có drift thật sự cũng không được xử lý kịp. Đặt threshold dựa trên baseline variability: nếu F1 tự nhiên dao động ±0.02, threshold drift nên ở 0.05+.

Nhầm seasonal drift với concept drift thật

Performance giảm vào Q4 mỗi năm do seasonality không phải concept drift cần xử lý. So sánh với cùng kỳ năm trước, không chỉ so với kỳ trước ngay đó.

• ✅ Concept drift = P(Y|X) thay đổi; khác data drift (P(X)) ở chỗ cần label ground truth để detect trực tiếp
• ✅ 4 loại theo tốc độ: sudden, gradual, incremental, recurring — mỗi loại cần strategy detect và xử lý khác nhau
• ✅ DDM và ADWIN từ thư viện river cho streaming detection khi có label real-time
• ✅ ECE (Expected Calibration Error) và confidence histogram shift là proxy signal khi label bị delay
• ✅ Label delay là vấn đề cốt lõi: giải quyết bằng proxy metric, sample labeling, active learning, hoặc LLM-as-judge
• ✅ Gom request theo bucket thời gian + T-test để detect gradual performance degradation
• ✅ LLM/RAG có drift đặc thù: answer quality, hallucination rate, tool success rate, user satisfaction
• ✅ Luôn log model_version + prompt_version — drift có thể do chính mình tạo ra khi deploy
• ✅ Canary deploy trước khi rollout toàn bộ; A/B test để confirm model mới sau retrain thực sự tốt hơn
• ✅ Điều tra nguyên nhân trước khi retrain — tránh catastrophic forgetting và overfit vào noise

Bài 49: Monitoring với Prometheus + Grafana — chuyển từ logging và drift detection sang hệ thống metrics tập trung: expose metrics bằng Prometheus, visualize và alert bằng Grafana.