AI 모델 서비스 건강 상태 점검과 자동 장애 전환 설계 완벽 가이드

HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교

서론: 왜 모델 서비스 건강 상태 점검과 자동 장애 전환이 중요한가

저는 최근 대량 트래픽을 처리하는 AI 기반 프로젝트를 수행하면서 모델服务的 가용성과 안정성이 얼마나 중요한지 뼈저리게 체감했습니다. 단일 API 엔드포인트에 의존하는 구조는 한 번의 서비스 중단으로 전체 시스템이 마비될 수 있습니다. HolySheep AI는 이러한 문제점을 해결하기 위해 기본적인 건강 상태 점검 기능을 제공하며, 개발자는 이를 활용하여 자동 장애 전환 체계를 구축할 수 있습니다.

건강 상태 점검(Health Check) 아키텍처 설계

건강 상태 점검은 크게 세 가지 유형으로 나눌 수 있습니다:

• 능동적 점검(Active Check): 주기적으로 서비스 엔드포인트에 요청을 보내 상태 확인
• 수동적 점검(Passive Check): 실제 요청의 성공/실패율을 기반으로 상태 판단
• 심층 점검(Deep Check): 실제 모델 추론 요청을 보내 응답 품질까지 검증

실전 구현: Python 기반 건강 상태 점검 및 자동 장애 전환 시스템

import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ServiceStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    UNKNOWN = "unknown"

@dataclass
class ModelEndpoint:
    name: str
    base_url: str
    api_key: str
    status: ServiceStatus = ServiceStatus.UNKNOWN
    consecutive_failures: int = 0
    consecutive_successes: int = 0
    last_check_time: float = 0
    last_success_time: float = 0
    average_latency: float = 0
    total_requests: int = 0
    failed_requests: int = 0

class HealthCheckConfig:
    def __init__(
        self,
        check_interval: int = 30,          # 健康检查间隔(秒)
        timeout: int = 10,                 # 请求超时时间(秒)
        healthy_threshold: int = 3,        # 连续成功次数阈值
        unhealthy_threshold: int = 3,      # 连续失败次数阈值
        latency_threshold: int = 5000,     # 延迟阈值(毫秒)
        degraded_threshold: int = 3000     # 降级阈值(毫秒)
    ):
        self.check_interval = check_interval
        self.timeout = timeout
        self.healthy_threshold = healthy_threshold
        self.unhealthy_threshold = unhealthy_threshold
        self.latency_threshold = latency_threshold
        self.degraded_threshold = degraded_threshold

@dataclass
class HealthCheckResult:
    endpoint: ModelEndpoint
    is_healthy: bool
    latency_ms: float
    error_message: Optional[str] = None
    timestamp: float = field(default_factory=time.time)

