RAG-Implementierungsmuster

Was macht RAG-Implementierungsmuster?

Umfassender Leitfaden zur Implementierung von Retrieval-Augmented Generation (RAG)-Systemen, einschließlich Vektordatenbank-Auswahl, Chunking-Strategien, Embedding-Modellen, Retrieval-Optimierung und Produktionsbereitstellungsmustern.

---

Kurzreferenz

Wann dieser Skill verwendet werden sollte:

• Aufbau von RAG-/semantischen Suchsystemen
• Implementierung von Dokumentenabruf-Pipelines
• Optimierung der Vektordatenbank-Leistung
• Behebung von Problemen mit der Retrieval-Qualität
• Auswahl zwischen Vektordatenbank-Optionen
• Entwerfen von Chunking-Strategien
• Implementierung von Hybrid-Suche

Abgedeckte Technologien:

• Vektor-DBs: Qdrant, Pinecone, Chroma, Weaviate, Milvus
• Embeddings: OpenAI, Sentence Transformers, Cohere
• Frameworks: LangChain, LlamaIndex, Haystack

---

Teil 1: Vektordatenbank-Auswahl

Datenbank-Vergleichsmatrix

Datenbank	Am besten geeignet für	Bereitstellung	Performance	Kosten
Qdrant	Selbst gehostet, Produktion	Docker/K8s	Exzellent (Rust)	Kostenlos (selbst gehostet)
Pinecone	Verwaltet, schnelles Prototyping	Cloud	Exzellent	Pay-per-use
Chroma	Lokale Entwicklung, eingebettet	In-Prozess	Gut (Python)	Kostenlos
Weaviate	Komplexe Schemata, GraphQL	Docker/Cloud	Exzellent (Go)	Kostenlos + Cloud
Milvus	Groß angelegt, verteilt	K8s	Exzellent (C++)	Kostenlos (selbst gehostet)

Qdrant-Setup (Empfohlen für die Produktion)

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

# Client initialisieren (lokal oder Cloud)
client = QdrantClient(url="http://localhost:6333")  # oder Cloud-URL

# Sammlung erstellen
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # OpenAI text-embedding-3-small Dimension
        distance=Distance.COSINE  # oder DOT, EUCLID
    )
)

# Vektoren mit Nutzdaten einfügen
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 1536 Dimensionen
            payload={
                "text": "Dokumenteninhalt",
                "source": "doc.pdf",
                "page": 1,
                "metadata": {...}
            }
        )
    ]
)

# Suche
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=5,
    score_threshold=0.7  # Minimale Ähnlichkeit
)

Pinecone-Setup (Verwalteter Dienst)

from pinecone import Pinecone, ServerlessSpec

# Initialisieren
pc = Pinecone(api_key="your-key")

# Index erstellen
pc.create_index(
    name="documents",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

# Index abrufen
index = pc.Index("documents")

# Vektoren einfügen
index.upsert(vectors=[
    ("doc1", [0.1, 0.2, ...], {"text": "...", "source": "..."})
])

# Abfrage
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

---

Teil 2: Chunking-Strategien

Strategie 1: Chunking mit fester Größe (Einfach, Schnell)

def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """
    Text in Blöcke fester Größe mit Überlappung aufteilen.

    Vorteile: Einfach, vorhersagbare Blockgrößen
    Nachteile: Kann Satzmitten aufbrechen, schlechte semantische Grenzen
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

# Verwendung
chunks = fixed_size_chunking(document, chunk_size=512, overlap=50)

Wann verwenden:

• Einfache Dokumente (Protokolle, Transkripte)
• Prototyping/MVP
• Gleichbleibende Token-Budgets erforderlich

Strategie 2: Semantisches Chunking (Bessere Qualität)

from langchain.text_splitter import RecursiveCharacterTextSplitter

def semantic_chunking(text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
    """
    Aufteilung an semantischen Grenzen (Absätze, Sätze).

    Vorteile: Bewahrt Bedeutung, bessere Retrieval-Qualität
    Nachteile: Unterschiedliche Blockgrößen, langsamerer Verarbeitung
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],  # Prioritätsreihenfolge
        length_function=len
    )

    return splitter.split_text(text)

# Verwendung
chunks = semantic_chunking(document, chunk_size=1000, overlap=200)

Wann verwenden:

• Lange Dokumente (Artikel, Bücher, Berichte)
• Qualität > Geschwindigkeit
• Natürlichsprachige Inhalte

Strategie 3: Hierarchisches Chunking (Am besten für strukturierte Dokumente)

