모델 서빙과 API

1추론 서비스 아키텍처

학습된 AI 모델을 실제 프로덕션 환경에 배포하기 위해서는 고성능 추론 서버가 필요합니다. LLM PaaS 플랫폼은 FastAPI 기반의 REST API와 Triton Inference Server를 결합한 하이브리드 아키텍처를 사용합니다.

Model Serving Architecture

클라이언트 계층

React Web App Mobile App Chatbot Interface External API

↓

API Gateway

• FastAPI REST Endpoints

• OAuth 2.0 Authentication

• Rate Limiting & Throttling

• Request Validation

↓

경량 모델 추론

FastAPI + PyTorch

• Classification

• Anomaly Detection

• Simple RAG Query

고성능 모델 추론

Triton Inference Server

• LLM Generation

• TensorRT Optimization

• Dynamic Batching

↓

모델 저장소

MLflow Model Registry HuggingFace Hub S3 Model Storage

추론 흐름

요청 수신 → 인증 검증 → 모델 선택 → 전처리 → 추론 → 후처리 → 응답

⚡

FastAPI 경량 추론

분류, 이상탐지 등 간단한 모델은 FastAPI로 직접 서빙하여 레이턴시 최소화

🚀

Triton 고성능 추론

LLM, Transformer 등 대형 모델은 Triton으로 TensorRT 최적화 및 배치 처리

🔐

OAuth 2.0 인증

JWT 토큰 기반 인증으로 API 접근 제어 및 사용량 추적

📊

실시간 모니터링

Prometheus + Grafana로 추론 레이턴시, 처리량, 에러율 실시간 모니터링

2FastAPI 추론 서비스 구현

FastAPI는 Python 기반의 고성능 웹 프레임워크로, 자동 API 문서 생성(OpenAPI)과 타입 검증을 지원합니다. 간단한 이상 탐지 모델 서빙 예제입니다.

          FastAPI 추론 서버
          Pseudocode
        
# === FastAPI 추론 서버 수도코드 ===

# 1. 모델 관리자
클래스 ModelManager:
    models = {}  // 로드된 모델 저장소

    함수 load_model(모델명, MLflow경로):
        models[모델명] = MLflow.로드(경로)

    함수 predict(모델명, 데이터):
        모델 = models[모델명]
        모델.평가모드()
        반환 모델(텐서변환(데이터))

# 2. 서버 시작 시 모델 로드
이벤트 서버시작:
    manager.load_model("anomaly_detector", "models:/이상탐지/Production")
    manager.load_model("quality_classifier", "models:/품질분류/Production")

# 3. 요청/응답 스키마
요청 = {sensor_data: [실수 10~50개], equipment_id: 문자열}
응답 = {is_anomaly: 불린, anomaly_score: 실수, confidence: 실수}

# 4. 추론 엔드포인트
POST /api/v1/predict/anomaly:
    토큰검증(요청.토큰)
    데이터 = 전처리(요청.sensor_data)
    점수 = manager.predict("anomaly_detector", 데이터)
    이상여부 = (점수 > 0.8)
    반환 {is_anomaly, anomaly_score, confidence, recommendation}

# 5. 기타 엔드포인트
GET /api/v1/models → 로드된 모델 목록
GET /health → 서버 상태 확인

자동 API 문서화

FastAPI는 OpenAPI 스펙을 자동 생성합니다. 서버 실행 후 http://localhost:8000/docs 에 접속하면 Swagger UI로 모든 엔드포인트를 테스트할 수 있습니다.

3Triton Inference Server 설정

대형 LLM 모델은 NVIDIA Triton Inference Server로 서빙합니다. Triton은 TensorRT 최적화, 동적 배칭, 멀티 GPU 지원을 제공합니다.

          Triton 서버 설정
          Pseudocode
        
# === Triton Inference Server 설정 수도코드 ===

# 1. 모델 설정 (config.pbtxt)
모델명: "llm_model"
플랫폼: PyTorch
최대배치크기: 32

입력:
  - input_ids: INT64, 가변크기
  - attention_mask: INT64, 가변크기
출력:
  - logits: FP32, 어휘크기(50257)

# 2. GPU 인스턴스 설정
instance_group: GPU 2개 (gpu[0], gpu[1])

# 3. 동적 배칭 (성능 최적화)
dynamic_batching:
  선호배치크기: [8, 16, 32]
  최대대기시간: 1ms

# 4. TensorRT 최적화
optimization: TensorRT, FP16 정밀도

# 5. 모델 워밍업 (첫 추론 지연 방지)
warmup: 배치16, 시퀀스128로 사전실행

          Triton 클라이언트
          Pseudocode
        
# === Triton 클라이언트 수도코드 ===