class ModelServiceHealthChecker:
    def __init__(
        self,
        config: HealthCheckConfig,
        on_status_change: Optional[callable] = None
    ):
        self.config = config
        self.endpoints: List[ModelEndpoint] = []
        self.on_status_change = on_status_change
        self._session: Optional[aiohttp.ClientSession] = None
        self._running = False

    def add_endpoint(self, name: str, base_url: str, api_key: str):
        endpoint = ModelEndpoint(name=name, base_url=base_url, api_key=api_key)
        self.endpoints.append(endpoint)
        logger.info(f"엔드포인트 추가됨: {name} - {base_url}")

    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            timeout = aiohttp.ClientTimeout(total=self.config.timeout)
            self._session = aiohttp.ClientSession(timeout=timeout)
        return self._session

    async def check_endpoint(self, endpoint: ModelEndpoint) -> HealthCheckResult:
        start_time = time.time()
        error_msg = None
        is_healthy = False

        # HolySheep AI 健康检查端点
        health_url = f"{endpoint.base_url}/health"

        try:
            session = await self._get_session()
            async with session.get(
                health_url,
                headers={"Authorization": f"Bearer {endpoint.api_key}"}
            ) as response:
                latency_ms = (time.time() - start_time) * 1000

                if response.status == 200:
                    is_healthy = True
                    endpoint.last_success_time = time.time()
                    endpoint.average_latency = (
                        endpoint.average_latency * 0.7 + latency_ms * 0.3
                    )
                    logger.info(
                        f"✓ {endpoint.name} 건강 상태 양호 - "
                        f"지연 시간: {latency_ms:.0f}ms"
                    )
                else:
                    error_msg = f"HTTP {response.status}"
                    logger.warning(
                        f"✗ {endpoint.name} 비정상 응답: {response.status}"
                    )

        except asyncio.TimeoutError:
            latency_ms = self.config.timeout * 1000
            error_msg = f"요청 시간 초과 ({self.config.timeout}초)"
            logger.warning(f"✗ {endpoint.name} 요청 시간 초과")
        except aiohttp.ClientError as e:
            latency_ms = (time.time() - start_time) * 1000
            error_msg = f"클라이언트 오류: {str(e)}"
            logger.error(f"✗ {endpoint.name} 클라이언트 오류: {e}")
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            error_msg = f"예상치 못한 오류: {str(e)}"
            logger.error(f"✗ {endpoint.name} 예상치 못한 오류: {e}")

        return HealthCheckResult(
            endpoint=endpoint,
            is_healthy=is_healthy,
            latency_ms=latency_ms,
            error_message=error_msg
        )

    def update_endpoint_status(self, result: HealthCheckResult):
        endpoint = result.endpoint
        endpoint.last_check_time = time.time()

        if result.is_healthy:
            endpoint.consecutive_successes += 1
            endpoint.consecutive_failures = 0

            # 지연 시간 기반 상태 평가
            if result.latency_ms > self.config.latency_threshold:
                new_status = ServiceStatus.UNHEALTHY
            elif result.latency_ms > self.config.degraded_threshold:
                if endpoint.status != ServiceStatus.HEALTHY:
                    new_status = ServiceStatus.DEGRADED
                else:
                    new_status = ServiceStatus.HEALTHY
            else:
                new_status = ServiceStatus.HEALTHY

            # 성공 횟수阈值 도달 시 상태 변경
            if endpoint.consecutive_successes >= self.config.healthy_threshold:
                if endpoint.status != ServiceStatus.HEALTHY:
                    old_status = endpoint.status
                    endpoint.status = ServiceStatus.HEALTHY
                    self._notify_status_change(endpoint, old_status)

        else:
            endpoint.consecutive_failures += 1
            endpoint.consecutive_successes = 0

            if endpoint.consecutive_failures >= self.config.unhealthy_threshold:
                if endpoint.status != ServiceStatus.UNHEALTHY:
                    old_status = endpoint.status
                    endpoint.status = ServiceStatus.UNHEALTHY
                    self._notify_status_change(endpoint, old_status)

    def _notify_status_change(self, endpoint: ModelEndpoint, old_status: ServiceStatus):
        logger.warning(
            f"⚠️ {endpoint.name} 상태 변경: {old_status.value} → {endpoint.status.value}"
        )
        if self.on_status_change:
            asyncio.create_task(self._call_status_change_handler(endpoint, old_status))

    async def _call_status_change_handler(self, endpoint: ModelEndpoint, old_status: ServiceStatus):
        try:
            await self.on_status_change(endpoint, old_status, endpoint.status)
        except Exception as e:
            logger.error(f"상태 변경 핸들러 실행 실패: {e}")

    async def health_check_loop(self):
        self._running = True
        logger.info("건강 상태 점검 루프 시작")

        while self._running:
            tasks = [self.check_endpoint(ep) for ep in self.endpoints]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for result in results:
                if isinstance(result, HealthCheckResult):
                    self.update_endpoint_status(result)
                elif isinstance(result, Exception):
                    logger.error(f"건강 상태 점검 중 예외 발생: {result}")

            await asyncio.sleep(self.config.check_interval)

    async def start(self):
        await self.health_check_loop()

    async def stop(self):
        self._running = False
        if self._session and not self._session.closed:
            await self._session.close()
        logger.info("건강 상태 점검 시스템 종료")

    def get_healthy_endpoints(self) -> List[ModelEndpoint]:
        return [ep for ep in self.endpoints if ep.status == ServiceStatus.HEALTHY]

    def get_best_endpoint(self) -> Optional[ModelEndpoint]:
        healthy = self.get_healthy_endpoints()
        if not healthy:
            return None
        return min(healthy, key=lambda ep: ep.average_latency)

    def get_status_report(self) -> Dict:
        return {
            "total_endpoints": len(self.endpoints),
            "healthy": len([ep for ep in self.endpoints if ep.status == ServiceStatus.HEALTHY]),
            "degraded": len([ep for ep in self.endpoints if ep.status == ServiceStatus.DEGRADED]),
            "unhealthy": len([ep for ep in self.endpoints if ep.status == ServiceStatus.UNHEALTHY]),
            "endpoints": [
                {
                    "name": ep.name,
                    "status": ep.status.value,
                    "latency_ms": ep.average_latency,
                    "total_requests": ep.total_requests,
                    "failed_requests": ep.failed_requests,
                    "last_check": ep.last_check_time
                }
                for ep in self.endpoints
            ]
        }

