RAG Multimodal : Bilder, PDFs und mehr als Text

Das traditionelle RAG beschränkt sich auf Text, aber Unternehmenswissen liegt in vielen Formen vor: PDFs mit Grafiken, Produktbilder, technische Schaubilder, PowerPoint-Präsentationen. Das multimodale RAG erweitert die Fähigkeiten Ihres Assistenten, all diese Formate zu verstehen und zu nutzen.

Warum multimodal?

Die Grenzen des text-only RAG

In einer typischen Unternehmensdokumentation:

Ergebnis: Ein text-only RAG kann 40-50% der relevanten Informationen verpassen.

Multimodale Anwendungsfälle

• E-Commerce: Visuelle Produktsuche, Fragen zu Bildern
• Technische Dokumentation: Technische Dokumentation, Schaubilder, Architekturdiagramme
• Support: Screenshots von Benutzerfehlern
• Schulung: Präsentationsfolien, Infografiken
• Recht: Gescannte Dokumente, Unterschriften

Multimodale Architektur

┌─────────────────────────────────────────────────────────────┐
│                    SOURCES MULTIMODALES                      │
├──────────┬──────────┬──────────┬──────────┬────────────────┤
│  Text    │   PDF    │  Bilder  │  Folien  │    Video       │
│  .md/.txt│  .pdf    │ .jpg/.png│  .pptx   │   .mp4         │
└────┬─────┴────┬─────┴────┬─────┴────┬─────┴───────┬────────┘
     │          │          │          │             │
     │    ┌─────▼─────┐    │    ┌─────▼─────┐      │
     │    │   PDF     │    │    │  Folien-  │      │
     │    │ Extractor │    │    │ Extractor │      │
     │    └─────┬─────┘    │    └─────┬─────┘      │
     │          │          │          │             │
     └──────────┴──────────┼──────────┴─────────────┘
                           │
              ┌────────────▼────────────┐
              │    Vision Encoder       │
              │  (CLIP, GPT-4V, etc.)   │
              └────────────┬────────────┘
                           │
              ┌────────────▼────────────┐
              │   Multimodal Index      │
              │  (Text + Image embeds)  │
              └────────────┬────────────┘
                           │
              ┌────────────▼────────────┐
              │   Multimodal LLM        │
              │  (GPT-4V, Claude 3)     │
              └─────────────────────────┘

Erweiterte PDF-Extraktion

Extraktion mit Erhalt der Struktur

DEVELOPERpython
import fitz  # PyMuPDF
from dataclasses import dataclass
from typing import List, Optional
import base64

@dataclass
class PDFElement:
    type: str  # "text", "table", "image", "heading"
    content: str
    page: int
    bbox: tuple  # (x0, y0, x1, y1)
    metadata: dict = None

