生产型RAG的数据清洗：编码、OCR识别、去重及个人信息处理
如何将解析后的企业文档转换为可用于检索的文本：编码修复、模板与重复内容删除、高风险OCR标记处理、语言处理以及个人身份信息遮蔽。
来源: Data Cleaning and Preprocessing for Production RAG: Building Retrieval-Ready Documents
想象这样一家银行：它在20万份内部文件和监管文件之上部署了知识助手。数据导入过程顺利，PDF文件能被解析，扫描后的反洗钱通告可通过OCR识别，SharePoint文件也会按时到达，而且表格会被单独提取出来。看来这个文档库似乎没有遗漏任何内容。
然而查询结果却参差不齐。一名合规官员询问针对高风险企业客户进行强化尽职调查需要哪些文件，检索系统返回了五段内容。其中三段包含相关的客户身份识别政策，另外两段则是来自该文件旧版本的几乎完全相同的文本。每段内容开头都有相同的标题：
CONFIDENTIAL - INTERNAL USE ONLY
Group Financial Crime Compliance
Page 47 of 132
其中一段文本还包含“所有高风险企业客户都必须接受更严格的尽职调查”这句话。仔细看：All变成了AII，diligence变成了diIigence，这是因为OCR将小写的l和大写的I搞混了。
在同一文件的其他地方，有一个不间断空格被保留了下来，Windows-1252编码的撇号也被错误地解码了，而页脚内容因为出现在每一页上而出现了132次。该政策的归档副本被保存在索引中，与已批准的版本放在一起，因为SharePoint保留了这两个版本。
从技术层面来看，数据导入流程是正常的。但从实际使用角度看，这些数据集已经受到污染，那些正在构建首个RAG系统的工程师往往低估了这种差异所带来的影响。
为何脏数据会隐藏起来
这类损坏通常很微妙，不会导致任何程序崩溃。格式错误的字节序列不会让向量数据库报错，重复的段落也不会使嵌入 API 失效。重复的页脚不会触发警报，而 OCR 识别中将 1 误识为 l 也不会中断索引任务。整个处理流程会愉快地嵌入所有数据，随后才会以检索失败的形式体现问题。
重复的文档会占据 top-k 结果中的多个位置。反复出现的模板内容会在无关的内容片段之间制造出人为的相似性。OCR 错误会改变关键实体、百分比、账号编号以及监管限制数值。编码错误则会引入没有实际业务意义的标记。当错误的答案出现时，各个下游组件可能都在正常工作：检索器返回了存储的内容，重排序器对收到的内容进行了排序，大语言模型则根据所获得的上下文进行处理。问题其实早在这三个环节之前就已经出现了。
采用这样一项反洗钱政策：金额超过10,000欧元的交易需要进行额外的资金来源核查。一旦出现错误的OCR识别和编码转换，被索引的内容就会显示为“EUR l0,000”。人们能立刻发现这个错误。而嵌入模型仍会将该句子视为与交易阈值和资金来源核查等概念相近的内容，因此检索功能依然运作良好，这恰恰令人担忧。由于语义检索能够容忍文本上的瑕疵，这条被损坏的陈述仍会传达到大型语言模型那里。系统最终找到的却是包含错误监管数值的正确政策条款。
正因如此，文本清洗工作绝不能仅靠text.strip().lower()来完成。
文本清洗在RAG流程中的位置
清洗工作位于文档导入与分块处理之间，这一顺序是有意安排的。在修复明显的编码错误之前，不应先测量文档的相似度；在识别出文本语言之前，也不应过度进行文本规范化处理；不能仅仅因为某些法律条款与模板相似就直接删除它们；此外，置信度低的OCR校正结果也不应擅自修改合规要求。
一旦含有杂质的文本被分割成块并嵌入系统后，进行清洗就会变得极为困难。重复的段落已经生成了重复的向量，破碎的单词影响了分块边界，重复的标题占用了标记空间，而OCR错误则已嵌入到语义表示中。因此，清洗工作应在分块处理之前完成，而非在质量开始下降后才作为补救措施来实施。
编码规范化：先修复文本
企业中的各种文档集合很少使用相同的编码规则。现代系统可能采用UTF-8编码，而客户通信档案可能使用Windows-1252编码，更早的导出文件则可能采用Latin-1编码。在进入处理流程之前，从SharePoint、电子邮件、Word以及旧版文档系统中复制的文本可能会经过多次编码和解码。最终的结果如下：
Customer’s identity must be verified before account opening.
The customer must provide proof of address.
Enhanced Due Diligence â€" High Risk Customers
这种损坏现象被称为“mojibake”，当用一种编码写入的字节被误读为另一种编码时就会发生。人们立刻就能看出customer’s存在问题。在数据检索流程中，损坏后的标记会以与customer's不同的形式传递给分词器，而大规模情况下这类不一致性会破坏词汇表结构，并为嵌入空间带来不必要的变化。
需谨慎处理修复的程度。有些鲁莽的函数会将任何非ASCII字符视为无效数据，但实际上欧洲的支付相关文本集中就存在Société Générale、Deutsche Bundesbank、€10,000和Überweisung这类合法内容。如果去掉重音符号和特殊字符，真实信息就会消失。目标并非仅使用ASCII文本，而是保持语义一致的Unicode格式。
利用ftfy修复Unicode
您无需亲自列出所有的乱码模式，因为 ftfy 已经能够修复常见的 Unicode 和编码错误。可以从这里开始进入正式的清理阶段：
import re
import unicodedata
from dataclasses import dataclass, field
import ftfy
@dataclass
class TextCleaningResult:
original_text: str
cleaned_text: str
transformations: list[str] = field(default_factory=list)
quality_flags: list[str] = field(default_factory=list)
def normalize_encoding(text: str) -> TextCleaningResult:
"""
Repair common encoding problems while preserving meaningful Unicode.
Suitable for policy, AML/KYC, payment, and regulatory documents where
accented names, currency symbols, and multilingual text must survive.
"""
transformations: list[str] = []
quality_flags: list[str] = []
if not text:
return TextCleaningResult(original_text=text, cleaned_text=text)
cleaned = text
# Repair mojibake and common Unicode encoding mistakes.
repaired = ftfy.fix_text(cleaned)
if repaired != cleaned:
transformations.append("ftfy_encoding_repair")
cleaned = repaired
# NFC preserves characters while producing a consistent Unicode
# representation. Safer than aggressively stripping accents.
normalized = unicodedata.normalize("NFC", cleaned)
if normalized != cleaned:
transformations.append("unicode_nfc_normalization")
cleaned = normalized
# Replace non-breaking spaces with normal spaces.
if "\u00a0" in cleaned:
cleaned = cleaned.replace("\u00a0", " ")
transformations.append("non_breaking_space_normalization")
# Remove zero-width characters that frequently leak from PDFs,
# web pages, and copied Office content.
zero_width_chars = {"\u200b", "\u200c", "\u200d", "\ufeff"}
if any(char in cleaned for char in zero_width_chars):
cleaned = "".join(char for char in cleaned if char not in zero_width_chars)
transformations.append("zero_width_character_removal")
# Remove control characters; preserve newline and tab because
# they may still carry document structure needed downstream.
cleaned_without_controls = "".join(
char for char in cleaned
if char in "\n\t" or unicodedata.category(char) != "Cc"
)
if cleaned_without_controls != cleaned:
transformations.append("control_character_removal")
cleaned = cleaned_without_controls
# Normalize horizontal whitespace without flattening paragraphs.
whitespace_normalized = re.sub(r"[ \t]+", " ", cleaned)
whitespace_normalized = re.sub(r"\n{3,}", "\n\n", whitespace_normalized)
if whitespace_normalized != cleaned:
transformations.append("whitespace_normalization")
cleaned = whitespace_normalized
cleaned = cleaned.strip()
# Keep suspicious replacement characters observable rather than silently deleting them.
if "\ufffd" in cleaned:
quality_flags.append("unicode_replacement_character_detected")
return TextCleaningResult(
original_text=text,
cleaned_text=cleaned,
transformations=transformations,
quality_flags=quality_flags,
)
注意该函数没有对哪些内容置之不理。它并没有将所有文字转为小写，也没有去掉Société Générale中的重音符号、欧元符号，更没有把文档压缩成单行。这些省略是刻意为之的。
为何过度清洗会带来危害
许多旧版自然语言处理流程是为词袋模型或TF-IDF算法设计的，它们会将文字转为小写、去除标点符号、删掉停用词并进行词干提取。在采用现代嵌入模型之前盲目套用这类方法会丢失有用信息。试想一下它会对这句话产生什么影响：
Customer must NOT be classified as low risk.
财经文本提供了更为微妙的例子。EUR 10,000与EUR 100,000在书写形式上相似，但其实际含义却大相径庭；而Article 6(1)(a)与Article 6(1)(b)虽然结构相同，但法律效力却不同。处理此类文本的准则是：积极实现表述标准化，同时谨慎保留原有含义。将customer’s改为customer's只是形式上的调整，旨在还原作者的原意。然而删除€符号、去掉客户姓名中的重音符号，或自动将l0%转换为10%，则可能会改变企业所依赖的事实信息。
去除模板文字：重在结构而非关键词
页眉和页脚与编码噪声并非同一类问题。132页中的第47页这种重复显然是模板内容，但出现在多个政策条款中的合规要求可能是有意重复的，并具有法律效力。仅凭某行出现的频率无法判断属于哪种情况。
更有效的识别依据是将重复性与位置结合考虑。页眉集中在页面顶部，页脚位于底部，而页码虽会略有变化，但仍保持相同的结构模式：
KYC Policy | Version 7.2 | Page 41 of 132
KYC Policy | Version 7.2 | Page 42 of 132
精确匹配逻辑会将两者视为不同的字符串，而结构感知逻辑则会将其视为一个模板的两个实例。这就是为什么在页面边界仍然存在时，模板检测效果最佳的原因之一；因此应在文档被读取后避免立即将其扁平化为一个巨大的字符串。
import re
from collections import Counter
from dataclasses import dataclass
@dataclass
class BoilerplatePattern:
normalized_text: str
occurrences: int
page_ratio: float
position: str
def normalize_boilerplate_candidate(line: str) -> str:
"""
Normalize variable fields so structurally identical headers and
footers can be compared across pages.
"""
normalized = line.strip()
normalized = re.sub(
r"\bpage\s+\d+\s+of\s+\d+\b", "page <n> of <n>",
normalized, flags=re.IGNORECASE,
)
normalized = re.sub(r"\bpage\s+\d+\b", "page <n>", normalized, flags=re.IGNORECASE)
normalized = re.sub(r"\s+", " ", normalized)
return normalized.casefold()
def detect_repeated_marginal_text(
pages: list[dict],
margin_lines: int = 3,
min_page_ratio: float = 0.6,
) -> list[BoilerplatePattern]:
"""
Detect repeated text in page-header or page-footer regions.
A candidate is considered boilerplate only when it appears in the same
marginal position across a substantial fraction of the document.
"""
if not pages:
return []
header_counts: Counter = Counter()
footer_counts: Counter = Counter()
for page in pages:
lines = [line.strip() for line in page["text"].splitlines() if line.strip()]
if not lines:
continue
header_counts.update(normalize_boilerplate_candidate(l) for l in lines[:margin_lines])
footer_counts.update(normalize_boilerplate_candidate(l) for l in lines[-margin_lines:])
total_pages = len(pages)
patterns: list[BoilerplatePattern] = []
for position, counts in (("header", header_counts), ("footer", footer_counts)):
for normalized_text, occurrences in counts.items():
page_ratio = occurrences / total_pages
if page_ratio >= min_page_ratio:
patterns.append(BoilerplatePattern(
normalized_text=normalized_text,
occurrences=occurrences,
page_ratio=page_ratio,
position=position,
))
return patterns
min_page_ratio阈值是最为关键的决策依据。规则并非“删除出现两次的任何行”，而是“如果某行在文档的大部分内容中始终出现在页边位置，就将其视为样板文本”。对于一个共100页的文档，若其中98页都带有CONFIDENTIAL - INTERNAL USE ONLY这样的标题，那么这一判断就有充分依据。
另一个相关要点是：从用于嵌入的文本中移除某些内容并不意味着要从系统中彻底删除它。有时信息应当从语义内容层转移到元数据层，诸如版本、所有者、分类和生效日期等信息作为元数据保存会比复制到每个独立片段中更为合适。
超越完全重复的去重方式
一份文档即便经过完美编码、没有OCR损坏，去除了冗余内容，但由于语料库中存在多个副本，仍然会影响检索效果。企业知识库中充斥着这类情况：
AML_Policy_v8_Final.docx
AML_Policy_v8_Final_Approved.docx
AML_Policy_v8_Final_Copy.docx
AML_Policy_v8_Approved_2026.docx
将这四个文件都作为独立文档进行索引，其数据块会在检索过程中相互竞争。如果要求加强对高风险客户的监控，一个政策条款就可能占满三个位置，只剩下两个位置用于真正不同的证据。重复数据带来的代价不仅仅是磁盘空间占用：它们还会扭曲“top-k”概念的实际含义。
通过内容哈希实现完全重复内容的识别
import hashlib
from dataclasses import dataclass
@dataclass
class DeduplicationRecord:
document_id: str
source: str
content_hash: str
duplicate_of: str | None = None
def canonicalize_for_hashing(text: str) -> str:
"""
Deterministic normalization before hashing.
Does not lowercase or remove punctuation because those transformations
could collapse documents that are not actually identical.
"""
lines = [" ".join(line.split()) for line in text.splitlines()]
return "\n".join(lines).strip()
def calculate_content_hash(text: str) -> str:
canonical_text = canonicalize_for_hashing(text)
return hashlib.sha256(canonical_text.encode("utf-8")).hexdigest()
def find_exact_duplicates(
documents: list[dict],
) -> tuple[list[dict], list[DeduplicationRecord]]:
"""
Keep one canonical copy of each identical document while preserving
duplicate lineage for auditability.
"""
seen_hashes: dict[str, str] = {}
unique_documents: list[dict] = []
records: list[DeduplicationRecord] = []
for document in documents:
content_hash = calculate_content_hash(document["clean_text"])
if content_hash in seen_hashes:
records.append(DeduplicationRecord(
document_id=document["document_id"],
source=document["source"],
content_hash=content_hash,
duplicate_of=seen_hashes[content_hash],
))
continue
seen_hashes[content_hash] = document["document_id"]
unique_documents.append(document)
records.append(DeduplicationRecord(
document_id=document["document_id"],
source=document["source"],
content_hash=content_hash,
))
return unique_documents, records
哈希值在增量索引中也十分有用。如果同一份标准化文档出现在明晚的数据导入过程中，其哈希值早已已知，因此处理流程可以跳过再次进行清洗、分块和嵌入的操作。
使用MinHash LSH检测近似重复内容
精确哈希适用于简单情况。更复杂的情况是，比如某项客户身份验证政策的版本7和版本8，文本内容有98%的相似度，但存在一条监管条款的不同。从合规性角度来看，它们绝不是重复内容，因为那条不同的条款改变了相应的控制要求。
在受监管的RAG系统中，近重复检测应首先回答另一个问题，即哪些文档的相似度足以值得进一步调查，而将删除操作作为单独的决策。对于20万份文档而言，两两比较需要约200亿次比对。MinHash结合局部敏感哈希技术，通过生成紧凑的签名并将相似的文档放入候选桶中，从而避免了这种庞大的比较量。
import re
from datasketch import MinHash, MinHashLSH
def create_word_shingles(text: str, shingle_size: int = 5) -> set[str]:
"""
Five-word shingles work well for long policy and regulatory documents
because they capture local textual structure without being overly
sensitive to isolated formatting changes.
"""
tokens = re.findall(r"\b\w+\b", text.casefold())
if len(tokens) < shingle_size:
return {" ".join(tokens)} if tokens else set()
return {
" ".join(tokens[i:i + shingle_size])
for i in range(len(tokens) - shingle_size + 1)
}
def create_minhash(text: str, num_perm: int = 128) -> MinHash:
shingles = create_word_shingles(text)
minhash = MinHash(num_perm=num_perm)
for shingle in shingles:
minhash.update(shingle.encode("utf-8"))
return minhash
def build_duplicate_index(
documents: list[dict],
threshold: float = 0.85,
num_perm: int = 128,
) -> tuple[MinHashLSH, dict[str, MinHash]]:
"""
Build an LSH index for candidate near-duplicate discovery.
The threshold identifies candidates. It does not automatically
determine whether a document is deleted.
"""
lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
signatures: dict[str, MinHash] = {}
for document in documents:
document_id = document["document_id"]
signature = create_minhash(document["clean_text"], num_perm=num_perm)
signatures[document_id] = signature
lsh.insert(document_id, signature)
return lsh, signatures
当 KYC 政策版本 v7 和 v8 被归入同一类别时，处理流程仅会判定需要进一步仔细检查，而不会决定删除其中某个版本。实际决策应结合文本相似度以及政策编号、版本、生效日期、审批状态和管辖区域等元数据，因为仅凭相似度无法做出此类判断。
OCR 错误：外观正确但内容错误的文本
编码修复功能用于修正存储方式有误的字符，而 OCR 错误则属于另一类问题。在这种情况下，识别引擎可能完全确信地读取了一个完全有效的 Unicode 字符，但却选错了。常见的易混淆字符对包括 0 和 O、1、l 和 I、5 和 S，以及 rn 和 m。
整体错误率可能较低，但对于受监管的RAG系统而言，综合准确率并非关键指标。一份40页、正确率为99.8%的文件，其最关键的数值仍可能被错误修改。关键在于OCR错误的风险程度各不相同：将custorner修正为customer属于低风险，因为上下文和词典能明确提示正确用词；而将EUR l0,000改为EUR 10,000则属于完全不同的风险等级，因为错误的转换可能会影响合规要求。
import re
from dataclasses import dataclass
@dataclass
class OCRIssue:
token: str
issue_type: str
risk_level: str
suggested_value: str | None
requires_review: bool
CURRENCY_PATTERN = re.compile(r"\b(EUR|USD|GBP|INR)\s+([A-Za-z0-9.,]+)\b")
PERCENTAGE_PATTERN = re.compile(r"\b([A-Za-z0-9.,]+)\s*%")
def detect_numeric_ocr_issues(text: str) -> list[OCRIssue]:
"""
Detect suspicious OCR substitutions inside financial values.
This function identifies candidates. It does not silently rewrite
high-risk values in the source text.
"""
issues: list[OCRIssue] = []
substitutions = {"O": "0", "o": "0", "I": "1", "l": "1"}
def inspect_numeric_token(token: str, issue_type: str) -> None:
if token.replace(",", "").replace(".", "").isdigit():
return
corrected = "".join(substitutions.get(char, char) for char in token)
numeric_candidate = corrected.replace(",", "").replace(".", "")
if numeric_candidate.isdigit() and corrected != token:
issues.append(OCRIssue(
token=token,
issue_type=issue_type,
risk_level="high",
suggested_value=corrected,
requires_review=True,
))
for match in CURRENCY_PATTERN.finditer(text):
inspect_numeric_token(match.group(2), issue_type="currency_value")
for match in PERCENTAGE_PATTERN.finditer(text):
inspect_numeric_token(match.group(1), issue_type="percentage")
return issues
该函数刻意不会修改文档内容。正则表达式可以识别出在货币语境中l0,000显得可疑，但无法证明原文实际写的是10,000。
处理普通词汇时，使用了解该领域的词典会更有效。而银行相关文本则需要一个通用词表，再加上beneficial ownership、counterparty、SWIFT、IBAN、SEPA以及sanctioned这类专业术语。否则，普通的拼写检查工具很可能会将合法的金融术语“纠正”成更错误的表达。
在实际应用中，OCR纠错功能会权衡引擎的识别置信度与错误可能造成的影响程度。对于风险较低的文本，即便识别结果不够准确，通常也无需人工干预即可修正。而涉及金钱金额、百分比、法规条款编号及日期等高风险内容，则无论OCR的置信度如何，都应交由人工审核。金融系统应当追求可控的正确性，而非极致的自动化。
多语言预处理
当语料库跨越不同国家时，语言检测就显得尤为重要。一个支付业务数据库中可能包含英文的操作文档、德文的客户信件、法文的法规通知以及西班牙文的合同，而且一份反洗钱调查文件中也可能在同一案件记录里包含多种语言的内容。
语言会影响后续的所有处理环节。拼写校正依赖于语言，OCR的错误模式因文字系统而异，命名实体所带的重音符号在标准化处理后仍需保留，而且嵌入模型本身的表现也会根据所索引的语言不同而有所差异。
from dataclasses import dataclass
from langdetect import DetectorFactory, LangDetectException, detect_langs
DetectorFactory.seed = 0 # Makes pipeline output reproducible.
@dataclass
class LanguagePrediction:
language: str
confidence: float
alternatives: list[tuple[str, float]]
needs_review: bool
def detect_document_language(
text: str,
confidence_threshold: float = 0.85,
) -> LanguagePrediction:
sample = text.strip()
if not sample:
return LanguagePrediction(language="unknown", confidence=0.0, alternatives=[], needs_review=True)
try:
predictions = detect_langs(sample)
except LangDetectException:
return LanguagePrediction(language="unknown", confidence=0.0, alternatives=[], needs_review=True)
best = predictions[0]
alternatives = [(p.lang, round(p.prob, 4)) for p in predictions]
return LanguagePrediction(
language=best.lang,
confidence=best.prob,
alternatives=alternatives,
needs_review=best.prob < confidence_threshold,
)
当文档包含多种语言时，按段落进行检测比对整个文件给出统一结论更为有效。此外，这还能为分块工具提供结构线索：如果分割发生在分块工具未曾识别的语言边界上，一个分块中就可能包含两种无关的语言语境。
有些团队会先将所有内容翻译成英语。虽然这样能让索引更简单，但会在权威文本与检索结果之间增加一次转换环节；而德国的合规条款往往使用具有精确法律含义的措辞，翻译可能会对此产生微妙的变化。更为稳健的设计是将原文视为真实来源，将各种译文作为附加表述存储，并在元数据中记录其来源信息。语言相关的元数据也应做到精确无误：
metadata = {
"primary_language": "en",
"languages": ["en", "de", "fr"],
"language_distribution": {"en": 0.72, "de": 0.21, "fr": 0.07},
"is_multilingual": True,
}
语言检测也可用作健康检查。假设一份预期为英文的英国反洗钱政策文件，突然显示出primary_language = “unknown”，且confidence = 0.31。该文件未必是多语言的；OCR可能严重破坏了文本结构，导致检测工具无法识别。
个人身份信息与敏感数据：在嵌入前进行检测
预处理是控制进入检索系统的内容的最后机会之一。一份客户身份验证文件可能包含客户姓名、护照号码、IBAN账号、电话号码和电子邮件地址。一旦这些数据经过嵌入服务并存入向量数据库，问题就会比保护原始文档复杂得多，因为现在这些数据的副本会存在于嵌入请求、向量值、元数据、缓存、索引、日志以及备份文件中。
问题不在于嵌入模型能否理解个人身份信息，因为它确实可以。关键在于这些信息是否有必要被嵌入。
删除检测器找到的所有实体确实能提升语料库的安全性，但会破坏基于实体的检索功能。如果每个公司名称都被替换为<ORGANIZATION>，那么当反洗钱调查人员查询Northbridge Trading GmbH时将一无所获。因此，实际的处理方式是在采取行动之前为每种检测到的实体类型制定相应的处理策略：公开名称和运营名称通常保留，客户个人身份信息则会被遮蔽、标记为令牌或加以限制，内部组织名称则需遵循访问规则，而凭证绝不能出现在索引中。
from dataclasses import dataclass, field
from presidio_analyzer import AnalyzerEngine
@dataclass
class SensitiveEntity:
entity_type: str
start: int
end: int
score: float
original_value: str
@dataclass
class PIIScanResult:
entities: list[SensitiveEntity] = field(default_factory=list)
needs_review: bool = False
class BankingPIIDetector:
def __init__(self) -> None:
self.analyzer = AnalyzerEngine()
def detect(
self,
text: str,
language: str = "en",
score_threshold: float = 0.70,
) -> PIIScanResult:
"""
Detect sensitive entities without modifying the source text.
Detection is deliberately separated from anonymization because
different entity types require different handling policies.
"""
results = self.analyzer.analyze(
text=text, language=language, score_threshold=score_threshold
)
entities = [
SensitiveEntity(
entity_type=r.entity_type,
start=r.start,
end=r.end,
score=r.score,
original_value=text[r.start:r.end],
)
for r in results
]
high_risk_types = {"CREDIT_CARD", "IBAN_CODE", "US_SSN", "IP_ADDRESS"}
return PIIScanResult(
entities=entities,
needs_review=any(e.entity_type in high_risk_types for e in entities),
)
对于RAG而言，使用带类型的占位符替换值通常比直接删除它们更有效：
Customer IBAN: <IBAN>
Customer Email: <EMAIL_ADDRESS>
该句子仍保留着其表达意义的结构，因此关于客户IBAN验证地点的查询依然可以与包含<IBAN>的文本匹配。当需要保留各提及元素之间的关联时，确定的标记比通用的占位符更合适。在40条案例记录中反复出现的CUSTOMER_7F29A1能够保持调查线索的完整性，而若将每个名称都替换为<PERSON>则会导致线索断裂。
元数据很容易被忽视而泄露。正文虽经精心处理，其字段中仍可能包含customer_name: "Lena Hoffmann"或类似/customers/Lena_Hoffmann/KYC/passport_review.pdf这样的源路径信息。因此，敏感数据管控必须覆盖所有被索引的内容，而不仅仅是文本本身。
生产级处理流程
在测试阶段，单独的清洗功能就已经足够了。但在实际生产环境中则需要更严格的标准：不能仅仅因为一系列步骤都顺利执行而没有出现异常，就直接将文档进入分块处理阶段。处理流程必须记录下它对文档做了哪些修改、哪些部分仍存在可疑之处、是否发现了敏感数据、文件是否与其他文件重复，以及最终结果是否值得被索引。
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class QualityStatus(str, Enum):
PASS = "pass"
REVIEW = "review"
REJECT = "reject"
@dataclass
class Transformation:
stage: str
operation: str
details: dict[str, Any] = field(default_factory=dict)
@dataclass
class QualityFlag:
code: str
severity: str
details: dict[str, Any] = field(default_factory=dict)
@dataclass
class RetrievalReadyDocument:
document_id: str
source: str
raw_text: str
clean_text: str
language: str | None = None
language_confidence: float | None = None
content_hash: str | None = None
duplicate_of: str | None = None
transformations: list[Transformation] = field(default_factory=list)
quality_flags: list[QualityFlag] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
status: QualityStatus = QualityStatus.PASS
raw_text与clean_text之间的差异至关重要。前者反映了数据摄取时实际提取的内容，而后者则表示我们愿意进行分块并嵌入的格式。转换日志会说明二者是如何从彼此转化而来的，当在六个月后出现检索失败时，这样的转换记录具有极高的价值。
import logging
logger = logging.getLogger(__name__)
class ProductionDocumentCleaner:
def __init__(
self,
pii_detector: BankingPIIDetector,
pii_anonymizer,
redact_pii_for_embeddings: bool = True,
) -> None:
self.pii_detector = pii_detector
self.pii_anonymizer = pii_anonymizer
self.redact_pii_for_embeddings = redact_pii_for_embeddings
def clean(self, document: dict) -> RetrievalReadyDocument:
result = RetrievalReadyDocument(
document_id=document["document_id"],
source=document["source"],
raw_text=document["text"],
clean_text=document["text"],
metadata=document.get("metadata", {}).copy(),
)
try:
self._normalize_text(result)
self._detect_language(result)
self._validate_ocr(result)
self._handle_sensitive_data(result)
self._calculate_hash(result)
self._apply_quality_gate(result)
except Exception as exc:
logger.exception("Preprocessing failed for document %s", result.document_id)
result.quality_flags.append(QualityFlag(
code="PREPROCESSING_EXCEPTION",
severity="critical",
details={"error": str(exc)},
))
result.status = QualityStatus.REJECT
return result
def _apply_quality_gate(self, document: RetrievalReadyDocument) -> None:
severities = {flag.severity for flag in document.quality_flags}
if "critical" in severities:
document.status = QualityStatus.REJECT
elif "high" in severities:
document.status = QualityStatus.REVIEW
else:
document.status = QualityStatus.PASS
该接口会输出三种明确状态中的一种。PASS表示文档可以继续分块处理，且不存在未解决的高风险问题。REVIEW意味着已生成可用的内容，但某些转换结果在索引前需要人工验证。REJECT则表示结果不可靠，无法用于索引。如果扫描后的圆形图表显示的l0%阈值存疑，则会被归类为REVIEW；而一份80页中仅有17个乱码字符的PDF很可能会被判定为REJECT；至于那些仅通过添加撇号和调整空格就能修复的文档，则属于PASS状态。
清洗阶段的各项指标也值得记录下来。如果text_reduction_ratio通常只能削减文档的4-8%，但在版本发布后却上升至63%，那就说明出现了问题。同样，如果OCR库升级后unknown_language_rate从0.3%骤升至18%，则说明上游的检测功能出现了退化。这类数据能让清洗过程变得可见且可控，而非依赖不可见的脚本。
生产环境中的权衡
没有一种清洗配置适合所有人，而这些选择都会带来实际后果。
- 编码修复。
ftfy能够以低成本高效地处理常见的乱码和Unicode损坏问题。应将其视为确定性Unicode规范化工具的辅助手段，而非随意重写文本的许可。
贯穿所有这些内容的思路是：让检索需求及避免语义失真的风险来指导预处理决策，而无需考虑处理后的文本看起来有多整洁。
回到初始政策
恢复到最初的KYC政策。数据摄取模块已成功提取相关内容，而清洗流程在将其纳入索引之前会执行更多处理步骤：修复损坏的编码而不删除合法的多语言字符，移除重复的页面标题同时保留文档元数据，识别存储在不同仓库中的完全相同的副本，并将旧版本的政策视为相关内容而非直接删除的对象。该流程会将EUR l0,000标记为严重的OCR异常，不会擅自猜测处理结果，能够识别多语言段落，在敏感客户信息到达嵌入层之前就将其检测出来，记录每一次转换过程，并决定最终结果是应通过、需审核还是被拒绝。
只有到这时，文档才会进入分块阶段。顺序很重要，因为分块无法恢复预处理过程中已丢失的信息，而嵌入模型也无法区分权威知识与未经检查就混入语料库的错误内容。
清洗之后的问题则是另一回事：如何将一份80页的AML政策合理分割？虽然创建一个包含500个标记的简单窗口很容易，但它却可能把规则与例外情况分开，或将表格的标题剥离，又或者仅仅因为某些条款共享相同的标记数量就将它们强行拼接在一起。分割方法多种多样，包括固定大小法、递归法以及基于语义、结构感知、父子关系和命题层面的方法，每种方法都需要单独处理。
分块前的检查清单
在文件进入分块和嵌入阶段之前，预处理环节应当能够对以下各项给出明确答复：
- 编码与Unicode规范化处理是否顺利完成，且没有丢失任何具有意义的字符？
- 控制字符、零宽度字符、重复的页眉页脚以及其他可能干扰检索的元素是否已得到处理？
- 是否检测到了完全相同或几乎相同的文档，同时保留了真实的文档版本？
- 对于金额、百分比、日期、标识符和法律引用等存在疑问的OCR识别结果，是否已上报给相关人员，而非擅自修改？
- 语言检测的精度是否足够，能否识别混合语言内容？
- 在嵌入之前，敏感信息是否已根据应用的安全策略进行分类？
- 每一处重大变更是否都能追溯到最初的解析文本？
- 在开始分块之前，是否有明确的通过、审查或拒绝的判定？
如果任何答案都不明确，说明该文档尚不具备检索条件。
如果一份损坏的策略文件已经被分割成80个分块，嵌入到向量索引中、复制到其他地方、被缓存并由实时应用提供服务，那么仅修复源文件已不够。必须找到所有派生出的分块，删除其对应的向量，使缓存失效，重新处理文档，并确认已移除过时的表示形式。在不良知识转化为嵌入项之前将其拒绝才是最经济的方式。
所示代码注重清晰性与正确性。在正式环境中，应添加结构化日志记录、配置与密钥管理功能，为数据转换过程保留持久审计轨迹，设置预处理指标，实现故障恢复，并对高风险修正操作进行人工审核。在受监管的环境中，每一项破坏性数据转换都应能追溯到原始数据源。
相关阅读
- 生产环境RAG的分块策略：固定大小法、语义分块法及父子节点分块法 —— 详细介绍从固定大小分割、递归分割到父子节点分块、命题分块、延迟分块、RAPTOR算法、表格处理、代码应用及基准测试等RAG分块方法的实践要点。