Update app.py

app.py

@@ -4,30 +4,16 @@
 # HF Spaces Secrets 설정:
 #   NAVER_CLIENT_ID     = 발급받은 Client ID
 #   NAVER_CLIENT_SECRET = 발급받은 Client Secret
-#
-# (선택) 임베딩 모델 변경:
-#   EMBEDDING_MODEL = sentence-transformers 호환 모델명
-#     예) jhgan/ko-sroberta-multitask (기본)
-#
-# 로컬 실행 시(선택):
-#   export NAVER_CLIENT_ID="..."
-#   export NAVER_CLIENT_SECRET="..."
-#   export EMBEDDING_MODEL="jhgan/ko-sroberta-multitask"
 
 import os
 import html
 import re
 from datetime import datetime
-from typing import Dict, Any, List, Tuple, Optional
-from functools import lru_cache
+from typing import Dict, Any, List, Tuple
 
 import requests
 import gradio as gr
 
-# 임베딩
-import numpy as np
-from sentence_transformers import SentenceTransformer
-
 
 NAVER_NEWS_ENDPOINT = "https://openapi.naver.com/v1/search/news.json"
 
@@ -122,10 +108,9 @@ def render_results(data: Dict[str, Any], max_items: int = 10) -> str:
         origin = it.get("originallink", "")
         pub = _format_pubdate(it.get("pubDate", ""))
 
-        # ✅ ordered list 문법을 깨지 않도록 이스케이프 제거
         lines.append(f"{i}. **{title}**")
 
-        # ✅ 하위 항목은 4칸 들여쓰기(ordered list 하위 리스트 인식 안정화)
+        # 하위 항목 4칸 들여쓰기
         if pub:
             lines.append(f"    - 발행: {pub}")
         if origin:
