这次我们继续“大模型RAG和Cursor实战”系列。系列前面几篇已经拆过RAG的整体链路、向量检索方式和提示词组织,这一篇落到组件篇里最容易被低估、却又直接决定知识库质量上限的一环——数据导入技术。
一句话总结:RAG能不能答得准,很多时候不是模型选得不够大,而是知识库里的原始数据没有经过有效加工。PDF里的表格会错位、扫描件是图片、HTML页面塞满导航和广告、关系数据库里是关系型结构而不是自然语言——这些内容如果直接丢给后续的向量化与检索环节,召回结果大概率是灾难。数据导入技术,核心任务就是把PDF、Word、网页、数据库记录这类杂乱输入,加工成大模型知识库可以加载、切分、检索的标准化文本块。
本文会围绕数据导入管道设计、多数据源解析、文本切分策略、增量入库、质量验证这几个方向展开,提供可复用的Python工程代码,并演示如何用Cursor把重复的解析脚本快速搭起来。如果你正在做RAG知识库项目,或者遇到了“文档切分后语义破碎”“PDF表格提取乱掉”“数据库怎么转成大模型能读懂的数据”这类问题,这篇文章可以直接收藏。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 组件定位 | RAG 数据预处理管道,负责加载、解析、清洗、切分、入库 |
| 负责数据源 | PDF、Word、Markdown、HTML、TXT、关系数据库等 |
| 核心输出 | 结构化文本块(chunk)+ 元数据,供向量化和检索使用 |
| 工程形态 | Python 脚本 / CLI 工具 / FastAPI 服务 / 定时增量任务 |
| 关键依赖 | pypdf、PyMuPDF、pdfplumber、python-docx、BeautifulSoup、SQLAlchemy、LangChain Text Splitters |
| 是否需要 GPU | 不需要,数据导入本身是 CPU 密集型任务 |
| 显存占用 | 0(除非在管道中调用本地大模型做清洗或抽取) |
| 支持批量任务 | 支持,目录扫描、流式解析、并发控制均可实现 |
| 是否支持 API | 支持,可封装为独立的数据导入服务 |
| 主要瓶颈 | PDF 复杂版面、扫描件 OCR、超大批量文件的吞吐与去重 |
| 适合场景 | 企业知识库、客服问答、文档检索、数据库内容加工、RAG 数据预处理 |
从工程角度看,数据导入是 RAG 中“性价比”最高的优化点。换模型、调提示词不一定能解决召回质量差的问题,但把数据切分策略和清洗规则改到位,通常立竿见影。
2. 数据导入在 RAG 链路中扮演什么角色
RAG 的完整链路可以简化为下面这条流水线:
原始数据加载 -> 内容解析 -> 数据清洗 -> 文本切分 -> 向量化 -> 写入向量库很多入门教程会把重点放在“向量化”和“检索”上,但向量化模型只能对一段文本做语义编码。如果进入向量的文本本身是杂乱的、截断的、混入大量噪声的内容,再强的 Embedding 模型也救不回来。
数据导入做的是流水线的前半段,约等于把原始数据变成“可以被大模型理解的知识单元”。这一步需要明确几个目标:
- 保留原文的语义完整性,不能让一句话被硬切到两个 chunk 里。
- 剥离噪声内容,例如页眉页脚、导航栏、广告、重复模板。
- 补充上下文元数据,例如来源文件、页码、章节标题,方便后续过滤和溯源。
- 保持知识单元粒度合适,既能被检索到,又不会引入过多无关信息。
在实际项目里,常见的问题是团队花大精力选了向量库和 Embedding 模型,结果文档加载解析这一步没有做版面分析,导致 PDF 表格、多栏文本完全错乱。最后检索出来的内容虽然“语义相似”,但答案依赖的关键数据已经丢了。所以说,数据导入是决定 RAG 效果上限的第一道闸门。
3. 适用场景与使用边界
3.1 适合解决的场景
- 企业制度文档、技术手册、产品说明书的批量入库。
- PDF、Word、网页混存的历史资料统一整理。
- 关系数据库中的业务数据需要转成自然语言描述,供大模型问答。
- 需要增量同步文档变化,定时重建知识库的长期维护项目。
- 需要把解析、切分、入库做成接口服务,供多个业务系统调用的场景。
3.2 不适合或需要特别控制的场景
- 对数据实时性要求极高的业务,例如秒级变化的交易数据,不应直接用离线导入管道,需要配合实时流处理。
- 涉及大量敏感个人信息、商业机密的文档,要优先做脱敏和权限控制,向量库本身不提供细粒度权限管理。
- 需要强多步推理的复杂问答,光靠“导入+检索”不够,还要配合 Agent 和工具调用。
- 扫描件占比过高的资料库,需要额外引入 OCR 流程,解析成本和错误率都会明显上升。
3.3 版权与合规边界
文档解析和数据处理必须基于合法获得的内容。不管是用 PyMuPDF 提取 PDF,还是从网页抓取内容,都需要确认是否有权使用这些材料。RAG 知识库如果用于企业内部或公开产品,要避免把来路不明的爬虫内容、盗版电子书、未授权转载直接灌入。涉及人脸、个人身份信息、声音等敏感数据时,需要脱敏并限制访问范围。
4. 环境准备与前置条件
数据导入主要是 CPU 密集型任务,普通笔记本即可跑起来。推荐环境如下:
- 操作系统:Windows 10/11、Ubuntu 20.04+、macOS 均可。
- Python:3.9 或更高版本。
- 依赖管理:建议使用 venv 或 conda 创建独立环境,避免污染系统 Python。
- 数据库访问:根据目标源库决定,例如 MySQL 需要 pymysql,PostgreSQL 需要 psycopg2。
- 向量库:本文代码示例不绑定具体向量库,入库环节按自己的选型替换即可。
创建虚拟环境并安装依赖:
# 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate # 安装基础依赖 pip install pymupdf pdfplumber python-docx beautifulsoup4 lxml sqlalchemy pymysql pip install langchain langchain-text-splitters langchain-community如果确定要用 LangChain 的加载器,依赖里还需要对应的文档加载库,例如unstructured或pypdf。这里更推荐先用直接库解析,再把结果转成统一的文档结构,原因是 LangChain 加载器更新频繁,部分加载器不稳定,且额外封装会拉高排查成本。
Cursor 这边需要做的是把项目目录打开,并确保 Python 解释器指向虚拟环境。如果 Cursor 界面是英文,可以在 Settings 里切换中文语言包,不过对写代码影响不大。比较有用的做法是给 Cursor 配置一个项目级规则文件,告诉它“生成的解析脚本要捕获异常、要有日志、要兼容批量输入”,这样后续生成的代码会稳定很多。
5. 用 Cursor 快速搭建数据导入管道骨架
数据导入脚本的代码模式高度重复:加载文件、解析内容、清洗、切分、入库。这类任务非常适合用 Cursor 做初始搭建,然后再人工 review。
在 Cursor 中新建项目目录后,可以先写一个任务说明文件,例如data_ingestion/README_TASK.md,内容大致是:
请生成一个 Python 数据管道脚本,包含以下模块: 1. loader.py:支持 PDF、DOCX、MD、HTML、TXT 文件加载,统一返回 Document 对象列表。 2. parser.py:对 HTML 做正文提取,对 PDF 做文本与表格提取。 3. cleaner.py:去除多余空白、页眉页脚、特殊控制字符。 4. splitter.py:使用递归字符切分,chunk_size=500,overlap=50。 5. metadata.py:为每个 chunk 补充来源文件名、页码、切分序号。 6. batch_runner.py:批量扫描目录,打印每个文件的处理结果和耗时。 所有脚本需要包含日志、异常捕获和类型标注。把任务说明交给 Cursor 后,它通常会生成一版初稿。这里不建议盲目接受全部输出,重点检查三件事:
- 文件读取路径用的是绝对路径还是相对路径,是否需要改成从配置读取。
- 异常处理是否完整,单个文件解析失败时会不会中断整个批量任务。
- 切分参数是否满足你的知识粒度需求,而不是照搬默认值。
用 Cursor 生成代码的效率优势不在于“一次写对”,而在于把重复的骨架代码快速铺开,把时间省下来处理文档解析的特殊情况。
6. 常见数据源导入实战
6.1 PDF 解析:最需要单独处理的一类
PDF 表面上是文档,内里可能是文本、矢量图形、扫描图片的混合体。解析 PDF 的第一件事是判断它属于文本型 PDF 还是扫描型 PDF。
文本型 PDF 可以直接用 PyMuPDF 提取文本:
# extract_pdf.py # 使用 PyMuPDF 提取文本型 PDF 内容,并保留页码信息 import fitz # PyMuPDF from dataclasses import dataclass, field @dataclass class Document: page_content: str metadata: dict = field(default_factory=dict) def extract_text_pdf(file_path: str, start_page: int = 0, end_page: int = None) -> list[Document]: docs = [] with fitz.open(file_path) as pdf: total_pages = len(pdf) if end_page is None or end_page > total_pages: end_page = total_pages for page_index in range(start_page, end_page): page = pdf[page_index] text = page.get_text("text") if text.strip(): docs.append(Document( page_content=text.strip(), metadata={ "source": file_path, "page": page_index + 1, "total_pages": total_pages } )) return docs这段代码的优点是解析结果稳定、速度快,缺点是无法还原多栏文本的正确阅读顺序。如果在真实项目中遇到双栏论文、复杂表格,仅靠get_text会得到错乱的文本顺序。这时候要么按坐标排序文本块,要么引入版面分析模型,例如使用 PaddleOCR 的版面分析能力,或者使用专门的文档解析服务。
扫描型 PDF 没有文本层,只能走 OCR。常见方案是转成图片后调 PaddleOCR:
# ocr_pdf.py # 对扫描型 PDF 做页面渲染并调用 PaddleOCR 识别文本 import fitz from paddleocr import PaddleOCR ocr = PaddleOCR(use_angle_cls=True, lang="ch", show_log=False) def ocr_pdf(file_path: str, dpi: int = 200) -> list[Document]: docs = [] with fitz.open(file_path) as pdf: for page_index, page in enumerate(pdf): pix = page.get_pixmap(dpi=dpi) img_bytes = pix.tobytes("png") result = ocr.ocr(img_bytes, cls=True) lines = [] if result: for line in result: if line: lines.append(line[1][0]) text = "\n".join(lines) if text.strip(): docs.append(Document( page_content=text.strip(), metadata={ "source": file_path, "page": page_index + 1, "ocr": True } )) return docsOCR 的注意点是:解析速度慢、识别结果不一定准确,尤其是公式、特殊符号、表格线。扫描件导入前要先做质检,不能直接相信 OCR 输出。
6.2 Word、Markdown、HTML 解析
Word 文件使用 python-docx 提取段落和表格:
# extract_docx.py from docx import Document as DocxDocument from dataclasses import dataclass, field @dataclass class Document: page_content: str metadata: dict = field(default_factory=dict) def extract_docx(file_path: str) -> list[Document]: doc = DocxDocument(file_path) blocks = [] for para in doc.paragraphs: text = para.text.strip() if text: blocks.append(Document( page_content=text, metadata={"source": file_path, "type": "paragraph"} )) for table in doc.tables: rows = [] for row in table.rows: cells = [cell.text.strip().replace("\n", " ") for cell in row.cells] rows.append(" | ".join(cells)) if rows: blocks.append(Document( page_content="\n".join(rows), metadata={"source": file_path, "type": "table"} )) return blocksWord 转出来的表格往往带有大量空单元格和重复内容,入库前可以再做一步压缩,只保留非空行。
Markdown 文件的特征是自带标题结构。标题本身是很好的切分边界,可以直接按#、##、###拆分,而不是按固定字符数硬切。大多数 LangChain 加载器会把 Markdown 里的大段内容一次性带出来,后续再用标题结构切分会更合理。
HTML 解析的重点是正文提取。直接用 BeautifulSoup 拿get_text()会把导航、脚注、广告全部带进来:
# extract_html.py from bs4 import BeautifulSoup from dataclasses import dataclass, field @dataclass class Document: page_content: str metadata: dict = field(default_factory=dict) def extract_html(file_path: str) -> list[Document]: with open(file_path, "r", encoding="utf-8") as f: soup = BeautifulSoup(f.read(), "lxml") # 移除脚本、样式、导航、广告等干扰节点 for tag in soup(["script", "style", "nav", "header", "footer", "aside", "iframe"]): tag.decompose() title = soup.title.string.strip() if soup.title else "" body = soup.body.get_text("\n", strip=True) if soup.body else "" chunks = [] if title: chunks.append(Document( page_content=title, metadata={"source": file_path, "type": "title"} )) if body: # 按空行切分出多个逻辑块,避免全文塞进一个 chunk logical_blocks = [b.strip() for b in body.split("\n") if b.strip()] for idx, block in enumerate(logical_blocks): chunks.append(Document( page_content=block, metadata={"source": file_path, "type": "content", "block_index": idx} )) return chunksHTML 正文提取是工程性很强的一环。如果只是内部少量页面,上面的过滤规则够用。要是需要从大量来源不同的网页里稳定提取正文,可以使用trafilatura或readability这类专门做正文抽取的库,它们对正文和噪声的识别更成熟。
6.3 关系数据库数据如何加工成大模型能读懂的数据
数据库里的记录是结构化字段,不是自然语言。直接把这些字段拼起来灌进向量库,检索效果通常不好,因为用户提问时使用的是自然语言,而一条用户记录可能长这样:
order_id: 202405001 user_name: 张三 product: 笔记本电脑 price: 5999 status: 已发货大模型读懂的自然语言版本应该是:
订单 202405001 由用户张三购买了一台笔记本电脑,订单金额为 5999 元,当前状态是已发货。把关系数据库加工成 RAG 可用的文本,核心思路是“字段到语义化文本”的映射转换。下面给出一套通用做法:
# db_to_rag.py # 从关系数据库读取订单数据,生成语义化的文本块 from sqlalchemy import create_engine, text DB_URL = "mysql+pymysql://user:password@localhost:3306/shop" ORDER_FIELDS = { "order_id": "订单编号", "user_name": "购买用户", "product": "商品名称", "price": "成交金额", "status": "订单状态", } def row_to_text(row, field_map: dict) -> str: parts = [] for db_field, label in field_map.items(): value = row.get(db_field) if value is not None: parts.append(f"{label}为{value}") return ",".join(parts) def load_orders_to_documents(batch_size: int = 1000): engine = create_engine(DB_URL) with engine.connect() as conn: # 按主键分批读取,避免一次性加载全表导致内存溢出 last_id = -1 while True: sql = text( "SELECT * FROM orders " "WHERE id > :last_id ORDER BY id LIMIT :batch_size" ) rows = conn.execute(sql, {"last_id": last_id, "batch_size": batch_size}).mappings().all() if not rows: break for row in rows: content = row_to_text(dict(row), ORDER_FIELDS) if content.strip(): print(content) # 这里替换成后续的切分与入库逻辑 last_id = rows[-1]["id"]数据库导入的几个工程要点:
- 多表关联时,优先在 SQL 中完成 JOIN,再在应用层生成描述文本。顺序是先聚合,再转文本,避免一条记录对应多个不完整块。
- 数值字段不要直接拼接,最好带上单位,模型才能理解“5999”是元还是分。
- 涉及时间、状态等枚举字段,翻译成中文可读描述,而不是直接写
status=2。 - 敏感字段入库前先做脱敏,例如手机号、身份证号不能原样进入向量库。
6.4 批量目录扫描与增量更新
数据导入不是一次性任务,文档库会持续更新。批量导入的骨架可以这样设计:
# batch_runner.py import hashlib import json import logging from pathlib import Path logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) SUPPORTED_SUFFIX = {".pdf", ".docx", ".md", ".html", ".htm", ".txt"} CACHE_FILE = Path("./processed_files.json") def file_md5(file_path: Path) -> str: hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def load_processed_cache() -> dict: if CACHE_FILE.exists(): return json.loads(CACHE_FILE.read_text(encoding="utf-8")) return {} def save_processed_cache(cache: dict): CACHE_FILE.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8") def scan_and_process(base_dir: Path): cache = load_processed_cache() for file_path in sorted(base_dir.rglob("*")): if file_path.suffix.lower() not in SUPPORTED_SUFFIX: continue try: md5 = file_md5(file_path) if cache.get(str(file_path)) == md5: logger.info(f"跳过未变更文件: {file_path}") continue logger.info(f"开始处理: {file_path}") # TODO: 调用对应解析器,得到 Document 列表后执行入库 cache[str(file_path)] = md5 except Exception as e: logger.error(f"处理失败 {file_path}: {e}", exc_info=True) save_processed_cache(cache) if __name__ == "__main__": scan_and_process(Path("./docs"))用文件 MD5 做增量判断,简单有效。如果文件数量特别大,可以做两层策略:先用文件大小和修改时间快速过滤,再对候选文件计算 MD5。对于数据库导入,增量更新则依赖业务表的时间戳或自增 ID,记录每个批次的游标位置。
7. 文本切分策略:数据导入最关键的一步
解析完成后,文档是一大段文本,不能直接向量化,必须切分成 chunk。切分不等于简单截断,核心目标是保持语义完整。
固定长度切分的问题很明显:一句话可能被从中间切断,检索回来的 chunk 往往是残缺的。工程中最常用的折中方案是 LangChain 的递归字符切分器:
# split_text.py from langchain_text_splitters import RecursiveCharacterTextSplitter # chunk_size 和 chunk_overlap 需要按实际文档调整 splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "!", "?", ",", ";", " ", ""], length_function=len, ) def split_document(doc, splitter): texts = splitter.split_text(doc.page_content) chunks = [] for idx, text in enumerate(texts): metadata = dict(doc.metadata) metadata["chunk_index"] = idx chunks.append({"text": text, "metadata": metadata}) return chunks指示切分器按“双换行、单换行、句末标点、中文逗号”的优先级去断句,比纯按字符数切块自然很多。
如果文档有清晰的 Markdown 标题,推荐按标题层级切分:
# split_by_headers.py # 使用 MarkdownHeaderTextSplitter 按标题结构切分,保留标题上下文 from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "H1"), ("##", "H2"), ("###", "H3"), ] splitter = MarkdownHeaderTextSplitter(headers_to_split_on) def split_markdown_by_header(markdown_text: str): docs = splitter.split_text(markdown_text) return [ { "text": doc.page_content, "metadata": doc.metadata } for doc in docs ]标题切分的优势在于每个 chunk 自带章节上下文,检索时命中某一段,可以知道它属于哪一章。
切分参数的选择并没有固定答案。chunk_size 太小会导致检索信息不足,太大则会引入无关内容、降低召回精度。更稳妥的做法是先用不同参数跑一批文档,抽样检查检索效果,再确定基线参数。
对于代码文档、配置文档这类内容,按代码函数或配置块切分比按字符切分更合理,这需要识别文档结构,属于更细粒度解析的范畴。
8. 数据清洗与元数据管理
文档解析后的原始文本通常不能直接用,至少要做以下清洗:
- 去除非打印字符、控制字符、连续空白。
- 删除页眉页脚、页码、日期水印。
- 合并因为 PDF 分页而断裂的段落。
- 对明显重复的模板文本去重。
- 去除 URL 中的跟踪参数、HTML 实体转义。
清洗逻辑建议写成独立的 cleaner 模块,方便复用和调试。每个清洗规则单独一个函数,并输出清洗前后的文本对比日志。
元数据管理同样重要。一个 chunk 至少应该带上来源文件、页码、切分序号。如果文档有标题结构,把标题层级也放进 metadata。这些信息在检索阶段有两大作用:
- 生产环境做权限和来源过滤,只搜索指定目录或指定文档类型。
- 给大模型提供引用来源,回答问题时可以追溯到具体文档和页码。
元数据拼接时要注意体积,超出 Embedding 模型输入上限的信息会拖慢推理速度。建议只保留必要字段,长章节内容不要整段塞进 metadata。
9. 效果验证与质量评估
数据导入做完了,不能只看“跑通了”,要验证导出的文本块是否真的适合 RAG。先做手工抽检,再做批量评估。
9.1 中间产物可视化
每批次导入后,把生成的 chunk 和 metadata 写成 JSON 落盘,打开肉眼检查:
{ "text": "订单 202405001 由用户张三购买了一台笔记本电脑,订单金额为 5999 元,当前状态是已发货。", "metadata": { "source": "mysql://shop/orders", "chunk_index": 0, "business_type": "order" } }如果 chunk 是截断的半句话,或者混进了页眉文本,就要回头调整切分参数或清洗规则。这一步比任何指标都有价值。
9.2 召回质量抽检
准备 10 到 20 条真实业务 query,对每个 query 执行“检索 top-k”,人工判断返回的 chunk 是否与问题相关。记录相关命中的比例,作为基线。
9.3 指标化评估
如果后续要量化优化,可以从三个维度评估知识库效果:
| 指标 | 关注点 | 一般做法 |
|---|---|---|
| 召回率(Recall) | 正确答案是否稳定出现在检索结果中 | 命中 / 总问题数 |
| 上下文相关性 | 返回的 chunk 是否紧密围绕问题 | 人工打分或 LLM 打分 |
| 答案准确率 | 大模型基于 chunk 的回答是否正确 | 与 golden answer 对比 |
热搜词里提到的“RAG 知识库指标”通常也是指这一组。数据导入阶段影响最大的是召回率,因为切分不合理或清洗不干净,正确内容根本没有进入向量库。
10. 接口 API 与批量任务
数据导入脚本积累到一定规模后,建议封装成服务,方便业务系统调用,也可以把“导入一批文件”暴露为异步任务。
用 FastAPI 提供服务的最小示例:
# api_service.py from fastapi import FastAPI, UploadFile, File, BackgroundTasks from pydantic import BaseModel, Field app = FastAPI() class ImportResponse(BaseModel): task_id: str = Field(description="任务ID") status: str = Field(description="任务状态") def run_import_task(file_path: str): # 这里执行实际的解析、切分、入库逻辑 print(f"开始导入: {file_path}") @app.post("/import/file", response_model=ImportResponse) async def import_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)): # 保存上传文件到临时目录,注意限制后缀和文件大小 temp_path = f"./uploads/{file.filename}" with open(temp_path, "wb") as f: f.write(await file.read()) background_tasks.add_task(run_import_task, temp_path) return ImportResponse(task_id=temp_path, status="queued")批量任务要加失败重试和任务表,比较通用的设计是:
- 文件上传后先落盘,记录文件名、大小、MD5。
- 任务进入队列,状态为 pending。
- 处理中记录日志,异常时状态改为 failed,捕获完整堆栈。
- 处理成功后标记 done,并保存向量库返回的 ID。
批量任务最怕的是单个坏文件拖垮整个流程,所以每个文件处理时都要单独 try/except,失败文件单独记录,不能中断整个批处理。
11. 资源占用与性能观察
数据导入本身不涉及 GPU,主要看 CPU、内存和磁盘 I/O。性能观察点集中在几个环节:
- PDF 解析:文本型 PDF 很快,扫描件 OCR 很慢,一张高分辨率页面可能需要几秒。
- 数据库读取:大批量查询会占内存,务必用游标或分批读取。
- 文本切分:纯文本操作,性能影响不大,但大批量文件切分时要观察内存。
- 向量化入库:如果 embedding 模型跑在本地 CPU 上,1000 个 chunk 可能比解析还耗时;如果接的是远程 API,则要看请求并发和响应速度。
批量导入时建议观察进程的内存和 CPU 曲线。如果内存持续上升,优先排查是否把所有文件内容都读进了内存而不是流式处理。更稳的做法是限制并发数,例如同时最多处理 4 个文件,每个文件内部串行解析。
增量导入能明显降低资源消耗。首次全量导入无可避免,之后只处理变更文件,可以减少无效解析时间。对于每天新增几十个文件的场景,定时任务配合 MD5 缓存已经足够。
12. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| PDF 提取的文字乱码 | PDF 是扫描件或使用了自定义编码 | 检查是否无文本层,查看单页渲染效果 | 改用 OCR 流程 |
| PDF 表格提取错位 | 表格线复杂,文本提取顺序乱 | 打印原始文本块坐标和内容 | 引入版面分析,或按坐标排序文本块 |
| 中文文本被切分成乱句 | 切分器未包含中文标点分隔符 | 检查 chunk 内容是否在标点处截断 | 调整 separators,加入中文标点 |
| HTML 提取到大量导航和广告 | 提取时未移除 header/nav 等干扰节点 | 打印提取后的全文头部内容 | 按正文抽取规则过滤,或使用 trafilatura |
| 数据库导入一条记录生成多个不完整 chunk | 字段直接拼成大段文本,切分切断了语义 | 检查生成文本是否在字段中间断开 | 改为按记录生成语义化短文本,再决定是否切分 |
| 重复文件反复导入了多次 | 缺少文件去重机制 | 检查向量库中是否有重复内容 | 用 MD5 或业务主键记录处理状态 |
| 批量导入时内存持续增长 | 一次性加载了过多文件到内存 | 观察进程内存曲线 | 改流式读取,控制并发 |
| OCR 识别率低 | 图片质量差、倾斜、文字太小 | 抽样看 OCR 结果与原图 | 提高渲染分辨率,先做图像矫正 |
| 接口服务超时 | 导入任务执行时间过长,同步等待 | 查看日志耗时分布 | 改为后台任务加轮询状态 |
排查数据导入问题最有效的方式是“中间落盘”。每一环节都输出一个可检查的中间文件,例如原始文本 JSON、切分后 chunk JSON,这样能快速定位是解析环节出错还是切分环节出错。
13. 最佳实践与使用建议
- 第一次跑通不要追求全量导入,先拿 10 个代表性文件做小样本测试,确认解析、切分、检索效果都没问题,再开全量。
- 保留一套最小可运行目录:原始文件、临时 JSON 产物、最终 chunk 输出、日志分别放在不同目录,方便回溯。
- 每个 chunk 都要带来源元数据,否则后续问题定位成本会很高。
- 数据库导入时,敏感字段必须先脱敏,向量库的访问权限也要单独控制。
- 使用 Cursor 生成数据管道代码时,不要直接接受全部输出,重点审查文件路径、异常处理和日志是否完整。
- 文档解析器和 Oracle 类的依赖经常更新,建议锁版本并定期回归测试,避免升级后解析结果变化。
- 涉及版权材料的文档,导入前确认获取渠道合法;涉及个人隐私的信息,做字段级脱敏;扫描件和图片数据要避免未经授权采集人脸等敏感信息。
- 发布或商用 RAG 产品前,抽检一批真实 query 的检索结果,确认没有导入错误造成的事实性误导。
14. 总结与下一步
数据导入技术是整个 RAG 知识库项目里投入产出比最高的一环。PDF 解析、数据库语义化、文本切分、增量入库这些环节虽然不涉及模型训练,却直接决定上游向量检索和下游大模型回答质量的基线。数据没洗干净,后续换模型、调提示词都只是修补。
如果读者接下来要动手做,建议按照“单个文件解析 -> 全文档切分 -> 检索抽检 -> 批量落库 -> 增量更新”的顺序推进。最容易踩的坑是 PDF 表格和扫描件,尽早确认自己的文档库里有没有这类内容,比等到上线后被用户发现要好得多。
这一篇之后,可以把数据导入的产物接到向量化与检索环节,验证 chunk 的召回效果。如果数据导入管道已经稳定,下一步可以继续拆 RAG 的检索重排、Agent 化方案,或者用 Cursor 继续构建后续的知识库管理界面。建议收藏备用,实际项目里大概率需要反复翻阅其中的代码和排查思路。