Update searchAsyncSingle.py

searchAsyncSingle.py

@@ -355,11 +355,11 @@ async def rebuild_faiss_index():
     def _build_ivf_index():
         dimension = item_vectors.shape[1]
         # IVF 클러스터 수 - 데이터 크기에 따라 조정 (√n 규칙 사용)
-        nlist = int(np.sqrt(total_items) * 1.5)  # 계수를 4에서 1.5로 감소
-        nlist = max(16, min(nlist, 256))  # 최대값을 1024에서 256으로 감소
+        nlist = int(np.sqrt(total_items) * 4)  # 클러스터 수 증가
+        nlist = max(32, min(nlist, 1024))  # 최소 32, 최대 1024개 제한
         
         # 양자화 파라미터 - 차원 수에 맞게 조정
-        M = min(32, dimension // 4)  # 서브벡터 수
+        M = min(64, dimension // 2)  # 서브벡터 수
         nbits = 8  # 비트 수
         
         # 고속 IVF 인덱스 생성
@@ -378,7 +378,7 @@ async def rebuild_faiss_index():
         
         # 검색 품질 향상을 위한 설정
         # nprobe = 몇 개의 클러스터를 검색할지 (높을수록 정확도 ↑, 속도 ↓)
-        index.nprobe = min(8, max(4, nlist // 16))  # 더 적은 클러스터 탐색 (nlist의 6.25%)
+        index.nprobe = min(32, nlist // 4)  # 클러스터의 25% 검색
         
         logger.info(f"✅ IVF 인덱스 구축 완료: clusters={nlist}, nprobe={index.nprobe}")
         return index
@@ -410,21 +410,16 @@ async def check_faiss_index():
         raise RuntimeError("FAISS 인덱스 초기화에 실패했습니다.")
 
 # ✅ 최적화된 키워드 추출 함수
-async def extract_keywords(query: str, top_n: int = 1):  # top_n을 2에서 1로 감소
+async def extract_keywords(query: str, top_n: int = 2):  # top_n 감소
     """KeyBERT 최적화 키워드 추출 (성능 중심)"""
     # 매우 짧은 쿼리는 그대로 반환 (처리 비용 절감)
     if len(query) <= 3:
         return [query]
     
-    # 단어가 2개 이하면 키워드 추출 과정 생략
-    words = query.split()
-    if len(words) <= 2:
-        return [words[0]] if words else [query]
-    
     loop = asyncio.get_event_loop()
     
     def _optimized_extract():
-        # 성능 중심 설정 (KeyBERT 파라미터 최적화)
+        # 성능 중심 설정
         return kw_model.extract_keywords(
             query, 
             keyphrase_ngram_range=(1, 1),  # 단일 단어만 추출
@@ -437,22 +432,20 @@ async def extract_keywords(query: str, top_n: int = 1):  # top_n을 2에서 1로
     try:
         keywords = await loop.run_in_executor(thread_pool, _optimized_extract)
         # 가중치가 너무 낮은 키워드 제외
-        filtered = [(k, s) for k, s in keywords if s > 0.3]  # 임계값 상향 (0.2 → 0.3)
+        filtered = [(k, s) for k, s in keywords if s > 0.2]
         return [k[0] for k in filtered]
     except Exception as e:
         logger.error(f"❌ 키워드 추출 오류: {str(e)}")
-        # 단어 분리로 폴백 - 첫 단어만 사용
-        return [words[0]] if words else [query]
-
+        # 단어 분리로 폴백
+        return query.split()[:2]
 
 
 # ✅ 최적화된 키워드 확장 함수
-async def expand_keywords_with_word2vec(keywords: list, max_new=1):  # max_new를 2에서 1로 감소
+async def expand_keywords_with_word2vec(keywords: list, max_new=2):  # max_new 감소
     """Word2Vec 키워드 확장 최적화"""
     global word2vec_model
     
-    # 키워드가 없거나, 모델이 없거나, 빈 문자열이면 빠르게 리턴
-    if not word2vec_model or not keywords or not keywords[0]:
+    if word2vec_model is None or not keywords:
         return keywords
     
     # 결과 저장을 위한 집합
@@ -461,28 +454,27 @@ async def expand_keywords_with_word2vec(keywords: list, max_new=1):  # max_new
     loop = asyncio.get_event_loop()
     
     def _expand_keywords():
-        # 첫 번째 키워드만 확장 (계산 비용 절감)
-        keyword = keywords[0]
-        
-        # 단일 단어인 경우
-        if keyword in word2vec_model:
-            # 유사도가 높은 단어만 선택 (임계값 적용)
-            similar_words = word2vec_model.most_similar(keyword, topn=max_new)
-            for word, score in similar_words:
-                if score > 0.8:  # 높은 유사도 임계값 적용 (0.7 → 0.8)
-                    expanded.add(word)
-        # 복합어 처리 (첫 단어만)
-        elif len(keyword.split()) > 1:
-            word = keyword.split()[0]
-            if word in word2vec_model and len(word) > 1:
-                similar = word2vec_model.most_similar(word, topn=1)
-                if similar and similar[0][1] > 0.85:  # 임계값 더 높임 (0.8 → 0.85)
-                    expanded.add(similar[0][0])
-        
-        # 결과 변환 - 최대 3개로 제한 (이전 5개)
+        for keyword in keywords:
+            # 단일 단어인 경우
+            if keyword in word2vec_model:
+                # 유사도가 높은 단어만 선택 (임계값 적용)
+                similar_words = word2vec_model.most_similar(keyword, topn=max_new)
+                for word, score in similar_words:
+                    if score > 0.7:  # 높은 유사도 임계값 적용
+                        expanded.add(word)
+            # 복합어 처리 (첫 단어만)
+            elif len(keyword.split()) > 1:
+                word = keyword.split()[0]
+                if word in word2vec_model and len(word) > 1:
+                    similar = word2vec_model.most_similar(word, topn=1)
+                    if similar and similar[0][1] > 0.8:  # 높은 임계값
+                        expanded.add(similar[0][0])
+        
+        # 결과 변환
         result = list(expanded)
-        if len(result) > 3:
-            return keywords + result[len(keywords):3]
+        # 키워드가 너무 많으면 제한
+        if len(result) > 5:
+            return keywords + result[len(keywords):5]
         return result
     
     try:
@@ -494,9 +486,8 @@ async def expand_keywords_with_word2vec(keywords: list, max_new=1):  # max_new
         return keywords  # 오류 시 원본 키워드 반환
 
 
-
 # ✅ 최적화된 search_faiss_with_keywords 함수
-async def search_faiss_with_keywords(query: str, top_k: int = 3, keywords=None, expanded_keywords=None):  # top_k를 5에서 3으로 감소
+async def search_faiss_with_keywords(query: str, top_k: int = 5, keywords=None, expanded_keywords=None):
     """고속 키워드 기반 FAISS 검색 수행"""
     global faiss_index, indexed_items
     
@@ -520,11 +511,8 @@ async def search_faiss_with_keywords(query: str, top_k: int = 3, keywords=None,
     # 2. 벡터 인코딩 최적화 - 쿼리와 키워드 한 번에 처리
     search_texts = [query] + expanded_keywords
     
-    # 짧은 텍스트는 벡터화 배치 크기 2배 증가 (성능 향상)
-    batch_size = 2048 if len(search_texts) < 10 else 1024
-    
     # 벡터 인코딩 - 최적화된 함수 사용
-    all_vectors = await encode_texts_parallel(search_texts, batch_size=batch_size)
+    all_vectors = await encode_texts_parallel(search_texts)
     
     # 벡터 정규화 - 최적화된 방식
     def normalize_batch(vectors):
@@ -544,47 +532,30 @@ async def search_faiss_with_keywords(query: str, top_k: int = 3, keywords=None,
     else:
         return []  # 벡터화 실패 시 빈 결과 반환
     
-    # 3. FAISS 검색 최적화 - 검색 파라미터 조정
+    # 3. FAISS 검색 최적화 - 일괄 배치 처리
     def _optimized_batch_search():
         all_results = {}
         
-        # 쿼리 벡터 검색 (가중치 유지)
+        # 쿼리 벡터 검색 (가중치 3배로 증가)
         if query_vector.shape[0] > 0:
-            # IVF 인덱스의 nprobe 감소 (더 적은 클러스터 검색)
-            original_nprobe = faiss_index.nprobe
-            faiss_index.nprobe = min(8, original_nprobe)  # nprobe 감소
-            
-            # top_k의 1.5배만 검색 (이전 2배)
-            search_k = int(top_k * 1.5)
-            distances, indices = faiss_index.search(query_vector, search_k)
-            
-            # 원래 nprobe로 복원
-            faiss_index.nprobe = original_nprobe
-            
-            # 쿼리 결과 가중치 적용
+            distances, indices = faiss_index.search(query_vector, top_k * 2)
+            # 쿼리 결과 가중치 적용 (중요도 상향)
             for idx, dist in zip(indices[0], distances[0]):
                 if idx < len(indexed_items):
-                    all_results[idx] = dist * 3.0  # 가중치 3.0 유지
+                    all_results[idx] = dist * 3.0  # 가중치 3.0
         
-        # 키워드 벡터 배치 검색 - 키워드 벡터가 적으면 생략 (1개 이하)
-        if keyword_vectors.shape[0] > 1:
-            # IVF 인덱스의 nprobe 감소 (더 적은 클러스터 검색)
-            original_nprobe = faiss_index.nprobe
-            faiss_index.nprobe = max(2, original_nprobe // 4)  # 더 낮은 nprobe
-            
-            # top_k만큼만 검색 (이전과 동일)
+        # 키워드 벡터 배치 검색
+        if keyword_vectors.shape[0] > 0:
+            # 배치 검색 한 번에 처리
             k_distances, k_indices = faiss_index.search(keyword_vectors, top_k)
             
-            # 원래 nprobe로 복원
-            faiss_index.nprobe = original_nprobe
-            
             # 키워드별 가중치 적용 및 결과 병합
             for i in range(keyword_vectors.shape[0]):
                 for j, (idx, dist) in enumerate(zip(k_indices[i], k_distances[i])):
                     if idx < len(indexed_items):
                         # 순위에 따라 가중치 차등 적용 (상위 결과 우대)
-                        rank_weight = 1.0 / (1 + j * 0.3)  # 순위별 가중치 감소 더 빠르게 (0.2 → 0.3)
-                        weight = 0.5 * rank_weight  # 가중치 감소 (0.6 → 0.5)
+                        rank_weight = 1.0 / (1 + j * 0.2)  # 순위별 가중치 감소
+                        weight = 0.6 * rank_weight  # 기본 가중치 0.6
                         
                         # 기존 점수에 추가
                         all_results[idx] = all_results.get(idx, 0) + dist * weight
@@ -598,50 +569,44 @@ async def search_faiss_with_keywords(query: str, top_k: int = 3, keywords=None,
     def _process_results():
         # 임계값 필터링 및 정렬
         filtered_items = [(idx, score) for idx, score in result_scores.items() 
-                        if score >= 0.4]  # 최소 점수 필터링 상향 (0.3 → 0.4)
+                        if score >= 0.3]  # 최소 점수 필터링
         
         # 점수 기준 내림차순 정렬
         sorted_items = sorted(filtered_items, key=lambda x: x[1], reverse=True)
         
-        # 최종 결과 변환 - 상위 N개만
+        # 최종 결과 변환
         recommendations = []
-        top_items = sorted_items[:top_k]  # top_k개만 처리
-        
-        # 결과가 충분하면 상품 정보 조회
-        if top_items:
-            item_indices = [idx for idx, _ in top_items]
-            item_names = [indexed_items[idx] for idx in item_indices if idx < len(indexed_items)]
-            
-            # 메모리 내 조회 최적화 - 한 번에 조회
-            items_df = active_sale_items[active_sale_items["ITEMNAME"].isin(item_names)]
-            
-            # 결과 매핑
-            for idx, score in top_items:
-                if idx < len(indexed_items):
-                    item_name = indexed_items[idx]
-                    mask = items_df["ITEMNAME"] == item_name
-                    if mask.any():
-                        item_seq = items_df.loc[mask, "ITEMSEQ"].values[0]
-                        recommendations.append({
-                            "ITEMSEQ": item_seq,
-                            "ITEMNAME": item_name,
-                            "score": float(score)
-                        })
-                        
+        for idx, score in sorted_items[:top_k]:  # top_k개만 처리
+            item_name = indexed_items[idx]
+            try:
+                # 메모리 내 조회 최적화
+                mask = active_sale_items["ITEMNAME"] == item_name
+                if mask.any():
+                    item_seq = active_sale_items.loc[mask, "ITEMSEQ"].values[0]
+                    recommendations.append({
+                        "ITEMSEQ": item_seq,
+                        "ITEMNAME": item_name,
+                        "score": float(score)
+                    })
+            except (IndexError, KeyError):
+                continue
+                
         return recommendations
     
     # 결과 처리 실행
     recommendations = await loop.run_in_executor(thread_pool, _process_results)
     
-    # 5. 직접 매칭 처리 - 결과가 없거나 매우 적을 때만 수행
-    if len(recommendations) == 0:  # 결과가 없을 때만 직접 매칭 수행
-        direct_matches = await find_direct_matches(query, min(2, top_k))  # 최대 2개만 가져오기
+    # 5. 직접 매칭 추가 최적화 (필요한 경우에만)
+    if len(recommendations) < top_k:
+        direct_matches = await find_direct_matches(query, 
+                                                  top_k - len(recommendations),
+                                                  [r["ITEMNAME"] for r in recommendations])
         if direct_matches:
             recommendations.extend(direct_matches)
     
-    # 처리 시간 로깅 필요한 경우만
+    # 처리 시간이 1초 이상인 경우에만 로깅
     elapsed = time.time() - start_time
-    if elapsed > 0.5:  # 임계값 하향 (1.0 → 0.5)
+    if elapsed > 1.0:
         logger.info(f"🔍 검색 완료 | 소요시간: {elapsed:.2f}초 | 결과: {len(recommendations)}개")
     
     return recommendations[:top_k]
@@ -708,18 +673,16 @@ async def recommend(request: RecommendRequest, background_tasks: BackgroundTasks
         if not search_query:
             raise HTTPException(status_code=400, detail="검색어를 입력해주세요")
             
-        top_k = min(max(1, request.top_k), 10)  # 1~10 범위로 제한 (이전 20)
-        
-        # 병렬 프로세싱을 위한 동시 실행 - 단순 쿼리는 키워드 처리 간소화
-        if len(search_query) <= 5:  # 짧은 쿼리는 모든 키워드 처리 건너뛰기 (길이 5 이하)
-            keywords = [search_query.split()[0] if search_query.split() else search_query]
-            expanded_keywords = keywords  # 확장 없이 동일하게 사용
-        else:
-            keywords = await extract_keywords(search_query)
-            if request.use_expansion:
-                expanded_keywords = await expand_keywords_with_word2vec(keywords, max_new=1)
-            else:
-                expanded_keywords = keywords
+        top_k = min(max(1, request.top_k), 20)  # 1~20 범위로 제한
+        
+        # 병렬 프로세싱을 위한 동시 실행
+        keywords, expanded_keywords = await asyncio.gather(
+            extract_keywords(search_query),
+            expand_keywords_with_word2vec(
+                [search_query.split()[0]] if search_query.split() else [search_query],
+                max_new=2
+            ) if request.use_expansion else None
+        )
         
         # 검색 실행 - 병렬 처리된 키워드 활용
         recommendations = await search_faiss_with_keywords(
@@ -737,9 +700,9 @@ async def recommend(request: RecommendRequest, background_tasks: BackgroundTasks
             "expanded_keywords": expanded_keywords if expanded_keywords and len(expanded_keywords) > 0 else None
         }
         
-        # 응답 시간 측정 (0.5초 이상만 로깅)
+        # 응답 시간 측정 (1초 이상만 로깅)
         elapsed = time.time() - start_time
-        if elapsed > 0.5:
+        if elapsed > 1.0:
             logger.info(f"⏱️ API 응답 시간: {elapsed:.2f}초 | 쿼리: '{search_query}'")
         
         return result
@@ -748,6 +711,47 @@ async def recommend(request: RecommendRequest, background_tasks: BackgroundTasks
         logger.error(f"❌ 추천 처리 오류: {str(e)}")
         raise HTTPException(status_code=500, detail=f"추천 처리 중 오류가 발생했습니다")
 
+# 인덱스 상태 확인 함수 (백그라운드 태스크용)
+async def check_index_health():
+    """인덱스 상태를 주기적으로 확인하는 백그라운드 태스크"""
+    try:
+        # 인덱스 사용 상태 확인
+        if faiss_index is None:
+            logger.warning("⚠️ 백그라운드 체크: FAISS 인덱스가 로드되지 않았습니다.")
+            await check_faiss_index()
+        
+        # 추가적인 상태 확인 로직을 여기에 구현할 수 있음
+        logger.debug("✅ 인덱스 상태 확인 완료")
+    except Exception as e:
+        logger.error(f"❌ 백그라운드 인덱스 체크 중 오류: {str(e)}")
+
+# ✅ 유사 단어 검색 API
+@app.post("/api/similar_words")
+async def similar_words(word: str, top_k: int = 10):
+    """Word2Vec 모델을 사용한 유사 단어 검색 API (비동기 지원)"""
+    try:
+        if word2vec_model is None:
+            return {"error": "Word2Vec 모델이 로드되지 않았습니다."}
+        
+        loop = asyncio.get_event_loop()
+        
+        def _get_similar():
+            if word not in word2vec_model:
+                return []
+            
+            similar = word2vec_model.most_similar(word, topn=top_k)
+            return [{"word": w, "similarity": float(s)} for w, s in similar]
+        
+        result = await loop.run_in_executor(thread_pool, _get_similar)
+        
+        if not result:
+            return {"word": word, "similar_words": [], "message": "단어가 모델에 없습니다."}
+        
+        return {"word": word, "similar_words": result}
+    except Exception as e:
+        logger.error(f"❌ 유사 단어 검색 중 오류: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"유사 단어 검색 오류: {str(e)}")
+
 # ✅ FAISS 인덱스 갱신 API (명시적으로 요청할 때만 실행)
 @app.post("/api/update_index")
 async def update_index(background_tasks: BackgroundTasks):
@@ -772,6 +776,108 @@ async def rebuild_and_log_index():
     except Exception as e:
         logger.error(f"❌ 백그라운드 인덱스 재구축 중 오류: {str(e)}")
 
+# ✅ 인덱스 디버깅 API
+@app.get("/api/debug_index")
+async def debug_index(query: str, top_k: int = 20):
+    """인덱스 디버깅을 위한 API (비동기 지원)"""
+    try:
+        await check_faiss_index()
+        
+        loop = asyncio.get_event_loop()
+        
+        # 원본 벡터 생성 (비동기)
+        def _get_vector():
+            vector = embedding_model.encode(query, convert_to_numpy=True).astype("float32")
+            norm = np.linalg.norm(vector)
+            normalized_vector = vector / norm
+            return normalized_vector, norm
+        
+        normalized_vector, norm = await loop.run_in_executor(thread_pool, _get_vector)
+        
+        # 원본 쿼리로 검색 (비동기)
+        def _search():
+            return faiss_index.search(np.array([normalized_vector]), top_k)
+        
+        distances, indices = await loop.run_in_executor(thread_pool, _search)
+        
+        # 결과 매핑
+        results = []
+        for i, (idx, dist) in enumerate(zip(indices[0], distances[0])):
+            if idx < len(indexed_items):
+                item_name = indexed_items[idx]
+                results.append({
+                    "rank": i + 1,
+                    "index": int(idx),
+                    "item_name": item_name,
+                    "distance/score": float(dist)
+                })
+        
+        # 데이터셋에 해당 단어가 있는지 확인 (비동기)
+        def _find_matches():
+            contains = [item for item in indexed_items if query.lower() in item.lower()][:5]
+            exact = [item for item in indexed_items if query.lower() == item.lower()]
+            return contains, exact
+        
+        contains_query, exact_matches = await loop.run_in_executor(thread_pool, _find_matches)
+        
+        return {
+            "query": query,
+            "vector_norm": float(norm),
+            "contains_query": contains_query,
+            "exact_matches": exact_matches,
+            "results": results
+        }
+    except Exception as e:
+        logger.error(f"❌ 인덱스 디버깅 중 오류: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"인덱스 디버깅 오류: {str(e)}")
+
+# ✅ 문자열 포함 검색 API
+@app.get("/api/text_search")
+async def text_search(query: str, top_k: int = 10):
+    """단순 텍스트 포함 검색 API (비동기 지원)"""
+    try:
+        loop = asyncio.get_event_loop()
+        
+        # 비동기 검색 함수
+        def _text_search():
+            # 단순 텍스트 포함 검색
+            matched_items = []
+            for idx, item_name in enumerate(indexed_items):
+                if query.lower() in item_name.lower():
+                    try:
+                        item_seq = active_sale_items.loc[active_sale_items["ITEMNAME"] == item_name, "ITEMSEQ"].values[0]
+                        matched_items.append({"ITEMSEQ": item_seq, "ITEMNAME": item_name, "match_type": "contains"})
+                    except (IndexError, KeyError):
+                        continue
+
+            # 정확히 일치하는 항목을 앞으로
+            exact_matches = []
+            partial_matches = []
+            
+            for item in matched_items:
+                if query.lower() == item["ITEMNAME"].lower():
+                    item["match_type"] = "exact"
+                    exact_matches.append(item)
+                else:
+                    partial_matches.append(item)
+            
+            # 결합 및 제한
+            return exact_matches + partial_matches
+        
+        # 비동기적으로 검색 실행
+        results = await loop.run_in_executor(thread_pool, _text_search)
+        
+        logger.info(f"🔍 텍스트 검색 결과: {len(results)}개 찾음, 쿼리: '{query}'")
+        
+        return {
+            "query": query,
+            "total_matches": len(results),
+            "results": results[:top_k]
+        }
+    except Exception as e:
+        logger.error(f"❌ 텍스트 검색 중 오류: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"텍스트 검색 오류: {str(e)}")
+
 # ✅ FastAPI 실행
 if __name__ == "__main__":
     # 서버 시작 시 저장된 인덱스 로드 시도