面向银行的RAG数据采集：PDF、OCR、表格及连接器
如何为RAG导入银行文档：对PDF进行分类，使用带审核标记的OCR工具处理，保留Word文档结构，妥善处理表格与图片，并构建可靠的连接器。
来源: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
一家银行的RAG系统悄然失灵
想象一下，某家中型银行在其内部知识库之上构建了一个检索增强生成系统。该系统的覆盖范围包括数千份文档：印度储备银行的监管通告、政策手册、《巴塞尔协议III》的相关指导文件、贷款处理流程、客户身份核实记录以及操作指南。这个项目起初进展顺利——一名工程师负责搭建向量数据库，另一名工程师编写检索流程，还有一名工程师连接大型语言模型并构建用户界面，随后系统便投入使用了。
三周后，用户开始抱怨答案有误，而原因并非幻觉。检索系统找到了正确的文档，模型也依据所给信息准确作答。问题其实早在更早的时候就出现了。一份关于非银行金融机构流动性标准的RBI通告仅以扫描版PDF形式存在，在OCR处理过程中，“10%”这个数字被错识别为“lO%”，其中的数字1被换成了小写的L。解析器未加检查就接受了这一错误数据。嵌入模型对有误的文本进行了编码，向量存储系统将其保存下来，检索系统又将其排在靠前的位置，最终大语言模型自信地告诉合规官员所需的比例是“lO percent.”
检索没有失败，生成也没有失败。每个组件都发挥了其设计功能。问题出在流程的第一阶段——数据导入，而这正是生产环境中的RAG系统中最常见且代价最高的隐患之一。团队们花费数周时间调整嵌入模型、分块方式、重排算法以及提示词，却将数据导入视为普通的预处理步骤。实际上，在任何查询到来之前，检索质量就已经决定了。如果源数据被错误解析，再好的嵌入模型、索引或大语言模型也无法还原原始事实。
接下来的内容将介绍企业级RAG系统中的数据导入处理：如何处理结构化、半结构化和非结构化数据，如何应对多模态文档，如何实现大规模扩展，如何识别解析错误，以及如何运用Python将这些功能整合起来。
数据导入的作用
在RAG流程中，数据摄取环节会接收各种格式和来源的原始文档，并将其转换为结构化、整洁的文本，以便后续进行分块处理和嵌入使用。
处理流程完全在离线管道中运行，早在任何用户提问之前就已开始。该管道输出经过解析和清洗的文本以及元数据，这些内容将成为后续所有步骤的输入。这里会进行三项处理：加载功能从存储位置获取原始文档，这些位置可能是文件系统、S3存储桶、SharePoint站点、REST API或数据库。解析功能则从原始格式中提取纯净文本，方式包括解码PDF、去除HTML标签、反序列化JSON或对图像进行OCR识别。初始元数据提取则会记录文档来源、创建日期、文档类型以及解析时能看到的所有与格式相关的细节。之后再进行文本清洗、分块和嵌入处理。整个流程的唯一目标就是从原始数据中提取出纯净的文本。
银行文档的复杂现实
银行的知识库并非以整洁的 Markdown 格式呈现，而是由数十年来积累的各种格式组成：20世纪90年代的扫描PDF文件、带有修订痕迹的Word文档、内部合规门户的HTML页面、政策管理系统输出的JSON数据、带有附件的电子邮件，以及结构复杂的SharePoint文件夹。每种格式都需要对应的解析器，而每个解析器又会有各自的故障表现。更糟糕的是，解析失败往往毫无征兆——无法正确解析的解析器不会抛出异常，只会返回乱码并继续处理下一项内容。
PDF文件：三种类型，三种处理方式
PDF在监管和合规文档中占据主导地位，同时也是最难解析的格式。针对三种不同的PDF类型，需要采用不同的处理方法：
- 文本层PDF来自文字处理软件或打印转PDF功能，其文本内容位于内容流中，可直接提取。
- 扫描版PDF是由实体纸张转换而成的图像文件，其中完全没有文本层，因此必须使用OCR技术。较旧的通告和历史政策大多属于此类。
- 混合版PDF则是经过扫描后添加了质量较差的OCR文本层的文件。这类文件风险最高，因为简单的解析工具会直接使用存在错误的文本层，并误以为一切正常。
首先需要区分这两类文件，下方的分类功能会检测PDF并告知其所属类型。
import fitz # PyMuPDF
from pathlib import Path
def classify_pdf(pdf_path: str) -> str:
"""
Classify a PDF as text-native, scanned, or hybrid.
Returns: 'text', 'scanned', or 'hybrid'
"""
doc = fitz.open(pdf_path)
total_pages = len(doc)
pages_with_text = 0
total_text_length = 0
for page in doc:
text = page.get_text("text").strip()
if text:
pages_with_text += 1
total_text_length += len(text)
doc.close()
# No text layer at all: scanned PDF
if pages_with_text == 0:
return "scanned"
# Text exists but average is suspiciously short: likely hybrid with poor OCR
avg_text_per_page = total_text_length / total_pages
if avg_text_per_page < 100 and pages_with_text < total_pages * 0.5:
return "hybrid"
return "text"
def load_text_pdf(pdf_path: str) -> list[dict]:
"""
Extract text from a text-native PDF using PyMuPDF.
Returns a list of page dicts with text and metadata.
"""
doc = fitz.open(pdf_path)
pages = []
for page_num, page in enumerate(doc, start=1):
text = page.get_text("text")
pages.append({
"page_number": page_num,
"text": text.strip(),
"source": pdf_path,
"total_pages": len(doc),
})
doc.close()
return pages
扫描后的文件需要OCR处理流程，该图表对比了三种实际可行的处理方案。
对于需要遵循RBI相关规定的银行而言，Azure Document Intelligence是更实用的选择。它能够处理质量参差的扫描文件，理解扫描页面中的表格结构，并为每个单词评估置信度，从而将置信度较低的提取结果标记出来以便人工审核。下方的加载器会调用该服务并应用相应的逻辑。
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
def load_scanned_pdf_azure(
pdf_path: str,
endpoint: str,
api_key: str,
) -> list[dict]:
"""
Extract text from a scanned PDF using Azure Document Intelligence.
Returns pages with text, confidence scores, and table data.
"""
client = DocumentAnalysisClient(
endpoint=endpoint,
credential=AzureKeyCredential(api_key),
)
with open(pdf_path, "rb") as f:
poller = client.begin_analyze_document("prebuilt-read", f)
result = poller.result()
pages = []
for page in result.pages:
page_text = ""
low_confidence_words = []
for line in page.lines:
page_text += line.content + "\n"
# Flag words with confidence below threshold
for word in page.words:
if word.confidence < 0.85:
low_confidence_words.append({
"word": word.content,
"confidence": word.confidence,
})
pages.append({
"page_number": page.page_number,
"text": page_text.strip(),
"low_confidence_words": low_confidence_words,
"needs_review": len(low_confidence_words) > 5,
"source": pdf_path,
})
return pages
在受监管的环境中，needs_review 标志非常重要。如果一个页面包含超过五个置信度较低的单词，就不应被悄悄传递下去，而应放入人工审核队列中，或者至少在其元数据中添加相应标志，以便后续评估时能予以适当考虑。
Word文档、HTML及结构化数据源
Word文件常用于撰写内部政策与流程文档。它们包含标题、表格和列表，这些元素有助于后续检索，但在简单的文本提取工具下会丢失。此处的加载器会遍历文档并保留其结构。
from docx import Document as DocxDocument
from docx.oxml.ns import qn
def load_docx(docx_path: str) -> list[dict]:
"""
Load a DOCX file preserving heading structure and tables.
Returns a list of content blocks with type and text.
"""
doc = DocxDocument(docx_path)
blocks = []
current_section = "Introduction"
for element in doc.element.body:
# Paragraphs
if element.tag == qn("w:p"):
para = element
style = para.style.name if hasattr(para, "style") else ""
text = para.text.strip()
if not text:
continue
if style.startswith("Heading"):
level = int(style.replace("Heading ", "")) if style != "Heading" else 1
current_section = text
blocks.append({
"type": "heading",
"level": level,
"text": text,
"section": current_section,
"source": docx_path,
})
else:
blocks.append({
"type": "paragraph",
"text": text,
"section": current_section,
"source": docx_path,
})
# Tables
elif element.tag == qn("w:tbl"):
table_data = []
for row in element.findall(f".//{qn('w:tr')}"):
row_data = []
for cell in row.findall(f".//{qn('w:tc')}"):
cell_text = " ".join(
p.text for p in cell.findall(f".//{qn('w:t')}")
if p.text
)
row_data.append(cell_text.strip())
table_data.append(row_data)
blocks.append({
"type": "table",
"data": table_data,
"section": current_section,
"source": docx_path,
})
return blocks
章节标记能带来显著效果。在包含四十个段落的KYC流程中，知道某部分内容来自“第3节：强化尽职调查”而非“第1节：标准KYC”，这样的元数据能大幅提升检索精度。
合规门户几乎都是网络应用，因此HTML加载时必须跳过导航、广告、Cookie提示栏以及实际内容周围的其他冗余元素。
import httpx
from bs4 import BeautifulSoup
def load_html_page(url: str, headers: dict | None = None) -> dict:
"""
Load an HTML page and extract main content, stripping boilerplate.
"""
response = httpx.get(url, headers=headers or {}, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
# Remove boilerplate elements common in compliance portals
for tag in soup(["nav", "header", "footer", "script", "style",
"aside", "advertisement", ".cookie-banner",
".sidebar", ".breadcrumb"]):
tag.decompose()
# Find main content area (adjust selectors for your portal)
main_content = (
soup.find("main")
or soup.find("article")
or soup.find(id="content")
or soup.find(class_="content")
or soup.find("body")
)
text = main_content.get_text(separator="\n", strip=True) if main_content else ""
return {
"url": url,
"title": soup.title.string if soup.title else "",
"text": text,
"source": url,
}
诸如政策管理系统、案例数据库和事务日志之类的结构化数据源需要不同的处理方式。这类数据的文本本身已经较为规整，因此关键问题在于确定应提取哪些字段，以及如何将数据行转换为可检索的文本形式。
import json
import psycopg2
from typing import Any
def load_json_policies(json_path: str, text_fields: list[str]) -> list[dict]:
"""
Load a JSON array of policy documents, extracting specified text fields.
Banking example: a JSON export from a policy management system.
"""
with open(json_path, "r", encoding="utf-8") as f:
policies = json.load(f)
documents = []
for policy in policies:
# Combine relevant text fields into a single document
text_parts = []
for field in text_fields:
if field in policy and policy[field]:
text_parts.append(f"{field.replace('_', ' ').title()}: {policy[field]}")
documents.append({
"text": "\n\n".join(text_parts),
"policy_id": policy.get("id"),
"policy_type": policy.get("type"),
"effective_date": policy.get("effective_date"),
"jurisdiction": policy.get("jurisdiction"),
"source": json_path,
})
return documents
def load_from_postgres(
connection_string: str,
query: str,
text_column: str,
metadata_columns: list[str],
) -> list[dict]:
"""
Load documents from a PostgreSQL database.
Banking example: regulatory requirement records from a compliance database.
"""
conn = psycopg2.connect(connection_string)
cursor = conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
cursor.close()
conn.close()
documents = []
for row in rows:
row_dict = dict(zip(columns, row))
documents.append({
"text": str(row_dict.get(text_column, "")),
**{col: row_dict.get(col) for col in metadata_columns},
"source": "postgresql",
})
return documents
银行将大多数工作文件存储在SharePoint或Google Workspace中，要访问这些文件需要OAuth以及相应的Graph或Drive API。
import httpx
from typing import Generator
def load_sharepoint_library(
tenant_id: str,
client_id: str,
client_secret: str,
site_id: str,
drive_id: str,
) -> Generator[dict, None, None]:
"""
Iterate over all DOCX and PDF files in a SharePoint document library.
Yields file metadata dicts; caller is responsible for downloading and parsing.
"""
# Get access token
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_response = httpx.post(token_url, data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "https://graph.microsoft.com/.default",
})
access_token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
# List files recursively
next_link = f"{base_url}/root/children"
while next_link:
response = httpx.get(next_link, headers=headers)
data = response.json()
for item in data.get("value", []):
if item.get("file") and item["name"].endswith((".pdf", ".docx")):
yield {
"name": item["name"],
"download_url": item.get("@microsoft.graph.downloadUrl"),
"created": item.get("createdDateTime"),
"modified": item.get("lastModifiedDateTime"),
"size": item.get("size"),
"web_url": item.get("webUrl"),
}
next_link = data.get("@odata.nextLink")
批量处理还是流式处理
每一种数据导入设计都始于一个决策：是在文档到达时立即处理，还是按预定时间批量处理？
- 批量导入按照预定时间表执行，通常是夜间或每周一次。当数据的时效性不是关键因素、数据量很大且成本需要控制时，这种方式很合适。例如，银行每月导入印度储备银行的通告时并不需要实时处理，而夜间执行任务成本更低。
- 流式导入则是在文档到达时通过事件队列进行处理。当数据的时效性至关重要时，比如需要在一个新通告发布后几分钟内就将其显示出来的合规系统，这种方式更为适用。
外部连接器需要具备防止速率限制的能力，下面的类通过重试和延迟机制来处理请求。
import time
import httpx
from pathlib import Path
class RateLimitedConnector:
"""
A base connector with retry logic and rate limiting.
Use this pattern for any external API that imposes rate limits.
Banking example: connecting to an RBI document portal or a Bloomberg data feed.
"""
def __init__(
self,
base_url: str,
api_key: str,
requests_per_minute: int = 60,
max_retries: int = 3,
backoff_factor: float = 2.0,
):
self.base_url = base_url
self.api_key = api_key
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0.0
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def _wait_for_rate_limit(self):
elapsed = time.monotonic() - self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.monotonic()
def fetch(self, endpoint: str, params: dict | None = None) -> dict:
url = f"{self.base_url}/{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.max_retries):
self._wait_for_rate_limit()
try:
response = httpx.get(url, headers=headers, params=params or {})
if response.status_code == 429:
# Rate limited: back off and retry
wait_time = self.backoff_factor ** attempt
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
if attempt == self.max_retries - 1:
raise
time.sleep(self.backoff_factor ** attempt)
raise RuntimeError(f"Failed to fetch {url} after {self.max_retries} attempts")
那种模式并非可选。包括金融数据提供商在内的大多数外部 API，一旦超出限制就会封锁该 IP 或撤销密钥。backoff_factor参数实现了指数级退避机制，即连续重试时的等待时间分别为2秒、4秒和8秒，这是生产环境连接器最基本的重试策略。
表格、图表与图片
2025年，生产环境中的RAG技术已从仅处理文本的流程转向多模态处理。银行文件并非纯文本形式，其中包含资本充足率表格、不良贷款趋势图表、客户身份核实流程图以及已签署协议的图片。忽略非文本内容会丢弃每份文件中大量信息，而在监管要求下，这些信息往往是最有价值的。印度储备银行通告中所列的精确数值阈值，正是合规团队所需要的内容。
表格有两种处理方式：以文本形式序列化，或提取为结构化数据。具体选择取决于后续用途。下面的提取函数可用于从PDF中获取表格。
import fitz
import pdfplumber
def extract_tables_from_pdf(pdf_path: str) -> list[dict]:
"""
Extract tables from a PDF using pdfplumber.
Returns tables as both structured data and serialised text.
Banking example: extracting capital ratio tables from Basel III compliance reports.
"""
tables_output = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
tables = page.extract_tables()
for table_idx, table in enumerate(tables):
if not table or len(table) < 2:
continue
# First row as headers
headers = [str(h).strip() if h else f"col_{i}"
for i, h in enumerate(table[0])]
rows = table[1:]
# Structured data representation
structured = []
for row in rows:
row_dict = {}
for header, cell in zip(headers, row):
row_dict[header] = str(cell).strip() if cell else ""
structured.append(row_dict)
# Text serialisation (markdown table format for LLM consumption)
header_row = " | ".join(headers)
separator = " | ".join(["---"] * len(headers))
data_rows = [
" | ".join(str(cell or "").strip() for cell in row)
for row in rows
]
serialised = "\n".join([header_row, separator] + data_rows)
tables_output.append({
"page_number": page_num,
"table_index": table_idx,
"headers": headers,
"structured_data": structured,
"serialised_text": serialised,
"source": pdf_path,
})
return tables_output
序列化后的 Markdown 适合让大型语言模型阅读并进行分析，比如“巴塞尔协议 III 要求的 CET1 比率是多少？”。结构化的 JSON 则适合在代码中查询或进行数值比较的场景。
图表、示意图和信息图同样不可或缺。按季度划分的不良贷款趋势图，或是 FEMA 交易审批流程的流程图，都包含着周围文字中找不到的信息。常见的 2025 年处理方式是将每张图片输入视觉语言模型，然后存储生成的描述以便后续嵌入和检索。
import fitz
import base64
import httpx
from io import BytesIO
def extract_and_describe_images(
pdf_path: str,
openai_api_key: str,
min_image_size: int = 5000, # bytes, to skip tiny decorative images
) -> list[dict]:
"""
Extract images from a PDF and generate text descriptions using GPT-4V.
Banking example: describing NPA trend charts and regulatory workflow diagrams.
"""
doc = fitz.open(pdf_path)
image_docs = []
for page_num in range(len(doc)):
page = doc[page_num]
image_list = page.get_images(full=True)
for img_idx, img_info in enumerate(image_list):
xref = img_info[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
# Skip tiny decorative images
if len(image_bytes) < min_image_size:
continue
# Encode to base64 for the vision API
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
image_ext = base_image["ext"]
# Generate description with GPT-4V
response = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {openai_api_key}"},
json={
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"This image is from a banking regulatory document. "
"Describe its content precisely, including any numerical "
"values, trends, labels, axes, or process steps shown. "
"If it is a chart, describe what metric is shown, the "
"time period, and the key data points. If it is a "
"diagram, describe the process or relationship shown."
),
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/{image_ext};base64,{image_b64}"
},
},
],
}
],
"max_tokens": 400,
},
)
description = response.json()["choices"][0]["message"]["content"]
image_docs.append({
"type": "image_description",
"page_number": page_num + 1,
"image_index": img_idx,
"description": description,
"source": pdf_path,
})
doc.close()
return image_docs
提示词在这里至关重要。泛泛的“描述这张图片”只会得到千篇一律的结果；而针对特定领域的指令，要求提供数值、趋势、标签或流程步骤，才能生成有实际参考价值的描述。例如，工程师若想了解第三季度报告中的NPA趋势线，就需要包含“NPA”“趋势”以及相应季度的描述，而非关于蓝色柱状图的描述。
音频与视频
音频和视频是当前银行领域RAG技术的前沿应用方向。董事会会议记录、分析师电话会议文字稿、印度储备银行新闻发布会内容以及培训视频中都包含着RAG系统目前还无法获取的宝贵信息。
import httpx
import time
def transcribe_audio(
audio_path: str,
openai_api_key: str,
language: str = "en",
) -> dict:
"""
Transcribe an audio file using OpenAI Whisper.
Banking example: transcribing an RBI monetary policy press conference recording.
"""
with open(audio_path, "rb") as audio_file:
response = httpx.post(
"https://api.openai.com/v1/audio/transcriptions",
headers={"Authorization": f"Bearer {openai_api_key}"},
data={
"model": "whisper-1",
"language": language,
"response_format": "verbose_json", # includes timestamps per segment
"timestamp_granularities[]": "segment",
},
files={"file": (audio_path, audio_file, "audio/mpeg")},
)
result = response.json()
return {
"transcript": result.get("text", ""),
"segments": [
{
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
}
for seg in result.get("segments", [])
],
"language": result.get("language", language),
"source": audio_path,
}
输出中的时间戳非常重要。当问题的答案出现在电话会议进行到34分钟时的某段对话中时，系统应标明确切的位置，而不仅仅是引用会议录音。
每种格式对应一个路由器
单独的加载器是不够的。生产流程必须将每份文档发送到正确的解析器，妥善处理故障，跟踪处理状态，并无论输入内容如何都能生成一致的输出格式。
from pathlib import Path
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class DocumentType(Enum):
PDF_TEXT = "pdf_text"
PDF_SCANNED = "pdf_scanned"
DOCX = "docx"
HTML = "html"
JSON = "json"
AUDIO = "audio"
UNKNOWN = "unknown"
def detect_document_type(file_path: str) -> DocumentType:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
classification = classify_pdf(file_path)
return DocumentType.PDF_SCANNED if classification == "scanned" else DocumentType.PDF_TEXT
elif suffix == ".docx":
return DocumentType.DOCX
elif suffix in (".html", ".htm"):
return DocumentType.HTML
elif suffix == ".json":
return DocumentType.JSON
elif suffix in (".mp3", ".mp4", ".wav", ".m4a"):
return DocumentType.AUDIO
return DocumentType.UNKNOWN
def ingest_document(
file_path: str,
azure_endpoint: str | None = None,
azure_api_key: str | None = None,
openai_api_key: str | None = None,
) -> list[dict]:
"""
Route a document to the correct parser and return a list of content blocks
with a consistent schema regardless of input format.
Every output block contains:
- text: the extracted text content
- source: origin path or URL
- doc_type: the detected document type
- page_number: where applicable
- needs_review: flag for human review queue
"""
doc_type = detect_document_type(file_path)
results = []
try:
if doc_type == DocumentType.PDF_TEXT:
pages = load_text_pdf(file_path)
for page in pages:
page["doc_type"] = "pdf_text"
page["needs_review"] = False
results.append(page)
elif doc_type == DocumentType.PDF_SCANNED:
if azure_endpoint and azure_api_key:
pages = load_scanned_pdf_azure(file_path, azure_endpoint, azure_api_key)
for page in pages:
page["doc_type"] = "pdf_scanned"
results.append(page)
else:
logger.warning(
f"Scanned PDF detected but no OCR credentials provided: {file_path}"
)
results.append({
"text": "",
"source": file_path,
"doc_type": "pdf_scanned",
"needs_review": True,
"error": "No OCR credentials available",
})
elif doc_type == DocumentType.DOCX:
blocks = load_docx(file_path)
for block in blocks:
block["doc_type"] = "docx"
block["needs_review"] = False
results.append(block)
elif doc_type == DocumentType.AUDIO:
if openai_api_key:
transcript = transcribe_audio(file_path, openai_api_key)
for segment in transcript["segments"]:
results.append({
"text": segment["text"],
"source": file_path,
"doc_type": "audio_transcript",
"start_seconds": segment["start"],
"end_seconds": segment["end"],
"needs_review": False,
})
else:
logger.warning(f"Unknown document type, skipping: {file_path}")
except Exception as e:
logger.error(f"Ingestion failed for {file_path}: {e}")
results.append({
"text": "",
"source": file_path,
"doc_type": str(doc_type.value),
"needs_review": True,
"error": str(e),
})
return results
统一的输出结构是最重要的设计决策。无论输入的是扫描后的圆形文档、DOCX格式的贷款政策文件，还是合规培训课程的音频转录内容，分块器、嵌入生成器和元数据增强器都会接收相同格式的词典。needs_review标志会将存在问题的文档放入人工处理队列，从而防止错误进一步扩散。
大多数问题的根源在于三种故障
银行领域RAG系统在检索质量方面出现的问题，大多可追溯到三种数据导入错误。
解析时忽略文档类型。 对扫描后的 PDF 进行文本提取时，往往不会得到任何结果，或仅有少量杂乱字符。这样的文件会变成空片段、接近零的向量值以及检索系统返回的空索引条目。大语言模型看到空上下文后就会编造答案。classify_pdf 函数仅需几行代码就能避免这一问题，省去数周的调试时间。
将表格视为正文文本处理。 被转换成普通文字的风险评级表会失去行和列结构，像“AAA 0.01% AA 0.05% A 0.10%”这样的字符串也无法被有效检索。但如果以带表头的 Markdown 表格形式序列化，就能实现有效检索。应单独提取表格并保留其结构。
连接器缺少重试机制。当 SharePoint 连接器遇到网络超时却没有任何反馈时，数据集中就会出现缺失，而且没人知道哪些文档不见了。RateLimitedConnector模式会在重试次数用尽时触发，并记录每次尝试的情况，这样缺失部分就能被识别出来。
工具的权衡
- PyMuPDF、pdfplumber 和 PyPDF2。PyMuPDF 是处理纯文本 PDF 的最快提取工具。pdfplumber 速度稍慢，但在处理表格方面表现更好。PyPDF2 是最老的版本，准确性也最低，因此在新流程中应避免使用。对于同一文件，可用 PyMuPDF 提取文本，用 pdfplumber 提取表格。
原始文本并非干净文本
数据导入后得到的是原始文本，而非干净文本。在数据导入与分块处理之间还有一个独立的清洗阶段，负责去除噪声、去重、修正OCR错误、实现多语言标准化以及删除冗余内容。分阶段的处理很重要，因为不同环节会出现不同的问题：数据导入失败会导致内容为空或混乱，而清洗失败则会使内容表面上看似正常，但实际上存在会影响嵌入质量的细微噪声，这类问题只有在执行检索测试时才会显现出来。
在正式生产环境中，代码示例需要具备日志记录功能、密钥管理机制（应将API密钥存储在环境变量或专用密钥管理系统中，绝不能硬编码），并在每个处理阶段添加延迟监测功能。