[HCR-425] 새로 배포

app/core/config.py

@@ -10,8 +10,12 @@
 class Settings(BaseSettings):
     app_name: str = "intelligence-server"
     app_env: str = "local"
+    app_mode: str = Field(default="", validation_alias=AliasChoices("APP_MODE"))
     debug: bool = False
-    cdc_analysis_enabled: bool = False
+    cdc_analysis_enabled: bool | None = Field(
+        default=None,
+        validation_alias=AliasChoices("CDC_ANALYSIS_ENABLED"),
+    )
 
     api_v1_prefix: str = "/api/v1"
     log_level: str = "INFO"
@@ -119,6 +123,13 @@ def effective_database_url(self) -> str:
             return dsn.replace("postgres://", "postgresql+asyncpg://", 1)
         return dsn
 
+    @property
+    def effective_cdc_analysis_enabled(self) -> bool:
+        if self.cdc_analysis_enabled is not None:
+            return self.cdc_analysis_enabled
+
+        return self.app_mode in {"server", "analysis-server"}
+
 
 @lru_cache
 def get_settings() -> Settings:

app/infra/kafka/recommendation_producer.py

@@ -0,0 +1,62 @@
+"""추천 결과 발행용 Kafka producer. 앱 lifespan 동안 1개 인스턴스를 재사용한다."""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from app.core.config import Settings, get_settings
+from app.infra.kafka.client_options import build_kafka_client_options
+
+logger = logging.getLogger(__name__)
+
+_producer: Any = None
+
+
+async def start_recommendation_kafka_producer(settings: Settings | None = None) -> None:
+    """bootstrap이 설정된 경우에만 producer를 생성·시작한다. 실패 시 로그만 남기고 계속 기동."""
+    global _producer
+    if _producer is not None:
+        return
+    try:
+        from aiokafka import AIOKafkaProducer
+    except ImportError:
+        logger.info("aiokafka 미설치, 추천용 Kafka producer 생략")
+        return
+
+    s = settings or get_settings()
+    bootstrap = (s.kafka_bootstrap_servers or "").strip()
+    if not bootstrap:
+        logger.info("kafka_bootstrap_servers 비어 있음, 추천용 Kafka producer 생략")
+        return
+
+    producer = AIOKafkaProducer(
+        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
+        **build_kafka_client_options(s),
+    )
+    try:
+        await producer.start()
+    except Exception:
+        logger.exception("추천용 Kafka producer 시작 실패 (bootstrap=%s)", bootstrap[:80])
+        return
+    _producer = producer
+    logger.info("추천용 Kafka producer 시작됨 topic=%s", getattr(s, "kafka_recommendation_topic", ""))
+
+
+async def stop_recommendation_kafka_producer() -> None:
+    global _producer
+    if _producer is None:
+        return
+    try:
+        await _producer.stop()
+    except Exception:
+        logger.exception("추천용 Kafka producer 종료 중 오류")
+    finally:
+        _producer = None
+        logger.info("추천용 Kafka producer 종료됨")
+
+
+def get_recommendation_kafka_producer() -> Any:
+    """시작된 공유 producer. 없으면 None (스크립트·미설정 환경)."""
+    return _producer

app/infra/openai/app_client.py

