RAG中PDF文档的处理和存储（二）：Markdown文档的分块处理与入库

上一篇当中介绍了将 PDF 转换为 Markdown 的过程，本文将介绍 RAG 中 PDF 文档处理和存储的第二阶段：Markdown 文档的分块处理与入库。

本文提供的实现过程和代码既可用于对通过上一篇由 PDF 转换得到的 Markdown 的进一步处理，也可用于直接对 Markdown 原件进行分块处理和入库。

一、总体分块策略

1.1 总体分块步骤

第一步：按标题进行结构化拆分

• 函数首先使用MarkdownHeaderTextSplitter工具，根据 Markdown 的标题层级（即#,##,###）将文档拆分成多个大的“章节块”（chunks）。
• 这样做的好处是保留了文档的原始结构，确保每个分块都带有其所属的标题信息（作为元数据），使得内容在语义上更加完整。

第二步：对长章节进行滑动窗口拆分

• 对于上一步生成的“章节块”，如果其内容长度超过了预设的chunk_size，函数会对其进行二次拆分。
• 采用滑动窗口的方法，将长文本按句子进行切分，并组合成符合chunk_size要求的小块。
• 为了避免上下文断裂，相邻的两个分块之间会保留chunk_overlap长度的重叠内容。

1.2 特殊元素的识别与独立处理

表格和算法块

• 这些元素会被从正文中识别并提取出来，作为独立的Document对象进行处理。
• 它们的metadata中的type字段会被分别标记为"table"和"algorithm"，并赋予唯一的id。
• 在原来的正文位置，会留下一个占位符（如{{table_id}}），以表明此处曾有一个表格或算法块。

代码块

• 函数会根据代码块的长度采取不同策略。短代码块直接保留在原始文本中，不进行独立拆分。长代码块与表格和算法块类似，会被提取为独立的Document对象，type标记为"code"，并在原文中留下占位符（如{{Code 0}}）。

图片

• 函数会预先扫描整个文档，提取所有图片的编号、名称和描述信息，并存储起来。
• 在拆分文本时，它会分析每个文本块，识别出其中引用的图片，并将这些图片的元数据记录在该文本块的dependencies元数据字段中。

二、分块部分的代码实现

2.1 markdown_splitter 函数定义、参数处理和有效性检查

def markdown_splitter(document_path: str, rename: str=None, chunk_size=500, chunk_overlap=100, extra_metadata: Dict[str, int|float|str|None]=None): """ Markdown 分块处理：先按标题分块，对于长度超过 "chunk_size" 的章节再进行滑动分块。 Args: document_path: Markdown 文档路径 rename: 文档别名 chunk_size: 分块的最小长度 chunk_overlap: 重叠部分的最小长度 extra_metadata: 文档元数据（例如版本号、修订时间、管理人、权限等） Returns: 处理后的 Document 列表 """ if extra_metadata is None: extra_metadata = dict() elif not extra_metadata.get("authority"): extra_metadata["authority"] = 0 # 默认权限为 0（公开）（生产环境不建议设置较低默认值） if not os.path.isfile(document_path): raise FileNotFoundError(f"文件 {document_path} 不存在") if not document_path.endswith(".md"): raise ValueError(f"文件 {document_path} 不是 markdown 文件")

2.2 读取 Markdown 文档，对图片统一提取和处理

# 接续markdown_splitter函数 # 读取 Markdown 文档 with open(document_path, 'r', encoding="utf-8") as file: markdown_text = file.read() # {"图 x-y": {"name": "图片名称", "description": "图片描述"}, ...} figures: Dict[str, Dict[str, str]] = dict() # 先对文中图片统一提取和处理 splitlines = markdown_text.splitlines() for i, splitline in enumerate(splitlines): if splitline.startswith("!["): figure_description = re.findall(r"!\[([^]]+)]", splitline)[0] j = i+1 while j<len(splitlines) and not splitlines[j]: j +=1 if j>=len(splitlines): continue # 图片位于文档末尾，无法提取编号 if splitlines[j].startswith(("*图", "*Fig", "*FIG")): figure_id, figure_name = re.findall(r"\*([^*]+)\*[\s|:：]*(.*)", splitlines[j])[0] elif re.match(r"(?:图\s*|figure |fig.\s*)[0-9A-Z.\-]+", splitlines[j], re.IGNORECASE): figure_id, figure_name = re.findall(r"((?:图\s*|figure |fig.\s*)[0-9A-Z.\-]+)[\s|:：]*(.*)", splitlines[j], re.IGNORECASE)[0] else: continue # 无编号图片，无需提取和处理 if not figure_id: continue # 无编号图片，无需提取和处理 if figure_id[-1] == '.': figure_id = figure_id[:-1] figures[figure_id] = {"name": figure_name, "description": figure_description}