class AdvancedPDFExtractor:
    def __init__(self, vision_model=None):
        self.vision = vision_model

    def extract(self, pdf_path: str) -> List[PDFElement]:
        """
        Extrahiert alle Elemente eines PDFs unter Erhalt ihrer Struktur
        """
        doc = fitz.open(pdf_path)
        elements = []

        for page_num in range(len(doc)):
            page = doc[page_num]

            # Text mit Struktur extrahieren
            text_elements = self._extract_text_blocks(page, page_num)
            elements.extend(text_elements)

            # Tabellen extrahieren
            tables = self._extract_tables(page, page_num)
            elements.extend(tables)

            # Bilder extrahieren
            images = self._extract_images(page, page_num)
            elements.extend(images)

        doc.close()
        return elements

    def _extract_text_blocks(self, page, page_num: int) -> List[PDFElement]:
        """
        Extrahiert Textblöcke mit Erkennung von Überschriften
        """
        elements = []
        blocks = page.get_text("dict")["blocks"]

        for block in blocks:
            if "lines" not in block:
                continue

            text_parts = []
            max_font_size = 0

            for line in block["lines"]:
                for span in line["spans"]:
                    text_parts.append(span["text"])
                    max_font_size = max(max_font_size, span["size"])

            text = " ".join(text_parts).strip()
            if not text:
                continue

            # Erkennen, ob es sich um eine Überschrift anhand der Schriftgröße handelt
            is_heading = max_font_size > 14

            elements.append(PDFElement(
                type="heading" if is_heading else "text",
                content=text,
                page=page_num,
                bbox=block["bbox"],
                metadata={"font_size": max_font_size}
            ))

        return elements

    def _extract_tables(self, page, page_num: int) -> List[PDFElement]:
        """
        Extrahiert Tabellen mit Strukturerkennung
        """
        elements = []

        # Tabellen-Erkennung von PyMuPDF verwenden
        tables = page.find_tables()

        for table in tables:
            # In Markdown für das RAG konvertieren
            markdown = self._table_to_markdown(table)

            elements.append(PDFElement(
                type="table",
                content=markdown,
                page=page_num,
                bbox=table.bbox,
                metadata={
                    "rows": len(table.cells),
                    "cols": len(table.cells[0]) if table.cells else 0
                }
            ))

        return elements

    def _table_to_markdown(self, table) -> str:
        """
        Konvertiert eine Tabelle in das Markdown-Format
        """
        rows = []
        for row_idx, row in enumerate(table.extract()):
            cells = [str(cell or "").strip() for cell in row]
            rows.append("| " + " | ".join(cells) + " |")

            # Trennzeile nach der Kopfzeile hinzufügen
            if row_idx == 0:
                separator = "|" + "|".join(["---"] * len(cells)) + "|"
                rows.append(separator)

        return "\n".join(rows)

    def _extract_images(self, page, page_num: int) -> List[PDFElement]:
        """
        Extrahiert Bilder und generiert Beschreibungen
        """
        elements = []
        images = page.get_images()

        for img_idx, img in enumerate(images):
            xref = img[0]
            base_image = page.parent.extract_image(xref)

            if base_image:
                image_data = base_image["image"]
                image_b64 = base64.b64encode(image_data).decode()

                # Eine Beschreibung mit dem Vision-Modell erzeugen
                description = ""
                if self.vision:
                    description = self._describe_image(image_b64)

                elements.append(PDFElement(
                    type="image",
                    content=description,
                    page=page_num,
                    bbox=img[1:5] if len(img) > 4 else (0, 0, 0, 0),
                    metadata={
                        "image_b64": image_b64,
                        "format": base_image.get("ext", "unknown"),
                        "width": base_image.get("width"),
                        "height": base_image.get("height")
                    }
                ))

        return elements

    async def _describe_image(self, image_b64: str) -> str:
        """
        Generiert eine Textbeschreibung des Bildes
        """
        prompt = """
        Beschreibe dieses Bild detailliert, damit es über die Textsuche gefunden werden kann.
        Enthalte:
        - Den Bildtyp (Foto, Schaubild, Grafik, Screenshot)
        - Die sichtbaren Hauptelemente
        - Sichtbaren Text, falls vorhanden
        - Den wahrscheinlichen Kontext
        """

        return await self.vision.analyze_image(image_b64, prompt)

OCR für gescannte Dokumente

DEVELOPERpython
import pytesseract
from PIL import Image
import cv2
import numpy as np

