SSAFY 기업연계 회고 — 1,129줄짜리 CLI 스크립트를 FastAPI 서비스로 옮긴 기록

SSAFY 기업연계 프로젝트에서 내가 맡은 건 AI 이미지 분석 API 서비스였다. 팀 전체 과제는 기존 NestJS 백엔드를 FastAPI로 갈아엎는 마이그레이션이었고, 그중 AI 분석 파트가 분리돼서 나한테 떨어졌다.

처음 레포를 열었을 때 algorithm_original/ 이라는 폴더가 먼저 눈에 들어왔다. 이게 원본이고, 이걸 서비스로 만드는 게 내 일이었다. 스크립트 1,129줄을 그대로 @app.post("/") 뒤에 붙이면 되는 줄 알았는데 그게 아니었다.

왜 Python / FastAPI 였나

분석 대상이 소변 스트립(urine strip) 이미지 다. 의료 현장에서 쓰는 그 종이 조각. 환자가 직접 찍어 올리면 6가지 파라미터 (ketone, glucose, protein, pH, blood, white) 를 색상 변화로 판정해야 한다.

처리 파이프라인을 쭉 보면 이렇게 된다:

 이미지 업로드 (S3)
    ↓
 [1] QR / BlackBox 검출          ← OpenCV (Contour, Morphology)
    ↓
 [2] ROI(Region of Interest) 추출 ← OpenCV
    ↓
 [3] Strip 인식 & 기울기 정정     ← scipy.optimize.least_squares, CLAHE
    ↓
 [4] 색상 보정 (그림자·레퍼런스)   ← NumPy 3차 회귀
    ↓
 [5] KNN 레벨 분류 (6 parameter)  ← scikit-learn / cv2.ml.KNearest
    ↓
 JSON 응답

단계 하나하나가 Python 과학 계산 스택에 특화된 작업이다. OpenCV, NumPy, SciPy, scikit-learn — 이것들 전부 C/C++ 구현체를 Python이 얇게 감싼 바인딩이다. 그래서:

1. NestJS(TypeScript) 에서 하려면 sharp/jimp 같은 Node 이미지 라이브러리를 억지로 끌어와야 함 — 레거시 알고리즘이 이미 Python인 마당에 재작성은 비효율
2. C 레벨 연산은 GIL을 해제한다 — OpenCV 내부 루프는 파이썬 인터프리터 락에 걸리지 않음. 그래서 프로세스 워커 4개로도 병렬 처리 가능
3. FastAPI + Pydantic 조합이 JSON 스키마를 타입 레벨에서 보장 — 프론트가 받을 응답 shape가 코드로 고정됨

JS 진영에서 AI 분석 서비스를 굳이 NestJS로 유지할 이유가 없었다. “AI 파트는 Python으로 떼어낸다” 가 애초에 팀 합의였고, 내 역할은 그 Python을 서비스 형태로 만들어 올리는 것이었다.

출발점 — algorithm_original/ 진입점을 열었을 때

원본 진입점은 이렇게 생겨있었다.

import pipeline as algo
from config import *

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

bbox = algo.BlackBox(bbox_points, roi_size, qr_centers, qr_size, min_area_ratio=0.01)
strip = algo.Strip(bbox_points, patch_points, patch_hsize, ref_points, strip_length, bbox_padding)

모듈 import 시점에 전역 객체 두 개가 생성된다. 즉, 이 파일을 import 만 해도 BlackBox, Strip 인스턴스가 즉시 만들어진다. 단일 프로세스 CLI에서는 문제가 없지만, FastAPI 앱에서 이 모듈을 import 하면:

• 앱 기동 시 자동으로 객체가 생성돼 버림 (초기화 순서 제어 불가)
• 테스트 fixture를 못 꽂음 (이미 bbox 가 특정 설정으로 고정됨)
• 설정을 바꾸려면 모듈 reload 해야 함

진입 함수도 예외 처리가 공격적이다.

try:
    result_bbox = bbox(filename)
except:
    result['error'] = "UNKNOWN"
    return result

if not result_bbox['success']:
    result['error'] = "QR_BBOX"
    return result

try:
    result_strip = strip(result_bbox['bbox'], filename, result_bbox['roi'])
except:
    result['error'] = "UNKNOWN"
    return result
# ...

베어 except: 가 세 군데. 무슨 에러가 났는지 스택트레이스를 읽을 방법이 없다. 디버깅하다가 OpenCV가 뱉는 cv2.error 인지, NumPy의 IndexError 인지, S3가 죽은 건지 구분이 안 됐다.

그 외에 눈에 띄는 것들:

