app-backup.py

import gradio as gr
import spaces
import os
from typing import List, Dict, Any, Optional, Tuple
import hashlib
from datetime import datetime
import numpy as np
from transformers import pipeline, TextIteratorStreamer
import torch
from threading import Thread
import re

# PDF 처리 라이브러리
try:
    import fitz  # PyMuPDF
    PDF_AVAILABLE = True
except ImportError:
    PDF_AVAILABLE = False
    print("⚠️ PyMuPDF not installed. Install with: pip install pymupdf")

try:
    from sentence_transformers import SentenceTransformer
    ST_AVAILABLE = True
except ImportError:
    ST_AVAILABLE = False
    print("⚠️ Sentence Transformers not installed. Install with: pip install sentence-transformers")

# Custom CSS
custom_css = """
.gradio-container {
    background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
    min-height: 100vh;
    font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif;
}

.main-container {
    background: rgba(255, 255, 255, 0.98);
    border-radius: 16px;
    padding: 24px;
    box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.1), 0 2px 4px -1px rgba(0, 0, 0, 0.06);
    border: 1px solid rgba(0, 0, 0, 0.05);
    margin: 12px;
}

.pdf-status {
    padding: 12px 16px;
    border-radius: 12px;
    margin: 12px 0;
    font-size: 0.95rem;
    font-weight: 500;
}

.pdf-success {
    background: linear-gradient(135deg, #d4edda 0%, #c3e6cb 100%);
    border: 1px solid #b1dfbb;
    color: #155724;
}

.pdf-error {
    background: linear-gradient(135deg, #f8d7da 0%, #f5c6cb 100%);
    border: 1px solid #f1aeb5;
    color: #721c24;
}

.pdf-info {
    background: linear-gradient(135deg, #d1ecf1 0%, #bee5eb 100%);
    border: 1px solid #9ec5d8;
    color: #0c5460;
}

.rag-context {
    background: linear-gradient(135deg, #fef3c7 0%, #fde68a 100%);
    border-left: 4px solid #f59e0b;
    padding: 12px;
    margin: 12px 0;
    border-radius: 8px;
    font-size: 0.9rem;
}

.thinking-section {
    background: rgba(0, 0, 0, 0.02);
    border: 1px solid rgba(0, 0, 0, 0.1);
    border-radius: 8px;
    padding: 12px;
    margin: 8px 0;
}
"""

class SimpleTextSplitter:
    """텍스트 분할기"""
    def __init__(self, chunk_size=800, chunk_overlap=100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_text(self, text: str) -> List[str]:
        """텍스트를 청크로 분할"""
        chunks = []
        sentences = text.split('. ')
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk) + len(sentence) < self.chunk_size:
                current_chunk += sentence + ". "
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + ". "
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks

class PDFRAGSystem:
    """PDF 기반 RAG 시스템"""
    
    def __init__(self):
        self.documents = {}
        self.document_chunks = {}
        self.embeddings_store = {}
        self.text_splitter = SimpleTextSplitter(chunk_size=800, chunk_overlap=100)
        
        # 임베딩 모델 초기화
        self.embedder = None
        if ST_AVAILABLE:
            try:
                self.embedder = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
                print("✅ 임베딩 모델 로드 성공")
            except Exception as e:
                print(f"⚠️ 임베딩 모델 로드 실패: {e}")
    
    def extract_text_from_pdf(self, pdf_path: str) -> Dict[str, Any]:
        """PDF에서 텍스트 추출"""
        if not PDF_AVAILABLE:
            return {
                "metadata": {
                    "title": "PDF Reader Not Available",
                    "file_name": os.path.basename(pdf_path),
                    "pages": 0
                },
                "full_text": "PDF 처리를 위해 'pip install pymupdf'를 실행해주세요."
            }
        
        try:
            doc = fitz.open(pdf_path)
            text_content = []
            metadata = {
                "title": doc.metadata.get("title", os.path.basename(pdf_path)),
                "pages": len(doc),
                "file_name": os.path.basename(pdf_path)
            }
            
            for page_num, page in enumerate(doc):
                text = page.get_text()
                if text.strip():
                    text_content.append(text)
            
            doc.close()
            
            return {
                "metadata": metadata,
                "full_text": "\n\n".join(text_content)
            }
        except Exception as e:
            raise Exception(f"PDF 처리 오류: {str(e)}")
    
    def process_and_store_pdf(self, pdf_path: str, doc_id: str) -> Dict[str, Any]:
        """PDF 처리 및 저장"""
        try:
            # PDF 텍스트 추출
            pdf_data = self.extract_text_from_pdf(pdf_path)
            
            # 텍스트를 청크로 분할
            chunks = self.text_splitter.split_text(pdf_data["full_text"])
            
            if not chunks:
                print("Warning: No chunks created from PDF")
                return {"success": False, "error": "No text content found in PDF"}
            
            print(f"Created {len(chunks)} chunks from PDF")
            
            # 청크 저장
            self.document_chunks[doc_id] = chunks
            
            # 임베딩 생성 (선택적)
            if self.embedder:
                try:
                    print("Generating embeddings...")
                    embeddings = self.embedder.encode(chunks)
                    self.embeddings_store[doc_id] = embeddings
                    print(f"Generated {len(embeddings)} embeddings")
                except Exception as e:
                    print(f"Warning: Failed to generate embeddings: {e}")
                    # 임베딩 실패해도 계속 진행
            
            # 문서 정보 저장
            self.documents[doc_id] = {
                "metadata": pdf_data["metadata"],
                "chunk_count": len(chunks),
                "upload_time": datetime.now().isoformat()
            }
            
            # 디버그: 첫 번째 청크 출력
            print(f"First chunk preview: {chunks[0][:200]}...")
            
            return {
                "success": True,
                "doc_id": doc_id,
                "chunks": len(chunks),
                "pages": pdf_data["metadata"]["pages"],
                "title": pdf_data["metadata"]["title"]
            }
            
        except Exception as e:
            print(f"Error processing PDF: {e}")
            return {"success": False, "error": str(e)}
    
    def search_relevant_chunks(self, query: str, doc_ids: List[str], top_k: int = 3) -> List[Dict]:
        """관련 청크 검색"""
        all_relevant_chunks = []
        
        print(f"Searching chunks for query: '{query[:50]}...' in {len(doc_ids)} documents")
        
        # 먼저 문서가 있는지 확인
        for doc_id in doc_ids:
            if doc_id not in self.document_chunks:
                print(f"Warning: Document {doc_id} not found in chunks")
                continue
                
            chunks = self.document_chunks[doc_id]
            print(f"Document {doc_id} has {len(chunks)} chunks")
            
            # 임베딩 기반 검색 시도
            if self.embedder and doc_id in self.embeddings_store:
                try:
                    query_embedding = self.embedder.encode([query])[0]
                    doc_embeddings = self.embeddings_store[doc_id]
                    
                    # 코사인 유사도 계산 (안전하게)
                    similarities = []
                    for i, emb in enumerate(doc_embeddings):
                        try:
                            query_norm = np.linalg.norm(query_embedding)
                            emb_norm = np.linalg.norm(emb)
                            
                            if query_norm > 0 and emb_norm > 0:
                                sim = np.dot(query_embedding, emb) / (query_norm * emb_norm)
                                similarities.append(sim)
                            else:
                                similarities.append(0.0)
                        except Exception as e:
                            print(f"Error calculating similarity for chunk {i}: {e}")
                            similarities.append(0.0)
                    
                    # 상위 청크 선택
                    if similarities:
                        top_indices = np.argsort(similarities)[-min(top_k, len(similarities)):][::-1]
                        
                        for idx in top_indices:
                            if idx < len(chunks):  # 인덱스 범위 확인
                                all_relevant_chunks.append({
                                    "content": chunks[idx],
                                    "doc_name": self.documents[doc_id]["metadata"]["file_name"],
                                    "similarity": similarities[idx]
                                })
                                print(f"Added chunk {idx} with similarity: {similarities[idx]:.3f}")
                except Exception as e:
                    print(f"Error in embedding search: {e}")
                    # 임베딩 실패시 폴백
            
            # 임베딩이 없거나 실패한 경우 - 간단히 처음 N개 청크 반환
            if not all_relevant_chunks:
                print(f"Falling back to simple chunk selection for {doc_id}")
                for i in range(min(top_k, len(chunks))):
                    all_relevant_chunks.append({
                        "content": chunks[i],
                        "doc_name": self.documents[doc_id]["metadata"]["file_name"],
                        "similarity": 1.0 - (i * 0.1)  # 순서대로 가중치
                    })
                    print(f"Added chunk {i} (fallback)")
        
        # 유사도 기준 정렬
        all_relevant_chunks.sort(key=lambda x: x.get('similarity', 0), reverse=True)
        
        # 상위 K개 선택
        result = all_relevant_chunks[:top_k]
        print(f"Returning {len(result)} chunks")
        
        # 디버그: 첫 번째 청크 내용 일부 출력
        if result:
            print(f"First chunk preview: {result[0]['content'][:100]}...")
        
        return result
    
    def create_rag_prompt(self, query: str, doc_ids: List[str], top_k: int = 3) -> tuple:
        """RAG 프롬프트 생성 - 쿼리와 컨텍스트를 분리하여 반환"""
        print(f"Creating RAG prompt for query: '{query[:50]}...' with docs: {doc_ids}")
        
        relevant_chunks = self.search_relevant_chunks(query, doc_ids, top_k)
        
        if not relevant_chunks:
            print("No relevant chunks found - checking if documents exist")
            # 문서가 있는데 청크를 못 찾은 경우, 첫 번째 청크라도 사용
            for doc_id in doc_ids:
                if doc_id in self.document_chunks and self.document_chunks[doc_id]:
                    print(f"Using first chunk from {doc_id} as fallback")
                    relevant_chunks = [{
                        "content": self.document_chunks[doc_id][0],
                        "doc_name": self.documents[doc_id]["metadata"]["file_name"],
                        "similarity": 0.5
                    }]
                    break
            
            if not relevant_chunks:
                print("No documents or chunks available")
                return query, ""
        
        print(f"Using {len(relevant_chunks)} chunks for context")
        
        # 컨텍스트 구성
        context_parts = []
        context_parts.append("Based on the following document context, please answer the question below:")
        context_parts.append("=" * 40)
        
        for i, chunk in enumerate(relevant_chunks, 1):
            context_parts.append(f"\n[Document Reference {i} - {chunk['doc_name']}]")
            # 청크 크기 증가
            content = chunk['content'][:1000] if len(chunk['content']) > 1000 else chunk['content']
            context_parts.append(content)
            print(f"Added chunk {i} ({len(content)} chars) with similarity: {chunk.get('similarity', 0):.3f}")
        
        context_parts.append("\n" + "=" * 40)
        
        context = "\n".join(context_parts)
        enhanced_query = f"{context}\n\nQuestion: {query}\n\nAnswer based on the document context provided above:"
        
        print(f"Enhanced query length: {len(enhanced_query)} chars (original: {len(query)} chars)")
        
        return enhanced_query, context