class OCRExtractor:
    def __init__(self, languages: list = ["fra", "eng"]):
        self.languages = "+".join(languages)

    def extract_from_image(self, image_path: str) -> dict:
        """
        Extrahiert Text aus einem Bild mittels OCR
        """
        # Vorverarbeitung zur Verbesserung der OCR
        image = cv2.imread(image_path)
        processed = self._preprocess(image)

        # OCR mit Tesseract
        text = pytesseract.image_to_string(
            processed,
            lang=self.languages,
            config='--psm 1'  # Automatic page segmentation
        )

        # Ebenfalls die Koordinaten abrufen
        data = pytesseract.image_to_data(
            processed,
            lang=self.languages,
            output_type=pytesseract.Output.DICT
        )

        return {
            "text": text.strip(),
            "words": self._extract_words_with_positions(data),
            "confidence": self._average_confidence(data)
        }

    def _preprocess(self, image: np.ndarray) -> np.ndarray:
        """
        Vorverarbeitung des Bildes zur Verbesserung der OCR
        """
        # In Graustufen konvertieren
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

        # Rauschunterdrückung
        denoised = cv2.fastNlMeansDenoising(gray)

        # Adaptive Binarisierung
        binary = cv2.adaptiveThreshold(
            denoised, 255,
            cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
            cv2.THRESH_BINARY, 11, 2
        )

        # Schräglagenkorrektur
        corrected = self._deskew(binary)

        return corrected

    def _deskew(self, image: np.ndarray) -> np.ndarray:
        """
        Korrigiert die Schräglage des Dokuments
        """
        coords = np.column_stack(np.where(image > 0))
        angle = cv2.minAreaRect(coords)[-1]

        if angle < -45:
            angle = -(90 + angle)
        else:
            angle = -angle

        (h, w) = image.shape[:2]
        center = (w // 2, h // 2)
        M = cv2.getRotationMatrix2D(center, angle, 1.0)
        rotated = cv2.warpAffine(
            image, M, (w, h),
            flags=cv2.INTER_CUBIC,
            borderMode=cv2.BORDER_REPLICATE
        )

        return rotated

    def _average_confidence(self, data: dict) -> float:
        """
        Berechnet die durchschnittliche OCR-Konfidenz
        """
        confidences = [c for c in data["conf"] if c > 0]
        return sum(confidences) / len(confidences) if confidences else 0

Multimodale Embeddings

CLIP für Bilder und Text

DEVELOPERpython
from transformers import CLIPProcessor, CLIPModel
import torch
from PIL import Image

class CLIPEmbedder:
    def __init__(self, model_name: str = "openai/clip-vit-large-patch14"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = CLIPModel.from_pretrained(model_name).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(model_name)

    def embed_image(self, image_path: str) -> np.ndarray:
        """
        Generiert ein Embedding für ein Bild
        """
        image = Image.open(image_path)
        inputs = self.processor(images=image, return_tensors="pt").to(self.device)

        with torch.no_grad():
            image_features = self.model.get_image_features(**inputs)

        return image_features.cpu().numpy().flatten()

    def embed_text(self, text: str) -> np.ndarray:
        """
        Generiert ein Embedding für Text
        """
        inputs = self.processor(text=text, return_tensors="pt", padding=True).to(self.device)

        with torch.no_grad():
            text_features = self.model.get_text_features(**inputs)

        return text_features.cpu().numpy().flatten()

    def similarity(self, image_path: str, text: str) -> float:
        """
        Berechnet die Ähnlichkeit zwischen einem Bild und einem Text
        """
        image_emb = self.embed_image(image_path)
        text_emb = self.embed_text(text)

        # Normalisieren und Skalarprodukt berechnen
        image_emb = image_emb / np.linalg.norm(image_emb)
        text_emb = text_emb / np.linalg.norm(text_emb)

        return float(np.dot(image_emb, text_emb))

Multimodaler Index mit Qdrant

DEVELOPERpython
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

class MultimodalIndex:
    def __init__(self, text_embedder, image_embedder, qdrant_url: str = "localhost"):
        self.text_embedder = text_embedder
        self.image_embedder = image_embedder
        self.client = QdrantClient(host=qdrant_url, port=6333)

    def create_collection(self, name: str):
        """
        Erstellt eine Collection mit Text- UND Bild-Vektoren
        """
        self.client.create_collection(
            collection_name=name,
            vectors_config={
                "text": VectorParams(size=768, distance=Distance.COSINE),
                "image": VectorParams(size=768, distance=Distance.COSINE),
            }
        )

    def index_document(self, doc_id: str, content: dict, collection: str):
        """
        Indexiert ein multimodales Dokument
        """
        vectors = {}

        # Text-Embedding falls vorhanden
        if content.get("text"):
            vectors["text"] = self.text_embedder.encode(content["text"]).tolist()

        # Image-Embedding falls vorhanden
        if content.get("image_path"):
            vectors["image"] = self.image_embedder.embed_image(content["image_path"]).tolist()

        point = PointStruct(
            id=hash(doc_id) % (2**63),
            vector=vectors,
            payload={
                "doc_id": doc_id,
                "text": content.get("text", ""),
                "image_path": content.get("image_path"),
                "metadata": content.get("metadata", {})
            }
        )

        self.client.upsert(collection_name=collection, points=[point])

    def search(
        self,
        query: str,
        collection: str,
        query_image_path: str = None,
        top_k: int = 10
    ) -> list:
        """
        Multimodale Suche: Text und/oder Bild
        """
        # Anfragevektoren vorbereiten
        query_vectors = []

        # Textsuche
        text_vector = self.text_embedder.encode(query).tolist()
        query_vectors.append(("text", text_vector))

        # Visuelle Suche, falls Bild bereitgestellt
        if query_image_path:
            image_vector = self.image_embedder.embed_image(query_image_path).tolist()
            query_vectors.append(("image", image_vector))

        # Suchen ausführen und zusammenführen
        all_results = []
        for vector_name, vector in query_vectors:
            results = self.client.search(
                collection_name=collection,
                query_vector=(vector_name, vector),
                limit=top_k
            )
            all_results.extend(results)

        # Deduplizieren und neu bewerten
        return self._merge_results(all_results, top_k)

    def _merge_results(self, results: list, top_k: int) -> list:
        """
        Führt die Ergebnisse der verschiedenen Modalitäten zusammen
        """
        scores = {}
        docs = {}

        for result in results:
            doc_id = result.payload["doc_id"]
            if doc_id not in scores:
                scores[doc_id] = 0
                docs[doc_id] = result.payload

            # RRF-Fusion
            scores[doc_id] += result.score

        # Nach zusammengeführtem Score sortieren
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)

        return [
            {"doc_id": doc_id, "score": score, **docs[doc_id]}
            for doc_id, score in sorted_docs[:top_k]
        ]

Multimodale Generierung mit GPT-4V / Claude 3

Multimodale Antwort-Pipeline

DEVELOPERpython
from openai import OpenAI
import base64

class MultimodalRAG:
    def __init__(self, index, llm_client=None):
        self.index = index
        self.client = llm_client or OpenAI()

    async def query(
        self,
        text_query: str,
        image_query_path: str = None,
        collection: str = "multimodal_kb"
    ) -> dict:
        """
        Multimodale RAG-Anfrage
        """
        # 1. Multimodale Suche
        results = self.index.search(
            query=text_query,
            collection=collection,
            query_image_path=image_query_path,
            top_k=5
        )

        # 2. Multimodalen Kontext vorbereiten
        context_parts = []
        images_for_llm = []

        for result in results:
            if result.get("text"):
                context_parts.append(f"Document: {result['text']}")

            if result.get("image_path"):
                # Bild für das LLM laden
                with open(result["image_path"], "rb") as f:
                    img_b64 = base64.b64encode(f.read()).decode()
                    images_for_llm.append({
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{img_b64}"
                        }
                    })

        # 3. Antwort mit dem multimodalen LLM generieren
        response = await self._generate_response(
            query=text_query,
            context="\n\n".join(context_parts),
            images=images_for_llm,
            query_image=image_query_path
        )

        return {
            "answer": response,
            "sources": results,
            "has_visual_context": len(images_for_llm) > 0
        }

    async def _generate_response(
        self,
        query: str,
        context: str,
        images: list,
        query_image: str = None
    ) -> str:
        """
        Generiert eine Antwort mithilfe eines multimodalen LLM
        """
        messages = [
            {
                "role": "system",
                "content": """Du bist ein multimodaler RAG-Assistent. Du hast Zugriff auf Textdokumente und Bilder.

Regeln:
1. Stütze deine Antworten auf die bereitgestellten Dokumente UND Bilder
2. Beschreibe, was du auf den Bildern siehst, falls relevant
3. Wenn der Nutzer ein Bild sendet, analysiere es zur Beantwortung
4. Zitiere deine Quellen (Dokumente oder Bilder)"""
            }
        ]

        # Benutzer-Nachricht aufbauen
        user_content = []

        # Anfragebild hinzufügen, falls vorhanden
        if query_image:
            with open(query_image, "rb") as f:
                img_b64 = base64.b64encode(f.read()).decode()
            user_content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}
            })

        # Kontextbilder hinzufügen
        user_content.extend(images)

        # Text hinzufügen
        user_content.append({
            "type": "text",
            "text": f"""Dokumentkontext:
{context}

Frage: {query}

Antworte auf Basis des Kontexts und der bereitgestellten Bilder."""
        })

        messages.append({"role": "user", "content": user_content})

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            max_tokens=2000
        )

        return response.choices[0].message.content

