add dockerfile, change async

Author: jinsoo

.gitignore

@@ -18,3 +18,12 @@ qdrant_data/
 **/qdrant_data/
 *.sqlite
 *.db
+
+tasteam_app_all_restaurants_ai_api_results.json
+tasteam_app_all_restaurants_ai_api_results.sql
+tasteam_all_seed_reviews.csv
+tasteam_app_all_review_data.json
+merged.csv
+260202_api_result.md
+tasteam_app_kr3_640k_even.json
+service_simul_data/

Dockerfile.cpu

@@ -0,0 +1,35 @@
+# CPU 전용 이미지 - GPU 없는 환경에서 src 애플리케이션 실행
+#
+# 빌드: docker build -f Dockerfile.cpu -t app-cpu .
+# 실행: docker run -p 8001:8001 app-cpu
+#
+FROM python:3.11-slim-bookworm
+
+ENV PYTHONUNBUFFERED=1
+# config.py에서 GPU 사용 여부를 "USE_GPU#" 환경 변수로 읽음
+ENV "USE_GPU#"=false
+WORKDIR /app
+
+# 시스템 의존성: 빌드 도구 + OpenJDK (PySpark/비교 파이프라인용)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    curl \
+    openjdk-17-jdk-headless \
+    && rm -rf /var/lib/apt/lists/*
+ENV JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
+
+# PyTorch CPU 버전 먼저 설치 (requirements보다 먼저 해야 충돌 방지)
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu
+
+COPY requirements.txt /app/
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 애플리케이션 코드 복사
+COPY . /app
+
+# 포트 노출 (app.py 기본값 8001)
+EXPOSE 8001
+
+# CPU 환경에서 실행
+CMD ["python", "app.py"]

src/api/main.py

@@ -14,6 +14,13 @@
 
 from .routers import sentiment, vector, llm, test
 from ..cpu_monitor import get_cpu_monitor
+from ..metrics_collector import app_queue_depth_inc, app_queue_depth_dec
+
+try:
+    from prometheus_fastapi_instrumentator import Instrumentator
+    _INSTRUMENTATOR_AVAILABLE = True
+except ImportError:
+    _INSTRUMENTATOR_AVAILABLE = False
 
 # 로거 설정 (콘솔 출력)
 # basicConfig는 한 번만 실행되므로, root 로거에 직접 핸들러 추가
@@ -74,6 +81,17 @@ async def add_request_id(request: Request, call_next):
     return response
 
 
+# Queue depth (in-flight 요청 수) — Prometheus app_queue_depth 집계용
+@app.middleware("http")
+async def track_queue_depth(request: Request, call_next):
+    app_queue_depth_inc()
+    try:
+        response = await call_next(request)
+        return response
+    finally:
+        app_queue_depth_dec()
+
+
 def _error_payload(*, code: int, message: str, details, request_id: str) -> dict:
     return {"code": code, "message": message, "details": details, "request_id": request_id}
 
@@ -158,3 +176,14 @@ async def health():
         "version": "1.0.0",
     }
 
+
+# Prometheus 메트릭 (요청 수, 지연 시간 등 자동 수집, 패키지 설치 시에만 노출)
+if _INSTRUMENTATOR_AVAILABLE:
+    Instrumentator().instrument(app).expose(app)
+else:
+    import logging
+    logging.getLogger(__name__).warning(
+        "prometheus_fastapi_instrumentator 미설치: /metrics 비활성화. "
+        "설치: pip install prometheus-client prometheus-fastapi-instrumentator"
+    )
+

src/api/routers/llm.py

@@ -337,14 +337,17 @@ async def summarize_reviews(
 
         # 메트릭 수집
         total_reviews_count = sum(len(hits_dict.get(cat, [])) for cat in ["service", "price", "food"])
+        processing_time_ms = (time.time() - start_time) * 1000
         request_id = metrics.collect_metrics(
             restaurant_id=request.restaurant_id,
             analysis_type="summary",
             start_time=start_time,
             tokens_used=None,
             batch_size=total_reviews_count,
         )
-        
+        # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+        metrics.record_llm_ttft(analysis_type="summary", ttft_ms=processing_time_ms)
+
         # 항상 SummaryDisplayResponse (positive_reviews 등 미사용 필드 제외)
         return SummaryDisplayResponse(
             restaurant_id=request.restaurant_id,
@@ -353,7 +356,7 @@ async def summarize_reviews(
             categories=categories_dict if categories_dict else None,
             debug=DebugInfo(
                 request_id=request_id,
-                processing_time_ms=(time.time() - start_time) * 1000,
+                processing_time_ms=processing_time_ms,
                 tokens_used=None,
                 model_version=None,
             ) if debug else None,
@@ -484,8 +487,11 @@ async def compare(
             analysis_type="comparison",
             start_time=start_time,
             batch_size=result.get("total_candidates", 0),
-            )
-            
+        )
+        # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+        ttfur_ms = (time.time() - start_time) * 1000
+        metrics.record_llm_ttft(analysis_type="comparison", ttft_ms=ttfur_ms)
+
         # 디버그 정보 추가
         if debug:
             result["debug"] = DebugInfo(
@@ -536,6 +542,7 @@ async def compare_batch(
     Returns:
         각 레스토랑별 비교 결과 리스트
     """
+    start_time = time.time()
     try:
         pipeline = ComparisonPipeline(
             llm_utils=llm_utils,
@@ -545,6 +552,9 @@ async def compare_batch(
             restaurants=request.restaurants,
             all_average_data_path=request.all_average_data_path,
         )
+        # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+        elapsed_ms = (time.time() - start_time) * 1000
+        metrics.record_llm_ttft(analysis_type="comparison", ttft_ms=elapsed_ms)
         return ComparisonBatchResponse(results=[ComparisonResponse(**r) for r in results])
     except Exception as e:
         logger.error(f"배치 비교 중 오류: {str(e)}", exc_info=True)
@@ -586,13 +596,17 @@ async def summarize_reviews_batch(
     Returns:
         각 레스토랑별 요약 결과 리스트 (categories 기반)
     """
+    start_time = time.time()
     try:
         seed_list = [DEFAULT_SERVICE_SEEDS, DEFAULT_PRICE_SEEDS, DEFAULT_FOOD_SEEDS]
         name_list = ["service", "price", "food"]
         logger.info("요약: 기본 시드만 사용")
 
         if Config.SUMMARY_SEARCH_ASYNC or Config.SUMMARY_RESTAURANT_ASYNC:
             results = await _batch_summarize_async(request, vector_search, llm_utils, seed_list, name_list)
+            # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+            elapsed_ms = (time.time() - start_time) * 1000
+            metrics.record_llm_ttft(analysis_type="summary", ttft_ms=elapsed_ms)
             return SummaryBatchResponse(results=[SummaryDisplayResponse(**r) for r in results])
 
         # search_async=false, restaurant_async=false: 레스토랑·aspect 완전 순차
@@ -644,7 +658,10 @@ async def summarize_reviews_batch(
                 per_category_max=request.limit,
             )
             results.append(_build_category_result(result, restaurant_id, restaurant_data.get("restaurant_name")))
-        
+
+        # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+        elapsed_ms = (time.time() - start_time) * 1000
+        metrics.record_llm_ttft(analysis_type="summary", ttft_ms=elapsed_ms)
         return SummaryBatchResponse(results=[
             SummaryDisplayResponse(**r) for r in results
         ])

src/api/routers/sentiment.py

@@ -116,13 +116,16 @@ async def analyze_sentiment(
             )
 
             # 메트릭 수집
+            processing_time_ms = (time.time() - start_time) * 1000
             request_id = metrics.collect_metrics(
                 restaurant_id=request.restaurant_id,
                 analysis_type="sentiment",
                 start_time=start_time,
                 tokens_used=result.get("tokens_used"),
                 batch_size=len(request.reviews),
             )
+            # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+            metrics.record_llm_ttft(analysis_type="sentiment", ttft_ms=processing_time_ms)
 
             # 디버그 모드에 따라 응답 반환 (restaurant_name은 요청에서 반환)
             result["restaurant_name"] = getattr(request, "restaurant_name", None)
@@ -167,6 +170,7 @@ async def analyze_sentiment(
 async def analyze_sentiment_batch(
     request: SentimentAnalysisBatchRequest,
     analyzer: SentimentAnalyzer = Depends(get_sentiment_analyzer),
+    metrics: MetricsCollector = Depends(get_metrics_collector),
 ):
     """
     여러 레스토랑의 **전체 리뷰**를 sentiment 모델로 분류하여 결과를 반환합니다.
@@ -180,6 +184,7 @@ async def analyze_sentiment_batch(
     Returns:
         각 레스토랑별 감성 분석 결과 리스트
     """
+    start_time = time.time()
     try:
         results = await analyzer.analyze_multiple_restaurants_async(restaurants_data=request.restaurants)
         # 각 결과에 restaurant_name 병합 (요청 항목 순서 대응)
@@ -190,6 +195,9 @@ async def analyze_sentiment_batch(
                 r["restaurant_name"] = item.get("restaurant_name") if isinstance(item, dict) else getattr(item, "restaurant_name", None)
             else:
                 r["restaurant_name"] = None
+        # TTFUR = t1 - t0 (요청 수신 시각 t0 → 응답 반환 직전 t1)
+        elapsed_ms = (time.time() - start_time) * 1000
+        metrics.record_llm_ttft(analysis_type="sentiment", ttft_ms=elapsed_ms)
         return SentimentAnalysisBatchResponse(results=[
             SentimentAnalysisResponse(**result) for result in results
         ])

src/comparison_pipeline.py

@@ -28,6 +28,26 @@
 except ImportError:
     Py4JNetworkError = Exception  # no py4j
 
+# Spark/JVM 실패 시 폴백용: JAVA_GATEWAY_EXITED 등 Py4J 이외 예외도 잡기 위함
+def _is_spark_or_jvm_error(e: Exception) -> bool:
+    msg = str(e).upper()
+    return (
+        "JAVA_GATEWAY" in msg
+        or "PY4J" in msg
+        or "SPARK" in msg
+        or isinstance(e, (Py4JNetworkError, BrokenPipeError, ConnectionError, OSError, EOFError))
+    )
+
+
+def _spark_disabled() -> bool:
+    """Docker 등 JVM 없는 환경에서 Spark 비활성화 시 True."""
+    try:
+        from .config import Config
+        return getattr(Config, "DISABLE_SPARK", False)
+    except Exception:
+        return False
+
+
 _spark_session = None
 
 
@@ -517,15 +537,18 @@ def calculate_single_restaurant_ratios(
     texts = [s for s in reviews if s and isinstance(s, str)]
     if not texts:
         return {"service": 0.0, "price": 0.0}
-    if SPARK_AVAILABLE:
+    if SPARK_AVAILABLE and not _spark_disabled():
         try:
             spark = _get_spark()
             rdd = spark.sparkContext.parallelize(texts, numSlices=max(1, min(len(texts) // 50, 32)))
             out = _spark_calculate_ratios(rdd, stopwords)
             return {"service": round(out["service"], 2), "price": round(out["price"], 2)}
-        except (Py4JNetworkError, BrokenPipeError, ConnectionError, OSError, EOFError) as e:
-            logger.warning("Spark/Py4J 오류, Python 폴백 사용: %s", e)
-            _reset_spark()
+        except Exception as e:
+            if _is_spark_or_jvm_error(e):
+                logger.warning("Spark/JVM 오류, Python 폴백 사용: %s", e)
+                _reset_spark()
+            else:
+                raise
     out = _python_calculate_ratios(texts, stopwords)
     return {"service": round(out["service"], 2), "price": round(out["price"], 2)}
 
@@ -596,6 +619,18 @@ def calculate_all_average_ratios_from_file(
         logger.warning("pyspark 미설치. calculate_all_average_ratios_from_file 불가.")
         return None
 
+    # Docker 등 JVM 없는 환경: Spark 건너뛰고 Python 경로만 사용
+    if _spark_disabled():
+        try:
+            rows = load_reviews_from_aspect_data_file(path, project_root)
+            texts = [(r.get("content") or r.get("text") or "").strip() for r in rows if isinstance(r, dict)]
+            texts = [t for t in texts if t]
+            if texts:
+                return _python_calculate_ratios(texts, stopwords)
+        except Exception as e:
+            logger.warning("DISABLE_SPARK 시 Python 경로 실패: %s", e)
+        return None
+
     try:
         from pyspark.sql.functions import col, length, explode
 
@@ -632,20 +667,20 @@ def calculate_all_average_ratios_from_file(
 
         texts_rdd = base_df.select("text").rdd.map(lambda r: r["text"])
         return _spark_calculate_ratios(texts_rdd, stopwords)
-    except (Py4JNetworkError, BrokenPipeError, ConnectionError, OSError, EOFError) as e:
-        logger.warning("Spark/Py4J 오류, Python 폴백 시도: %s", e)
-        _reset_spark()
-        try:
-            rows = load_reviews_from_aspect_data_file(path, project_root)
-            texts = [(r.get("content") or r.get("text") or "").strip() for r in rows if isinstance(r, dict)]
-            texts = [t for t in texts if t]
-            if texts:
-                return _python_calculate_ratios(texts, stopwords)
-        except Exception as fb:
-            logger.warning("Python 폴백 실패: %s", fb)
-        return None
     except Exception as e:
-        logger.warning("calculate_all_average_ratios_from_file 실패: %s — %s", path, e)
+        if _is_spark_or_jvm_error(e):
+            logger.warning("Spark/JVM 오류, Python 폴백 시도: %s", e)
+            _reset_spark()
+            try:
+                rows = load_reviews_from_aspect_data_file(path, project_root)
+                texts = [(r.get("content") or r.get("text") or "").strip() for r in rows if isinstance(r, dict)]
+                texts = [t for t in texts if t]
+                if texts:
+                    return _python_calculate_ratios(texts, stopwords)
+            except Exception as fb:
+                logger.warning("Python 폴백 실패: %s", fb)
+        else:
+            logger.warning("calculate_all_average_ratios_from_file 실패: %s — %s", path, e)
         return None
 
 
@@ -727,16 +762,19 @@ def calculate_all_average_ratios_from_reviews(
     texts = [t for t in texts if t and isinstance(t, str)]
     if not texts:
         return {"service": 0.0, "price": 0.0}
-    if SPARK_AVAILABLE:
+    if SPARK_AVAILABLE and not _spark_disabled():
         try:
             spark = _get_spark()
             rdd = spark.sparkContext.parallelize(
                 texts, numSlices=max(1, min(len(texts) // 100, 256))
             )
             return _spark_calculate_ratios(rdd, stopwords)
-        except (Py4JNetworkError, BrokenPipeError, ConnectionError, OSError, EOFError) as e:
-            logger.warning("Spark/Py4J 오류, Python 폴백 사용: %s", e)
-            _reset_spark()
+        except Exception as e:
+            if _is_spark_or_jvm_error(e):
+                logger.warning("Spark/JVM 오류, Python 폴백 사용: %s", e)
+                _reset_spark()
+            else:
+                raise
     return _python_calculate_ratios(texts, stopwords)
 
 

src/config.py

@@ -132,7 +132,8 @@ class _CacheConfig:
 
 # --- Spark (Comparison 전체 평균 데이터 등) ---
 class _SparkConfig:
-    """Spark/배치: 전체 평균 데이터 경로, 비율"""
+    """Spark/배치: 전체 평균 데이터 경로, 비율. DISABLE_SPARK=true 시 JVM 없이 Kiwi만 사용 (Docker 등)."""
+    DISABLE_SPARK: bool = os.getenv("DISABLE_SPARK", "false").lower() == "true"
     ALL_AVERAGE_ASPECT_DATA_PATH: Optional[str] = os.getenv("ALL_AVERAGE_ASPECT_DATA_PATH", "data/test_data_sample.json")
     ALL_AVERAGE_SERVICE_RATIO: float = float(os.getenv("ALL_AVERAGE_SERVICE_RATIO", "0.60"))
     ALL_AVERAGE_PRICE_RATIO: float = float(os.getenv("ALL_AVERAGE_PRICE_RATIO", "0.55"))