Update app.py

app.py

@@ -26,6 +26,23 @@ import PyPDF2
 ##############################################################################
 SERPHOUSE_API_KEY = "V38CNn4HXpLtynJQyOeoUensTEYoFy8PBUxKpDqAW1pawT1vfJ2BWtPQ98h6"
 
+##############################################################################
+# [새로 추가] 사용자 메시지로부터 간단히 키워드 추출하는 함수 예시
+# - 실제 환경에 맞게 stopwords, 형태소 분석 등 고도화 가능
+##############################################################################
+def extract_keywords(text: str, top_k: int = 5) -> str:
+    # 1) 소문자로
+    text = text.lower()
+    # 2) 알파벳/숫자/공백 제외 문자 제거
+    text = re.sub(r"[^a-z0-9\s]", "", text)
+    # 3) 공백단위 토큰
+    tokens = text.split()
+    # 4) 우선은 앞에서 몇 개 토큰만 사용 (top_k=5)
+    #    - 필요시 stopword 제거나 빈도수 계산 후 상위 k개 추출하도록 변경 가능
+    key_tokens = tokens[:top_k]
+    # 5) 공백으로 join
+    return " ".join(key_tokens)
+
 ##############################################################################
 # Simple function to call the SERPHouse Live endpoint
 # https://api.serphouse.com/serp/live
@@ -33,7 +50,7 @@ SERPHOUSE_API_KEY = "V38CNn4HXpLtynJQyOeoUensTEYoFy8PBUxKpDqAW1pawT1vfJ2BWtPQ98h
 def do_web_search(query: str) -> str:
     """
     Calls SERPHouse live endpoint with the given query (q).
-    Returns a simple text summary or error message.
+    Returns top-20 results' titles as a bullet list, or an error message.
     """
     try:
         url = "https://api.serphouse.com/serp/live"
@@ -43,27 +60,26 @@ def do_web_search(query: str) -> str:
             "lang": "en",
             "device": "desktop",
             "serp_type": "web",
+            "num_result": "20",  # [새로 추가] 상위 20개 결과
             "api_token": SERPHOUSE_API_KEY,
         }
         resp = requests.get(url, params=params, timeout=30)
         resp.raise_for_status()  # Raise an exception for 4xx/5xx errors
         data = resp.json()
 
-        # For demonstration, let's extract top 3 organic results:
         results = data.get("results", {})
         organic = results.get("results", {}).get("organic", [])
         if not organic:
             return "No web search results found."
 
+        # 상위 20개 제목만 뽑아서 정리
         summary_lines = []
-        for item in organic[:3]:
-            rank = item.get("position", "-")
+        for idx, item in enumerate(organic[:20], start=1):
             title = item.get("title", "No Title")
-            link = item.get("link", "No Link")
-            snippet = item.get("snippet", "(No snippet)")
-            summary_lines.append(f"**Rank {rank}:** [{title}]({link})\n\n> {snippet}")
+            summary_lines.append(f"{idx}. {title}")
 
-        return "\n\n".join(summary_lines) if summary_lines else "No web search results found."
+        # 20개를 \n 으로 연결
+        return "\n".join(summary_lines)
     except Exception as e:
         logger.error(f"Web search failed: {e}")
         return f"Web search failed: {str(e)}"
@@ -87,15 +103,10 @@ MAX_NUM_IMAGES = int(os.getenv("MAX_NUM_IMAGES", "5"))
 # CSV, TXT, PDF 분석 함수
 ##################################################
 def analyze_csv_file(path: str) -> str:
-    """
-    CSV 파일을 전체 문자열로 변환. 너무 길 경우 일부만 표시.
-    """
     try:
         df = pd.read_csv(path)
-        # 데이터 프레임 크기 제한 (행/열 수가 많은 경우)
         if df.shape[0] > 50 or df.shape[1] > 10:
             df = df.iloc[:50, :10]
