Aufbau einer Produktions-ETL-Pipeline für RAG mit LangChain und FAISS

Stellen Sie sich vor, Sie sind der GenAI-Engineer in einem mittelgroßen FinTech-Unternehmen. Das Compliance-Team besitzt mehr als 40.000 PDF-Dateien, die regulatorische Unterlagen, Zirkulare der Zentralbanken, Richtlinien der Wertpapierbehörden sowie interne Richtlinien umfassen und über SharePoint, S3 sowie lokale Laufwerke verteilt sind. Die Analysten verbringen vier bis sechs Stunden pro Tag damit, diese manuell zu durchsuchen. Der CTO wünscht einen Chatbot, der jede Frage zur Compliance aus dieser Datenbank in weniger als drei Sekunden beantworten kann.

Es handelt sich offensichtlich um eine Aufgabe der Retrieval-Augmented Generation. Was leicht übersehen wird, ist, dass die Qualität der Antworten bereits lange vor dem Eintreffen einer Frage im ETL-Prozess entschieden wird, der die Wissensdatenbank vorbereitet. Falsches Aufteilen in Fragmente zerstört den Kontext, naive Embeddings führen dazu, dass die Bedeutung verloren geht, und eine schlechte Indizierung verlangsamt das Abrufen von Informationen. Eine grobe Erfahrungregel besagt, dass etwa 80 Prozent der Qualitätsprobleme bei RAG tatsächlich ETL-Probleme sind. In der Regel ist das Modell nicht die Engstelle, sondern der gesamte Prozessfluss.

Die verwendeten Technologien sind Python, LangChain, FAISS, Pinecone sowie OpenAI-Embeddings.

Die betroffene Skala

• Etwa 40.000 Dokumente in gemischten Formaten: PDF, DOCX, HTML und reiner Text.
• Ungefähr 120 Fragmente pro Dokument nach rekursivem Aufteilen in Blöcken von 512 Tokenn.
• Ein Vektorindex mit 4,8 Millionen Embeddings in Pinecone, bei 1536 Dimensionen.

P95-Auslieferungszeit von 180 ms, nachdem der ungefähre nächstgelegene Nachbarindex mit ef=200 abgestimmt wurde.

Pipeline-Architektur

Es handelt sich um den klassischen Ansatz extrahieren, transformieren, laden. Jede Stufe verbirgt domain-spezifische Komplexitäten, die leicht unterschätzt werden können.

┌────────────────────────────────────────────────────────────────────────────┐
│                    RAG ETL Pipeline                                        │
│                                                                            │
│  [EXTRACT]       [TRANSFORM 1]    [TRANSFORM 2]    [LOAD]                  │
│  Doc Loader  →   Chunker      →   Embedder      →  VectorDB  →  RAG API    │
│  PDF·DOCX·HTML   Semantic splits  Batch encode     Pinecone/FAISS          │
└────────────────────────────────────────────────────────────────────────────┘

Jede Phase läuft als eigenständiger, idempotenter Prozess und schreibt in einen Zwischenspeicher, bevor die nächste Phase startet. Dadurch wird Fehlertoleranz gewährleistet. Wenn die Extraktion bei Dokument 30.000 fehlschlägt, zwingt nichts dazu, alle 40.000 Dokumente erneut zu extrahieren.

Der Orchesterer ist ein durch Dataclasses gesteuerter Phase-Executor:

# pipeline/orchestrator.py
# Top-level pipeline orchestration using a dataclass-driven stage runner

from dataclasses import dataclass, field
from typing import List, Optional
import logginglogger = logging.getLogger(__name__)@dataclass
class PipelineConfig:
    source_bucket: str
    chunk_size: int = 512           # tokens, not characters
    chunk_overlap: int = 64         # token overlap between chunks
    embedding_model: str = "text-embedding-3-small"
    embedding_batch_size: int = 256
    vector_store: str = "pinecone"
    index_name: str = "compliance-docs"
    namespace: str = "v2"
    max_workers: int = 8            # parallel PDF extraction workersclass ETLPipeline:
    def __init__(self, config: PipelineConfig):
        self.config = config
        self.loader   = DocumentLoader(config)
        self.chunker  = AdaptiveChunker(config)
        self.embedder = BatchEmbedder(config)
        self.indexer  = VectorIndexer(config)    def run(self, doc_ids: Optional[List[str]] = None) -> dict:
        """Full ETL run. Pass doc_ids for incremental updates."""
        stats = {}        # Stage 1: Extract (parallelised — 2-5s per PDF, 8 workers)
        raw_docs = self.loader.load(doc_ids=doc_ids)
        stats["extracted"] = len(raw_docs)
        logger.info(f"Extracted {len(raw_docs)} documents")        # Stage 2a: Chunk
        chunks = self.chunker.chunk_batch(raw_docs)
        stats["chunks"] = len(chunks)        # Stage 2b: Embed
        embedded = self.embedder.embed_batch(chunks)        # Stage 3: Index
        upserted = self.indexer.upsert(embedded)
        stats["upserted"] = upserted
        return stats

