A/B Testing ML Production Setup Guide — ระบบทดสอบ ML Model ใน Production

A/B Testing สำหรับ ML Model คืออะไร

A/B Testing ในบริบทของ Machine Learning Production คือการทดสอบ ML Model หลายเวอร์ชันพร้อมกันโดยแบ่ง traffic จริงจากผู้ใช้ไปยังแต่ละ model แล้ววัดผลลัพธ์ด้วยวิธีทางสถิติเพื่อตัดสินใจว่า model ไหนทำงานได้ดีกว่า

ความแตกต่างจาก A/B Testing ของเว็บไซต์ทั่วไปคือ ML A/B Testing ต้องพิจารณา metrics เฉพาะทางเช่น Prediction Accuracy, Latency, Model Drift และ Business Metrics เช่น Conversion Rate หรือ Revenue per User นอกจากนี้ยังต้องจัดการกับ Data Feedback Loop ที่ผลจากการ predict อาจส่งผลกระทบต่อข้อมูลที่ใช้ train model ต่อไป

ระบบ A/B Testing สำหรับ ML Production ประกอบด้วย 4 ส่วนหลักคือ Model Registry ที่เก็บ model ทุกเวอร์ชัน, Model Serving Layer ที่ serve prediction requests, Traffic Router ที่แบ่ง traffic ระหว่าง model versions และ Metrics Collection ที่เก็บผลลัพธ์สำหรับวิเคราะห์ทางสถิติ

การออกแบบ A/B Test ที่ดีต้องกำหนด Hypothesis ให้ชัดเจน เลือก Primary Metric ที่สะท้อน business value เลือก Sample Size ที่มากพอสำหรับ Statistical Significance และกำหนดระยะเวลาที่เหมาะสมสำหรับการทดสอบ

สถาปัตยกรรมระบบ A/B Testing สำหรับ ML Production

สถาปัตยกรรมที่แนะนำใช้ Kubernetes เป็น platform หลักร่วมกับ Seldon Core สำหรับ model serving และ Istio สำหรับ traffic management

# สถาปัตยกรรมของระบบ
# Client → Istio Gateway → VirtualService (Traffic Split)
#                              ├── Model A (v1) - 80% traffic
#                              └── Model B (v2) - 20% traffic
#                                      ↓
#                              Prometheus (Metrics)
#                                      ↓
#                              Grafana (Dashboard)
#                                      ↓
#                              Statistical Analysis Job
#                                      ↓
#                              MLflow (Model Registry)

# ติดตั้ง Seldon Core บน Kubernetes
helm repo add seldon https://storage.googleapis.com/seldon-charts
helm repo update

kubectl create namespace seldon-system
helm install seldon-core seldon/seldon-core-operator \
  --namespace seldon-system \
  --set usageMetrics.enabled=true \
  --set istio.enabled=true \
  --set istio.gateway=istio-system/seldon-gateway

# ตรวจสอบการติดตั้ง
kubectl get pods -n seldon-system
# NAME                                        READY   STATUS
# seldon-controller-manager-xxx-yyy            1/1     Running

# ติดตั้ง Istio (ถ้ายังไม่มี)
istioctl install --set profile=default -y
kubectl label namespace default istio-injection=enabled

ตั้งค่า MLflow เป็น Model Registry สำหรับเก็บและจัดการ model versions

# docker-compose.yml สำหรับ MLflow
version: "3.9"
services:
  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.12.1
    ports:
      - "5000:5000"
    environment:
      MLFLOW_BACKEND_STORE_URI: postgresql://mlflow:password@postgres:5432/mlflow
      MLFLOW_DEFAULT_ARTIFACT_ROOT: s3://mlflow-artifacts/
      AWS_ACCESS_KEY_ID: 
      AWS_SECRET_ACCESS_KEY: 
    command: >
      mlflow server
      --host 0.0.0.0
      --port 5000
      --backend-store-uri postgresql://mlflow:password@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: mlflow
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: password
    volumes:
      - mlflow_db:/var/lib/postgresql/data

volumes:
  mlflow_db:

# Register model version
# python3 -c "
# import mlflow
# mlflow.set_tracking_uri('http://localhost:5000')
# mlflow.register_model('runs:/RUN_ID/model', 'recommendation-model')
# "

ติดตั้ง ML Model Serving ด้วย Seldon Core

สร้าง SeldonDeployment ที่ serve model 2 เวอร์ชันพร้อมกันสำหรับ A/B Testing

