Startseite / Artikel / Multimodales RAG in eine Gradio-Chat-Oberfläche einbetten

Multimodales RAG in eine Gradio-Chat-Oberfläche einbetten

Refaktorieren Sie einen CLI-basierten multimodalen RAG-Engine in eine mit Drag-and-Drop-funktion arbeitende Gradio-App mit Fortschrittsanzeige, Zitaten und streambasierten Antworten.

1866 Wörter

Von einer CLI-basierten multimodalen RAG-Lösung zu einer Gradio-Chat-Oberfläche

Teil 1 stellte einen Befehlszeilen-basierten multimodalen RAG-Stack vor: Eingabe von PDFs, Extraktion von Text und Diagrammen sowie lokale Antworten. Mächtig – aber für nicht-technische Kollegen unhandlich. In diesem Folgeartikel wird derselbe Motor in eine Web-Oberfläche eingebettet: Man kann PDFs per Drag-and-Drop einfügen, den Fortschritt verfolgen und mit Zitaten chatten.

Zusammenfassung des CLI-Kerns

Der Motor wusste bereits, wie man Daten einliest, indiziert, Eltern- und Kind-Chunks abruft sowie mit Quellen antwortet. Die Benutzeroberfläche sollte diese Logik nicht duplizieren – sie sollte sie lediglich aufrufen.

Refaktorierung: `rag_engine.py`

Die Funktionen für die Eingabe und Abfrage werden in klare Funktionen umgewandelt, die von der Benutzeroberfläche aufgerufen werden können:

# rag_engine.py - Core functions

def ingest_pdf(pdf_path: str, progress=None, status=None):
    """
    Parse PDF, extract text, process images, and store in Chroma.

    Args:
        pdf_path: Path to the PDF file
        progress: Gradio progress object (optional)
        status: Callback for status updates (optional)
    """
    # ... same logic as Part 1, but with progress callbacks

def query_rag(question: str) -> dict:
    """
    Retrieve relevant context and generate an answer.

    Args:
        question: User's question

    Returns:
        dict: {"answer": str, "sources": list[str]}
    """
    # ... returns answer with source parent IDs

def clear_database():
    """Reset the database and reinitialize collections."""
    # ... handles collection re-creation

def get_ingested_documents():
    """Return list of ingested PDFs."""
    # ... utility for status display

Fortschritts-Callback-Funktionen

Beim Einlesen langer PDFs sind Statusaktualisierungen im Browser erforderlich:

# In rag_engine.py
def ingest_pdf(pdf_path: str, progress=None, status=None):
    if status:
        status(f"📄 Processing page {page_num + 1}/{total_pages}...")
    if progress:
        progress((page_num + 1) / total_pages, desc=f"Processing page {page_num + 1}/{total_pages}")
    # ... rest of the logic

Übergeben Sie die Gradio-Funktionen progress / status, damit die Benutzer die Seitenanzahl statt eines feststeckenden Spinners sehen.

Metadaten für Zitate

Speichern Sie lesbare Quellfelder während des Chunking-Vorgangs:

# During ingestion in rag_engine.py
metadata = {
    "source": os.path.basename(pdf_path),
    "page": page_num + 1,
    "parent_id": f"Page_{page_num + 1}",
    "section": "Unknown",  # Optional: extract from heading
    "chunk_index": chunk_idx,
    "type": "hybrid"
}

Aufruf von Kind- und Elternobjekten (kurze Zusammenfassung)

Kind-Chunkes werden präzise abgerufen; Elternfenster liefern den Kontext für die Erstellung. Die Benutzeroberfläche sollte dieselben Quellen anzeigen wie die CLI.

Abfragepfad

def query_rag(question: str) -> dict:
    # 1. Retrieve top matching child chunks
    results = child_collection.query(
        query_texts=[question],
        n_results=3
    )

    # 2. Fetch the full Parent contexts with metadata
    parent_ids = list(set([m["parent_ref"] for m in results["metadatas"][0]]))
    parent_results = parent_collection.get(ids=parent_ids)

    # 3. Build full context from parent documents
    full_context = "\n\n---\n\n".join(parent_results["documents"])

    # 4. Get metadata for source display
    source_metadata = []
    for meta in parent_results["metadatas"]:
        page = meta.get("page", "Unknown")
        section = meta.get("section", "Section")
        source_metadata.append({"page": page, "section": section})

    # 5. Build prompt and generate answer
    prompt = f"""
    You are a financial research assistant. Answer the question based strictly on the context below.
    If the context contains chart summaries or tables, use those numbers specifically.
    If you cannot answer from the context, say "I don't have that information."

    Context:
    {full_context}

    Question: {question}
    Answer:
    """

    response = ollama.chat(
        model=TEXT_MODEL,
        messages=[{"role": "user", "content": prompt}]
    )

    return {
        "answer": response["message"]["content"],
        "sources": source_metadata  # Now contains page and section info
    }