-            
         df_str = df.to_string()
         if len(df_str) > MAX_CONTENT_CHARS:
             df_str = df_str[:MAX_CONTENT_CHARS] + "\n...(truncated)..."
@@ -105,9 +116,6 @@ def analyze_csv_file(path: str) -> str:
 
 
 def analyze_txt_file(path: str) -> str:
-    """
-    TXT 파일 전문 읽기. 너무 길면 일부만 표시.
-    """
     try:
         with open(path, "r", encoding="utf-8") as f:
             text = f.read()
@@ -119,25 +127,19 @@ def analyze_txt_file(path: str) -> str:
 
 
 def pdf_to_markdown(pdf_path: str) -> str:
-    """
-    PDF → Markdown. 페이지별로 간단히 텍스트 추출.
-    """
     text_chunks = []
     try:
         with open(pdf_path, "rb") as f:
             reader = PyPDF2.PdfReader(f)
-            # 최대 5페이지만 처리
             max_pages = min(5, len(reader.pages))
             for page_num in range(max_pages):
                 page = reader.pages[page_num]
                 page_text = page.extract_text() or ""
                 page_text = page_text.strip()
                 if page_text:
-                    # 페이지별 텍스트도 제한
                     if len(page_text) > MAX_CONTENT_CHARS // max_pages:
                         page_text = page_text[:MAX_CONTENT_CHARS // max_pages] + "...(truncated)"
                     text_chunks.append(f"## Page {page_num+1}\n\n{page_text}\n")
-            
             if len(reader.pages) > max_pages:
                 text_chunks.append(f"\n...(Showing {max_pages} of {len(reader.pages)} pages)...")
     except Exception as e:
@@ -181,14 +183,6 @@ def count_files_in_history(history: list[dict]) -> tuple[int, int]:
 
 
 def validate_media_constraints(message: dict, history: list[dict]) -> bool:
-    """
-    - 비디오 1개 초과 불가
-    - 비디오와 이미지 혼합 불가
-    - 이미지 개수 MAX_NUM_IMAGES 초과 불가
-    - <image> 태그가 있으면 태그 수와 실제 이미지 수 일치
-    - CSV, TXT, PDF 등은 여기서 제한하지 않음
-    """
-    # 이미지와 비디오 파일만 필터링
     media_files = []
     for f in message["files"]:
         if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE) or f.endswith(".mp4"):
@@ -213,9 +207,7 @@ def validate_media_constraints(message: dict, history: list[dict]) -> bool:
         gr.Warning(f"You can upload up to {MAX_NUM_IMAGES} images.")
         return False
     
-    # 이미지 태그 검증 (실제 이미지 파일만 계산)
     if "<image>" in message["text"]:
-        # 이미지 파일만 필터링
         image_files = [f for f in message["files"] if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE)]
         image_tag_count = message["text"].count("<image>")
         if image_tag_count != len(image_files):
@@ -232,9 +224,7 @@ def downsample_video(video_path: str) -> list[tuple[Image.Image, float]]:
     vidcap = cv2.VideoCapture(video_path)
     fps = vidcap.get(cv2.CAP_PROP_FPS)
     total_frames = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))
-
-    # 더 적은 프레임을 추출하도록 조정
-    frame_interval = max(int(fps), int(total_frames / 10))  # 초당 1프레임 또는 최대 10프레임
+    frame_interval = max(int(fps), int(total_frames / 10))  
     frames = []
 
     for i in range(0, total_frames, frame_interval):
@@ -245,8 +235,6 @@ def downsample_video(video_path: str) -> list[tuple[Image.Image, float]]:
             pil_image = Image.fromarray(image)
             timestamp = round(i / fps, 2)
             frames.append((pil_image, timestamp))
-            
-            # 최대 5프레임만 사용
             if len(frames) >= 5:
                 break
 