자동 장애 전환 및 회귀 시스템 구현

import asyncio
import aiohttp
import json
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import time
import random

@dataclass
class RequestMetrics:
    endpoint_name: str
    start_time: float
    latency_ms: float
    success: bool
    error_type: Optional[str] = None

class CircuitBreakerState(Enum):
    CLOSED = "closed"      # 정상 작동
    OPEN = "open"          # 차단됨
    HALF_OPEN = "half_open"  # 테스트 중

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        half_open_requests: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_requests = half_open_requests
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self.state = CircuitBreakerState.CLOSED
        self.half_open_successes = 0

    def record_success(self):
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.half_open_successes += 1
            if self.half_open_successes >= self.half_open_requests:
                self.state = CircuitBreakerState.CLOSED
                self.failure_count = 0
                self.half_open_successes = 0
        else:
            self.failure_count = max(0, self.failure_count - 1)

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.state == CircuitBreakerState.HALF_OPEN:
            self.state = CircuitBreakerState.OPEN
            self.half_open_successes = 0
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN

    def can_execute(self) -> bool:
        if self.state == CircuitBreakerState.CLOSED:
            return True

        if self.state == CircuitBreakerState.OPEN:
            if self.last_failure_time:
                elapsed = time.time() - self.last_failure_time
                if elapsed >= self.recovery_timeout:
                    self.state = CircuitBreakerState.HALF_OPEN
                    self.half_open_successes = 0
                    return True
            return False

        if self.state == CircuitBreakerState.HALF_OPEN:
            return self.half_open_successes < self.half_open_requests

        return False