Diagramm als Quelle der Wahrheit

Ein ASCII-Flussdiagramm zeigt jeden Entscheidungspunkt dar – von der Eingabe des Rohdokuments über Wiederholungsversuche und Gruppierung bis zur endgültigen Speicherung im Vektorlagern.

                        ┌──────────────────────┐
                        │  START: Doc Queue    │
                        └──────────┬───────────┘
                                   │
                        ┌──────────▼───────────┐
                        │  Detect Doc Type     │  ← PDF · DOCX · HTML · MD · TXT
                        └──────────┬───────────┘
                                   │
                        ┌──────────▼───────────┐
                        │   Parseable? (OCR)   │
                        └────┬──────────┬──────┘
                         YES │          │ NO
                             │          ▼
                             │   ┌──────────────────┐
                             │   │ Dead-Letter Queue │  ← PagerDuty alert
                             │   └──────────────────┘
                ┌────────────▼────────────────────┐
   EXTRACT      │  Text Extraction + Cleaning     │
                │  PyMuPDF · python-docx · BS4    │
                │  2–5s / PDF · Parallel workers  │
                └────────────┬────────────────────┘
                             │
                ┌────────────▼────────────────────┐
                │   Select Chunking Strategy      │
                └──────┬────────────┬─────────────┘
                       │            │              │
              Code/Tables    Default (prose)   Long-form
                       │            │              │
              ┌────────▼──┐  ┌──────▼──────┐ ┌───▼─────────┐
 TRANSFORM    │ Fixed-Size│  │  Recursive  │ │  Semantic   │
 (CHUNK)      │ char split│  │  Char Split │ │  (NLTK+emb) │
              └────────┬──┘  └──────┬──────┘ └───┬─────────┘
                       └────────────┼─────────────┘
                                    │ (converge)
                        ┌───────────▼─────────────┐
                        │ Attach Metadata Per Chunk│
                        │ doc_id·page·source·auth  │
                        └───────────┬─────────────┘
                                    │
                        ┌───────────▼─────────────┐
                        │  Token Length Validation │
                        │  Assert chunk ≤ 8191 tok │
                        └───────────┬─────────────┘
                                    │
                        ┌───────────▼─────────────┐
                        │ Group Chunks → Batches   │
                        │ batch_size=256           │
                        └───────────┬─────────────┘
                                    │
                        ┌───────────▼─────────────┐
 TRANSFORM              │   Call Embedding API    │ ◄─────────┐
 (EMBED)                │   async · retry · rate  │           │
                        └───────────┬─────────────┘           │
                                    │                         │
                        ┌───────────▼─────────────┐     ┌─────┴────── ┐
                        │    API Call Success?     │ NO │  Exp Backoff│
                        └──────┬──────────┬────────┘    │  2^n+jitter │
                           YES │          └────────────►└─────────────┘
                               │
                        ┌──────▼──────────────────┐
                        │  Normalize Embeddings   │
                        │  L2 norm · float32      │
                        └──────┬──────────────────┘
                               │
                        ┌──────▼──────────────────┐
                        │  Write to Redis Cache   │
                        │  hash(text)→vec · TTL7d │
                        └──────┬──────────────────┘
                               │
                        ┌──────▼──────────────────┐
 LOAD                   │  Upsert to Vector Store │
 (INDEX)                │  Pinecone · batch=100   │
                        └──────┬──────────────────┘
                               │
                        ┌──────▼──────────────────┐
                        │    Index Updated?        │ NO ──► Retry DLQ (max 5)
                        └──────┬──────────────────┘
                           YES │
                        ┌──────▼──────────────────┐
                        │   Emit Pipeline Metrics  │
                        │   Prometheus · Grafana   │
                        └──────┬──────────────────┘
                               │
                        ┌──────▼──────────────────┐
                        │   Pipeline Complete ✓   │
                        └─────────────────────────┘

Jeder Diamant in diesem Diagramm entspricht einem tatsächlichen if/else-Zweig im Code. Seine Zeichnung macht den Flussdiagramm bereits vor der eigentlichen Implementierung zur Referenz für die Architektur – und nicht nur zu Dokumentation.

Blochbildung

Die Blochbildung ist vermutlich die einflussreichste Entscheidung in der Verarbeitungskette. Ein Block ist die Abrufeinheit. Ist er zu groß, wird das Kontextfenster mit irrelevantem Text verschwendet; ist er zu klein, zerfällt die Bedeutung. Es gibt keine universell richtige Blockgröße.

Trennung mit fester Größe

Der einfachste Ansatz besteht darin, alle N Zeichen oder Token abzutrennen, gegebenenfalls mit Überschneidung. Er eignet sich für einheitliche Daten wie Protokolle und Tabellen, zerstört aber die Grenzen zwischen Sätzen und Abschnitten in prosaischen Texten.