@@ -0,0 +1,46 @@
+"""앱 lifespan에서 유지하는 AsyncOpenAI 클라이언트 (연결 풀 재사용)."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.core.config import Settings, get_settings
+
+logger = logging.getLogger(__name__)
+
+_client: Any = None
+
+
+def start_openai_client(settings: Settings | None = None) -> None:
+    """OPENAI_API_KEY가 있을 때만 공유 클라이언트를 만든다. 동기 함수(내부 I/O 없음)."""
+    global _client
+    if _client is not None:
+        return
+    s = settings or get_settings()
+    api_key = (getattr(s, "openai_api_key", "") or "").strip()
+    if not api_key:
+        logger.info("OPENAI_API_KEY 비어 있음, 공유 AsyncOpenAI 클라이언트 생략")
+        return
+    from openai import AsyncOpenAI
+
+    _client = AsyncOpenAI(api_key=api_key)
+    logger.info("공유 AsyncOpenAI 클라이언트 생성됨")
+
+
+async def stop_openai_client() -> None:
+    global _client
+    if _client is None:
+        return
+    try:
+        await _client.close()
+    except Exception:
+        logger.exception("AsyncOpenAI close 중 오류")
+    finally:
+        _client = None
+        logger.info("공유 AsyncOpenAI 클라이언트 종료됨")
+
+
+def get_openai_client() -> Any:
+    """lifespan에서 초기화된 클라이언트. 없으면 None (스크립트·키 미설정 기동)."""
+    return _client

app/realtime/api/v1/recommendation.py

@@ -1,4 +1,8 @@
-from fastapi import APIRouter, BackgroundTasks, Depends
+import logging
+import time
+import uuid
+
+from fastapi import APIRouter, BackgroundTasks, Depends, Request
 from fastapi.responses import Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -7,18 +11,35 @@
 from app.services.recommendation_service import run_recommendation_and_publish_to_kafka
 
 router = APIRouter()
+logger = logging.getLogger(__name__)
 
 
 @router.post("/recommendations", status_code=202)
 async def post_recommendations(
     body: RecommendationRequest,
     background_tasks: BackgroundTasks,
+    request: Request,
     session: AsyncSession = Depends(get_db_session),
 ) -> Response:
     """
     202 Accepted 즉시 반환. 백그라운드에서 추천 생성 후 Kafka recommendation-topic 발행.
     Spring이 Kafka consume → persona_recommendation 적재 → CompletableFuture.complete(결과).
     """
     _ = session
-    background_tasks.add_task(run_recommendation_and_publish_to_kafka, body.member_id)
+    t0 = time.monotonic()
+    trace_id = (
+        (request.headers.get("x-trace-id") or "").strip()
+        or uuid.uuid4().hex[:12]
+    )
+    member_id = body.member_id
+
+    logger.info("[REC][trace_id=%s][member_id=%s] fastapi_start", trace_id, member_id)
+    background_tasks.add_task(run_recommendation_and_publish_to_kafka, member_id, trace_id)
+    total_ms = int((time.monotonic() - t0) * 1000)
+    logger.info(
+        "[REC][trace_id=%s][member_id=%s] fastapi_end total_ms=%s",
+        trace_id,
+        member_id,
+        total_ms,
+    )
     return Response(status_code=202, content=None)

app/realtime/main.py

@@ -11,6 +11,15 @@
 
 from app.core.config import get_settings
 from app.core.logging import configure_logging
+from app.infra.kafka.recommendation_producer import (
+    start_recommendation_kafka_producer,
+    stop_recommendation_kafka_producer,
+)
+from app.infra.openai.app_client import (
+    get_openai_client,
+    start_openai_client,
+    stop_openai_client,
+)
 from app.realtime.api.router import api_router
 from app.services.cdc_analysis_service import CdcAnalysisService
 
@@ -30,29 +39,48 @@ async def lifespan(application: FastAPI):
     runtime_settings = get_settings()
     cdc_service = None
 
-    if runtime_settings.cdc_analysis_enabled:
+    if runtime_settings.effective_cdc_analysis_enabled:
         cdc_service = CdcAnalysisService(runtime_settings)
         await cdc_service.start()
         application.state.cdc_service = cdc_service
-        logging.info("CDC analysis service enabled inside unified intelligence runtime.")
+        logging.info("APP_MODE: %s", runtime_settings.app_mode or "(unset)")
+        logging.info(
+            "CDC analysis enabled: %s",
+            runtime_settings.effective_cdc_analysis_enabled,
+        )
+
+    start_openai_client(runtime_settings)
+    application.state.openai_client = get_openai_client()
+
+    await start_recommendation_kafka_producer(runtime_settings)
 
     try:
         yield
     finally:
+        await stop_recommendation_kafka_producer()
+        await stop_openai_client()
+        application.state.openai_client = None
         if cdc_service is not None:
             await cdc_service.stop()
 
 
 def create_app() -> FastAPI:
-    application = FastAPI(title=settings.app_name, lifespan=lifespan)
+    application = FastAPI(
+        title=settings.app_name,
+        lifespan=lifespan,
+    )
     application.include_router(api_router, prefix=settings.api_v1_prefix)
 
     @application.on_event("startup")
     async def log_db_target() -> None:
         runtime_settings = get_settings()
         url = runtime_settings.effective_database_url
         logging.info("DB 연결 대상: %s", _mask_database_url(url))
-        logging.info("CDC analysis enabled: %s", runtime_settings.cdc_analysis_enabled)
+        logging.info("APP_MODE: %s", runtime_settings.app_mode or "(unset)")
+        logging.info(
+            "CDC analysis enabled: %s",
+            runtime_settings.effective_cdc_analysis_enabled,
+        )
 
     @application.get("/")
     async def root() -> dict[str, str]:

app/services/persona_recommendation_prompts.py

@@ -3,6 +3,10 @@
 명세 8장: segment가 추천 제약, persona가 마케팅·reason 스타일 결정.
 """
 
