A/B Testing ML Production Setup Guide —

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

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Dagster Pipeline MLOps Workflow —

# สถาปัตยกรรมของระบบ

# 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

แนะนำเพิ่มเติม — อ่านเพิ่มเติมที่ SiamCafeBook

# 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

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง AI Cha

# 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

แนะนำเพิ่มเติม — iCafeForex

# .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 ต่างกันอย่างไร?

เนื้อหาเกี่ยวข้อง — อ่านต่อ: penetration testing คือ

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 หลายตัวอย่างไร?

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Prometheus PromQL Zero Downtime Deployment

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 ทั่วไป