# chunking/fixed_size.py

from langchain.text_splitter import CharacterTextSplitter
import tiktokenclass FixedSizeChunker:
    """
    Use for: Code files, CSV-like data, log files.
    Avoid for: Prose documents, legal text, RBI circulars.
    """
    def __init__(self, chunk_size: int = 512, overlap: int = 50):
        self.enc = tiktoken.get_encoding("cl100k_base")
        self.splitter = CharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=overlap,
            separator="",              # hard cut — no respect for sentences
            length_function=lambda t: len(self.enc.encode(t)),
        )    def chunk(self, text: str, metadata: dict) -> list[dict]:
        chunks = self.splitter.split_text(text)
        return [
            {
                "text": c,
                "token_count": len(self.enc.encode(c)),
                "chunk_idx": i,
                **metadata,
            }
            for i, c in enumerate(chunks)
        ]

Rekursives Zeichensegmentieren (Standard)

Dies ist die gängigste Methode. Zunächst wird der sinnvollste Trenner ausprobiert (leere Zeile, dann Zeilenumbruch, dann Punkt, dann Leerzeichen, dann einzelnes Zeichen), und erst wenn es unbedingt notwendig ist, wird auf ein strengeres Segmentierungsverfahren zurückgegriffen. LangChain’s RecursiveCharacterTextSplitter implementiert diese Methode. Die unten stehende angepasste Version zählt die Tokens mit tiktoken, sodass keine Teilmenge die 8192-Token-Grenze des Embedding-Modells überschreitet.

# chunking/recursive_splitter.py

from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken
from typing import Listclass AdaptiveChunker:
    """
    Production-grade chunker for compliance PDFs.
    Uses tiktoken to count tokens (not characters), ensuring
    we never exceed the 8192 embedding model context limit.
    """    # Separator hierarchy: try broader splits first
    SEPARATORS = [
        "\n\n\n",   # Section breaks
        "\n\n",     # Paragraph breaks
        ".\n",      # Sentence at line end
        ". ",       # Sentence in paragraph
        "\n",       # Line break
        " ",        # Word boundary (last resort)
    ]    def __init__(self, config):
        self.enc = tiktoken.encoding_for_model("text-embedding-3-small")
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=config.chunk_size,        # 512 tokens
            chunk_overlap=config.chunk_overlap,  # 64 tokens
            separators=self.SEPARATORS,
            length_function=self._token_len,
            is_separator_regex=False,
            keep_separator=True,                 # preserve paragraph structure
        )    def _token_len(self, text: str) -> int:
        return len(self.enc.encode(text, disallowed_special=()))    def chunk_document(self, raw_doc: dict) -> List[dict]:
        text = raw_doc["text"]
        splits = self.splitter.split_text(text)        chunks = []
        for idx, chunk_text in enumerate(splits):
            token_count = self._token_len(chunk_text)            # Safety guard — should not happen, but log if it does
            if token_count > 8191:
                raise ValueError(
                    f"Chunk {idx} has {token_count} tokens — exceeds model limit. "
                    f"doc_id={raw_doc['doc_id']}"
                )            chunks.append({
                "chunk_id": f"{raw_doc['doc_id']}_chunk_{idx}",
                "text": chunk_text,
                "token_count": token_count,
                "chunk_index": idx,
                "total_chunks": len(splits),   # for reconstruction context
                # Metadata propagated to vector store
                "doc_id": raw_doc["doc_id"],
                "source": raw_doc["source"],
                "page_number": raw_doc.get("page_number", 0),
                "doc_type": raw_doc["doc_type"],
                "created_date": raw_doc["created_date"],
                "authority": raw_doc.get("authority", "unknown"),  # RBI, SEBI, etc.
            })        return chunks    def chunk_batch(self, raw_docs: List[dict]) -> List[dict]:
        all_chunks = []
        for doc in raw_docs:
            try:
                all_chunks.extend(self.chunk_document(doc))
            except Exception as e:
                logger.error(f"Chunking failed for {doc['doc_id']}: {e}")
        return all_chunks

Semantisches Chunking

Hier wird jeder Satz eingebettet, und der Text wird an allen Stellen geteilt, an denen die Kosinusähnlichkeit zwischen benachbarten Teilen stark abfällt. Die entstandenen Blöcke sind kohärent, doch der Prozess verläuft 10 bis 15 Mal langsamer und erfordert zusätzliche Aufrufe zur Einbettung nur für das Chunking, wodurch die API-Nutzung effektiv verdoppelt wird. Dies macht nur bei hochwertigen narrativen Dokumenten wie langen Rechtsmeinungen, Jahresberichten und Strategiepapieren Sinn.

