Создание промышленной пайплайн-системы 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 и локальных дисках. Аналитики тратят от четырех до шести часов в день на их ручной поиск. Технический директор просит создать чат-бота, который сможет ответить на любой вопрос, связанный с соблюдением нормативов, на основе этих файлов менее чем за три секунды.
Очевидно, что речь идет о задаче генерации с использованием поиска информации. Что часто упускается из виду, так это то, что качество ответов определяется задолго до поступления вопроса — еще на этапе 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 ГБ ОЗУ) с асинхронными процессами встраивания.
Извлечение данных из 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 подразумевает полную перестройку индекса, поэтому решение следует принять заранее, исходя из ожидаемого размера корпуса данных.