feat(backend): 문서 타입별 모델 선택 및 프리로드 추가

app/main.py

@@ -26,6 +26,7 @@ from sqlalchemy.orm import Session
 from .database import engine, get_db, init_db, test_connection
 from . import models
 from .routers import analysis, downloads, pages, projects
+from .services.model_registry import model_registry
 
 # 환경 변수 로드
 load_dotenv()
@@ -126,6 +127,19 @@ async def startup_event():
             print("✅ Database tables initialized")
         except Exception as e:
             print(f"⚠️ Table initialization warning: {e}")
+
+    preload_env = os.getenv("MODEL_PRELOAD", "SmartEyeSsen")
+    preload_targets = [
+        name.strip()
+        for name in preload_env.split(",")
+        if name.strip()
+    ]
+    if preload_targets:
+        try:
+            model_registry.preload(preload_targets)
+            print(f"🧠 Preloaded models: {', '.join(preload_targets)}")
+        except Exception as e:
+            print(f"⚠️ Model preload failed: {e}")
     
     print("=" * 60)
     print("✅ SmartEyeSsen Backend Ready!")

app/routers/analysis.py

@@ -15,6 +15,8 @@ from ..services.batch_analysis import (
     analyze_project_batch_async_parallel,
     _get_analysis_service,
     _process_single_page_async,
+    is_supported_model,
+    resolve_model_choice,
 )
 from ..services.formatter import TextFormatter
 
@@ -37,12 +39,14 @@ class ProjectAnalysisRequest(BaseModel):
     api_key: Optional[str] = None
     use_parallel: bool = True  # False → True (병렬 처리 기본값)
     max_concurrent_pages: int = 8  # 4 → 8 (성능 최적화)
+    analysis_model: Optional[str] = None
 
 
 class PageAnalysisRequest(BaseModel):
     """단일 페이지 비동기 분석 요청"""
     use_ai_descriptions: bool = True
     api_key: Optional[str] = None
