企业官网建设流程全解析

目录

第一步：文本切块（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_size=35, chunk_overlap=12) # 模拟一个上游传入的基础元数据（比如文件名和行业分类） 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_size=500, chunk_overlap=80) # 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(texts=batch_texts, embedding=self.embeddings, metadatas=batch_metadatas) else: # 后续批次：增量追加，在内存中动态拼接矩阵，无任何硬盘 I/O 锁死风险 db.add_texts(texts=batch_texts, metadatas=batch_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", encoding="utf-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_workers=max_workers) as executor: executor.map(self.process_single_file, pure_file_names) # 4. 【全量大捷后追加账本】将本次成功动工的指纹追加写入物理账本，固化成果 # 注意：为了简化逻辑，默认本次多线程运行全部成功。工业上会严格配合 try-catch 成功后才写入 os.makedirs(self.db_dir, exist_ok=True) with open(log_file_path, "a", encoding="utf-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_dir=INPUT_DIRECTORY, db_dir=VECTOR_DB_DIRECTORY) pipeline.run_pipeline(max_workers=2)

第六步：配置智谱大模型接口

智谱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）