目录1 切割核心原则2 RAG 文档切分核心思路3 不同类型文档的切分形式4 补充切分的核心原则5 代码实现步骤5.1 导入与配置5.2 主流程定义5.3 步骤 1: 获取输入 (Step 1: Get Inputs)5.4 步骤 2: 标题初切 (Step 2: Split by Titles)5.5 步骤 3: 精细化 (Step 3: Refine Chunks)5.6 步骤 4: 备份与更新 (Step 4: Backup Update)5.7 单元测试 (Unit Test)1 切割核心原则文档切分 (node_document_split)文件:app/import_process/agent/nodes/node_document_split.py核心是将长篇、无结构 / 半结构化的 Markdown 文档转化为「可直接向量化入库、可精准检索」的微观语义单元Chunk是 RAG 知识库的核心基础 —— 切分质量直接决定后续检索召回率、答案准确率没有合理切分再好的向量模型也无法发挥作用。语义完整不切断句子、代码块、表格、条款保证单个 Chunk 是「完整的知识点 / 逻辑单元」长度可控Chunk 大小适配大模型上下文窗口和 Embedding 模型输入限制不超上限、不碎片化边界清晰相邻 Chunk 保留合理重叠避免关键信息落在两块交界处可追溯每个 Chunk 携带完整元数据标题、父标题、文件名便于后续溯源和引用。2 RAG 文档切分核心思路文档切分的核心逻辑是「先保语义再控大小最后补细节」全程遵循「3 步核心流程」所有操作都围绕「不破坏语义、适配检索」展开具体步骤如下在这里插入图片描述第一步先做「语义分层」最核心避免切碎知识点无论什么类型的文档先不考虑长度优先按「文档本身的语义边界」拆分这是 RAG 切分的核心原则 ——语义完整永远比 “凑长度” 更重要。核心逻辑找到文档的「天然语义边界」标题、章节、段落、条款、对话轮次按边界拆分确保每个拆分后的单元能独立表达一个完整的意思比如一个章节、一个步骤、一个条款、一轮对话。关键动作先识别文档结构是否有标题、条款、表格、代码块再按结构拆分不盲目按字符数切割。第二步再做「长度优化」适配模型不超限制语义拆分后会出现两种情况① 单个语义单元过长超过模型输入限制② 单个语义单元过短碎片化检索无意义此时进行针对性优化超长处理对超过模型适配长度的语义单元进行「递归切分」按更小的语义边界比如段落→句子→标点不硬断句子、不破坏逻辑过短处理对过短的语义单元如单个句子、零散短语在「同语义层级」内合并比如同一章节下的短段落合并避免碎片化。具体切割大小,根据不同的文档类型有差距会单独说明第三步最后补「细节保障」落地性、可追溯长度优化后补充 2 个关键细节确保切分结果可落地、可调试重叠设置相邻 Chunk 保留一定重叠5%~10%避免关键信息如步骤、参数、条款落在切口上导致检索漏招元数据补齐给每个 Chunk 添加元数据标题、父标题、文件名、块序号便于后续检索溯源、问题定位。3 不同类型文档的切分形式不同文档的结构、用途不同切分形式和侧重点会不同介绍每类「切分形式、核心要求、操作细节」类型 1技术文档 / 手册 / API 文档有清晰的标题层级# 一级标题、## 二级标题……包含步骤、代码块、参数说明核心是「步骤完整、代码可读、参数不遗漏」。切分思路:语义拆分第一步按「标题层级」切分一级标题#拆分为大模块二级标题##拆分为子模块以此类推确保每个模块对应一个完整的知识点代码块保护识别代码块标记/~~~代码块整体保留不拆分、不切割哪怕代码块过长也单独作为一个 Chunk避免代码不可读长度优化第二步若单个标题下的内容过长如超过 2000 (自行设定)字符按「段落」拆分优先在段落空行处切割不切断步骤、不切断代码若单个标题下的内容过短如 300 字符且与相邻子标题属于同一知识点合并为一个 Chunk细节补充第三步重叠率5%~8%如 2000 字符的 Chunk重叠 100~160 字符元数据保留「父标题上级标题、当前标题、文件名」便于追溯知识点所属模块。核心要求:不切断步骤、不切断代码块、不切断参数说明确保每个 Chunk 能独立体现一个 “操作 / 知识点”。类型 2论文 / 研究报告 / 长叙述文本有摘要、引言、章节、结论以文字叙述为主核心是「论点与论据连贯、逻辑完整」无代码块多为段落式结构。切分思路语义拆分第一步按「章节标题 段落」切分先按章节标题拆分为大模块再按段落拆分为子模块确保每个子模块对应一个完整的论点 / 论据如 “2.1 实验方法” 下的某一段实验描述长度优化第二步若单个段落过长如超过 3000 字符按「句子」拆分优先在句号、感叹号处切割不硬断句子若单个段落过短如 400 字符且与相邻段落属于同一论点合并为一个 Chunk细节补充第三步重叠率8%~12%如 2500 字符的 Chunk重叠 200~300 字符避免论点边界丢失元数据保留「章节标题、段落序号、文件名」标注论点所属章节。核心要求:论点与论据不分离不切断句子确保每个 Chunk 能完整表达一个观点 / 一段论述。类型 3对话记录 / 日志 / 工单特殊场景无标题、无章节多为 “角色 内容” 的轮次结构如客服对话、系统日志核心是「上下文承接、轮次完整」。切分思路语义拆分第一步按「轮次 / 时间戳」切分优先按 “完整轮次” 拆分如 “用户提问→客服回复” 为一个完整轮次日志按 “时间戳分段” 拆分长度优化第二步若轮次过长如多轮对话累计超过 1500 字符按「单轮对话」拆分确保每轮对话独立成块若轮次过短如单句对话 200 字符合并相邻轮次如连续 3 轮短对话合并为一个 Chunk避免碎片化细节补充第三步重叠率15%~20%如 1000 字符的 Chunk重叠 150~200 字符确保上下文承接不丢失元数据保留「角色如用户 / 客服、时间戳、文件名」标注对话 / 日志的时间顺序。核心要求:轮次完整、上下文承接不切断单轮对话确保每个 Chunk 能体现一段完整的交互 / 日志片段。类型 4法律 / 合同 / 制度文档严谨场景有清晰的条款编号如 “第一条、第二条”语言严谨核心是「条款完整、不破坏引用关系」不可拆分单条条款。切分思路:语义拆分第一步按「条款编号」切分单条条款如 “第一条 定义”作为一个基础语义单元若条款过长如包含多个子条款按「子条款」拆分如 “第一条 1.1 定义”长度优化第二步若单条条款过长如超过 2000 字符按「条款内的分句」拆分优先在分号、句号处切割不破坏条款逻辑不合并任何条款哪怕条款过短避免条款边界模糊、引用错误细节补充第三步重叠率5%~10%如 1500 字符的 Chunk重叠 75~150 字符元数据保留「条款编号、父条款若有、文件名」标注条款所属章节。核心要求:不拆分单条条款、不破坏引用关系确保每个 Chunk 是完整的一条 / 一段条款可直接用于法律检索、条款引用。4 补充切分的核心原则语义优先原则无论长度如何先保证语义完整不硬断句子、代码、条款这是 RAG 切分的底线模型适配原则Chunk 大小不超过 Embedding 模型输入限制总和不超过大模型上下文窗口的 70%不碎片化原则单个 Chunk 不小于 300 字符纯中文(条款除外)避免零散碎片导致检索噪声边界保护原则代码块、表格、条款整体保留不拆分、不切割。5 代码实现步骤本节点负责将 Markdown 文本切分为适合向量检索的语义块Chunk采用「先按标题语义切分 → 再按长度精细化切割」的稳定策略。获取与清洗内容 (Step 1)从state中提取 Markdown 内容与文件标题统一换行符格式\r\n/\r→\n保证跨平台兼容。按标题语义初切 (Step 2)基于 Markdown 标题语法#~######进行语义级切分自动跳过代码块内的标题匹配避免误切注释保证每个块语义完整。无标题文档兜底处理 (Step 2 内置)若文档无任何标题自动生成默认标题无主题确保内容不丢失、流程不中断。超长块递归精细化切割 (Step 3)使用RecursiveCharacterTextSplitter对超过指定长度的语义块进行二次切割按「段落 → 换行 → 句子 → 空格」优先级切割不产生碎片、不硬断句子、无需手动合并。构建标准 Chunk 结构为每个切片补充完整元数据title、content、file_title、parent_title、part序号保证可检索、可溯源。本地备份与状态更新 (Step 4)将切分结果备份到本地chunks.json文件同时将最终 chunks 存入state供后续向量入库使用。5.1 导入与配置引入必要的正则表达式、JSON 处理库并定义切分相关的阈值常量。importjsonimportosimportrefrompathlibimportPathfromtypingimportTuple,List,Dictfromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromapp.core.loggerimportlogger,node_log,step_logfromapp.import_process.agent.stateimportImportGraphStatefromapp.utils.task_utilsimportadd_running_task,add_done_task# 全局配置可根据模型调整# 单个文本块最大长度控制不超过模型上下文CHUNK_SIZE200# 小值方便测试切割# 块之间重叠长度保证语义不丢失CHUNK_OVERLAP205.2 主流程定义定义 LangGraph 的节点入口函数串联所有步骤。 1. **获取与清洗内容 (Step 1)** 从 state 中提取 Markdown 内容与文件标题统一换行符格式\r\n / \r → \n保证跨平台兼容。 2. **按标题语义初切 (Step 2)** 基于 Markdown 标题语法# ~ ######进行**语义级切分**自动跳过代码块内的标题匹配避免误切注释保证每个块语义完整。 3. **无标题文档兜底处理 (Step 2 内置)** 若文档无任何标题自动生成默认标题 无主题确保内容不丢失、流程不中断。 4. **超长块递归精细化切割 (Step 3)** 使用 RecursiveCharacterTextSplitter 对**超过指定长度**的语义块进行二次切割按「段落 → 换行 → 句子 → 空格」优先级切割**不产生碎片、不硬断句子、无需手动合并**。 5. **构建标准 Chunk 结构** 为每个切片补充完整元数据title、content、file_title、parent_title、part 序号保证可检索、可溯源。 6. **本地备份与状态更新 (Step 4)** 将切分结果备份到本地 chunks.json 文件同时将最终 chunks 存入 state供后续向量入库使用。 node_log(node_document_split)defnode_document_split(state:ImportGraphState)-ImportGraphState: 节点: 文档切分 (node_document_split) 为什么叫这个名字: 将长文档切分成小的 Chunks (切片) 以便检索。 # 1. 进行任务和日志处理add_running_task(state[task_id],node_document_split)# 2. 进行state中数据清晰(md_content / file_title (做标题兜底))md_content,file_titlestep_1_get_content(state)# 3. 按标题语义初切# [{content:标题的内容,title标题,file_title文件名},{},{}]sections,title_count,lines_countstep_2_split_by_title(md_content,file_title)# 4. 进行语义内递归切割# [{content:标题的内容,title标题,file_title文件名,parent_title,part},{},{}]final_chunksstep_3_refine_chunks(sections)# 5. 数据备份和修改state chunksstate[chunks]final_chunks step_4_backup_chunks(final_chunks,state)add_done_task(state[task_id],node_document_split)returnstate5.3 步骤 1: 获取输入 (Step 1: Get Inputs)从 State 中提取必要的数据并进行基础清洗。step_log(step_1_get_content)defstep_1_get_content(state)-Tuple[str,str]: 数据清晰,处理md_content中不同系统的换成分割! 统一处理 并且获取文件file_tile用于整个内容title兜底 :param state: :return: # 1. 获取md_content内容md_contentstate[md_content]ifnotmd_content:logger.error(f没有输入内容,请检查输入内容是否正确!)raiseRuntimeError(没有输入内容,请检查输入内容是否正确!)# 2.清晰数据统一换行符号 window \r\n linux/mac \n 老mac \r md_contentmd_content.replace(\r\n,\n).replace(\r,\n)file_titlestate.get(file_title,default_file)returnmd_content,file_title5.4 步骤 2: 标题初切 (Step 2: Split by Titles)基于 Markdown 的标题语法#进行第一轮粗略切分。step_log(step_2_split_by_title)defstep_2_split_by_title(md_content,file_title)-List[Dict]: 语义切割,根据标题,进行内容切割! :param md_content: :param file_title: :return: [{content,title,file_title}] # 1. 定义切割正则 / md_content按行切割# \s* 空格 tab * 0 - n# #{1,6} 匹配1-6个 ## \s 1-n #### 标题名# . .任意字符串 1-n [空格]###[空格]标题描述title_patternre.compile(r^\s*#{1,6}\s.)linesmd_content.split(\n)# 准备存储数据容器chunks[]# 最终结果current_title# 当前标题current_lines[]# 当前标题还行内容in_code_blockFalse# 记录是否在代码块中title_count0# 2. 循环处理每行数据forlineinlines:strip_lineline.strip()# 判断是否在代码块中ifstrip_line.startswith()orstrip_line.startswith(~~~):in_code_blocknotin_code_block# 取反即可current_lines.append(line)continue# 不是代码块,判断是不是标题ifnotin_code_blockandtitle_pattern.match(strip_line):# 到了新的标题,将上一次标题进行除虫脲ifcurrent_title:# 存储上一次标题chunks.append({title:current_title,content:\n.join(current_lines),file_title:file_title})# 重置变量(记录当前行)current_titlestrip_line current_lines[strip_line]title_count1else:# 添加行内容 [还是标题内容,非代码行]current_lines.append(strip_line)# 3. 最后一块存储 (最后一次跳出循环没有保存)ifcurrent_title:chunks.append({title:current_title.strip(),content:\n.join(current_lines),file_title:file_title})# 4. 进行无标题处理# --------------------# 兜底文档无标题时# --------------------ifnotchunks:chunks[{title:无主题,content:md_content,file_title:file_title}] md - ## # - ######[空格]标题名称 ## 开篇 内容 \n ![]() ~~~python 代码块 # 注释 # 注释 python 内容 \n ## 中篇 内容 \n xxxxx 内容 \n ## 下篇 内容 \n 内容 \n returnchunks,title_count,len(lines)5.5 步骤 3: 精细化 (Step 3: Refine Chunks)调用辅助函数对过长或过短的 Chunk 进行二次处理。step_log(step_3_refine_chunks)defstep_3_refine_chunks(sections)-List[Dict]: 同一标题下,同一语义,进行二次超长切割!! :param sections: 按标题切割数据 :return: 二次切割数据 spliterRecursiveCharacterTextSplitter(chunk_sizeCHUNK_SIZE,chunk_overlapCHUNK_OVERLAP,# 切割优先级段落 → 换行 → 句子 → 空格separators[\n\n,\n,。,,, ])# 进行切割final_chunks[]forsectioninsections:# 进行二次切割sub_chunksspliter.split_text(section[content])has_multiple_chunkslen(sub_chunks)1# 生成带编号的子块foridx,chunkinenumerate(sub_chunks,start1):current_titlef{section[title]}_{idx}ifhas_multiple_chunkselsesection[title]final_chunks.append({title:current_title,content:chunk.strip(),file_title:section[file_title],parent_title:section[title],part:idx})returnfinal_chunks5.6 步骤 4: 备份与更新 (Step 4: Backup Update)更新 State 并将结果备份到本地。defstep_4_backup_chunks(final_chunks,state): 进行最终数据备份 :param final_chunks: 要备份的数据 :param state: 获取local_dir文件夹 :return: backup_file_pathPath(state[md_path]).parent/backup_chunks.jsonwithopen(backup_file_path,w,encodingutf-8)asf:json.dump(final_chunks,f,ensure_asciiFalse,#中文直接原文存储indent4# json带有缩进 4)logger.debug(f数据备份成功,备份地址:{backup_file_path})5.7 单元测试 (Unit Test)您可以在node_document_split.py文件底部直接运行以下测试代码。包含了模拟数据测试和联合 Step 3 的真实文件测试。if__name____main__: 单元测试联合node_md_img图片处理节点进行集成测试 测试条件1.已配置.envMinIO/大模型环境 2.存在测试MD文件 3.能导入node_md_img 测试流程先运行图片处理→再运行文档切分验证端到端流程 本地测试入口单独运行该文件时执行MD图片处理全流程测试fromapp.utils.path_utilimportPROJECT_ROOTfromapp.import_process.agent.nodes.node_md_imgimportnode_md_img logger.info(f本地测试 - 项目根目录{PROJECT_ROOT})# 测试MD文件路径需手动将测试文件放入对应目录test_md_nameos.path.join(routput\hak180产品安全手册,hak180产品安全手册.md)test_md_pathos.path.join(PROJECT_ROOT,test_md_name)# 校验测试文件是否存在ifnotos.path.exists(test_md_path):logger.error(f本地测试 - 测试文件不存在{test_md_path})logger.info(请检查文件路径或手动将测试MD文件放入项目根目录的output目录下)else:# 构造测试状态对象模拟流程入参test_state{md_path:test_md_path,task_id:test_task_123456,md_content:,file_title:hak180产品安全手册,local_dir:os.path.join(PROJECT_ROOT,output),}logger.info(开始本地测试 - MD图片处理全流程)# 执行核心处理流程result_statenode_md_img(test_state)logger.info(f本地测试完成 - 处理结果状态{result_state})logger.info(\n 开始执行文档切分节点集成测试 )logger.info( 开始运行当前节点node_document_split文档切分)final_statenode_document_split(result_state)final_chunksfinal_state.get(chunks,[])logger.info(f✅ 测试成功最终生成{len(final_chunks)}个有效Chunk{final_chunks})