打造可配置、可扩展的自动化预处理流水线从PDF到RAG知识库在构建RAG检索增强生成系统的过程中文档预处理是决定检索质量的基础环节。然而面对多源异构文档如PDF、图片、扫描件手动处理不仅效率低下还难以保证一致性和可维护性。为此我们需要构建一条自动化预处理流水线将OCR识别、文本清洗、分段、元数据绑定、知识库上传等步骤串联起来并具备可配置和可扩展的能力从而适应不同业务场景和未来需求的变化。本文将以一个实际案例为主线展示如何从PDF或图片开始经过PaddleOCR提取文本、预处理分段、上传至Dify知识库并绑定元数据最终实现全流程自动化。我们将重点介绍配置驱动和插件化扩展的设计思想并提供核心代码片段帮助你快速搭建自己的RAG预处理流水线。一、整体架构概览整个预处理流水线的流程如下textPDF/图片 → [PaddleOCR API] → 提取纯文本 → [预处理/分段] → [Dify API] → 知识库同步 → [元数据绑定]所有模块的参数均通过一个集中配置文件config.yaml管理保证修改灵活同时支持插件化新增预处理规则并通过Celery实现异步任务队列满足大规模文档处理需求。二、核心模块实现1. OCR文本提取模块支持多引擎切换我们封装了一个OCREngine类通过配置文件指定使用的OCR引擎如PaddleOCR及其API地址。这样可以轻松替换为其他OCR服务。pythonimport requests class OCREngine: def __init__(self, config): self.api_url config[ocr][api_url] # 从配置加载OCR API地址 self.high_precision config[ocr].get(high_precision, False) def extract_text(self, file_path): 调用OCR API提取文本返回JSON格式结果 with open(file_path, rb) as f: files {file: f} response requests.post(self.api_url, filesfiles) if response.status_code 200: return response.json() # 包含识别出的文本及其他信息 else: raise Exception(fOCR提取失败: {response.text})设计要点通过配置文件指定API地址支持切换本地部署或云服务。返回的JSON包含完整信息便于下游模块进一步筛选如只取文本字段。2. 文本预处理与分段模块该模块负责清洗OCR结果并按规则进行分段。我们将其设计为插件式方便新增预处理规则如去空格、去URL、去特殊字符等。pythonimport re class TextProcessor: def __init__(self, config): self.config config self.rules config[dify][process_rules] # 插件列表可动态加载 self.plugins [ RemoveExtraSpaces(enabledTrue), RemoveSpecialChars(enabledTrue), # 后续可扩展 ] def preprocess(self, text): 应用所有启用的预处理插件 for plugin in self.plugins: if plugin.enabled: text plugin.apply(text) return text def segment(self, text): 根据配置的分隔符切分文本 separator self.rules[segmentation][separator] # 简单的分隔符分割实际可用更复杂的逻辑 return re.split(separator, text) # 示例插件去除多余空格 class RemoveExtraSpaces: def __init__(self, enabledTrue): self.enabled enabled def apply(self, text): return re.sub(r\s, , text).strip() # 示例插件去除特殊字符 class RemoveSpecialChars: def __init__(self, enabledTrue): self.enabled enabled def apply(self, text): return re.sub(r[^\w\s], , text)设计要点预处理规则通过插件类实现可灵活开关和组合。分段规则如分隔符、最大token数由配置文件定义便于调整。3. 上传至Dify知识库模块封装Dify API支持通过文本直接创建文档并可指定分段规则复用配置中的规则。pythonimport requests import json class DifyUploader: def __init__(self, config): self.api_key config[dify][api_key] self.dataset_id config[dify][dataset_id] self.base_url fhttps://api.dify.ai/v1/datasets/{self.dataset_id} self.rules config[dify][process_rules] def upload_by_text(self, segments): 将分段后的文本列表上传为Dify文档 headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } document_ids [] for i, content in enumerate(segments): payload { name: fsegment_{i}.txt, text: content, indexing_technique: high_quality, process_rule: self.rules # 复用配置中的分段规则 } response requests.post( f{self.base_url}/document/create_by_text, headersheaders, datajson.dumps(payload) ) if response.status_code 200: doc_id response.json()[document][id] document_ids.append(doc_id) print(fSegment {i} uploaded, ID: {doc_id}) else: print(fUpload failed: {response.text}) return document_ids设计要点使用Dify官方API支持通过文本创建文档。分段规则直接取自配置保证与预处理阶段一致。4. 动态元数据绑定模块元数据可以帮助后续检索时进行过滤或溯源。我们通过配置定义元数据字段及其值来源如固定值、文件名等。pythonclass MetadataManager: def __init__(self, config): self.api_key config[dify][api_key] self.dataset_id config[dify][dataset_id] self.base_url fhttps://api.dify.ai/v1/datasets/{self.dataset_id} self.metadata_fields config[metadata][fields] def bind_metadata(self, document_id, filename): 为指定文档绑定元数据 url f{self.base_url}/documents/{document_id}/metadata headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} metadata_list [] for field in self.metadata_fields: value field[value] if field.get(value_from) filename: value filename # 从文件名动态取值 metadata_list.append({ name: field[name], value: value }) payload {metadata: metadata_list} response requests.post(url, headersheaders, datajson.dumps(payload)) if response.status_code 200: print(fMetadata bound for document {document_id}) else: print(fMetadata binding failed: {response.text})设计要点支持从文件名、固定值等来源动态生成元数据。调用Dify的元数据绑定API将字段与文档关联。三、可配置设计集中管理参数所有模块的配置统一存放在config.yaml文件中示例如下yamlocr: api_url: http://localhost:8080/ocr # PaddleOCR API地址 high_precision: true dify: api_key: your_api_key dataset_id: your_dataset_id process_rules: pre_processing: - id: remove_extra_spaces enabled: true segmentation: separator: \n\n # 按段落分隔 max_tokens: 500 metadata: fields: - name: author value: default_author - name: source value_from: filename # 从文件名提取在Python中加载配置pythonimport yaml def load_config(): with open(config.yaml, r, encodingutf-8) as f: return yaml.safe_load(f)通过这种方式修改OCR地址、切换预处理规则、调整分段参数等均无需改动代码只需更新配置文件。四、可扩展设计插件与异步任务1. 插件化预处理规则如前所述我们将每个预处理操作封装为独立的插件类在TextProcessor中统一调用。新增规则时只需创建新的插件类并添加到插件列表即可无需修改核心逻辑。2. 异步任务支持Celery当需要批量处理大量文档时同步处理会阻塞主流程。我们可以引入Celery将任务异步化提高吞吐量。pythonfrom celery import Celery app Celery(tasks, brokerredis://localhost:6379/0) app.task def process_pdf_async(pdf_path): config load_config() # 1. OCR提取 ocr_engine OCREngine(config) raw_result ocr_engine.extract_text(pdf_path) raw_text raw_result.get(text, ) # 假设返回JSON中有text字段 # 2. 预处理与分段 processor TextProcessor(config) cleaned_text processor.preprocess(raw_text) segments processor.segment(cleaned_text) # 3. 上传至Dify uploader DifyUploader(config) doc_ids uploader.upload_by_text(segments) # 4. 绑定元数据 metadata_mgr MetadataManager(config) for doc_id in doc_ids: metadata_mgr.bind_metadata(doc_id, pdf_path)通过Celery我们可以将处理任务放入队列由worker异步执行支持定时调度如每天凌晨处理新增文件。五、完整流程演示将上述模块串联起来main.py如下pythondef main(pdf_path): config load_config() # 1. OCR提取 ocr_engine OCREngine(config) raw_result ocr_engine.extract_text(pdf_path) raw_text raw_result[text] # 假设返回JSON中包含text字段 # 2. 预处理与分段 processor TextProcessor(config) cleaned_text processor.preprocess(raw_text) segments processor.segment(cleaned_text) # 3. 上传至Dify uploader DifyUploader(config) doc_ids uploader.upload_by_text(segments) # 4. 绑定元数据 metadata_mgr MetadataManager(config) for doc_id in doc_ids: metadata_mgr.bind_metadata(doc_id, pdf_path) if __name__ __main__: main(example.png) # 支持PDF或图片执行后你将看到类似如下的输出OCR提取完成得到文本。文本被预处理并分为若干段。各段依次上传至Dify知识库返回文档ID。元数据成功绑定。在Dify后台刷新即可看到新创建的文档及其元数据。六、总结与最佳实践通过本文的实践我们构建了一条可配置、可扩展的自动化预处理流水线其核心设计思想值得借鉴配置驱动将所有可变参数外置到YAML文件避免硬编码便于调整和维护。模块化封装将OCR、预处理、上传、元数据等独立为类职责清晰便于替换和升级。插件化扩展预处理规则以插件形式组织新增功能只需添加新插件无需修改原有代码。异步任务结合Celery实现高并发处理支持大规模文档批量导入。元数据动态绑定通过配置定义元数据来源增强文档的可管理性和检索能力。这套流水线不仅适用于Dify稍作修改即可对接其他RAG平台如RAGFlow、LlamaIndex等。它解放了人工操作让RAG系统的数据接入变得高效、规范、可持续。未来你还可以在此基础上增加更多功能如多模态OCR、自动语言检测、自定义分块算法等使其成为企业级RAG预处理的标准组件。