# chunking/semantic_chunker.py

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddingsclass SemanticChunkerWrapper:
    """
    Use ONLY for: Long-form legal opinions, annual reports,
    policy white papers where paragraph coherence is critical.    Cost warning: This calls the embedding API TWICE per document
    (once for chunking, once for indexing). Cache aggressively.
    """
    def __init__(self, breakpoint_threshold: float = 0.85):
        embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.splitter = SemanticChunker(
            embeddings=embeddings,
            # Splits when cosine similarity between sentences drops below 0.85
            breakpoint_threshold_type="cosine",
            breakpoint_threshold_amount=breakpoint_threshold,
            buffer_size=1,   # include 1 sentence of context across boundaries
        )    def chunk(self, text: str) -> list[str]:
        return self.splitter.split_text(text)

Vergleich der Methoden:

• Festes Format: am besten für Code, Protokolle und tabellarische Daten. Schnell und günstig, aber mit geringer Kohärenz.
• Rekursiv (Standard): in den meisten Fällen am besten für PDFs, DOCX und HTML. Schnell und günstig, mit mittlerer Kohärenz.
• Semantisch: am besten für lange rechtliche und Richtliniendokumente. Hohe Kohärenz, aber 10 bis 15 Mal langsamer sowie etwa doppelt so viele API-Aufrufe erforderlich.

Erstellung von Embeddings

Embeddings wandeln einen Textabschnitt in einen dichten Vektor um, der seine Bedeutung erfassen kann. Zu den entscheidenden Fragen gehören, welches Modell verwendet werden soll, wie die Verarbeitung in Batches erfolgen soll, ob gespeichert werden sollte und wie Rate Limits asynchron gehandhabt werden können.

Die Rechenlogik zur Ratebegrenzung ist ernüchternd. OpenAIs text-embedding-3-small erlaubt auf der Stufe 2 eine Million Tokens pro Minute. Bei 4,8 Millionen Blöcken à durchschnittlich 300 Tokens ergibt sich insgesamt 1,44 Milliarden Tokens. Die sequenzielle Verarbeitung bei dieser Grenze würde etwa 24 Stunden in Anspruch nehmen, während asynchrone Batching-Verfahren mit parallelen Prozessen die Zeit auf etwa zweieinhalb Stunden verkürzen.

# embedding/batch_embedder.py

import asyncio
import hashlib
import json
import time
from typing import List
import openai
import redis
import numpy as np
from tenacity import retry, stop_after_attempt, wait_exponentialclass BatchEmbedder:
    """
    Async batch embedder with:
    - Redis-backed cache (avoid re-embedding unchanged chunks)
    - Exponential backoff on rate limit errors
    - Token budget management to avoid 429s
    """
    TOKEN_BUDGET_PER_MINUTE = 900_000   # Stay under 1M limit with headroom    def __init__(self, config):
        self.client = openai.AsyncOpenAI()
        self.model = config.embedding_model
        self.batch_size = config.embedding_batch_size  # 256
        self.redis = redis.Redis(host="localhost", decode_responses=True)
        self.tokens_this_minute = 0
        self.minute_start = time.monotonic()    def _cache_key(self, text: str) -> str:
        return f"emb:{hashlib.sha256(text.encode()).hexdigest()[:16]}"    def _from_cache(self, text: str):
        val = self.redis.get(self._cache_key(text))
        return json.loads(val) if val else None    def _to_cache(self, text: str, vector: list, ttl: int = 604800):  # 7 days
        self.redis.setex(self._cache_key(text), ttl, json.dumps(vector))    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=60),
    )
    async def _call_api(self, texts: List[str]) -> List[List[float]]:
        response = await self.client.embeddings.create(
            model=self.model,
            input=texts,
            encoding_format="float",
        )
        return [item.embedding for item in response.data]    async def _embed_batch(self, chunks: List[dict]) -> List[dict]:
        texts = [c["text"] for c in chunks]
        cache_hits, uncached_texts, uncached_idxs = [], [], []        # Separate cache hits from misses
        for i, text in enumerate(texts):
            cached = self._from_cache(text)
            if cached:
                cache_hits.append((i, cached))
            else:
                uncached_texts.append(text)
                uncached_idxs.append(i)        # Token budget enforcement
        elapsed = time.monotonic() - self.minute_start
        if elapsed > 60:
            self.tokens_this_minute = 0
            self.minute_start = time.monotonic()        estimated_tokens = sum(len(t.split()) * 1.3 for t in uncached_texts)
        if self.tokens_this_minute + estimated_tokens > self.TOKEN_BUDGET_PER_MINUTE:
            sleep_time = 60 - elapsed + 2
            logger.info(f"Token budget hit, sleeping {sleep_time:.1f}s")
            await asyncio.sleep(sleep_time)
            self.tokens_this_minute = 0
            self.minute_start = time.monotonic()        # Embed uncached chunks
        new_vectors = []
        if uncached_texts:
            new_vectors = await self._call_api(uncached_texts)
            self.tokens_this_minute += estimated_tokens
            for text, vec in zip(uncached_texts, new_vectors):
                self._to_cache(text, vec)        # Assemble results in original order
        all_vectors = [None] * len(chunks)
        for i, vec in cache_hits:
            all_vectors[i] = vec
        for i, vec in zip(uncached_idxs, new_vectors):
            all_vectors[i] = vec        # L2 normalize for cosine similarity in FAISS / Pinecone
        for chunk, vector in zip(chunks, all_vectors):
            arr = np.array(vector, dtype=np.float32)
            arr /= np.linalg.norm(arr) + 1e-9
            chunk["embedding"] = arr.tolist()        return chunks    async def embed_batch_async(self, all_chunks: List[dict]) -> List[dict]:
        batches = [
            all_chunks[i : i + self.batch_size]
            for i in range(0, len(all_chunks), self.batch_size)
        ]
        tasks = [self._embed_batch(batch) for batch in batches]
        results = await asyncio.gather(*tasks, return_exceptions=True)        embedded_chunks = []
        for r in results:
            if isinstance(r, Exception):
                logger.error(f"Batch embedding failed: {r}")
            else:
                embedded_chunks.extend(r)
        return embedded_chunks    def embed_batch(self, chunks: List[dict]) -> List[dict]:
        return asyncio.run(self.embed_batch_async(chunks))

