# AI 경품 추천 구현 방안 ## 1. 개요 ### 1.1 목적 매장의 업종, 지역, 시즌 등 컨텍스트를 분석하여 최적의 경품을 추천하고, 추천 이유를 명확히 설명하는 AI 기반 추천 시스템 구현 ### 1.2 핵심 전략 **RAG(Retrieval-Augmented Generation) 기반 하이브리드 추천** - 벡터 DB에서 유사 성공 사례 검색 → Claude API로 컨텍스트 기반 재해석 및 창의적 변형 - Redis Vector Search + Claude API 3.5 Sonnet 활용 - Prompt Caching으로 비용 절감 및 응답 속도 개선 --- ## 2. 데이터 수집 및 처리 ### 2.1 매장 데이터 수집 #### 2.1.1 입력 데이터 구조 ```json { "store_id": "string", "business_number": "string", // 사업자등록번호 "category": { "main": "string", // 대분류: 음식점, 소매업, 서비스업 "sub": "string", // 중분류: 한식, 양식, 카페 "detail": "string" // 소분류: 고깃집, 베이커리 }, "location": { "sido": "string", // 서울특별시 "sigungu": "string", // 강남구 "dong": "string", // 역삼동 "latitude": "float", "longitude": "float" }, "budget": { "min": "integer", "max": "integer" }, "target_customer": { "age_group": ["string"], // ["20대", "30대"] "gender": "string", // "전체", "남성", "여성" "income_level": "string" // "중상", "중", "중하" }, "event_purpose": "string", // "신규 고객 유치", "단골 유지", "재방문 유도" "season": "string" // "2025-Q1", auto-calculated } ``` #### 2.1.2 데이터 자동 수집 프로세스 ```mermaid graph LR A[사업자등록번호 입력] --> B[공공데이터 API 조회] B --> C[업종/주소 자동 매핑] C --> D[좌표 변환 Kakao Map API] D --> E[상권 분석 데이터 결합] E --> F[매장 프로필 생성] ``` **외부 API 연동** 1. **공공데이터포털 - 상권정보 조회** - 엔드포인트: `https://api.odcloud.kr/api/StoreInfo/v1/getStoreInfo` - 수집 항목: 업종, 주소, 개업일 2. **Kakao Local API - 좌표 변환** - 엔드포인트: `https://dapi.kakao.com/v2/local/search/address.json` - 수집 항목: 위경도, 행정구역 코드 3. **서울 열린데이터광장 - 상권 분석** - 엔드포인트: `https://data.seoul.go.kr/dataList/OA-15572/S/1/datasetView.do` - 수집 항목: 유동인구, 매출 추정, 경쟁 업체 수 ### 2.2 경품 데이터 수집 #### 2.2.1 데이터 소스 **1) 내부 데이터 (과거 이벤트 성과)** ```json { "event_id": "string", "store_profile": { /* 2.1.1 구조 동일 */ }, "prize": { "name": "string", "category": "string", // "식음료", "생활용품", "체험권", "할인쿠폰" "unit_price": "integer", "quantity": "integer", "supplier": "string" }, "performance": { "participant_count": "integer", "conversion_rate": "float", // 참여자 중 구매 전환율 "roi": "float", // 매출 증대액 / 경품 비용 "customer_satisfaction": "float" // 1~5점 } } ``` **2) 외부 트렌드 데이터** - **네이버 쇼핑 트렌드**: 주간/월간 인기 상품 키워드 - API: `https://developers.naver.com/docs/serviceapi/datalab/shopping/shopping.md` - **SNS 트렌드 (Instagram, TikTok)**: 해시태그 순위 - 크롤링 또는 서드파티 API (예: Apify) - **시즌 이벤트 캘린더**: 공휴일, 기념일, 페스티벌 - 수동 관리 또는 공공 캘린더 API #### 2.2.2 경품 카탈로그 구축 ```json { "prize_id": "string", "name": "string", "category": "string", "subcategory": "string", "price_range": { "min": "integer", "max": "integer" }, "target_demographic": { "age": ["string"], "gender": "string" }, "seasonality": ["string"], // ["봄", "여름", "크리스마스"] "trend_score": "float", // 최근 1개월 트렌드 점수 (0~1) "compatibility": { "food": 0.9, "retail": 0.7, "service": 0.5 } } ``` --- ## 3. 데이터 정제 및 벡터화 ### 3.1 정제 파이프라인 ```mermaid graph TD A[원본 데이터] --> B[결측치 처리] B --> C[이상치 제거] C --> D[정규화] D --> E[카테고리 인코딩] E --> F[시계열 특성 추출] F --> G[벡터화] ``` #### 3.1.1 매장 데이터 정제 **1) 결측치 처리** - `budget`: 업종별 평균값으로 대체 - `target_customer`: "전체"로 기본값 설정 - `location.dong`: 시군구 단위로 통합 **2) 정규화** - 예산: Min-Max Scaling (0~1) - 좌표: Standard Scaling (평균 0, 분산 1) **3) 카테고리 인코딩** ```python # One-Hot Encoding for categorical features 업종_대분류 = OneHotEncoder(['음식점', '소매업', '서비스업']) 업종_중분류 = LabelEncoder() # 100+ categories → Integer 지역_시도 = OneHotEncoder(['서울', '경기', '인천', ...]) ``` **4) 시계열 특성 (Cyclical Encoding)** ```python # 계절성 표현: 봄(3~5월) = 0°, 여름 = 90°, 가을 = 180°, 겨울 = 270° month = datetime.now().month season_sin = np.sin(2 * np.pi * month / 12) season_cos = np.cos(2 * np.pi * month / 12) ``` #### 3.1.2 경품 데이터 정제 **1) 트렌드 점수 계산** ```python # 네이버 쇼핑 트렌드 API 응답 기반 trend_score = ( 0.4 * 검색량_증가율 + 0.3 * 클릭률 + 0.3 * SNS_언급_빈도 ) ``` **2) 호환성 매트릭스** ```python # 과거 성과 데이터 기반 업종별 선호도 compatibility_matrix = { ('음식점', '식음료 쿠폰'): 0.95, ('음식점', '생활용품'): 0.6, ('소매업', '할인쿠폰'): 0.9, ... } ``` ### 3.2 벡터화 전략 #### 3.2.1 매장 임베딩 (Store Embedding) **차원 구성 (총 128차원)** ```python store_vector = np.concatenate([ 업종_one_hot, # 20차원 지역_demographic, # 30차원 (인구 밀도, 소득 수준, 연령 분포 등) 예산_normalized, # 1차원 타겟고객_encoded, # 10차원 시즌_cyclical, # 2차원 (sin, cos) 이벤트_목적_one_hot, # 5차원 상권_특성 # 60차원 (유동인구, 경쟁 업체 수, 매출 추정치 등) ]) ``` **임베딩 모델** - **Option 1**: 수동 Feature Engineering (위 구조) - **Option 2**: Pre-trained Model (추후 고도화) - `sentence-transformers/all-MiniLM-L6-v2` 활용 - 매장 설명 텍스트를 자연어로 생성 후 임베딩 #### 3.2.2 경품 임베딩 (Prize Embedding) **차원 구성 (총 128차원 - Store와 동일 공간)** ```python prize_vector = np.concatenate([ 카테고리_one_hot, # 20차원 가격_normalized, # 1차원 타겟_demographic, # 10차원 시즌_cyclical, # 2차원 트렌드_score, # 1차원 호환성_matrix, # 20차원 설명_embedding # 74차원 (sentence-transformers) ]) ``` #### 3.2.3 성공 사례 임베딩 (Case Embedding) **하이브리드 접근** ```python case_vector = np.concatenate([ store_vector, # 128차원 prize_vector, # 128차원 performance_features # 16차원 (ROI, 전환율, 만족도 등) ]) # 총 272차원 ``` ### 3.3 Redis Vector Search 저장 #### 3.3.1 인덱스 생성 ```python from redis.commands.search.field import VectorField, TextField, NumericField from redis.commands.search.indexDefinition import IndexDefinition, IndexType schema = ( VectorField("case_vector", "FLAT", { "TYPE": "FLOAT32", "DIM": 272, "DISTANCE_METRIC": "COSINE" }), TextField("store_category"), TextField("prize_name"), NumericField("roi"), NumericField("conversion_rate"), NumericField("timestamp") ) redis_client.ft("idx:prize_cases").create_index( schema, definition=IndexDefinition(prefix=["case:"], index_type=IndexType.HASH) ) ``` #### 3.3.2 데이터 저장 예시 ```python import numpy as np import pickle case_id = "case:12345" redis_client.hset(case_id, mapping={ "case_vector": pickle.dumps(case_vector.astype(np.float32)), "store_category": "음식점 > 한식 > 고깃집", "prize_name": "한우 선물세트", "roi": 3.2, "conversion_rate": 0.15, "timestamp": "2025-01-15" }) ``` --- ## 4. Claude API 호출 구조 ### 4.1 프롬프트 설계 #### 4.1.1 시스템 프롬프트 (Prompt Caching 적용) ```json { "role": "system", "content": [ { "type": "text", "text": "당신은 소상공인 이벤트 기획 전문가입니다. 매장 정보와 유사 성공 사례를 바탕으로 최적의 경품을 추천하고, 그 이유를 명확히 설명해야 합니다.\n\n[역할]\n- 업종, 지역, 예산, 타겟 고객을 종합 분석\n- 과거 성공 사례 패턴 학습\n- 창의적이면서도 실현 가능한 경품 제안\n- 추천 이유를 구체적 데이터로 뒷받침\n\n[제약사항]\n- 예산 범위 내 경품만 추천\n- 법적 문제 없는 경품 (사행성 금지)\n- 조달 가능한 경품 우선", "cache_control": {"type": "ephemeral"} } ] } ``` #### 4.1.2 Few-Shot Examples (Prompt Caching 적용) ```json { "role": "user", "content": [ { "type": "text", "text": "[사례 1]\n매장: 강남역 고깃집, 예산 50만원, 타겟 20~30대 직장인\n유사 성공 사례: 역삼동 삼겹살집이 '소주 1+1 쿠폰' 이벤트로 재방문율 40% 증가\n\n[사례 2]\n매장: 홍대 카페, 예산 30만원, 타겟 10~20대 학생\n유사 성공 사례: 신촌 디저트카페가 'AirPods 추첨' 이벤트로 SNS 공유 200% 증가\n\n[사례 3]\n매장: 이태원 양식당, 예산 100만원, 타겟 30~40대 고소득층\n유사 성공 사례: 청담동 스테이크 하우스가 '와인 시음권' 이벤트로 객단가 25% 상승", "cache_control": {"type": "ephemeral"} } ] } ``` #### 4.1.3 User Request (실시간 입력) ```json { "role": "user", "content": [ { "type": "text", "text": "## 매장 정보\n- 업종: 음식점 > 한식 > 고깃집\n- 위치: 서울 강남구 역삼동\n- 예산: 50만원\n- 타겟 고객: 20~30대 직장인, 남녀 모두\n- 이벤트 목적: 신규 고객 유치\n- 시즌: 2025년 1분기 (겨울)\n\n## 유사 성공 사례 (벡터 검색 결과)\n1. [ROI 3.2] 역삼동 삼겹살집 - 소주 1+1 쿠폰 (예산 40만원, 전환율 15%)\n2. [ROI 2.8] 서초동 한우집 - 한우 선물세트 추첨 (예산 60만원, 전환율 12%)\n3. [ROI 2.5] 강남역 숯불구이 - 배달앱 할인쿠폰 (예산 30만원, 전환율 18%)\n\n## 요청사항\n위 정보를 바탕으로 다음을 제공하세요:\n1. 최적 경품 1개 (구체적 상품명, 수량, 예상 비용)\n2. 대안 경품 2개\n3. 각 경품의 추천 이유 (과거 사례 근거 포함)\n4. 예상 효과 (참여자 수, 전환율, ROI)\n\nJSON 형식으로 응답해주세요." } ] } ``` ### 4.2 API 요청 구조 #### 4.2.1 Request 예시 ```json { "model": "claude-3-5-sonnet-20241022", "max_tokens": 4096, "temperature": 0.7, "system": [ { "type": "text", "text": "당신은 소상공인 이벤트 기획 전문가입니다...", "cache_control": {"type": "ephemeral"} } ], "messages": [ { "role": "user", "content": "[사례 1]\n...", "cache_control": {"type": "ephemeral"} }, { "role": "user", "content": "## 매장 정보\n..." } ], "stream": true } ``` **Prompt Caching 효과** - System Prompt (약 500 tokens): 캐싱 후 재사용 시 90% 비용 절감 - Few-Shot Examples (약 800 tokens): 캐싱 후 재사용 시 90% 비용 절감 - 5분 내 동일 프롬프트 재사용 시 캐시 히트 #### 4.2.2 Response 구조 (Streaming) ```json { "id": "msg_01XYZ...", "type": "message", "role": "assistant", "content": [ { "type": "text", "text": "{\n \"recommendations\": [\n {\n \"rank\": 1,\n \"prize\": {\n \"name\": \"1++등급 한우 등심 200g 교환권\",\n \"category\": \"식음료\",\n \"quantity\": 20,\n \"unit_price\": 25000,\n \"total_cost\": 500000,\n \"supplier\": \"한우자조금\"\n },\n \"reasoning\": {\n \"why_this_prize\": \"고깃집 업종 특성상 '고기 품질'에 대한 직접 경험이 재방문 유도에 가장 효과적입니다. 역삼동 삼겹살집 사례(ROI 3.2)에서 '소주 쿠폰'보다 '한우 선물세트'가 고급화 전략에 적합합니다.\",\n \"target_alignment\": \"20~30대 직장인은 '가성비'보다 '특별한 경험'을 선호하는 경향(네이버 쇼핑 트렌드 2025.01 기준, 프리미엄 식품 검색량 35% 증가)\",\n \"budget_efficiency\": \"건당 2.5만원으로 20명 당첨 시 예산 100% 활용. 서초동 한우집 사례 대비 20% 저렴한 비용으로 유사 효과 기대\",\n \"seasonal_relevance\": \"겨울철 보양식 수요 증가 (1분기 한우 판매량 전년 대비 18% 상승, 농림축산식품부)\"\n },\n \"expected_outcome\": {\n \"participant_count\": 300,\n \"conversion_rate\": 0.14,\n \"new_customers\": 42,\n \"roi\": 3.0,\n \"rationale\": \"유사 사례(역삼동 삼겹살집) 전환율 15% 대비 소폭 하락 예상. 단, 한우 브랜드 신뢰도로 고객 만족도 상승 효과\"\n }\n },\n {\n \"rank\": 2,\n \"prize\": {\n \"name\": \"배달앱 5천원 할인쿠폰 (100매)\",\n \"category\": \"할인쿠폰\",\n \"quantity\": 100,\n \"unit_price\": 5000,\n \"total_cost\": 500000\n },\n \"reasoning\": {\n \"why_this_prize\": \"강남역 숯불구이 사례(ROI 2.5, 전환율 18%)에서 입증된 효과. 즉시 사용 가능한 혜택으로 참여 장벽 낮춤\",\n \"target_alignment\": \"배달 이용률 높은 직장인 타겟 적합 (20~30대 배달앱 이용률 주 3회 이상 68%, 통계청)\",\n \"budget_efficiency\": \"100명 당첨 가능, 경품 수량 많아 '당첨 가능성' 마케팅 효과\",\n \"seasonal_relevance\": \"겨울철 배달 수요 증가 (기온 5도 하락 시 배달 주문 12% 증가, 배민 데이터)\"\n },\n \"expected_outcome\": {\n \"participant_count\": 500,\n \"conversion_rate\": 0.18,\n \"new_customers\": 90,\n \"roi\": 2.5\n }\n },\n {\n \"rank\": 3,\n \"prize\": {\n \"name\": \"프리미엄 소주/와인 세트 (10세트)\",\n \"category\": \"식음료\",\n \"quantity\": 10,\n \"unit_price\": 50000,\n \"total_cost\": 500000\n },\n \"reasoning\": {\n \"why_this_prize\": \"고급화 전략 + SNS 공유 유도. '언박싱' 콘텐츠로 바이럴 효과 기대\",\n \"target_alignment\": \"MZ세대 '프리미엄 주류' 소비 증가 (위스키 수입액 2024년 전년 대비 23% 증가, 관세청)\",\n \"budget_efficiency\": \"고가 경품으로 '프리미엄 이미지' 구축, 브랜드 가치 상승 효과\",\n \"seasonal_relevance\": \"연말연시 선물 수요 연계 (1~2월 선물 세트 검색량 연중 최고치)\"\n },\n \"expected_outcome\": {\n \"participant_count\": 200,\n \"conversion_rate\": 0.10,\n \"new_customers\": 20,\n \"roi\": 2.2,\n \"rationale\": \"당첨자 수 적어 전환율 낮지만, SNS 공유로 브랜드 인지도 상승 효과 (간접 효과 미포함 시)\"\n }\n }\n ],\n \"metadata\": {\n \"analysis_timestamp\": \"2025-01-20T10:30:00Z\",\n \"vector_search_cases\": 3,\n \"trend_data_sources\": [\"네이버 쇼핑 트렌드\", \"농림축산식품부\", \"통계청\"]\n }\n}\n" } ], "model": "claude-3-5-sonnet-20241022", "usage": { "input_tokens": 2150, "cache_creation_input_tokens": 1300, "cache_read_input_tokens": 0, "output_tokens": 1024 } } ``` ### 4.3 응답 파싱 및 후처리 #### 4.3.1 JSON 검증 ```python import json from jsonschema import validate, ValidationError response_schema = { "type": "object", "required": ["recommendations"], "properties": { "recommendations": { "type": "array", "minItems": 3, "maxItems": 3, "items": { "type": "object", "required": ["rank", "prize", "reasoning", "expected_outcome"], "properties": { "rank": {"type": "integer", "minimum": 1, "maximum": 3}, "prize": { "type": "object", "required": ["name", "category", "quantity", "unit_price", "total_cost"] }, "reasoning": { "type": "object", "required": ["why_this_prize", "target_alignment", "budget_efficiency"] }, "expected_outcome": { "type": "object", "required": ["participant_count", "conversion_rate", "roi"] } } } } } } try: parsed_response = json.loads(claude_response) validate(instance=parsed_response, schema=response_schema) except (json.JSONDecodeError, ValidationError) as e: # Fallback: 룰 기반 추천 호출 logger.error(f"Claude response validation failed: {e}") return fallback_recommendation(store_data) ``` #### 4.3.2 예산 초과 필터링 ```python def filter_budget_overrun(recommendations, max_budget): filtered = [] for rec in recommendations: if rec['prize']['total_cost'] <= max_budget: filtered.append(rec) else: # 수량 조정으로 예산 내 맞추기 adjusted_quantity = max_budget // rec['prize']['unit_price'] if adjusted_quantity > 0: rec['prize']['quantity'] = adjusted_quantity rec['prize']['total_cost'] = adjusted_quantity * rec['prize']['unit_price'] rec['reasoning']['budget_efficiency'] += f" (수량 조정: {adjusted_quantity}개)" filtered.append(rec) return filtered[:3] # 최대 3개 유지 ``` --- ## 5. 전체 시스템 플로우 ### 5.1 추천 요청 흐름도 ```mermaid sequenceDiagram participant User as 사용자 participant API as API Gateway participant Svc as Prize Recommendation Service participant Cache as Redis Cache participant Vector as Redis Vector Search participant Claude as Claude API participant Fallback as Fallback Engine User->>API: POST /api/v1/prizes/recommend API->>Svc: 매장 정보 전달 alt 캐시 히트 Svc->>Cache: 동일 매장 프로필 조회 Cache-->>Svc: 캐시된 추천 결과 (TTL 1시간) Svc-->>API: 추천 결과 반환 else 캐시 미스 Svc->>Svc: 매장 데이터 정제 및 벡터화 Svc->>Vector: 유사 성공 사례 검색 (Top 3) Vector-->>Svc: 사례 데이터 반환 Svc->>Claude: Prompt 생성 및 API 호출 Note over Svc,Claude: System Prompt (캐싱)
Few-Shot Examples (캐싱)
User Request (실시간) alt Claude 성공 Claude-->>Svc: Streaming Response (JSON) Svc->>Svc: JSON 파싱 및 검증 Svc->>Cache: 추천 결과 캐싱 (TTL 1시간) Svc-->>API: 추천 결과 반환 else Claude 실패 Claude-->>Svc: Error (Timeout/Rate Limit) Svc->>Fallback: 룰 기반 추천 요청 Fallback-->>Svc: 기본 추천 결과 Svc-->>API: Fallback 결과 반환 end end API-->>User: HTTP 200 OK + JSON ``` ### 5.2 데이터 파이프라인 스케줄 ```mermaid gantt title 데이터 수집 및 갱신 스케줄 dateFormat YYYY-MM-DD section 외부 데이터 네이버 트렌드 수집 :active, trend, 2025-01-20, 7d SNS 트렌드 크롤링 :active, sns, 2025-01-20, 7d 공공데이터 동기화 :active, public, 2025-01-20, 7d section 벡터 재계산 성공 사례 임베딩 :active, embed, 2025-01-20, 1d 경품 카탈로그 갱신 :active, catalog, 2025-01-20, 1d section 모델 업데이트 Claude Prompt 최적화 :milestone, prompt, 2025-01-27, 0d ``` **스케줄 상세** - **외부 데이터 수집**: 매주 월요일 02:00 (Airflow DAG) - **벡터 재계산**: 매일 03:00 (신규 사례 추가 시) - **캐시 갱신**: - L1 (업종별 시즌 추천): TTL 7일 - L2 (개인화 추천): TTL 1시간 - **Prompt 튜닝**: 월 1회 (성과 데이터 기반 Few-Shot 사례 갱신) --- ## 6. Fallback 전략 ### 6.1 Claude API 장애 대응 #### 6.1.1 룰 기반 추천 엔진 ```python class FallbackRecommendation: def __init__(self): self.rules = self._load_rules() def recommend(self, store_data): # 1단계: 업종별 인기 경품 필터링 category_prizes = self.rules['category_mapping'][store_data['category']['main']] # 2단계: 예산 필터링 budget_filtered = [p for p in category_prizes if p['price'] <= store_data['budget']['max']] # 3단계: 시즌 가중치 적용 season = store_data['season'] for prize in budget_filtered: prize['score'] = prize['base_score'] * self.rules['season_weight'][season].get(prize['id'], 1.0) # 4단계: 상위 3개 선정 top_prizes = sorted(budget_filtered, key=lambda x: x['score'], reverse=True)[:3] return self._format_response(top_prizes, store_data) ``` #### 6.1.2 과거 인기 경품 순위 ```python def get_popular_prizes(category, budget_range, limit=3): """최근 3개월 성과 데이터 기반 인기 경품 조회""" query = """ SELECT prize_id, name, AVG(roi) as avg_roi, COUNT(*) as usage_count FROM event_performance WHERE category = %s AND budget BETWEEN %s AND %s AND created_at >= NOW() - INTERVAL 3 MONTH GROUP BY prize_id ORDER BY avg_roi DESC, usage_count DESC LIMIT %s """ return db.execute(query, [category, budget_range[0], budget_range[1], limit]) ``` ### 6.2 에러 처리 우선순위 ```mermaid graph TD A[Claude API 호출] --> B{성공?} B -->|Yes| C[JSON 검증] B -->|No| D{재시도 가능?} D -->|Yes| E[Exponential Backoff 재시도] E --> A D -->|No| F[Fallback: 룰 기반 추천] C --> G{검증 통과?} G -->|Yes| H[응답 반환] G -->|No| F F --> I[과거 인기 경품 조회] I --> J{데이터 있음?} J -->|Yes| K[기본 추천 반환] J -->|No| L[에러 메시지 + 수동 선택 유도] ``` --- ## 7. 성능 최적화 ### 7.1 캐싱 전략 #### 7.1.1 L1 캐시 (업종별 시즌 추천) ```python # Redis Key: "prize:season:{category}:{season}" # TTL: 7일 # 예시: "prize:season:음식점-한식:2025-Q1" cache_key = f"prize:season:{store_category}:{current_season}" cached = redis_client.get(cache_key) if cached: return json.loads(cached) else: result = claude_api.recommend(...) redis_client.setex(cache_key, 604800, json.dumps(result)) # 7일 = 604800초 return result ``` #### 7.1.2 L2 캐시 (개인화 추천) ```python # Redis Key: "prize:personalized:{store_id}:{hash(request)}" # TTL: 1시간 import hashlib request_hash = hashlib.md5(json.dumps(store_data, sort_keys=True).encode()).hexdigest() cache_key = f"prize:personalized:{store_id}:{request_hash}" cached = redis_client.get(cache_key) if cached: return json.loads(cached) else: result = claude_api.recommend(...) redis_client.setex(cache_key, 3600, json.dumps(result)) # 1시간 = 3600초 return result ``` ### 7.2 벡터 검색 최적화 #### 7.2.1 인덱스 파라미터 튜닝 ```python # FLAT vs HNSW 비교 # FLAT: 정확도 100%, 검색 속도 O(n) - 데이터 10만 건 이하 # HNSW: 정확도 95%+, 검색 속도 O(log n) - 대규모 데이터 # 초기 (성공 사례 < 10만 건): FLAT 사용 schema = VectorField("case_vector", "FLAT", {...}) # 확장 시 (성공 사례 > 10만 건): HNSW 전환 schema = VectorField("case_vector", "HNSW", { "TYPE": "FLOAT32", "DIM": 272, "DISTANCE_METRIC": "COSINE", "M": 16, # 연결 수 (높을수록 정확, 느림) "EF_CONSTRUCTION": 200 # 구축 시간 vs 정확도 트레이드오프 }) ``` #### 7.2.2 Pre-filtering ```python # 벡터 검색 전 메타데이터 필터링으로 검색 공간 축소 query = ( Query("(@category:{음식점}) (@budget:[0 500000])=>[KNN 3 @case_vector $vec AS score]") .sort_by("score") .return_fields("store_category", "prize_name", "roi", "score") .dialect(2) ) ``` ### 7.3 Claude API 비용 최적화 #### 7.3.1 Prompt Caching 효과 분석 ```python # Prompt Caching 적용 전 base_cost_per_request = ( (500 + 800 + 850) * 0.003 / 1000 # System + Few-Shot + User (입력) + 1024 * 0.015 / 1000 # 출력 ) = 0.00642 + 0.01536 = $0.02178 # Prompt Caching 적용 후 (캐시 히트 시) optimized_cost_per_request = ( (500 + 800) * 0.0003 / 1000 # 캐싱된 토큰 (90% 할인) + 850 * 0.003 / 1000 # 신규 User 토큰 + 1024 * 0.015 / 1000 # 출력 ) = 0.00039 + 0.00255 + 0.01536 = $0.0183 # 비용 절감율: (0.02178 - 0.0183) / 0.02178 = 16% ``` **월간 비용 추정 (일 1,000 요청 기준)** - 캐시 히트율 70% 가정 - 월 비용: (1000 * 30) * (0.0183 * 0.7 + 0.02178 * 0.3) = **$581** - Caching 미적용 시: (1000 * 30) * 0.02178 = **$653** - **절감액: $72/월 (11%)** #### 7.3.2 배치 처리 ```python # 동일 시간대 다수 요청 발생 시 배치 처리 async def batch_recommend(store_list, batch_size=5): """ 5개씩 묶어서 Claude API 호출 - 단일 프롬프트에 여러 매장 정보 포함 - 응답을 파싱하여 개별 결과로 분리 """ batches = [store_list[i:i+batch_size] for i in range(0, len(store_list), batch_size)] results = [] for batch in batches: combined_prompt = "\n\n".join([f"[매장 {i+1}]\n{format_store(s)}" for i, s in enumerate(batch)]) response = await claude_api.recommend_batch(combined_prompt) results.extend(parse_batch_response(response, len(batch))) return results ``` --- ## 8. 모니터링 및 개선 ### 8.1 성능 지표 #### 8.1.1 추천 정확도 (Recommendation Accuracy) ```python # 사용자 행동 추적 metrics = { "ctr": "추천 경품 클릭률", # 추천 3개 중 클릭한 비율 "selection_rate": "추천 선택률", # 추천 경품을 실제 이벤트에 사용한 비율 "satisfaction_score": "만족도", # 이벤트 종료 후 설문 (1~5점) "roi_accuracy": "ROI 예측 정확도" # 예측 ROI vs 실제 ROI 오차율 } # 목표 KPI targets = { "ctr": 0.6, # 60% 이상 "selection_rate": 0.4, # 40% 이상 "satisfaction_score": 4.0, # 5점 만점 중 4점 이상 "roi_accuracy": 0.8 # 오차 20% 이내 } ``` #### 8.1.2 시스템 성능 ```python # Prometheus + Grafana 모니터링 system_metrics = { "claude_api_latency": "Claude API 응답 시간 (p50, p95, p99)", "vector_search_latency": "벡터 검색 응답 시간", "cache_hit_rate": "캐시 히트율 (L1, L2 별도)", "fallback_rate": "Fallback 발생률", "daily_api_cost": "일일 Claude API 비용" } # 알림 임계값 alerts = { "claude_api_latency_p95": 3000, # 3초 초과 시 알림 "cache_hit_rate": 0.5, # 50% 미만 시 알림 "fallback_rate": 0.1, # 10% 초과 시 알림 "daily_api_cost": 100 # $100 초과 시 알림 } ``` ### 8.2 피드백 루프 #### 8.2.1 사용자 피드백 수집 ```python class FeedbackCollector: def collect(self, event_id, feedback_type): """ feedback_type: - "thumbs_up": 추천 만족 - "thumbs_down": 추천 불만족 - "custom_prize": 사용자가 직접 경품 변경 - "roi_feedback": 이벤트 종료 후 실제 ROI 입력 """ feedback = { "event_id": event_id, "type": feedback_type, "timestamp": datetime.now(), "store_profile": get_store_profile(event_id), "recommended_prizes": get_recommendations(event_id), "selected_prize": get_selected_prize(event_id), "actual_performance": get_performance(event_id) if feedback_type == "roi_feedback" else None } # Kafka로 전송 → 데이터 파이프라인에서 벡터 DB 갱신 kafka_producer.send("feedback.prize_recommendation", feedback) ``` #### 8.2.2 Few-Shot 사례 자동 갱신 ```python def update_few_shot_examples(): """ 월 1회 실행: 최근 3개월 성과 데이터 기반 Few-Shot 사례 갱신 """ top_cases = db.execute(""" SELECT store_profile, prize, performance FROM event_performance WHERE created_at >= NOW() - INTERVAL 3 MONTH AND roi >= 2.5 AND customer_satisfaction >= 4.0 ORDER BY roi DESC, conversion_rate DESC LIMIT 5 """) # Few-Shot 템플릿 재생성 new_examples = format_few_shot_examples(top_cases) # S3 또는 DB에 버전 관리 save_prompt_template("few_shot_v2.json", new_examples) # Claude API 호출 시 새 템플릿 적용 update_system_prompt(new_examples) ``` --- ## 9. 구현 우선순위 ### Phase 1: MVP (2주) - [x] 매장 데이터 수집 API 연동 (공공데이터포털) - [x] 경품 카탈로그 수동 구축 (100개 경품) - [x] 수동 Feature Engineering 벡터화 - [x] Redis Vector Search 기본 구현 - [x] Claude API 기본 프롬프트 (Few-Shot 3개) - [x] 룰 기반 Fallback ### Phase 2: 최적화 (1주) - [ ] Prompt Caching 적용 - [ ] L1/L2 캐싱 전략 구현 - [ ] 외부 트렌드 데이터 수집 자동화 (네이버 트렌드) - [ ] 성능 모니터링 대시보드 (Grafana) ### Phase 3: 고도화 (2주) - [ ] 사용자 피드백 수집 시스템 - [ ] Few-Shot 자동 갱신 파이프라인 - [ ] Sentence Transformer 임베딩 전환 (Option 2) - [ ] A/B 테스트 프레임워크 - [ ] SNS 트렌드 크롤링 --- ## 10. 참조 자료 ### 10.1 관련 문서 - 유저스토리: `design/userstory.md` - UI/UX 프로토타입: `design/uiux/prototype/05-AI경품추천.html` - API 설계서: `design/backend/api/prize-recommendation-api.yaml` ### 10.2 외부 리소스 - Claude API 문서: https://docs.anthropic.com/claude/docs/intro-to-claude - Redis Vector Search: https://redis.io/docs/stack/search/reference/vectors/ - 공공데이터포털: https://www.data.go.kr/ - 네이버 쇼핑 트렌드: https://datalab.naver.com/shoppingInsight/sCategory.naver --- ## 부록: 코드 예시 ### A. 전체 추천 API 엔드포인트 ```python from fastapi import APIRouter, HTTPException from pydantic import BaseModel import asyncio router = APIRouter() class StoreProfile(BaseModel): store_id: str business_number: str category: dict location: dict budget: dict target_customer: dict event_purpose: str class RecommendationRequest(BaseModel): store_profile: StoreProfile @router.post("/api/v1/prizes/recommend") async def recommend_prizes(request: RecommendationRequest): """ 경품 추천 메인 엔드포인트 Flow: 1. 캐시 조회 (L2: 개인화) 2. 벡터 검색 (유사 성공 사례 Top 3) 3. Claude API 호출 (Streaming) 4. 응답 검증 및 후처리 5. 캐시 저장 """ try: # 1. 캐시 조회 cache_key = generate_cache_key(request.store_profile) cached_result = await redis_cache.get(cache_key) if cached_result: logger.info(f"Cache hit: {cache_key}") return cached_result # 2. 매장 데이터 정제 및 벡터화 store_vector = await vectorize_store(request.store_profile) # 3. 벡터 검색 (유사 사례) similar_cases = await vector_search.find_similar(store_vector, top_k=3) # 4. Claude API 호출 prompt = build_prompt( system=SYSTEM_PROMPT, few_shot=FEW_SHOT_EXAMPLES, store_profile=request.store_profile, similar_cases=similar_cases ) claude_response = await claude_client.recommend(prompt, stream=True) # 5. 응답 파싱 및 검증 recommendations = parse_and_validate(claude_response) # 6. 예산 필터링 filtered = filter_budget(recommendations, request.store_profile.budget.max) # 7. 캐시 저장 (TTL 1시간) await redis_cache.setex(cache_key, 3600, filtered) return { "status": "success", "data": filtered, "metadata": { "cache_hit": False, "vector_search_count": len(similar_cases), "model": "claude-3-5-sonnet-20241022" } } except ClaudeAPIError as e: logger.error(f"Claude API failed: {e}") # Fallback fallback_result = fallback_engine.recommend(request.store_profile) return { "status": "fallback", "data": fallback_result, "metadata": {"error": str(e)} } except Exception as e: logger.exception("Unexpected error in prize recommendation") raise HTTPException(status_code=500, detail=str(e)) ``` ### B. 벡터 검색 구현 ```python import redis from redis.commands.search.query import Query import numpy as np import pickle class VectorSearchEngine: def __init__(self, redis_client): self.client = redis_client self.index_name = "idx:prize_cases" async def find_similar(self, query_vector: np.ndarray, top_k: int = 3): """ 코사인 유사도 기반 벡터 검색 Args: query_vector: 272차원 매장 임베딩 top_k: 반환할 유사 사례 개수 Returns: List[dict]: 유사 사례 데이터 """ # 벡터를 bytes로 직렬화 query_bytes = pickle.dumps(query_vector.astype(np.float32)) # Redis Vector Search 쿼리 query = ( Query(f"*=>[KNN {top_k} @case_vector $vec AS score]") .sort_by("score") .return_fields("store_category", "prize_name", "roi", "conversion_rate", "case_vector", "score") .paging(0, top_k) .dialect(2) ) results = self.client.ft(self.index_name).search( query, query_params={"vec": query_bytes} ) # 결과 파싱 similar_cases = [] for doc in results.docs: similar_cases.append({ "store_category": doc.store_category, "prize_name": doc.prize_name, "roi": float(doc.roi), "conversion_rate": float(doc.conversion_rate), "similarity_score": 1 - float(doc.score) # 코사인 거리 → 유사도 }) return similar_cases ``` ### C. Claude API 클라이언트 ```python import anthropic import json class ClaudeRecommendationClient: def __init__(self, api_key: str): self.client = anthropic.Anthropic(api_key=api_key) self.model = "claude-3-5-sonnet-20241022" async def recommend(self, prompt: dict, stream: bool = True): """ Claude API 호출 (Streaming) Args: prompt: {system, messages} 구조 stream: Streaming 여부 Returns: dict: 파싱된 JSON 응답 """ try: response = self.client.messages.create( model=self.model, max_tokens=4096, temperature=0.7, system=prompt['system'], messages=prompt['messages'], stream=stream ) if stream: # Streaming 응답 수집 full_text = "" async for chunk in response: if chunk.type == "content_block_delta": full_text += chunk.delta.text # JSON 파싱 return json.loads(full_text) else: return json.loads(response.content[0].text) except anthropic.APIError as e: raise ClaudeAPIError(f"Claude API failed: {e.status_code} - {e.message}") except json.JSONDecodeError as e: raise ClaudeAPIError(f"Invalid JSON response: {e}") ```