一、文档分块先按照markdown文档结构构建合适的分块策略- 将Markdown格式的公司文档按照语义进行智能分块- 支持三种分块类型employee员工信息、organization组织架构、business业务流程- 自动提取标题路径和元数据- 统计token数量和分块信息分块实例{ content: ### 姓名 - 岗位\n**电话**: 电话\n**岗位职责**: 岗位职责\n**项目经历**: 项目经历\n**薪资结构**: 薪资结构月薪, chunk_id: 0, title_path: 模拟公司数据适配RAG人事智能体知识库构建 一、详细人员资料15人覆盖核心岗位 姓名, section_type: employee, metadata: { employee_name: 姓名, phone: 电话, position: 岗位, section: 人员资料 }, tokens: 47 },作者提问为什么要分的这么零碎和打标签方便后期的元数据筛选在处理库原始文档时提前分块只要元数据标签打的好回答准确性嘎嘎好。代码import re import time from typing import List, Dict, Optional from dataclasses import dataclass from pathlib import Path import json dataclass class DocumentChunk: 文档分块数据类存储单个分块的内容和元数据 content: str # 分块内容 chunk_id: int # 分块ID title_path: str # 标题路径如## 一、详细人员资料 ### 张三 section_type: str # 分块类型employee/organization/business metadata: Dict # 元数据员工姓名、部门等 tokens: int # token数量估算 class DocumentChunker: 文档分块器负责将Markdown文档按照语义分块策略进行分块 def __init__(self, file_path: str): 初始化文档分块器 Args: file_path: Markdown文档路径 self.file_path Path(file_path) # 文档路径 self.chunks: List[DocumentChunk] [] # 存储分块结果的列表 self.title_stack: List[str] [] # 标题栈用于维护标题路径 def load_document(self) - str: 加载文档内容 Returns: 文档内容字符串 print(f 正在加载文档: {self.file_path.name}) start_time time.time() try: with open(self.file_path, r, encodingutf-8) as f: content f.read() load_time time.time() - start_time print(f✅ 文档加载成功耗时: {load_time:.2f}秒) print(f 文档大小: {len(content)} 字符) return content except Exception as e: print(f❌ 文档加载失败: {str(e)}) raise def extract_title_level(self, line: str) - Optional[int]: 提取标题级别 Args: line: 文本行 Returns: 标题级别1-3如果不是标题则返回None match re.match(r^(#{1,3})\s(.)$, line)# 匹配标题行 if match: return len(match.group(1)) return None def update_title_stack(self, line: str, level: int): 更新标题栈维护标题路径 Args: line: 标题行 level: 标题级别 title_text re.sub(r^#\s, , line).strip() # 清空当前级别以下的标题 self.title_stack self.title_stack[:level-1] # 添加当前标题 if len(self.title_stack) level: self.title_stack.append(title_text) else: self.title_stack[level-1] title_text def get_title_path(self) - str: 获取当前标题路径 Returns: 标题路径字符串 return .join(self.title_stack) def is_employee_row(self, line: str) - bool: 判断是否为员工数据行表格行 Args: line: 文本行 Returns: 是否为员工数据行 return | in line and not line.strip().startswith(|---) def parse_employee_row(self, line: str, title_path: str) - Optional[DocumentChunk]: 解析员工数据行创建分块 Args: line: 员工数据行 title_path: 标题路径 Returns: 员工分块对象 cells [cell.strip() for cell in line.split(|)[1:-1]] if len(cells) 6: return None name, phone, position, duties, projects, salary cells # 构建分块内容 content f### {name} - {position} **电话**: {phone} **岗位职责**: {duties} **项目经历**: {projects} **薪资结构**: {salary} # 提取元数据 metadata { employee_name: name, phone: phone, position: position, section: 人员资料 } # 估算token数量中文约1.5字符1token tokens len(content) // 1.5 return DocumentChunk( contentcontent, chunk_idlen(self.chunks), title_pathf{title_path} {name}, section_typeemployee, metadatametadata, tokensint(tokens) ) def parse_organization_block(self, lines: List[str], start_idx: int, title_path: str) - DocumentChunk: 解析组织架构块 Args: lines: 所有文本行 start_idx: 起始行索引 title_path: 标题路径 Returns: 组织架构分块对象 content_lines [] current_idx start_idx # 提取部门信息 department_match re.search(r├──\s*(.?)\d人, lines[start_idx]) if department_match: department_name department_match.group(1) else: department_name 未知部门 # 收集部门内容直到遇到下一个部门或章节 while current_idx len(lines): line lines[current_idx] # 遇到新的部门或章节则停止 if (line.strip().startswith(├──) or line.strip().startswith(└──) or self.extract_title_level(line) is not None): if current_idx start_idx: break content_lines.append(line) current_idx 1 content \n.join(content_lines) # 提取元数据 metadata { department: department_name, section: 组织架构 } tokens len(content) // 1.5 return DocumentChunk( contentcontent, chunk_idlen(self.chunks), title_pathtitle_path, section_typeorganization, metadatametadata, tokensint(tokens) ) def parse_business_block(self, lines: List[str], start_idx: int, title_path: str) - DocumentChunk: 解析业务逻辑块 Args: lines: 所有文本行 start_idx: 起始行索引 title_path: 标题路径 Returns: 业务逻辑分块对象 content_lines [] current_idx start_idx # 提取业务类型 business_type_match re.search(r\*\*(.?)\*\*, lines[start_idx]) if business_type_match: business_type business_type_match.group(1) else: business_type 未知业务 # 收集业务内容直到遇到下一个业务点或章节 while current_idx len(lines): line lines[current_idx] # 遇到新的业务点或章节则停止 if (re.search(r^\d\.\s*\*\*, line) or self.extract_title_level(line) is not None): if current_idx start_idx: break content_lines.append(line) current_idx 1 content \n.join(content_lines) # 提取元数据 metadata { business_type: business_type, section: 业务逻辑 } tokens len(content) // 1.5 return DocumentChunk( contentcontent, chunk_idlen(self.chunks), title_pathtitle_path, section_typebusiness, metadatametadata, tokensint(tokens) ) def chunk_document(self, content: str) - List[DocumentChunk]: 对文档进行分块处理 Args: content: 文档内容 Returns: 分块列表 print(f\n 开始文档分块处理...) start_time time.time() lines content.split(\n)# 按行分割文档内容 current_section None # 当前章节类型 i 0 while i len(lines): line lines[i] # 检查是否为标题 title_level self.extract_title_level(line) if title_level: self.update_title_stack(line, title_level)# 更新标题栈 title_path self.get_title_path() # 判断当前章节类型 if 详细人员资料 in title_path: current_section employee elif 组织架构 in title_path: current_section organization elif 核心业务逻辑 in title_path: current_section business i 1 continue # 根据章节类型进行分块 if current_section employee and self.is_employee_row(line): chunk self.parse_employee_row(line, title_path) if chunk: self.chunks.append(chunk)# 加入分块列表 print(f ✓ 分块 {chunk.chunk_id 1}: {chunk.metadata[employee_name]} - {chunk.metadata[position]}) elif current_section organization and ├── in line: chunk self.parse_organization_block(lines, i, title_path) self.chunks.append(chunk) print(f ✓ 分块 {chunk.chunk_id 1}: {chunk.metadata[department]}) i 1 # 跳过已处理的行 elif current_section business and re.search(r^\d\.\s*\*\*, line): chunk self.parse_business_block(lines, i, title_path) self.chunks.append(chunk) print(f ✓ 分块 {chunk.chunk_id 1}: {chunk.metadata[business_type]}) i 1 chunk_time time.time() - start_time print(f\n✅ 分块完成共生成 {len(self.chunks)} 个分块) print(f⏱️ 分块耗时: {chunk_time:.2f}秒) return self.chunks def print_statistics(self): 打印分块统计信息 print(f\n 分块统计信息:) print(f{*60}) # 按类型统计 type_counts {} for chunk in self.chunks: type_counts[chunk.section_type] type_counts.get(chunk.section_type, 0) 1 print(f分块类型分布:) for section_type, count in type_counts.items(): print(f • {section_type}: {count} 个) # Token统计 total_tokens sum(chunk.tokens for chunk in self.chunks) avg_tokens total_tokens / len(self.chunks) if self.chunks else 0 print(f\nToken统计:) print(f • 总Token数: {total_tokens}) print(f • 平均Token数: {avg_tokens:.1f}) print(f • 最大Token数: {max(chunk.tokens for chunk in self.chunks) if self.chunks else 0}) print(f • 最小Token数: {min(chunk.tokens for chunk in self.chunks) if self.chunks else 0}) print(f{*60}\n) def export_chunks(self, output_path: str chunks.json): 导出分块到JSON文件 Args: output_path: 输出文件路径 print(f 正在导出分块到: {output_path}) start_time time.time() # 转换为可序列化的格式 chunks_data [] for chunk in self.chunks: chunks_data.append({ content: chunk.content, chunk_id: chunk.chunk_id, title_path: chunk.title_path, section_type: chunk.section_type, metadata: chunk.metadata, tokens: chunk.tokens }) # 写入JSON文件 with open(output_path, w, encodingutf-8) as f: json.dump(chunks_data, f, ensure_asciiFalse, indent2) export_time time.time() - start_time print(f✅ 导出完成耗时: {export_time:.2f}秒) print(f 文件路径: {Path(output_path).absolute()}) def preview_chunks(self, num_chunks: int 3): 预览前几个分块 Args: num_chunks: 预览数量 print(f\n 分块预览前 {min(num_chunks, len(self.chunks))} 个:) print(f{*60}) for i, chunk in enumerate(self.chunks[:num_chunks]): print(f\n【分块 {chunk.chunk_id 1}】) print(f标题路径: {chunk.title_path}) print(f分块类型: {chunk.section_type}) print(fToken数: {chunk.tokens}) print(f元数据: {chunk.metadata}) print(f\n内容预览:) print(chunk.content[:200] ... if len(chunk.content) 200 else chunk.content) print(f{-*60}) print(f{*60}\n) def main(): 主函数演示文档分块流程 print(f\n{*60}) print(f 文档分块处理系统) print(f{*60}\n) total_start_time time.time() # 创建分块器实例 chunker DocumentChunker(公司数据.md) # 加载文档 content chunker.load_document() # 分块处理 chunker.chunk_document(content) # 打印统计信息 chunker.print_statistics() # 预览分块 chunker.preview_chunks(num_chunks3) # 导出分块 chunker.export_chunks(chunks.json) total_time time.time() - total_start_time print(f\n 全部处理完成总耗时: {total_time:.2f}秒) print(f{*60}\n) if __name__ __main__: main()二、代码解析说明不使用递归分块 而是采用自定义的线性扫描流程。它通过逐行分析文档内容根据章节类型和内容特征进行分块。代码流程图流程说明初始化阶段创建DocumentChunker实例传入Markdown文档路径初始化分块列表和标题栈文档加载与预处理读取文档内容并统计基本信息将文档按行分割便于逐行处理核心分块处理循环标题识别识别#、##、###标题更新标题路径栈章节类型判断根据标题路径确定当前处理章节类型员工/组织/业务分块解析员工分块解析表格行提取姓名、电话、职位等信息组织分块解析树状结构提取部门信息业务分块解析编号列表提取业务逻辑点分块创建为每个识别出的语义单元创建DocumentChunk对象后处理与输出统计分块数量、类型分布、Token信息提供分块内容预览功能将分块结果导出为结构化JSON文件