+from __future__ import annotations
+
+import re
+
 
 SEGMENT_SYSTEM_PROMPTS = {
     "CHURN_RISK": """당신은 통신사 고객 이탈 방지 전문 추천 AI예요.
@@ -81,6 +85,62 @@ def get_persona_style_prompt(persona_code: str | None) -> str:
     return PERSONA_STYLE_PROMPTS.get(code, PERSONA_STYLE_PROMPTS["SPACE_EXPLORER"])
 
 
+def _extract_block(raw: str, label: str) -> str:
+    # [라벨] 다음 줄부터 다음 [라벨] 이전까지를 추출한다.
+    pattern = rf"{re.escape(label)}\s*(.*?)(?=\n\[[^\]]+\]|\Z)"
+    m = re.search(pattern, raw, flags=re.DOTALL)
+    return (m.group(1).strip() if m else "")
+
+
+def _compact_embedding_text(text: str, max_chars: int = 250, max_target_reasons: int = 2) -> str:
+    raw = (text or "").strip()
+    if not raw:
+        return ""
+
+    name = _extract_block(raw, "[상품명]")
+    price = _extract_block(raw, "[가격]")
+    ptype = _extract_block(raw, "[상품유형]")
+    tags = _extract_block(raw, "[태그]")
+    target_section = _extract_block(raw, "[추천 대상/상황]")
+
+    compact_lines: list[str] = []
+    if name:
+        compact_lines.append(f"[상품명] {name}")
+    if price:
+        compact_lines.append(f"[가격] {price}")
+    if ptype:
+        compact_lines.append(f"[상품유형] {ptype}")
+    if tags:
+        tag_items = [t.strip() for t in tags.split(",") if t.strip()]
+        if tag_items:
+            compact_lines.append(f"[태그] {', '.join(tag_items[:2])}")
+
+    if target_section:
+        pairs: list[str] = []
+        chunks = [c.strip() for c in target_section.split("|") if c.strip()]
+        for chunk in chunks:
+            t = re.search(r"\[대상\]\s*([^\[\]|·]+)", chunk)
+            r = re.search(r"\[추천 이유\]\s*([^\[\]|]+)", chunk)
+            target = (t.group(1).strip() if t else "")
+            reason = (r.group(1).strip() if r else "")
+            if target and reason:
+                pairs.append(f"[대상/이유] {target} - {reason}")
+            if len(pairs) >= max_target_reasons:
+                break
+        compact_lines.extend(pairs)
+
+    compact = "\n".join(compact_lines).strip()
+    if compact:
+        return compact[:max_chars]
+
+    # 구조화 태그가 없는 경우 문장 기반 폴백
+    chunks = [s.strip() for s in re.split(r"[.!?\n]+", raw) if s and s.strip()]
+    fallback = ". ".join(chunks[:3]).strip()
+    if fallback and not fallback.endswith("."):
+        fallback += "."
+    return fallback[:max_chars]
+
+
 def format_products(products: list[dict]) -> str:
     """LLM 유저 프롬프트에 넣을 상품 목록 문자열. 명세 8장 포맷."""
     lines = []
@@ -94,7 +154,11 @@ def format_products(products: list[dict]) -> str:
             tags_str = ", ".join(str(t) for t in tags)
         else:
             tags_str = str(tags) if tags else ""
-        embedding_text = (p.get("embedding_text") or "").strip()
+        embedding_text = _compact_embedding_text(
+            (p.get("embedding_text") or "").strip(),
+            max_chars=250,
+            max_target_reasons=2,
+        )
         lines.append(
             f"{i}. product_id: {p.get('product_id')}\n"
             f"   product_name: {product_name}\n"
@@ -128,7 +192,7 @@ def build_user_prompt(ctx: dict, products_text: str) -> str:
     ]
     if recent_counseling:
         blocks.append("## 최근 상담 이력")
-        blocks.append(recent_counseling[:800])
+        blocks.append(recent_counseling[:300])
     blocks.append("## 후보 상품 목록 (이 중에서 최대 3개를 골라 추천하고, 각 product_id에 대해 reason을 작성하세요)")
     blocks.append(products_text)
 

app/services/recommendation/__init__.py

@@ -0,0 +1,17 @@
+"""개인화 추천 파이프라인 (컨텍스트 로드·검색·LLM·Kafka)."""
+
+from .kafka_publish import publish_recommendation_to_kafka
+from .service import (
+    RecommendationService,
+    get_recommendation,
+    run_recommendation_and_publish_to_kafka,
+)
+from .weights import compute_product_type_weights
+
+__all__ = [
+    "RecommendationService",
+    "compute_product_type_weights",
+    "get_recommendation",
+    "publish_recommendation_to_kafka",
+    "run_recommendation_and_publish_to_kafka",
+]