# Initialize model and RAG system
model_id = "openai/gpt-oss-20b"
pipe = pipeline(
    "text-generation",
    model=model_id,
    torch_dtype="auto",
    device_map="auto",
)

rag_system = PDFRAGSystem()

# Global state for RAG
rag_enabled = False
selected_docs = []
top_k_chunks = 3
last_context = ""

def format_conversation_history(chat_history):
    """Format conversation history for the model"""
    messages = []
    for item in chat_history:
        role = item["role"]
        content = item["content"]
        if isinstance(content, list):
            content = content[0]["text"] if content and "text" in content[0] else str(content)
        messages.append({"role": role, "content": content})
    return messages

@spaces.GPU()
def generate_response(input_data, chat_history, max_new_tokens, system_prompt, temperature, top_p, top_k, repetition_penalty):
    """Generate response with optional RAG enhancement"""
    global last_context, rag_enabled, selected_docs, top_k_chunks
    
    # Debug logging
    print(f"RAG Enabled: {rag_enabled}")
    print(f"Selected Docs: {selected_docs}")
    print(f"Available Docs: {list(rag_system.documents.keys())}")
    
    # Apply RAG if enabled
    if rag_enabled and selected_docs:
        doc_ids = [doc.split(":")[0] for doc in selected_docs]
        enhanced_input, context = rag_system.create_rag_prompt(input_data, doc_ids, top_k_chunks)
        last_context = context
        actual_input = enhanced_input
        print(f"RAG Applied - Original: {len(input_data)} chars, Enhanced: {len(enhanced_input)} chars")
    else:
        actual_input = input_data
        last_context = ""
        print("RAG Not Applied")
    
    # Prepare messages
    new_message = {"role": "user", "content": actual_input}
    system_message = [{"role": "system", "content": system_prompt}] if system_prompt else []
    processed_history = format_conversation_history(chat_history)
    messages = system_message + processed_history + [new_message]
    
    # Setup streaming
    streamer = TextIteratorStreamer(pipe.tokenizer, skip_prompt=True, skip_special_tokens=True)
    generation_kwargs = {
        "max_new_tokens": max_new_tokens,
        "do_sample": True,
        "temperature": temperature,
        "top_p": top_p,
        "top_k": top_k,
        "repetition_penalty": repetition_penalty,
        "streamer": streamer
    }
    
    thread = Thread(target=pipe, args=(messages,), kwargs=generation_kwargs)
    thread.start()
    
    # Process streaming output
    thinking = ""
    final = ""
    started_final = False
    
    for chunk in streamer:
        if not started_final:
            if "assistantfinal" in chunk.lower():
                split_parts = re.split(r'assistantfinal', chunk, maxsplit=1)
                thinking += split_parts[0]
                final += split_parts[1]
                started_final = True
            else:
                thinking += chunk
        else:
            final += chunk
        
        clean_thinking = re.sub(r'^analysis\s*', '', thinking).strip()
        clean_final = final.strip()
        
        # Add RAG context indicator if used
        rag_indicator = ""
        if rag_enabled and selected_docs and last_context:
            rag_indicator = "<div class='rag-context'>📚 RAG Context Applied</div>\n\n"
        
        formatted = f"{rag_indicator}<details open><summary>Click to view Thinking Process</summary>\n\n{clean_thinking}\n\n</details>\n\n{clean_final}"
        yield formatted