@@ -141,299 +126,6 @@ def render_results(data: Dict[str, Any], max_items: int = 10) -> str:
     return "\n".join(lines).strip()
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# Sentence 입력 -> (규칙/통계) 후보 생성 -> (임베딩) 키워드 선별/확장 -> 쿼리 생성
-# ─────────────────────────────────────────────────────────────────────────────
-
-EMBEDDING_MODEL_NAME = _get_env("EMBEDDING_MODEL") or "jhgan/ko-sroberta-multitask"
-
-
-@lru_cache(maxsize=1)
-def _get_embedder() -> SentenceTransformer:
-    # HF Spaces에서 최초 로드 시간이 있을 수 있습니다.
-    return SentenceTransformer(EMBEDDING_MODEL_NAME)
-
-
-# 간단 불용어(필요 시 확장)
-STOPWORDS = {
-    "그리고", "또는", "및", "관련", "대한", "에서", "으로", "하는", "합니다", "해주세요",
-    "알려", "알려줘", "뉴스", "기사", "최근", "요즘", "이번", "오늘", "어제", "내일",
-    "정리", "분석", "동향", "현황", "이슈", "내용", "정보", "보기", "보고", "싶어",
-    "싶습니다", "합니다", "해줘", "해주세요", "어떻게", "가능", "가능한", "형태",
-}
-
-# 제외 의도 트리거(문장에 포함되면 제외어를 강화)
-NEGATION_TRIGGERS = [
-    "제외", "빼고", "빼줘", "말고", "아닌", "원치", "싫", "배제", "제거",
-]
-
-# 기본 제외 후보(문서/홍보성 잡음 방지 목적: 필요 시 조정)
-DEFAULT_EXCLUDE_CANDIDATES = [
-    "보도자료", "홍보", "PR", "프로모션", "광고", "협찬",
-]
-
-
-# 동의어/표기 확장(작게 시작해서 운영하면서 늘리는 것을 권장)
-SYNONYM_GROUPS = [
-    ["AI", "인공지능", "A.I."],
-    ["LLM", "거대언어모델", "대규모언어모델", "생성형AI", "생성형 AI"],
-    ["감사원", "감사원(BAI)", "Board of Audit and Inspection"],
-    ["방위사업청", "방사청", "DAPA"],
-]
-
-
-def _tokenize_korean_like(text: str) -> List[str]:
-    """
-    MVP용 토크나이저:
-    - 한글/영문/숫자 연속 토큰을 추출
-    - 지나치게 짧은 토큰(1자)은 제거
-    """
-    if not text:
-        return []
-    tokens = re.findall(r"[가-힣A-Za-z0-9]+", text)
-    tokens = [t.strip() for t in tokens if len(t.strip()) >= 2]
-    return tokens
-
-
-def _generate_ngrams(tokens: List[str], n: int) -> List[str]:
-    if n <= 1:
-        return tokens[:]
-    out = []
-    for i in range(len(tokens) - n + 1):
-        out.append(" ".join(tokens[i:i+n]))
-    return out
-
-
-def extract_candidates(sentence: str, max_candidates: int = 60) -> List[str]:
-    """
-    규칙/통계 기반 후보 생성:
-    - 토큰(2자 이상) + 2-gram을 후보로 생성
-    - 단순 빈도 기반 점수로 상위 후보만 반환
-    """
-    tokens = _tokenize_korean_like(sentence)
-    # 불용어 제거(토큰 단위)
-    tokens = [t for t in tokens if t not in STOPWORDS]
-
-    unigrams = tokens
-    bigrams = _generate_ngrams(tokens, 2)
-
-    # 통계(빈도) 기반 스코어링: bigram에 약간 가중치
-    freq: Dict[str, float] = {}
-
-    for t in unigrams:
-        freq[t] = freq.get(t, 0.0) + 1.0
-    for bg in bigrams:
-        freq[bg] = freq.get(bg, 0.0) + 1.5
-
-    # 너무 긴 후보는 제외(검색식 과도 복잡화 방지)
-    def _ok(c: str) -> bool:
-        if len(c) > 25:
-            return False
-        # 숫자만으로 된 후보는 제외
-        if re.fullmatch(r"\d+", c):
-            return False
-        return True
-
-    ranked = sorted(
-        [(c, s) for c, s in freq.items() if _ok(c)],
-        key=lambda x: x[1],
-        reverse=True,
-    )
-    return [c for c, _ in ranked[:max_candidates]]
-
-
-def _embed_texts(texts: List[str]) -> np.ndarray:
-    model = _get_embedder()
-    emb = model.encode(texts, normalize_embeddings=True, show_progress_bar=False)
-    return np.asarray(emb, dtype=np.float32)
-
-
-def select_keywords_by_embedding(sentence: str, candidates: List[str], top_n: int = 10) -> List[str]:
-    """
-    임베딩으로 후보 키워드 선별:
-    - 입력 문장과 후보(짧은 구/단어)를 임베딩 유사도로 점수화
-    """
-    if not sentence.strip() or not candidates:
-        return []
-
-    # 후보가 너무 많으면 속도 저하 → 상한
-    candidates = candidates[:80]
-
-    sent_emb = _embed_texts([sentence])[0]
-    cand_emb = _embed_texts(candidates)
-    sims = cand_emb @ sent_emb  # normalize_embeddings=True 이므로 내적=cosine
-
-    idx = np.argsort(sims)[::-1][:max(1, top_n)]
-    selected = [candidates[i] for i in idx]
-
-    # 중복/포함관계 정리(짧은 토큰이 긴 후보에 포함되면 긴 후보 우선)
-    dedup: List[str] = []
-    for s in selected:
-        if any(s != x and s in x for x in selected):
-            continue
-        if s not in dedup:
-            dedup.append(s)
-    return dedup[:top_n]
-
-
-def detect_excludes(sentence: str) -> List[str]:
-    """
-    제외어 추출:
-    - 문장에 제외 의도 트리거가 있으면 기본 제외 후보를 활성화
-    - 문장 내에서 "X 제외/빼고/말고" 패턴도 단순 추출
-    """
-    s = sentence.strip()
-    if not s:
-        return []
-
-    excludes: List[str] = []
-
-    # 1) 제외 의도 감지 시 기본 제외어 추가
-    if any(t in s for t in NEGATION_TRIGGERS):
-        excludes.extend(DEFAULT_EXCLUDE_CANDIDATES)
-
-    # 2) "OO 제외", "OO 빼고" 등의 단순 패턴 추출
-    #    너무 공격적으로 뽑으면 오탐이 늘어 MVP에서는 보수적으로(2자 이상 토큰)
-    for m in re.findall(r"([가-힣A-Za-z0-9]{2,})\s*(제외|빼고|말고|배제|제거)", s):
-        token = m[0].strip()
-        if token and token not in excludes:
-            excludes.append(token)
-
-    # 정리
-    excludes = [e for e in excludes if e not in STOPWORDS]
-    # 과도 확장 방지
-    return excludes[:8]
-
-
-def expand_synonyms(keywords: List[str]) -> List[List[str]]:
-    """
-    키워드가 동의어 그룹 항목과 '완전 일치'하지 않아도,
-    그룹 항목이 키워드(문구) 안에 포함되면 치환 확장을 만들어 OR 그룹 후보를 생성합니다.
-
-    예)
-      "AI 기본법" -> ["AI 기본법", "인공지능 기본법", "A.I. 기본법"]
-    """
-    groups: List[List[str]] = []
-
-    for k in keywords:
-        k_str = (k or "").strip()
-        if not k_str:
-            groups.append([k_str])
-            continue
-
-        expanded = [k_str]
-        matched = False
-
-        for g in SYNONYM_GROUPS:
-            for term in g:
-                # 부분 포함 매칭(대소문자 무시)
-                if term.lower() in k_str.lower():
-                    matched = True
-                    for alt in g:
-                        # 포함된 term 부분을 alt로 치환
-                        cand = re.sub(re.escape(term), alt, k_str, flags=re.IGNORECASE).strip()
-                        if cand and cand not in expanded:
-                            expanded.append(cand)
-                    break
-            if matched:
-                break  # 첫 매칭 그룹만 적용(확장 폭발 방지)
-
-        # 폭발 방지: 최대 3개까지만
-        groups.append(expanded[:3])
-
-    # 중복 그룹 병합(대소문자 무시)
-    merged: List[List[str]] = []
-    seen = set()
-    for g in groups:
-        key = tuple(sorted([x.lower() for x in g]))
-        if key in seen:
-            continue
-        seen.add(key)
-        merged.append(g)
-
-    return merged
-
-
-def _dedup_keywords_preserve_order(keywords: List[str]) -> List[str]:
-    """
-    검색어 정규화 + 의미 중복 제거:
-    - 양끝 공백 제거, 내부 연속 공백 1개로 축소
-    - 동일(정규화 기준) 중복 제거
-    - 짧은 토큰이 긴 토큰(정규화 기준)에 포함되면 제거
-    - 순서 유지
-    """
-    def norm(s: str) -> str:
-        s = (s or "").strip()
-        s = re.sub(r"\s+", " ", s)  # 연속 공백 정리
-        return s
-
-    # 1) 정규화 + 동일 중복 제거(순서 유지)
-    out: List[str] = []
-    seen = set()
-    normalized = [norm(k) for k in keywords if norm(k)]
-
-    for k in normalized:
-        if k in seen:
-            continue
-        seen.add(k)
-        out.append(k)
-
-    # 2) 포함관계 제거(짧은 토큰이 긴 토큰에 포함되면 제거)
-    final: List[str] = []
-    for k in out:
-        if any(k != x and k in x for x in out):
-            continue
-        final.append(k)
-
-    return final
-
-
-def build_queries(
-    sentence: str,
-    selected_keywords: List[str],
-    excludes: List[str],   # (호환 유지: 인자는 남겨둠)
-    max_queries: int = 6,  # (호환 유지: 인자는 남겨둠)
-) -> List[str]:
-    """
-    최적 전략:
-    - Q1: 기본 AND 쿼리 1개
-    - Q2: 동의어/표기 치환이 명확할 때만 1개 생성
-    - Q3 이상 생성하지 않음
-    - 제외(-)는 사용하지 않음
-    """
-    if not selected_keywords:
-        selected_keywords = extract_candidates(sentence, max_candidates=10)[:4]
-
-    selected_keywords = _dedup_keywords_preserve_order(selected_keywords)  # gk
-
-    # Q1: 기본 AND
-    q1 = " ".join(selected_keywords).strip()
-    queries: List[str] = [q1] if q1 else []
-
-    # Q2: 동의어/표기 치환이 '실제로 발생'한 경우에만 1개 생성
-    groups = expand_synonyms(selected_keywords)
-
-    # 어떤 키워드라도 확장(치환) 후보가 2개 이상 있으면 "명확"하다고 보고 Q2 생성 시도
-    has_clear_substitution = any(len(g) >= 2 for g in groups)
-
-    if has_clear_substitution:
-        # Q2는 각 그룹에서 "대체 후보"를 하나씩 골라 Q1과 다른 조합이 되게 만듦
-        combo = []
-        for g in groups:
-            # g[0]은 원문 유지, g[1]이 있으면 치환된 후보를 우선 사용
-            combo.append(g[1] if len(g) >= 2 else g[0])
-
-        combo = _dedup_keywords_preserve_order(combo)  # gk
-        q2 = " ".join(combo).strip()
-
-        # Q2가 Q1과 다르고, 비어있지 않으면 추가
-        if q2 and (not queries or q2 != queries[0]):
-            queries.append(q2)
-
-    # 최대 2개(Q1, Q2)만 반환
-    return queries[:2]
-
-
 def dedup_items(all_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """
     결과 중복 제거:
@@ -457,60 +149,36 @@ def dedup_items(all_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     return out
 
 
-def rerank_items_by_embedding(sentence: str, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """
-    입력 문장과 (title+description)의 임베딩 유사도로 재정렬
-    """
-    if not sentence.strip() or not items:
-        return items
-
-    texts = []
-    for it in items:
-        title = _strip_tags(it.get("title", ""))
-        desc = _strip_tags(it.get("description", ""))
-        texts.append((title + " " + desc).strip())
-
-    sent_emb = _embed_texts([sentence])[0]
-    doc_emb = _embed_texts(texts)
-    sims = doc_emb @ sent_emb
-
-    order = np.argsort(sims)[::-1]
-    reranked = [items[i] for i in order.tolist()]
-    return reranked
-
-
 def aggregate_search(
     sentence: str,
     display: int,
     sort: str,
-) -> Tuple[List[str], List[Dict[str, Any]]]:
+) -> Tuple[List[str], List[Dict[str, Any]], int]:
     """
-    문장 입력 -> (현재는) 사용자 입력 문장을 그대로 query로 사용하여 API 호출
-    반환: (생성된 쿼리 목록, 최종 아이템 목록)
+    사용자 입력 문장을 그대로 query로 사용하여 API 호출
+    반환: (사용된 쿼리 목록, 최종 아이템 목록, total)
     """
-
-    # ✅ 변경된 핵심: 사용자 입력 문장을 그대로 query로 사용
     queries = [sentence]
 
     all_items: List[Dict[str, Any]] = []
-    for q in queries:
-        data = naver_news_search(query=q, display=int(display), sort=sort, start=1)
-        all_items.extend(data.get("items", []))
+    total: int = 0
 
-    # 통합/중복 제거(단일 쿼리라도 유지)
-    merged = dedup_items(all_items)
+    # 단일 쿼리 호출
+    data = naver_news_search(query=sentence, display=int(display), sort=sort, start=1)
+    total = int(data.get("total", 0) or 0)
+    all_items.extend(data.get("items", []))
 
-    # 임베딩 재랭킹(기존 동작 유지)
-    reranked = rerank_items_by_embedding(sentence, merged)
+    # 중복 제거(단일 쿼리라도 유지)
+    merged = dedup_items(all_items)
 
     # 최종 개수 절단
-    final_items = reranked[:display]
-    return queries, final_items
+    final_items = merged[:display]
+    return queries, final_items, total
 
 
 def render_results_from_items(items: List[Dict[str, Any]]) -> str:
     """
-    통합/재랭킹된 items 리스트를 동일 스타일로 출력
+    items 리스트를 동일 스타일로 출력
     """
     lines: List[str] = []
     lines.append(f"- 최종 반환 개수: {len(items)}건")
@@ -523,10 +191,8 @@ def render_results_from_items(items: List[Dict[str, Any]]) -> str:
         origin = it.get("originallink", "")
         pub = _format_pubdate(it.get("pubDate", ""))
 
-        # ✅ ordered list 문법 유지
         lines.append(f"{i}. **{title}**")
 
-        # ✅ 하위 항목 4칸 들여쓰기
         if pub:
             lines.append(f"    - 발행: {pub}")
         if origin:
@@ -541,7 +207,6 @@ def render_results_from_items(items: List[Dict[str, Any]]) -> str:
     return "\n".join(lines).strip()
 
 
-
 def handle_search(
     user_query: str,
     chat_history: List[Dict[str, str]],
@@ -556,14 +221,23 @@ def handle_search(
     chat_history = chat_history + [{"role": "user", "content": q}]
 
     try:
-        queries, items = aggregate_search(sentence=q, display=int(display), sort=sort)
+        queries, items, total = aggregate_search(sentence=q, display=int(display), sort=sort)
+
+        # total이 0이거나 없을 때는 items 개수로 대체
+        total_to_show = total if total > 0 else len(items)
 
-        lines = []
+        lines: List[str] = []
+
+        # ✅ 요청 문구로 변경
+        lines.append(f"\"{q}\"에 대한 검색 결과는 {total_to_show}건 이며 내용은 다음과 같습니다.")
         lines.append("")
+
+        # (기존 이력 출력 유지)
         lines.append("API 호출에 사용된 검색어(query)는 다음과 같습니다:")
         for i, qq in enumerate(queries, start=1):
             lines.append(f"- Q{i}: `{qq}`")
         lines.append("")
+
         lines.append(render_results_from_items(items))
 
         assistant_text = "\n".join(lines).strip()