Створення промислового каналу ETL для RAG за допомогою LangChain та FAISS
Розробка схем чанкінгу, імбеддингу та векторного індексування для 40 000 PDF-документів з вимогами до конфіденційності, із можливістю повторних спроб, чергами для некоректних повідомлень, метриками Prometheus та вимірюваними показниками продуктивності.
Джерела: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
Уявіть себе інженером з генеративного ШІ у середнього за розміром фінтех-компанії. Команда з питань дотримання правил володіє понад 40 000 PDF-файлами, які містять інформацію про регуляторні звіти, циркуляри центрального банку, рекомендації регуляторів ринків цінних паперів та внутрішні політики; ці файли розташовані у SharePoint, S3 та локальних дисках. Аналітики щодня витрачають від чотирьох до шести годин на їх ручний пошук. Запит CTO — створити чат-бота, який зможе відповісти на будь-яке запитання щодо цих документів протягом менше трьох секунд.
Це, очевидно, завдання на генерацію з підтримкою пошуку інформації. Однак легко не помітити, що якість відповідей визначається задовго до появи запитання — у процесі ETL, який підготовлює базу знань. Неправильне розділення тексту на фрагменти руйнує контекст, примітивні методи вкладення призводять до зміни значень, а погана індексація уповільнює пошук. Згідно з досвідом, близько 80% проблем із якістю у таких системах насправді є проблемами ETL. Зазвичай вузьким місцем є не сама модель, а сам процес обробки даних.
У цьому проекті використовуються Python, LangChain, FAISS, Pinecone та моделі вкладення від OpenAI.
Масштаби
- Приблизно 40 000 документів у різних форматах: PDF, DOCX, HTML та простий текст.
- Після рекурсивного розділення на фрагменти по 512 токенів у кожному документі утворюється близько 120 фрагментів.
- Векторна індексація у Pinecone з 4,8 мільйона ембеддингів у 1536 вимірах.
Архітектура конвеєру
Це класична схема «витягування, перетворення, завантаження». Кожен етап приховує специфічну складність домену, яку легко недооцінити.
┌────────────────────────────────────────────────────────────────────────────┐
│ 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 │
└────────────────────────────────────────────────────────────────────────────┘
Кожен етап виконується як окремий ідемпотентний процес та записує дані у тимчасовий сховище перед початком наступного етапу. Це забезпечує стійкість до несправностей. Якщо процес вбудовування даних зупиниться на документі 30 000, ніщо не змушує переекстрагувати всі 40 000 документів.
Оркестратор – це виконувач етапів, який керується класом dataclass:
# 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
Діаграма потоку як джерело істини
ASCII-діаграма потоку відображає кожну точку прийняття рішень — від отримання сирого документа, через повторні спроби та групування, до остаточного зберігання у сховищі векторних даних.
┌──────────────────────┐
│ 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 ✓ │
└─────────────────────────┘
Кожен діамант у цьому діаграматичному зображенні відповідає реальній гілці if/else у коді. Спочатку її накреслення робить діаграму потоку схемою архітектури ще до створення будь-якої реалізації, а не просто документацією.
Блокування
Блокування може бути найвпливовішим рішенням у всьому процесі обробки. Блок — це одиниця отримання даних. Якщо зробити його занадто великим, ви марнуєте простір контексту на нерелевантний текст; якщо занадто малим — сенс розпадається. Не існує універсально правильного розміру блоку.
Розділення фіксованого розміру
Найпростіший підхід — розділяти на кожні N символів або токенів, за бажанням з перекриттям. Він підходить для однорідних даних, таких як журнали та таблиці, але руйнує межі речень та абзаців у прозовому тексті.
# 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)
]
Рекурсивне розділення символів (за замовчуванням)
Це основний метод. Спочатку пробується найбільш змістовний розділювач (порожній рядок, потім новий рядок, потім крапка, потім пробіл, потім один символ), і лише у разі необхідності вдається до більш жорсткого розділення. Його реалізує RecursiveCharacterTextSplitter від LangChain. Адаптивна версія нижче підраховує кількість токенів за допомогою tiktoken, тож жоден фрагмент не перевищує ліміт у 8192 токенів для моделі ембеддингу.
# 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
Семантичне чанкування
Тут кожне речення вбудовується, а текст розділяється там, де косинусова схожість між сусідніми елементами різко знижується. Чанки є послідовними, проте процес виконується на 10–15 разів повільніше та вимагає додаткових викликів для формування чанків, що фактично подвоює використання API. Це має сенс лише для документів з високою цінністю у вигляді оповідей, таких як довгі юридичні висновки, щорічні звіти та документи з політики.
# 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)
Як вони порівнюються:
- Фіксованого розміру: найкраще підходить для коду, журналів подій та табличних даних. Швидкий та найдешевший варіант, але з низькою кохерентністю.
- Рекурсивний (за замовчуванням): у більшості випадків найкраще підходить для PDF, DOCX та HTML. Швидкий та дешевий, але з середньою кохерентністю.
- Семантичний: найкраще підходить для довгих юридичних та політичних документів. Висока кохерентність, але швидкість обробки на 10–15 разів нижча та кількість викликів API приблизно вдвічі більша.
Генерація ембеддингів
Ембеддинг перетворює фрагмент тексту на щільний вектор, який відображає його значення. Ключові питання при розробці — яку модель використовувати, як обробляти дані пакетами, чи потрібно використовувати кеш та як асинхронно працювати з обмеженнями швидкості.
Арифметика обмежень швидкості є приголомшливою. Модель OpenAI text-embedding-3-small дозволяє обробляти мільйон токенів на хвилину у другому рівні. Якщо взяти 4,8 мільйона фрагментів по 300 токенів кожен, загальна кількість становить 1,44 мільярда токенів. Послідовна обробка за цим лімітом зайняла б близько 24 годин, тоді як асинхронна обробка з паралельними працівниками скорочує час до приблизно двох з половиною годин.
# 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))
Правило, яке потрібно дотримуватися, — це ніколи не вбудовувати один і той самий фрагмент двічі. Кеш, підтримуваний Redis, скорочує кількість інкрементальних обробок на 70–90 відсотків у колекціях, які рідко змінюються.
Індексування
Під час індексування фрагменти потрапляють у пошуковий сховище. Тут вирішуються два основні питання: схема метаданих (для фільтрованого пошуку) та тип індексу (Flat, HNSW або IVF).
Метадані відрізняють просту систему RAG від професійної. Візьмемо запит щодо того, що сказав регулятор цінних паперів про інсайдерську торгівлю у 2023 році. Система повинна спочатку фільтрувати за параметрами authority=SEBI та year=2023, перш ніж взагалі проводити пошук за принципом найближчого сусіда, що зменшує обсяг пошуку та підвищує точність.
# 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 для локальних рішень
Коли дані не можуть бути передані до хмарної векторної бази даних, що є поширеним явищем у банківській сфері, фінансах та урядових структурах, FAISS є стандартним рішенням. Вибір типу індексу має велике значення.
# 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підходить для корпусів з менше ніж 1 мільйоном векторів, забезпечує точний (100%) рівень відтворення даних, працює з середньою швидкістю та використовує багато пам’яті.IndexHNSWFlatпідходить для корпусів з 1 до 10 мільйонів векторів, досягає приблизно 98,5% рівня відтворення даних@10, швидко обробляє запити та використовує середню кількість пам’яті.IndexIVFPQпідходить для корпусів з більш ніж 10 мільйонами векторів, досягає рівня відтворення даних від 94 до 96%, працює дуже швидко та використовує мало пам’яті завдяки компресії 4–8 біт.
Обробка помилок та повторні спроби
Кожен етап виробничого процесу іноді може зазнати невдачі. Важливо робити це граційно: відокремлювати проблемні документи, не зупиняючи виконання процесу, та залишати чіткий запис для ручної перевірки. Для цього використовується черга для непереданих повідомлень, яка позначає кожну помилку етапом, на якому вона сталась.
# 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
Моніторинг
Канал обробки даних без метрик є „чорною скринькою“. Відстежуйте принаймні пропускну здатність, затримку, частоту помилок та вартість обробки кожного документа; ці дані передаються до Prometheus та візуалізуються у Grafana.
# 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)
Корисні панелі панелі керування:
- Кількість документів за хвилину протягом часу.
- Гістограма затримки вбудовування P95.
- Коефіцієнт успішних запитів до кешу Redis, мета — понад 80%.
- Глибина черги документів з помилками за етапами, сповіщення, якщо вона не дорівнює нулю протягом більше п’яти хвилин.
- Витрати на один документ, тобто кількість токенів помножена на ціну за токен.
- Затримка запитів FAISS або Pinecone під час отримання даних (P50, P95, P99).
Отримані результати
Ці показники надані системою контролю відповідності після налаштувань на сервері AWS c5.4xlarge (16 vCPU, 32 GB оперативної пам’яті) із асинхронними працівниками для вбудовування даних.
Вилучення PDF
PyMuPDF без OCR витрачає від 2 до 5 секунд на один документ залежно від кількості сторінок та форматування, причому робота паралелізується ефективно. З 8 одночасними працівниками вилучення 40 000 PDF займає приблизно:
40,000 docs × avg 3.5s / 8 workers = ~4.9 hours wall-clock
(scales linearly — 16 workers → ~2.4 hours)
За типом файлу: простий круговий документ з 1 до 5 сторінок обробляється за 2–3 секунди, оскільки шар тексту є чистим. Документ-довідник з 20 до 50 сторінок із кількома колонками та вбудованими таблицями обробляється за 3–4 секунди. Щорічний звіт із понад 100 сторінок обробляється за 4–5 секунд, оскільки потрібно пропустити щільні макети та багато зображень. Шифровані чи відскановані PDF-файли потрапляють у чергу для обробки, оскільки їм потрібен OCR.
OCR — це окремий процес. Відскановані файли, які потребують програми Tesseract чи AWS Textract, обробляються по 15–45 секунд кожен та мають бути у ізольованому пулі працівників, щоб вони не уповільнювали швидкий процес.
Етап за етапом: простий підхід порівняно з оптимізованим
- Вилучення даних з PDF: у будь-якому випадку 2–5 секунд на документ, але з використанням 8 паралельних працівників у
ThreadPoolExecutorзамість послідовного циклу. Робота обмежується часом очікування на диску та мережі, а не потужністю CPU.
Паралельне вилучення
Завантажувач розподіляє документи між пулом потоків:
# 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
Уроки роботи в масштабі
Створення продакшн-систем типу RAG полягає переважно у роботі з даними за методом ETL, і є кілька ключових уроків.
- Стратегія розділення даних впливає на якість пошуку більше, ніж вибір моделі. Дослідіть типи документів перед тим, як обрати інструмент для їх розділення.
- Вимірюйте довжину у токенах, а не у символах. Використовуйте
tiktoken. Кількість символів не має значення для моделей ембеддингу. - Спроєктуйте схему метаданих ще до написання коду ETL. До вже індексованого корпусу неможливо додати фільтри без повторної обробки даних.
- Активно використовуйте кешування ембеддингів. За допомогою Redis інкрементальна обробка оновлених документів коштує від 10 до 15 відсотків у порівнянні з першою повною обробкою.
- Використовуйте чергу для некоректних повідомлень. Один несправний PDF не повинен зупинити весь процес. Ізолюйте його, задокументуйте, надішліть сповіщення та обробіть пізніше.
IndexFlatIP до IndexHNSWFlat означає повну перебудову, тому потрібно заздалегідь вирішити це, враховуючи розмір проєктованого корпусу даних.