Erhalten Sie die Kind-Objekte, erweitern Sie die Elternobjekte, erstellen Sie die Anfrage und geben Sie Antwort + Quellen zurück.

Ausgabe der Quellen in der Benutzeroberfläche

# In app.py
if result["sources"]:
    source_text = "\n\n📚 **Sources:** "
    sources_list = []
    for source in result["sources"][:3]:
        # Assuming source contains metadata
        page = source.get("page", "Unknown")
        section = source.get("section", "Section")
        sources_list.append(f"Page {page} ({section})")
    answer += source_text + ", ".join(sources_list)

if result["sources"]:
    source_text = "\n\n **Sources:** "
    sources_list = []
    for source in result["sources"][:3]:
        page = source.get("page", "Unknown")
        section = source.get("section", "Section")
        sources_list.append(f"Page {page} ({section})")
    answer += source_text + ", ".join(sources_list)

Beispiel für den Austausch:

User: What was the total revenue shown in the financial statements?

The total revenue shown in the financial statements is €180,462 for the year ended 31 December 2025 and €160,465 for the year ended 31 December 2024.

📚 Sources: Page 35 (Section 1), Page 79 (Section 2), Page 159 (Section 3)

User: What was the total revenue shown in the financial statements?

The total revenue shown in the financial statements is €180,462 for the year ended 31 December 2025 and €160,465 for the year ended 31 December 2024.

📚 Sources: Page 35 (Section 1), Page 79 (Section 2), Page 159 (Section 3)

Erstellung von `app.py` mit Gradio

Gradio bietet Upload-Widgets, Chatverlauf sowie Streaming-Funktionen mit wenig Vorlagencode.

Streaming von Antworten

Verwenden Sie lieber die Methode yield für eine schnellere Wahrnehmung der Geschwindigkeit:

def chat_response(message, history):
    # ... retrieve context ...

    # Instead of returning, use yield to stream tokens
    full_response = ""
    for chunk in ollama.chat(model=TEXT_MODEL, messages=[...], stream=True):
        full_response += chunk["message"]["content"]
        yield history + [("user", message), ("assistant", full_response)]

Vollständiger Entwurf der App

#!/usr/bin/env python3
"""
Multimodal RAG Gradio UI
Run with: python app.py
"""

import os
import shutil
from pathlib import Path

import gradio as gr

# Import the core engine
from rag_engine import ingest_pdf, query_rag, clear_database, get_ingested_documents

# ---------- CONFIG ----------
UPLOAD_DIR = Path("./data")
UPLOAD_DIR.mkdir(exist_ok=True)

# ---------- UI FUNCTIONS ----------
def process_upload(file_obj, progress=gr.Progress()):
    """Handle PDF upload and ingestion with progress bar."""
    if file_obj is None:
        return " Please upload a PDF file first."

    pdf_path = UPLOAD_DIR / os.path.basename(file_obj.name)
    shutil.copy(file_obj.name, pdf_path)

    try:
        result = ingest_pdf(str(pdf_path), progress=progress)
        return f"{result}\n\n📄 File saved to: {pdf_path}"
    except Exception as e:
        return f" Error during ingestion: {str(e)}"

def chat_response(message, history):
    """
    Handle user questions and return responses.

    Note: This uses synchronous return. For streaming responses,
    consider using yield with ollama's stream=True parameter.
    """
    if not message or not message.strip():
        return history

    docs = get_ingested_documents()
    if not docs:
        history.append({"role": "user", "content": message})
        history.append({"role": "assistant", "content": " No documents ingested. Please upload and ingest a PDF first."})
        return history

    result = query_rag(message)
    answer = result["answer"]

    # Display readable sources with page and section info
    if result["sources"]:
        source_text = "\n\n **Sources:** "
        sources_list = []
        for source in result["sources"][:3]:
            page = source.get("page", "Unknown")
            section = source.get("section", "Section")
            sources_list.append(f"Page {page} ({section})")
        answer += source_text + ", ".join(sources_list)

    history.append({"role": "user", "content": message})
    history.append({"role": "assistant", "content": answer})
    return history

def reset_database():
    """Clear the database, chat history, and reset file upload."""
    result = clear_database()
    return result, [], gr.update(value=None)

def get_status():
    """Get current system status."""
    docs = get_ingested_documents()
    if docs:
        return f" {len(docs)} document(s) ingested: {', '.join(docs)}"
    return " No documents ingested. Upload and ingest a PDF to get started."

