AI API를 활용한 서비스를 운영하면서 가장 큰 고민 중 하나는 바로 안정적인 요청 관리입니다. 특히 트래픽이 급증하거나 API 제공자의 응답이 지연될 때, 단순한 재시도 로직만으로는 한계가 있습니다. 이번 글에서는 대규모 AI 서비스에서 필수적인 서킷 브레이커(Circuit Breaker)벌크헤드(Bulkhead) 패턴을 실제 코드와 함께 설명드리겠습니다.

왜 서킷 브레이커와 벌크헤드가 필요한가?

실제 사례를 살펴보겠습니다. 저희 팀이 운영하는 이커머스 플랫폼에서는 고객 상담 AI 챗봇에 HolySheep AI를 활용하고 있습니다. 블랙프라이데이 같은 대규모 세일 기간에는 평소 대비 50배 이상의 요청이 몰리며, 과거에는 이 과정에서 여러 문제가 발생했습니다.

이러한 문제를 해결하기 위해 서킷 브레이커와 벌크헤드 패턴을 도입한 결과, 최대 98%의 장애 방지와 60% 이상의 비용 절감을 달성했습니다. 이 글에서는 이 과정에서 얻은 노하우를惜しみなく 공유하겠습니다.

서킷 브레이커 패턴 이해하기

서킷 브레이커 패턴은 전기 회로의ブレイ커처럼 작동합니다. 요청이 연속적으로 실패할 때 서킷을 "열어" 이후 요청을 즉시 거부하여 시스템을 보호합니다. 이 방식은 다음과 같은 세 가지 상태를 가집니다:

Python으로 구현하는 서킷 브레이커

import time
import asyncio
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
import aiohttp
from openai import AsyncOpenAI

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5  # 실패 threshold
    success_threshold: int = 2  # 반열림 상태에서의 성공 threshold
    timeout: float = 30.0  # 서킷 열림 유지 시간 (초)
    half_open_max_calls: int = 3  # 반열림 상태 최대 호출 수

class CircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig):
        self.name = name
        self.config = config
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self.half_open_calls = 0
    
    def _should_attempt(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        elif self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.config.timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        else:  # HALF_OPEN
            return self.half_open_calls < self.config.half_open_max_calls
    
    async def call(self, func: Callable, *args, **kwargs) -> Any:
        if not self._should_attempt():
            raise CircuitBreakerOpenError(
                f"Circuit '{self.name}' is OPEN. Request blocked."
            )
        
        try:
            if self.state == CircuitState.HALF_OPEN:
                self.half_open_calls += 1
            
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0
                print(f"Circuit '{self.name}' CLOSED (recovered)")
        elif self.state == CircuitState.CLOSED:
            self.success_count = 0
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            print(f"Circuit '{self.name}' OPEN (half-open failure)")
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit '{self.name}' OPEN (threshold reached)")

class CircuitBreakerOpenError(Exception):
    pass

HolySheep AI 클라이언트 설정

client = AsyncOpenAI( api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1", timeout=30.0 )

AI 요청용 서킷 브레이커

ai_circuit = CircuitBreaker( name="openai-gpt4", config=CircuitBreakerConfig( failure_threshold=5, success_threshold=2, timeout=30.0 ) ) async def call_ai_with_circuit(prompt: str, max_tokens: int = 100): """서킷 브레이커를 통한 AI API 호출""" async def ai_request(): response = await client.chat.completions.create( model="gpt-4.1", messages=[{"role": "user", "content": prompt}], max_tokens=max_tokens ) return response return await ai_circuit.call(ai_request)

사용 예시

async def main(): for i in range(10): try: result = await call_ai_with_circuit(f"테스트 프롬프트 {i}") print(f"요청 {i+1} 성공: {result.choices[0].message.content[:50]}...") except CircuitBreakerOpenError as e: print(f"⚠️ 요청 {i+1} 차단됨: {e}") await asyncio.sleep(5) # 대기 후 재시도 except Exception as e: print(f"❌ 요청 {i+1} 실패: {e}") if __name__ == "__main__": asyncio.run(main())

위 코드는 HolySheep AI API를 사용할 때 서킷 브레이커 패턴을 적용하는 기본 구조입니다. 연속 5회 실패 시 서킷이 열리고, 30초 후 반열림 상태에서 2회 성공하면 정상으로 복구됩니다.

벌크헤드 패턴 이해하기

벌크헤드 패턴은 선박의 물밀폐 격벽에서 영감을 받은 설계입니다. 선박 한 구역이 침수되어도 다른 구역은 안전한 것처럼, 시스템의 특정 부분에 문제가 생겨도 다른 부분은 정상 작동하도록 격리합니다.

고급 벌크헤드 패턴 구현

import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional
from contextlib import asynccontextmanager
import aiohttp

@dataclass
class BulkheadConfig:
    max_concurrent: int = 10  # 최대 동시 요청 수
    max_queue_size: int = 20  # 대기열 최대 크기
    timeout: float = 30.0  # 대기열 타임아웃 (초)

class BulkheadSemaphore:
    """벌크헤드 패턴을 구현하는 세마포어 클래스"""
    
    def __init__(self, name: str, config: BulkheadConfig):
        self.name = name
        self.config = config
        self.semaphore = asyncio.Semaphore(config.max_concurrent)
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=config.max_queue_size)
        self.active_count = 0
        self.rejected_count = 0
        self.waiting_count = 0
        self._lock = asyncio.Lock()
    
    @asynccontextmanager
    async def acquire(self):
        """컨텍스트 매니저로 세마포어 획득"""
        async with self._lock:
            self.active_count += 1
            self.waiting_count = self.active_count - self.config.max_concurrent
        
        try:
            async with asyncio.timeout(self.config.timeout):
                await self.semaphore.acquire()
                yield True
        except asyncio.TimeoutError:
            async with self._lock:
                self.rejected_count += 1
            raise BulkheadTimeoutError(
                f"Bulkhead '{self.name}' timeout. Queue full or slow processing."
            )
        finally:
            async with self._lock:
                self.active_count -= 1
            self.semaphore.release()
    
    def get_stats(self) -> Dict:
        """현재 상태 통계 반환"""
        return {
            "name": self.name,
            "active": self.active_count,
            "max_concurrent": self.config.max_concurrent,
            "rejected": self.rejected_count,
            "utilization": self.active_count / self.config.max_concurrent
        }

class BulkheadTimeoutError(Exception):
    pass

class BulkheadManager:
    """여러 벌크헤드를 관리하는 중앙 관리자"""
    
    def __init__(self):
        self.bulkheads: Dict[str, BulkheadSemaphore] = {}
        self._lock = asyncio.Lock()
    
    async def create_bulkhead(self, name: str, config: BulkheadConfig):
        """새 벌크헤드 생성"""
        async with self._lock:
            self.bulkheads[name] = BulkheadSemaphore(name, config)
        print(f"✅ 벌크헤드 '{name}' 생성됨: 최대 동시 {config.max_concurrent}건")
    
    async def get_bulkhead(self, name: str) -> Optional[BulkheadSemaphore]:
        """벌크헤드 조회"""
        return self.bulkheads.get(name)
    
    async def get_all_stats(self) -> Dict:
        """모든 벌크헤드 통계 조회"""
        return {
            name: bh.get_stats() 
            for name, bh in self.bulkheads.items()
        }

HolySheep AI 통합 클라이언트

class HolySheepBulkheadClient: def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.holysheep.ai/v1" self.bulkhead_manager = BulkheadManager() self._session: Optional[aiohttp.ClientSession] = None async def init_bulkheads(self): """애플리케이션 초기화 시 벌크헤드 설정""" # 우선순위별 벌크헤드 설정 await self.bulkhead_manager.create_bulkhead( "critical", # 결제, 핵심 기능 BulkheadConfig(max_concurrent=20, max_queue_size=50, timeout=10.0) ) await self.bulkhead_manager.create_bulkhead( "standard", # 일반 검색, 추천 BulkheadConfig(max_concurrent=15, max_queue_size=30, timeout=20.0) ) await self.bulkhead_manager.create_bulkhead( "background", # 분석, 로깅 BulkheadConfig(max_concurrent=5, max_queue_size=10, timeout=30.0) ) async def call_api( self, bulkhead_name: str, endpoint: str, data: dict, priority: str = "standard" ): """우선순위에 따른 벌크헤드 처리 API 호출""" bh = await self.bulkhead_manager.get_bulkhead(priority) if not bh: raise ValueError(f"Unknown bulkhead: {priority}") async with bh.acquire(): headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } async with self._session.post( f"{self.base_url}{endpoint}", json=data, headers=headers ) as response: return await response.json()

사용 예시

async def main(): client = HolySheepBulkheadClient(api_key="YOUR_HOLYSHEEP_API_KEY") await client.init_bulkheads() # 동시에 여러 요청 시뮬레이션 tasks = [] # 중요도 높은 요청 for i in range(5): tasks.append(client.call_api( bulkhead_name="critical-api", endpoint="/chat/completions", data={"model": "gpt-4.1", "messages": [{"role": "user", "content": f"중요 {i}"}]}, priority="critical" )) # 일반 우선순위 요청 for i in range(10): tasks.append(client.call_api( bulkhead_name="standard-api", endpoint="/chat/completions", data={"model": "claude-sonnet-4-20250514", "messages": [{"role": "user", "content": f"일반 {i}"}]}, priority="standard" )) #后台 처리 요청 for i in range(3): tasks.append(client.call_api( bulkhead_name="background-api", endpoint="/embeddings", data={"model": "text-embedding-3-small", "input": f"배경 {i}"}, priority="background" )) results = await asyncio.gather(*tasks, return_exceptions=True) # 통계 출력 stats = await client.bulkhead_manager.get_all_stats() print("\n📊 벌크헤드 통계:") for name, stat in stats.items(): print(f" {name}: 활성 {stat['active']}/{stat['max_concurrent']}, " f"거부 {stat['rejected']}, 활용률 {stat['utilization']:.1%}") if __name__ == "__main__": asyncio.run(main())

기업 RAG 시스템에 적용하기

저희 팀에서 실제 운영 중인 엔터프라이즈 RAG(Retrieval-Augmented Generation) 시스템에 이 패턴들을 적용한 사례를 공유하겠습니다. 해당 시스템은:

초기에는 단일 API 클라이언트로 모든 요청을 처리했으나, 특정 부서의 과도한 사용이 다른 부서에게 영향을 미치는 문제가 반복되었습니다. 벌크헤드 패턴 도입 후:

비용 최적화와의 시너지

서킷 브레이커와 벌크헤드는 단순한 장애 방지 외에도 비용 최적화에 큰 도움이 됩니다. HolySheep AI의 가격표를 활용하면:

벌크헤드로 중요도 낮은 요청을 Gemini나 DeepSeek으로 라우팅하면:

import asyncio
from typing import Literal

class SmartRouter:
    """우선순위 기반 모델 라우팅 + 벌크헤드 통합"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.bulkhead_manager = BulkheadManager()
    
    async def init(self):
        """고가용성 모델용 벌크헤드"""
        await self.bulkhead_manager.create_bulkhead(
            "premium", 
            BulkheadConfig(max_concurrent=10, timeout=15.0)
        )
        await self.bulkhead_manager.create_bulkhead(
            "economy",
            BulkheadConfig(max_concurrent=30, timeout=30.0)
        )
    
    async def route_and_execute(
        self, 
        prompt: str, 
        priority: Literal["critical", "normal", "low"] = "normal"
    ):
        """우선순위에 따른 모델 선택 및 실행"""
        
        # 모델 선택 로직
        if priority == "critical":
            model = "gpt-4.1"
            bulkhead = "premium"
            estimated_cost_per_1k = 0.008  # $8/1M 토큰
        elif priority == "normal":
            model = "gemini-2.5-flash"
            bulkhead = "economy"
            estimated_cost_per_1k = 0.0025  # $2.50/1M 토큰
        else:
            model = "deepseek-v3.2"
            bulkhead = "economy"
            estimated_cost_per_1k = 0.00042  # $0.42/1M 토큰
        
        bh = await self.bulkhead_manager.get_bulkhead(bulkhead)
        
        async with bh.acquire():
            # HolySheep AI API 호출
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    f"{self.base_url}/chat/completions",
                    json={
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "max_tokens": 500
                    },
                    headers={"Authorization": f"Bearer {self.api_key}"}
                ) as resp:
                    result = await resp.json()
                    
                    # 비용 계산
                    tokens_used = result.get("usage", {}).get("total_tokens", 0)
                    actual_cost = (tokens_used / 1_000_000) * (estimated_cost_per_1k * 1000)
                    
                    return {
                        "model": model,
                        "response": result["choices"][0]["message"]["content"],
                        "tokens": tokens_used,
                        "estimated_cost_usd": actual_cost,
                        "priority": priority
                    }
    
    async def batch_process(self, requests: list):
        """배치 처리 - 자동 우선순위 분배"""
        results = await asyncio.gather(
            *[self.route_and_execute(**req) for req in requests],
            return_exceptions=True
        )
        
        # 비용 보고서 생성
        total_cost = sum(
            r.get("estimated_cost_usd", 0) 
            for r in results if isinstance(r, dict)
        )
        
        print(f"배치 처리 완료: {len(results)}건, 총 비용: ${total_cost:.4f}")
        return results

