当前位置：首页 > news >正文

构建自动化RAG数据提取流水线：从文档解析到向量化存储的工程实践

news 2026/7/12 11:54:08

1. 项目概述：为什么我们需要一个自动化的RAG数据提取工具？

如果你正在构建一个RAG（检索增强生成）系统，无论是用于内部知识库问答、智能客服，还是文档分析，你肯定遇到过这个最头疼的环节：数据准备。RAG的核心魅力在于让大语言模型（LLM）能够“引用”你提供的专属知识，但前提是，你得把这些知识——海量的PDF、Word、网页、数据库记录——变成LLM能理解和高效检索的格式。这个过程，我们称之为“数据提取与向量化流水线”。

市面上有很多RAG框架，比如LangChain、LlamaIndex，它们提供了强大的检索和生成链条。但当你真正把一堆杂乱无章的原始文档丢进去时，往往会发现，框架本身并不负责帮你把文档“收拾”干净。一个高度结构化、功能完整的Python程序，其核心定位就是填补这个空白：自动化地、可靠地、高质量地完成从原始数据到RAG就绪知识库的转换。它不是一个简单的脚本，而是一个具备工程化思维的解决方案，需要考虑格式解析、文本清洗、智能分块、元数据提取、向量化调度、错误处理与监控等全链路问题。简单来说，它要把“脏活累活”自动化，让你能专注于RAG应用的上层业务逻辑。

2. 核心需求解析：一个优秀的数据提取工具应该解决哪些问题？

在动手设计之前，我们必须明确这个工具要应对的具体挑战。根据我的经验，一个粗糙的数据处理流程会直接导致RAG系统表现不佳，出现“答非所问”、“幻觉严重”、“检索不准”等问题。以下是几个核心痛点，也是我们工具必须攻克的目标：

2.1 格式兼容性与解析鲁棒性

用户的数据不可能只有一种格式。一个合格的工具必须能处理：

文档类：PDF（扫描件/可复制文本）、Word（.docx）、PowerPoint（.pptx）、Excel（.xlsx）、纯文本（.txt）、Markdown（.md）。
网页类：HTML页面，可能需要处理JavaScript渲染后的内容。
结构化数据：从数据库（如MySQL、PostgreSQL）或API接口中提取的文本字段。

难点在于，每种格式的解析库（如PyPDF2、pdfplumber、python-docx）行为不同，错误处理机制各异。PDF中的扫描件需要OCR（光学字符识别），表格和图片中的文字需要特殊处理。工具必须封装这些复杂性，提供统一的接口。

2.2 文本清洗与标准化

从原始文档中提取的文本通常包含大量“噪音”：

无意义的页眉、页脚、页码、水印。
复杂的排版符号、乱码、不可见字符。
对于中文，全角/半角标点、多余的空格和换行。
从网页抓取的广告、导航栏等无关内容。

清洗的目标是保留核心语义内容，去除干扰信息，并将文本标准化为干净、连贯的段落。这一步的质量直接影响到后续“分块”的效果和向量嵌入的质量。

2.3 智能分块策略

这是RAG数据准备的灵魂。简单地将文档按固定字符数（如512个token）切割，会严重破坏语义的完整性。想象一下，一个问题的答案刚好被切在两块之间，检索系统就永远找不到完整信息。我们的工具需要实现更智能的分块：

基于语义的分块：利用句子边界检测（如sentence-transformers），确保每个块是完整的语义单元。
递归分块：先按大标题分，再按小标题分，最后按段落分，形成层次结构。
重叠分块：在块与块之间设置一定的重叠区域（如50个字符），防止关键信息被切断。
特殊内容处理：对于代码块、表格、列表，应尽量保持其完整性，单独成块或做特殊标记。

2.4 元数据提取与关联

为了让检索更精准，我们需要为每个文本块（chunk）附加丰富的元数据（metadata）。这些元数据未来会跟向量一起存入数据库，用于过滤和增强检索。

基础元数据：来源文件路径、文件名、文件类型、最后修改时间。
内容元数据：所属章节标题、页码、在文档中的顺序索引。
自定义元数据：根据业务需求提取，如文档类别、作者、部门、项目编号等。

关于“RAG系统嵌入chunk时需要把标题信息嵌入吗？”这个热门问题：答案是强烈建议嵌入。标题是理解段落上下文的关键。最佳实践是将标题信息作为元数据关联到chunk，并且在生成该chunk的文本向量时，可以考虑将标题文本与chunk内容拼接后进行嵌入，或者至少将标题作为一个独立的、高权重的过滤字段。这能极大提升检索的相关性。