@@ -275,7 +263,6 @@ def process_interleaved_images(message: dict) -> list[dict]:
     content = []
     image_index = 0
     
-    # 이미지 파일만 필터링
     image_files = [f for f in message["files"] if re.search(r"\.(png|jpg|jpeg|gif|webp)$", f, re.IGNORECASE)]
     
     for part in parts:
@@ -285,7 +272,6 @@ def process_interleaved_images(message: dict) -> list[dict]:
         elif part.strip():
             content.append({"type": "text", "text": part.strip()})
         else:
-            # 공백이거나 \n 같은 경우
             if isinstance(part, str) and part != "<image>":
                 content.append({"type": "text", "text": part})
     return content
@@ -295,66 +281,50 @@ def process_interleaved_images(message: dict) -> list[dict]:
 # PDF + CSV + TXT + 이미지/비디오
 ##################################################
 def is_image_file(file_path: str) -> bool:
-    """이미지 파일인지 확인"""
     return bool(re.search(r"\.(png|jpg|jpeg|gif|webp)$", file_path, re.IGNORECASE))
 
-
 def is_video_file(file_path: str) -> bool:
-    """비디오 파일인지 확인"""
     return file_path.endswith(".mp4")
 
-
 def is_document_file(file_path: str) -> bool:
-    """문서 파일인지 확인 (PDF, CSV, TXT)"""
     return (file_path.lower().endswith(".pdf") or 
             file_path.lower().endswith(".csv") or 
             file_path.lower().endswith(".txt"))
 
-
 def process_new_user_message(message: dict) -> list[dict]:
     if not message["files"]:
         return [{"type": "text", "text": message["text"]}]
 
-    # 1) 파일 분류
     video_files = [f for f in message["files"] if is_video_file(f)]
     image_files = [f for f in message["files"] if is_image_file(f)]
     csv_files = [f for f in message["files"] if f.lower().endswith(".csv")]
     txt_files = [f for f in message["files"] if f.lower().endswith(".txt")]
     pdf_files = [f for f in message["files"] if f.lower().endswith(".pdf")]
 
-    # 2) 사용자 원본 text 추가
     content_list = [{"type": "text", "text": message["text"]}]
 
-    # 3) CSV
     for csv_path in csv_files:
         csv_analysis = analyze_csv_file(csv_path)
         content_list.append({"type": "text", "text": csv_analysis})
 
-    # 4) TXT
     for txt_path in txt_files:
         txt_analysis = analyze_txt_file(txt_path)
         content_list.append({"type": "text", "text": txt_analysis})
 
-    # 5) PDF
     for pdf_path in pdf_files:
         pdf_markdown = pdf_to_markdown(pdf_path)
         content_list.append({"type": "text", "text": pdf_markdown})
 
-    # 6) 비디오 (한 개만 허용)
     if video_files:
         content_list += process_video(video_files[0])
         return content_list
 
-    # 7) 이미지 처리
     if "<image>" in message["text"] and image_files:
-        # interleaved
         interleaved_content = process_interleaved_images({"text": message["text"], "files": image_files})
-        # 원본 content_list 앞부분(텍스트)을 제거하고 interleaved로 대체
         if content_list[0]["type"] == "text":
-            content_list = content_list[1:]  # 원본 텍스트 제거
-        return interleaved_content + content_list  # interleaved + 나머지 문서 분석 내용
+            content_list = content_list[1:]
+        return interleaved_content + content_list
     else:
-        # 일반 여러 장
         for img_path in image_files:
             content_list.append({"type": "image", "url": img_path})
 
@@ -369,14 +339,11 @@ def process_history(history: list[dict]) -> list[dict]:
     current_user_content: list[dict] = []
     for item in history:
         if item["role"] == "assistant":
-            # user_content가 쌓여있다면 user 메시지로 저장
             if current_user_content:
                 messages.append({"role": "user", "content": current_user_content})
                 current_user_content = []