클래스 TritonClient:
    함수 __init__(서버주소, 모델명):
        client = Triton.연결(서버주소)
        tokenizer = 로드("gpt2")

    함수 generate_text(프롬프트):
        // 1. 토크나이징
        input_ids = tokenizer.encode(프롬프트)
        attention_mask = [1] * len(input_ids)

        // 2. Triton 입력 생성
        triton입력 = [InferInput(input_ids), InferInput(attention_mask)]

        // 3. 추론 요청
        응답 = client.infer(모델명, 입력=triton입력)
        logits = 응답.as_numpy("logits")

        // 4. 디코딩
        예측토큰 = argmax(logits)
        생성텍스트 = tokenizer.decode(예측토큰)
        반환 생성텍스트

# 사용 예시
client = TritonClient()
만약 client.서버준비() AND client.모델준비():
    결과 = client.generate_text("품질 불량의 주요 원인은")
    출력(결과)

4gRPC 고성능 서빙

낮은 레이턴시가 중요한 경우 REST API 대신 gRPC를 사용합니다. gRPC는 Protocol Buffers 기반으로 HTTP/2보다 빠른 통신이 가능합니다.

          gRPC 서비스 정의
          Pseudocode
        
# === gRPC 서비스 수도코드 ===

# 1. 서비스 정의 (inference.proto)
서비스 InferenceService:
    PredictAnomaly(요청) → 응답          // 단일 추론
    PredictBatch(배치요청) → 배치응답    // 배치 추론
    StreamPredict(스트림) → 스트림      // 실시간 스트리밍

# 2. 메시지 구조
요청 = {sensor_data: [float], equipment_id: string}
응답 = {is_anomaly: bool, anomaly_score: float, confidence: float}

# 3. gRPC 서버 구현
클래스 InferenceServicer:
    함수 __init__():
        모델 = torch.load("anomaly_model.pt")

    함수 PredictAnomaly(요청):
        데이터 = numpy변환(요청.sensor_data)
        점수 = 모델(텐서(데이터))
        이상여부 = (점수 > 0.8)
        반환 응답(이상여부, 점수, 신뢰도)

    함수 StreamPredict(요청스트림):
        각 요청 에 대해:
            점수 = 모델(요청.데이터)
            yield 응답(점수, 타임스탬프)

# 4. 서버 시작
server = gRPC.서버(스레드풀=10)
server.포트추가(":50051")
server.시작()

gRPC vs REST 선택 기준

gRPC 사용: 마이크로서비스 간 내부 통신, 실시간 스트리밍, 극저 레이턴시 요구
REST 사용: 외부 클라이언트 연동, 브라우저 호환성, 디버깅 편의성

5배치 추론 최적화

여러 요청을 묶어 한 번에 처리하는 동적 배칭(Dynamic Batching)은 GPU 활용률을 높이고 처리량을 극대화합니다.

          동적 배칭 처리기 (수도코드)
          Pseudocode
        
# === 동적 배칭 처리기 수도코드 ===

클래스 DynamicBatcher:
    // 설정값
    model = 추론모델
    max_batch_size = 32      // 최대 배치 크기
    max_wait_ms = 10         // 최대 대기 시간 (ms)
    queue = []               // 요청 대기열
    processing = false       // 처리 중 플래그

    함수 predict(데이터):
        // 1. 요청을 큐에 추가
        future = 비동기퓨처.생성()
        queue.추가(데이터, future)

        // 2. 처리 중이 아니면 배치 처리 시작
        만약 NOT processing:
            비동기실행(process_batch)

        반환 대기(future)

    함수 process_batch():
        processing = true
        시작시간 = 현재시간()
        batch_items = []

        // 1. 큐에서 요청 수집 (최대 크기 또는 대기시간까지)
        반복 (batch_items.크기 < max_batch_size):
            경과시간 = (현재시간 - 시작시간) * 1000
            만약 경과시간 > max_wait_ms OR queue.비어있음:
                중단
            batch_items.추가(queue.꺼내기())

        // 2. 배치 생성 및 추론
        batch_data = 텐서스택(batch_items의 데이터들)
        batch_output = model.추론(batch_data)

        // 3. 결과를 각 요청에 분배
        각 i, future 에 대해 batch_items:
            future.결과설정(batch_output[i])

        processing = false

        // 4. 남은 요청이 있으면 재귀 처리
        만약 queue.비어있지않음:
            비동기실행(process_batch)

# 사용 예시
batcher = DynamicBatcher(모델, 배치크기=32, 대기시간=10ms)
결과들 = 병렬실행([batcher.predict(데이터) 각 데이터 100개])
출력("동적 배칭으로 100개 요청 처리 완료")

8x

처리량 향상

15ms

평균 레이턴시

85%

GPU 활용률

2000

QPS (초당 쿼리)

6API 문서화와 테스트