Die zu befolgende Regel lautet, niemals denselben Datenblock zweimal einzubetten. Ein mit Redis betriebener Cache reduziert die Anzahl der Iterationen bei Sammlungen, die selten ändern, um 70 bis 90 Prozent.

Indizierung

Durch die Indizierung landen die eingebetteten Datenblöcke in einem suchbaren Speicher. Zwei Entscheidungen spielen dabei eine entscheidende Rolle: das Metadaten-Schema (für gefilterte Abfragen) sowie der Indextyp (Flat, HNSW oder IVF).

Metadaten unterscheiden ein einfaches RAG-System von einem produktionsreifen System. Nehmen wir die Frage, was die Wertpapierbehörde 2023 zu Insiderhandel gesagt hat. Das System sollte zunächst mit authority=SEBI und year=2023 vorfiltern, bevor überhaupt eine Suche nach dem nächsten Nachbarn durchgeführt wird – dadurch wird der Suchraum eingeschränkt und die Genauigkeit erhöht.

# indexing/vector_indexer.py

from pinecone import Pinecone, ServerlessSpec
from typing import List
import uuid# ── Pinecone Metadata Schema ──────────────────────────────────────────
# These fields are indexed for filtered retrieval.
# Only add fields you will actually filter on — excess metadata costs money.
METADATA_SCHEMA = {
    "doc_id":       str,    # for document-level retrieval
    "chunk_index":  int,    # for ordered reconstruction
    "total_chunks": int,    # for context window sizing
    "source":       str,    # S3 path or SharePoint URL
    "doc_type":     str,    # "circular", "guideline", "policy"
    "authority":    str,    # "RBI", "SEBI", "IRDAI", "internal"
    "year":         int,    # publication year (for temporal filters)
    "page_number":  int,    # for citation in UI
    "token_count":  int,    # for context window management at query time
}class VectorIndexer:
    def __init__(self, config):
        self.pc = Pinecone(api_key=config.pinecone_api_key)
        self.index_name = config.index_name
        self.namespace = config.namespace
        self._ensure_index()    def _ensure_index(self):
        existing = [i.name for i in self.pc.list_indexes()]
        if self.index_name not in existing:
            self.pc.create_index(
                name=self.index_name,
                dimension=1536,                  # text-embedding-3-small output dim
                metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )
            logger.info(f"Created Pinecone index: {self.index_name}")
        self.index = self.pc.Index(self.index_name)    def _build_vectors(self, embedded_chunks: List[dict]) -> list:
        """
        Convert embedded chunk dicts to Pinecone upsert format.
        Each vector = (id, embedding, metadata).
        """
        vectors = []
        for chunk in embedded_chunks:
            vector_id = chunk.get("chunk_id") or str(uuid.uuid4())
            metadata = {
                k: chunk[k]
                for k in METADATA_SCHEMA.keys()
                if k in chunk
            }
            # Store a snippet for quick UI display without fetching full doc
            metadata["text_preview"] = chunk["text"][:200]            vectors.append({
                "id": vector_id,
                "values": chunk["embedding"],
                "metadata": metadata,
            })
        return vectors    def upsert(self, embedded_chunks: List[dict]) -> int:
        vectors = self._build_vectors(embedded_chunks)
        total_upserted = 0        # Pinecone upsert in batches of 100
        for i in range(0, len(vectors), 100):
            batch = vectors[i : i + 100]
            self.index.upsert(vectors=batch, namespace=self.namespace)
            total_upserted += len(batch)        logger.info(f"Upserted {total_upserted} vectors to '{self.namespace}'")
        return total_upserted    def query(
        self,
        query_vector: list,
        top_k: int = 10,
        filters: dict = None
    ) -> list:
        """
        Filtered ANN search. Example:
        filters = {"authority": {"$in": ["SEBI", "RBI"]}, "year": {"$gte": 2022}}
        """
        result = self.index.query(
            vector=query_vector,
            top_k=top_k,
            namespace=self.namespace,
            filter=filters,
            include_metadata=True,
            include_values=False,    # we don't need the vectors back
        )
        return result.matches

