[PYTHON] A/B Testing을 위한 모델 트래픽 스플리팅 구현 7가지 방법과 기술적 차이 해결

Traffic Splitting

프로덕션 환경에서 리스크를 최소화하고 데이터 기반 의사결정을 가능하게 하는 고급 트래픽 제어 아키텍처 가이드

1. 모델 트래픽 스플리팅(Traffic Splitting)의 전략적 가치

새로운 머신러닝 모델을 배포할 때 가장 두려운 것은 무엇일까요? 바로 '성능 하락'입니다. 오프라인 테스트에서 아무리 점수가 좋았어도, 실제 사용자의 데이터 앞에서는 예측하지 못한 결과가 나올 수 있습니다. 이를 해결하기 위해 필수적인 기술이 바로 트래픽 스플리팅입니다. 트래픽 스플리팅은 전체 사용자 중 일부(예: 10%)에게만 신규 모델(Challenger)을 노출하고, 나머지 90%는 기존 모델(Champion)을 유지하며 두 그룹의 지표를 비교하는 A/B Testing의 핵심 메커니즘입니다. 본 가이드에서는 단순한 난수 생성을 넘어, 고성능 서빙 환경에서 즉시 적용 가능한 7가지 구현 방법을 상세히 다룹니다.

---

2. 트래픽 스플리팅 구현 방식별 차이점 비교

구현 위치(Application vs Infrastructure)와 방식에 따라 유연성과 복잡도의 차이가 발생합니다.

구현 방식	장점	단점	권장 시나리오
Application Layer	사용자 ID 기반 로직 커스텀이 자유로움	애플리케이션 코드에 분기 로직이 섞임	유저 특성(Tier, Region) 기반 정교한 A/B 테스트
API Gateway / Proxy	코드 수정 없이 인프라단에서 제어 가능	고급 유저 세그먼트 구분이 제한적임	단순 랜덤 샘플링 및 카나리 배포
Service Mesh (Istio 등)	쿠버네티스 환경에서 극강의 가용성	설정 및 운영 러닝 커브가 매우 높음	마이크로서비스 아키텍처(MSA) 대규모 서비스
Sidecar Pattern	메인 앱 로직과 스플리팅 로직의 완전 분리	컨테이너 자원 소모량이 증가함	독립적인 실험 모듈 운용 시

---

3. 실무형 트래픽 스플리팅 해결 Example 7선

Python 개발자가 실무 파이프라인에 즉시 통합할 수 있는 단계별 코드 예제입니다.

Example 1: Hash 기반 결정론적 유저 스플리팅 (Deterministic Splitting)

사용자 ID를 해싱하여 동일한 유저는 실험 기간 내내 항상 같은 모델을 보게 만드는 방법입니다.

import hashlib

def get_model_version(user_id, salt="experiment_v1"):
    # 유저 ID와 소금을 결합하여 해시 생성
    combined = f"{user_id}:{salt}".encode()
    hash_val = int(hashlib.md5(combined).hexdigest(), 16)
    
    # 하위 2자리를 취해 0~99 사이의 값 생성
    bucket = hash_val % 100
    
    if bucket < 10: # 10% 트래픽
        return "model_challenger"
    return "model_champion"

# 적용 예시
print(f"User A: {get_model_version('user_123')}")
print(f"User B: {get_model_version('user_456')}")
        

Example 2: FastAPI 미들웨어를 활용한 투명한 라우팅

비즈니스 로직 수정 없이 API 엔드포인트 도달 전 트래픽을 가로채 분배하는 해결책입니다.

from fastapi import FastAPI, Request
import random

app = FastAPI()

@app.middleware("http")
async def dispatch_model_traffic(request: Request, call_next):
    # 헤더나 쿠키에 실험군 정보 주입
    traffic_type = "B" if random.random() < 0.2 else "A"
    request.state.model_group = traffic_type
    
    response = await call_next(request)
    response.headers["X-Experiment-Group"] = traffic_type
    return response

@app.get("/predict")
async def predict(request: Request):
    group = request.state.model_group
    return {"message": f"Results from {group} model"}
        

Example 3: BentoML Runner를 이용한 멀티 모델 서빙

서빙 프레임워크 수준에서 리소스를 격리하면서 트래픽을 제어하는 고성능 아키텍처입니다.

import bentoml
import random

champion_runner = bentoml.sklearn.get("clf_v1:latest").to_runner()
challenger_runner = bentoml.sklearn.get("clf_v2:latest").to_runner()

