Прадакшн-ETL-канвеер для RAG на LangChain і FAISS
Праектуем разбіўку, эмбедынгі і вектарную індэксацыю для 40 000 PDF па камплаенсе з паўторамі, чэргамі нядастаўленых паведамленняў, метрыкамі Prometheus і вымеранымі бенчмаркамі.
Крыніцы: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
Уявіце сябе інжынерам GenAI у мяжовай фінтэх-компаніі. Команда з пытанняў салідарнасці адпаведае за аднароджэнне болей чым 40 000 PDF-файлов, якія включаюць дакументы, неабходныя для салідарнасці, цыркуляры цэнтральнага банку, рэкамендаціі рэгулятора рынку цэнных папероў і внутрашнія правілы; гэтыя файлы розмешчаны на SharePoint, S3 і локальных драйвах. Аналітыкі ўсё дзень вітрачаюць чатыры-шасць гадзін на ўручную пошук у гэтых файлах. Керавальнік тэхнічных задачаў просі стварыць чат-бота, які зможаў бы адпаведаць на будзь-яя запытанне, касаючыся салідарнасці, з гэтай бібліятэкі за менш чым тры секунды.
Гэта явна задача генеравання з дапамогай падзеяў. Што лёгка працісціць, так гэта тое, што якасць адказаў вялікай меры вяліцца ўжо задаўна, ў працэсе ETL, які падготавлівае базу знанняў. Некорэктная разбівка на часткі паслабляе кантэкст, простыя методы включэння дадзеных спакойваюць значэнне, а паслабленая індэксацыя спамошчае швайнае падзею. Па джэрантыях, адпаведна 80% проблем з якасцю RAG на самай справе ёсць проблемамі ETL. Зазвычай не мадэль являецца вузкім месцам, а сам працэс.
У гэтым прыкладзе выкарыстоўваюцца Python, LangChain, FAISS, Pinecone і методы включэння дадзеных ад OpenAI.
Масштаб задачы
- Апошнё 40 000 дакументаў у разных фарматах: PDF, DOCX, HTML і простым тэкстам.
- Прыблізна 120 частак кожны дакумент пасля рэкурсыўнага раздзелення на фрагменты по 512 токэнаў.
- Вектарная індэксацыя з 4,8 мільйонаў включэнняў у Pinecone, з 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 RAM) з асінхроннымі працоўнікамі для вбудоўвання.
Выяўленне 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-файлы патрапляюць у чэргу з нераспакаванымі паштовымі справамі, таму што ім трэба адмахоўкаванне тэксту.
Адмахоўкаванне тэксту — гэта самае аднае працэса. Сканаваныя файлы, якім трэба 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 означае павальную перабудову, таму з самага пачатку трэба вырашыць на адважанні па велічыне прагнозаванага корпусу дакументаў.