def hierarchical_chunking(document: dict) -> list[dict]:
    """
    Blockbildung basierend auf Dokumentenstruktur (Abschnitte, Unterabschnitte).

    Vorteile: Bewahrt Hierarchie, ermöglicht Eltern-Kind-Retrieval
    Nachteile: Erfordert strukturierte Eingabe, komplexer
    """
    chunks = []

    for section in document['sections']:
        # Eltern-Block (Abschnittszusammenfassung)
        chunks.append({
            'text': section['title'] + '\n' + section['summary'],
            'type': 'parent',
            'section_id': section['id']
        })

        # Kinder-Blöcke (Absätze)
        for para in section['paragraphs']:
            chunks.append({
                'text': para,
                'type': 'child',
                'parent_id': section['id']
            })

    return chunks

Wann verwenden:

• Technische Dokumentation
• Bücher mit Inhaltsverzeichnis
• Juristische Dokumente
• Notwendigkeit, Kontexthierarchie zu bewahren

Strategie 4: Gleitendes Fenster (Maximale Kontexterhaltung)

def sliding_window_chunking(text: str, window_size: int = 512, stride: int = 256) -> list[str]:
    """
    Überlappende Fenster für maximalen Kontext.

    Vorteile: Kein Informationsverlust an Grenzen
    Nachteile: Speichermehraufwand (doppelte Inhalte)
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words) - window_size + 1, stride):
        chunk = ' '.join(words[i:i + window_size])
        chunks.append(chunk)

    return chunks

Wann verwenden:

• Kritische Retrieval-Genauigkeit erforderlich
• Kurze Abfragen benötigen breiteren Kontext
• Speicherkosten spielen keine Rolle

---

Teil 3: Embedding-Modelle

Modell-Auswahlhilfe

Modell	Dimensionen	Geschwindigkeit	Qualität	Kosten	Anwendungsfall
OpenAI text-embedding-3-small	1536	Schnell	Exzellent	$0,02/1M Tokens	Produktion, allgemeiner Zweck
OpenAI text-embedding-3-large	3072	Mittel	Am besten	$0,13/1M Tokens	Hochwertiges Retrieval
all-MiniLM-L6-v2	384	Sehr schnell	Gut	Kostenlos	Selbst gehostet, Prototyping
all-mpnet-base-v2	768	Schnell	Sehr gut	Kostenlos	Selbst gehostet, Qualität
Cohere embed-english-v3.0	1024	Schnell	Exzellent	$0,10/1M Tokens	Fokus auf semantische Suche

OpenAI Embeddings (Empfohlen)

from openai import OpenAI

client = OpenAI(api_key="your-key")

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """
    Embeddings mit OpenAI generieren.

    Batch-Größe: Bis zu 2048 Eingaben pro Anfrage
    Ratenbegrenzungen: Tier-Grenzen prüfen
    """
    response = client.embeddings.create(
        model=model,
        input=texts
    )

    return [item.embedding for item in response.data]

# Verwendung
chunks = ["Block 1", "Block 2", ...]
embeddings = get_embeddings(chunks)

Sentence Transformers (Selbst gehostet)

from sentence_transformers import SentenceTransformer

# Modell laden (nach erstem Download zwischengespeichert)
model = SentenceTransformer('all-MiniLM-L6-v2')

def get_embeddings_local(texts: list[str]) -> list[list[float]]:
    """
    Embeddings lokal generieren (keine API-Kosten).

    GPU für Batches > 100 empfohlen
    CPU für kleine Batches akzeptabel
    """
    return model.encode(texts, show_progress_bar=True).tolist()

# Verwendung
embeddings = get_embeddings_local(chunks)

---

Teil 4: Retrieval-Optimierung

Technik 1: Hybrid-Suche (Dicht + Dünnbesetzt)

from qdrant_client.models import Filter, FieldCondition, MatchValue

def hybrid_search(query: str, query_vector: list[float], top_k: int = 10):
    """
    Kombiniert dichte (Vektor) und dünnbesetzte (Schlüsselwort) Suche.

    Dicht: Semantische Ähnlichkeit
    Dünnbesetzt: Exakte Schlüsselwortübereinstimmungen
    """
    # Dichte Suche
    dense_results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=top_k * 2  # Mehr Kandidaten abrufen
    )

    # Dünnbesetzte Suche (BM25 über Metadaten)
    sparse_results = client.search(
        collection_name="documents",
        query_filter=Filter(
            must=[
                FieldCondition(
                    key="text",
                    match=MatchValue(value=query)
                )
            ]
        ),
        limit=top_k * 2
    )

    # Zusammenführen und neu bewerten
    combined = merge_results(dense_results, sparse_results, weights=(0.7, 0.3))
    return combined[:top_k]

Technik 2: Abfrageerweiterung

def expand_query(query: str) -> list[str]:
    """
    Abfragevarianten für bessere Abdeckung generieren.

    Techniken:
    - Synonym-Erweiterung
    - Frageumformulierung
    - Entitätsextraktion
    """
    from openai import OpenAI
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "Generiere 3 alternative Formulierungen der Benutzerabfrage."
        }, {
            "role": "user",
            "content": query
        }]
    )

    expanded = [query] + response.choices[0].message.content.split('\n')
    return expanded

# Verwendung
queries = expand_query("Wie trainiere ich neuronale Netze?")
# → ["Wie trainiere ich neuronale Netze?",
#    "Was sind Trainingstechniken für neuronale Netze?",
#    "Optimierungsmethoden für neuronale Netze",
#    "Deep-Learning-Modelltraining"]

Technik 3: Neubewertung (Reranking)

from sentence_transformers import CrossEncoder

# Cross-Encoder laden (besser als Bi-Encoder zur Neubewertung)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: list[dict], top_k: int = 5) -> list[dict]:
    """
    Anfangsergebnisse mit Cross-Encoder neu bewerten.

    Genauer, aber langsamer als das anfängliche Retrieval
    Nur auf den Top-20-50 Kandidaten anwenden
    """
    # Jedes Abfrage-Dokument-Paar bewerten
    pairs = [(query, result['text']) for result in results]
    scores = reranker.predict(pairs)

    # Ergebnisse mit Bewertungen kombinieren
    for result, score in zip(results, scores):
        result['rerank_score'] = float(score)

    # Sortieren und top_k zurückgeben
    reranked = sorted(results, key=lambda x: x['rerank_score'], reverse=True)
    return reranked[:top_k]

Technik 4: Metadaten-Filterung

def filtered_search(
    query_vector: list[float],
    filters: dict,
    top_k: int = 5
):
    """
    Suche nach Metadaten filtern (Datum, Kategorie, Autor, usw.)

    Vor-Filter: Schneller, kann aber Ergebnisse übersehen
    Nach-Filter: Mehr Ergebnisse, aber langsamer
    """
    from qdrant_client.models import Filter, FieldCondition, Range

    # Filterbedingungen erstellen
    conditions = []

    if 'date_range' in filters:
        conditions.append(
            FieldCondition(
                key="date",
                range=Range(
                    gte=filters['date_range']['start'],
                    lte=filters['date_range']['end']
                )
            )
        )

    if 'category' in filters:
        conditions.append(
            FieldCondition(
                key="category",
                match=MatchValue(value=filters['category'])
            )
        )

    # Mit Filtern suchen
    results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=Filter(must=conditions) if conditions else None,
        limit=top_k
    )

    return results

---

Teil 5: Kontextverwaltung

Muster 1: Optimierung des abgerufenen Kontexts

def optimize_context(query: str, retrieved_docs: list[str], max_tokens: int = 4000) -> str:
    """
    Abgerufenen Kontext optimieren, um in das LLM-Kontextfenster zu passen.

    Strategien:
    1. Relevanzbasierte Kürzung
    2. Extraktive Zusammenfassung
    3. Überlappungsentfernung
    """
    # Nach Relevanz sortieren
    sorted_docs = sorted(retrieved_docs, key=lambda d: d['score'], reverse=True)

    # Kontext innerhalb des Token-Budgets aufbauen
    context_parts = []
    total_tokens = 0

    for doc in sorted_docs:
        doc_tokens = estimate_tokens(doc['text'])

        if total_tokens + doc_tokens <= max_tokens:
            context_parts.append(f"[Quelle: {doc['source']}]\n{doc['text']}")
            total_tokens += doc_tokens
        else:
            # Letztes Dokument kürzen, um zu passen
            remaining = max_tokens - total_tokens
            truncated = truncate_to_tokens(doc['text'], remaining)
            context_parts.append(f"[Quelle: {doc['source']}]\n{truncated}")
            break

    return "\n\n---\n\n".join(context_parts)

Muster 2: Zitierungsverfolgung

def generate_with_citations(query: str, context: str, sources: list[dict]) -> dict:
    """
    Antwort mit Zitierungsverfolgung generieren.

    Rückgabe:
    - answer: Generierter Text
    - citations: Liste der verwendeten Quelldokumente
    """
    from openai import OpenAI
    client = OpenAI()

    # Quellenkarte erstellen
    source_map = {i+1: source for i, source in enumerate(sources)}
    numbered_context = "\n\n".join([
        f"[{i+1}] {source['text']}"
        for i, source in enumerate(sources)
    ])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "Antworten Sie unter Verwendung der bereitgestellten Quellen. Zitieren Sie Quellen als [1], [2] usw."
        }, {
            "role": "user",
            "content": f"Kontext:\n{numbered_context}\n\nFrage: {query}"
        }]
    )

    answer = res