프로덕션 배포 전에 철저한 API 테스트와 문서화가 필수입니다. OpenAPI 스펙 생성과 pytest 기반 자동 테스트를 수행합니다.

          API 테스트 코드 (수도코드)
          Pseudocode
        
# === 추론 API 테스트 수도코드 ===

client = 테스트클라이언트(앱)

# 1. 헬스 체크 테스트
테스트 헬스체크:
    응답 = client.GET("/health")
    검증(응답.상태코드 == 200)
    검증(응답.데이터.status == "healthy")

# 2. 이상 탐지 추론 테스트
테스트 이상탐지추론:
    요청데이터 = {sensor_data: [10개 센서값], equipment_id: "PRESS_001"}
    응답 = client.POST("/api/v1/predict/anomaly", 데이터=요청데이터, 토큰="valid")

    검증(응답.상태코드 == 200)
    검증("is_anomaly" 존재)
    검증("anomaly_score" 존재)
    검증("confidence" 존재)

# 3. 인증 실패 테스트
테스트 잘못된토큰:
    응답 = client.POST("/api/v1/predict/anomaly", 토큰="invalid")
    검증(응답.상태코드 == 401)  // 인증 실패

# 4. 입력 검증 테스트
테스트 입력크기오류:
    요청데이터 = {sensor_data: [2개만]}  // 최소 10개 미만
    응답 = client.POST("/api/v1/predict/anomaly", 데이터=요청데이터)
    검증(응답.상태코드 == 422)  // 유효성 검사 오류

# 5. 동시 요청 처리 테스트
테스트 동시요청처리:
    응답들 = 병렬실행([요청() 반복 100번])
    검증(모든응답.상태코드 == 200)

# 6. 성능 테스트 (레이턴시)
테스트 레이턴시:
    시작 = 현재시간()
    응답 = client.POST("/api/v1/predict/anomaly", 데이터)
    레이턴시 = (현재시간 - 시작) * 1000
    검증(레이턴시 < 100ms)  // 100ms 이하

CI/CD 파이프라인 통합

이러한 테스트는 GitHub Actions 또는 GitLab CI에서 자동 실행됩니다. 모든 테스트가 통과해야만 프로덕션 배포가 가능하며, 성능 저하나 정확도 감소를 조기에 감지할 수 있습니다.

7API 엔드포인트 사양

LLM PaaS 플랫폼이 제공하는 주요 API 엔드포인트 목록입니다.

메서드	엔드포인트	설명	인증
POST	/api/v1/predict/anomaly	이상 탐지 추론	필수
POST	/api/v1/predict/quality	품질 예측	필수
POST	/api/v1/rag/query	RAG 지식 검색	필수
POST	/api/v1/llm/generate	LLM 텍스트 생성	필수
GET	/api/v1/models	로드된 모델 목록	필수
GET	/api/v1/metrics	추론 성능 메트릭	필수
POST	/api/v1/batch/predict	배치 추론 (비동기)	필수
GET	/health	헬스 체크	불필요
GET	/docs	API 문서 (Swagger UI)	불필요

8모니터링과 로깅

프로덕션 환경에서는 실시간 모니터링과 상세 로깅이 필수입니다. Prometheus로 메트릭을 수집하고 Grafana로 시각화합니다.

          추론 모니터링 (수도코드)
          Pseudocode
        
# === 추론 모니터링 수도코드 ===

# 1. Prometheus 메트릭 정의
메트릭:
    inference_requests_total  = Counter(모델, 엔드포인트, 상태)   // 총 요청 수
    inference_latency         = Histogram(모델, 엔드포인트)       // 레이턴시 분포
    model_gpu_utilization     = Gauge(gpu_id)                    // GPU 사용률
    active_requests           = Gauge()                          // 현재 활성 요청

# 2. 모니터링 데코레이터
함수 monitor_inference(모델명, 엔드포인트):
    반환 데코레이터:
        함수 wrapper(원본함수):
            active_requests.증가()
            시작시간 = 현재시간()
            상태 = "success"

            시도:
                결과 = 원본함수 실행
                반환 결과
            예외발생시:
                상태 = "error"
                예외 재발생
            마지막:
                // 메트릭 기록
                레이턴시 = 현재시간 - 시작시간
                inference_latency.기록(레이턴시)
                inference_requests_total.증가(모델명, 엔드포인트, 상태)
                active_requests.감소()

# 3. 사용 예시
@monitor_inference("anomaly_detector", "/predict/anomaly")
POST /api/v1/predict/anomaly:
    // 추론 로직 (자동으로 메트릭 수집됨)

# 4. 메트릭 서버 시작
Prometheus메트릭서버.시작(포트=8001)

핵심 모니터링 지표

레이턴시 (Latency): P50, P95, P99 백분위수
처리량 (Throughput): 초당 요청 수 (QPS)
에러율: 4xx, 5xx 에러 비율
모델 성능: 정확도, F1 스코어 실시간 추적