FAISS für lokale Umgebungen

Wenn Daten nicht an eine Cloud-Vektordatenbank übertragen werden können – was in der Bankwesen, Finanzbranche und im Staatseinsatz häufig vorkommt – ist FAISS die gängige Lösung. Die Wahl des Index-Typs ist von großer Bedeutung.

# indexing/faiss_indexer.py

import faiss
import numpy as npclass FAISSIndexer:
    def __init__(self, dim: int = 1536, n_vectors: int = 5_000_000):
        self.dim = dim
        # Index selection logic based on corpus size:
        # < 1M vectors    → IndexFlatIP         (exact, fast enough)
        # 1M–10M vectors  → IndexHNSWFlat       (M=32, efConstruction=64)
        # > 10M vectors   → IndexIVFPQ          (nlist=4096, m=32)
        if n_vectors < 1_000_000:
            self.index = faiss.IndexFlatIP(dim)
        elif n_vectors < 10_000_000:
            self.index = faiss.IndexHNSWFlat(dim, 32)   # M=32 edges per node
            self.index.hnsw.efConstruction = 64
            self.index.hnsw.efSearch = 200               # higher = more accurate, slower
        else:
            quantizer = faiss.IndexFlatIP(dim)
            self.index = faiss.IndexIVFPQ(quantizer, dim, 4096, 32, 8)        # GPU offload for faster build (comment out if no GPU)
        res = faiss.StandardGpuResources()
        self.index = faiss.index_cpu_to_gpu(res, 0, self.index)    def add(self, vectors: np.ndarray):
        # Vectors must be float32, L2 normalized for cosine similarity
        faiss.normalize_L2(vectors)
        self.index.add(vectors)    def search(self, query: np.ndarray, k: int = 10):
        faiss.normalize_L2(query)
        distances, indices = self.index.search(query, k)
        return distances, indices    def save(self, path: str):
        cpu_index = faiss.index_gpu_to_cpu(self.index)
        faiss.write_index(cpu_index, path)    def load(self, path: str):
        self.index = faiss.read_index(path)

• IndexFlatIP eignet sich für Korpora mit weniger als 1 Million Vektoren, bietet eine genaue (100%) Rückrufrate, arbeitet mit mittlerer Geschwindigkeit und verbraucht viel Speicher.
• IndexHNSWFlat ist geeignet für 1 bis 10 Millionen Vektoren, erreicht eine Rückrufrate von etwa 98,5% bei 10 Abfragen, ist schnell und verbraucht mittelgroßen Speicher.
• IndexIVFPQ eignet sich für mehr als 10 Millionen Vektoren, erreicht eine Rückrufrate von 94 bis 96%, ist sehr schnell und verbraucht dank 4- bis 8-Bit-Kompression nur wenig Speicher.

Fehlerbehandlung und Wiederholungsversuche

Jede Stufe eines Produktionspipelines wird manchmal fehlschlagen. Entscheidend ist ein fehlerfreies Versagen: Die fehlerhaften Dokumente sollten beiseitegelegt werden, ohne den Ablauf zu stoppen, und es sollte ein klarer Prüfverlauf für eine manuelle Überprüfung vorhanden sein. Eine Dead-Letter-Queue erfüllt diese Aufgabe, indem sie jeden Fehler mit der Stufe kennzeichnet, in der er aufgetreten ist.

# pipeline/fault_handler.py

from enum import Enum
from dataclasses import dataclass, field
from datetime import datetimeclass FailureStage(Enum):
    EXTRACTION = "extraction"
    CHUNKING   = "chunking"
    EMBEDDING  = "embedding"
    INDEXING   = "indexing"@dataclass
class PipelineFailure:
    doc_id: str
    stage: FailureStage
    error_type: str
    error_message: str
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    retry_count: int = 0
    is_retriable: bool = Trueclass DeadLetterQueue:
    """Stores failed items for later inspection and retry."""
    def __init__(self, redis_client, key_prefix="dlq"):
        self.r = redis_client
        self.prefix = key_prefix    def push(self, failure: PipelineFailure):
        key = f"{self.prefix}:{failure.stage.value}"
        self.r.lpush(key, json.dumps(failure.__dict__))
        logger.warning(
            f"DLQ push: doc={failure.doc_id} stage={failure.stage.value} "
            f"err={failure.error_type}: {failure.error_message[:80]}"
        )    def pop_batch(self, stage: FailureStage, n: int = 50) -> list:
        key = f"{self.prefix}:{stage.value}"
        items = self.r.lrange(key, 0, n - 1)
        self.r.ltrim(key, n, -1)
        return [json.loads(i) for i in items]# ── Categorize errors for smart retry decisions ────────────────────