2.3 第一轮分块：按标题分块（MarkdownHeaderTextSplitter）

# 接续markdown_splitter函数 # 第一步：按标题分块 headers_to_split_on = [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ("####", "article_id")] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) chunks0 = splitter.split_text(markdown_text)

2.4 第二轮分块：较长章节滑动分块

总体和每个大块分块前的准备

# 接续markdown_splitter函数 # 第二步：较长章节滑动分块 chunks1: List[Document] = [] # 处理后的 Document 列表 table_list: Dict[str, Document] = dict() # 表格列表 algorithm_list: Dict[str, Document] = dict() # 算法块列表 code_list: List[Document] = [] # 代码块列表 for chunk in tqdm(chunks0): page_content = "" page_content1 = "" # 滑动分块重叠部分 # 相关图片、表格、算法、代码标记（存入 metadata 的 dependencies字段） related_figures: Dict[str, Dict[str, str]] = dict() related_tables: List[str] = [] related_algorithms: List[str] = [] related_codes: List[int] = [] # 大块（章节）元数据 metadata = chunk.metadata # 文档级元数据 for k, v in extra_metadata.items(): metadata[k] = v if metadata.get("Header 1", "").lower() in ("目录", "目 录", "table of contents"): continue metadata["document_name"] = document_name

逐个段落区分类型处理：

• 表格：检测前面是否有表名，将表格单独划分为Document并存入表格列表，在正文相应位置插入占位符
• 算法块：检测编号，将算法块单独划分为Document并存入算法块列表，在正文在相应位置插入占位符
• 代码块：短代码直接插入正文，长代码则单独划分为Document存入代码列表
• 普通文本：提取文本中的图片、表格、算法编号，列入Metadata中的依赖项字段，然后使用spacy将段落划分为若干句子

随后将文本块逐句加入到 page_content 缓冲区当中，当 page_content 的长度大于等于 chunk_size - chunk_overlap 时，进入块间重叠部分 page_content1，文本块存入重叠部分；当重叠部分page_content1 的长度大于等于 chunk_overlap 时，将两个缓冲区内容合并，并添加元数据和依赖项信息，作为文档的一个 Document 分块对象。