# ---------- BUILD UI ----------
def create_ui():
    with gr.Blocks(title="Multimodal RAG Assistant") as demo:
        gr.Markdown("""
        # Zero-Cost Local Multimodal RAG

        A privacy-first assistant that answers questions from complex PDFs containing text, tables, and charts.

        **How it works:**
        1. Upload a PDF and click **Ingest**
        2. Wait for processing (charts will be analyzed by the VLM)
        3. Ask questions about the document content

        **100% local** — No data ever leaves your machine.

        **Sources** shown in answers include page numbers and sections from your PDF.
        """)

        # Status Bar
        with gr.Row():
            status_bar = gr.Textbox(value=get_status(), label=" Status", interactive=False, scale=3)
            refresh_btn = gr.Button(" Refresh", size="sm", scale=0)
            clear_db_btn = gr.Button(" Clear Database", size="sm", variant="stop", scale=0)

        # Upload & Chat
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### Upload & Ingest")
                file_upload = gr.File(label="Upload PDF", file_types=[".pdf"], height=100)
                ingest_btn = gr.Button(" Ingest PDF", variant="primary", size="lg")
                upload_status = gr.Textbox(label="Upload Status", interactive=False, lines=3)

            with gr.Column(scale=2):
                gr.Markdown("### 💬 Ask Questions")
                chatbot = gr.Chatbot(label="Chat", height=400, avatar_images=(None, "🤖"))
                with gr.Row():
                    msg = gr.Textbox(label="Your question", placeholder="e.g., What was the total revenue?", scale=4, container=False)
                    send_btn = gr.Button("Send", variant="primary", scale=0)
                clear_chat_btn = gr.Button(" Clear Chat", size="sm")

        # Event Handlers
        ingest_btn.click(fn=process_upload, inputs=[file_upload], outputs=[upload_status]).then(
            fn=get_status, inputs=[], outputs=[status_bar])
        msg.submit(fn=chat_response, inputs=[msg, chatbot], outputs=[chatbot]).then(fn=lambda: "", outputs=[msg])
        send_btn.click(fn=chat_response, inputs=[msg, chatbot], outputs=[chatbot]).then(fn=lambda: "", outputs=[msg])
        clear_chat_btn.click(fn=lambda: [], outputs=[chatbot])
        clear_db_btn.click(fn=reset_database, inputs=[], outputs=[upload_status, chatbot, file_upload]).then(
            fn=get_status, inputs=[], outputs=[status_bar])
        refresh_btn.click(fn=get_status, inputs=[], outputs=[status_bar])
        demo.load(fn=get_status, inputs=[], outputs=[status_bar])

    return demo

if __name__ == "__main__":
    print(" Starting Gradio UI...")
    print(" Opening at: http://127.0.0.1:7860")
    demo = create_ui()
    demo.launch(server_name="127.0.0.1", server_port=7860, share=False, theme=gr.themes.Soft(), css="footer {visibility: hidden}")

def chat_response(message, history):
    """
    Note: This uses synchronous return. For streaming responses,
    consider using yield with ollama's stream=True parameter.
    """

Führen Sie sie aus

python app.py

git clone https://github.com/froilan-sia/m1_multimodal_rag.git
cd m1_multimodal_rag
python app.py

Klonen Sie die Dateien, installieren Sie die Abhängigkeiten aus Teil 1, starten Sie app.py, laden Sie ein PDF hoch, stellen Sie eine konkrete Frage und erweitern Sie die Quellen.

Fazit

Die CLI hat den Ablauf überprüft; Gradio macht ihn teilenbar. Behalten Sie rag_engine.py als einzige Quelle der Wahrheit, damit die UI-Entwicklung niemals die Qualität der Informationsabrufung beeinträchtigt.

Sorgen Sie dafür, dass die Demos reproduzierbar bleiben: Fixieren Sie die Modelltags, dokumentieren Sie die Ports und bevorzugen Sie Mock-Modi, wenn der CI keine GPU hat.

Sicherstellen Sie, dass Demos reproduzierbar bleiben: Festschreiben Sie die Modelle-Tags, dokumentieren Sie die Ports und bevorzugen Sie Mock-Modi, wenn der CI keine GPU hat.

Sicherstellen Sie die Wiederholbarkeit von Demos: Festschreiben Sie die Modellebenstags, dokumentieren Sie die Ports und bevorzugen Sie Mock-Modi, wenn der CI keine GPU hat.

Sichern Sie die Wiederholbarkeit von Demos: Festschreiben Sie die Modellebenen, dokumentieren Sie die Ports und bevorzugen Sie Mock-Modi, wenn der CI keine GPU hat.

Alle Artikel Brief senden