카나리 배포, 섀도우 테스팅, A/B 테스트, 블루-그린 배포, 롤백 전략 등 AI 시스템을 프로덕션에 안전하게 배포하는 전략을 다룹니다.
전통 소프트웨어의 배포는 코드 한 가지만 관리하면 됩니다. 하지만 AI 시스템의 배포는 모델, 프롬프트, 설정, 가드레일, 평가 기준 등 여러 요소가 동시에 변경될 수 있어 훨씬 복잡합니다. 모델을 업그레이드했는데 기존 프롬프트와 호환되지 않거나, 가드레일 규칙을 바꿨는데 예상치 못한 차단이 발생하는 상황은 흔합니다. 배포 하네스는 이런 위험을 통제하면서 변경을 안전하게 릴리즈합니다.
전통 소프트웨어와 AI 시스템의 배포가 다른 근본적인 이유는 변경 단위의 다양성 때문입니다.
| 변경 유형 | 예시 | 위험도 |
|---|---|---|
| 모델 업그레이드 | Claude 3.5 -> Claude 4 | 높음 |
| 프롬프트 변경 | 시스템 프롬프트 수정 | 중간-높음 |
| 설정 변경 | temperature, max_tokens 조정 | 낮음-중간 |
| 가드레일 변경 | 필터링 규칙 추가/수정 | 중간 |
| 도구 변경 | 새 도구 추가, 기존 도구 수정 | 중간-높음 |
| RAG 데이터 변경 | 벡터 DB 업데이트 | 중간 |
이 모든 변경이 개별적으로도 출력에 영향을 미치고, 조합되면 예측하기 어려운 상호작용을 일으킬 수 있습니다.
카나리 배포(Canary Deployment)는 새 버전을 전체 사용자의 일부에게만 먼저 배포하고, 문제가 없으면 점진적으로 비율을 늘리는 전략입니다.
from dataclasses import dataclass
import random
@dataclass
class DeploymentConfig:
"""배포 설정"""
stable_version: str
canary_version: str
canary_percentage: float # 0.0 ~ 1.0
promotion_thresholds: list[float] # 예: [0.05, 0.25, 0.50, 1.0]
rollback_criteria: dict
class CanaryRouter:
"""카나리 배포 라우터"""
def __init__(self, config: DeploymentConfig):
self.config = config
self.metrics = {
"stable": {"requests": 0, "errors": 0, "latency_sum": 0},
"canary": {"requests": 0, "errors": 0, "latency_sum": 0},
}
def route(self, request_id: str) -> str:
"""요청을 안정 버전 또는 카나리로 라우팅"""
# 동일 사용자는 항상 같은 버전으로 (세션 고정)
hash_value = hash(request_id) % 100
if hash_value < self.config.canary_percentage * 100:
return self.config.canary_version
return self.config.stable_version
def record_metric(
self, version: str, latency_ms: float, is_error: bool
):
key = "canary" if version == self.config.canary_version else "stable"
self.metrics[key]["requests"] += 1
self.metrics[key]["latency_sum"] += latency_ms
if is_error:
self.metrics[key]["errors"] += 1
def should_rollback(self) -> bool:
"""롤백 조건 확인"""
canary = self.metrics["canary"]
stable = self.metrics["stable"]
if canary["requests"] < 100:
return False # 충분한 데이터 수집 전
canary_error_rate = canary["errors"] / canary["requests"]
stable_error_rate = (
stable["errors"] / stable["requests"]
if stable["requests"] > 0 else 0
)
# 카나리 에러율이 안정 버전의 2배 이상이면 롤백
threshold = self.config.rollback_criteria.get(
"error_rate_multiplier", 2.0
)
if canary_error_rate > stable_error_rate * threshold:
return True
# 카나리 지연시간이 안정 버전의 1.5배 이상이면 롤백
canary_avg_latency = canary["latency_sum"] / canary["requests"]
stable_avg_latency = (
stable["latency_sum"] / stable["requests"]
if stable["requests"] > 0 else float("inf")
)
latency_threshold = self.config.rollback_criteria.get(
"latency_multiplier", 1.5
)
if canary_avg_latency > stable_avg_latency * latency_threshold:
return True
return False
def should_promote(self) -> bool:
"""승격 조건 확인"""
canary = self.metrics["canary"]
if canary["requests"] < 500:
return False
canary_error_rate = canary["errors"] / canary["requests"]
return canary_error_rate < 0.01 # 에러율 1% 미만AI 시스템의 카나리 배포에서는 에러율뿐 아니라 품질 메트릭도 모니터링해야 합니다. 에러 없이 동작하지만 응답 품질이 떨어지는 것은 전통적인 카나리 모니터링으로는 감지하기 어렵습니다. 5장에서 다룬 평가 하네스를 카나리 모니터링에 통합하세요.
섀도우 테스팅(Shadow Testing)은 실제 프로덕션 트래픽을 새 버전에도 전달하되, 사용자에게는 기존 버전의 응답만 반환하는 방식입니다. 새 버전의 응답은 기록만 하고 사용자에게 노출하지 않으므로, 위험 없이 실전 데이터로 평가할 수 있습니다.
import asyncio
from dataclasses import dataclass
@dataclass
class ShadowResult:
primary_response: str
shadow_response: str
primary_latency_ms: float
shadow_latency_ms: float
responses_match: bool
similarity_score: float
class ShadowTestHarness:
"""섀도우 테스팅 하네스"""
def __init__(self, primary_fn, shadow_fn, comparator_fn):
self.primary = primary_fn
self.shadow = shadow_fn
self.comparator = comparator_fn
self.results: list[ShadowResult] = []
async def process(self, request: str) -> str:
"""
두 버전을 동시에 실행하되,
사용자에게는 primary 결과만 반환
"""
import time
# 병렬 실행
primary_start = time.monotonic()
primary_task = asyncio.create_task(self.primary(request))
shadow_task = asyncio.create_task(self.shadow(request))
primary_response = await primary_task
primary_latency = (time.monotonic() - primary_start) * 1000
try:
shadow_start = time.monotonic()
shadow_response = await asyncio.wait_for(
shadow_task, timeout=10.0
)
shadow_latency = (time.monotonic() - shadow_start) * 1000
except (asyncio.TimeoutError, Exception) as e:
shadow_response = f"[SHADOW ERROR] {e}"
shadow_latency = -1
# 비교 및 기록 (비동기, 메인 흐름 차단 없음)
similarity = await self.comparator(
primary_response, shadow_response
)
self.results.append(ShadowResult(
primary_response=primary_response,
shadow_response=shadow_response,
primary_latency_ms=primary_latency,
shadow_latency_ms=shadow_latency,
responses_match=similarity > 0.9,
similarity_score=similarity,
))
# 사용자에게는 primary 응답만 반환
return primary_response
def generate_report(self) -> dict:
"""섀도우 테스트 결과 리포트"""
if not self.results:
return {"error": "결과 없음"}
total = len(self.results)
matches = sum(1 for r in self.results if r.responses_match)
avg_similarity = sum(
r.similarity_score for r in self.results
) / total
return {
"total_requests": total,
"match_rate": matches / total,
"avg_similarity": avg_similarity,
"primary_avg_latency": sum(
r.primary_latency_ms for r in self.results
) / total,
"shadow_avg_latency": sum(
r.shadow_latency_ms for r in self.results
if r.shadow_latency_ms > 0
) / max(1, sum(
1 for r in self.results
if r.shadow_latency_ms > 0
)),
}섀도우 테스팅은 모델 업그레이드 시 가장 유용합니다. 실제 사용자 요청에 대해 구 모델과 신 모델의 응답을 비교하여, 어떤 유형의 요청에서 차이가 나는지를 파악할 수 있습니다.
A/B 테스트는 두 버전의 성능을 통계적으로 비교하는 방법입니다. 카나리 배포와 유사하지만, 목적이 다릅니다. 카나리가 "새 버전에 문제가 없는가?"를 확인하는 것이라면, A/B 테스트는 "어떤 버전이 더 나은가?"를 판단합니다.
from dataclasses import dataclass, field
import random
from scipy import stats
@dataclass
class ABTestConfig:
name: str
variant_a: str # 예: "system_prompt_v1"
variant_b: str # 예: "system_prompt_v2"
traffic_split: float = 0.5 # A의 트래픽 비율
min_samples: int = 1000
confidence_level: float = 0.95
@dataclass
class ABTestResult:
variant: str
metric_values: list[float] = field(default_factory=list)
@property
def mean(self) -> float:
return sum(self.metric_values) / len(self.metric_values) if self.metric_values else 0
@property
def count(self) -> int:
return len(self.metric_values)
class ABTestHarness:
"""A/B 테스트 하네스"""
def __init__(self, config: ABTestConfig):
self.config = config
self.results_a = ABTestResult(variant=config.variant_a)
self.results_b = ABTestResult(variant=config.variant_b)
def assign_variant(self, user_id: str) -> str:
"""사용자를 변형에 할당 (결정적)"""
hash_value = hash(user_id + self.config.name) % 100
if hash_value < self.config.traffic_split * 100:
return self.config.variant_a
return self.config.variant_b
def record(self, variant: str, metric_value: float):
"""메트릭 기록"""
if variant == self.config.variant_a:
self.results_a.metric_values.append(metric_value)
else:
self.results_b.metric_values.append(metric_value)
def analyze(self) -> dict:
"""통계적 유의성 분석"""
if (self.results_a.count < self.config.min_samples or
self.results_b.count < self.config.min_samples):
return {
"status": "insufficient_data",
"samples_a": self.results_a.count,
"samples_b": self.results_b.count,
"min_required": self.config.min_samples,
}
# 독립 표본 t-검정
t_stat, p_value = stats.ttest_ind(
self.results_a.metric_values,
self.results_b.metric_values,
)
significant = p_value < (1 - self.config.confidence_level)
winner = None
if significant:
winner = (
self.config.variant_a
if self.results_a.mean > self.results_b.mean
else self.config.variant_b
)
return {
"status": "conclusive" if significant else "inconclusive",
"mean_a": self.results_a.mean,
"mean_b": self.results_b.mean,
"p_value": p_value,
"significant": significant,
"winner": winner,
"improvement": abs(
self.results_a.mean - self.results_b.mean
) / max(self.results_a.mean, 0.001) * 100,
}블루-그린 배포(Blue-Green Deployment)는 두 개의 동일한 프로덕션 환경을 유지하고, 트래픽을 한 번에 전환하는 방식입니다. 롤백이 단순히 트래픽을 다시 전환하는 것으로 완료되므로, 가장 빠른 롤백이 가능합니다.
블루-그린 배포의 장점은 빠른 전환과 롤백이지만, 비용이 2배입니다. 항상 두 환경을 유지해야 하므로 인프라 비용이 증가합니다. AI 시스템에서는 GPU 자원이 비싸므로, 전환 시점 전후로만 두 환경을 동시에 유지하는 타협안도 고려하세요.
AI 시스템의 롤백은 코드 롤백만으로는 부족합니다. 모델, 프롬프트, 설정, 가드레일을 모두 이전 상태로 되돌릴 수 있어야 합니다.
from dataclasses import dataclass
from datetime import datetime
@dataclass
class SystemSnapshot:
"""시스템 상태 스냅샷"""
version: str
timestamp: datetime
model_config: dict
prompt_versions: dict[str, str]
guardrail_rules: dict
feature_flags: dict
deployment_hash: str
class RollbackManager:
"""다차원 롤백 관리"""
def __init__(self):
self.snapshots: list[SystemSnapshot] = []
def capture(self, version: str, config: dict) -> SystemSnapshot:
"""현재 상태 스냅샷 캡처"""
snapshot = SystemSnapshot(
version=version,
timestamp=datetime.now(),
model_config=config["model"],
prompt_versions=config["prompts"],
guardrail_rules=config["guardrails"],
feature_flags=config["features"],
deployment_hash=compute_hash(config),
)
self.snapshots.append(snapshot)
return snapshot
async def rollback_to(self, version: str) -> bool:
"""특정 버전으로 롤백"""
target = next(
(s for s in reversed(self.snapshots)
if s.version == version),
None,
)
if not target:
return False
# 모든 구성 요소를 원자적으로 롤백
await self._apply_model_config(target.model_config)
await self._apply_prompts(target.prompt_versions)
await self._apply_guardrails(target.guardrail_rules)
await self._apply_feature_flags(target.feature_flags)
return True
async def rollback_component(
self,
component: str,
version: str,
) -> bool:
"""특정 구성 요소만 선택적 롤백"""
target = next(
(s for s in reversed(self.snapshots)
if s.version == version),
None,
)
if not target:
return False
if component == "model":
await self._apply_model_config(target.model_config)
elif component == "prompts":
await self._apply_prompts(target.prompt_versions)
elif component == "guardrails":
await self._apply_guardrails(target.guardrail_rules)
return TrueAI 시스템의 프롬프트와 설정은 코드처럼 버전 관리되어야 합니다. 이를 PaC(Prompt as Code) 또는 CaC(Configuration as Code)라 부릅니다.
# AI 시스템 배포 설정 (버전 관리됨)
version: "2.3.0"
model:
provider: anthropic
model_id: claude-sonnet-4-20250514
temperature: 0.7
max_tokens: 4096
prompts:
system: "prompts/system_v5.txt"
few_shot: "prompts/few_shot_v3.json"
guardrails:
input:
max_length: 50000
injection_detection: true
topic_restriction:
allowed: [product, billing, support]
blocked: [politics, religion]
output:
pii_masking: true
toxicity_threshold: 0.8
max_length: 10000
evaluation:
golden_dataset: "eval/golden_v12.json"
min_pass_rate: 0.95
required_metrics:
- exact_match: 0.85
- semantic_similarity: 0.90# 1. 설정 변경을 커밋
git add deployment/config.yaml prompts/
git commit -m "feat: 시스템 프롬프트 v6 및 가드레일 규칙 업데이트"
# 2. CI에서 자동 평가 실행
# (golden dataset 통과 여부 확인)
# 3. 카나리 배포 시작
deploy --strategy canary --percentage 5
# 4. 모니터링 후 승격 또는 롤백
deploy --promote # 또는 deploy --rollback프롬프트 변경이 코드 변경만큼 (또는 그 이상으로) 시스템 동작에 영향을 미칩니다. 프롬프트 변경에도 코드 리뷰, 자동 테스트, 점진적 배포를 동일하게 적용하세요.
9장에서는 프로덕션에 배포된 AI 시스템을 지속적으로 관측하는 모니터링 하네스를 다룹니다. 토큰 사용량, 지연시간, 비용 추적, 드리프트 감지, 품질 모니터링, 피드백 루프까지 관측 가능성 파이프라인 전체를 살펴봅니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
토큰 사용량, 지연시간, 비용 추적, 드리프트 감지, 품질 모니터링, 알림 설계, 피드백 루프 등 AI 시스템의 관측 가능성 파이프라인을 다룹니다.
에이전트 라이프사이클 관리, 도구 오케스트레이션, 서브에이전트 관리, 상태 관리, 에러 복구 등 복잡한 AI 워크플로우를 조율하는 방법을 다룹니다.
전체 하네스 계층 통합, 하네스 성숙도 모델, CI/CD 파이프라인 통합, CLAUDE.md와 AGENTS.md 설계, 팀 협업 전략까지 하네스 엔지니어링의 완결편입니다.