# 接续 markdown_splitter -> for chunk in tqdm(chunks0): paragraphs = chunk.page_content.split('\n') i = 0 while i<len(paragraphs): paragraph = paragraphs[i].strip() #logger.info(paragraph) if not paragraph or (paragraph[0] == "*" and paragraph[:2] != "**"): i += 1 continue # 跳过图名、表名段落 # 表格 elif paragraph[0] == "|": # 检测表名 table_name = paragraphs[i-1] if i>0 else "" i+=1 while i<len(paragraphs) and paragraphs[i][0] == "|": paragraph += "\n" + paragraphs[i] i += 1 i -= 1 if table_name[:2] in ("*表", "*Tab", "*TAB"): table_id = re.findall(r"\*([^*]+)\*", table_name)[0] if table_id[-1] == '.': table_id = table_id[:-1] table_list[table_id] = Document(table_name + "\n" + paragraph, metadata=metadata) else: match = re.match(r"(?:表\s*|table |tab.\s*)[0-9A-Z.\-]+", table_name, re.IGNORECASE) if match: table_id = match.group() if table_id[-1] == '.': table_id = table_id[:-1] table_list[table_id] = Document(table_name + "\n" + paragraph, metadata=metadata) else: # 无名表格 table_id = f"Unnamed table {len(table_list)}" table_list[table_id] = Document(paragraph, metadata=metadata) # 插入表格占位符 sentences = [f"{{{{{table_id}}}}}"] # 算法块 elif paragraph.startswith("*****Algorithm Block*****"): while not paragraph.strip().endswith("*****End Algorithm*****"): i+=1 paragraph += "\n" + paragraphs[i] paragraph = paragraph[25:-23].strip() # 检测算法编号 algorithm_id = re.match(r"算法\s?[0-9A-Z.\-]+|algorithm [0-9A-Z.\-]+", paragraph, re.IGNORECASE).group() if algorithm_id[-1] == '.': algorithm_id = algorithm_id[:-1] algorithm_list[algorithm_id] = Document(paragraph, metadata=metadata) # 插入算法占位符 sentences = [f"{{{{{algorithm_id}}}}}"] # 代码块 elif paragraph.startswith("```"): while len(paragraph.strip())<=6 or not paragraph.strip().endswith("```"): i+=1 paragraph += "\n" + paragraphs[i] # 短代码直接插入正文，长代码则存入代码列表 if len(paragraph) > 100: related_codes.append(len(code_list)) sentences = [f"{{{{Code {len(code_list)}}}}}"] code_list.append(Document(paragraph, metadata=metadata)) else: sentences = [paragraph] # 普通文本（可能包含前述未涵盖的其他特殊类型） else: # 处理图片、表格、算法标号 for figure_id in re.findall(r"Figure [0-9A-Z.\-]+|图\s*[0-9A-Z.\-]+", paragraph): if figure_id[-1] == ".": figure_id = figure_id[:-1] if figures.get(figure_id): related_figures[figure_id] = figures[figure_id] for table_id in re.findall(r"Table [0-9A-Z.\-]+|表\s*[0-9A-Z.\-]+", paragraph): if table_id[-1] == ".": table_id = table_id[:-1] related_tables.append(table_id) for algorithm_id in re.findall(r"Algorithm [0-9A-Z.\-]+|算法\s*[0-9A-Z.\-]+", paragraph): if algorithm_id[-1] == ".": algorithm_id = algorithm_id[:-1] related_algorithms.append(algorithm_id) doc1 = nlp(paragraph) sentences = [sent.text for sent in doc1.sents] # 分句 # 逐句加入文本块 for sentence in sentences: if sentence[-1] in ".?!:'\"”)]": # 给英文标点后面加空格 sentence += " " if len(page_content) < chunk_size - chunk_overlap: page_content += sentence else: if len(page_content1) < chunk_overlap: page_content1 += sentence else: metadata["type"] = "text" # 添加依赖信息 dependencies: Dict[str, Any] = dict() dependencies["related_figures"] = related_figures dependencies["related_tables"] = related_tables dependencies["related_algorithms"] = related_algorithms dependencies["related_codes"] = related_codes metadata["dependencies"] = json.dumps(dependencies, ensure_ascii=False) chunks1.append(Document(page_content + page_content1, metadata=metadata)) page_content = page_content1 + sentence page_content1 = "" related_figures = dict() related_tables = [] related_algorithms = [] related_codes = [] if len(page_content) < chunk_size - chunk_overlap: page_content += "\n" else: if len(page_content1) < chunk_overlap: page_content1 += "\n" i += 1

全部大块处理完毕后，处理缓冲区中残余内容

# 接续 markdown_splitter -> for chunk in tqdm(chunks0): # 如果大块已经读取完毕，page_content仍然有剩余未保存的内容，则保存 if page_content: metadata["type"] = "text" # 添加依赖信息 dependencies: Dict[str, Any] = dict() dependencies["related_figures"] = related_figures dependencies["related_tables"] = related_tables dependencies["related_algorithms"] = related_algorithms dependencies["related_codes"] = related_codes metadata["dependencies"] = json.dumps(dependencies, ensure_ascii=False) chunks1.append(Document(page_content + page_content1, metadata=metadata))

