Delete app.py

app.py

@@ -1,462 +0,0 @@
-#!/usr/bin/env python
-
-import os
-import re
-import tempfile
-from collections.abc import Iterator
-from threading import Thread
-
-import cv2
-import gradio as gr
-import spaces
-import torch
-from loguru import logger
-from PIL import Image
-from transformers import AutoProcessor, Gemma3ForConditionalGeneration, TextIteratorStreamer
-
-# [PDF] PyPDF2 추가
-import PyPDF2
-# [CSV] Pandas 추가
-import pandas as pd
-
-model_id = os.getenv("MODEL_ID", "google/gemma-3-27b-it")
-processor = AutoProcessor.from_pretrained(model_id, padding_side="left")
-model = Gemma3ForConditionalGeneration.from_pretrained(
-    model_id, device_map="auto", torch_dtype=torch.bfloat16, attn_implementation="eager"
-)
-
-MAX_NUM_IMAGES = int(os.getenv("MAX_NUM_IMAGES", "5"))
-
-###################################################################
-# CSV를 Markdown으로 변환하는 유틸 함수
-###################################################################
-def csv_to_markdown(csv_path: str) -> str:
-    """
-    CSV 파일 전체를 문자열로 변환하여 Markdown 형태로 반환.
-    (매우 큰 CSV라면 전체를 넘기는 것이 위험할 수 있음 -> 필요 시 잘라낼 것)
-    """
-    try:
-        df = pd.read_csv(csv_path)
-        df_str = df.to_string()
-        # 필요하다면 길이 제한을 걸어도 됨
-        # if len(df_str) > 10000:
-        #     df_str = df_str[:10000] + "\n...(truncated)..."
-        
-        return f"**[CSV File: {os.path.basename(csv_path)}]**\n\n```\n{df_str}\n```"
-    except Exception as e:
-        return f"Failed to read CSV ({os.path.basename(csv_path)}): {str(e)}"
-
-###################################################################
-# PDF -> Markdown 변환 함수 (기존)
-###################################################################
-def pdf_to_markdown(pdf_path: str) -> str:
-    """
-    PDF 파일을 텍스트로 추출 후, 간단한 Markdown 형태로 반환.
-    """
-    text_chunks = []
-    with open(pdf_path, "rb") as f:
-        reader = PyPDF2.PdfReader(f)
-        for page_num, page in enumerate(reader.pages, start=1):
-            page_text = page.extract_text()
-            page_text = page_text.strip() if page_text else ""
-            if page_text:
-                # 페이지별로 간단한 헤더와 본문을 Markdown으로 합침
-                text_chunks.append(f"## Page {page_num}\n\n{page_text}\n")
-    return "\n".join(text_chunks)
-
-###################################################################
-# 이미지/비디오 개수 카운트 (기존)
-###################################################################
-def count_files_in_new_message(paths: list[str]) -> tuple[int, int]:
-    image_count = 0
-    video_count = 0
-    for path in paths:
-        if path.endswith(".mp4"):
-            video_count += 1
-        else:
-            image_count += 1
-    return image_count, video_count
-
-def count_files_in_history(history: list[dict]) -> tuple[int, int]:
-    image_count = 0
-    video_count = 0
-    for item in history:
-        if item["role"] != "user" or isinstance(item["content"], str):
-            continue
-        if item["content"][0].endswith(".mp4"):
-            video_count += 1
-        else:
-            image_count += 1
-    return image_count, video_count
-
-###################################################################
-# 미디어(이미지/비디오) 제한 검사 + PDF/CSV 예외 (기존/수정)
-###################################################################
-def validate_media_constraints(message: dict, history: list[dict]) -> bool:
-    """
-    이미지/비디오 개수와 혼합 여부 등을 검사하는 함수.
-    PDF, CSV 등은 검사 로직에서 제외하여 업로드만 허용.
-    """
-    # pdf, csv 파일 제외
-    pdf_files = [f for f in message["files"] if f.endswith(".pdf")]
-    csv_files = [f for f in message["files"] if f.lower().endswith(".csv")]
-    non_pdf_csv_files = [f for f in message["files"] 
-                         if not f.endswith(".pdf") and not f.lower().endswith(".csv")]
-
-    # 기존 로직은 이미지/비디오에 대해서만 체크
-    new_image_count, new_video_count = count_files_in_new_message(non_pdf_csv_files)
-    history_image_count, history_video_count = count_files_in_history(history)
-    image_count = history_image_count + new_image_count
-    video_count = history_video_count + new_video_count
-
-    if video_count > 1:
-        gr.Warning("Only one video is supported.")
-        return False
-    if video_count == 1:
-        if image_count > 0:
-            gr.Warning("Mixing images and videos is not allowed.")
-            return False
-        if "<image>" in message["text"]:
-            gr.Warning("Using <image> tags with video files is not supported.")
-            return False
-
-    if video_count == 0 and image_count > MAX_NUM_IMAGES:
-        gr.Warning(f"You can upload up to {MAX_NUM_IMAGES} images.")
-        return False
-
-    # <image> 태그가 있을 경우, 이미지 수와 태그 수 일치
-    if "<image>" in message["text"]:
-        if message["text"].count("<image>") != new_image_count:
-            gr.Warning("The number of <image> tags in the text does not match the number of images.")
-            return False
-
-    return True
-
-###################################################################
-# 동영상 처리 (기존)
-###################################################################
-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 = int(fps / 3)
-    frames = []
-
-    for i in range(0, total_frames, frame_interval):
-        vidcap.set(cv2.CAP_PROP_POS_FRAMES, i)
-        success, image = vidcap.read()
-        if success:
-            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
-            pil_image = Image.fromarray(image)
-            timestamp = round(i / fps, 2)
-            frames.append((pil_image, timestamp))
-
-    vidcap.release()
-    return frames
-
-def process_video(video_path: str) -> list[dict]:
-    content = []
-    frames = downsample_video(video_path)
-    for frame in frames:
-        pil_image, timestamp = frame
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
-            pil_image.save(temp_file.name)
-            content.append({"type": "text", "text": f"Frame {timestamp}:"})
-            content.append({"type": "image", "url": temp_file.name})
-    logger.debug(f"{content=}")
-    return content
-
-###################################################################
-# <image> 태그 interleaved 이미지 처리 (기존)
-###################################################################
-def process_interleaved_images(message: dict) -> list[dict]:
-    logger.debug(f"{message['files']=}")
-    parts = re.split(r"(<image>)", message["text"])
-    logger.debug(f"{parts=}")
-
-    content = []
-    image_index = 0
-    for part in parts:
-        logger.debug(f"{part=}")
-        if part == "<image>":
-            content.append({"type": "image", "url": message["files"][image_index]})
-            logger.debug(f"file: {message['files'][image_index]}")
-            image_index += 1
-        elif part.strip():
-            content.append({"type": "text", "text": part.strip()})
-        elif isinstance(part, str) and part != "<image>":
-            content.append({"type": "text", "text": part})
-    logger.debug(f"{content=}")
-    return content
-
-###################################################################
-# 새 user message 처리 (PDF + CSV + 이미지/비디오)
-###################################################################
-def process_new_user_message(message: dict) -> list[dict]:
-    if not message["files"]:
-        return [{"type": "text", "text": message["text"]}]
-
-    # PDF 파일 목록
-    pdf_files = [f for f in message["files"] if f.endswith(".pdf")]
-    # CSV 파일 목록
-    csv_files = [f for f in message["files"] if f.lower().endswith(".csv")]
-    # 이미지/비디오 (기존)
-    other_files = [f for f in message["files"] 
-                   if not f.endswith(".pdf") and not f.lower().endswith(".csv")]
-
-    # 일단 사용자의 text를 먼저 넣는다
-    content_list = [{"type": "text", "text": message["text"]}]
-
-    # [PDF] 변환 후 추가
-    for pdf_path in pdf_files:
-        pdf_markdown = pdf_to_markdown(pdf_path)
-        if pdf_markdown.strip():
-            content_list.append({"type": "text", "text": pdf_markdown})
-        else:
-            content_list.append({"type": "text", "text": "(PDF에서 텍스트 추출 실패)"})
-
-    # [CSV] 변환 후 추가
-    for cfile in csv_files:
-        csv_md = csv_to_markdown(cfile)
-        content_list.append({"type": "text", "text": csv_md})
-
-    # 영상 처리
-    video_files = [f for f in other_files if f.endswith(".mp4")]
-    if video_files:
-        content_list += process_video(video_files[0])
-        return content_list
-
-    # interleaved 이미지
-    if "<image>" in message["text"]:
-        return process_interleaved_images(message)
-
-    # 일반 이미지(여러 장)
-    image_files = [f for f in other_files if not f.endswith(".mp4")]
-    if image_files:
-        content_list += [{"type": "image", "url": path} for path in image_files]
-
-    return content_list
-
-###################################################################
-# 히스토리 -> LLM용 메시지 변환 (기존)
-###################################################################
-def process_history(history: list[dict]) -> list[dict]:
-    messages = []
-    current_user_content: list[dict] = []
-    for item in history:
-        if item["role"] == "assistant":
-            if current_user_content:
-                messages.append({"role": "user", "content": current_user_content})
-                current_user_content = []
-            messages.append({"role": "assistant", "content": [{"type": "text", "text": item["content"]}]})
-        else:
-            content = item["content"]
-            if isinstance(content, str):
-                current_user_content.append({"type": "text", "text": content})
-            else:
-                current_user_content.append({"type": "image", "url": content[0]})
-    return messages
-
-###################################################################
-# 메인 추론 함수 (기존)
-###################################################################
-@spaces.GPU(duration=120)
-def run(message: dict, history: list[dict], system_prompt: str = "", max_new_tokens: int = 512) -> Iterator[str]:
-    if not validate_media_constraints(message, history):
-        yield ""
-        return
-
-    messages = []
-    if system_prompt:
-        messages.append({"role": "system", "content": [{"type": "text", "text": system_prompt}]})
-    messages.extend(process_history(history))
-    messages.append({"role": "user", "content": process_new_user_message(message)})
-
-    inputs = processor.apply_chat_template(
-        messages,
-        add_generation_prompt=True,
-        tokenize=True,
-        return_dict=True,
-        return_tensors="pt",
-    ).to(device=model.device, dtype=torch.bfloat16)
-
-    streamer = TextIteratorStreamer(processor, timeout=30.0, skip_prompt=True, skip_special_tokens=True)
-    generate_kwargs = dict(
-        inputs,
-        streamer=streamer,
-        max_new_tokens=max_new_tokens,
-    )
-    t = Thread(target=model.generate, kwargs=generate_kwargs)
-    t.start()
-
-    output = ""
-    for delta in streamer:
-        output += delta
-        yield output
-
-###################################################################
-# 예시들 (기존 그대로)
-###################################################################
-examples = [
-    [
-        {
-            "text": "I need to be in Japan for 10 days, going to Tokyo, Kyoto and Osaka. Think about number of attractions in each of them and allocate number of days to each city. Make public transport recommendations.",
-            "files": [],
-        }
-    ],
-    [
-        {
-            "text": "Write the matplotlib code to generate the same bar chart.",
-            "files": ["assets/additional-examples/barchart.png"],
-        }
-    ],
-    [
-        {
-            "text": "What is odd about this video?",
-            "files": ["assets/additional-examples/tmp.mp4"],
-        }
-    ],
-    [
-        {
-            "text": "I already have this supplement <image> and I want to buy this one <image>. Any warnings I should know about?",
-            "files": ["assets/additional-examples/pill1.png", "assets/additional-examples/pill2.png"],
-        }
-    ],
-    [
-        {
-            "text": "Write a poem inspired by the visual elements of the images.",
-            "files": ["assets/sample-images/06-1.png", "assets/sample-images/06-2.png"],
-        }
-    ],
-    [
-        {
-            "text": "Compose a short musical piece inspired by the visual elements of the images.",
-            "files": [
-                "assets/sample-images/07-1.png",
-                "assets/sample-images/07-2.png",
-                "assets/sample-images/07-3.png",
-                "assets/sample-images/07-4.png",
-            ],
-        }
-    ],
-    [
-        {
-            "text": "Write a short story about what might have happened in this house.",
-            "files": ["assets/sample-images/08.png"],
-        }
-    ],
-    [
-        {
-            "text": "Create a short story based on the sequence of images.",
-            "files": [
-                "assets/sample-images/09-1.png",
-                "assets/sample-images/09-2.png",
-                "assets/sample-images/09-3.png",
-                "assets/sample-images/09-4.png",
-                "assets/sample-images/09-5.png",
-            ],
-        }
-    ],
-    [
-        {
-            "text": "Describe the creatures that would live in this world.",
-            "files": ["assets/sample-images/10.png"],
-        }
-    ],
-    [
-        {
-            "text": "Read text in the image.",
-            "files": ["assets/additional-examples/1.png"],
-        }
-    ],
-    [
-        {
-            "text": "When is this ticket dated and how much did it cost?",
-            "files": ["assets/additional-examples/2.png"],
-        }
-    ],
-    [
-        {
-            "text": "Read the text in the image into markdown.",
-            "files": ["assets/additional-examples/3.png"],
-        }
-    ],
-    [
-        {
-            "text": "Evaluate this integral.",
-            "files": ["assets/additional-examples/4.png"],
-        }
-    ],
-    [
-        {
-            "text": "caption this image",
-            "files": ["assets/sample-images/01.png"],
-        }
-    ],
-    [
-        {
-            "text": "What's the sign says?",
-            "files": ["assets/sample-images/02.png"],
-        }
-    ],
-    [
-        {
-            "text": "Compare and contrast the two images.",
-            "files": ["assets/sample-images/03.png"],
-        }
-    ],
-    [
-        {
-            "text": "List all the objects in the image and their colors.",
-            "files": ["assets/sample-images/04.png"],
-        }
-    ],
-    [
-        {
-            "text": "Describe the atmosphere of the scene.",
-            "files": ["assets/sample-images/05.png"],
-        }
-    ],
-]
-
-###################################################################
-# PDF + CSV를 허용하는 Gradio ChatInterface
-###################################################################
-demo = gr.ChatInterface(
-    fn=run,
-    type="messages",
-    chatbot=gr.Chatbot(type="messages", scale=1, allow_tags=["image"]),
-    textbox=gr.MultimodalTextbox(
-        file_types=["image", ".mp4", ".pdf", ".csv"],  # pdf & csv 허용
-        file_count="multiple",
-        autofocus=True
-    ),
-    multimodal=True,
-    additional_inputs=[
-        gr.Textbox(
-            label="System Prompt",
-            value=(
-                "You are a deeply thoughtful AI. Consider problems thoroughly and derive correct "
-                "solutions through systematic reasoning. Please answer in korean."
-            )
-        ),
-        gr.Slider(
-            label="Max New Tokens",
-            minimum=100,
-            maximum=8000,
-            step=50,
-            value=2000
-        ),
-    ],
-    stop_btn=False,
-    title="Gemma 3 27B IT",
-    examples=examples,
-    run_examples_on_click=False,
-    cache_examples=False,
-    css_paths="style.css",
-    delete_cache=(1800, 1800),
-)
-
-if __name__ == "__main__":
-    demo.launch()