Daten vorbereiten

Die Qualität von RAG hängt von Ihren Daten ab

Eine RAG-Pipeline ist nur so gut wie die Daten, die sie indiziert. Sie können den perfekten Prompt entwerfen und das beste Generierungsmodell auswählen, aber wenn das abgerufene Fragment eine Definition in der Hälfte abschneidet oder einen sich wiederholenden Fußzeilentext statt nützlichen Texts mitnimmt, leidet Ihre Antwort. Die Vorbereitung erfolgt in drei Phasen: Holen Sie die Daten dort ab, wo sie liegen, entfernen Sie alles, was bedeutungslos ist, und teilen Sie sie in Fragmente auf, die für die Indizierung optimiert sind.

Daten abrufen

Echte Korpora sind selten homogen. Sie werden typischerweise Markdown-Dokumentation in einem Repository, einige Seiten einer öffentlichen Website und eine Handvoll regulatorischer PDFs haben. Jede Quelle benötigt ihren eigenen Loader, aber sie müssen alle dieselbe Struktur erzeugen—inhalt, Quelle und Typ—damit der Rest der Pipeline gegenüber dem Ursprung des Textes agnostisch bleibt.

Bei Textdateien reicht ein rekursiver Verzeichnislauf aus, und das Filtern nach Dateiendung verhindert, dass versehentlich Binärdateien geladen werden.

import os

def load_text_files(directory, extensions=(".txt", ".md", ".rst")):
    """Lädt alle Textdateien aus einem Verzeichnis."""
    documents = []
    for root, dirs, files in os.walk(directory):
        for f in files:
            if any(f.endswith(ext) for ext in extensions):
                filepath = os.path.join(root, f)
                with open(filepath, "r", encoding="utf-8") as fh:
                    content = fh.read()
                documents.append({
                    "content": content,
                    "source": filepath,
                    "type": "text",
                })
    return documents

docs = load_text_files("./data/")
print(f"{len(docs)} Dokumente geladen")

Webseiten kommen über HTTP, aber was Sie erhalten, ist HTML. Die folgende Extraktion entfernt Tags mithilfe von regulären Ausdrücken und fasst mehrere Leerzeichen zusammen. Dies reicht für einen Prototyp aus, aber der Kommentar erinnert Sie daran, dass BeautifulSoup in der Produktion bei Tabellen und verschachtelten Listen eine deutlich bessere Arbeit leistet.

import requests

def fetch_web_page(url):
    """Holt den Textinhalt einer Webseite."""
    response = requests.get(url)
    response.raise_for_status()

    # Basis-Textextraktion (verwenden Sie BeautifulSoup für die Produktion)
    text = response.text
    # HTML-Tags entfernen
    import re
    text = re.sub(r'<[^>]+>', ' ', text)
    text = re.sub(r'\s+', ' ', text).strip()

    return {
        "content": text,
        "source": url,
        "type": "web",
    }

# Beispiel
page = fetch_web_page("https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt")
print(f"Inhalt: {len(page['content'])} Zeichen")

PDFs verdienen eine besondere Behandlung. Anstatt das gesamte Dokument zu concatenieren, halten Sie pro Seite einen Eintrag und codieren die Seitennummer in der Quelle als file.pdf#page=12. Wenn Ihr Assistent diesen Abschnitt zitiert, wird der Benutzer genau wissen, wo er dies überprüfen kann.

# pip install PyPDF2
from PyPDF2 import PdfReader

def load_pdf(filepath):
    """Extrahiert Text aus einem PDF."""
    reader = PdfReader(filepath)
    pages = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text()
        if text.strip():
            pages.append({
                "content": text,
                "source": f"{filepath}#page={i+1}",
                "type": "pdf",
                "page": i + 1,
            })
    return pages

Daten bereinigen

Die Bereinigung ist kritisch für die Abrufqualität, da jedes fremde Zeichen Platz im Fragment verbraucht und das resultierende Vektorfeld trübt. Die folgende Funktion verknüpft vier Behandlungen: Sie entfernt Steuerzeichen, die von Formatkonvertierungen übrig geblieben sind, normalisiert Sequenzen von Zeilenumbrüchen und Leerzeichen, löscht Paginierungsmarker, die typisch für PDFs sind, und ersetzt endlose URLs durch einen einfachen Platzhalter, damit sie nicht ein ganzes Fragment verschlingen.

import re

def clean_text(text):
    """Bereinigt Text für die RAG-Indizierung."""
    # Entfernt Steuerzeichen
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)

    # Normalisiert Leerzeichen und Zeilenumbrüche
    text = re.sub(r'\n{3,}', '\n\n', text)
    text = re.sub(r' {2,}', ' ', text)

    # Entfernt sich wiederholende Kopf-/Fußzeilen (PDF)
    text = re.sub(r'Page \d+ of \d+', '', text)

    # Entfernt sehr lange URLs
    text = re.sub(r'https?://\S{100,}', '[URL]', text)

    return text.strip()

# Bereinigung anwenden
for doc in docs:
    doc["content"] = clean_text(doc["content"])

Chunking-Strategien

