Budowa pipeline’u ETL do produkcji w technologii RAG przy użyciu LangChain i FAISS.
Zaprojektowanie mechanizmów dzielenia na fragmenty, wplecania i indeksowania wektorowego dla 40 000 plików PDF z wymogami prawnymi, wraz z mechanizmami ponawiania prób, kolejkami wiadomości nieudanych, metrykami Prometheus oraz mierzonymi wskaźnikami wydajności.
Źródła: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
Załóżmy, że jesteś inżynierem GenAI w średniej firmie z sektora fintech. Zespół ds. zgodności posiada ponad 40 000 plików PDF, obejmujących dokumenty regulacyjne, obwieszczenia banku centralnego, wytyczne organów nadzoru nad rynkami finansowymi oraz wewnętrzne zasady, rozmieszczone na SharePoint, S3 i lokalnych dyskach. Analitycy spędzają od czterech do sześciu godzin dziennie na ręcznym przeszukiwaniu tych plików. Prośba CTO dotyczy bota konwersacyjnego, który potrafi odpowiedzieć na każde pytanie związane z zgodnością z tych plików w mniej niż trzy sekundy.
Oczywiście jest to zadanie typu Retrieval-Augmented Generation. Trudno zauważyć, że jakość odpowiedzi jest określana na długo przed pojawieniem się pytania, w procesie ETL przygotowującym bazę wiedzy. Niewłaściwe dzielenie tekstu na fragmenty niszczy kontekst, proste metody embedding sprawiają, że znaczenie ulega zmianie, a słaba indeksacja spowalnia proces wyszukiwania. Z doświadczenia wynika, że około 80% problemów z jakością w RAG to w rzeczywistości problemy ETL. Zazwyczaj to nie model stanowi wąskie gardło, lecz cały proces.
Użyte narzędzia to Python, LangChain, FAISS, Pinecone oraz embeddingi od OpenAI.
Masowa skala
- Około 40 000 dokumentów w różnych formatach: PDF, DOCX, HTML i tekst prosty.
- Mniej więcej 120 fragmentów na dokument po rekurencyjnym dzieleniu na kawałki po 512 tokenach.
- Indeks wektorowy zawierający 4,8 miliona embeddingów w Pinecone, o wymiarowości 1536.
Architektura pipeline’a
Klasyczny schemat obejmuje kroki wyodrębniania, przekształcania i ładowania. Każdy z tych etapów ukrywa specyficzną dla danej dziedziny złożoność, którą łatwo jest niedocenić.
┌────────────────────────────────────────────────────────────────────────────┐
│ 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 │
└────────────────────────────────────────────────────────────────────────────┘
Każdy etap działa jako odrębny proces idempotentny i zapisuje dane do przechowalni tymczasowej przed rozpoczęciem następnego. Dzięki temu osiąga się tolerancję na awarie. Jeśli proces wstawiania danych zakończy się przy dokumencie numer 30 000, nic nie zmusza do ponownego wydobywania wszystkich 40 000 dokumentów.
Orkiestrator to narzędzie do uruchamiania etapów oparte na klasach danych:
# 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
Schemat przepływu jako źródło prawdy
Schemat przepływu w formacie ASCII przedstawia każdy punkt decyzyjny, począwszy od przyjmowania surowych dokumentów, przez próby ponowne i grupowanie zadań, aż po ostateczne zapisanie ich w magazynie wektorowym.
┌──────────────────────┐
│ 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 ✓ │
└─────────────────────────┘
Każdy diament na tym schemacie odpowiada rzeczywistej gałęzi if/else w kodzie. Jej narysowanie najpierw sprawia, że schemat przepływu staje się punktem odniesienia architektury, zanim powstanie jakakolwiek implementacja, a nie tylko dokumentacją.
Dzielenie na fragmenty
Dzielenie na fragmenty może być najważniejszą decyzją w całym procesie. Fragment to jednostka pobierania danych. Jeśli będzie zbyt duży, marnujemy przestrzeń kontekstową na nieistotny tekst; jeśli zbyt mały, sens treści ulega rozproszeniu. Nie ma jednego uniwersalnie poprawnego rozmiaru fragmentu.
Dzielenie o stałej wielkości
Najprostsze podejście polega na dzieleniu co N znaków lub tokenów, opcjonalnie z nakładaniem się. Pasuje do jednolitych danych, takich jak logi i tabele, ale niszczy granice zdań i akapitów w tekście prozatorskim.
# 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)
]
Rekurencyjne dzielenie na znaki (standardowe)
To rozwiązanie najczęściej wykorzystywane. Najpierw próbuje najbardziej odpowiedniego separatora (linia pusta, następnie nowa linia, potem kropka, następnie spacja, a na końcu pojedynczy znak) i tylko w razie konieczności ucieka się do bardziej agresywnego dzielenia. Implementację tego rozwiązania dostarcza RecursiveCharacterTextSplitter z LangChain. Poniższa wersja adaptacyjna liczy tokeny za pomocą tiktoken, dzięki czemu żaden fragment nie przekracza limitu 8192 tokenów modelu embeddingowego.
# 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
Semantyczne dzielenie na fragmenty
Tutaj każde zdanie jest włączane do struktury, a tekst dzieli się w tych miejscach, gdzie kąt podobieństwa kosinowego pomiędzy sąsiadującymi fragmentami gwałtownie spada. Fragmenty te są spójne, ale proces ten jest 10 do 15 razy wolniejszy i wymaga dodatkowych wywołań służących do włączania danych tylko po to, by podzielić tekst na fragmenty, co faktycznie podwaja zużycie API. Ma to sens tylko w przypadku dokumentów narracyjnych o dużej wartości, takich jak długie opinie prawne, raporty roczne i dokumenty polityczne.
# 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)
Jak się porównują:
- O stałej wielkości: najlepszy do kodu, logów i danych tabelarycznych. Szybki i najtańszy, ale o niskiej spójności.
- Rekurencyjny (domyślny): najlepszy w większości przypadków do plików PDF, DOCX i HTML. Szybki i tani, z umiarkowaną spójnością.
- Semantyczny: najlepszy do długich dokumentów prawniczych i regulaminowych. Wysoka spójność, ale 10 do 15 razy wolniejszy oraz wymagający około dwukrotnie więcej wywołań API.
Generowanie embeddingów
Embedding przekształca fragment tekstu w gęsty wektor, który oddaje jego znaczenie. Kluczowe pytania projektowe dotyczą wyboru modelu, sposobu grupowania danych, konieczności cacheowania oraz asynchronicznego radzenia sobie z ograniczeniami szybkości.
Matematyka ograniczeń szybkości jest przygnębiająca. Model text-embedding-3-small firmy OpenAI umożliwia użycie miliona tokenów na minutę w poziomie Tier 2. Przy 4,8 miliona fragmentów o średniej długości 300 tokenów łączna liczba wynosi 1,44 miliarda tokenów. Proces embeddingu sekwencyjnego przy tym limicie zajęłby około 24 godzin, natomiast zastosowanie batchowania asynchronicznego z pracownikami równoległymi skraca ten czas do około dwóch i pół godziny.
# 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))
Zasada, którą należy stosować, brzmi: nigdy nie umieszczać tego samego fragmentu dwa razy. Pamięć cache oparta na Redis redukuje czas wykonywania zadań incrementalnych o 70 do 90 procent w przypadku kolekcji, które rzadko się zmieniają.
Indeksowanie
Indeksowanie to proces, w którym fragmenty trafiają do magazynu umożliwiającego wyszukiwanie. Decydują o tym dwie kwestie: schemat metadanych (do filtrowania wyników) oraz typ indeksu (Flat, HNSW lub IVF).
Metadane odróżniają prosty system RAG od systemu produkcyjnego. Weźmy na przykład pytanie o to, co organ regulacyjny ds. papierów wartościowych powiedział na temat handlu informacjami niejawnymi w 2023 roku. System powinien najpierw przefiltrować wyniki według kryteriów authority=SEBI i year=2023, zanim w ogóle przeprowadzi wyszukiwanie najbliższych sąsiadów, co zmniejsza obszar poszukiwań i poprawia dokładność.
# 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 dla rozwiązań lokalnych
Gdy dane nie mogą trafić do chmurowej bazy wektorowej, co jest powszechne w bankowości, finansach i administracji publicznej, FAISS stanowi standardowe rozwiązanie. Wybór typu indeksu ma duże znaczenie.
# 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)
IndexFlatIPnadaje się do korpusów z mniej niż 1 milionem wektorów, zapewnia dokładną (100%) powtarzalność wyników, działa z umiarkowaną prędkością i wymaga dużej ilości pamięci.IndexHNSWFlatnadaje się do korpusów z od 1 do 10 milionów wektorów, osiąga około 98,5% powtarzalności wyników@10, szybko obsługuje zapytania i wymaga umiarkowanej ilości pamięci.IndexIVFPQnadaje się do korpusów z ponad 10 milionami wektorów, osiąga od 94 do 96% powtarzalności wyników, działa bardzo szybko i wymaga niewielkiej ilości pamięci dzięki kompresji 4 do 8 bitów.
Zarządzanie błędami i ponawianie prób
Każdy etap procesu produkcyjnego czasami może zawieść. Ważne jest, aby błędy występowały w sposób uporządkowany: należy odizolować nieprawidłowe dokumenty bez przerywania całego procesu oraz zachować wyraźny zapis działań w celu manualnej analizy. Do tego służy kolejka wiadomości błędowych, która oznacza każdy błąd etapem, w którym się on wystąpił.
# 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
Monitorowanie
Przepływ danych bez żadnych metryk to czarna skrzynka. Należy śledzić przynajmniej przepustowość, opóźnienie, wskaźnik błędów oraz koszt za dokument, które są udostępniane w Prometheus i wykresowane w Grafanie.
# 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)
Korzystne panele na panelu sterowania:
- Liczba dokumentów na minutę w czasie.
- Histogram latencji embedowania P95.
- Wskaźnik trafień w pamięci cache Redis, celujący powyżej 80%.
- Głębokość kolejki wiadomości błędnych według etapu, z powiadamianiem w przypadku wartości różnej od zera przez ponad pięć minut.
- Koszt za dokument, czyli liczba tokenów pomnożona przez cenę jednego tokena.
- Latencja zapytań FAISS lub Pinecone podczas wyszukiwania (P50, P95, P99).
Wyniki pomiarów
Te wartości pochodzą z systemu kontroli zgodności po dostosowaniach, na serwerze AWS c5.4xlarge (16 vCPU, 32 GB pamięci RAM) z asynchronicznymi procesorami do embedowania.
Ekstrakcja PDF
PyMuPDF bez OCR zajmuje od 2 do 5 sekund na dokument, w zależności od liczby stron i układu, a praca może być realizowana równolegle. Przy 8 jednoczesnych procesorach ekstrakcja 40 000 plików PDF zajmuje około:
40,000 docs × avg 3.5s / 8 workers = ~4.9 hours wall-clock
(scales linearly — 16 workers → ~2.4 hours)
Według typu pliku: prosty krąglicowy dokument składający się z 1 do 5 stron zajmuje od 2 do 3 sekund, ponieważ warstwa tekstu jest prosta. Przewodnik liczący od 20 do 50 stron z wieloma kolumnami i wstawionymi tabelami potrzebuje od 3 do 4 sekund. Roczny raport liczący ponad 100 stron zajmuje od 4 do 5 sekund, ze względu na konieczność pominięcia złożonych układów i licznych obrazków. Zaszyfrowane lub zeskanowane pliki PDF trafiają do kolejki błędów, ponieważ wymagają OCR.
OCR to odrębny proces. Zeskanowane pliki, które wymagają narzędzi Tesseract lub AWS Textract, potrzebują po 15 do 45 sekund każdy i powinny znajdować się w izolowanej grupie zadań, aby nie spowalniały szybkiego procesu.
Krok po kroku: prosty sposób versus optymalizacja
- Ekstrakcja z PDF: w obu przypadkach 2 do 5 sekund na dokument, ale z użyciem 8 równoległych pracowników w
ThreadPoolExecutorzamiast sekwencyjnego pętli. Praca jest ograniczona czasem oczekiwania na dysku i sieci, a nie przez moc procesora.
Wydobywanie równoległe
Loader rozkłada dokumenty wśród grup wątków:
# 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
Lekcje z wdrażania w dużych skali
Budowa systemu RAG do użycia w produkcji to przede wszystkim inżynieria ETL, przy czym kilka kwestii wyróżnia się szczególnie.
- Strategia dzielenia na fragmenty ma większy wpływ na jakość wyszukiwania niż wybór modelu. Przed wybraniem narzędzia do dzielenia dokumentów przeanalizuj ich rodzaje.
- Mierz długość w tokenach, a nie w znakach. Użyj
tiktoken. Liczba znaków nie ma żadnego znaczenia dla modelu embeddingów. - Zaprojektuj schemat metadanych przed napisaniem jakiegokolwiek kodu ETL. Do już zindeksowanego korpusu nie można dodać filtrów bez ponownego jego indeksowania.
- Agresywnie cacheuj embeddingi. Z użyciem Redis, iteracyjne przetwarzanie zaktualizowanych dokumentów kosztuje od 10 do 15 procent czasu potrzebnego na pierwsze pełne przetworzenie.
- Użyj kolejki wiadomości błędowych. Jeden uszkodzony plik PDF nie powinien sparaliżować całego procesu. Izoluj go, zapisz informacje o błędzie, wyślij powiadomienie i przetrzyj go później.
IndexFlatIP na IndexHNSWFlat oznacza pełną rekonstrukcję, więc należy wcześnie podjąć decyzję na podstawie przewidywanej wielkości korpusu.