-            # 그 뒤 item은 assistant
             messages.append({"role": "assistant", "content": [{"type": "text", "text": item["content"]}]})
         else:
-            # user
             content = item["content"]
             if isinstance(content, str):
                 current_user_content.append({"type": "text", "text": content})
@@ -385,10 +352,8 @@ def process_history(history: list[dict]) -> list[dict]:
                 if is_image_file(file_path):
                     current_user_content.append({"type": "image", "url": file_path})
                 else:
-                    # 비이미지 파일은 텍스트로 처리
                     current_user_content.append({"type": "text", "text": f"[File: {os.path.basename(file_path)}]"})
-    
-    # 마지막 사용자 메시지가 처리되지 않은 경우 추가
+
     if current_user_content:
         messages.append({"role": "user", "content": current_user_content})
         
@@ -407,60 +372,54 @@ def run(
     use_web_search: bool = False,
     web_search_query: str = "",
 ) -> Iterator[str]:
-    """
-    The main inference function. Now extended with optional web_search arguments:
-    - use_web_search: bool
-    - web_search_query: str
-    If `use_web_search` is True, calls SERPHouse for the given `web_search_query`.
-    """
-    # Validate media constraints first
+
     if not validate_media_constraints(message, history):
         yield ""
         return
 
     try:
-        # If user opted for "Web Search", do it here and yield a prefix message
-        if use_web_search and web_search_query.strip():
-            ws_result = do_web_search(web_search_query.strip())
-            yield f"**[Web Search Results for '{web_search_query.strip()}':]**\n\n{ws_result}\n\n---\n"
+        # [새로 추가] web search 체크된 경우, 사용자가 입력한 "web_search_query" 대신
+        #             사용자의 메시지에서 키워드를 추출하여 검색
+        if use_web_search:
+            user_text = message["text"]
+            # 키워드 추출
+            ws_query = extract_keywords(user_text, top_k=5)
+            logger.info(f"[Auto WebSearch Keyword] {ws_query!r}")
+            # 상위 20개 결과 가져오기
+            ws_result = do_web_search(ws_query)
+            # 검색된 20개 제목을 system 메시지에 추가
+            system_search_content = f"[Search top-20 Titles Based on user prompt]\n{ws_result}\n"
+            # system 메시지로 추가
+            # (LLM이 이 정보를 참고하도록)
+            if system_search_content.strip():
+                history_system_msg = {
+                    "role": "system",
+                    "content": [{"type": "text", "text": system_search_content}]
+                }
+            else:
+                history_system_msg = {
+                    "role": "system",
+                    "content": [{"type": "text", "text": "No web search results"}]
+                }
+        else:
+            history_system_msg = None
 
         messages = []
         if system_prompt:
             messages.append({"role": "system", "content": [{"type": "text", "text": system_prompt}]})
+        # 만약 web search가 있었다면, 그 결과를 추가 system 메시지로 삽입
+        if history_system_msg:
+            messages.append(history_system_msg)
+
         messages.extend(process_history(history))
         
-        # 사용자 메시지 처리
         user_content = process_new_user_message(message)
-        
-        # 토큰 수를 줄이기 위해 너무 긴 텍스트는 잘라내기
         for item in user_content:
             if item["type"] == "text" and len(item["text"]) > MAX_CONTENT_CHARS:
                 item["text"] = item["text"][:MAX_CONTENT_CHARS] + "\n...(truncated)..."
         
         messages.append({"role": "user", "content": user_content})
 