+    analysis_model: Optional[str] = None
 
 
 @router.post(
@@ -71,9 +75,14 @@ async def analyze_project(
     - 모델: 싱글톤 패턴으로 메모리 효율적 (중복 로드 방지)
     - 권장: 모든 환경 (CPU 4코어 이상, RAM 4GB+)
     """
-    project_exists = db.query(Project.project_id).filter(Project.project_id == project_id).scalar()
-    if not project_exists:
+    project = db.query(Project).filter(Project.project_id == project_id).first()
+    if not project:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="프로젝트를 찾을 수 없습니다.")
+    if payload.analysis_model and not is_supported_model(payload.analysis_model):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"지원하지 않는 모델입니다: {payload.analysis_model}",
+        )
 
     if payload.use_parallel:
         logger.info(f"병렬 분석 시작: project_id={project_id}, max_concurrent={payload.max_concurrent_pages}")
@@ -83,6 +92,7 @@ async def analyze_project(
             use_ai_descriptions=payload.use_ai_descriptions,
             api_key=payload.api_key,
             max_concurrent_pages=payload.max_concurrent_pages,
+            analysis_model=payload.analysis_model or None,
         )
     else:
         logger.info(f"순차 분석 시작: project_id={project_id}")
@@ -91,6 +101,7 @@ async def analyze_project(
             project_id=project_id,
             use_ai_descriptions=payload.use_ai_descriptions,
             api_key=payload.api_key,
+            analysis_model=payload.analysis_model or None,
         )
     
     return analysis_result
@@ -133,6 +144,11 @@ def analyze_page_async(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"페이지 ID {page_id}를 찾을 수 없습니다."
         )
+    if payload.analysis_model and not is_supported_model(payload.analysis_model):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"지원하지 않는 모델입니다: {payload.analysis_model}",
+        )
     
     # 작업 ID 생성
     job_id = str(uuid.uuid4())
@@ -142,6 +158,7 @@ def analyze_page_async(
         "page_id": page_id,
         "page_number": page.page_number,
         "project_id": page.project_id,
+        "analysis_model": payload.analysis_model,
         "result": None,
         "error": None,
         "progress": "작업 대기 중...",
@@ -156,6 +173,7 @@ def analyze_page_async(
         page_id=page_id,
         use_ai_descriptions=payload.use_ai_descriptions,
         api_key=payload.api_key,
+        analysis_model=payload.analysis_model,
     )
     
     return {
@@ -196,6 +214,7 @@ async def _run_async_page_analysis(
     page_id: int,
     use_ai_descriptions: bool,
     api_key: Optional[str],
+    analysis_model: Optional[str],
 ) -> None:
     """
     백그라운드에서 실행되는 단일 페이지 비동기 분석 작업
@@ -227,7 +246,8 @@ async def _run_async_page_analysis(
             raise ValueError(f"프로젝트 ID {page.project_id}를 찾을 수 없습니다.")
         
         # AnalysisService 및 TextFormatter 초기화
-        analysis_service = _get_analysis_service()
+        model_choice = resolve_model_choice(project.doc_type_id, analysis_model)
+        analysis_service = _get_analysis_service(model_choice)
         formatter = TextFormatter(
             doc_type_id=project.doc_type_id,
             db=db,

app/services/analysis_service.py

@@ -27,12 +27,12 @@ import openai
 import pytesseract
 import torch
 from PIL import Image
-from huggingface_hub import hf_hub_download
 from loguru import logger
 from openai import AsyncOpenAI
 from sqlalchemy.orm import Session
 
 from .. import models
+from .model_registry import model_registry
 
 # --- 신규: 이미지 설명을 위한 프롬프트 템플릿 추가 ---
 figure_prompt = """
@@ -239,77 +239,34 @@ class AnalysisService:
             model_choice: 사용할 모델 선택 (기본값: "SmartEyeSsen")
             auto_load: True이면 초기화 시 자동으로 모델 로드 (기본값: False, 하위 호환성 유지)
         """
-        self.model = None
         self.device = device
         self.model_choice = model_choice
+        self.model_registry = model_registry
+        self._model_handle = None
         self._model_loaded = False
 
         # 자동 로드 옵션이 활성화된 경우 즉시 모델 로드
         if auto_load:
             self._ensure_model_loaded()
 
-    def download_model(self, model_choice="SmartEyeSsen"):
-        """모델 다운로드 (기존과 동일)"""
-        models = {
-            "doclaynet_docsynth": {
-                "repo_id": "juliozhao/DocLayout-YOLO-DocLayNet-Docsynth300K_pretrained",
-                "filename": "doclayout_yolo_doclaynet_imgsz1120_docsynth_pretrain.pt",
-            },
-            "docstructbench": {
-                "repo_id": "juliozhao/DocLayout-YOLO-DocStructBench",
-                "filename": "doclayout_yolo_docstructbench_imgsz1024.pt",
-            },
-            "docsynth300k": {
-                "repo_id": "juliozhao/DocLayout-YOLO-DocSynth300K-pretrain",
-                "filename": "doclayout_yolo_docsynth300k_imgsz1600.pt",
-            },
-            "SmartEyeSsen": {"repo_id": "AkJeond/SmartEye", "filename": "best.pt"},
-        }
-        selected_model = models.get(model_choice, models["SmartEyeSsen"])
-        try:
-            logger.info(f"모델 다운로드 중: {selected_model['repo_id']}")
-            filepath = hf_hub_download(
-                repo_id=selected_model["repo_id"], filename=selected_model["filename"]
-            )
-            logger.info(f"모델 다운로드 완료: {filepath}")
-            return filepath
-        except Exception as e:
-            logger.error(f"모델 다운로드 실패: {e}")
-            raise
-
-    def load_model(self, model_path):
-        """모델 로드 (기존과 동일)"""
-        try:
-            try:
-                from doclayout_yolo import YOLOv10
-            except ImportError:
-                logger.error("DocLayout-YOLO가 설치되지 않았습니다.")
-                return False
-            logger.info("모델 로드 중...")
-            self.model = YOLOv10(model_path, task="predict")
-            self.model.to(self.device)
-            if hasattr(self.model, "training"):
-                self.model.training = False
-            logger.info("모델 로드 완료!")
-            return True
-        except Exception as e:
-            logger.error(f"모델 로드 실패: {e}")
-            return False
-
-    def _ensure_model_loaded(self):
+    def _ensure_model_loaded(self, model_choice: Optional[str] = None):
         """
         Lazy Loading: 모델이 로드되지 않았으면 자동으로 로드
         (다중 페이지 처리 시 모델을 한 번만 로드하도록 최적화)
         """
-        if self._model_loaded and self.model is not None:
-            return  # 이미 로드됨
-
-        logger.info(f"모델 자동 로드 시작 (선택: {self.model_choice})...")
-        model_path = self.download_model(self.model_choice)
-        if not self.load_model(model_path):
-            raise RuntimeError(f"모델 로드 실패: {self.model_choice}")
+        target_model = model_choice or self.model_choice
+        if (
+            self._model_loaded
+            and self._model_handle is not None
+            and self._model_handle.name == target_model
+        ):
+            return self._model_handle
+
+        handle = self.model_registry.get_model(target_model, device=self.device)
+        self._model_handle = handle
+        self.model_choice = target_model
         self._model_loaded = True
-        logger.info("모델 자동 로드 완료!")
+        return handle
 
     def analyze_layout(
         self,
@@ -341,27 +298,24 @@ class AnalysisService:
                 self._model_loaded = False
 
             # Lazy Loading: 모델이 없으면 자동 로드
-            self._ensure_model_loaded()
+            handle = self._ensure_model_loaded(active_model)
+            model = handle.model
+            model_spec = handle.spec
 
             logger.info("레이아웃 분석 시작...")
             temp_path = "temp_image.jpg"
             cv2.imwrite(temp_path, image)
 
-            if active_model == "SmartEyeSsen":
-                imgsz, conf = 1024, 0.25
-            elif active_model == "docsynth300k":
-                imgsz, conf = 1600, 0.15
-            else:
-                imgsz, conf = 1024, 0.25
+            imgsz, conf = model_spec.imgsz, model_spec.conf
 
-            results = self.model.predict(
+            results = model.predict(
                 temp_path, imgsz=imgsz, conf=conf, iou=0.45, device=self.device
             )
 
             boxes = results[0].boxes.xyxy.cpu().numpy()  # [x1, y1, x2, y2]
             classes = results[0].boxes.cls.cpu().numpy()
             confs = results[0].boxes.conf.cpu().numpy()
-            class_names = self.model.names  # 클래스 ID → 이름
+            class_names = model.names  # 클래스 ID → 이름
 
             detection_records: List[Dict[str, float]] = []
 

app/services/batch_analysis.py

@@ -38,7 +38,7 @@ import time
 from contextlib import asynccontextmanager
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Set
 
 import aiofiles
 import cv2
@@ -49,6 +49,7 @@ from sqlalchemy.orm import Session, selectinload
 
 from ..models import LayoutElement, Page, Project
 from .analysis_service import AnalysisService
+from .model_registry import model_registry
 from .formatter import TextFormatter
 from .mock_models import MockElement
 from .sorter import save_sorting_results_to_db, sort_layout_elements
@@ -67,6 +68,51 @@ DEFAULT_MAX_CONCURRENT_PAGES = int(os.getenv("MAX_CONCURRENT_PAGES", "8"))  # CP
 _model_instances: Dict[str, AnalysisService] = {}
 _model_lock = threading.Lock()
 
+# 문서 타입별 기본 모델 매핑
+DOC_TYPE_MODEL_MAP = {
+    1: "SmartEyeSsen",
+    2: "docstructbench",
+}
+DEFAULT_MODEL_CHOICE = "SmartEyeSsen"
+
+
+def _available_model_names() -> Set[str]:
+    return set(model_registry.list_registered().keys())
+
+
+def is_supported_model(model_name: str) -> bool:
+    return model_name in _available_model_names()
+
+
+def resolve_model_choice(
+    doc_type_id: Optional[int],
+    requested_model: Optional[str] = None,
+) -> str:
+    """
+    doc_type 또는 사용자 요청에 맞는 모델명을 반환합니다.
+
+    Args:
+        doc_type_id: document_types.doc_type_id
+        requested_model: 사용자가 명시적으로 지정한 모델 이름
+
+    Raises:
+        ValueError: 지원되지 않는 모델명이 요청된 경우
+    """
+    if requested_model:
+        if not is_supported_model(requested_model):
+            raise ValueError(f"지원하지 않는 AI 모델입니다: {requested_model}")
+        return requested_model
+
+    if doc_type_id in DOC_TYPE_MODEL_MAP:
+        return DOC_TYPE_MODEL_MAP[doc_type_id]
+
+    logger.warning(
+        "알 수 없는 doc_type_id ({})에 대해 기본 모델({})을 사용합니다.",
+        doc_type_id,
+        DEFAULT_MODEL_CHOICE,
+    )
+    return DEFAULT_MODEL_CHOICE
+
 
 def _get_analysis_service(model_choice: str = "SmartEyeSsen") -> AnalysisService:
     """
@@ -515,6 +561,7 @@ async def analyze_project_batch_async(
     use_ai_descriptions: bool = True,
     api_key: Optional[str] = None,
     ai_max_concurrency: int = DEFAULT_AI_CONCURRENCY,
+    analysis_model: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     프로젝트 내 'pending' 상태 페이지를 순차적으로 분석하고 결과 요약을 반환합니다.
@@ -555,7 +602,14 @@ async def analyze_project_batch_async(
     _update_project_status(project, "in_progress")
     db.commit()
 
-    analysis_service = _get_analysis_service()
+    model_choice = resolve_model_choice(project.doc_type_id, analysis_model)
+    logger.info(
+        "프로젝트 분석 모델 선택: project_id={}, doc_type_id={}, model={}",
+        project.project_id,
+        project.doc_type_id,
+        model_choice,
+    )
+    analysis_service = _get_analysis_service(model_choice)
     formatter = TextFormatter(
         doc_type_id=project.doc_type_id,
         db=db,
@@ -612,6 +666,7 @@ def analyze_project_batch(
     use_ai_descriptions: bool = True,
     api_key: Optional[str] = None,
     ai_max_concurrency: int = DEFAULT_AI_CONCURRENCY,
+    analysis_model: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     동기 컨텍스트 호환용 래퍼.
@@ -623,6 +678,7 @@ def analyze_project_batch(
             use_ai_descriptions=use_ai_descriptions,
             api_key=api_key,
             ai_max_concurrency=ai_max_concurrency,
+            analysis_model=analysis_model,
         )
     )
 
@@ -635,6 +691,7 @@ async def analyze_project_batch_async_parallel(
     api_key: Optional[str] = None,
     ai_max_concurrency: int = DEFAULT_AI_CONCURRENCY,
     max_concurrent_pages: int = 8,
+    analysis_model: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     프로젝트 내 'pending' 상태 페이지를 병렬로 분석하고 결과 요약을 반환합니다.
@@ -696,7 +753,14 @@ async def analyze_project_batch_async_parallel(
     _update_project_status(project, "in_progress")
     db.commit()
 
-    analysis_service = _get_analysis_service()
+    model_choice = resolve_model_choice(project.doc_type_id, analysis_model)
+    logger.info(
+        "병렬 프로젝트 분석 모델 선택: project_id={}, doc_type_id={}, model={}",
+        project.project_id,
+        project.doc_type_id,
+        model_choice,
+    )
+    analysis_service = _get_analysis_service(model_choice)
     formatter = TextFormatter(
         doc_type_id=project.doc_type_id,
         db=db,
@@ -798,6 +862,7 @@ def analyze_project_batch_parallel(
     api_key: Optional[str] = None,
     ai_max_concurrency: int = DEFAULT_AI_CONCURRENCY,
     max_concurrent_pages: int = DEFAULT_MAX_CONCURRENT_PAGES,
+    analysis_model: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     동기 컨텍스트 호환용 래퍼 (병렬 처리 버전).
@@ -810,6 +875,7 @@ def analyze_project_batch_parallel(
             api_key=api_key,
             ai_max_concurrency=ai_max_concurrency,
             max_concurrent_pages=max_concurrent_pages,
+            analysis_model=analysis_model,
         )
     )
 
@@ -823,4 +889,6 @@ __all__ = [
     "_process_single_page",
     "_process_single_page_async",
     "DEFAULT_AI_CONCURRENCY",
+    "is_supported_model",
+    "resolve_model_choice",
 ]

app/services/model_registry.py

@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from shutil import copy2
+from threading import Lock
+from typing import Dict, Iterable, Optional
+
+import torch
+from huggingface_hub import hf_hub_download
+from loguru import logger
+
+
+try:
+    from doclayout_yolo import YOLOv10
+except ImportError as exc:  # pragma: no cover - 환경 의존
+    YOLOv10 = None  # type: ignore[assignment]
+    _IMPORT_ERROR = exc
+else:
+    _IMPORT_ERROR = None
+
+
+@dataclass(frozen=True)
+class ModelSpec:
+    name: str
+    repo_id: str
+    filename: str
+    imgsz: int = 1024
+    conf: float = 0.25
+
+
+@dataclass
+class ModelHandle:
+    name: str
+    spec: ModelSpec
+    model: "YOLOv10"
+    device: str
+    weight_path: Path
+
+
+class ModelRegistry:
+    """
+    DocLayout-YOLO 계열 모델을 전역으로 캐싱/재사용하기 위한 레지스트리.
+    - 모델별 가중치 다운로드는 한 번만 수행
+    - 디바이스(CPU/GPU)별 인스턴스를 필요 시 별도로 유지
+    """
+
+    def __init__(self) -> None:
+        self._specs: Dict[str, ModelSpec] = {}
+        self._models: Dict[str, ModelHandle] = {}
+        self._locks: Dict[str, Lock] = {}
+        self._default_device = "cuda" if torch.cuda.is_available() else "cpu"
+
+    @staticmethod
+    def _make_key(name: str, device: str) -> str:
+        return f"{name}:{device}"
+
+    def register(self, spec: ModelSpec) -> None:
+        self._specs[spec.name] = spec
+        self._locks.setdefault(spec.name, Lock())
+        logger.debug(f"📘 모델 스펙 등록: {spec.name} (imgsz={spec.imgsz}, conf={spec.conf})")
+
+    def list_registered(self) -> Dict[str, ModelSpec]:
+        return dict(self._specs)
+
+    def preload(self, targets: Optional[Iterable[str]] = None, *, device: Optional[str] = None) -> None:
+        names = list(targets) if targets else list(self._specs.keys())
+        for name in names:
+            try:
+                self.get_model(name, device=device)
+            except Exception as exc:  # pragma: no cover - 초기화 단계
+                logger.error(f"❌ 모델 프리로드 실패 ({name}): {exc}")
+                raise
+
+    def get_model(self, name: str, *, device: Optional[str] = None) -> ModelHandle:
+        if name not in self._specs:
+            raise KeyError(f"등록되지 않은 모델입니다: {name}")
+
+        if _IMPORT_ERROR is not None:
+            raise RuntimeError(
+                "doclayout_yolo 패키지가 설치되지 않아 모델을 로드할 수 없습니다."
+            ) from _IMPORT_ERROR
+
+        resolved_device = device or self._default_device
+        key = self._make_key(name, resolved_device)
+
+        if key in self._models:
+            return self._models[key]
+
+        lock = self._locks.setdefault(name, Lock())
+        with lock:
+            if key in self._models:
+                return self._models[key]
+
+            spec = self._specs[name]
+            weight_path = self._download_weights(name, spec)
+            model = self._load_model(weight_path, resolved_device)
+
+            handle = ModelHandle(
+                name=name,
+                spec=spec,
+                model=model,
+                device=resolved_device,
+                weight_path=weight_path,
+            )
+            self._models[key] = handle
+            logger.info(f"✅ 모델 로드 완료: {name} (device={resolved_device})")
+            return handle
+
+    @staticmethod
+    def _download_weights(name: str, spec: ModelSpec) -> Path:
+        override_env = os.getenv(f"{name.upper()}_MODEL_PATH")
+        if override_env:
+            override_path = Path(override_env)
+            if override_path.exists():
+                logger.info(f"📂 {name} 가중치 경로 override 사용: {override_path}")
+                return override_path.resolve()
+            logger.warning(
+                f"⚠️ {name.upper()}_MODEL_PATH 가 지정되었지만 파일을 찾을 수 없습니다: {override_path}"
+            )
+
+        cache_root = Path(
+            os.getenv("MODEL_CACHE_DIR", Path.home() / ".cache" / "smarteye_models")
+        ).resolve()
+        target_dir = (cache_root / name).resolve()
+        target_dir.mkdir(parents=True, exist_ok=True)
+        target_path = target_dir / spec.filename
+
+        if target_path.exists():
+            logger.debug(f"📦 캐시된 가중치 사용: {target_path}")
+            return target_path
+
+        logger.info(f"⬇️ {name} 가중치 다운로드 중 ({spec.repo_id}/{spec.filename})")
+        downloaded_path = hf_hub_download(
+            repo_id=spec.repo_id,
+            filename=spec.filename,
+            local_dir=str(target_dir),
+            local_dir_use_symlinks=False,
+        )
+
+        downloaded_path = Path(downloaded_path).resolve()
+        if downloaded_path != target_path:
+            copy2(downloaded_path, target_path)
+            logger.debug(f"📁 가중치 복사: {downloaded_path.name} -> {target_path}")
+
+        return target_path
+
+    @staticmethod
+    def _load_model(weight_path: Path, device: str) -> "YOLOv10":
+        if YOLOv10 is None:  # pragma: no cover
+            raise RuntimeError("doclayout_yolo 패키지가 없습니다.")
+
+        logger.info(f"🧠 모델 로딩: {weight_path.name} (device={device})")
+        model = YOLOv10(str(weight_path), task="predict")
+        model.to(device)
+        if hasattr(model, "training"):
+            model.training = False
+        return model
+
+
+# ---------------------------------------------------------------------------
+# 전역 레지스트리 인스턴스 및 기본 모델 스펙 등록
+# ---------------------------------------------------------------------------
+DEFAULT_MODEL_SPECS = [
+    ModelSpec(
+        name="SmartEyeSsen",
+        repo_id="AkJeond/SmartEye",
+        filename="best.pt",
+        imgsz=1024,
+        conf=0.25,
+    ),
+    ModelSpec(
+        name="docstructbench",
+        repo_id="juliozhao/DocLayout-YOLO-DocStructBench",
+        filename="doclayout_yolo_docstructbench_imgsz1024.pt",
+        imgsz=1024,
+        conf=0.25,
+    )
+]
+
+model_registry = ModelRegistry()
+for spec in DEFAULT_MODEL_SPECS:
+    model_registry.register(spec)