Is Your LLM Quietly Getting Worse? Build a Tiny Drift Detector for Your AI Features
Tuần này trên Hacker News có một thread hơn 800 điểm, xoay quanh đúng một câu hỏi: "Model này đã bị nerf chưa?". Ai từng đưa tính năng AI lên production chắc cũng gặp cảnh này: hôm qua prompt chạy ngon, hôm nay output tự dưng dài hơn, lệch format JSON, hoặc từ chối những request trước đó vẫn xử lý bình thường. Provider có thể đổi model phía sau, đổi system prompt mặc định, đổi quantization hay routing, và bạn thường không được báo trước. Nhiều team lúc đó chỉ biết đoán, hoặc lên Twitter xem có ai than giống mình không.
Bài này hướng dẫn dựng một drift detector nhỏ, khoảng 150 dòng Python và một cron job, để trả lời câu hỏi đó bằng số liệu thay vì cảm giác. Mình đã chạy cách này cho vài dự án chatbot và pipeline extract dữ liệu. Nó bắt được regression sớm hơn user ít nhất vài lần.
Vì sao unit test thông thường không đủ
Với code thường, f(x) luôn trả về cùng một kết quả. Với LLM thì không: cùng input, output có thể khác nhau giữa hai lần gọi, kể cả khi đặt temperature=0. Vì vậy test kiểu assert output == "..." sẽ fail liên tục (flaky), rồi team tắt nó đi luôn.
Thứ cần đo không phải output có giống hệt hôm qua không, mà là phân phối chất lượng có dịch chuyển không. Cụ thể có 4 nhóm tín hiệu:
- Format compliance: tỉ lệ output parse được JSON, đúng schema.
-
- Task accuracy: với các case có đáp án rõ (phân loại, extract field), tỉ lệ đúng là bao nhiêu.
-
- Behavior: tỉ lệ từ chối (refusal), độ dài output trung bình.
-
- Operational: latency p50/p95, số token tiêu thụ.
Chỉ số nào lệch khỏi baseline quá ngưỡng thì báo động.
flowchart LR
A[Golden set<br/>50-200 cases] --> B[Runner<br/>cron mỗi 6h]
B --> C[LLM API]
C --> D[Scorer<br/>format, accuracy, refusal]
D --> E[(SQLite<br/>lịch sử metrics)]
E --> F{So với baseline}
F -->|Lệch > ngưỡng| G[Alert Slack/TG]
F -->|OK| H[Không làm gì]
```
## Bước 1: Xây golden set từ traffic thật
Đừng tự nghĩ test case ngồi trong phòng. Hãy lấy từ log production, vì đó là thứ user thật sự gửi lên. Mình thường chọn khoảng 100 case theo tỉ lệ: 60% case phổ biến, 30% edge case từng gây bug, 10% case "nhạy cảm" dễ bị model từ chối.
Mỗi case là một dòng JSONL:
```json
{"id": "inv-012", "prompt": "Trích xuất số hóa đơn và tổng tiền: ...", "expect": {"invoice_no": "HD-2291", "total": 1250000}, "type": "extract"}
{"id": "cls-044", "prompt": "Phân loại ticket: 'App bị crash khi thanh toán MoMo'", "expect": {"label": "payment_bug"}, "type": "classify"}
```
Mẹo: nhớ mask dữ liệu cá nhân (số điện thoại, CCCD) trước khi đưa vào golden set, nhất là khi file nằm trong repo.
## Bước 2: Runner và scorer
Đoạn code dưới đây dùng `httpx` 0.27 và gọi một endpoint theo chuẩn OpenAI-compatible, nên đổi provider hay chạy model local qua vLLM/Ollama đều được. Mỗi case chạy 3 lần để giảm nhiễu.
```python
# drift_runner.py - Python 3.11+
import json, time, sqlite3, statistics, os
import httpx
API_URL = os.environ["LLM_API_URL"]
API_KEY = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "default")
REFUSAL_MARKERS = ("i can't", "i cannot", "tôi không thể", "xin lỗi, tôi")
def call_llm(client, prompt):
t0 = time.perf_counter()
r = client.post(API_URL, headers={"Authorization": f"Bearer {API_KEY}"}, json={
"model": MODEL,
"temperature": 0,
"messages": [
{"role": "system", "content": "Chỉ trả về JSON hợp lệ."},
{"role": "user", "content": prompt},
],
}, timeout=60)
r.raise_for_status()
data = r.json()
return data["choices"][0]["message"]["content"], time.perf_counter() - t0
def score(case, text):
s = {"parsed": 0, "correct": 0, "refused": 0, "length": len(text)}
if any(m in text.lower() for m in REFUSAL_MARKERS):
s["refused"] = 1
return s
try:
out = json.loads(text.strip().removeprefix("```json").removesuffix("```"))
s["parsed"] = 1
s["correct"] = int(all(out.get(k) == v for k, v in case["expect"].items()))
except (json.JSONDecodeError, AttributeError):
pass
return s
def run(golden_path="golden.jsonl", repeats=3):
cases = [json.loads(l) for l in open(golden_path, encoding="utf-8")]
rows, latencies = [], []
with httpx.Client() as client:
for case in cases:
for _ in range(repeats):
text, lat = call_llm(client, case["prompt"])
rows.append(score(case, text))
latencies.append(lat)
n = len(rows)
metrics = {
"parse_rate": sum(r["parsed"] for r in rows) / n,
"accuracy": sum(r["correct"] for r in rows) / n,
"refusal_rate": sum(r["refused"] for r in rows) / n,
"avg_len": statistics.mean(r["length"] for r in rows),
"p95_latency": statistics.quantiles(latencies, n=20)[18],
}
db = sqlite3.connect("drift.db")
db.execute("CREATE TABLE IF NOT EXISTS runs (ts INTEGER, model TEXT, metrics TEXT)")
db.execute("INSERT INTO runs VALUES (?, ?, ?)", (int(time.time()), MODEL, json.dumps(metrics)))
db.commit()
return metrics
if __name__ == "__main__":
print(json.dumps(run(), indent=2))
```
Có hai điểm đáng lưu ý. Thứ nhất, refusal detection bằng keyword khá thô nhưng đủ dùng lúc đầu, về sau có thể thay bằng một classifier nhỏ. Thứ hai, `avg_len` nghe vô hại nhưng lại là tín hiệu rất nhạy: model bị đổi phía sau thường nói dài hơn hoặc ngắn hơn hẳn, trước cả khi accuracy kịp tụt.
## Bước 3: So baseline và bắn alert
Baseline nên là **median của 7 ngày gần nhất**, không phải lần chạy hôm qua, để một lần chạy xấu ngẫu nhiên không kéo baseline lệch theo. Ngưỡng mình hay đặt: accuracy giảm hơn 5 điểm %, parse_rate giảm hơn 3 điểm %, refusal_rate tăng gấp đôi, avg_len lệch quá 30%.
```mermaid
sequenceDiagram
participant Cron
participant Runner
participant DB as SQLite
participant Alert
Cron->>Runner: chạy mỗi 6h
Runner->>DB: ghi metrics mới
Runner->>DB: lấy median 7 ngày
Runner->>Runner: so sánh với ngưỡng
alt có metric lệch
Runner->>Alert: gửi diff chi tiết
end
```
```python
# drift_check.py
import json, sqlite3, statistics, time, sys
RULES = {
"accuracy": lambda base, now: base - now > 0.05,
"parse_rate": lambda base, now: base - now > 0.03,
"refusal_rate": lambda base, now: now > max(base * 2, 0.02),
"avg_len": lambda base, now: abs(now - base) / base > 0.30,
}
def check(db_path="drift.db"):
db = sqlite3.connect(db_path)
week_ago = int(time.time()) - 7 * 86400
rows = [json.loads(m) for (m,) in db.execute(
"SELECT metrics FROM runs WHERE ts > ? ORDER BY ts", (week_ago,))]
if len(rows) < 5:
return []
latest, history = rows[-1], rows[:-1]
alerts = []
for key, is_bad in RULES.items():
base = statistics.median(r[key] for r in history)
if is_bad(base, latest[key]):
alerts.append(f"{key}: baseline={base:.3f} now={latest[key]:.3f}")
return alerts
if __name__ == "__main__":
alerts = check()
if alerts:
print("DRIFT DETECTED\n" + "\n".join(alerts))
sys.exit(1)
```
Phần lên lịch thì một crontab là đủ. Không cần Airflow cho việc này:
```bash
# crontab -e
0 */6 * * * cd /opt/drift && ./venv/bin/python drift_runner.py >> run.log 2>&1 \
&& ./venv/bin/python drift_check.py > alert.txt \
|| curl -s -X POST "$SLACK_WEBHOOK" -H 'Content-Type: application/json' \
-d "{\"text\": \"$(cat alert.txt | sed 's/\"/\\\"/g')\"}"
```
Nếu team đã dùng GitHub Actions thì dùng `schedule: - cron: '0 */6 * * *'` với cùng logic, để exit code 1 làm fail workflow và nhận notification có sẵn.
## Những cái bẫy mình đã dính
- **Chi phí**: 100 case × 3 lần × 4 lượt/ngày = 1.200 request/ngày. Với model lớn, con số này không nhỏ. Hãy tính trước, hoặc chỉ chạy full set mỗi ngày một lần và chạy subset 20 case mỗi 6h.
- **Golden set bị mục**: sau vài tháng, traffic thật thay đổi mà golden set vẫn đứng yên. Mỗi quý nên bổ sung case mới từ log.
- **Nhầm lẫn với lỗi của chính mình**: có lần alert kêu inh ỏi, hóa ra là do team đổi system prompt mà quên chạy lại baseline. Nên log luôn hash của prompt template vào bảng `runs`, và reset baseline mỗi khi hash đổi.
- **Pin version**: nếu provider cho chọn model snapshot có ngày cụ thể thay vì alias kiểu `latest`, hãy pin lại. Drift detector lúc đó chủ yếu dùng để bắt thay đổi hạ tầng, và làm test khi bạn chủ động nâng version.
## Kết luận
Nghi ngờ model "bị nerf" là chuyện bình thường, nhưng đoán mò thì không nên. Checklist để bắt đầu ngay trong tuần này:
1. **Trích 50-100 case từ log production**, mask PII, lưu thành `golden.jsonl` có trường `expect`.
2. **Đo 4 nhóm chỉ số**: parse rate, accuracy, refusal rate, độ dài output (cộng thêm latency nếu quan tâm SLA).
3. **So với median 7 ngày**, không so với lần chạy trước, để tránh báo động giả.
4. **Chạy bằng cron hoặc GitHub Actions** mỗi 6-24h, fail thì bắn alert ra kênh chat của team.
5. **Log hash của prompt template** để phân biệt model đổi hay team tự đổi.
6. **Pin model version** nếu có thể, và luôn chạy drift check trước khi nâng cấp.
Khoảng 150 dòng code, một file SQLite và một dòng crontab là đủ để lần sau có người hỏi "model có tệ đi không?", bạn mở dashboard ra trả lời bằng số liệu thay vì ngồi lướt Hacker News.All rights reserved