• 타입 힌팅 0건 — 함수 시그니처만 보고는 뭐가 들어가고 뭐가 나오는지 모름
• print() 만 남김 — 로깅 시스템 없음
• getopt.getopt(argv[1:], 'i:o:', ...) + CSV 출력 — 완전한 CLI 도구
• 결과물을 파일로 저장 — cv2.imwrite(filename_split + '_mask.jpg', bbox) 같은 부작용 곳곳에

1,129 줄을 한 번 쭉 읽고 나서 결정했다. 계층을 나누자.

6개 파라미터, 5개 에러 코드

리팩토링 들어가기 전에 도메인 정리부터.

그리고 실패 케이스가 5개. 이게 도메인 지식이다.

원본 코드에서는 UNKNOWN 이 전체 실패의 꽤 큰 비중이었다. except: 에 걸려서 뭐든 “UNKNOWN” 으로 갈 수 있었으니까. 리팩토링 목표 하나는 UNKNOWN 비율 줄이기 였다.

리팩토링 — 스크립트에서 서비스로

계층 분리

4개 층으로 나눴다.

 ┌───────────────────────────────────────────────┐
 │  app/api/analyze.py        ← FastAPI 엔드포인트 │
 └───────────────────┬───────────────────────────┘
                     ▼
 ┌───────────────────────────────────────────────┐
 │  app/schemas/              ← Pydantic 요청/응답 │
 └───────────────────┬───────────────────────────┘
                     ▼
 ┌───────────────────────────────────────────────┐
 │  app/services/             ← 비즈니스 로직       │
 │    - analysis_service.py   (분석 서비스)        │
 │    - s3_service.py         (이미지 다운로드)    │
 │    - knn_cache.py          (KNN 싱글톤)        │
 └───────────────────┬───────────────────────────┘
                     ▼
 ┌───────────────────────────────────────────────┐
 │  app/algorithm/            ← 순수 알고리즘      │
 │    (BlackBox · Strip · Estimate 모듈)          │
 └───────────────────────────────────────────────┘

원칙은 단순하다. 아래 층은 위 층을 모른다. 알고리즘 층은 numpy.ndarray 를 받아서 dict 를 뱉는 순수 함수. 서비스 층은 알고리즘을 조립해서 로깅·타이밍·에러 매핑을 얹는다. API 층은 FastAPI 의존성 주입과 Pydantic 검증만 담당.

전역 객체 제거

원본의 모듈 레벨 전역 객체 두 개를 서비스 클래스 안으로 밀어 넣었다.

class ImageAnalysisService:
    """Image analysis service"""

    def __init__(self):
        self.bbox = algo.BlackBox(bbox_points, roi_size, qr_centers, qr_size, min_area_ratio=0.01)
        self.strip = algo.Strip(bbox_points, patch_points, patch_hsize, ref_points, strip_length, bbox_padding)

그리고 파일 맨 아래에:

# 파일 하단의 싱글톤 인스턴스
analysis_service = ImageAnalysisService()

결과적으로 싱글톤이긴 한데, 만드는 책임과 쓰는 책임이 분리됐다. 테스트에서는 ImageAnalysisService() 를 새로 만들어 fixture로 꽂을 수 있다.

베어 except → 명시적 에러 매핑

원본의 try: ... except: result['error'] = "UNKNOWN" 패턴을 다 들어냈다. 대신 실패 경로마다 어떤 에러 코드로 번역될지를 명시했다.

bbox_start = time.time()
result_bbox = self.bbox(image)
bbox_elapsed = (time.time() - bbox_start) * 1000
logger.info(f"[TIMING] BlackBox: {bbox_elapsed:.0f}ms")

if not result_bbox['success']:
    result['error'] = "QR_BBOX"
    logger.error("BlackBox analysis failed: QR or BBOX not found")
    return result

# ... Strip 단계
result_strip = self.strip(result_bbox['bbox'], image, result_bbox['roi'])

if not result_strip['success']:
    if 'error' in result_strip:
        result['error'] = result_strip['error']   # NOSTRIP / REF / SHADOW
    else:
        result['error'] = 'STRIP'
    logger.error(f"Strip analysis failed: {result.get('error')}")
    return result

맨 바깥에 최종 그물 하나만 둔다.

except Exception as e:
    logger.error(f"Unexpected error during analysis: {e}", exc_info=True)
    result['error'] = "UNKNOWN"
    return result

이 UNKNOWN 은 “예상 밖의 진짜 예외” 전용. 도메인 실패는 모두 구조화된 에러 코드로 빠진다. 덕분에 UNKNOWN 이 찍히면 진짜 버그 라는 신호가 됐다.

파일 I/O 제거