2.5 表格、算法、代码块统一处理，保存为 Document 对象

# 接续markdown_splitter函数 # 表格、算法、代码块统一处理 logger.info("正在处理表格...") for table_id, table_document in table_list.items(): table_document.metadata["type"] = "table" table_document.metadata["id"] = table_id chunks1.append(table_document) logger.info("正在处理算法块...") for algorithm_id, algorithm_document in algorithm_list.items(): algorithm_document.metadata["type"] = "algorithm" algorithm_document.metadata["id"] = algorithm_id chunks1.append(algorithm_document) logger.info("正在处理代码块...") for i, code_document in enumerate(code_list): code_document.metadata["type"] = "code" code_document.metadata["id"] = f"Code {i}" chunks1.append(code_document) logger.info(f"文档切片完毕，共 {len(chunks1)} 个大块") return chunks1

三、分块部分的本地测试

if __name__ == '__main__': parser = argparse.ArgumentParser(description="PDF文档分块处理工具") # 必填参数 parser.add_argument("document_path", type=str, help="PDF文档路径") # 可选参数 parser.add_argument("--rename", type=str, default=None, help="文档别名") parser.add_argument("--chunk-size", type=int, default=500, help="分块的最小长度 (默认: 500)") parser.add_argument("--chunk-overlap", type=int, default=100, help="重叠部分的最小长度 (默认: 100)") parser.add_argument("--title-format", type=str, nargs="+", default=None, help='标题格式列表，例如: "第一章" "一、" "1. "') parser.add_argument("--extra-metadata", type=str, default=None, help='额外元数据，JSON格式字符串，例如: \'{"version": "1.0", "authority": 1}\'') parser.add_argument("--output", type=str, default="temp/output.md", help="输出文件路径 (默认: temp/output.md)") args = parser.parse_args() # 解析 extra_metadata JSON 字符串为字典 extra_metadata = None if args.extra_metadata: try: extra_metadata = json.loads(args.extra_metadata) except json.JSONDecodeError as e: raise ValueError(f"extra-metadata 必须是有效的JSON格式: {e}") # 调用 pdf_splitter documents = pdf_splitter( document_path=args.document_path, rename=args.rename, chunk_size=args.chunk_size, chunk_overlap=args.chunk_overlap, title_format=args.title_format, extra_metadata=extra_metadata ) # 输出结果 import os os.makedirs(os.path.dirname(args.output), exist_ok=True) with open(args.output, "w", encoding="utf-8") as output: for i, document in enumerate(documents): output.write(f"\n\n# Document {i + 1}:") output.write("\n\n**Page Content:**\n{}".format( document.page_content.replace('\n', '\n\n').replace('|\n\n|', '|\n|') )) output.write(f"\n\n**Metadata:**\n{document.metadata}") print(f"处理完成，共生成 {len(documents)} 个文档块，输出至: {args.output}")

使用方法示例：

# 基本用法 python data_splitter.py path/to/document.pdf # 完整参数 python data_splitter.py path/to/document.pdf \ --rename "我的文档" \ --chunk-size 800 \ --chunk-overlap 150 \ --title-format "第一章" "一、" "1." \ --extra-metadata '{"version": "2.0", "authority": 1}' \ --output result/output.md

四、文档入库

4.1 入库策略

1. 预处理：基于内容指纹的去重

在向数据库写入任何数据之前，函数会首先对传入的文档块列表进行一次内部去重。

• 去重依据：它并非简单地比较整个文档对象，而是提取每个文档块的document_name（文档名）和page_content（文本内容）组合成一个唯一的“内容指纹”（content_fingerprint）。
• 处理逻辑：函数会遍历所有文档块，如果一个“内容指纹”是首次出现，则保留该文档块；如果该指纹已存在，则视为重复内容并跳过。
• 目的：确保进入后续存储流程的文档块列表本身没有重复项，避免了因上游处理不当导致的数据冗余。

2. 索引：生成基于内容的唯一ID

为了实现精准的数据同步，函数为每个文档块生成了一个稳定且唯一的ID。

