利用LangChain和FAISS构建用于RAG的生产级ETL管道。
为40,000份合规PDF设计分块、嵌入及向量索引方案，包含重试机制、死信队列、Prometheus指标以及性能基准测试。
来源: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
想象一下，你是一名中型金融科技公司的生成式AI工程师。合规团队拥有超过40,000份PDF文件，内容包括监管申报材料、中央银行通知、证券监管机构指南以及内部政策，这些文件分散存储在SharePoint、S3和本地驱动器中。分析师们每天要花费四到六小时时间手动搜索这些文件。首席技术官的要求是打造一个聊天机器人，能够在三秒内回答来自这些文档库的任何合规相关问题。
这显然属于检索增强生成任务。容易被忽视的是，答案的质量早在问题出现之前，即在用于构建知识库的ETL流程中就已经决定了。糟糕的分块方式会破坏上下文连贯性，简单的嵌入方法会导致语义偏差，而低效的索引机制则会使检索速度变慢。根据经验，大约80%的RAG质量问题实际上都是ETL流程的问题。模型通常并非瓶颈，真正的瓶颈在于整个处理流程。
此处使用的技术栈包括Python、LangChain、FAISS、Pinecone以及OpenAI的嵌入模型。
规模情况
- 共有约40,000份不同格式的文档：PDF、DOCX、HTML和纯文本。
- 每份文档经过512个标记的递归分割后，大约会生成120个分块。
- 在Pinecone中构建的向量索引包含480万个嵌入向量，维度为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份文档。
协调器是一种由数据类驱动的阶段执行器：
# 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)
]
递归字符分割（默认方式）
这是最常用的方法。它会首先尝试最有意义的分隔符（空行、换行符、句号、空格，再到单个字符），只有在必要时才会采用更激进的分割方式。LangChain的RecursiveCharacterTextSplitter实现了这一功能。下面的自适应版本会使用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在二级套餐下允许每分钟处理一百万个标记。若以平均每个数据块300个标记、共计480万个数据块计算，总处理量将达到14.4亿个标记。按照该限制进行顺序嵌入处理需要大约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%。
索引构建
索引构建就是将嵌入的片段存储到可搜索的系统中。其中有两个关键决策：元数据架构（用于过滤检索）以及索引类型（扁平型、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适用于向量数量少于100万的语料库，能够实现100%的召回率，运行速度中等，但内存占用较高。IndexHNSWFlat适用于100万到1000万个向量的语料库，召回率可达约98.5%@10，查询速度快，内存占用处于中等水平。IndexIVFPQ适用于向量数量超过1000万的语料库，召回率为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提取
不使用OCR功能的PyMuPDF处理每份文档需要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秒，不同之处在于优化方案会使用
ThreadPoolExecutor中的8个并行工作线程，而非顺序循环处理。该任务的瓶颈在于磁盘和网络等待时间，而非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意味着需要重新构建整个索引，因此应根据预期的数据量尽早做出决定。