02 — RAG Infrastructure Pipeline at Scale

Role: AI Specialist / Senior AI Engineer
Problem: Design a RAG pipeline over 100 GB of heterogeneous infrastructure documents that achieves > 90% retrieval precision with sub-2s retrieval latency
Key challenges: Multiple file formats, OCR required, Arabic + English, incremental updates, metadata-filtered retrieval

1. Clarifying Questions

Data characteristics

• File formats? (PDFs — some scanned, some text-native; Excel; Word/DOCX; JSON metadata exports from GIS; XML SCADA exports)
• Languages? (English-primary, some Arabic documents and bilingual)
• Update frequency? (New documents weekly; revisions to existing docs monthly; some real-time log exports daily)
• Are documents classified? (Some are RESTRICTED — must not be mixed with UNCLASSIFIED in retrieval)

Query characteristics

• Who queries? (Engineers, operators, project managers — different vocabulary levels)
• Query types? (Factual: "what is the max flow rate for PUMP-007?"; Procedural: "how do I replace the bearing?"; Regulatory: "what standard governs pipeline inspection intervals?")
• Expected QPS? (50 users, ~0.5 QPS peak — but batch ingest may spike to 5 QPS during corpus refresh)

SLOs

• Retrieval precision (fraction of top-4 chunks that are relevant): > 90%
• End-to-end retrieval latency (embed + search + rerank): < 2 seconds
• Corpus freshness: new documents searchable within 48 hours of upload

2. Capacity Estimation

Corpus size

• 100 GB PDFs; typical infrastructure PDF: 1–5 MB per document → ~20,000–100,000 documents
• Average extracted text per document: 5,000–15,000 characters
• Use conservative estimate: 50,000 docs × 8,000 chars = 400M characters
• Chunks at 1,000 chars, 150 overlap (stride 850): 400M / 850 ≈ 470,000 chunks

Index size

• Multilingual-e5-large: 1024 dims × 4 bytes × 470,000 = 1.9 GB in-memory
• Plus metadata overhead (~500 bytes/chunk): ~235 MB
• Total index: ~2.1 GB — fits comfortably in 16 GB server RAM

Ingest time

• multilingual-e5-large on CPU: ~50ms/chunk → 470K chunks = 6.5 hours serial
• With 8 parallel workers: ~50 minutes for full re-ingest
• Incremental (only changed docs, typically 500 docs/week): ~3 minutes

OCR overhead

• Scanned PDFs: ~2–5 seconds per page with Tesseract
• Assume 20% of docs are scanned, 20 pages avg: 10,000 docs × 20 pages × 3s = 167 hours with single-threaded OCR
• With 8 workers: ~21 hours — schedule during weekend maintenance window
• After initial OCR, incremental OCR is fast (only new scanned docs)

3. High-Level Architecture

INGESTION PIPELINE (offline, scheduled)
══════════════════════════════════════════════════════════════════════

SharePoint        GIS Portal       SCADA Historian   Local disk
(PDFs, DOCX)     (JSON, XML)      (CSV exports)     (manuals)
      │                │                 │                │
      └────────────────┴─────────────────┴────────────────┘
                                │
                                ▼
                    ┌─────────────────────┐
                    │   Document Watcher  │  polls every hour
                    │   (hash-based dedup)│  new hash → ingest
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Parser / OCR      │
                    │   ├── PyPDF2        │  text-native PDF
                    │   ├── Tesseract     │  scanned PDF
                    │   ├── python-docx   │  DOCX
                    │   ├── openpyxl      │  Excel (→ markdown table)
                    │   └── json/xml      │  GIS metadata
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Chunker           │
                    │   ├── recursive     │  text docs
                    │   └── structural    │  headers, tables
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │  Metadata Extractor │
                    │  {doc_type, lang,   │
                    │   classification,   │
                    │   asset_tag, date}  │
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Embedder          │  multilingual-e5-large
                    │   8× parallel       │  ONNX Runtime (CPU)
                    │   ~50ms/chunk       │
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │  ChromaDB (local)   │  upsert by doc_id+chunk_idx
                    │  HNSW, cosine       │  delete old → insert new
                    └─────────────────────┘


QUERY PIPELINE (online, real-time)
══════════════════════════════════════════════════════════════════════

User Query (Arabic or English)
      │
      ▼
Language Detection (~0ms)
      │
      ▼
Multilingual-e5-large embed (~15ms)           BM25 index build (~0ms, pre-built)
      │                                              │
      ▼                                              ▼
ChromaDB cosine search, top-20 (~40ms)      BM25 keyword search, top-20 (~5ms)
      │                                              │
      └──────────────────┬───────────────────────────┘
                         ▼
            Reciprocal Rank Fusion (RRF) → top-20 unique chunks
                         │
                         ▼
            Cross-encoder reranker → top-4 (~300ms)
                         │
                         ▼
            Build prompt with numbered citations
                         │
                         ▼
            LLM generation (streaming)