-        # 모델 입력 생성 전 최종 확인
-        for msg in messages:
-            if msg["role"] != "user":
-                continue
-                
-            filtered_content = []
-            for item in msg["content"]:
-                if item["type"] == "image":
-                    if is_image_file(item["url"]):
-                        filtered_content.append(item)
-                    else:
-                        # 이미지 파일이 아닌 경우 텍스트로 변환
-                        filtered_content.append({
-                            "type": "text", 
-                            "text": f"[Non-image file: {os.path.basename(item['url'])}]"
-                        })
-                else:
-                    filtered_content.append(item)
-            
-            msg["content"] = filtered_content
-
-        # 모델 입력 생성
         inputs = processor.apply_chat_template(
             messages,
             add_generation_prompt=True,
@@ -469,7 +428,6 @@ def run(
             return_tensors="pt",
         ).to(device=model.device, dtype=torch.bfloat16)
 
-        # 텍스트 생성 스트리머 설정
         streamer = TextIteratorStreamer(processor, timeout=30.0, skip_prompt=True, skip_special_tokens=True)
         gen_kwargs = dict(
             inputs,
@@ -477,11 +435,9 @@ def run(
             max_new_tokens=max_new_tokens,
         )
         
-        # 별도 스레드에서 텍스트 생성
         t = Thread(target=model.generate, kwargs=gen_kwargs)
         t.start()
 
-        # 결과 스트리밍
         output = ""
         for new_text in streamer:
             output += new_text
@@ -493,9 +449,6 @@ def run(
 
 
 
-##################################################
-# 예시들 (한글화 버전)
-##################################################
 examples = [
 
     [
@@ -601,12 +554,6 @@ examples = [
 ]
 
 
-
-
-
-##############################################################################
-# Custom CSS similar to second example (colorful background, panel, etc.)
-##############################################################################
 css = """
 body {
     background: linear-gradient(135deg, #667eea, #764ba2);
@@ -668,11 +615,6 @@ title_html = """
 </p>
 """
 
-##############################################################################
-# Build a Blocks layout that includes:
-#   - A left sidebar with "Web Search" controls
-#   - The main ChatInterface in the center or right
-##############################################################################
 with gr.Blocks(css=css, title="Vidraft-Gemma-3-27B") as demo:
     gr.Markdown(title_html)
 
@@ -686,10 +628,11 @@ with gr.Blocks(css=css, title="Vidraft-Gemma-3-27B") as demo:
                     value=False,
                     info="Check to enable a SERPHouse web search before the chat reply"
                 )
+            # [중요] web_search_text는 사실상 사용 안 함 (자동추출로 검색)
             web_search_text = gr.Textbox(
                 lines=1,
-                label="Web Search Query",
-                placeholder="Enter search keywords..."
+                label="(Unused) Web Search Query",
+                placeholder="No direct input needed"
             )
 
             gr.Markdown("---")
@@ -710,9 +653,8 @@ with gr.Blocks(css=css, title="Vidraft-Gemma-3-27B") as demo:
                 value=2000,
             )
 
-            gr.Markdown("<br><br>")  # spacing
+            gr.Markdown("<br><br>")
 
-        # Main ChatInterface to the right
         with gr.Column(scale=7):
             chat = gr.ChatInterface(
                 fn=run,
@@ -731,7 +673,7 @@ with gr.Blocks(css=css, title="Vidraft-Gemma-3-27B") as demo:
                     system_prompt_box,
                     max_tokens_slider,
                     web_search_checkbox,
-                    web_search_text,
+                    web_search_text,  # 실제로는 사용 안함
                 ],
                 stop_btn=False,
                 title="Vidraft-Gemma-3-27B",
@@ -745,10 +687,9 @@ with gr.Blocks(css=css, title="Vidraft-Gemma-3-27B") as demo:
     with gr.Row(elem_id="examples_row"):
         with gr.Column(scale=12, elem_id="examples_container"):
             gr.Markdown("### Example Inputs (click to load)")
-            # The fix: pass an empty list to avoid the "None" error, so we keep the code structure.
             gr.Examples(
                 examples=examples,
-                inputs=[],  # Instead of None or chat.
+                inputs=[],  # Gradio가 dataset에 연결할 inputs가 없으므로 빈 리스트
                 cache_examples=False
             )