RETRIABLE_ERRORS = {
    "openai.RateLimitError",
    "openai.APITimeoutError",
    "pinecone.exceptions.PineconeApiException",
    "ConnectionError",
}NON_RETRIABLE_ERRORS = {
    "UnicodeDecodeError",     # Corrupted doc — send to DLQ permanently
    "ValueError",             # Bad chunking logic — needs code fix
    "pdfminer.pdfpage.PDFTextExtractionNotAllowed",  # Encrypted PDF
}def is_retriable(exc: Exception) -> bool:
    return type(exc).__name__ in RETRIABLE_ERRORS

Überwachung

Ein Pipeline ohne Metriken ist eine Black Box. Verfolgen Sie zumindest die Durchsatzrate, Latenzzeit, Fehlerquote sowie die Kosten pro Dokument – diese werden an Prometheus übermittelt und in Grafana dargestellt.

# monitoring/metrics.py

from prometheus_client import Counter, Histogram, Gauge, start_http_server# ── Define metrics ─────────────────────────────────────────────────
docs_processed = Counter(
    "etl_docs_processed_total",
    "Total documents processed",
    ["status"],           # labels: "success" or "failed"
)
chunks_created = Counter(
    "etl_chunks_created_total",
    "Total chunks created"
)
embedding_latency = Histogram(
    "etl_embedding_seconds",
    "Embedding API call latency",
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)
index_queue_depth = Gauge(
    "etl_index_queue_depth",
    "Vectors waiting to be indexed"
)
embedding_cache_hit_rate = Gauge(
    "etl_cache_hit_rate",
    "Redis cache hit rate for embeddings"
)
dlq_depth = Gauge(
    "etl_dlq_depth",
    "Dead-letter queue depth",
    ["stage"]
)# Usage in BatchEmbedder:
with embedding_latency.time():   # measures duration automatically
    vectors = await self._call_api(uncached_texts)# Start Prometheus metrics endpoint on port 8001
start_http_server(8001)

Nützliche Dashboard-Panels:

• Dokumente pro Minute im Laufe der Zeit.
• Ein Histogramm zur Darstellung der P95-Embedding-Latenz.
• Hit-Rate des Redis-Caches, mit Zielwert von über 80 %.
• Tiefe der Dead-Letter-Queue nach Phase; es wird gewarnt, wenn sie länger als fünf Minuten nicht null ist.
• Kosten pro Dokument, also Anzahl der Tokens multipliziert mit dem Preis pro Token.
• Abruf-Latenz von FAISS oder Pinecone bei der Suche (P50, P95, P99).

Messergebnisse

Diese Werte stammen vom Konformitätsystem nach Anpassungen auf einem AWS c5.4xlarge-Server (16 vCPU, 32 GB RAM) mit asynchronen Embedding-Arbeitern.

PDF-Extraktion

PyMuPDF ohne OCR benötigt 2 bis 5 Sekunden pro Dokument, abhängig von der Anzahl der Seiten und dem Layout; die Arbeit lässt sich gut parallelisieren. Mit 8 gleichzeitigen Arbeitern dauert das Extrahieren von 40.000 PDFs etwa:

40,000 docs × avg 3.5s / 8 workers = ~4.9 hours wall-clock
(scales linearly — 16 workers → ~2.4 hours)

Nach Dateityp: Ein einfacher, 1 bis 5 Seiten umfassender Kreisbericht benötigt 2 bis 3 Sekunden, da die Textschicht klar strukturiert ist. Ein Leitfaden mit 20 bis 50 Seiten, mehreren Spalten und eingebetteten Tabellen, braucht 3 bis 4 Sekunden. Ein Jahresbericht mit über 100 Seiten benötigt 4 bis 5 Sekunden, da dichte Layouts und viele Bilder übersprungen werden müssen. Verschlüsselte oder gescannte PDFs landen in der Warteschlange für unverarbeitete Dateien, da sie eine OCR-Verarbeitung benötigen.

OCR ist ein eigenständiger Verarbeitungsprozess. Gescannte Dateien, die Tesseract oder AWS Textract benötigen, brauchen jeweils 15 bis 45 Sekunden und gehören in einen isolierten Arbeitspool, damit sie den schnellen Verarbeitungsweg nicht verlangsamen.

Schritt für Schritt: Naiv gegen optimiert