원본은 스트립 인식 결과를 _mask.jpg 파일로 디스크에 저장했다. API 서버에서 이건 의미가 없다. 매 요청마다 디스크를 쓰면 오히려 병목이다. 그래서:

• 알고리즘 층은 경로 대신 numpy 배열을 받는다
• 결과물은 dict로 반환하고 파일 저장 안 함
• S3에서 받은 이미지도 디스크 거치지 않고 메모리에서 처리

# 메모리 버퍼로 다운로드
buffer = io.BytesIO()
self.s3_client.download_fileobj(bucket, key, buffer)
buffer.seek(0)

# Decode image from buffer
image_array = np.frombuffer(buffer.read(), dtype=np.uint8)
image = cv2.imdecode(image_array, cv2.IMREAD_COLOR)

BytesIO → np.frombuffer → cv2.imdecode 조합으로 S3 → OpenCV 까지 디스크를 한 번도 안 거친다. 컨테이너의 디스크 IOPS에도 의존하지 않음.

측정 가능한 로깅

모든 단계에 타이밍 로그를 달았다.

[TIMING] BlackBox: 87ms
[TIMING] Strip: 142ms
[TIMING] Estimate: 4ms
Analysis completed successfully. Strip type: 6-parameter

production에서 이걸 수집하면 어느 단계가 느린지 바로 보인다. 나중에 튜닝할 때 이 숫자가 기준선이 됐다.

성능 튜닝 — 측정 가능한 개선

KNN 모델 싱글톤 캐싱 (가장 큰 효과)

원본에서 estimate() 함수는 예측할 때마다 CSV를 읽고 KNN을 재학습했다. 6 파라미터 중 5개가 KNN을 쓰니까 요청 한 건당 5번씩 반복됐다.

서비스 계층에 KNN 싱글톤 캐시를 만들었다.

class KNNModelCache:
    """KNN 모델 메모리 캐시 (싱글톤)"""

    _instance = None
    _models: Dict[str, cv2.ml.KNearest] = {}
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

앱 기동 시(startup 이벤트) 한 번만 initialize() 를 부른다.

for category in categories:
    rows = train_data['category'] == category
    data = train_data.loc[rows]
    data_rgb = data.loc[:, ['R', 'G', 'B']]
    data_levels = data.loc[:, 'expected']

    array_data_rgb = data_rgb.to_numpy().astype(np.float32)
    array_data_levels = data_levels.to_numpy().astype(np.int32)

    knn = cv2.ml.KNearest_create()
    knn.train(array_data_rgb, cv2.ml.ROW_SAMPLE, array_data_levels)

    self._models[category] = knn

예측 시에는 학습 단계가 통째로 사라진다.

knn = self._models[category]
_, results, _, dist = knn.findNearest(rgb, k)
predicted_level = int(results[0, 0])

효과: 예측당 50~100ms → 1~5ms. 단일 요청에서 5번 반복되던 걸 생각하면 전체 응답 시간이 눈에 띄게 내려갔다.

싱글톤으로 만든 이유는 하나 더 있다. Uvicorn workers=4 로 띄우면 프로세스가 4개로 나뉘는데, 각 프로세스가 기동 시에 KNN을 1회씩만 학습하면 된다. 요청이 쇄도해도 학습은 총 4번이 끝. 프로세스 격리의 이점을 살린 셈.

CLAHE 객체 재사용

알고리즘 모듈에서 CLAHE (Contrast Limited Adaptive Histogram Equalization) 객체를 매 __call__ 마다 생성하던 걸 __init__ 으로 올렸다. 동시에 타일 그리드를 8×8 → 4×4 로 줄였다. 도메인 실험으로 확인한 결과 분석 대상 사이즈에 4×4가 더 적합했고, 연산량도 준다. 단순한 변경인데 요청당 0.2~0.5초 절약.

Docker 멀티스테이지 빌드

pyzbar 가 libzbar0 시스템 패키지를 요구한다. OpenCV도 libglib2.0, libgl1-mesa-glx 등이 필요. 이걸 runtime 이미지에 다 넣으면 1GB를 훌쩍 넘는다.

멀티스테이지로 나눴다:

• Builder stage: build-essential, libzbar-dev, Poetry 설치 후 의존성 빌드
• Runtime stage: 런타임 라이브러리만, 빌드된 휠을 복사

이미지 크기가 체감으로 반 이하가 됐다. CI 빌드 속도, 배포 속도 둘 다 이득.

단계별 지연 (실측 기반 추정)

Locust 200 users / 6분 부하 테스트에서 대체로 이 범위 안에서 수렴했다.