Das Chunking ist der wichtigste Schritt in der Vorbereitung: Chunk-Größe und -Methode wirken sich direkt auf die Abrufqualität aus. Es gibt drei Ansätze, die von mechanisch bis strukturbewusst reichen.

Das Chunking fester Größe bewegt sich durch den Text in regelmäßigen Schnitten. Es ignoriert die Bedeutung vollständig, was es für jede Quelle anwendbar macht, aber es schneidet oft Sätze in der Hälfte ab. Überlappung korrigiert diesen Fehler teilweise: Mit overlap=200 wiederholt sich das Ende jedes Fragments am Anfang des nächsten, sodass eine geteilte Idee in mindestens einem der beiden Teile vollständig bleibt.

def chunk_fixed_size(text, chunk_size=2048, overlap=200):
    """Chunking fester Größe mit Überlappung."""
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start += chunk_size - overlap
    return chunks

Das Chunking nach Absätzen, das für Dokumentation empfohlen wird, stützt sich auf die natürlichen Grenzen des Textes. Es sammelt Absätze, solange sie unter max_size bleiben, und öffnet dann ein neues Fragment. Die min_size-Schwelle verhindert, dass ein isolierter Titel oder eine einzelne Zeile indiziert wird, was ein Vektor ohne nutzbaren Inhalt erzeugen würde.

def chunk_by_paragraphs(text, max_size=1500, min_size=100):
    """Intelligentes Chunking nach Absätzen."""
    paragraphs = text.split("\n\n")
    chunks = []
    current = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue

        if len(current) + len(para) + 2 <= max_size:
            current = current + "\n\n" + para if current else para
        else:
            if len(current) >= min_size:
                chunks.append(current)
            current = para

    if current and len(current) >= min_size:
        chunks.append(current)

    return chunks

Das Chunking nach Markdown-Sektionen nutzt Informationen, die der Autor bereits bereitgestellt hat: seine Überschriften. Jede Sektion wird zu einem eigenständigen Fragment, was garantiert, dass eine Frage zu „Authentifizierung” die gesamte Sektion statt eines Fragments abruft. Sektionen, die zu lang sind, fallen auf das Chunking nach Absätzen zurück, sodass kein Fragment das Limit überschreitet.

def chunk_by_markdown_sections(text, max_size=2000):
    """Chunking nach Markdown-Überschriften (## und ###)."""
    sections = re.split(r'\n(?=##\s)', text)
    chunks = []

    for section in sections:
        section = section.strip()
        if not section:
            continue
        if len(section) <= max_size:
            chunks.append(section)
        else:
            # Unterteilt überlange Sektionen
            sub_chunks = chunk_by_paragraphs(section, max_size=max_size)
            chunks.extend(sub_chunks)

    return chunks

Komplette Vorbereitungs-Pipeline

Verknüpfen Sie schließlich die drei Operationen und wählen Sie die Chunking-Strategie basierend auf der Dateityp: .md-Dateien durchlaufen Sektionen, alles andere durchläuft Absätze. Jedes Fragment trägt dann seine Metadaten—Quelle, Position im Dokument und Gesamtzahl der Fragmente aus diesem Dokument. Ohne diese können Sie eine Antwort nicht nachvollziehen. Die letzten beiden Zeilen zeigen die Expansionsrate und die durchschnittliche Größe an: Wenn dieser Durchschnitt weit unter Ihrem chunk_size liegt, ist das Dokument mehr fragmentiert als erwartet, und Sie müssen Ihre Chunking-Strategie überdenken.

def prepare_documents(raw_docs, chunk_size=1500, overlap=200):
    """Komplette Pipeline: bereinigen, chunken, mit Metadaten anreichern."""
    all_chunks = []

    for doc in raw_docs:
        # 1. Bereinigen
        clean = clean_text(doc["content"])

        # 2. Chunken basierend auf Typ
        if doc.get("source", "").endswith(".md"):
            chunks = chunk_by_markdown_sections(clean, max_size=chunk_size)
        else:
            chunks = chunk_by_paragraphs(clean, max_size=chunk_size)

        # 3. Mit Metadaten anreichern
        for i, chunk in enumerate(chunks):
            all_chunks.append({
                "text": chunk,
                "source": doc["source"],
                "chunk_index": i,
                "total_chunks": len(chunks),
            })

    return all_chunks

# Ausführung
chunks = prepare_documents(docs)
print(f"{len(docs)} Dokumente -> {len(chunks)} Fragmente")
print(f"Durchschnittliche Größe: {sum(len(c['text']) for c in chunks) / len(chunks):.0f} Zeichen")

Kernaussagen

• Die Qualität von RAG hängt direkt von der Datenqualität und der Chunking-Strategie ab
• Bereinigen Sie Daten vor der Indizierung (entfernen Sie fremde Zeichen, lange URLs, Duplikate)
• Das Chunking nach Absätzen oder Markdown-Sektionen bewahrt semantische Kohärenz
• Eine Überlappung von 10–15 % zwischen Fragmenten verhindert das Aufteilen von Ideen
• Halten Sie Metadaten (Quelle, Seite, Position) für die Nachvollziehbarkeit von Antworten bei