class AutomaticFailoverRouter:
    def __init__(
        self,
        endpoints: List[ModelEndpoint],
        health_checker: ModelServiceHealthChecker,
        circuit_breaker_config: Optional[Dict] = None
    ):
        self.endpoints = endpoints
        self.health_checker = health_checker
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}

        cb_config = circuit_breaker_config or {}
        for endpoint in endpoints:
            self.circuit_breakers[endpoint.name] = CircuitBreaker(
                failure_threshold=cb_config.get('failure_threshold', 5),
                recovery_timeout=cb_config.get('recovery_timeout', 60),
                half_open_requests=cb_config.get('half_open_requests', 3)
            )

        self.current_index = 0
        self.metrics: List[RequestMetrics] = []
        self.total_requests = 0
        self.failed_requests = 0

    def _get_next_endpoint_index(self) -> int:
        """순환 방식 엔드포인트 선택"""
        return (self.current_index + 1) % len(self.endpoints)

    def _select_endpoint(self) -> Optional[ModelEndpoint]:
        """엔드포인트 선택 로직"""
        # Circuit Breaker 상태 확인
        available = []
        for endpoint in self.endpoints:
            cb = self.circuit_breakers.get(endpoint.name)
            if cb and cb.can_execute():
                available.append(endpoint)

        if not available:
            logger.error("사용 가능한 엔드포인트 없음")
            return None

        # 상태별 우선순위 정렬
        def sort_key(ep):
            cb = self.circuit_breakers.get(ep.name)
            if cb and cb.state == CircuitBreakerState.HALF_OPEN:
                return 0  # 테스트 중인 엔드포인트 우선
            if ep.status == ServiceStatus.HEALTHY:
                return 1
            elif ep.status == ServiceStatus.DEGRADED:
                return 2
            return 3

        available.sort(key=sort_key)

        # 지연 시간 기반 가중치 선택
        if len(available) > 1:
            weights = []
            for ep in available:
                # 평균 지연 시간의 역수를 가중치로 사용
                weight = 1 / (ep.average_latency + 1)
                weights.append(weight)

            total_weight = sum(weights)
            rand = random.uniform(0, total_weight)

            cumulative = 0
            for i, w in enumerate(weights):
                cumulative += w
                if rand <= cumulative:
                    return available[i]

        return available[0]

    async def request(
        self,
        model: str,
        messages: List[Dict],
        temperature: float = 0.7,
        max_tokens: int = 1000,
        **kwargs
    ) -> Dict[str, Any]:
        """API 요청 실행 및 자동 장애 전환"""
        start_time = time.time()
        endpoint = self._select_endpoint()

        if not endpoint:
            raise Exception("요청 가능한 엔드포인트가 없습니다")

        cb = self.circuit_breakers.get(endpoint.name)
        last_error = None

        # 최대 재시도 횟수
        max_retries = len(self.endpoints) + 2
        retry_count = 0

        while retry_count < max_retries:
            if not endpoint:
                retry_count += 1
                endpoint = self._select_endpoint()
                if not endpoint:
                    continue

            cb = self.circuit_breakers.get(endpoint.name)

            if cb and not cb.can_execute():
                logger.warning(f"{endpoint.name} Circuit Breaker 열림, 다음 엔드포인트 시도")
                endpoint = None
                retry_count += 1
                continue

            try:
                result = await self._execute_request(
                    endpoint, model, messages, temperature, max_tokens, **kwargs
                )

                # 성공 시 Circuit Breaker 업데이트
                if cb:
                    cb.record_success()

                # 메트릭 기록
                latency = (time.time() - start_time) * 1000
                self.metrics.append(RequestMetrics(
                    endpoint_name=endpoint.name,
                    start_time=start_time,
                    latency_ms=latency,
                    success=True
                ))

                endpoint.total_requests += 1
                self.total_requests += 1
                return result

            except Exception as e:
                last_error = e
                logger.error(f"{endpoint.name} 요청 실패: {e}")

                # 실패 시 Circuit Breaker 업데이트
                if cb:
                    cb.record_failure()

                endpoint.consecutive_failures += 1
                endpoint.failed_requests += 1
                self.failed_requests += 1

                # 메트릭 기록
                latency = (time.time() - start_time) * 1000
                self.metrics.append(RequestMetrics(
                    endpoint_name=endpoint.name,
                    start_time=start_time,
                    latency_ms=latency,
                    success=False,
                    error_type=type(e).__name__
                ))

                endpoint = None
                retry_count += 1

        raise Exception(f"모든 엔드포인트 장애 전환 실패: {last_error}")

    async def _execute_request(
        self,
        endpoint: ModelEndpoint,
        model: str,
        messages: List[Dict],
        temperature: float,
        max_tokens: int,
        **kwargs
    ) -> Dict[str, Any]:
        """실제 API 요청 실행"""
        session = await self.health_checker._get_session()

        headers = {
            "Authorization": f"Bearer {endpoint.api_key}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            **kwargs
        }

        url = f"{endpoint.base_url}/chat/completions"

        async with session.post(url, headers=headers, json=payload) as response:
            if response.status != 200:
                error_text = await response.text()
                raise Exception(f"API 오류: {response.status} - {error_text}")

            return await response.json()

    def get_failover_stats(self) -> Dict:
        """장애 전환 통계 반환"""
        total = len(self.metrics)
        successful = len([m for m in self.metrics if m.success])
        failed = total - successful

        endpoint_stats = {}
        for endpoint in self.endpoints:
            endpoint_metrics = [m for m in self.metrics if m.endpoint_name == endpoint.name]
관련 리소스
📚 AI API 기술 문서
💰 요금제 보기
📖 개발자 문서
🚀 무료 가입
관련 문서
Flutter AI 대화 애플리케이션 개발: HolySheep AI로跨플랫폼 모바일 통합하기
Kotlin Ktor로 AI API 호출하기: 코루틴 동시성 완벽 가이드
MLflow로 파인튜닝 모델 버전 관리 및 자동 배포 파이프라인 구축