# ab-test-deployment.yaml
apiVersion: machinelearning.seldon.io/v1
kind: SeldonDeployment
metadata:
  name: recommendation-ab-test
  namespace: default
spec:
  predictors:
    - name: model-a
      replicas: 3
      traffic: 80
      annotations:
        seldon.io/engine-separate-pod: "true"
      componentSpecs:
        - spec:
            containers:
              - name: model-a
                image: myregistry/recommendation-model:v1.2.0
                resources:
                  requests:
                    memory: "1Gi"
                    cpu: "500m"
                  limits:
                    memory: "2Gi"
                    cpu: "1000m"
                env:
                  - name: MODEL_VERSION
                    value: "v1.2.0"
                  - name: MLFLOW_TRACKING_URI
                    value: "http://mlflow:5000"
                readinessProbe:
                  httpGet:
                    path: /health/status
                    port: 9000
                  initialDelaySeconds: 10
      graph:
        name: model-a
        type: MODEL
        endpoint:
          type: REST

    - name: model-b
      replicas: 2
      traffic: 20
      componentSpecs:
        - spec:
            containers:
              - name: model-b
                image: myregistry/recommendation-model:v2.0.0
                resources:
                  requests:
                    memory: "1Gi"
                    cpu: "500m"
                  limits:
                    memory: "2Gi"
                    cpu: "1000m"
                env:
                  - name: MODEL_VERSION
                    value: "v2.0.0"
                  - name: MLFLOW_TRACKING_URI
                    value: "http://mlflow:5000"
      graph:
        name: model-b
        type: MODEL
        endpoint:
          type: REST

# Deploy
# kubectl apply -f ab-test-deployment.yaml

# ตรวจสอบสถานะ
# kubectl get seldondeployments
# kubectl get pods -l seldon-deployment-id=recommendation-ab-test

ทดสอบ prediction endpoint

# ส่ง prediction request
curl -X POST "http://istio-gateway/seldon/default/recommendation-ab-test/api/v1.0/predictions" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "ndarray": [[25, "male", "electronics", 3, 150.0]]
    }
  }'

# Response จะมี metadata บอกว่าถูก route ไป model ไหน
# {
#   "data": {"ndarray": [[0.85, 0.12, 0.03]]},
#   "meta": {
#     "routing": {"model-a": 1},
#     "requestPath": {"model-a": "myregistry/recommendation-model:v1.2.0"}
#   }
# }

# ทดสอบด้วย load test เพื่อตรวจสอบ traffic distribution
for i in $(seq 1 100); do
  curl -s -X POST "http://istio-gateway/seldon/default/recommendation-ab-test/api/v1.0/predictions" \
    -H "Content-Type: application/json" \
    -d '{"data":{"ndarray":[[25,"male","electronics",3,150.0]]}}' | \
    python3 -c "import sys, json; d=json.load(sys.stdin); print(list(d['meta']['routing'].keys())[0])"
done | sort | uniq -c
# Output:
#   81 model-a
#   19 model-b

ตั้งค่า Traffic Splitting ระหว่าง Model Version

ใช้ Istio VirtualService สำหรับ traffic management ที่ละเอียดกว่า Seldon built-in routing

# istio-virtual-service.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: recommendation-ab-test
spec:
  hosts:
    - recommendation-ab-test.default.svc.cluster.local
  http:
    - match:
        - headers:
            x-model-version:
              exact: "v2"
      route:
        - destination:
            host: recommendation-ab-test-model-b
            port:
              number: 8000
    - route:
        - destination:
            host: recommendation-ab-test-model-a
            port:
              number: 8000
          weight: 80
        - destination:
            host: recommendation-ab-test-model-b
            port:
              number: 8000
          weight: 20
---
# DestinationRule สำหรับ circuit breaking
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: recommendation-circuit-breaker
spec:
  host: recommendation-ab-test-model-b
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: DEFAULT
        http1MaxPendingRequests: 50
        http2MaxRequests: 100
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 60s
      maxEjectionPercent: 50

สร้าง Script สำหรับปรับ traffic weight แบบ gradual

#!/usr/bin/env python3
# adjust_traffic.py — ปรับ traffic split แบบ gradual
import subprocess
import json
import time
import sys

