v1.2.0

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/main.py

@@ -11,7 +11,15 @@
 
 from app.core.config import get_settings
 from app.core.logging import configure_logging
-from app.infra.pinpoint import build_fastapi_pinpoint_middleware
+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
 
@@ -42,9 +50,17 @@ async def lifespan(application: FastAPI):
             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()
 

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",
+]

app/services/recommendation/constants.py

@@ -0,0 +1,126 @@
+"""추천 파이프라인 SQL·상수."""
+
+from sqlalchemy import bindparam, text
+
+FETCH_MEMBER_LLM_CONTEXT_SQL = text("""
+SELECT
+    member_id, membership, age_group, join_months, children_count,
+    family_group_num, family_role, persona_code, segment,
+    current_subscriptions, current_product_types, product_type_clicks, current_data_usage_ratio,
+    data_usage_pattern, churn_score, churn_tier, recent_counseling,
+    recent_viewed_tags_top_3, contract_expiry_within_3m, updated_at
+FROM member_llm_context
+WHERE member_id = :member_id
+""")
+
+FETCH_SUBSCRIPTION_PRICES_SQL = text("""
+SELECT product_id, price, sale_price, product_type
+FROM product
+WHERE product_id IN :ids
+""").bindparams(bindparam("ids", expanding=True))
+
+SEARCH_SIMILAR_SQL = text("""
+SELECT product_id
+FROM product
+WHERE embedding_vector IS NOT NULL
+  AND (NOT (product_id = ANY(:exclude_ids)))
+ORDER BY embedding_vector <#> :query_vec
+LIMIT :k
+""")
+
+SEARCH_SIMILAR_WITH_TYPE_BOOST_SQL = text("""
+SELECT product_id
+FROM product
+WHERE embedding_vector IS NOT NULL
+  AND (NOT (product_id = ANY(:exclude_ids)))
+ORDER BY (embedding_vector <#> :query_vec)
+  - (CASE
+       WHEN product_type = :boost_type1 THEN :boost1
+       WHEN product_type = :boost_type2 THEN :boost2
+       ELSE 0
+     END)
+LIMIT :k
+""")
+
+SEARCH_SIMILAR_MAIN_TYPES_WINDOW_SQL = text("""
+SELECT product_id, product_type, rn
+FROM (
+  SELECT
+    product_id,
+    product_type,
+    ROW_NUMBER() OVER (
+      PARTITION BY product_type
+      ORDER BY embedding_vector <#> :query_vec
+    ) AS rn
+  FROM product
+  WHERE embedding_vector IS NOT NULL
+    AND product_type IN :main_types
+    AND (NOT (product_id = ANY(:exclude_ids)))
+) ranked
+WHERE ranked.rn <= :per_type_k
+""").bindparams(bindparam("main_types", expanding=True))
+
+FETCH_PRODUCTS_FULL_SQL = text("""
+SELECT
+    p.product_id, p.name, p.product_type, p.price, p.sale_price, p.tags, p.embedding_text,
+    COALESCE(mp.data_amount, tw.data_amount) AS data_amount
+FROM product p
+LEFT JOIN mobile_plan mp ON p.product_id = mp.product_id
+LEFT JOIN tab_watch_plan tw ON p.product_id = tw.product_id
+WHERE p.product_id IN :ids
+""").bindparams(bindparam("ids", expanding=True))
+
+FETCH_DEFAULT_PRODUCTS_SQL = text("""
+SELECT
+    p.product_id, p.name, p.product_type, p.price, p.sale_price, p.tags, p.embedding_text,
+    NULL::integer AS data_amount
+FROM product p
+WHERE p.embedding_text IS NOT NULL
+ORDER BY p.product_id
+LIMIT :k
+""")
+
+RETRIEVAL_CANDIDATES_K = 50
+
+MAIN_PRODUCT_TYPES: tuple[str, ...] = (
+    "MOBILE_PLAN",
+    "INTERNET",
+    "IPTV",
+    "TAB_WATCH_PLAN",
+    "ADDON",
+)
+RETRIEVAL_PER_TYPE_K = 10
+
+DEFAULT_RETRIEVAL_QUERY = "통신 요금제, 데이터 요금제, 부가서비스 추천"
+
+EMBEDDING_DIMENSION = 1536
+
+UNLIMITED_DATA_TAG_MARKER = "무제한"
+
+MAX_PRODUCTS_PER_TYPE = 2
+
+DEFAULT_PRODUCT_REASON_TEXT = "고객님께 가장 적합한 상품을 추천드립니다!"
+NO_CANDIDATE_MESSAGE = "추천할 수 있는 상품이 없습니다."
+NO_MATCHED_MESSAGE = "조건에 맞는 추천 상품이 없습니다."
+FALLBACK_CACHED_MESSAGE = "고객님의 이용 패턴과 관심사를 반영한 요금제·부가서비스 추천입니다."
+FALLBACK_VECTOR_SUMMARY = "고객님께 어울리는 상품을 추천해드릴게요!"
+
+CHURN_MAX_PRICE_RATIO = 1.1
+
+SEGMENT_WEIGHT_CONFIG: dict[str, dict[str, float]] = {
+    "CHURN_RISK": {
+        "current_type": 1.5,
+        "click": 0.7,
+        "tag": 0.5,
+    },
+    "UPSELL": {
+        "current_type": 1.0,
+        "click": 1.5,
+        "tag": 1.2,
+    },
+    "NORMAL": {
+        "current_type": 0.8,
+        "click": 1.3,
+        "tag": 1.3,
+    },
+}