2.5 流程自动化与可观测性

处理成千上万的文档，手动操作是不可想象的。工具需要：

增量更新：能够识别哪些文件是新增加的或修改过的，只处理变化的部分，避免全量重复处理。
任务队列与并行处理：利用multiprocessing或celery等工具并行处理多个文件，提升吞吐量。
日志与监控：详细记录每个文件的处理状态（成功、失败、跳过）、消耗时间、产生的chunk数量。对于失败的文件，要记录具体的错误原因，方便排查。
配置化管理：所有参数（如分块大小、重叠长度、清洗规则、向量模型）都应通过配置文件（如YAML）管理，便于不同场景的切换。

3. 系统架构设计与技术选型

基于以上需求，我们来设计这个自动化工具的整体架构。我将它分为五个核心模块，形成一个清晰的数据流水线。

原始数据源 --> [输入与路由模块] --> [解析与提取模块] --> [清洗与分块模块] --> [向量化与存储模块] --> [调度与监控模块]

3.1 输入与路由模块

这个模块负责发现和分发待处理的数据。它监听一个或多个“数据源目录”，或者从消息队列中接收任务。

技术选型：使用watchdog库监控文件夹变化，或使用Redis/RabbitMQ作为任务队列。对于简单的本地批处理，一个递归遍历目录的函数就足够了。
关键设计：为每个文件生成一个唯一任务ID，并附带文件路径和类型。根据文件后缀名（如.pdf）将任务路由到对应的解析器。

3.2 解析与提取模块

这是格式处理的核心，我们为每种文件类型实现一个解析器类（Parser Class），它们继承自一个统一的基类。

PDF解析：优先使用pdfplumber，因为它对表格和布局的分析能力更强。对于扫描件，集成pytesseract（Tesseract OCR的Python封装）或easyocr。
Word/PPT/Excel解析：使用python-docx、python-pptx、openpyxl或pandas。注意提取文本的同时，尽可能保留样式信息（如标题级别）。
网页解析：使用BeautifulSoup处理静态HTML。对于动态页面，集成playwright或selenium进行自动化渲染后再提取。playwright是当前比较流行的选择，因为它支持多浏览器且API现代。
数据库/API解析：根据数据源定制，使用对应的驱动（如psycopg2、requests）并编写SQL或调用API。

实操心得：一定要为每个解析器实现健壮的错误处理。比如，一个PDF文件可能加密了，或者损坏了。解析器应该捕获所有异常，记录错误日志，并将任务标记为失败，而不是让整个程序崩溃。

3.3 清洗与分块模块

接收解析后的原始文本，进行清洗和智能分割。

清洗链：设计一系列清洗函数，如去除多余空白、规范化换行符、移除特定正则表达式匹配的噪声（如“第X页”）。可以引入html2text来处理HTML转纯文本的残留问题。
分块器：这是核心算法所在。我们可以实现多种分块策略，并通过配置选择。
- 固定大小分块器：最简单，使用LangChain的RecursiveCharacterTextSplitter或自定义实现。
- 语义分块器：利用sentence-transformers计算句子嵌入，根据嵌入的余弦相似度进行动态分割。LangChain的SemanticChunker是一个参考。
- 递归分块器：按“\n\n”、“。”、“；”等分隔符递归切割，直到块大小符合要求。这是最常用的平衡策略。
元数据附着：在分块过程中，将当前块的上下文信息（如上一级标题）作为元数据记录下来。

3.4 向量化与存储模块

将文本块转化为向量，并存入向量数据库。

嵌入模型：选择适合你语种和场景的模型。对于中文，BAAI/bge-small-zh-v1.5、moka-ai/m3e-base都是不错的选择。使用sentence-transformers库可以方便地加载和使用这些模型。
向量数据库：根据数据量和运维需求选择。轻量级可选ChromaDB（内存/文件），生产环境可选Qdrant、Weaviate、Milvus或PGVector（基于PostgreSQL）。我们的工具需要封装与向量数据库的交互，实现“插入”、“更新”、“按ID删除”等操作。
批处理：嵌入模型调用通常是流水线的瓶颈。务必使用批处理（batch）方式向模型送入文本，而不是一条一条处理，这能带来数十倍的性能提升。

3.5 调度与监控模块

协调整个流水线，提供可观测性。