def get_current_weights():
    result = subprocess.run(
        ["kubectl", "get", "seldondeployment", "recommendation-ab-test",
         "-o", "json"],
        capture_output=True, text=True
    )
    spec = json.loads(result.stdout)
    predictors = spec["spec"]["predictors"]
    return {p["name"]: p["traffic"] for p in predictors}

def set_weights(model_a_weight, model_b_weight):
    patch = json.dumps({
        "spec": {
            "predictors": [
                {"name": "model-a", "traffic": model_a_weight},
                {"name": "model-b", "traffic": model_b_weight}
            ]
        }
    })
    subprocess.run([
        "kubectl", "patch", "seldondeployment", "recommendation-ab-test",
        "--type=merge", "-p", patch
    ])
    print(f"Updated: model-a={model_a_weight}%, model-b={model_b_weight}%")

def gradual_increase(target_b_weight, step=10, interval_minutes=30):
    current = get_current_weights()
    current_b = current.get("model-b", 20)

    while current_b < target_b_weight:
        current_b = min(current_b + step, target_b_weight)
        current_a = 100 - current_b
        set_weights(current_a, current_b)
        if current_b < target_b_weight:
            print(f"Waiting {interval_minutes} minutes before next increase...")
            time.sleep(interval_minutes * 60)

if __name__ == "__main__":
    target = int(sys.argv[1]) if len(sys.argv) > 1 else 50
    gradual_increase(target)

เก็บ Metrics และวิเคราะห์ผลด้วย Statistical Test

ตั้งค่า Prometheus เพื่อเก็บ metrics จาก Seldon Core และสร้าง Statistical Analysis Job

# prometheus-seldon-rules.yaml
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: seldon-ab-test-rules
spec:
  groups:
    - name: seldon-ab-test
      rules:
        - record: seldon_model_prediction_latency_p99
          expr: |
            histogram_quantile(0.99,
              sum(rate(seldon_api_executor_client_requests_seconds_bucket{
                deployment_name="recommendation-ab-test"
              }[5m])) by (le, predictor_name)
            )
        - record: seldon_model_error_rate
          expr: |
            sum(rate(seldon_api_executor_client_requests_seconds_count{
              deployment_name="recommendation-ab-test",
              code!="200"
            }[5m])) by (predictor_name)
            /
            sum(rate(seldon_api_executor_client_requests_seconds_count{
              deployment_name="recommendation-ab-test"
            }[5m])) by (predictor_name)

---
# statistical_analysis.py — วิเคราะห์ผล A/B Test
import numpy as np
from scipy import stats
import requests
import json

PROMETHEUS_URL = "http://prometheus:9090"

def query_prometheus(query, start, end, step="1h"):
    r = requests.get(f"{PROMETHEUS_URL}/api/v1/query_range", params={
        "query": query,
        "start": start,
        "end": end,
        "step": step
    })
    return r.json()["data"]["result"]

def get_conversion_data(model_name, days=7):
    query = f'sum(seldon_prediction_conversion{{predictor_name="{model_name}"}})'
    total_query = f'sum(seldon_prediction_total{{predictor_name="{model_name}"}})'
    # ดึงข้อมูลจาก business metrics database
    conversions = 1250  # ตัวอย่าง
    total = 5000
    return conversions, total

def run_ab_test_analysis():
    conv_a, total_a = get_conversion_data("model-a")
    conv_b, total_b = get_conversion_data("model-b")

    rate_a = conv_a / total_a
    rate_b = conv_b / total_b

    # Two-proportion z-test
    p_pool = (conv_a + conv_b) / (total_a + total_b)
    se = np.sqrt(p_pool * (1 - p_pool) * (1/total_a + 1/total_b))
    z_stat = (rate_b - rate_a) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

    # Effect size (Cohen's h)
    h = 2 * np.arcsin(np.sqrt(rate_b)) - 2 * np.arcsin(np.sqrt(rate_a))

    print(f"=== A/B Test Results ===")
    print(f"Model A: {rate_a:.4f} ({conv_a}/{total_a})")
    print(f"Model B: {rate_b:.4f} ({conv_b}/{total_b})")
    print(f"Lift: {((rate_b - rate_a) / rate_a * 100):.2f}%")
    print(f"Z-statistic: {z_stat:.4f}")
    print(f"P-value: {p_value:.6f}")
    print(f"Cohen's h: {h:.4f}")
    print(f"Significant (p<0.05): {'YES' if p_value < 0.05 else 'NO'}")

    return {
        "winner": "model-b" if (p_value < 0.05 and rate_b > rate_a) else "model-a",
        "p_value": p_value,
        "lift": (rate_b - rate_a) / rate_a * 100,
        "significant": p_value < 0.05
    }