• ID生成方式：使用SHA256哈希算法。
• 哈希内容：将document_name和page_content拼接后进行哈希计算。
• 策略优势：这种策略保证了只要文档名和内容不变，生成的ID就永远不变。这是实现后续“增量更新”和“精准删除”的基石。即使文档块在列表中的顺序发生变化，其ID依然稳定。

3. 同步：智能的增、删、改策略（核心部分）

检测目标 Chroma 数据库是否存在，并据此采取不同的同步策略。

如果目标目录中没有检测到现有的数据库，函数会直接调用Chroma.from_documents方法，将所有传入的文档块一次性全部存入，创建一个全新的向量数据库。

当检测到目标目录中已有数据库时，函数会执行一套完整的同步逻辑，而不是盲目地覆盖或追加。

1. 精准删除：

   • 识别目标：首先，确定本次操作涉及哪些文档（通过incoming_doc_names集合）。
   • 执行逻辑：遍历数据库中所有已存在的文档块，如果某个文档块满足以下两个条件，就会被标记为“待删除”：
     • 它属于本次正在更新的文档（即其document_name在incoming_doc_names中）。
     • 它的内容ID（基于旧内容生成）不在本次新生成的ID列表中。
   • 目的：这能精准地清理掉原文档中已被删除或修改的旧段落，保证数据库内容与最新文档保持一致。

2. 精准新增：

   • 识别目标：在执行删除操作后，检查数据库中当前剩余的所有ID。
   • 执行逻辑：将本次新生成的ID列表与数据库中现有的ID列表进行比对。只有那些在数据库中完全不存在的新ID，才会被当作“待新增”的内容。
   • 目的：这确保了只有真正新增的段落或被修改过的段落（其内容变化导致ID也变化）才会被向量化并写入数据库，避免了对未变动内容的重复计算和存储，极大地提升了更新效率。

4.2 代码实现（my_chroma.py）