모니터링 및 복구 전략

패턴을 적용했다면 반드시 모니터링 시스템도 함께 구축해야 합니다. 저자가 추천하는 핵심 모니터링 지표:

import time
from dataclasses import dataclass, asdict
from typing import List
import json

@dataclass
class HealthReport:
    timestamp: float
    circuit_states: dict
    bulkhead_stats: dict
    total_requests: int
    failed_requests: int
    avg_latency_ms: float
    p95_latency_ms: float
    
    def to_json(self) -> str:
        return json.dumps(asdict(self), indent=2)
    
    def is_healthy(self) -> bool:
        """전체 시스템 건강 상태 판별"""
        # 서킷이 모두 닫혀있는지 확인
        circuits_ok = all(
            state == "closed" 
            for state in self.circuit_states.values()
        )
        
        # 벌크헤드 활용률이 80% 이하인지 확인
        bulkheads_ok = all(
            stat["utilization"] < 0.80 
            for stat in self.bulkhead_stats.values()
        )
        
        # 실패율이 5% 이하인지 확인
        failure_rate = self.failed_requests / max(self.total_requests, 1)
        failure_ok = failure_rate < 0.05
        
        return circuits_ok and bulkheads_ok and failure_ok

class HealthMonitor:
    def __init__(self):
        self.reports: List[HealthReport] = []
        self.request_times: List[float] = []
    
    def record_request(self, latency_ms: float, success: bool):
        self.request_times.append(latency_ms)
    
    def generate_report(
        self, 
        circuit_breakers: dict, 
        bulkhead_manager
    ) -> HealthReport:
        now = time.time()
        
        # P95 계산
        sorted_times = sorted(self.request_times)
        p95_index = int(len(sorted_times) * 0.95)
        p95 = sorted_times[p95_index] if sorted_times else 0
        
        report = HealthReport(
            timestamp=now,
            circuit_states={
                name: cb.state.value 
                for name, cb in circuit_breakers.items()
            },
            bulkhead_stats=bulkhead_manager.get_all_stats(),
            total_requests=len(self.request_times),
            failed_requests=sum(1 for t in self.request_times if t < 0),
            avg_latency_ms=sum(self.request_times) / max(len(self.request_times), 1),
            p95_latency_ms=p95
        )
        
        self.reports.append(report)
        return report