svc = bentoml.Service("model_ab_test", runners=[champion_runner, challenger_runner])

@svc.api(input=bentoml.io.NumpyNdarray(), output=bentoml.io.JSON())
async def classify(input_data):
    if random.random() < 0.15: # 15% 실험군
        result = await challenger_runner.predict.async_run(input_data)
        tag = "challenger"
    else:
        result = await champion_runner.predict.async_run(input_data)
        tag = "champion"
    return {"prediction": result.tolist(), "version": tag}
        

Example 4: Redis를 이용한 동적 가중치 제어 (Hot Reload)

서버 재시작 없이 실시간으로 트래픽 비율을 0%에서 100%까지 조정하는 고급 방법입니다.

import redis
import random

r = redis.Redis(host='localhost', port=6379, db=0)

def get_dynamic_weight():
    # Redis에서 'challenger_weight' 키를 읽어옴 (기본값 5)
    weight = r.get("exp:challenger:weight")
    return int(weight) if weight else 5

def route_request():
    current_weight = get_dynamic_weight()
    if random.random() * 100 < current_weight:
        return "v2_model"
    return "v1_model"
        

Example 5: Weights & Biases(W&B) 연동 실험 추적 로깅

스플리팅된 트래픽의 결과를 외부 플랫폼에 기록하여 실시간 성능 차이를 분석하는 방법입니다.

import wandb

# API 호출 시마다 로깅 (배치 처리 권장)
def log_ab_result(user_id, model_tag, metric_value):
    wandb.log({
        "user_id": user_id,
        "model_tag": model_tag,
        "accuracy_score": metric_value
    })
        

Example 6: Nginx Split Clients 모듈 설정을 통한 인프라 레벨 제어

Python 코드 도달 전에 로드밸런서에서 트래픽을 물리적으로 분리하는 해결법입니다.

# nginx.conf 예시
http {
    split_clients "${remote_addr}AAA" $variant {
        10%     "challenger_upstream";
        * "champion_upstream";
    }

    server {
        location /predict {
            proxy_pass http://$variant;
        }
    }
}
        

Example 7: 쉐도우 배포(Shadow Deployment) 아키텍처

실제 응답은 Champion이 주지만, 백그라운드에서 Challenger에게도 데이터를 보내 성능만 측정하는 리스크 제로 방법입니다.

import asyncio

async def shadow_predict(data):
    # 실제 응답에는 영향을 주지 않는 비동기 실행
    try:
        await challenger_model.predict(data)
    except Exception as e:
        print(f"Shadow model failed: {e}")

async def main_predict(data):
    # 1. 챔피언 모델 응답 생성
    response = await champion_model.predict(data)
    
    # 2. 챌린저 모델에게 '그림자' 요청 (Fire and Forget)
    asyncio.create_task(shadow_predict(data))
    
    return response
        

---

4. 성공적인 A/B 테스트를 위한 엔지니어링 체크리스트

단순히 트래픽을 나누는 것보다 중요한 것은 데이터의 오염(Data Contamination)을 막는 것입니다.

• Sticky Sessions: 유저가 사이트를 새로고침할 때마다 모델이 바뀌면 사용자 경험이 저하됩니다. 해시 기반 스플리팅을 우선 고려하십시오.
• Latency Monitoring: 신규 모델이 기존 모델보다 느리다면 트래픽을 늘리기 전에 성능 최적화가 선행되어야 합니다.
• Sample Size Calculator: 통계적으로 유의미한 결론을 내리기 위해 필요한 최소 샘플 수를 미리 계산하십시오.
• Kill Switch: 신규 모델에서 에러가 급증할 경우 즉시 모든 트래픽을 롤백할 수 있는 자동화된 장치를 마련하십시오.

5. 결론 및 향후 전망

트래픽 스플리팅은 단순한 기술적 선택이 아닌, 조직의 실험 문화와 직결됩니다. Application Layer에서의 유연함과 Infrastructure Layer에서의 안정성 사이에서 최적의 균형점을 찾는 것이 시니어 개발자의 역할입니다. 최근에는 Feature Flag 시스템과 연동하여 비개발 직군도 트래픽 비율을 조절할 수 있는 환경으로 진화하고 있습니다.

참고 문헌 및 정보 출처

• "Patterns for Machine Learning Operations (MLOps)", Google Cloud Architecture.
• Istio Traffic Management Documentation - istio.io
• BentoML Model Serving Guide - docs.bentoml.org
• Kohavi, R., "Trustworthy Online Controlled Experiments", Cambridge University Press.