app/services/recommendation/context_loader.py

@@ -0,0 +1,36 @@
+"""member_llm_context 로딩."""
+
+import logging
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .constants import FETCH_MEMBER_LLM_CONTEXT_SQL
+
+logger = logging.getLogger(__name__)
+
+
+async def load_member_llm_context(session: AsyncSession, member_id: int) -> dict | None:
+    try:
+        ctx_result = await session.execute(
+            FETCH_MEMBER_LLM_CONTEXT_SQL,
+            {"member_id": member_id},
+        )
+        row = ctx_result.fetchone()
+        if row is None:
+            logger.info(
+                "recommendation: member_llm_context 행 없음 (member_id=%s). 테이블은 있으나 해당 회원 데이터 없음 → 폴백",
+                member_id,
+            )
+            return None
+        return dict(row._mapping) if hasattr(row, "_mapping") else dict(row)
+    except Exception as e:
+        logger.info(
+            "recommendation: member_llm_context 조회 실패 → 폴백 예정 (member_id=%s, error=%s)",
+            member_id,
+            e,
+        )
+        try:
+            await session.rollback()
+        except Exception:
+            pass  # best-effort: 이미 닫힌 세션 등에서 rollback 실패 가능
+        return None

app/services/recommendation/kafka_publish.py

@@ -0,0 +1,55 @@
+"""추천 결과 Kafka 발행."""
+
+from __future__ import annotations
+
+import json
+import logging
+
+try:
+    from aiokafka import AIOKafkaProducer
+except Exception:  # pragma: no cover
+    AIOKafkaProducer = None  # type: ignore[assignment]
+
+from app.core.config import get_settings
+from app.infra.kafka.client_options import build_kafka_client_options
+from app.infra.kafka.recommendation_producer import get_recommendation_kafka_producer
+from app.schemas.recommendation import RecommendationResponse
+
+logger = logging.getLogger(__name__)
+
+
+async def publish_recommendation_to_kafka(
+    member_id: int,
+    response: RecommendationResponse,
+    trace_id: str | None = None,
+) -> None:
+    settings = get_settings()
+    topic = getattr(settings, "kafka_recommendation_topic", "recommendation")
+    bootstrap = getattr(settings, "kafka_bootstrap_servers", "").strip()
+    if AIOKafkaProducer is None:
+        logger.warning("recommendation: aiokafka 미설치, Kafka 발행 스킵 member_id=%s", member_id)
+        return
+    if not bootstrap:
+        logger.warning("recommendation: Kafka 미설정, 발행 스킵 member_id=%s", member_id)
+        return
+    tid = (trace_id or "").strip()
+    payload = {"traceId": tid, "memberId": member_id, **response.model_dump(by_alias=True)}
+
+    shared = get_recommendation_kafka_producer()
+    owned = False
+    producer = shared
+    if producer is None:
+        producer = AIOKafkaProducer(
+            value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
+            **build_kafka_client_options(settings),
+        )
+        await producer.start()
+        owned = True
+    try:
+        await producer.send_and_wait(topic, value=payload, key=str(member_id).encode("utf-8"))
+        logger.info("recommendation: Kafka 발행 완료 member_id=%s topic=%s", member_id, topic)
+    except Exception as e:
+        logger.error("recommendation: Kafka 발행 실패 member_id=%s: %s", member_id, e, exc_info=True)
+    finally:
+        if owned:
+            await producer.stop()