# app.py - v1.2 # Beschreibung: State-of-the-Art RAG-Pipeline mit Gemma-3, FAISS, # semantischem Chunking und Streaming-Antworten. # (Fix: Korrekte FAISS-Indexierung mit langchain.Document) import os import torch import gradio as gr from typing import List, Tuple, Generator, Dict from threading import Thread # ML / Transformers from sentence_transformers import SentenceTransformer from transformers import AutoProcessor, Gemma3ForConditionalGeneration, TextStreamer # Dokumentenverarbeitung & RAG from pypdf import PdfReader from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS # -------------------------------------------------------------------- # Konfiguration # -------------------------------------------------------------------- EMBED_MODEL_ID = "google/embeddinggemma-300m" LLM_MODEL_ID = "google/gemma-3-4b-it" # Globale States EMBED_MODEL: SentenceTransformer = None LLM_MODEL: Gemma3ForConditionalGeneration = None LLM_PROCESSOR: AutoProcessor = None VECTOR_STORE: FAISS = None # -------------------------------------------------------------------- # Model Loading # -------------------------------------------------------------------- def get_device() -> torch.device: """Gibt das verfügbare torch-Device zurück (CUDA, wenn verfügbar).""" if torch.cuda.is_available(): print("CUDA ist verfügbar. Nutze GPU.") return torch.device("cuda") print("CUDA nicht verfügbar. Nutze CPU.") return torch.device("cpu") def get_embedding_model() -> SentenceTransformer: """Lädt und initialisiert das Embedding-Modell.""" global EMBED_MODEL if EMBED_MODEL is None: device = "cuda" if torch.cuda.is_available() else "cpu" EMBED_MODEL = SentenceTransformer(EMBED_MODEL_ID, device=device) return EMBED_MODEL def get_llm() -> Tuple[Gemma3ForConditionalGeneration, AutoProcessor]: """Lädt und initialisiert das LLM und den zugehörigen Prozessor.""" global LLM_MODEL, LLM_PROCESSOR if LLM_MODEL is None or LLM_PROCESSOR is None: device = get_device() dtype = torch.bfloat16 if device.type == "cuda" else torch.float32 LLM_MODEL = Gemma3ForConditionalGeneration.from_pretrained( LLM_MODEL_ID, torch_dtype=dtype, device_map="auto", ).eval() LLM_PROCESSOR = AutoProcessor.from_pretrained(LLM_MODEL_ID) return LLM_MODEL, LLM_PROCESSOR # -------------------------------------------------------------------- # Datei-Handling & Chunking # -------------------------------------------------------------------- def extract_text_from_file(path: str) -> str: """Extrahiert Text aus verschiedenen Dateitypen.""" ext = os.path.splitext(path)[1].lower() if ext in [".txt", ".md", ".markdown"]: with open(path, "r", encoding="utf-8", errors="ignore") as f: return f.read() if ext == ".pdf": text_parts = [] try: reader = PdfReader(path) for page in reader.pages: page_text = page.extract_text() if page_text: text_parts.append(page_text) return "\n\n".join(text_parts) except Exception as e: print(f"Fehler beim Lesen von PDF {path}: {e}") return "" try: with open(path, "r", encoding="utf-8", errors="ignore") as f: return f.read() except Exception: return "" def get_text_splitter() -> RecursiveCharacterTextSplitter: """Erstellt einen semantisch bewussten Text-Splitter.""" return RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, add_start_index=True, ) # -------------------------------------------------------------------- # Indexing / RAG mit FAISS # -------------------------------------------------------------------- def index_files(file_paths: List[str], progress=gr.Progress(track_tqdm=True)) -> str: """ Liest Dateien, erstellt Chunks und baut/aktualisiert einen FAISS-Vektorindex. """ global VECTOR_STORE if not file_paths: return "Keine Dateien zum Indexieren ausgewählt." embed_model = get_embedding_model() text_splitter = get_text_splitter() documents: List[Document] = [] for path in progress.tqdm(file_paths, desc="1/2: Dateien verarbeiten & chunken"): if path is None: continue text = extract_text_from_file(path) if not text.strip(): continue chunks = text_splitter.split_text(text) if not chunks: continue source_name = os.path.basename(path) for chunk in chunks: doc = Document(page_content=chunk, metadata={"source": source_name}) documents.append(doc) if not documents: return "Kein Text in den Dateien gefunden, der indexiert werden konnte." progress(0.5, desc="2/2: Embeddings erstellen & FAISS Index aufbauen...") # FAISS.from_documents kümmert sich um Embedding und Indexierung new_store = FAISS.from_documents(documents, embed_model) if VECTOR_STORE is None: VECTOR_STORE = new_store else: # Fügt neue Dokumente zum bestehenden Index hinzu VECTOR_STORE.add_documents(documents) return f"Index aktualisiert: {VECTOR_STORE.index.ntotal} Chunks insgesamt." def clear_index() -> str: """Leert den Vektorindex.""" global VECTOR_STORE VECTOR_STORE = None # Garbage Collection anstoßen, um Speicher freizugeben import gc gc.collect() return "Index geleert." def retrieve_relevant_chunks(query: str, top_k: int = 5) -> List[Dict]: """Sucht die relevantesten Chunks mit FAISS.""" if VECTOR_STORE is None: return [] # similarity_search_with_score gibt Dokumente und L2-Distanz-Scores zurück results_with_scores = VECTOR_STORE.similarity_search_with_score(query, k=top_k) formatted_results = [] for doc, score in results_with_scores: formatted_results.append({ "content": doc.page_content, "source": doc.metadata.get("source", "Unbekannt"), "score": 1 - score # Konvertiert Distanz in eine Ähnlichkeits-Metrik }) return formatted_results # -------------------------------------------------------------------- # LLM-Generierung mit Streaming # -------------------------------------------------------------------- def build_rag_prompt(user_question: str, retrieved_chunks: List[Dict]) -> str: """Baut den RAG-Prompt für das LLM.""" if not retrieved_chunks: context_str = "Es wurden keine relevanten Dokumente im Kontext gefunden." else: context_parts = [] for i, ch in enumerate(retrieved_chunks, start=1): context_parts.append( f"Dokument [{i}] (Quelle: {ch['source']}, Relevanz: {ch['score']:.3f}):\n" f"\"{ch['content']}\"" ) context_str = "\n\n".join(context_parts) prompt = ( f"Du bist ein präziser, hilfreicher Assistent. Deine Aufgabe ist es, die folgende Benutzerfrage ausschließlich " f"basierend auf den unten stehenden Kontext-Dokumenten zu beantworten. " f"Wenn die Antwort nicht in den Dokumenten enthalten ist, gib klar an: 'Die Information ist in den bereitgestellten Dokumenten nicht enthalten.' " f"Antworte auf Deutsch und fasse die relevanten Informationen zusammen, anstatt die Dokumente wörtlich zu zitieren.\n\n" f"--- Kontext-Dokumente ---\n" f"{context_str}\n\n" f"--- Benutzerfrage ---\n" f"{user_question}\n\n" f"--- Deine Antwort ---\n" ) return prompt def answer_with_rag(question: str, history: List[Tuple[str, str]]) -> Generator[str, None, None]: """Führt RAG durch und generiert eine gestreamte Antwort.""" model, processor = get_llm() streamer = TextStreamer(processor, skip_prompt=True, skip_special_tokens=True) retrieved = retrieve_relevant_chunks(question, top_k=5) prompt = build_rag_prompt(question, retrieved) messages = [{"role": "user", "content": prompt}] inputs = processor.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, return_tensors="pt" ).to(model.device) generation_kwargs = dict( inputs, streamer=streamer, max_new_tokens=1024, do_sample=True, temperature=0.7, top_p=0.9, ) thread = Thread(target=model.generate, kwargs=generation_kwargs) thread.start() for new_text in streamer: yield new_text # -------------------------------------------------------------------- # Gradio UI # -------------------------------------------------------------------- def build_demo() -> gr.Blocks: with gr.Blocks(title="Gemma 3 RAG mit FAISS", theme="soft") as demo: gr.Markdown( """ # 🔍 Gemma 3 RAG v1.2 – mit FAISS & Streaming **Eine "State of the Art" RAG-Pipeline mit `google/embeddinggemma-300m` und `google/gemma-3-4b-it`** 1. Lade deine Dokumente hoch und klicke auf "Index aktualisieren". 2. Stelle deine Fragen im Chatfenster. Die Antworten werden live gestreamt. """ ) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 📁 Dokumenten-Management") file_uploader = gr.File( label="Dateien hochladen (.pdf, .txt, .md)", file_count="multiple", type="filepath", ) with gr.Row(): index_button = gr.Button("🔄 Index aktualisieren", variant="primary") clear_index_button = gr.Button("🧹 Index leeren") index_status = gr.Markdown("Index ist leer.") index_button.click( fn=index_files, inputs=file_uploader, outputs=index_status, ) clear_index_button.click( fn=clear_index, inputs=None, outputs=index_status, ) with gr.Column(scale=2): gr.Markdown("### 💬 Chat über deine Dokumente") chatbot = gr.Chatbot( label="Gemma-3 Chat", bubble_full_width=False, show_copy_button=True, height=600, ) with gr.Row(): msg_textbox = gr.Textbox( label="Deine Frage", placeholder="Stelle eine Frage zu den Dokumenten...", scale=4, autofocus=True, ) send_btn = gr.Button("Senden", variant="primary", scale=1) def chat_interface(message: str, history: list): if not message or not message.strip(): # Leere Eingaben ignorieren return history # Neue Nachricht zur History hinzufügen mit Platzhalter für Antwort history.append((message, "")) # Stream die LLM-Antwort in den Platzhalter for token in answer_with_rag(message, history): history[-1] = (message, history[-1][1] + token) yield history # Event Listeners für das Senden einer Nachricht msg_textbox.submit( fn=chat_interface, inputs=[msg_textbox, chatbot], outputs=chatbot, ).then( fn=lambda: gr.update(value=""), # Textbox leeren outputs=msg_textbox ) send_btn.click( fn=chat_interface, inputs=[msg_textbox, chatbot], outputs=chatbot ).then( fn=lambda: gr.update(value=""), # Textbox leeren outputs=msg_textbox ) return demo if __name__ == "__main__": app_demo = build_demo() # Share=True erzeugt einen öffentlichen Link, nützlich für Colab/Remote app_demo.launch(share=False)