• PDF-Extraktion: In jedem Fall 2 bis 5 Sekunden pro Dokument, jedoch mit 8 parallelen Arbeitern in einem ThreadPoolExecutor anstelle einer sequenziellen Schleife. Die Verarbeitungsgeschwindigkeit wird durch Wartezeiten auf Festplatte und Netzwerk begrenzt, nicht durch die CPU.

Chunking: von 850 auf 6.200 Chunks pro Sekunde, indem ein vorab kompiliierter tiktoken-Encoder verwendet wird und dieser nicht für jedes Dokument neu initialisiert wird.

Embedding: von 120 auf 14.400 Chunks pro Minute, mit asynchronen Batch-Größen von 256 sowie einer Redis-Cache-Nutzung von 87 %.

FAISS-Indexierung: von 400 auf 8.000 Vektoren pro Sekunde durch GPU-gestützte HNSW-Bauvorgänge.

Anfragedauer (P95): von 820 ms auf 180 ms, bei ef=200, einem vorab aufgewärmten In-Memory-Index sowie Vorfiltern der Metadaten.

Parallelisierte Extraktion

Der Lader verteilt die Dokumente auf einen Thread-Pool:

# loader/parallel_pdf_loader.py

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional
import fitz  # PyMuPDFclass DocumentLoader:
    """
    Parallel PDF extraction using ThreadPoolExecutor.
    PDF parsing is I/O-bound (disk reads + text decode),
    so threads outperform processes here.    Benchmarks (c5.4xlarge, non-OCR PDFs):
      - 1 worker:  ~2–5s per doc, sequential
      - 8 workers: ~2–5s per doc, 8x throughput
      - 16 workers: diminishing returns past disk I/O saturation
    """    def __init__(self, config):
        self.max_workers = config.max_workers  # default: 8    def _extract_single(self, doc_path: str) -> dict:
        """Extract text from a single non-OCR PDF via PyMuPDF."""
        doc = fitz.open(doc_path)
        pages_text = []
        for page in doc:
            pages_text.append(page.get_text("text"))
        doc.close()        full_text = "\n\n".join(pages_text)
        return {
            "doc_id": doc_path,
            "text": full_text,
            "page_count": len(pages_text),
            "source": doc_path,
            "doc_type": "pdf",
        }    def load(
        self,
        doc_paths: List[str],
        doc_ids: Optional[List[str]] = None
    ) -> List[dict]:
        raw_docs, failed = [], []        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_path = {
                executor.submit(self._extract_single, path): path
                for path in doc_paths
            }
            for future in as_completed(future_to_path):
                path = future_to_path[future]
                try:
                    raw_docs.append(future.result())
                except Exception as e:
                    logger.error(f"Extraction failed: {path} → {e}")
                    failed.append(path)        logger.info(
            f"Extracted {len(raw_docs)} docs, "
            f"{len(failed)} failures routed to DLQ"
        )
        return raw_docs

Lektionen aus der Skalierung

Der Aufbau eines produktionsreifen RAG-Systems besteht hauptsächlich aus ETL-Arbeit, wobei sich einige Lektionen besonders hervorheben.

• Die Chunking-Strategie beeinflusst die Qualität der Suche stärker als die Wahl des Modells. Studieren Sie Ihre Dokumenttypen, bevor Sie einen Splitter auswählen.
• Messen Sie die Länge in Tokens und nicht in Zeichen. Verwenden Sie tiktoken. Zeichenzählungen bedeuten für ein Embedding-Modell nichts.
• Entwerfen Sie das Metadatenschema bereits vor dem Schreiben von ETL-Code. An ein bereits indiziertes Korpus können keine Filter hinzugefügt werden, ohne es erneut einzuspeisen.
• Cachen Sie Embeddings intensiv. Mit Redis kosten inkrementelle Verarbeitungen aktualisierter Dokumente 10 bis 15 Prozent der Zeit einer ersten vollständigen Verarbeitung.
• Verwenden Sie eine Dead-Letter-Queue. Ein fehlerhaftes PDF sollte die gesamte Verarbeitung nicht zum Abbruch bringen. Isolieren, protokollieren, benachrichtigen und später erneut verarbeiten.

Parallelisieren Sie die nicht-OCR-basierte Extraktion. Bei 2 bis 5 Sekunden pro PDF über 8 Prozesse können 40.000 Dokumente über Nacht verarbeitet werden. Vermischen Sie niemals die OCR-Verarbeitung von gescannten Dokumenten mit derselben Gruppe.

Betrachten Sie den Indextyp als dauerhaft. Der Wechsel von IndexFlatIP zu IndexHNSWFlat bedeutet einen vollständigen Neubau, daher sollten Sie frühzeitig auf der Grundlage der erwarteten Dokumentenmenge entscheiden.

Instrumentieren Sie ab dem ersten Tag. Um bereits um 3 Uhr morgens herauszufinden, warum die Suchqualität nachgelassen hat, müssen Latenz-Histogramme sowie Warnungen bei der Warteschlangentiefe bereits aktiv sein.