4. Deep Dives

4.1 Multi-Format Document Parsing

Each file type requires a different extraction strategy:

PDF (text-native):

import pypdf

def extract_pdf_text(path: str) -> list[dict]:
    reader = pypdf.PdfReader(path)
    pages = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text()
        if len(text.strip()) < 50:  # probably scanned
            text = ocr_page(path, i)
        pages.append({"page": i+1, "text": text})
    return pages

PDF (scanned) — Tesseract OCR:

import pytesseract
from pdf2image import convert_from_path

def ocr_page(pdf_path: str, page_num: int) -> str:
    images = convert_from_path(pdf_path, first_page=page_num+1, last_page=page_num+1, dpi=300)
    # For Arabic + English: use both language packs
    return pytesseract.image_to_string(images[0], lang="eng+ara", config="--psm 3")

Excel (convert to markdown tables for better chunking):

import openpyxl

def excel_to_markdown(path: str) -> str:
    wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
    result = []
    for sheet in wb.sheetnames:
        ws = wb[sheet]
        rows = list(ws.iter_rows(values_only=True))
        if not rows:
            continue
        headers = [str(h) if h else "" for h in rows[0]]
        result.append(f"## Sheet: {sheet}\n")
        result.append("| " + " | ".join(headers) + " |\n")
        result.append("| " + " | ".join(["---"] * len(headers)) + " |\n")
        for row in rows[1:]:
            result.append("| " + " | ".join(str(v) if v else "" for v in row) + " |\n")
    return "".join(result)

Why markdown tables? Chunking on markdown table rows preserves row context. Chunking raw CSV ("A1,B1,C1\nA2,B2,...") strips column headers from middle rows — the most common RAG data quality bug for tabular data.

4.2 Hybrid Search with Reciprocal Rank Fusion

Why hybrid? Dense (embedding) search excels at paraphrase and concept matching. Sparse (BM25) search excels at exact-term matching — critical for:

• Asset IDs: "PUMP-007" must be found even in documents that discuss it by ID
• Standards numbers: "ISO 9001", "IEC 61511"
• Proper nouns: "Al Ain Water Distribution System"

Dense search may embed these as conceptually similar to other IDs (all pump IDs cluster together), losing the specificity.

BM25 scoring for term $t$ in document $d$ with corpus of $N$ documents:

$$\text{BM25}(t, d) = \text{IDF}(t) \cdot \frac{f(t,d) \cdot (k_1 + 1)}{f(t,d) + k_1 \cdot \left(1 - b + b \cdot \frac{|d|}{\text{avgdl}}\right)}$$

Where $\text{IDF}(t) = \ln!\left(\frac{N - n_t + 0.5}{n_t + 0.5} + 1\right)$, $k_1 = 1.5$, $b = 0.75$ (standard defaults).

Reciprocal Rank Fusion:

def rrf(rankings: list[list[str]], k: int = 60) -> list[str]:
    """Fuse multiple ranked lists. k=60 is the standard constant."""
    scores: dict[str, float] = defaultdict(float)
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, 1):
            scores[doc_id] += 1.0 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)

Why RRF instead of score normalisation? RRF is robust to calibration differences between dense and sparse scores. Cosine similarity values (0.2–0.9) and BM25 scores (0–20) are on incomparable scales. RRF uses only rank positions, which are always comparable.

4.3 Cross-Encoder Reranker

A cross-encoder processes (query, chunk) pairs together — it sees both simultaneously and can capture fine-grained relevance signals that bi-encoders miss.

Bi-encoder (retrieval):
  embed(query) → 1024-dim vector
  embed(chunk) → 1024-dim vector
  similarity = dot product of two vectors computed independently

Cross-encoder (reranking):
  score(query + [SEP] + chunk) → relevance score
  Both texts processed together through all transformer layers
  Much more accurate; too slow for full corpus (~500ms for 20 candidates)

For Arabic + English: use cross-encoder/msmarco-MiniLM-L-6-v2 (multilingual variant) or fine-tune on domain pairs if budget allows.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", max_length=512)

def rerank(query: str, candidates: list[dict]) -> list[dict]:
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, candidates), reverse=True)
    return [c for _, c in ranked[:4]]  # top-4

Latency budget: 20 candidates × 512 tokens × cross-encoder → ~300ms on CPU (MiniLM-L-6 is small). Total retrieval pipeline: 15 (embed) + 40 (HNSW) + 5 (BM25) + 300 (rerank) = 360ms. Well within 2s SLO.

4.4 Metadata Filtering

At 470K chunks, metadata filtering is essential to prevent cross-contamination — a procedure manual's steps being returned for a factual engineering question.

