Accueil / Articles / Envelopper Multimodal RAG dans une interface de chat Gradio

Envelopper Multimodal RAG dans une interface de chat Gradio

Refactoriser un moteur RAG multimodal en ligne de commande en une application Gradio à glisser-déposer avec suivi des progrès, citations et réponses en streaming.

1866 mots

De RAG multimodal en ligne de commande à une interface de chat Gradio

La première partie présentait une pile RAG multimodale en ligne de commande : ingestion de PDF, extraction de texte et de graphiques, réponses locales. Puissante — mais peu pratique pour les collègues non techniques. Cette suite enveloppe le même moteur dans une interface web : glisser-déposer un PDF, suivre l’avancement, discuter avec des citations.

Rappel du noyau en ligne de commande

Le moteur savait déjà comment ingérer, indexer, récupérer des fragments parents/enfants et répondre en citant les sources. L’interface ne doit pas dupliquer cette logique — elle doit l’appeler.

Refacturation : `rag_engine.py`

Répartir les fonctions d’ingestion et de requête en fonctions claires que l’interface peut appeler :

# 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

Appels de retour sur l’avancement

L’ingestion de PDF longs nécessite des mises à jour d’état dans le navigateur :

# 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

Transmettez les fonctions progress / status de Gradio afin que les utilisateurs voient le nombre de pages plutôt qu’un indicateur en chargement bloqué.

Métadonnées pour les citations

Stockez des champs sources lisible tout en effectuant le chunking :

# 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"
}

Récupération enfant-parent (bref résumé)

Les chunks enfants sont récupérés avec précision ; les fenêtres parents fournissent le contexte nécessaire à la génération. L’interface utilisateur doit afficher les mêmes sources que celles imprimées par la ligne de commande.

Chemin de requête

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
    }

Récupérez les enfants, développez les parents, construisez la requête, puis retournez la réponse ainsi que les sources.

Rendu des sources dans l’UI

# 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)

Exemple de structure d’échange :

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)

Création de `app.py` avec Gradio

Gradio propose des widgets de téléchargement, un historique de chat et une fonctionnalité de streaming avec peu de code de base.

Streaming des réponses

Préférez l’utilisation de yield pour une vitesse perçue plus élevée :

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)]

Esquisse complète de l’application

#!/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.
    """

Exécuter l’application

python app.py

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

Cloner le projet, installer les dépendances mentionnées dans la première partie, lancer app.py, télécharger un PDF, poser une question concrète, et approfondir les sources.

Conclusion

La ligne de commande a prouvé le fonctionnement du pipeline ; Gradio permet de le partager facilement. Conservez rag_engine.py comme source unique de vérité afin que les modifications de l’interface n’affectent pas la qualité des résultats obtenus.

Assurez que les démos soient reproductibles : fixez les identifiants des modèles, documentez les ports, et préférez les modes d’imitation lorsque l’environnement de test ne dispose pas de GPU.

Assurez la reproductibilité des démos : fixez les étiquettes des modèles, documentez les ports, et préférez les modes simulés lorsque le CI ne dispose pas de GPU.

Tous les articles Envoyer le brief