Trong bối cảnh các mô hình AI ngày càng đa dạng, việc lựa chọn đúng tổ hợp mô hình cho production không chỉ ảnh hưởng đến chất lượng đầu ra mà còn tác động trực tiếp đến chi phí vận hành. Bài viết này sẽ hướng dẫn bạn xây dựng một Multi-Model A/B Testing Framework hoàn chỉnh, giúp tự động đánh giá và so sánh hiệu suất của nhiều mô hình LLM khác nhau.
Tại sao cần Multi-Model A/B Testing?
Mỗi mô hình LLM có điểm mạnh yếu khác nhau. GPT-4.1 xuất sắc trong reasoning phức tạp, Claude Sonnet 4.5 mạnh về sáng tạo nội dung, Gemini 2.5 Flash nhanh và tiết kiệm, DeepSeek V3.2 thể hiện tốt trên các tác vụ coding. Vấn đề là làm sao biết được tổ hợp nào tối ưu cho use-case cụ thể của bạn?
Với HolySheep AI, bạn có thể truy cập tất cả các mô hình này qua một API duy nhất với chi phí tiết kiệm đến 85% so với các nhà cung cấp truyền thống — chỉ ¥1 = $1. Tính năng thanh toán qua WeChat/Alipay và độ trễ dưới 50ms giúp việc testing trở nên thực tế hơn bao giờ hết.
Kiến trúc tổng quan
Framework của chúng ta bao gồm 4 thành phần chính:
- Model Gateway: Quản lý kết nối đến nhiều provider
- Traffic Router: Phân phối request theo chiến lược A/B
- Metrics Collector: Thu thập metrics về chất lượng và chi phí
- Optimizer Engine: Phân tích và đề xuất tổ hợp tối ưu
Triển khai Production-Level Code
1. Cấu hình Models và Providers
"""
Multi-Model A/B Testing Framework
Production-ready implementation với HolySheep AI
"""
import asyncio
import httpx
import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import hashlib
class ModelProvider(Enum):
HOLYSHEEP = "holysheep"
OPENAI = "openai"
ANTHROPIC = "anthropic"
@dataclass
class ModelConfig:
"""Cấu hình cho mỗi mô hình trong A/B test"""
model_id: str
provider: ModelProvider
api_key: str
base_url: str = "https://api.holysheep.ai/v1" # Chỉ dùng HolySheep
max_tokens: int = 4096
temperature: float = 0.7
cost_per_1k_tokens: float = 0.0 # Input + Output
expected_latency_ms: float = 0.0
weight: float = 1.0 # Trọng số phân phối traffic
@dataclass
class ABTestConfig:
"""Cấu hình A/B test"""
test_name: str
models: list[ModelConfig]
traffic_split: dict[str, float] = field(default_factory=dict)
min_samples: int = 100
confidence_level: float = 0.95
max_test_duration_hours: int = 24
Định nghĩa các models với pricing 2026
MODELS = {
"gpt4.1": ModelConfig(
model_id="gpt-4.1",
provider=ModelProvider.HOLYSHEEP,
api_key="YOUR_HOLYSHEEP_API_KEY",
cost_per_1k_tokens=8.0, # $8/MTok
expected_latency_ms=2500
),
"claude_sonnet_45": ModelConfig(
model_id="claude-sonnet-4.5",
provider=ModelProvider.HOLYSHEEP,
api_key="YOUR_HOLYSHEEP_API_KEY",
cost_per_1k_tokens=15.0, # $15/MTok
expected_latency_ms=3000
),
"gemini_flash_25": ModelConfig(
model_id="gemini-2.5-flash",
provider=ModelProvider.HOLYSHEEP,
api_key="YOUR_HOLYSHEEP_API_KEY",
cost_per_1k_tokens=2.50, # $2.50/MTok
expected_latency_ms=800
),
"deepseek_v32": ModelConfig(
model_id="deepseek-v3.2",
provider=ModelProvider.HOLYSHEEP,
api_key="YOUR_HOLYSHEEP_API_KEY",
cost_per_1k_tokens=0.42, # $0.42/MTok
expected_latency_ms=1200
),
}
def calculate_traffic_split(models: list[ModelConfig]) -> dict[str, float]:
"""Tính toán phân phối traffic dựa trên trọng số"""
total_weight = sum(m.weight for m in models)
return {
m.model_id: (m.weight / total_weight) * 100
for m in models
}2. Model Gateway với Concurrency Control
import asyncio
from typing import Any
import json
from collections import defaultdict
import statistics
@dataclass
class Response:
"""Standardized response từ mô hình"""
model_id: str
content: str
latency_ms: float
input_tokens: int
output_tokens: int
cost: float
error: Optional[str] = None
metadata: dict = field(default_factory=dict)
@dataclass
class MetricsSnapshot:
"""Metrics tại một thời điểm"""
total_requests: int
successful_requests: int
failed_requests: int
avg_latency_ms: float
p95_latency_ms: float
avg_cost_per_request: float
quality_score: float = 0.0
class ModelGateway:
"""
Gateway quản lý kết nối đến nhiều mô hình
với concurrency control và rate limiting
"""
def __init__(self):
self.client: Optional[httpx.AsyncClient] = None
self.semaphores: dict[str, asyncio.Semaphore] = {}
self.rate_limits: dict[str, tuple[int, float]] = {} # (count, window)
self.metrics: dict[str, list[float]] = defaultdict(list)
self._lock = asyncio.Lock()
async def __aenter__(self):
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0, connect=10.0),
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
return self
async def __aexit__(self, *args):
if self.client:
await self.client.aclose()
async def call_model(
self,
config: ModelConfig,
prompt: str,
system_prompt: str = "",
max_concurrent: int = 10
) -> Response:
"""Gọi mô hình với concurrency control"""
# Khởi tạo semaphore cho mỗi model
if config.model_id not in self.semaphores:
async with self._lock:
if config.model_id not in self.semaphores:
self.semaphores[config.model_id] = asyncio.Semaphore(max_concurrent)
semaphore = self.semaphores[config.model_id]
async with semaphore:
start_time = time.perf_counter()
try:
# Kiểm tra rate limit
await self._check_rate_limit(config.model_id)
# Build request
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": config.model_id,
"messages": messages,
"max_tokens": config.max_tokens,
"temperature": config.temperature,
}
headers = {
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
# Gọi API
response = await self.client.post(
f"{config.base_url}/chat/completions",
json=payload,
headers=headers
)
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status_code != 200:
return Response(
model_id=config.model_id,
content="",
latency_ms=latency_ms,
input_tokens=0,
output_tokens=0,
cost=0.0,
error=f"HTTP {response.status_code}: {response.text}"
)
data = response.json()
content = data["choices"][0]["message"]["content"]
# Tính tokens và cost
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
total_tokens = input_tokens + output_tokens
cost = (total_tokens / 1000) * config.cost_per_1k_tokens
# Lưu metrics
self.metrics[config.model_id].append(latency_ms)
return Response(
model_id=config.model_id,
content=content,
latency_ms=latency_ms,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost=cost,
metadata=data.get("usage", {})
)
except asyncio.TimeoutError:
return Response(
model_id=config.model_id,
content="",
latency_ms=(time.perf_counter() - start_time) * 1000,
input_tokens=0,
output_tokens=0,
cost=0.0,
error="Request timeout"
)
except Exception as e:
return Response(
model_id=config.model_id,
content="",
latency_ms=(time.perf_counter() - start_time) * 1000,
input_tokens=0,
output_tokens=0,
cost=0.0,
error=str(e)
)
async def _check_rate_limit(self, model_id: str):
"""Implement rate limiting đơn giản"""
# Simplified rate limit check
await asyncio.sleep(0) # Yield control
class ABTestRunner:
"""Chạy A/B test với traffic routing thông minh"""
def __init__(
self,
config: ABTestConfig,
gateway: ModelGateway,
quality_evaluator=None
):
self.config = config
self.gateway = gateway
self.quality_evaluator = quality_evaluator
self.results: dict[str, list[Response]] = defaultdict(list)
self.start_time: Optional[float] = None
def select_model(self, user_id: str) -> ModelConfig:
"""Chọn model dựa trên consistent hashing cho user"""
# Consistent hashing để đảm bảo same user luôn vào same bucket
hash_value = int(hashlib.md5(f"{user_id}_{self.config.test_name}".encode()).hexdigest(), 16)
models = self.config.models
total_weight = sum(m.weight for m in models)
normalized_hash = (hash_value % total_weight) / total_weight
cumulative = 0.0
for model in models:
cumulative += model.weight / total_weight
if normalized_hash < cumulative:
return model
return models[0]
async def run_single_request(
self,
user_id: str,
prompt: str,
system_prompt: str = ""
) -> Response:
"""Chạy một request A/B test"""
if self.start_time is None:
self.start_time = time.time()
model_config = self.select_model(user_id)
response = await self.gateway.call_model(
config=model_config,
prompt=prompt,
system_prompt=system_prompt
)
self.results[model_config.model_id].append(response)
return response
def get_metrics(self) -> dict[str, MetricsSnapshot]:
"""Thu thập metrics tổng hợp"""
snapshots = {}
for model_id, responses in self.results.items():
if not responses:
continue
successful = [r for r in responses if not r.error]
failed = [r for r in responses if r.error]
latencies = [r.latency_ms for r in successful]
costs = [r.cost for r in successful]
snapshots[model_id] = MetricsSnapshot(
total_requests=len(responses),
successful_requests=len(successful),
failed_requests=len(failed),
avg_latency_ms=statistics.mean(latencies) if latencies else 0,
p95_latency_ms=(
sorted(latencies)[int(len(latencies) * 0.95)]
if len(latencies) > 10 else 0
),
avg_cost_per_request=statistics.mean(costs) if costs else 0,
quality_score=self._calculate_quality_score(successful)
)
return snapshots
def _calculate_quality_score(self, responses: list[Response]) -> float:
"""Tính quality score dựa trên heuristics"""
if not responses:
return 0.0
scores = []
for r in responses:
# Basic heuristics - có thể thay bằng ML model
score = 1.0
# Penalize nếu response quá ngắn
if len(r.content) < 50:
score *= 0.5
# Penalize nếu có error markers
error_markers = ["sorry", "cannot", "unable", "error"]
if any(marker in r.content.lower() for marker in error_markers):
score *= 0.7
# Reward nếu có structured content
if "``" in r.content or "``json" in r.content:
score *= 1.1
scores.append(min(score, 1.0))
return statistics.mean(scores)
def should_continue(self) -> bool:
"""Kiểm tra xem test có nên tiếp tục không"""
if self.start_time is None:
return True
elapsed_hours = (time.time() - self.start_time) / 3600
if elapsed_hours >= self.config.max_test_duration_hours:
return False
for model_id, responses in self.results.items():
if len(responses) < self.config.min_samples:
return True
Tài nguyên liên quan
Bài viết liên quan