任务调度：可以使用APScheduler实现定时扫描，或者由外部系统触发。
状态管理：在数据库中维护一张processing_jobs表，记录每个文件的任务ID、状态（pending, processing, success, failed）、开始时间、结束时间、错误信息、产生的chunk数量等。
日志系统：使用Python标准的logging模块，配置不同的handler（如文件、控制台），并设置合理的日志级别（INFO记录流程，ERROR记录失败）。

4. 核心代码实现与关键细节

下面，我将以几个关键代码片段为例，展示如何实现上述模块的核心功能。请注意，这是一个高度简化的示例，真实项目需要更完善的错误处理和配置管理。

4.1 解析器工厂与基类设计

首先，我们定义一个解析器基类，所有具体解析器都必须实现parse方法。

from abc import ABC, abstractmethod from typing import Dict, Any, Optional import logging logger = logging.getLogger(__name__) class BaseParser(ABC): """解析器基类""" def __init__(self, file_path: str, **kwargs): self.file_path = file_path self.kwargs = kwargs @abstractmethod def parse(self) -> Dict[str, Any]: """ 解析文件，返回结构化数据。 返回格式示例： { "content": "完整的文本内容", "metadata": { "title": "文档标题", "pages": 10, "author": "作者", "sections": [{"title": "章节1", "level": 1}, ...] } } """ pass def safe_parse(self) -> Optional[Dict[str, Any]]: """安全的解析方法，包含异常处理""" try: return self.parse() except Exception as e: logger.error(f"解析文件 {self.file_path} 时发生错误: {e}", exc_info=True) return None class ParserFactory: """解析器工厂，根据文件扩展名返回对应的解析器实例""" _parsers = { '.pdf': 'PDFParser', '.docx': 'DocxParser', '.txt': 'TextParser', '.html': 'HtmlParser', # ... 注册更多解析器 } @staticmethod def get_parser(file_path: str, parser_map: Dict = None) -> Optional[BaseParser]: import os from .parsers import PDFParser, DocxParser, TextParser, HtmlParser # 假设这些类在parsers模块中 ext = os.path.splitext(file_path)[1].lower() map_to_use = parser_map or ParserFactory._parsers parser_class_name = map_to_use.get(ext) if not parser_class_name: logger.warning(f"不支持的文件格式: {ext} for file {file_path}") return None # 这里简化了，实际项目中可以用更动态的导入方式 class_map = { 'PDFParser': PDFParser, 'DocxParser': DocxParser, 'TextParser': TextParser, 'HtmlParser': HtmlParser, } parser_class = class_map.get(parser_class_name) if parser_class: return parser_class(file_path) return None

4.2 智能分块器的实现

我们实现一个结合递归分割和重叠的通用分块器。

from typing import List, Dict, Any import re class RecursiveOverlapChunker: """递归重叠分块器""" def __init__( self, chunk_size: int = 500, chunk_overlap: int = 50, separators: List[str] = None, length_function = len, # 计算长度的函数，可以是len或token计数器 ): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap # 默认按段落、换行、句子、逗号等分割 self.separators = separators or ["\n\n", "\n", "。", "；", "，", " ", ""] self.length_function = length_function def split_text(self, text: str, metadata: Dict[str, Any] = None) -> List[Dict[str, Any]]: """将文本分割成块，并携带元数据""" final_chunks = [] metadata = metadata or {} def _recursive_split(chunk_text: str, current_metadata: Dict) -> List[Dict]: # 如果文本已经足够短，直接返回 if self.length_function(chunk_text) <= self.chunk_size: return [{"text": chunk_text, "metadata": current_metadata.copy()}] # 尝试用不同的分隔符进行分割 split_success = False for sep in self.separators: if sep == "": # 最后一个分隔符是空字符串，表示按字符硬切 splits = [chunk_text[i:i+self.chunk_size] for i in range(0, len(chunk_text), self.chunk_size - self.chunk_overlap)] # 为每个分割块创建结果 result = [] for split in splits: result.append({"text": split, "metadata": current_metadata.copy()}) return result if sep in chunk_text: splits = chunk_text.split(sep) # 确保分割后还有内容 if len(splits) > 1: # 重新加上分隔符（除了最后一个空的分隔符情况） combined_splits = [] for i, s in enumerate(splits): if i < len(splits) - 1: combined_splits.append(s + sep) else: combined_splits.append(s) # 递归处理每个部分 result = [] current_split = "" for combined in combined_splits: if self.length_function(current_split + combined) <= self.chunk_size: current_split += combined else: if current_split: result.extend(_recursive_split(current_split, current_metadata)) current_split = combined if current_split: result.extend(_recursive_split(current_split, current_metadata)) return result split_success = True break # 如果没有找到合适的分隔符，按字符硬切（理论上不会走到这里，因为空字符串分隔符会兜底） return [{"text": chunk_text, "metadata": current_metadata}] # 初始调用 chunks = _recursive_split(text, metadata) # 处理重叠：简单实现，在每个块末尾添加下一个块开头的一部分 for i in range(len(chunks) - 1): current_text = chunks[i]["text"] next_text = chunks[i + 1]["text"] if self.chunk_overlap > 0 and len(next_text) >= self.chunk_overlap: overlap_part = next_text[:self.chunk_overlap] # 检查重叠部分是否已包含在当前文本末尾，避免重复 if not current_text.endswith(overlap_part): chunks[i]["text"] = current_text + overlap_part final_chunks = chunks # 为每个块添加顺序索引 for idx, chunk in enumerate(final_chunks): chunk["metadata"]["chunk_index"] = idx chunk["metadata"]["total_chunks"] = len(final_chunks) return final_chunks

4.3 向量化与存储的封装

这里展示如何用sentence-transformers和ChromaDB完成向量化存储。

from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings import hashlib from typing import List, Dict class VectorStoreManager: """向量存储管理器""" def __init__(self, embedding_model_name: str = 'BAAI/bge-small-zh-v1.5', persist_directory: str = "./chroma_db", collection_name: str = "rag_documents"): self.embedding_model = SentenceTransformer(embedding_model_name) self.persist_directory = persist_directory self.collection_name = collection_name # 初始化Chroma客户端 self.client = chromadb.PersistentClient(path=persist_directory, settings=Settings(anonymized_telemetry=False)) # 获取或创建集合 self.collection = self.client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} # 使用余弦相似度 ) def _generate_id(self, text: str, metadata: Dict) -> str: """根据文本和元数据生成唯一ID""" # 一个简单的方法：对文件路径和块索引做哈希 unique_string = f"{metadata.get('source_file', '')}_{metadata.get('chunk_index', 0)}" return hashlib.md5(unique_string.encode()).hexdigest() def add_documents(self, chunks: List[Dict[str, Any]]): """将一批文本块添加到向量数据库""" if not chunks: return texts = [chunk["text"] for chunk in chunks] metadatas = [chunk["metadata"] for chunk in chunks] ids = [self._generate_id(chunk["text"], chunk["metadata"]) for chunk in chunks] # 批量生成向量嵌入 logger.info(f"正在为 {len(texts)} 个文本块生成嵌入向量...") embeddings = self.embedding_model.encode(texts, normalize_embeddings=True, batch_size=32).tolist() # 注意batch_size logger.info("嵌入向量生成完毕。") # 批量插入到ChromaDB self.collection.add( embeddings=embeddings, documents=texts, metadatas=metadatas, ids=ids ) logger.info(f"成功插入 {len(ids)} 个文本块到集合 '{self.collection_name}'。") def search(self, query: str, n_results: int = 5, filter_metadata: Dict = None) -> List[Dict]: """语义搜索""" query_embedding = self.embedding_model.encode([query], normalize_embeddings=True).tolist()[0] results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results, where=filter_metadata # 可以用元数据过滤，如 {"category": "技术文档"} ) # 格式化返回结果 returned_docs = [] if results['documents']: for i in range(len(results['documents'][0])): returned_docs.append({ 'text': results['documents'][0][i], 'metadata': results['metadatas'][0][i], 'distance': results['distances'][0][i] }) return returned_docs

5. 完整工作流串联与配置化

将上述模块串联起来，形成一个完整的Pipeline类。这个类从配置文件（如config.yaml）中读取参数，并顺序执行各个步骤。

import yaml from pathlib import Path import time from .parser_factory import ParserFactory from .chunker import RecursiveOverlapChunker from .vector_store import VectorStoreManager from .utils import setup_logging, calculate_md5 class RAGDataPipeline: """RAG数据预处理流水线""" def __init__(self, config_path: str): with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) setup_logging(self.config.get('logging', {})) self.parser_factory = ParserFactory() self.chunker = RecursiveOverlapChunker( chunk_size=self.config['chunking']['size'], chunk_overlap=self.config['chunking']['overlap'] ) self.vector_store = VectorStoreManager( embedding_model_name=self.config['embedding']['model_name'], persist_directory=self.config['vector_db']['persist_dir'], collection_name=self.config['vector_db']['collection'] ) self.processed_cache = {} # 简单缓存，记录已处理文件的MD5，用于增量更新 def process_file(self, file_path: str) -> bool: """处理单个文件的核心流程""" logger.info(f"开始处理文件: {file_path}") start_time = time.time() # 1. 检查增量更新（示例：通过文件MD5） file_md5 = calculate_md5(file_path) if file_path in self.processed_cache and self.processed_cache[file_path] == file_md5: logger.info(f"文件未变化，跳过: {file_path}") return True # 2. 路由并解析 parser = self.parser_factory.get_parser(file_path) if not parser: logger.error(f"无法找到对应的解析器: {file_path}") return False parsed_data = parser.safe_parse() if not parsed_data: logger.error(f"解析文件失败: {file_path}") return False # 3. 清洗（此处省略具体清洗函数调用） raw_text = parsed_data['content'] cleaned_text = self._clean_text(raw_text) # 4. 准备分块元数据 base_metadata = parsed_data.get('metadata', {}) base_metadata.update({ 'source_file': str(Path(file_path).absolute()), 'file_type': Path(file_path).suffix, 'processed_time': time.time() }) # 5. 智能分块 chunks = self.chunker.split_text(cleaned_text, base_metadata) logger.info(f"文件分割为 {len(chunks)} 个文本块。") # 6. 向量化并存储 if chunks: self.vector_store.add_documents(chunks) # 更新缓存 self.processed_cache[file_path] = file_md5 elapsed = time.time() - start_time logger.info(f"文件处理完成: {file_path}, 耗时: {elapsed:.2f}秒， 生成{len(chunks)}个块。") return True else: logger.warning(f"文件未产生任何文本块: {file_path}") return False def process_directory(self, data_dir: str): """处理整个目录""" data_path = Path(data_dir) if not data_path.exists(): logger.error(f"数据目录不存在: {data_dir}") return supported_extensions = ['.pdf', '.docx', '.txt', '.md', '.html'] # 从配置读取更好 file_list = [] for ext in supported_extensions: file_list.extend(data_path.rglob(f"*{ext}")) total = len(file_list) success = 0 for idx, file_path in enumerate(file_list, 1): logger.info(f"处理进度: {idx}/{total}") if self.process_file(str(file_path)): success += 1 logger.info(f"目录处理完成。总计{total}个文件，成功{success}个，失败{total-success}个。") def _clean_text(self, text: str) -> str: """文本清洗函数示例""" # 移除多余的空格和换行 text = re.sub(r'\s+', ' ', text).strip() # 移除常见的页眉页脚模式（需要根据实际数据调整正则表达式） text = re.sub(r'第\s*\d+\s*页', '', text) # 更多清洗规则... return text

对应的config.yaml配置文件示例：

# config.yaml data_source: input_dir: "./data/raw_documents" watch_for_changes: false # 是否启用文件夹监控 chunking: size: 500 overlap: 50 separators: ["\n\n", "\n", "。", "；", "，", " ", ""] embedding: model_name: "BAAI/bge-small-zh-v1.5" device: "cpu" # 或 "cuda" vector_db: type: "chroma" persist_dir: "./data/vector_db" collection: "my_knowledge_base" logging: level: "INFO" file: "./logs/pipeline.log"

6. 高级特性与优化方向

一个基础流水线搭建完成后，可以考虑以下高级特性来提升其工业级强度。

6.1 分布式任务处理

当文档量极大时，单机处理会成为瓶颈。可以使用Celery或Dask搭建分布式任务队列。

设计：将process_file函数包装成一个Celery任务。输入模块将每个文件路径作为任务发布到Redis消息队列。多个Worker进程（可以在不同机器上）从队列中消费任务，并行处理文件，并将结果（chunks）批量发送回主进程或直接写入一个共享的向量数据库。
优势：水平扩展能力强，可以轻松应对百万级文档的处理。

6.2 复杂元数据与关联提取

除了基础元数据，可以集成NLP工具提取更丰富的信息。

实体识别：使用spaCy或StanfordNLP识别文本中的人名、地名、组织名、日期等，作为元数据字段。
关键词提取：使用TF-IDF或TextRank算法提取每个chunk的关键词，便于后续的混合检索（关键词+向量）。
摘要生成：为每个chunk生成一个简短摘要，既可以作为元数据，也可以在检索结果中预览。

6.3 混合检索与重排序

我们的工具产出的是向量数据库。但在真正的RAG系统中，检索阶段可以更复杂。

工具集成：在工具配置中，可以指定是否同时生成用于关键词检索（如Elasticsearch）的索引。实现“混合检索”，即同时进行向量相似度搜索和关键词匹配，然后合并结果。
重排序集成：可以配置在存储chunk时，也使用一个更强大的“重排序模型”为每个chunk生成一个分数（与一个虚拟查询相比）。或者在工具内提供一个测试接口，输入查询，返回经过重排序模型（如BGE-reranker）精排后的结果。这能让你在构建阶段就评估数据准备的质量。

6.4 质量评估与数据探查

构建一个简单的Web界面或命令行工具，用于：

抽样检查：随机查看一些文档被解析、分块后的结果，直观判断分块质量。
统计信息：展示处理了多少文件，生成了多少chunk，平均chunk长度，元数据分布等。
检索测试：输入一个查询，查看工具构建的知识库能返回哪些chunk，模拟RAG的检索环节。

7. 避坑指南与常见问题排查

在实际开发和运行中，我踩过不少坑，这里总结一下，希望能帮你节省时间。

7.1 解析阶段常见问题

PDF解析乱码或空白：
- 原因：PDF是扫描图片或使用了特殊字体编码。
- 解决：优先使用pdfplumber，并检查其extract_text方法的layout参数。对于扫描件，必须集成OCR。可以先用pdfplumber尝试提取，如果提取出的文本长度极短，则自动切换到OCR流程。
Word文档格式丢失：
- 原因：python-docx提取的是纯文本，丢失了标题级别等结构信息。
- 解决：遍历文档的段落（document.paragraphs），通过paragraph.style.name判断是否是标题（如Heading 1），并将其作为元数据提取出来。
网页提取大量无关内容：
- 原因：直接提取了整个HTML的body文本。
- 解决：使用readability或trafilatura这类专门用于提取主体内容的库。或者，用playwright渲染后，通过更精准的CSS选择器来定位正文区域。

7.2 分块阶段常见问题

块大小不均匀或语义被切断：
- 原因：固定长度分块对中文不友好，因为中文没有明显的单词分隔。
- 解决：采用递归字符分块（RecursiveCharacterTextSplitter）是很好的起点。对于中文，分隔符可以设置为["\n\n", "\n", "。", "；", "，", " ", ""]。更高级的方案是使用基于句子嵌入的语义分块。
元数据关联错误：
- 问题：某个chunk的section_title元数据指向了错误的章节。
- 解决：在解析阶段就构建文档的层次结构树。分块时，每个chunk都记录下它在树中的路径（如[“第一章”， “第一节”， “第一段”]）。这比只记录上一个标题更精确。

7.3 向量化与存储阶段常见问题

嵌入速度慢：
- 原因：单条文本调用模型。
- 解决：务必使用批处理。sentence-transformers的encode函数天然支持批处理，将batch_size设置为32、64或更大（取决于GPU内存）。同时，可以考虑使用多进程，将文本列表分片后并行编码。
向量数据库写入慢或内存溢出：
- 原因：一次性插入数十万条记录。
- 解决：实现分批插入。每处理完一定数量（如1000个）的chunk，就插入一次向量数据库。对于ChromaDB，注意其在内存中缓存数据，如果数据量极大，考虑使用ClickHouse或Qdrant等支持磁盘索引的数据库。
重复内容被多次插入：
- 原因：同一文件被多次处理。
- 解决：实现增量处理逻辑。如上面代码所示，记录文件的哈希值（MD5或最后修改时间）。在每次处理前进行比对，只有文件内容真正发生变化时才重新处理。

7.4 流程与运维问题

如何处理失败的任务：
- 策略：不要因为一个文件解析失败就停止整个流程。使用try...except捕获每个文件的异常，记录到错误日志文件和任务状态表中。定期检查失败任务，人工介入排查原因（是文件损坏还是解析器bug）。
如何监控流水线健康度：
- 方案：在关键步骤（开始处理文件、解析成功、分块完成、存储成功）输出结构化的日志（JSON格式）。使用Logstash或Fluentd收集日志，并导入Elasticsearch+Kibana（ELK栈）或Grafana进行可视化监控，可以清晰看到处理速度、成功率、各阶段耗时等指标。