import os, sys, glob, json, tempfile, threading import gradio as gr from gradio_pdf import PDF from typing import Optional from gradio import Request as GradioRequest from llama_index.core import Settings, VectorStoreIndex, StorageContext, load_index_from_storage, SimpleDirectoryReader from llama_index.core.llms import ChatMessage, MessageRole from llama_index.llms.google_genai import GoogleGenAI from llama_index.core.node_parser import SentenceSplitter from llama_index.embeddings.fastembed import FastEmbedEmbedding # ----- 1) Global IP Limiter ----- IP_DATA_FILE = "user_limits.json" lock = threading.Lock() def save_ip_data(data): with open(IP_DATA_FILE, 'w') as f: json.dump(data, f, indent=4) def load_ip_data(): if not os.path.exists(IP_DATA_FILE): return {} try: with open(IP_DATA_FILE, 'r') as f: return json.load(f) except json.JSONDecodeError: return {} def get_client_ip(request: Optional[GradioRequest]) -> str: if request: x_forwarded_for = request.headers.get('x-forwarded-for') if x_forwarded_for: return x_forwarded_for.split(',')[0].strip() if request.client and request.client.host: return request.client.host return "127.0.0.1" # ----- 2) API Key setup ----- #os.environ.get("GOOGLE_API_KEY", None) key = os.environ.get("GOOGLE_API_KEY", None) if not key: print("ERROR: GOOGLE_API_KEY environment variable not set.") sys.exit(2) os.environ["GOOGLE_API_KEY"] = key print("✅ API key set!") # ----- 3) Set up Gemini Models ----- print("API Key --> Gemini Model") try: Settings.llm = GoogleGenAI(model="models/gemini-2.5-flash") Settings.embed_model = FastEmbedEmbedding(model_name = "BAAI/bge-small-en-v1.5") print("✅ Global Settings configured for Gemini.") except Exception as e: print(f"Error setting up Gemini Models : {e}") sys.exit(2) # ----- 4) Load Index Portfolio ----- print("Gemini Model --> Index Loading") try: storage_context = StorageContext.from_defaults(persist_dir = "./index_storage") index = load_index_from_storage(storage_context) print("✅ Portfolio index loaded!") except FileNotFoundError: print("Error: './index_storage' folder not found.") print("Please run 'python build_index.py' first.") sys.exit(2) except Exception as e: print(f"Error loading index: {e}") sys.exit(2) # ----- 5) Chat Engine Set-Up ----- print("Index Loading --> Chat Engine Set-Up") chat_engine = index.as_chat_engine(chat_mode = "condense_question", verbose = True) # ----- 6) Load Hobby Pictures ----- print("Chat Engine Set-Up --> Load Hobby Pictures") folder = "./pictures/" hobby_pic_list = glob.glob(os.path.join(folder, "*.jpg")) if not hobby_pic_list: print("Warning: No pictures found in './pictures/'. Hobbies tab will be empty.") else: print(f"✅ found {len(hobby_pic_list)} pictures") # ----- 7) Load About Me text and RAG Functions ----- print("Load Hobby Pictures --> Load RAG Functions") try: with open("./about_me.txt", encoding="utf-8") as f: about_me_text = f.read() print("✅ About Me text loaded!") except FileNotFoundError: print("❌ Error: './about_me.txt' not found.") about_me_text = "Welcome to my portfolio! (Error: about_me.txt not found.)" def get_rag_core(pdf_file_path): documents = SimpleDirectoryReader(input_files=[pdf_file_path], encoding="utf-8").load_data() splitter = SentenceSplitter(chunk_size=500) nodes = splitter.get_nodes_from_documents(documents) temp_index = VectorStoreIndex(nodes=nodes, show_progress=False) return temp_index.as_chat_engine( chat_mode = "condense_question", verbose = False ) SAMPLE_PDF_PATH = "./Musterdokument.pdf" if os.path.exists(SAMPLE_PDF_PATH): print("✅ Initializing RAG Engine...") default_rag_engine = get_rag_core(SAMPLE_PDF_PATH) else: default_rag_engine = None def load_pdf_and_update(file_obj): if file_obj is None: return None, gr.update(value="./Musterdokument.pdf") file_path = file_obj.name print(f"✅ PDF uploaded: {file_path}") return file_path, gr.update(value=file_path) def demo_chat_response_wrapper(message, history, uploaded_pdf_path, request: Optional[GradioRequest]): if not message.strip(): return "", history updated_history = demo_chat_response(message, history, uploaded_pdf_path, request) return "", updated_history MAX_MESSAGES = 10 rag_engine_cache = {} cache_lock = threading.Lock() def demo_chat_response(message, history, uploaded_pdf_path, request: Optional[GradioRequest] = None): ip_address = get_client_ip(request) with lock: ip_data = load_ip_data() current_count = ip_data.get(ip_address, 0) if current_count >= MAX_MESSAGES: bot_response = f"❌ **Limit erreicht:** Diese system hat bereits {MAX_MESSAGES} Abfragen verwendet." history.append({"role": "assistant", "content": bot_response}) return history ip_data[ip_address] = current_count + 1 save_ip_data(ip_data) if uploaded_pdf_path: with cache_lock: if uploaded_pdf_path not in rag_engine_cache: print(f"Building RAG engine for {uploaded_pdf_path}...") rag_engine_cache[uploaded_pdf_path] = get_rag_core(uploaded_pdf_path) MAX_CACHE_SIZE = 5 if len(rag_engine_cache) > MAX_CACHE_SIZE: # Get the first key (oldest) and delete it oldest_key = next(iter(rag_engine_cache)) del rag_engine_cache[oldest_key] print(f"Deleted old cache entry: {oldest_key}") current_engine = rag_engine_cache[uploaded_pdf_path] elif default_rag_engine: current_engine = default_rag_engine else: bot_response = "Bitte laden Sie eine PDF-Datei hoch oder stellen Sie sicher, dass die Beispiel-PDF verfügbar ist." history.append({"role": "assistant", "content": bot_response}) return history llama_history = [] for msg in history: role = MessageRole.USER if msg["role"] == "user" else MessageRole.ASSISTANT llama_history.append(ChatMessage(role=role, content=msg["content"])) try: response = current_engine.chat(message, chat_history=llama_history) bot_response = str(response) except Exception as e: bot_response = f"Error during query: {e}" history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": bot_response}) return history # ----- 8) Define chatbot function ----- print("Load About Me text --> Chatbot Functions") def portfolio_chat_response(message, history, request: Optional[GradioRequest] = None): if not message.strip(): return "", history ip_address = get_client_ip(request) with lock: ip_data = load_ip_data() current_count = ip_data.get(ip_address, 0) if current_count >= MAX_MESSAGES: bot_response = f"❌ **Limit erreicht:** Diese System hat bereits {MAX_MESSAGES} Abfragen verwendet." history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": bot_response}) return "", history ip_data[ip_address] = current_count + 1 save_ip_data(ip_data) print(f"Received message: {message}. IP: {ip_address}, Count: {current_count + 1}") llama_history = [] for msg in history: role = MessageRole.USER if msg["role"] == "user" else MessageRole.ASSISTANT llama_history.append(ChatMessage(role=role, content=msg["content"])) try: response = chat_engine.chat(message, chat_history=llama_history) bot_response = str(response) except Exception as e: print(f"Error during chat: {e}") bot_response = f"Error: {e}" history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": bot_response}) return "", history # ----- 9) Tab switching functions ----- print("Chatbot Functions --> Tab Switching Functions") def show_home_tab(): return { home_ui: gr.update(visible = True), resume_ui: gr.update(visible = False), chat_ui: gr.update(visible = False), hobbies_ui: gr.update(visible = False), demo_ui: gr.update(visible = False) } def show_resume_tab(): return { home_ui: gr.update(visible = False), resume_ui: gr.update(visible = True), chat_ui: gr.update(visible = False), hobbies_ui: gr.update(visible = False), demo_ui: gr.update(visible = False) } def show_demo_tab(): return { home_ui: gr.update(visible = False), resume_ui: gr.update(visible = False), chat_ui: gr.update(visible = False), hobbies_ui: gr.update(visible = False), demo_ui: gr.update(visible = True) } def show_chat_tab(): return { home_ui: gr.update(visible = False), resume_ui: gr.update(visible = False), chat_ui: gr.update(visible = True), hobbies_ui: gr.update(visible = False), demo_ui: gr.update(visible = False) } def show_hobbies_tab(): return { home_ui: gr.update(visible = False), resume_ui: gr.update(visible = False), chat_ui: gr.update(visible = False), hobbies_ui: gr.update(visible = True), demo_ui: gr.update(visible = False) } # ----- 10) Build Gradio UI Blocks ----- print("Tab Switching Functions --> Gradio UI Blocks") app_css = """ * { font-family: Calibri, Candara, Segoe, "Segoe UI", Optima, Arial, sans-serif !important; } #app-container { border-radius: 10px; overflow: hidden; display: flex; flex-direction: column; height: auto !important; } #title-text { text-align: center !important; width: 100% !important; } #main-row { flex: 1 1 auto; display: flex; height: auto !important; min-height: 0; gap: 45px; align-items: stretch; } #main-row > .gr-column:last-child { display: flex; flex-direction: column; min-height: 0; overflow: hidden; } #left-nav { background: linear-gradient(to bottom, #2a2a4e, #1a1a2e); padding: 10px; height: auto; overflow-y: auto; border-top-left-radius: 1px; border-bottom-left-radius: 1px; } #left-nav .gr-button { background-color: transparent !important; color: #dcdcdc !important; font-size: 18px !important; font-weight: 500; margin: 5px 0; text-align: left; padding-left: 5px !important; border: none !important; box-shadow: none !important; } #left-nav .gr-button:hover { background-color: rgba(255, 255, 255, 0.1) !important; color: #ffffff !important; } #profile-pic { max-width: 150px; max-height: 150px; margin: 10px auto; border-radius: 50%; border: 2px solid #fff; overflow: hidden; } .no-clear > div > button.clear-button { display: none !important; } .content-tab { background: transparent !important; border: none !important; box-shadow: none !important; padding: 0px !important; height: auto !important; flex: 1 1 auto; display: flex; flex-direction: column; overflow-y: auto !important; } .content-tab .gr-markdown { background: transparent !important; border: none !important; box-shadow: none !important; } .content-tab .gr-markdown h1 { display: block; width: 100%; text-align: center !important; font-size: 32px !important; font-weight: 600; margin-bottom: 20px; border-bottom: 1px solid rgba(255, 255, 255, 0.2); padding-bottom: 15px; } #home-content-body { height: auto !important; overflow-y: auto !important; } #home-content-body p, #home-content-body div { font-size: 15px !important; line-height: 1.4 !important; } #footer-container { flex-grow: 0; flex-shrink: 0; } .footer-block { background-color: #002266; color: #dcdcdc; text-align: center; padding: 7px; width: 100%; border-radius: 10px; } .footer-block p { margin: 0 0 5px 0; font-size: 16px; } .footer-block a { color: #79c0ff; font-size: 18px; font-weight: bold; text-decoration: underline; } .footer-block a:hover { color: #a9d0ff; } #chat-button-row { gap: 10px; } .chat-button, .chat-button-rag { border-radius: 8px !important; } .load-button-rag { width: auto !important; display: inline-flex !important; align-items: center !important; justify-content: center !important; white-space: nowrap !important; border-radius: 8px !important; font-size: 14px !important; } #chat-input-box textarea { height: 60px !important; } #resume-viewer-pdf { height: auto; } #chatbot-component { flex: 1 1 auto; min-height: 300px; } .demo-title-wrapper { background-color: transparent !important; margin: 0 0 10px 0 !important; padding: 0 !important; flex-shrink: 0; } .demo-title-wrapper .gr-markdown, .demo-title-wrapper h1 { background: transparent !important; border: none !important; margin: 0 !important; } #demo-content-area { background: transparent !important; border: none !important; flex: 1 1 auto; min-height: 0; overflow: hidden; } .demo-center-wrapper { padding: 0 2%; height: 100%; width: 100%; display: flex; flex-direction: column; min-height: 0; } .demo-row { padding: 10px 0; gap: 3% !important; width: 100%; flex: 1 1 auto; min-height: 0; display: flex !important; flex-direction: row !important; flex-wrap: nowrap !important; justify-content: space-between !important; background: transparent !important; } .demo-row > * { flex: 1 1 48.5% !important; min-width: 0 !important; max-width: 48.5% !important; } .demo-tile { background-color: transparent !important; border: 2px solid rgba(45, 45, 68, 0.5) !important; border-radius: 12px !important; padding: 15px !important; box-shadow: none !important; flex: 1 1 48.5% !important; min-height: 0; max-height: 100% !important; min-width: 0; max-width: 48.5% !important; display: flex; flex-direction: column; overflow: hidden; } .demo-tile h2 { text-align: center; font-size: 26px; margin: 0 0 10px 0; padding: 10px 0; flex-shrink: 0; } #demo-pdf-viewer { flex: 1 1 auto; min-height: 0; overflow: hidden; } #demo-chatbot { flex: 1 1 auto; min-height: 400px !important; max-height: 100% !important; display: flex !important; flex-direction: column !important; overflow: auto; } #demo-chatbot > div { flex: 1 1 auto !important; min-height: 0 !important; } #demo-chat-column { flex: 1 1 auto !important; min-height: 400px !important; display: flex !important; flex-direction: column !important; } #demo-chat-column > div:last-child { margin-top: 10px !important; } .rag-row-note { align-items: center; } .rag-row-note > div:first-child { flex-grow: 0; flex-shrink: 1; } #pdf-upload-box { padding: 0; max-height: 150px; overflow: hidden; flex-shrink: 0; } #pdf-upload-box .drop-file-zone { height: 100px !important; } #hobbies-gallery-component { min-height: 60vh !important; height: auto !important; overflow-y: auto !important; } @media (max-width: 768px) { #main-row { flex-direction: column; height: auto; } .demo-row { flex-direction: row !important; flex-wrap: nowrap !important; gap: 2% !important; } .demo-tile { min-width: 49% !important; max-width: 49% !important; flex: 1 1 49% !important; } .rag-row-note { flex-wrap: wrap; } .rag-row-note > div:first-child { flex-basis: 70%; } .rag-row-note .note-text { flex-basis: 30%; font-size: 10px !important; white-space: normal !important; } .load-button-rag { padding: 5px 8px !important; font-size: 14px !important; white-space: normal !important; } } """ with gr.Blocks(theme = gr.themes.Soft(), css = app_css, title = "Suhas’ KI-Portfolio", fill_height = True, fill_width = True) as demo: with gr.Column(elem_id = "app-container"): with gr.Row(equal_height = False, elem_id = "main-row"): with gr.Column(scale = 0, elem_id = "left-nav"): gr.Image( "Foto_SuhasKamuni.jpg", elem_id = "profile-pic", interactive = False, show_label = False, show_download_button = False, elem_classes = "no-clear" ) gr.Markdown( """
Data Scientist/Analyst
""", elem_id = "title-text" ) home_btn = gr.Button("🏠 Startseite", variant = "transparent") resume_btn = gr.Button("📄 Lebenslauf", variant = "transparent") chat_btn = gr.Button("🤖 Chat Assistant", variant = "transparent") demo_btn = gr.Button("🧠 Projekt-Demo", variant = "transparent") hobbies_btn = gr.Button("🎨 Hobbys", variant = "transparent") # --- Right Content Column --- with gr.Column(scale = 7): with gr.Group(visible = True, elem_classes="content-tab", elem_id="home-content-body") as home_ui: gr.Markdown("RAG (Retrieval-Augmented Generation) kombiniert die Dokumentensuche mit KI-Chat-Unterstützung. Das System ruft relevante Abschnitte ab und generiert präzise Antworten auf der Grundlage des Dokumentinhalts. Obwohl PDF-Dokumente heutzutage in die meisten großen LLMs hochgeladen werden können, bietet RAG eine privatere Option, da die Daten gelöscht werden, sobald der Tab geschlossen wird.
""", elem_classes = "demo-description" ) with gr.Column(elem_classes = "demo-center-wrapper", elem_id = "demo-content-area"): with gr.Row(equal_height = True, elem_classes = "demo-row"): with gr.Group(elem_classes = "demo-tile"): gr.Markdown("* Hinweis: PDF-Dateien werden nirgendwo gespeichert, sondern nach Beendigung der Sitzung gelöscht.
", elem_classes = "note-text" ) demo_pdf_viewer = PDF( "./Musterdokument.pdf", label = None, visible = True, elem_id = "demo-pdf-viewer" ) uploaded_file_path = gr.State(value=None) with gr.Group(elem_classes="demo-tile"): gr.Markdown("