def upload_pdf(file):
    """PDF 파일 업로드 처리"""
    if file is None:
        return (
            gr.update(value="<div class='pdf-status pdf-info'>📁 파일을 선택해주세요</div>"),
            gr.update(choices=[])
        )
    
    try:
        # 파일 해시를 ID로 사용
        with open(file.name, 'rb') as f:
            file_hash = hashlib.md5(f.read()).hexdigest()[:8]
        
        doc_id = f"doc_{file_hash}"
        
        # PDF 처리 및 저장
        result = rag_system.process_and_store_pdf(file.name, doc_id)
        
        if result["success"]:
            status_html = f"""
            <div class="pdf-status pdf-success">
                ✅ PDF 업로드 완료!<br>
                📄 {result['title']}<br>
                📑 {result['pages']} 페이지 | 🔍 {result['chunks']} 청크
            </div>
            """
            
            # 문서 목록 업데이트
            doc_choices = [f"{doc_id}: {rag_system.documents[doc_id]['metadata']['file_name']}" 
                          for doc_id in rag_system.documents.keys()]
            
            return (
                status_html,
                gr.update(choices=doc_choices, value=doc_choices)
            )
        else:
            return (
                f"<div class='pdf-status pdf-error'>❌ 오류: {result['error']}</div>",
                gr.update()
            )
            
    except Exception as e:
        return (
            f"<div class='pdf-status pdf-error'>❌ 오류: {str(e)}</div>",
            gr.update()
        )

def clear_documents():
    """문서 초기화"""
    global selected_docs
    rag_system.documents = {}
    rag_system.document_chunks = {}
    rag_system.embeddings_store = {}
    selected_docs = []
    
    return (
        gr.update(value="<div class='pdf-status pdf-info'>🗑️ 모든 문서가 삭제되었습니다</div>"),
        gr.update(choices=[], value=[])
    )

def update_rag_settings(enable, docs, k):
    """Update RAG settings"""
    global rag_enabled, selected_docs, top_k_chunks
    rag_enabled = enable
    selected_docs = docs if docs else []
    top_k_chunks = k
    
    # Debug logging
    print(f"RAG Settings Updated - Enabled: {rag_enabled}, Docs: {selected_docs}, Top-K: {top_k_chunks}")
    
    status = "✅ Enabled" if enable and docs else "⭕ Disabled"
    status_html = f"<div class='pdf-status pdf-info'>🔍 RAG: <strong>{status}</strong></div>"
    
    # Show context preview if RAG is enabled
    if enable and docs:
        preview = f"<div class='rag-context'>📚 Using {len(docs)} document(s) with {k} chunks per query</div>"
        return gr.update(value=status_html), gr.update(value=preview, visible=True)
    else:
        return gr.update(value=status_html), gr.update(value="", visible=False)

