Create app.py

app.py

@@ -0,0 +1,495 @@
+import gradio as gr
+import spaces
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM, TextIteratorStreamer
+from threading import Thread
+import re
+import json
+from datetime import datetime
+import math
+import os
+
+# ═══════════════════════════════════════════════════════════
+# 🔧 모델 로딩
+# ═══════════════════════════════════════════════════════════
+
+MODEL_ID = "zai-org/GLM-4.7-Flash"
+
+print(f"[Init] Loading tokenizer from {MODEL_ID}...")
+tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
+
+print(f"[Init] Loading model from {MODEL_ID}...")
+model = None  # 지연 로딩
+
+def get_model():
+    global model
+    if model is None:
+        print("[Model] Loading model with bfloat16...")
+        model = AutoModelForCausalLM.from_pretrained(
+            MODEL_ID,
+            torch_dtype=torch.bfloat16,
+            device_map="auto",
+            trust_remote_code=True,
+            low_cpu_mem_usage=True,
+        )
+        print(f"[Model] Model loaded on {model.device}")
+    return model
+
+# ═══════════════════════════════════════════════════════════
+# 📄 파일 처리 함수
+# ═══════════════════════════════════════════════════════════
+
+def extract_text_from_pdf(file_path: str) -> str:
+    """PDF 파일에서 텍스트 추출"""
+    try:
+        import fitz  # PyMuPDF
+        doc = fitz.open(file_path)
+        text_parts = []
+        for page_num, page in enumerate(doc, 1):
+            text = page.get_text()
+            if text.strip():
+                text_parts.append(f"[페이지 {page_num}]\n{text}")
+        doc.close()
+        return "\n\n".join(text_parts) if text_parts else "[PDF에서 텍스트를 추출할 수 없습니다]"
+    except ImportError:
+        try:
+            from pypdf import PdfReader
+            reader = PdfReader(file_path)
+            text_parts = []
+            for page_num, page in enumerate(reader.pages, 1):
+                text = page.extract_text()
+                if text and text.strip():
+                    text_parts.append(f"[페이지 {page_num}]\n{text}")
+            return "\n\n".join(text_parts) if text_parts else "[PDF에서 텍스트를 추출할 수 없습니다]"
+        except Exception as e:
+            return f"[PDF 읽기 오류: {str(e)}]"
+    except Exception as e:
+        return f"[PDF 읽기 오류: {str(e)}]"
+
+def extract_text_from_docx(file_path: str) -> str:
+    """DOCX 파일에서 텍스트 추출"""
+    try:
+        from docx import Document
+        doc = Document(file_path)
+        
+        text_parts = []
+        
+        # 문단 추출
+        for para in doc.paragraphs:
+            if para.text.strip():
+                text_parts.append(para.text)
+        
+        # 테이블 추출
+        for table_idx, table in enumerate(doc.tables, 1):
+            table_text = [f"\n[표 {table_idx}]"]
+            for row in table.rows:
+                row_text = " | ".join(cell.text.strip() for cell in row.cells)
+                if row_text.strip():
+                    table_text.append(row_text)
+            if len(table_text) > 1:
+                text_parts.append("\n".join(table_text))
+        
+        return "\n\n".join(text_parts) if text_parts else "[DOCX에서 텍스트를 추출할 수 없습니다]"
+    except Exception as e:
+        return f"[DOCX 읽기 오류: {str(e)}]"
+
+def extract_text_from_txt(file_path: str) -> str:
+    """TXT 파일에서 텍스트 추출"""
+    try:
+        encodings = ['utf-8', 'cp949', 'euc-kr', 'latin-1']
+        for encoding in encodings:
+            try:
+                with open(file_path, 'r', encoding=encoding) as f:
+                    return f.read()
+            except UnicodeDecodeError:
+                continue
+        return "[텍스트 파일 인코딩을 인식할 수 없습니다]"
+    except Exception as e:
+        return f"[TXT 읽기 오류: {str(e)}]"
+
+def process_uploaded_file(file) -> tuple[str, str]:
+    """업로드된 파일 처리"""
+    if file is None:
+        return "", ""
+    
+    file_path = file.name if hasattr(file, 'name') else str(file)
+    file_name = os.path.basename(file_path)
+    file_ext = os.path.splitext(file_name)[1].lower()
+    
+    if file_ext == '.pdf':
+        content = extract_text_from_pdf(file_path)
+    elif file_ext == '.docx':
+        content = extract_text_from_docx(file_path)
+    elif file_ext in ['.txt', '.md', '.py', '.js', '.html', '.css', '.json', '.xml', '.csv']:
+        content = extract_text_from_txt(file_path)
+    else:
+        content = f"[지원하지 않는 파일 형식: {file_ext}]"
+    
+    # 텍스트 길이 제한
+    max_chars = 50000
+    if len(content) > max_chars:
+        content = content[:max_chars] + f"\n\n... [텍스트가 {max_chars}자로 잘렸습니다. 원본: {len(content)}자]"
+    
+    return file_name, content
+
+# ═══════════════════════════════════════════════════════════
+# 🛠️ Tool Definitions
+# ═══════════════════════════════════════════════════════════
+
+def execute_tool(tool_name: str, arguments: dict) -> str:
+    """도구 실행"""
+    try:
+        if tool_name == "calculator":
+            expr = arguments.get("expression", "")
+            allowed_names = {
+                "abs": abs, "round": round, "min": min, "max": max,
+                "sum": sum, "pow": pow, "sqrt": math.sqrt,
+                "sin": math.sin, "cos": math.cos, "tan": math.tan,
+                "log": math.log, "log10": math.log10, "exp": math.exp,
+                "pi": math.pi, "e": math.e,
+                "floor": math.floor, "ceil": math.ceil,
+            }
+            expr = re.sub(r'[^0-9+\-*/().a-zA-Z_ ]', '', expr)
+            result = eval(expr, {"__builtins__": {}}, allowed_names)
+            return f"계산 결과: {expr} = {result}"
+        
+        elif tool_name == "get_current_time":
+            tz = arguments.get("timezone", "UTC")
+            now = datetime.now()
+            return f"현재 시간 ({tz}): {now.strftime('%Y-%m-%d %H:%M:%S')}"
+        
+        elif tool_name == "unit_converter":
+            value = arguments.get("value", 0)
+            from_unit = arguments.get("from_unit", "").lower()
+            to_unit = arguments.get("to_unit", "").lower()
+            
+            conversions = {
+                ("km", "m"): lambda x: x * 1000,
+                ("m", "km"): lambda x: x / 1000,
+                ("kg", "g"): lambda x: x * 1000,
+                ("g", "kg"): lambda x: x / 1000,
+                ("c", "f"): lambda x: x * 9/5 + 32,
+                ("f", "c"): lambda x: (x - 32) * 5/9,
+                ("km", "mile"): lambda x: x * 0.621371,
+                ("mile", "km"): lambda x: x * 1.60934,
+                ("kg", "lb"): lambda x: x * 2.20462,
+                ("lb", "kg"): lambda x: x * 0.453592,
+            }
+            
+            key = (from_unit, to_unit)
+            if key in conversions:
+                result = conversions[key](value)
+                return f"변환 결과: {value} {from_unit} = {result:.4f} {to_unit}"
+            else:
+                return f"지원하지 않는 단위 변환: {from_unit} -> {to_unit}"
+        
+        elif tool_name == "code_executor":
+            code = arguments.get("code", "")
+            local_vars = {}
+            exec(code, {"__builtins__": {"print": print, "range": range, "len": len, "str": str, "int": int, "float": float, "list": list, "dict": dict}}, local_vars)
+            if "result" in local_vars:
+                return f"실행 결과: {local_vars['result']}"
+            return "코드 실행 완료"
+        
+        else:
+            return f"알 수 없는 도구: {tool_name}"
+    
+    except Exception as e:
+        return f"도구 실행 오류: {str(e)}"
+
+def parse_tool_calls(response: str) -> list:
+    """응답에서 도구 호출 파싱"""
+    tool_calls = []
+    
+    patterns = [
+        r'<\|tool_call\|>(\{.*?\})<\|/tool_call\|>',
+        r'```json\s*(\{[^`]*"name"[^`]*\})\s*```',
+        r'\{"name":\s*"(\w+)",\s*"arguments":\s*(\{[^}]+\})\}',
+    ]
+    
+    for pattern in patterns:
+        matches = re.findall(pattern, response, re.DOTALL)
+        for match in matches:
+            try:
+                if isinstance(match, tuple):
+                    tool_call = {"name": match[0], "arguments": json.loads(match[1])}
+                else:
+                    tool_call = json.loads(match)
+                tool_calls.append(tool_call)
+            except:
+                continue
+    
+    return tool_calls
+
+# ═══════════════════════════════════════════════════════════
+# 💬 스트리밍 채팅 함수
+# ═══════════════════════════════════════════════════════════
+
+file_context = {"name": "", "content": ""}
+
+@spaces.GPU(duration=120)
+def chat_streaming(
+    message: str,
+    history: list,
+    system_prompt: str,
+    max_tokens: int,
+    temperature: float,
+    top_p: float,
+    enable_thinking: bool,
+    enable_tools: bool,
+):
+    """스트리밍 채팅 생성"""
+    global file_context
+    
+    if not message.strip():
+        yield history, ""
+        return
+    
+    model = get_model()
+    messages = []
+    
+    # 시스템 프롬프트 구성
+    sys_content = system_prompt if system_prompt.strip() else "You are a helpful AI assistant."
+    
+    # 파일 컨텍스트 추가
+    if file_context["content"]:
+        sys_content += f"\n\n[업로드된 파일: {file_context['name']}]\n아래는 사용자가 업로드한 파일의 내용입니다. 이 내용을 참고하여 질문에 답변하세요.\n\n---\n{file_context['content']}\n---"
+    
+    # Tool 정보 추가
+    if enable_tools:
+        tool_desc = """
+You have access to the following tools:
+1. calculator: Perform mathematical calculations
+2. get_current_time: Get current date and time  
+3. unit_converter: Convert between units
+4. code_executor: Execute Python code
+
+To use a tool, respond with: {"name": "tool_name", "arguments": {...}}
+"""
+        sys_content += f"\n\n{tool_desc}"
+    
+    messages.append({"role": "system", "content": sys_content})
+    
+    # 히스토리 추가
+    for h in history:
+        if h[0]:
+            messages.append({"role": "user", "content": h[0]})
+        if h[1]:
+            messages.append({"role": "assistant", "content": h[1]})
+    
+    # 현재 메시지
+    user_content = message
+    if enable_thinking:
+        user_content = f"<think>\nLet me think step by step.\n</think>\n\n{message}"
+    
+    messages.append({"role": "user", "content": user_content})
+    
+    # 토크나이즈
+    try:
+        inputs = tokenizer.apply_chat_template(
+            messages,
+            add_generation_prompt=True,
+            tokenize=True,
+            return_dict=True,
+            return_tensors="pt",
+        ).to(model.device)
+    except Exception as e:
+        yield history + [[message, f"토크나이즈 오류: {str(e)}"]], ""
+        return
+    
+    # 스트리머 설정
+    streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+    
+    generation_kwargs = {
+        **inputs,
+        "streamer": streamer,
+        "max_new_tokens": max_tokens,
+        "temperature": temperature if temperature > 0 else 0.01,
+        "top_p": top_p,
+        "do_sample": temperature > 0,
+        "pad_token_id": tokenizer.pad_token_id or tokenizer.eos_token_id,
+    }
+    
+    thread = Thread(target=model.generate, kwargs=generation_kwargs)
+    thread.start()
+    
+    partial_response = ""
+    new_history = history + [[message, ""]]
+    
+    for new_token in streamer:
+        partial_response += new_token
+        new_history[-1][1] = partial_response
+        yield new_history, ""
+    
+    thread.join()
+    
+    # Tool 호출 처리
+    if enable_tools:
+        tool_calls = parse_tool_calls(partial_response)
+        if tool_calls:
+            tool_results = []
+            for tc in tool_calls:
+                result = execute_tool(tc.get("name", ""), tc.get("arguments", {}))
+                tool_results.append(result)
+            
+            if tool_results:
+                final_response = partial_response + "\n\n📌 **도구 실행 결과:**\n" + "\n".join(tool_results)
+                new_history[-1][1] = final_response
+    
+    yield new_history, ""
+
+def handle_file_upload(file):
+    """파일 업로드 처리"""
+    global file_context
+    
+    if file is None:
+        file_context = {"name": "", "content": ""}
+        return "📂 파일이 제거되었습니다."
+    
+    file_name, content = process_uploaded_file(file)
+    
+    if content.startswith("[") and "오류" in content:
+        file_context = {"name": "", "content": ""}
+        return f"❌ {content}"
+    
+    file_context = {"name": file_name, "content": content}
+    
+    preview = content[:500] + "..." if len(content) > 500 else content
+    char_count = len(content)
+    
+    return f"""✅ **파일 로드 완료: {file_name}**
+- 문자 수: {char_count:,}자
+- 미리보기:
+```
+{preview}
+```"""
+
+def clear_file():
+    """파일 컨텍스트 초기화"""
+    global file_context
+    file_context = {"name": "", "content": ""}
+    return None, "📂 파일이 제거되었습니다."
+
+def clear_chat():
+    """채팅 초기화"""
+    return [], ""
+
+# ═══════════════════════════════════════════════════════════
+# 🎨 Gradio UI
+# ═══════════════════════════════════════════════════════════
+
+CUSTOM_CSS = """
+.chatbot { height: 500px !important; }
+.contain { max-width: 1200px !important; margin: auto !important; }
+.title { text-align: center; margin-bottom: 1rem; }
+.tool-badge {
+    display: inline-block;
+    padding: 4px 8px;
+    margin: 2px;
+    border-radius: 12px;
+    font-size: 12px;
+    background: #e0e7ff;
+    color: #3730a3;
+}
+"""
+
+with gr.Blocks(css=CUSTOM_CSS, title="GLM-4.7-Flash Chatbot") as demo:
+    gr.HTML("""
+    <div class="title">
+        <h1>🤖 GLM-4.7-Flash Chatbot</h1>
+        <p>30B-A3B MoE 모델 기반 스트리밍 챗봇 | 문서 분석 | Tool Calling</p>
+        <div>
+            <span class="tool-badge">📄 PDF</span>
+            <span class="tool-badge">📝 DOCX</span>
+            <span class="tool-badge">📃 TXT</span>
+            <span class="tool-badge">🧮 계산기</span>
+            <span class="tool-badge">🕐 시간조회</span>
+            <span class="tool-badge">📐 단위변환</span>
+            <span class="tool-badge">🐍 코드실행</span>
+        </div>
+    </div>
+    """)
+    
+    with gr.Row():
+        with gr.Column(scale=3):
+            chatbot = gr.Chatbot(
+                label="대화",
+                elem_classes=["chatbot"],
+                height=500,
+                show_copy_button=True,
+            )
+            
+            with gr.Row():
+                message = gr.Textbox(
+                    label="메시지 입력",
+                    placeholder="메시지를 입력하세요... (Shift+Enter로 전송)",
+                    lines=3,
+                    scale=4,
+                )
+                submit_btn = gr.Button("전송 📤", variant="primary", scale=1)
+            
+            with gr.Row():
+                clear_btn = gr.Button("대화 초기화 🗑️")
+                stop_btn = gr.Button("생성 중지 ⏹️")
+            
+            # 파일 업로드
+            with gr.Accordion("📁 문서 업로드 (PDF / DOCX / TXT)", open=True):
+                file_upload = gr.File(
+                    label="파일 선택",
+                    file_types=[".pdf", ".docx", ".txt", ".md", ".py", ".js", ".html", ".css", ".json", ".xml", ".csv"],
+                    file_count="single",
+                )
+                file_status = gr.Markdown("📂 파일을 업로드하면 내용을 분석할 수 있습니다.")
+                clear_file_btn = gr.Button("📂 파일 제거", size="sm")
+        
+        with gr.Column(scale=1):
+            gr.Markdown("### ⚙️ 설정")
+            
+            system_prompt = gr.Textbox(
+                label="시스템 프롬프트",
+                value="You are a helpful AI assistant. Answer questions accurately and concisely in the same language as the user. You can analyze uploaded documents and use tools when needed.",
+                lines=4,
+            )
+            
+            max_tokens = gr.Slider(64, 4096, value=1024, step=64, label="최대 토큰 수")
+            temperature = gr.Slider(0, 2, value=0.7, step=0.1, label="Temperature")
+            top_p = gr.Slider(0.1, 1.0, value=0.9, step=0.05, label="Top-P")
+            
+            enable_thinking = gr.Checkbox(label="🧠 Thinking 모드", value=False)
+            enable_tools = gr.Checkbox(label="🛠️ Tool Calling 활성화", value=True)
+            
+            gr.Markdown("### 📝 예시 질문")
+            examples = gr.Examples(
+                examples=[
+                    ["안녕하세요! 자기소개 해주세요."],
+                    ["업로드한 문서를 요약해줘"],
+                    ["이 문서의 핵심 내용이 뭐야?"],
+                    ["문서에서 중요한 키워드를 추출해줘"],
+                    ["123 * 456 + 789를 계산해줘"],
+                    ["현재 시간이 몇 시야?"],
+                    ["100 킬로미터는 몇 마일이야?"],
+                ],
+                inputs=message,
+            )
+    
+    # 이벤트 핸들러
+    submit_event = submit_btn.click(
+        fn=chat_streaming,
+        inputs=[message, chatbot, system_prompt, max_tokens, temperature, top_p, enable_thinking, enable_tools],
+        outputs=[chatbot, message],
+    )
+    
+    message.submit(
+        fn=chat_streaming,
+        inputs=[message, chatbot, system_prompt, max_tokens, temperature, top_p, enable_thinking, enable_tools],
+        outputs=[chatbot, message],
+    )
+    
+    clear_btn.click(fn=clear_chat, outputs=[chatbot, message])
+    stop_btn.click(fn=None, cancels=[submit_event])
+    
+    file_upload.change(fn=handle_file_upload, inputs=[file_upload], outputs=[file_status])
+    clear_file_btn.click(fn=clear_file, outputs=[file_upload, file_status])
+
+if __name__ == "__main__":
+    demo.queue().launch(server_name="0.0.0.0", server_port=7860, share=False)