Spezialisierte Anwendungsfälle

Visuelle Produktsuche

DEVELOPERpython
class VisualProductSearch:
    def __init__(self, product_index, clip_embedder):
        self.index = product_index
        self.clip = clip_embedder

    async def find_similar_products(
        self,
        image_path: str,
        top_k: int = 10,
        filters: dict = None
    ) -> list:
        """
        Findet Produkte, die einem Bild ähnlich sind
        """
        # Embedding des Anfragebilds
        query_embedding = self.clip.embed_image(image_path)

        # Suche im Produkt-Index
        results = self.index.search(
            query_vector=query_embedding,
            top_k=top_k,
            filters=filters
        )

        return results

    async def answer_product_question(
        self,
        product_image_path: str,
        question: str
    ) -> str:
        """
        Beantwortet eine Frage zu einem Produkt anhand seines Bildes
        """
        # Produkt finden
        similar = await self.find_similar_products(product_image_path, top_k=1)

        if not similar:
            return "Ich habe dieses Produkt nicht in unserem Katalog gefunden."

        product = similar[0]

        # Antwort mit Produktkontext erzeugen
        prompt = f"""
        Identifiziertes Produkt: {product['name']}
        Beschreibung: {product['description']}
        Preis: {product['price']}€
        Merkmale: {product['specs']}

        Kundenfrage: {question}

        Antworte hilfreich und verkaufsorientiert.
        """

        return await self.llm.generate(prompt)

