目录第一步文本切块chunker.py第二步虚拟环境第三步chunker.py测试运行第四步Git同步第五步多线程自动化管道src/pipeline.py第六步配置智谱大模型接口第七步填入PDF文献⚠️ 只有一个小小的“避坑点”第八步:Pipeline.py测试运行第九步Git同步下一步知识检索与大模型智能问答Retrieval Generation接下来我们要开始一行一行地编写chunker.py里面的核心清洗与切块逻辑了。第一步文本切块chunker.py完整代码如下# 语义级文本切块算法 import re from typing import List, Dict from loguru import logger # 工业级日志库比普通的 print 高级很多 class SemanticChunker: def __init__(self, chunk_size: int 500, chunk_overlap: int 100): 初始化语义切块器 :param chunk_size: 每个文本块的最大字符数控制模型的上下文窗口负载 :param chunk_overlap: 相邻文本块的重叠字符数核心防止切缝处的语义断层 self.chunk_size chunk_size self.chunk_overlap chunk_overlap def split_text(self, text: str, metadata_base: Dict) - List[Dict]: 核心方法对原始文本进行清洗、按语义边界切分并注入元数据Metadata logger.info(f开始对文件 {metadata_base.get(source, 未知)} 进行智能化语义切块...) # 1. 基础清洗ETL第一步利用正则表达式去除异常的不可见字符、控制字符以及连续的空行 text re.sub(re.compile(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\xff]), , text) text re.sub(r\n, \n, text).strip() # 2. 语义边界切分核心难点抛弃硬切按照中文的标点符号。或换行符进行切分保留完整的句子边界 sentences re.split(r(?[。\n]), text) chunks [] current_chunk for sentence in sentences: if not sentence.strip(): continue # 动态滑窗逻辑如果当前累加的句子长度没超过限制就继续往里装 if len(current_chunk) len(sentence) self.chunk_size: current_chunk sentence else: # 超过限制了说明当前块已经饱和立刻归档 if current_chunk: chunks.append(self._create_chunk_dict(current_chunk, metadata_base, len(chunks))) # 【高能踩坑点】处理重叠度Overlap # 新的一块不能直接从新句子开始而是要向前回溯保留一部分老文本chunk_overlap长度 # 这样可以确保前后两个 Chunk 之间的语义在数学向量空间上具有连续性 if len(current_chunk) self.chunk_overlap: current_chunk current_chunk[-self.chunk_overlap:] sentence else: current_chunk sentence # 别忘了收尾把最后剩下的一点尾巴文本装进最后一个 Chunk 归档 if current_chunk: chunks.append(self._create_chunk_dict(current_chunk, metadata_base, len(chunks))) logger.success(f切块完成共生成 {len(chunks)} 个高聚合、带元数据的数据块。) return chunks def _create_chunk_dict(self, content: str, base_meta: Dict, index: int) - Dict: 辅助方法为每一个切出来的文本块注入多维度的元数据Metadata标签 meta base_meta.copy() meta[chunk_id] index # 文本块的序号 meta[content_length] len(content) # 文本块的字符长度 return { page_content: content, metadata: meta } # # 本地局部测试确保这个组件能单独跑通典型的软件工程单元测试思维面试加分项 # if __name__ __main__: # 模拟一段行业原始语料 test_text 人工智能是核心生产力。大模型需要吃进高质量的语料燃料所以AI数据工程师的岗位在当前极其重要这就是我们开发自动化ETL系统的初衷。希望这个项目能帮你拿到满意的Offer。 # 实例化切块器故意把最大长度设小方便看切块效果 chunker SemanticChunker(chunk_size35, chunk_overlap12) # 模拟一个上游传入的基础元数据比如文件名和行业分类 mock_metadata {source: interview_tips.txt, category: AI_Tech} # 跑一下切块流程 results chunker.split_text(test_text, mock_metadata) # 打印出来看看成果 print(\n--- 单元测试查看切块结果 ---) for index, item in enumerate(results): print(f【Chunk {index}】内容 - {item[page_content]}) print(f 元数据 - {item[metadata]}\n)第二步虚拟环境在运行测试之前先在 Git Bash 终端里执行三行命令1.创建虚拟环境做好隔离 python -m venv .venv 2.激活虚拟环境 source .venv/Scripts/activate # 激活成功后会发现命令行开头多了一个括号 (.venv) 3.安装包 pip install -r requirements.txt安装包的过程比较长耐心等待2-5mins左右。中间需要等很久下载完成之后还有解压安装配置过程小电脑在背后默默工作哟~第三步chunker.py测试运行在终端里确保在rag-data-pipeline的根目录下做简单的单元测试python src/chunker.py结果如下第四步Git同步把这行具有里程碑意义的代码存入 Git 账本# 1. 检查当前状态你会看到 src/chunker.py 变成了红色 git status # 2. 把修改并测试通过的 chunker.py 添加到暂存区 git add src/chunker.py # 3. 签字画押feat 标签代表这是一个开发成功的新功能 git commit -m feat: implement semantic chunker with sentence boundary and overlap retention第五步多线程自动化管道src/pipeline.py作用是调用刚才写好的chunker.py批量、多线程地去把某个文件夹下的海量 PDF 文献读进来然后自动调用本地免费的 Embedding 模型变成向量一键灌入向量数据库Chroma里。pipeline.py完整代码如下import os from concurrent.futures import ThreadPoolExecutor from loguru import logger from langchain_community.vectorstores import FAISS from langchain_zhipu import ZhipuAIEmbeddings from pypdf import PdfReader # 优雅复用你的高级语义切块器 from chunker import SemanticChunker class AutoDataPipeline: def __init__(self, input_dir: str, db_dir: str): 自动化高性能数据资产管道初始化 :param input_dir: 存放待处理原始 PDF 文献的本地目录 :param db_dir: 向量数据库最终在本地持久化导出的资产目录 self.input_dir input_dir self.db_dir db_dir # 1. 实例化高级切块器组件单片上限500字重叠80字保证语义连续 self.chunker SemanticChunker(chunk_size500, chunk_overlap80) # 2. 建立云端安全网关连接完全解耦不消耗本地CPU/GPU算力 logger.info( 正在连接云端免费高维 Embedding 模型网关...) self.embeddings ZhipuAIEmbeddings( api_key替换 ) logger.success(✅ 云端模型网关握手成功) def process_single_file(self, file_name: str): 核心ETL单文件闭环从PDF中压榨文本 - 智能化切块 - 自动流控落库 if not file_name.endswith(.pdf): return file_path os.path.join(self.input_dir, file_name) logger.warning(f [流水线启动] 正在纯净并行处理文件: {file_name}) try: # ----------------------------------------------------------------- # STEP 1: Extraction (数据抽取) # 使用纯 Python 编写的 PdfReader 逐页读取文本100% 避开本地任何重量级模型库 # ----------------------------------------------------------------- reader PdfReader(file_path) text_list [] for page in reader.pages: page_text page.extract_text() if page_text: text_list.append(page_text) full_text .join(text_list) # 严格边界拦截如果是纯扫描件图片提取不出文本直接安全熔断防止污染上游 if not full_text.strip(): logger.error(f❌ 文件 {file_name} 提取内容为空可能是纯扫描件图片已拦截。) return # ----------------------------------------------------------------- # STEP 2: Transformation (数据清洗与语义切块) # 自动灌入结构化元数据Metadata方便未来在大模型检索时进行精确溯源 # ----------------------------------------------------------------- base_metadata {source: file_name, data_type: unstructured_report} chunks self.chunker.split_text(full_text, base_metadata) # ----------------------------------------------------------------- # STEP 3: Loading (工业级流控向量沉淀) # ----------------------------------------------------------------- # 【核心防御性机制】物理过滤一切空字符串或纯空格块防止触发云端参数有误报错 clean_chunks [c for c in chunks if c[page_content] and c[page_content].strip()] texts [c[page_content] for c in clean_chunks] metadatas [c[metadata] for c in clean_chunks] if not texts: logger.warning(f⚠️ 文件 {file_name} 经过纯净清洗后未检测到合法文本跳过落库。) return logger.info(f 正在将 {file_name} 的 {len(texts)} 个精纯向量节点批量沉淀至本地库...) # 【★ 终极破案手写高并发窗口滑动机制 ★】 # 旧版集成库在 from_texts 内部存在不透传参数的 Bug。我们直接在最前端手写分批 # 强行每次最多只打包 64 条数据送往智谱网关完美兼容并对齐云端的物理吞吐硬限制。 db None CHUNK_LIMIT 64 for i in range(0, len(texts), CHUNK_LIMIT): batch_texts texts[i:i CHUNK_LIMIT] batch_metadatas metadatas[i:i CHUNK_LIMIT] logger.info(f 正在发送当前网络分批: 正在写入第 {i} 到 {i len(batch_texts)} 个高维向量节点...) if db is None: # 第一批次在内存中直接拉起 FAISS 基础矩阵地基 db FAISS.from_texts(textsbatch_texts, embeddingself.embeddings, metadatasbatch_metadatas) else: # 后续批次增量追加在内存中动态拼接矩阵无任何硬盘 I/O 锁死风险 db.add_texts(textsbatch_texts, metadatasbatch_metadatas) # ----------------------------------------------------------------- # STEP 4: Serialization (原子化资产持久化) # 整个文件所有切块在内存里完美转化完毕后一次性原子落盘在 D 盘生成精纯资产 # ----------------------------------------------------------------- if db: db.save_local(self.db_dir) logger.success(f [处理成功] 知识资产已完美录入本地 FAISS 向量库: {file_name}) except Exception as e: logger.error(f❌ [处理失败] 文件 {file_name} 在流水线中发生异常: {str(e)}) def get_file_md5(self, file_path: str) - str: 【新增物理审计】计算文件的唯一MD5指纹防止同名不同内容或者改名重复执行 import hashlib hasher hashlib.md5() with open(file_path, rb) as f: # 每次读取 4KB防止大文件吃爆内存 for chunk in iter(lambda: f.read(4096), b): hasher.update(chunk) return hasher.hexdigest() def run_pipeline(self, max_workers: int 4): 高性能并发总控引擎引入工业级断点续传与智能去重审计 if not os.path.exists(self.input_dir): os.makedirs(self.input_dir) logger.warning(f输入目录 {self.input_dir} 不存在已自动创建。) return # 1. 初始化/读取成功落库的历史文献 log_file_path os.path.join(self.db_dir, processed_files.txt) processed_hashes set() if os.path.exists(log_file_path): with open(log_file_path, r, encodingutf-8) as lf: processed_hashes set(line.strip() for line in lf if line.strip()) raw_files [f for f in os.listdir(self.input_dir) if f.endswith(.pdf)] if not raw_files: logger.error(f 发现待处理的 PDF 文件数量为 0。请先往 {self.input_dir} 文件夹中放置 PDF 文献) return # 2. 【核心去重漏斗】遍历文件计算指纹剔除已经处理过的文献 files_to_process [] for file_name in raw_files: file_path os.path.join(self.input_dir, file_name) file_hash self.get_file_md5(file_path) if file_hash in processed_hashes: # 瞬间跳过控制台亮起优雅的白色日志 logger.info(f⏭️ [智能跳过] 检测到文献 {file_name} 历史已成功落库无需重复计算。) else: # 登记进入本次动工的“生死簿” files_to_process.append((file_name, file_hash)) if not files_to_process: logger.success(✨ [完美对齐] 目录内所有文献均已在本地向量库中无任何新任务需要执行) return logger.info(f 自动化并行数据管道触发待冲锋新任务: {len(files_to_process)} 个分配线程数: {max_workers}) # 3. 剥离出纯文件名列表扔给线程池并行动工 pure_file_names [item[0] for item in files_to_process] with ThreadPoolExecutor(max_workersmax_workers) as executor: executor.map(self.process_single_file, pure_file_names) # 4. 【全量大捷后追加账本】将本次成功动工的指纹追加写入物理账本固化成果 # 注意为了简化逻辑默认本次多线程运行全部成功。工业上会严格配合 try-catch 成功后才写入 os.makedirs(self.db_dir, exist_okTrue) with open(log_file_path, a, encodingutf-8) as lf: for _, file_hash in files_to_process: lf.write(f{file_hash}\n) logger.success( [大获全胜] 管道全线畅通新版增量向量资产追加完毕) if __name__ __main__: # 配置统一输入输出目录 INPUT_DIRECTORY ./data/input VECTOR_DB_DIRECTORY ./data/vector_db pipeline AutoDataPipeline(input_dirINPUT_DIRECTORY, db_dirVECTOR_DB_DIRECTORY) pipeline.run_pipeline(max_workers2)第六步配置智谱大模型接口智谱AI开放平台1. 在终端安装智谱 AI 的免费开源官方桥接件 pip install langchain-community 2. 找到下面的代码行替换成自己的 self.embeddings ZhipuAIEmbeddings( zhipuai_api_key替换 )第七步填入PDF文献打开rag-data-pipeline文件夹。创建一个data文件夹点进去再建一个名为input的子文件夹。从你的电脑里随便找一两篇普通的 PDF 格式文件比如你以前的大学课件、论文、说明书或者任何 PDF 报告都行把它们复制并粘贴到data/input/文件夹里面。⚠️ 只有一个小小的“避坑点”千万不要找那种“纯图片扫描版”的 PDF就是那种里面的文字用鼠标完全没办法选中、像一张张照片拼起来的 PDF。因为我们目前使用的是轻量级的文本提取器它只能读取 PDF 里的“原生文本”。如果是纯图片它会识别出空内容。第八步:Pipeline.py测试运行python src/pipeline.py大获全胜第一次测试用一个文件做测试即可没问题之后再多加入几个pdf文献比如三个文件一起测试多线程并发。代码中加入来筛选和清洗避免文件重复解析录入浪费资源。再额外加入另一个文件测试是否会被准确跳过。可以观察到之前的三个已经被跳过了非常成功再次运行可以发现全部跳过。第九步Git同步# 1. 检查当前状态你会看到 src/chunker.py 变成了红色 git status # 2. 把修改并测试通过的 chunker.py 添加到暂存区 git add src/pipeline.py # 3. 签字画押feat 标签代表这是一个开发成功的新功能 git commit -m feat: implement semantic pipeline with sentence boundary and overlap retention下一篇知识检索与大模型智能问答部分Retrieval Generation