if __name__ == "__main__":
    result = run_ab_test_analysis()
    if result["significant"]:
        print(f"\nWINNER: {result['winner']} with {result['lift']:.2f}% lift")
    else:
        print("\nNo significant difference detected. Continue testing.")

Automate Model Promotion ด้วย CI/CD Pipeline

สร้าง GitHub Actions Pipeline ที่ promote model อัตโนมัติเมื่อ A/B Test มี statistical significance

# .github/workflows/ml-ab-test-promote.yml
name: ML A/B Test Auto-Promote

on:
  schedule:
    - cron: '0 6 * * *'  # รันทุกวัน 6 โมงเช้า
  workflow_dispatch:

jobs:
  analyze-and-promote:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install numpy scipy requests mlflow kubernetes

      - name: Run Statistical Analysis
        id: analysis
        env:
          PROMETHEUS_URL: }
          MLFLOW_TRACKING_URI: }
        run: |
          python3 statistical_analysis.py > result.json
          WINNER=$(python3 -c "import json; print(json.load(open('result.json'))['winner'])")
          SIGNIFICANT=$(python3 -c "import json; print(json.load(open('result.json'))['significant'])")
          echo "winner=$WINNER" >> $GITHUB_OUTPUT
          echo "significant=$SIGNIFICANT" >> $GITHUB_OUTPUT

      - name: Promote Winner Model
        if: steps.analysis.outputs.significant == 'True'
        env:
          KUBECONFIG_DATA: }
        run: |
          echo "$KUBECONFIG_DATA" | base64 -d > /tmp/kubeconfig
          export KUBECONFIG=/tmp/kubeconfig
          WINNER="}"
          echo "Promoting $WINNER to 100% traffic"
          kubectl patch seldondeployment recommendation-ab-test \
            --type=merge \
            -p "{\"spec\":{\"predictors\":[{\"name\":\"$WINNER\",\"traffic\":100}]}}"

      - name: Notify Slack
        if: always()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {
              "text": "ML A/B Test Result: Winner=}, Significant=}"
            }
        env:
          SLACK_WEBHOOK_URL: }

FAQ คำถามที่พบบ่อย

Q: A/B Testing กับ Multi-Armed Bandit ต่างกันอย่างไร?

A: A/B Testing แบ่ง traffic คงที่ตลอดการทดสอบ (เช่น 80/20) แล้วใช้ statistical test วิเคราะห์ผลตอนจบ ส่วน Multi-Armed Bandit ปรับ traffic แบบ dynamic ตาม performance ที่สังเกตได้ระหว่างทดสอบ ข้อดีของ Bandit คือลด regret (สูญเสียจากการส่ง traffic ไป model ที่แย่กว่า) แต่ A/B Testing ให้ผลทางสถิติที่ชัดเจนและตีความง่ายกว่า

Q: ต้องใช้ sample size เท่าไหร่ถึงจะเพียงพอ?

A: ขึ้นอยู่กับ Minimum Detectable Effect (MDE) ที่ต้องการตรวจจับ ถ้าต้องการตรวจจับ lift 1% ด้วย statistical power 80% และ significance level 5% จะต้องใช้ประมาณ 15,000-20,000 samples ต่อ group ใช้ online sample size calculator หรือคำนวณด้วย scipy.stats.power

Q: จะจัดการ Feature Interaction ระหว่าง A/B Test หลายตัวอย่างไร?

A: ใช้ Layered Experiment Framework แบบที่ Google อธิบายใน paper "Overlapping Experiment Infrastructure" แบ่ง traffic เป็น layers แต่ละ layer รัน experiment อิสระจากกัน หรือใช้ Factorial Design ที่ทดสอบหลาย factors พร้อมกันเพื่อวัด interaction effects

Q: ML Model A/B Testing ต่างจากการทดสอบ UI อย่างไร?

A: ML A/B Testing ต้องพิจารณา metrics เพิ่มเติมเช่น Prediction Latency, Model Drift, Feature Distribution Shift และ Feedback Loop Effects นอกจากนี้ผลลัพธ์จาก ML model อาจใช้เวลานานกว่าจะเห็นผลกระทบต่อ business metrics ทำให้ต้องรัน test นานกว่าการทดสอบ UI ทั่วไป