사용 예시

monitor = HealthMonitor() circuits = {"gpt4": ai_circuit, "claude": CircuitBreaker(...)}

주기적 헬스체크

async def health_check_loop(): while True: report = monitor.generate_report(circuits, bulkhead_manager) status = "✅ Healthy" if report.is_healthy() else "⚠️ Degraded" print(f"{status} | P95: {report.p95_latency_ms:.0f}ms | " f" Circuits: {report.circuit_states}") await asyncio.sleep(60) # 1분마다 체크

자주 발생하는 오류 해결

1. 서킷 브레이커가 너무 자주 열릴 때

failure_threshold가 너무 낮거나 timeout이 너무 짧은 경우, 정상적인 요청도 차단됩니다.

# ❌ 잘못된 설정 (너무 민감)
config_too_sensitive = CircuitBreakerConfig(
    failure_threshold=2,  # 2회 실패만으로 열림
    timeout=5.0  # 5초만 대기
)

✅ 권장 설정 (안정적)

config_recommended = CircuitBreakerConfig( failure_threshold=5, # 5회 연속 실패 시 success_threshold=2, # 반열림 상태에서 2회 성공 후 복구 timeout=30.0 # 30초 대기 후 복구 시도 )

💡 상황별 조정 가이드

높은 트래픽 API: failure_threshold=10, timeout=60

배치 처리: failure_threshold=3, timeout=120

실시간 채팅: failure_threshold=5, timeout=15

2. 벌크헤드 대기열이 가득 찼을 때

대기열이 꽉 차면 TimeoutError가 발생합니다. 이때는 즉시 실패처리보다 폴백(fallback) 응답을 제공하는 것이 좋습니다.

async def call_with_fallback(
    prompt: str, 
    primary_bulkhead: str = "standard",
    fallback_response: str = "죄송합니다. 현재 혼잡하여 나중에 다시 시도해주세요."
):
    bh = await bulkhead_manager.get_bulkhead(primary_bulkhead)
    
    try:
        async with bh.acquire():
            # 실제 API 호출
            return await holy_sheep_client.chat(prompt)
    
    except BulkheadTimeoutError:
        # 폴백: 캐시된 응답 또는 기본 메시지 반환
        cached = await check_cache(prompt)
        if cached:
            return cached
        return {"content": fallback_response, "from_cache": False, "fallback": True}
    
    except Exception as e:
        # 최종 폴백
        return {"content": fallback_response, "error": str(e)}

3. HolySheep API 연결 시간 초과

network 지연이나 HolySheep 게이트웨이 문제로 타임아웃이 발생합니다. 재시도 로직과 함께 사용하세요.

from tenacity import retry, stop_after_attempt, wait_exponential

async def resilient_api_call(prompt: str, max_retries: int = 3):
    """재시도 로직이 포함된 안정적 API 호출"""
    
    @retry(
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError))
    )
    async def _call():
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.holysheep.ai/v1/chat/completions",
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": prompt}]
                },
                headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
                timeout=aiohttp.ClientTimeout(total=30)
            ) as resp:
                if resp.status == 429:  # Rate limit
                    raise RateLimitError("Rate limit exceeded")
                return await resp.json()
    
    try:
        return await _call()
    except RateLimitError:
        # Rate limit 시에는 서킷 브레이커에 위임
        await circuit_breaker.call(_call)