import hashlib import os from typing import List, Set, Dict from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.embeddings import Embeddings from summary import doc_summary, markdown_summary def textblock_chroma(documents: List[Document], embedding_model: Embeddings, save_dir: str | None="chroma") -> Chroma | None: """ 将分块后的文档存入 Chroma 数据库（支持按 document_name 增量更新/覆盖） Args: documents: 分块后的文档 embedding_model: 文本嵌入模型，例如 OpenAIEmbeddings、DashScopeEmbeddings、OllamaEmbeddings、HuggingFaceEmbeddings等 save_dir: 存储目录 Returns: 向量数据 """ if not documents: print("传入的文档列表为空，跳过处理。") return None # --- 修复：基于 ID 生成规则进行去重 (精准去重) --- # 核心思想：去重的判断标准必须和 generate_ids 的逻辑完全一致 seen_content_keys = set() unique_docs = [] for doc in documents: # 1. 提取用于生成 ID 的核心要素 (与 generate_ids 函数内部逻辑保持一致) content = doc.page_content doc_name = doc.metadata.get("document_name", "unknown") # 2. 创建一个“内容指纹”，仅包含影响 ID 生成的因素 content_fingerprint = (doc_name, content) # 3. 如果这个“指纹”没有出现过，则保留该文档 if content_fingerprint not in seen_content_keys: seen_content_keys.add(content_fingerprint) unique_docs.append(doc) else: # 可选：打印被忽略的重复项信息，方便调试 print(f"发现重复内容块 (文档: {doc_name}, 内容: {content[:15]}...), 已跳过。") print(f"原始文档数量: {len(documents)}，去重后文档数量: {len(unique_docs)}") documents = unique_docs # --- 去重结束 --- if save_dir == "documents_summary": print("当前使用的向量数据库名称 'documents_summary' 为文档摘要专用，已自动切换为默认名称 'chroma'。") save_dir = "chroma" if save_dir and not os.path.isdir(save_dir): os.mkdir(save_dir) def generate_ids(documents): """为文档列表生成基于内容的唯一哈希ID""" ids = [] for doc in documents: # 使用文档内容生成哈希，确保相同内容ID相同 content = doc.page_content.encode('utf-8') # 建议将 document_name 也纳入哈希计算，防止不同文档出现完全相同的段落导致ID冲突 doc_name = doc.metadata.get("document_name", "unknown").encode('utf-8') unique_content = doc_name + b"|||" + content doc_id = hashlib.sha256(unique_content).hexdigest() ids.append(doc_id) return ids print("正在构建内容索引...") # 2. 为传入的 documents 生成 ID new_ids = generate_ids(documents) new_contents = [doc.page_content for doc in documents] new_metadatas = [doc.metadata for doc in documents] # 提取本次涉及的所有 document_name（去重） incoming_doc_names: Set[str] = set( doc.metadata.get("document_name") for doc in documents if doc.metadata.get("document_name") ) if save_dir and os.path.isfile(os.path.join(save_dir, "chroma.sqlite3")): # --- 情况 A: 数据库已存在 --- print(f"检测到现有数据库，正在按 document_name 同步数据...") vectorstore = Chroma(persist_directory=save_dir, embedding_function=embedding_model, collection_name="document_fulltext") # 2. 获取数据库中现有的所有 ID 和 metadata existing_items = vectorstore.get(include=['metadatas']) existing_ids_all = existing_items['ids'] existing_metadatas = existing_items['metadatas'] # 3. 【精准删除逻辑】找出需要淘汰的旧段落 # 规则：属于本次更新的文档，且其 ID 不在本次的新数据列表中 ids_to_delete = [] for i, meta in enumerate(existing_metadatas): doc_name = meta.get("document_name") if meta else None # 只有当该向量属于本次更新的文档，且它的指纹(new_ids)没有出现在新版本中时，才视为被删除的旧章节 if doc_name in incoming_doc_names and existing_ids_all[i] not in new_ids: ids_to_delete.append(existing_ids_all[i]) if ids_to_delete: print(f"发现 {len(ids_to_delete)} 个被淘汰的旧段落（如被删减的章节），正在清理...") vectorstore.delete(ids=ids_to_delete) # 4. 【精准新增逻辑】只向量化并写入真正的新增/修改段落 # 获取数据库里目前已有的所有 ID（包含刚才没被删的同名段落） current_db_ids = set(vectorstore.get(include=[])['ids']) # 只有本次生成的新 ID 中，那些数据库里完全不存在的，才需要执行 add ids_to_add = [id for id in new_ids if id not in current_db_ids] if ids_to_add: # 根据 ids_to_add 过滤出对应的文本和元数据 id_to_content_map = dict(zip(new_ids, new_contents)) id_to_meta_map = dict(zip(new_ids, new_metadatas)) add_contents = [id_to_content_map[i] for i in ids_to_add] add_metadatas = [id_to_meta_map[i] for i in ids_to_add] print(f"正在向量化并添加 {len(ids_to_add)} 个新段落（如新增的章节或改动的内容）...") vectorstore.add_texts( texts=add_contents, metadatas=add_metadatas, ids=ids_to_add ) else: print("所有段落均为最新版本，无需新增或修改。") else: # --- 情况 B: 数据库不存在，直接创建 --- print("未检测到数据库，正在创建新数据库...") vectorstore = Chroma.from_documents( documents=documents, embedding=embedding_model, persist_directory=save_dir, # 注意：这里直接传目录即可，Chroma会自动处理文件名 collection_name="document_fulltext" ) if save_dir else Chroma.from_documents( documents=documents, embedding=embedding_model, collection_name="document_fulltext" ) return vectorstore

五、可选功能：提取摘要和目录，存入文档级向量库

当向量库中文档较多时，如果直接针对所有文档进行检索，大量与用户查询无关的文档将明显降低检索的效率。本节考虑了先筛选文档缩小检索范围，再在筛选出的文档中检索分块的方式，在进行文档分块处理之前，先提取文档的摘要和附录，存入文档级向量数据库。

在markdown_splitter函数的“nlp = spacy.load("zh_core_web_sm")”前面添加：