Metadata schema per chunk:

{
    "doc_id": "pipeline-ops-manual-v3.2",
    "chunk_idx": 14,
    "doc_type": "maintenance_manual",      # or "standard", "procedure", "drawing", "alarm_report"
    "asset_tag": "PUMP-007",               # or null if not asset-specific
    "language": "en",                      # or "ar", "ar-en" (bilingual)
    "classification": "UNCLASSIFIED",      # or "RESTRICTED"
    "date": "2024-03-15",
    "page": 12,
    "section": "3.4 Bearing Replacement"
}

Filtered query examples:

# "What does the maintenance procedure say about replacing bearings?"
collection.query(
    query_embeddings=[embed(query)],
    n_results=20,
    where={"doc_type": {"$in": ["maintenance_manual", "procedure"]}}
)

# "What is the current inspection standard for gas pipelines?"
collection.query(
    query_embeddings=[embed(query)],
    n_results=20,
    where={"doc_type": "standard"}
)

# Security: never mix classification levels in retrieval
collection.query(
    query_embeddings=[embed(query)],
    n_results=20,
    where={"classification": user.clearance_level}  # UNCLASSIFIED users never see RESTRICTED
)

The router populates the filter: the query classifier doesn't just route to "doc_question" — it also identifies the likely document type and asset context to narrow retrieval.

4.5 Incremental Ingest and Corpus Freshness

The corpus must update without full re-ingest (which takes ~50 minutes).

Hash-based change detection:

import hashlib, json
from pathlib import Path

MANIFEST_PATH = Path("corpus/manifest.json")

def load_manifest() -> dict:
    if MANIFEST_PATH.exists():
        return json.loads(MANIFEST_PATH.read_text())
    return {}

def doc_hash(path: str) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()[:16]

def get_changed_docs(doc_dir: str) -> tuple[list, list]:
    manifest = load_manifest()
    current = {f: doc_hash(f) for f in Path(doc_dir).rglob("*") if f.is_file()}
    
    new_or_changed = [f for f, h in current.items() if manifest.get(f) != h]
    deleted = [f for f in manifest if f not in current]
    
    return new_or_changed, deleted

Upsert logic (delete + re-embed for changed docs):

def incremental_ingest(doc_dir: str):
    new_docs, deleted_docs = get_changed_docs(doc_dir)
    
    # Delete stale chunks
    for doc_id in deleted_docs + new_docs:  # new docs: delete old version first
        collection.delete(where={"doc_id": doc_id})
    
    # Re-embed changed/new docs
    for doc_path in new_docs:
        chunks = parse_and_chunk(doc_path)
        embeddings = embedder.encode([c["text"] for c in chunks])
        collection.add(
            ids=[f"{doc_id}_{i}" for i, _ in enumerate(chunks)],
            embeddings=embeddings,
            documents=[c["text"] for c in chunks],
            metadatas=[c["metadata"] for c in chunks]
        )
    
    # Update manifest
    update_manifest(new_docs, deleted_docs)
    log(f"Ingest complete: {len(new_docs)} updated, {len(deleted_docs)} deleted")

Scheduling: run hourly during business hours via Windows Task Scheduler or cron. Alert if no successful ingest in 72 hours.

5. Trade-offs and Alternatives

6. Monitoring

Key metrics to track:

• Retrieval precision: fraction of top-4 chunks judged relevant by LLM for weekly sample of 50 queries. Target: > 0.90.
• Corpus freshness: time since last successful ingest. Alert if > 72h.
• Missing asset tags: chunks with asset_tag=null that contain an asset ID in the text → metadata extraction failure. Review weekly.
• Query distribution by doc_type: if most queries route to "maintenance_manual" but that category has low precision → review chunking strategy for that doc type.
• BM25 hit rate: what % of returned chunks came from BM25-only (missed by dense)? If < 5%, BM25 may not be adding value. If > 30%, dense embeddings may be poorly calibrated for this domain.

7. Interview Talking Points

1. "Hybrid search is not optional for infrastructure data." Asset IDs, standards numbers, and proper nouns are the queries that matter most for safety-critical decisions. Dense search misses them. This is not theoretical — the first production failure I plan for is "PUMP-007 not found" because the embedding model conflated it with PUMP-008.

2. "OCR is the unsexy bottleneck that determines your corpus quality." At 20% scanned PDFs, bad OCR doubles the chunk error rate. I run a quality check: randomly sample 20 OCR-extracted pages, manually verify accuracy, and track character error rate. If CER > 5%, I tune Tesseract parameters or switch to PaddleOCR.

3. "Metadata is the filter that makes retrieval trustworthy at scale." Without metadata, a query about maintenance procedures retrieves regulatory standards and alarm history — all technically "relevant" by embedding distance but wrong in context. The metadata schema is part of the system design, not an afterthought.