企业微调数据管理：内部文档如何转化成高质量训练样本
一、个性化深度引言
做企业微调最尴尬的不是模型效果不好，而是效果不稳定——今天用这批数据训好了，下周新数据加入后直接退化。追查下去，问题十有八九出在数据处理环节。
内部文档转训练样本这件事，听起来像是格式转换，做起来才知道是个系统工程。企业文档的特点是：格式混乱（Word/PDF/Excel/扫描件混在一起）、内容非结构化或半结构化、信息密度不均匀（有的段落信息量极高，有的纯粹是套话）、部门间术语不统一。直接拿去微调，模型会学到很多噪声。
我经历过一个项目：用3000份内部技术文档微调一个代码辅助模型，训练集 loss 降到0.12，但上线后用户反馈"有时候很准，有时候胡言乱语"。排查了三天，发现原因是数据里掺杂了大量会议纪要——模型从纪要里学到了"李经理说周五之前要交"，然后把这个当成技术回答输出了。
数据质量决定了微调的上限，这句话不是套话。下面把这条工程链路讲清楚。
二、个性化原理剖析
从原始企业文档到可用的训练样本，需要经过五个步骤。这张图概括了整个流程：
每一步都要回答一个问题：这步做不好会有什么后果？
格式统一做不好，模型会学到 PDF 的换行符和编码噪声。去噪清洗不彻底，PII（个人身份信息）泄露是法律风险。信息提取不到位，高质量段落被埋没在噪声里。质量分级不准确，低质量样本会拖累整个模型的输出水平。
最关键的是质量分级这一步。信息密度评分可以用 LLM 做 auto-grading，给每个段落打分（1-5分），再按分数段分配不同的样本生成策略。
三、个性化代码实践
下面是一段文档预处理和样本生成的 Python 代码：
import hashlib import re from dataclasses import dataclass, field from typing import List, Tuple, Optional from collections import Counter @dataclass class DocumentChunk: """文档分块——设计原因：chunk是数据处理的最小单位，统一结构方便管线化""" chunk_id: str text: str source_file: str page_num: Optional[int] = None quality_score: float = 0.0 def __hash__(self): # 用文本hash做去重——设计原因：MD5比字符串比较快10倍以上 return hash(hashlib.md5(self.text.encode()).hexdigest()) class EnterpriseDocProcessor: """企业文档处理管线——设计原因：链式调用，每步可独立开关""" # PII正则模式——设计原因：编译一次复用，避免重复编译开销 PII_PATTERNS = { "phone": re.compile(r"1[3-9]\d{9}"), "id_card": re.compile(r"\d{17}[\dXx]"), "email": re.compile(r"[\w.-]+@[\w.-]+\.\w+"), "bank_card": re.compile(r"\d{16,19}"), } # 低质量模式——设计原因：识别"套话"段落，降低其权重 BOILERPLATE_PATTERNS = [ r"^本[文档文档].*仅供参考", r"^以上内容.*最终解释权", r"^如有.*问题.*请.*联系", r"^此致\s*$", r"^第\d+页.*共\d+页", ] def __init__(self, enable_pii_mask: bool = True): self.enable_pii_mask = enable_pii_mask self._boilerplate_re = [re.compile(p) for p in self.BOILERPLATE_PATTERNS] def clean_text(self, text: str) -> str: """文本清洗——设计原因：集中清洗逻辑，方便统一调整规则不散落各处""" # 1. 统一换行符 text = text.replace("\r\n", "\n").replace("\r", "\n") # 2. 去除多余空白行——设计原因：过多空行影响chunk语义完整性 text = re.sub(r"\n{3,}", "\n\n", text) # 3. 去除特殊Unicode字符——设计原因：某些PDF转换会产生不可见字符 text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text) # 4. PII脱敏——设计原因：必须在清洗阶段做，样本生成阶段来不及 if self.enable_pii_mask: for pii_type, pattern in self.PII_PATTERNS.items(): text = pattern.sub(f"[{pii_type.upper()}_MASKED]", text) return text.strip() def split_chunks(self, text: str, source: str, chunk_size: int = 512) -> List[DocumentChunk]: """智能分块——设计原因：固定512字+语义边界，比纯按字数切更合理""" paragraphs = text.split("\n\n") chunks = [] current_chunk = [] current_len = 0 for para in paragraphs: para = para.strip() if not para: continue para_len = len(para) # 超过chunk_size的长段落单独成块——设计原因：长段落通常是完整逻辑单元 if para_len > chunk_size * 1.5: if current_chunk: chunks.append(self._make_chunk( "\n".join(current_chunk), source, len(chunks) )) current_chunk = [] current_len = 0 chunks.append(self._make_chunk(para, source, len(chunks))) continue # 积累到接近chunk_size时切分——设计原因：512字对大部分embedding模型是合理上限 if current_len + para_len > chunk_size and current_chunk: chunks.append(self._make_chunk( "\n".join(current_chunk), source, len(chunks) )) current_chunk = [para] current_len = para_len else: current_chunk.append(para) current_len += para_len # 收尾——设计原因：不能丢失剩余文本 if current_chunk: chunks.append(self._make_chunk( "\n".join(current_chunk), source, len(chunks) )) return chunks def score_quality(self, chunk: DocumentChunk) -> float: """信息密度评分——设计原因：多维度加权评分比单维度更稳定""" text = chunk.text score = 5.0 # 满分5分，从满分往下扣 # 1. 长度惩罚——设计原因：小于100字的chunk通常是标题或分隔符 if len(text) < 100: score -= 2.0 elif len(text) < 200: score -= 1.0 # 2. 套话检测——设计原因：法律声明/免责条款/格式说明信息量为零 for pattern in self._boilerplate_re: if pattern.search(text): score -= 1.5 break # 3. 数字和术语密度——设计原因：技术文档的有效信息通常包含专有名词和数值 tech_term_count = len(re.findall(r"[A-Z][a-z]+|[0-9]+[a-zA-Z]*", text)) term_density = tech_term_count / max(len(text.split()), 1) if term_density < 0.05: score -= 1.0 elif term_density > 0.15: score += 0.5 # 技术密度高加分 # 4. 重复度惩罚——设计原因：全是"的的的"这类文本信息量为零 words = text.split() if words: unique_ratio = len(set(words)) / len(words) if unique_ratio < 0.3: score -= 2.0 return max(0.0, min(5.0, score)) def generate_samples(self, chunk: DocumentChunk, quality_level: str) -> List[dict]: """根据质量等级生成不同格式的训练样本——设计原因：好料精做，差料粗用""" text = chunk.text if quality_level == "A": # A级：生成指令跟随样本（问答对）——设计原因：SFT阶段最需要的格式 return [{ "instruction": f"请根据以下文档内容回答问题: {text[:100]}...", "input": "请总结这段内容的核心要点。", "output": f"[需要人工标注]" }] elif quality_level == "B": # B级：生成补全样本——设计原因：用于continue pretraining return [{ "prefix": text[:len(text)//2], "suffix": text[len(text)//2:] }] else: # C级：直接作为预训练语料——设计原因：清洗过就行，不生成instruction return [{"text": text}] def _make_chunk(self, text: str, source: str, idx: int) -> DocumentChunk: chunk_id = f"{source}_chunk_{idx:04d}" return DocumentChunk(chunk_id=chunk_id, text=text, source_file=source) def deduplicate(self, chunks: List[DocumentChunk]) -> List[DocumentChunk]: """去重——设计原因：企业内部文档大量重复引用，不去重会导致过拟合""" seen_hashes = set() unique = [] for chunk in chunks: h = hash(chunk) if h not in seen_hashes: seen_hashes.add(h) unique.append(chunk) return unique def pipeline(self, raw_texts: List[Tuple[str, str]]) -> List[dict]: """完整处理管线——设计原因：单一入口，方便集成到调度系统""" all_chunks = [] # 第一阶段：清洗+分块 for text, source in raw_texts: cleaned = self.clean_text(text) chunks = self.split_chunks(cleaned, source) all_chunks.extend(chunks) # 第二阶段：去重 all_chunks = self.deduplicate(all_chunks) # 第三阶段：评分+分级 for chunk in all_chunks: chunk.quality_score = self.score_quality(chunk) # 统计分布——设计原因：方便评估数据质量，输出报告 scores = [c.quality_score for c in all_chunks] print(f"总chunk数: {len(all_chunks)}") print(f"平均质量分: {sum(scores)/len(scores):.2f}") print(f"质量分布: {Counter(round(s,1) for s in scores)}") return all_chunks # 执行示例 processor = EnterpriseDocProcessor() raw_data = [ ("这是一份技术文档的内容...具体的技术参数包括CPU架构...", "doc_001.txt"), ("会议纪要：李经理说周五之前要交方案。", "doc_002.txt"), ] chunks = processor.pipeline(raw_data) print(f"处理完成，共{len(chunks)}个chunk")代码中最容易被忽略的细节是 PII 脱敏的时机——必须在清洗阶段做，不能在样本生成阶段补。因为数据会在多个环节被缓存和传递，脱敏越早，风险面越小。
四、个性化边界权衡
清洗力度 vs 信息保留
过度清洗会丢失有用的上下文。比如去除页眉页脚时，可能把章节标题也误删了。我们的做法是：第一版保守清洗（只去明确的噪声），训练后用 attention 可视化分析模型关注了哪些 token，再根据分析结果微调清洗规则。
自动化评分 vs 人工标注
LLM 自动打分快但不够准，人工标注准确但贵。A/B 测试的结果是：对 B/C 级样本，自动评分足够；对 A 级样本（用于 SFT 的指令样本），必须人工校验。这样成本控制在可以接受的范围内，质量也没打折扣。
样本多样性 vs 规模优先
3000份文档不代表有3000份"不同的"文档。实际去重后可能只剩1500份有效内容。与其追求数据量的数字好看，不如确保覆盖度——每个业务场景至少50条高质量样本，比每个场景500条低质量样本效果更好。
五、总结
企业文档转训练样本的核心步骤为：格式统一与 OCR 识别、文本清洗与 PII 脱敏、语义分块、信息密度评分、按质量等级生成差异化样本。评分体系从长度、套话检测、术语密度、重复度四个维度综合打分。代码实现需在清洗阶段完成脱敏以保证安全性边界最小化。实施中需平衡清洗力度与信息保留、自动评分与人工校验、样本多样性与数据规模的关系。最终目标是构建一个可持续迭代的数据管线，而非一次性数据清理项目。