# 先提取摘要和目录，存入文档级向量库 # 向量模型可根据需要选用OpenAIEmbeddings、DashScopeEmbeddings、OllamaEmbeddings、HuggingFaceEmbeddings等 document_chroma(document_path, embedding_model = OllamaEmbeddings(model="qwen3-embedding:0.6b", base_url="http://localhost:11434"), rename=rename, extra_metadata=extra_metadata)

my_chroma.py中对document_chroma函数的实现（这里还考虑了docx等其他格式文档的处理，不在本文讨论范围内）：

def document_chroma(document_path: str, rename: str = None, extra_metadata: Dict[str, int | float | str | None] = None, save_dir="documents_summary"): """摘要数据存储""" if not os.path.isfile(document_path): print(f"文件 {document_path} 不存在！") return if document_path.endswith("pdf"): print(f"文件 {document_path} 是 PDF 文件，为了准确提取摘要内容，请先转换为 Markdown 格式。") return if document_path.endswith(("docx", "doc")): summary_document = doc_summary(document_path, rename=rename, extra_metadata=extra_metadata) # 这一段是本文主要涉及的 markdown 文档 elif document_path.endswith("md"): summary_document = markdown_summary(document_path, rename=rename, extra_metadata=extra_metadata) else: print(f"暂不支持的文件格式 {document_path}") return vector_store = Chroma( collection_name="document_summaries", # 集合名称，可根据需求自定义 embedding_function=embedding_model, persist_directory=save_dir ) # 2. 将新提取的摘要文档添加到现有的向量数据库中 # add_documents 方法会在原有数据的基础上进行追加，不会删除原有数据 vector_store.add_documents([summary_document]) print(f"成功将文档 {document_path} 的摘要和目录存入向量数据库。")

其中 markdown_summary 是为 Markdown 文档创建摘要+目录数据的函数，在 summary.py 中实现：

# summary.py import os from typing import Dict from langchain_core.documents import Document from langchain_text_splitters import MarkdownHeaderTextSplitter def markdown_summary(document_path: str, rename: str=None, extra_metadata: Dict[str, int|float|str|None]=None): """为 Markdown 文档创建摘要数据""" metadata = extra_metadata or dict() document_name = rename or os.path.basename(document_path) metadata["document_name"] = document_name with open(document_path, 'r', encoding="utf-8") as file: markdown_text = file.read() headers_to_split_on = [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3")] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) chunks = splitter.split_text(markdown_text) extracted_toc = [] # 存储目录结构 title_buffer = ["", "", ""] # 用于检测是否有新的标题 summary = "" # 摘要内容 alterative_summary = "" # 备选摘要内容（前部一定长度内容） for chunk in chunks: # 目录的生成 h1 = chunk.metadata.get("Header 1", "") if h1 and h1!= title_buffer[0]: title_buffer[0] = h1 extracted_toc.append("- " + h1) h2 = chunk.metadata.get("Header 2", "") if h2 and h2!= title_buffer[1]: title_buffer[1] = h2 extracted_toc.append(" - " + h2) h3 = chunk.metadata.get("Header 3", "") if h3 and h3!= title_buffer[2]: title_buffer[2] = h3 extracted_toc.append(" - " + h3) # 摘要的生成：优先提取“前言”或“摘要”部分 if h1.lower() in ("前言", "前 言", "forward", "摘要", "摘 要", "abstract")\ or h2.lower() in ("前言", "前 言", "forward", "摘要", "摘 要", "abstract")\ or h3.lower() in ("前言", "前 言", "forward", "摘要", "摘 要", "abstract"): summary += chunk.page_content + '\n' # 未遇到过“前言”或“摘要”部分，则先提取正文 elif not summary: alterative_summary += chunk.page_content + '\n' # 文档不含“前言”或“摘要”部分，则从正文中提取前 5000 个字符 if not summary: summary = alterative_summary[:5000] page_content = "# 摘要\n{}\n# 目录\n{}".format(summary, "\n".join(extracted_toc)) return Document(page_content=page_content, metadata=metadata)

本文源代码获取方式

git clone https://gitee.com/dsy0221/my-rag.git