# Build the interface
with gr.Blocks(theme=gr.themes.Soft(), css=custom_css, fill_height=True) as demo:
    gr.Markdown("# 🚀 GPT-OSS-20B with PDF RAG System")
    gr.Markdown("Enhanced AI assistant with document-based context understanding")
    
    with gr.Row():
        # Left sidebar for RAG controls
        with gr.Column(scale=1):
            with gr.Group(elem_classes="main-container"):
                gr.Markdown("### 📚 Document RAG Settings")
                
                pdf_upload = gr.File(
                    label="Upload PDF",
                    file_types=[".pdf"],
                    type="filepath"
                )
                
                upload_status = gr.HTML(
                    value="<div class='pdf-status pdf-info'>📤 Upload a PDF to enable document-based answers</div>"
                )
                
                document_list = gr.CheckboxGroup(
                    choices=[],
                    label="📄 Select Documents",
                    info="Choose documents to use as context"
                )
                
                clear_btn = gr.Button("🗑️ Clear All Documents", size="sm", variant="secondary")
                
                enable_rag = gr.Checkbox(
                    label="✨ Enable RAG",
                    value=False,
                    info="Use documents for context-aware responses"
                )
                
                top_k_slider = gr.Slider(
                    minimum=1,
                    maximum=5,
                    value=3,
                    step=1,
                    label="Context Chunks",
                    info="Number of document chunks to use"
                )
                
                # RAG status display
                rag_status = gr.HTML(
                    value="<div class='pdf-status pdf-info'>🔍 RAG: <strong>Disabled</strong></div>"
                )
                
                context_preview = gr.HTML(value="", visible=False)
        
        # Right side for chat interface
        with gr.Column(scale=3):
            with gr.Group(elem_classes="main-container"):
                # Create ChatInterface with custom function
                chat_interface = gr.ChatInterface(
                    fn=generate_response,
                    additional_inputs=[
                        gr.Slider(label="Max new tokens", minimum=64, maximum=4096, step=1, value=2048),
                        gr.Textbox(
                            label="System Prompt",
                            value="You are a helpful assistant. Reasoning: medium",
                            lines=4,
                            placeholder="Change system prompt"
                        ),
                        gr.Slider(label="Temperature", minimum=0.1, maximum=2.0, step=0.1, value=0.7),
                        gr.Slider(label="Top-p", minimum=0.05, maximum=1.0, step=0.05, value=0.9),
                        gr.Slider(label="Top-k", minimum=1, maximum=100, step=1, value=50),
                        gr.Slider(label="Repetition Penalty", minimum=1.0, maximum=2.0, step=0.05, value=1.0)
                    ],
                    examples=[
                        [{"text": "Explain Newton laws clearly and concisely"}],
                        [{"text": "Write a Python function to calculate the Fibonacci sequence"}],
                        [{"text": "What are the benefits of open weight AI models"}],
                    ],
                    cache_examples=False,
                    type="messages",
                    description="""Chat with GPT-OSS-20B. Upload PDFs to enhance responses with document context. 
                    Click to view thinking process (default is on).""",
                    textbox=gr.Textbox(
                        label="Query Input",
                        placeholder="Type your prompt (RAG will be applied if enabled)"
                    ),
                    stop_btn="Stop Generation",
                    multimodal=False
                )
    
    # Event handlers
    pdf_upload.upload(
        fn=upload_pdf,
        inputs=[pdf_upload],
        outputs=[upload_status, document_list]
    )
    
    clear_btn.click(
        fn=clear_documents,
        outputs=[upload_status, document_list]
    )
    
    # Update RAG settings when changed
    enable_rag.change(
        fn=update_rag_settings,
        inputs=[enable_rag, document_list, top_k_slider],
        outputs=[rag_status, context_preview]
    )
    
    document_list.change(
        fn=update_rag_settings,
        inputs=[enable_rag, document_list, top_k_slider],
        outputs=[rag_status, context_preview]
    )
    
    top_k_slider.change(
        fn=update_rag_settings,
        inputs=[enable_rag, document_list, top_k_slider],
        outputs=[rag_status, context_preview]
    )

if __name__ == "__main__":
    demo.launch(share=True)