4. 여러 벌크헤드 간의 교착 상태

여러 벌크헤드를 순차적으로 획득할 때 교착 상태가 발생할 수 있습니다. 항상 동일한 순서로 획득하세요.

# ❌ 교착 상태 가능성 있는 코드
async def bad_order(bh1, bh2):
    async with bh1.acquire():
        await expensive_operation()
        async with bh2.acquire():  # bh1을 보유한 상태로 bh2 대기
            ...

✅ 항상 동일한 순서로 획득

BULKHEAD_ORDER = ["critical", "standard", "background"] async def safe_acquire_multiple(*bulkheads): """안전한 다중 벌크헤드 획득""" acquired = [] try: for bh_name in sorted(bulkheads, key=lambda x: BULKHEAD_ORDER.index(x)): bh = await bulkhead_manager.get_bulkhead(bh_name) await bh.acquire() acquired.append(bh_name) # 실제 작업 수행 yield finally: # 역순으로 해제 for bh_name in reversed(acquired): bh = await bulkhead_manager.get_bulkhead(bh_name) bh.semaphore.release()

5. API 응답 형식 불일치

HolySheep AI의 응답 구조와 OpenAI 표준이 다를 수 있습니다. 응답 정규화가 필요합니다.

from typing import Dict, Any

class ResponseNormalizer:
    """HolySheep AI 응답을 표준 형식으로 변환"""
    
    @staticmethod
    def normalize_chat_response(response: Dict[str, Any]) -> Dict[str, Any]:
        # HolySheep AI가 OpenAI 호환 형식을 반환하지만,
        # 추가 검증과 정규화를 통해 일관성 확보
        return {
            "id": response.get("id", ""),
            "model": response.get("model", ""),
            "content": response["choices"][0]["message"]["content"],
            "usage": {
                "prompt_tokens": response.get("usage", {}).get("prompt_tokens", 0),
                "completion_tokens": response.get("usage", {}).get("completion_tokens", 0),
                "total_tokens": response.get("usage", {}).get("total_tokens", 0)
            },
            "finish_reason": response["choices"][0].get("finish_reason", "stop")
        }
    
    @staticmethod
    def validate_response(response: Dict[str, Any]) -> bool:
        required_fields = ["choices"]
        return all(field in response for field in required_fields)

async def safe_api_call(prompt: str) -> Dict[str, Any]:
    """응답 검증이 포함된 안전한 API 호출"""
    response = await client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}]
    )
    
    # 응답을 dict로 변환
    response_dict = response.model_dump() if hasattr(response, 'model_dump') else response
    
    if not ResponseNormalizer.validate_response(response_dict):
        raise ValueError("Invalid API response format")
    
    return ResponseNormalizer.normalize_chat_response(response_dict)

결론

AI API를 안정적으로 운영하려면 단순한 에러 핸들링을 넘어서 체계적인 보호 패턴이 필요합니다. 서킷 브레이커는 연쇄적 장애를 방지하고, 벌크헤드는 리소스 경합을 해결합니다. 이 두 패턴을 함께 사용하면:

저는 개인 프로젝트부터 대규모 엔터프라이즈 시스템까지 다양한 규모에서 이 패턴들을 적용해 왔고, HolySheep AI의 안정적인 게이트웨이 인프라와 결합하면 훨씬 효과적인 결과를 얻을 수 있습니다.

특히 HolySheep AI는 단일 API 키로 여러 모델을 통합 관리할 수 있어, 벌크헤드 패턴과 결합하면 우선순위 기반 모델 라우팅을 간편하게 구현할 수 있습니다. 이제 안정적인 AI 서비스를 구축해 보세요!

👉 HolySheep AI 가입하고 무료 크레딧 받기