Support mit Screenshots

DEVELOPERpython
class VisualSupportAssistant:
    def __init__(self, rag_pipeline, vision_model):
        self.rag = rag_pipeline
        self.vision = vision_model

    async def analyze_screenshot(
        self,
        screenshot_path: str,
        user_description: str = ""
    ) -> dict:
        """
        Analysiert einen Screenshot eines Fehlers
        """
        # Bild mit dem Vision-Modell analysieren
        analysis = await self.vision.analyze(
            screenshot_path,
            prompt="""Analysiere diesen Screenshot und identifiziere:
            1. Den sichtbaren Anwendungs-/Oberflächentyp
            2. Jede Fehlermeldung oder Anomalie
            3. Den Kontext der laufenden Aktion
            4. Sichtbare technische Details (Codes, Versionen)
            """
        )

        # Mit der Nutzerbeschreibung kombinieren
        combined_query = f"""
        Nutzerbeschreibung: {user_description}

        Screenshot-Analyse:
        {analysis}
        """

        # In der Support-KB suchen
        results = await self.rag.query(
            text_query=combined_query,
            image_query_path=screenshot_path
        )

        return {
            "visual_analysis": analysis,
            "suggested_solutions": results["answer"],
            "related_articles": results["sources"]
        }

Best Practices

1. Vorverarbeitung der Bilder

• Größen vor dem Embedding normalisieren
• Konsistentes Preprocessing anwenden (Crop, Resize)
• Bilder schlechter Qualität herausfiltern

2. Metadatenverwaltung

Immer die Metadaten der Bilder aufbewahren:

• Quelle und ursprünglicher Kontext
• Abmessungen und Format
• Erstellungsdatum
• Generierte Beschreibung

3. Sanfter Fallback

DEVELOPERpython
async def safe_multimodal_query(query: str, image: str = None):
    """
    Behandelt Fälle, in denen der multimodale Pfad fehlschlägt
    """
    try:
        if image:
            return await multimodal_rag.query(query, image)
    except Exception as e:
        logger.warning(f"Multimodal failed: {e}, falling back to text-only")

    # Fallback zum text-only RAG
    return await text_rag.query(query)

Um weiterzugehen

• Retrieval-Grundlagen - Grundlagen der Suche
• Einführung in RAG - Überblick
• Dense Retrieval - Fortgeschrittene Embeddings

RAG Multimodal schlüsselfertig mit Ailog

Die Implementierung eines multimodalen RAG erfordert die Integration zahlreicher Technologien. Mit Ailog erhalten Sie diese Fähigkeiten ohne Komplexität:

• Erweiterte PDF-Extraktion mit Erkennung von Tabellen und Grafiken
• Integriertes OCR für gescannte Dokumente
• Bildindexierung mit automatischen Beschreibungen
• Hybride Suche Text + visuell
• Multimodale LLMs (GPT-4V, Claude 3) vorkonfiguriert

Testen Sie Ailog kostenlos und indexieren Sie alle Ihre Inhalte, unabhängig vom Format.