동시성 — 왜 worker 4개, 왜 thread가 아닌가

팀 리뷰에서 질문을 두 번쯤 받았다. “async 안 쓰는 이유?”, “thread pool 쓰면 더 빠르지 않나?”.

매번 같은 답을 하려니 지쳐서 python-concurrency-patterns.md 라는 문서를 따로 970줄 정도 썼다. 요약하면:

1. 요청 간 독립성: 각 요청이 서로 다른 이미지를 들고 옴. 공유 상태 없음. → 프로세스 격리가 안전
2. 요청 내부는 순차 의존: BBox 결과가 Strip의 입력, Strip 결과가 Estimate의 입력. 병렬화 여지가 적음
3. OpenCV/NumPy는 GIL을 해제한다: thread로도 되긴 하는데, 디버깅 난이도와 메모리 공유 이슈 고려하면 process worker 가 나음
4. Uvicorn workers=4: CPU 4 core 컨테이너에 정확히 맞춤. 워커당 KNN 캐시 1셋 = 메모리 4배. 수용 가능한 트레이드오프

make load-test (200 users, 6분), make load-test-stress (0→50→100→200 단계적), make load-test-spike (200→500) 세 시나리오로 검증했다. docker stats 로 CPU/메모리 추이 찍어보면 worker 4개가 균등하게 로드를 나눠 가진다.

뭘 놓쳤나 — 회고 3포인트

1. 알고리즘 단계 병렬화를 포기했다

“단계 간 의존성 때문” 이라고 합리화했는데, 실제로는 S3 다운로드와 KNN warm-up은 독립적이다. FastAPI 쪽에서 asyncio.gather([download_s3(), warm_knn()]) 로 묶었으면 첫 요청의 cold-start 지연을 줄일 수 있었다. 깊게 들어가면 BlackBox 내부의 QR 검출과 ROI 추출도 일부 병렬화 여지가 있다. 이건 내가 “단계”라는 단어에 갇혀서 못 본 부분.

2. S3 다운로드가 동기(sync) 호출이다

boto3 의 download_fileobj 는 블로킹 I/O 다. FastAPI 는 async 프레임워크인데 여기서 이벤트 루프가 네트워크 대기 동안 멈춘다. 지금은 Uvicorn worker process 4개로 우회 중이라 체감 문제가 없지만, 단일 워커 내부에서의 concurrent 요청은 직렬화되는 셈. 제대로 하려면 두 가지 옵션 중 하나였다:

• aiobotocore 로 바꿔서 진짜 async 다운로드
• 최소한 asyncio.to_thread(self.s3_client.download_fileobj, ...) 로 스레드풀에 위임

프로세스 워커 4개로 커버되니까 “async 이점의 절반만 살리는” 형태로 타협했는데, 커넥션 풀이나 동시 요청 수가 늘면 S3 대기가 곧 병목이 된다. “async 프레임워크 쓴다” 와 “async 하게 동작한다” 는 별개라는 걸 놓쳤다.

3. Observability 가 로그뿐

[TIMING] BlackBox: 87ms 같은 걸 logger.info 로 찍는 수준에 그쳤다. 이걸 히스토그램으로 쌓아야 튜닝 근거가 된다 — “p50/p95/p99 가 어디냐, CLAHE 4×4 바꿨을 때 분포가 어떻게 이동했냐”. Prometheus exporter 하나 붙이고 Grafana로 받았으면 성능 최적화의 대화가 숫자 위에서 가능했을 것. 지금은 “체감상 빨라졌다” 수준의 주장만 남아있다.

마치며

1,129줄짜리 레거시 스크립트를 @app.post("/") 뒤에 그대로 꽂는 건 10분이면 끝났을 것이다. 실제로는 그 스크립트를 서비스로 만드는 과정이 프로젝트 전체의 절반 이상 걸렸다. 계층 분리, 전역 객체 제거, 에러 코드 구조화, 파일 I/O 제거, KNN 캐싱 — 이게 다 “CLI → 서비스” 로 오는 길 위에서 어느 하나 건너뛸 수 없었다.

그리고 “서비스” 라는 단어의 무게도 배웠다. 동시성, 부하 분산, 관측 가능성 — 이게 다 서비스가 된 이후에만 의미가 있는 개념들이다. Python 스크립트를 FastAPI로 감쌌다고 자동으로 이게 따라오지 않는다. 하나하나 의식적으로 얹어야 했다.

다음 읽을거리

• Fintech 회고 시리즈 — 비슷하게 Python 마이크로서비스를 직접 설계한 기록
• IoT 데이터 파이프라인 회고 — 또 다른 “스크립트 → 서비스” 이야기