1. 拆解需求批量导入知识库到底难在哪1.1 一个真实场景引出的痛点我手上有一批资料散在十几个文件夹里产品文档是 Word技术白皮书是 PDF日常笔记和部分规范是 Markdown还有一些从网页复制下来存成 txt 的零碎内容。过去往知识库里放东西基本靠手工——打开文件、复制正文、粘贴到录入框、填标题、选标签一条一条来。几十个文件还能忍上百个文件就是纯体力活而且人一累就会漏、会错标题和内容对不上号的低级错误我都犯过好几次。这就是我想写个脚本的起因。目标很朴素给脚本一个目录它自己把里面所有受支持的文件读出来、清洗、切分然后按知识库接口要求批量提交我只在旁边看日志就行。说白了就是把重复劳动一次性交给机器。这里需要先明确一点“批量导入知识库”不是一个单一动作它其实是一条流水线文件扫描 → 格式解析 → 文本清洗 → 内容分块 → 元数据组装 → 接口调用 → 结果回执 → 失败重试。任何一环没处理好整体都会出问题。下面我按这条流水线的顺序把每一段的设计和取舍讲清楚。1.2 为什么不用现成工具非要自己写先说清楚我为什么没直接找个现成的导入工具。原因有三个都是实际用下来才体会到的。第一是格式太杂。市面上不少工具对单一格式支持很好但对“Word PDF Markdown 混着来”的支持参差不齐经常出现某类文件解析出来一堆乱码。第二是分块策略不能自己定。知识库检索效果好不好很大程度取决于文本怎么切。现成工具往往固定一个 chunk 大小改不了。我这边有些政策条款必须整段保留有些技术文档又适合切细一点固定策略满足不了。第三是接口不透明。我想知道每一条到底提交成功没有、失败原因是什么、能不能只重试失败的那部分。自己写日志我想怎么打就怎么打。所以说自建脚本的本质不是“重复造轮子”而是把导入过程变成一个可控、可观测、可重跑的流程。这是后面所有设计的出发点。1.3 整体思路和方案选型整体方案我定成这样用 Python 做主脚本负责遍历、解析、分块、提交用 shell 脚本做一个简单的入口包装方便定时执行或手动触发。选 Python 而不是纯 shell理由很直接——文本解析生态好。读 docx 有 python-docx读 PDF 有 pdfplumber字符串处理、正则、JSON 组装都是现成的用 shell 去解析二进制文档基本是自找麻烦。选 shell 做入口是因为它启动快、写起来简单配合 crontab 做增量导入很顺手。注意格式解析库只负责“把内容尽量取出来”不保证 100% 还原排版。表格、图文混排、多栏 PDF 是最容易出问题的三类内容后面会专门讲怎么处理。参数上我一开始定了几条硬指标单次运行要能处理上千个文件、失败率控制在个位数、单文件解析超时要能跳过而不是卡死整个任务。这些约束直接决定了后面分块大小、超时设置和重试逻辑的写法。2. 脚本骨架拆解与关键参数怎么定2.1 目录扫描与文件类型过滤第一步是找出“要处理哪些文件”。最省事的做法是递归遍历目录把扩展名在白名单里的文件收进列表。我用一个集合来做白名单这样判断速度更快也方便后续扩展。这里有个容易忽略的点遍历顺序会带来隐性重复。如果目录里有软链接指向同一份文件或者同名文件出现在多个子目录直接按路径导入会重复。我的做法是对每个文件先算一个内容哈希把哈希放进集合做去重已经在集合里的直接跳过。另一个细节是隐藏文件和临时文件要过滤掉。像~$开头的 Word 临时文件、.DS_Store这类系统文件混进去会导致解析报错。过滤逻辑很简单但在实际跑的时候能省掉一大堆莫名其妙的报错。2.2 Word、PDF、Markdown 三种格式怎么解析这三类文件的解析思路不一样我分开说。Word.docx用 python-docx 读段落。要注意它读的是“段落对象”表格里的内容默认不在段落流里需要单独遍历表格单元格。我的处理是把段落和表格都读出来按出现顺序拼接虽然会丢一点原始排版但正文完整性有保障。PDF这是最麻烦的。PDF 本质是排版描述不是结构化文档所以解析出来经常是“一行被拆成好几段”。我的做法是用 pdfplumber 按页提取文字然后按行合并再根据空行和标点做二次分段。扫描版 PDF图片型纯文本库读不出来这种情况我会在日志里标记出来靠人工补不硬凑。Markdown最友好直接读文本就行。我会顺手把代码块、超链接的冗余语法清理掉但保留标题层级因为标题对分块很有用——它是天然的段落边界。实操心得解析库的版本要固定。我在不同机器上跑pdfplumber 换了个小版本同一份 PDF 出来的换行位置就变了。生产用的脚本依赖版本最好锁死别用“随最新版”。2.3 文本分块chunk 大小和重叠怎么定分块是决定检索质量的关键。切太大一段里混了好几个主题检索时命中不精准切太小一句话被拆散语义不完整。我的经验值是这样定的参数取值说明chunk_size500 字符中文场景下大约 300~400 字语义较完整chunk_overlap80 字符给相邻块留一点重叠避免边界信息丢失最小块长度50 字符低于这个长度的碎片直接丢弃或合并到上一块为什么要留重叠因为句子被硬切断时关键词可能正好落在切口上。留一段重叠检索时至少有一个块能包含完整语义。重叠太大又会导致冗余80 字符是我反复试下来比较平衡的值。切分边界上我优先按标题、空行、句号这类“硬边界”切实在没有边界才按字符数硬切。这个逻辑在代码里就是先找分隔符找不到再退化到长度切分。2.4 接口调用与失败重试提交环节要解决三个问题限流、超时、幂等。限流方面知识库接口一般都有速率限制短时间内大量请求会被拒。我在脚本里加了一个简单的节流每提交 N 条 sleep 一下或者用令牌桶控制速率。超时方面每条请求都要设超时时间网络抖动时不能让整个脚本挂住。超时后不直接判失败而是放进重试队列。幂等方面这条最重要。脚本可能会被重跑比如中途挂了再执行如果没有幂等控制同一份内容会重复入库。我的做法是给每个文件算哈希把“已成功导入的哈希”记到一个本地记录文件里重跑时先查记录已经成功的跳过。注意幂等靠的是“内容哈希 成功记录”不是靠文件名。文件改名后内容没变仍然应该被识别为已导入。3. 拿 AI 手搓脚本的真实过程3.1 怎么给 AI 描述需求才有效这是我这次最想分享的部分。让 AI 写代码效果好坏九成取决于你怎么描述。我踩过的坑是一开始只丢一句“帮我写个批量导入知识库的脚本”结果它给我的代码假设了一个根本不存在的接口格式还用了我看不懂的库。后来我改成“分块描述需求”效果好很多。具体是这样拆的先告诉它输入是什么一个目录里面有 docx、pdf、markdown 三类文件。再告诉它输出接口长什么样接口地址、请求方法、请求体的字段名、认证方式。然后告诉它关键约束chunk 大小、超时、去重、日志格式。最后要求它按函数拆分每个函数只干一件事不要写成一个巨型 main。把接口字段名、请求示例清楚地写进提示词AI 生成的代码可用率会大幅提升。因为它是照着你的规格在填空而不是凭空猜。3.2 分阶段生成别一次性要完整代码一次性要一整套代码出来的东西通常“看起来对、跑起来错”。我的做法是分阶段第一阶段只要文件遍历和解析让它先能打印出每个文件读到了多少字。跑通之后再进入下一阶段。第二阶段只要分块逻辑我拿几份典型文件验证切分结果合不合理。第三阶段才是接口提交和重试。这时候前面解析、分块已经稳定出问题就只可能出在网络和接口部分排查范围小很多。这个“小步验证”的思路是我觉得比代码本身更值钱的经验。AI 写代码快但你必须有能力分阶段验收否则就是在赌。3.3 完整脚本拆解与逐段注释下面是我最终的脚本骨架去掉业务细节后大概是这个结构import os import json import time import hashlib import requests from pathlib import Path SUPPORTED_EXT {.docx, .pdf, .md, .txt} CHUNK_SIZE 500 CHUNK_OVERLAP 80 RECORD_FILE imported_hashes.json API_URL https://your-kb-endpoint/api/documents REQUEST_TIMEOUT 15 MAX_RETRY 3 def load_imported(): if os.path.exists(RECORD_FILE): with open(RECORD_FILE, r, encodingutf-8) as f: return set(json.load(f)) return set() def save_imported(hashes): with open(RECORD_FILE, w, encodingutf-8) as f: json.dump(list(hashes), f, ensure_asciiFalse) def file_hash(path): h hashlib.md5() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() def iter_files(root): for dirpath, _, filenames in os.walk(root): for name in filenames: if name.startswith(~$) or name.startswith(.): continue if Path(name).suffix.lower() in SUPPORTED_EXT: yield os.path.join(dirpath, name)这段是“扫描 去重”的部分。file_hash用分块读取避免大文件一次性读进内存iter_files用生成器文件多的时候不会把整个列表撑在内存里。def parse_docx(path): from docx import Document doc Document(path) parts [p.text for p in doc.paragraphs if p.text.strip()] for table in doc.tables: for row in table.rows: cells [c.text.strip() for c in row.cells] if any(cells): parts.append( | .join(cells)) return \n.join(parts) def parse_pdf(path): import pdfplumber parts [] with pdfplumber.open(path) as pdf: for page in pdf.pages: text page.extract_text() or parts.append(text) return \n.join(parts) def parse_markdown(path): with open(path, r, encodingutf-8, errorsignore) as f: return f.read() def parse_file(path): ext Path(path).suffix.lower() if ext .docx: return parse_docx(path) if ext .pdf: return parse_pdf(path) if ext in (.md, .txt): return parse_markdown(path) return 解析部分的重点是每种格式单独一个函数出错时好定位。PDF 那里加or 是防止某页读到None直接抛异常。def split_text(text, sizeCHUNK_SIZE, overlapCHUNK_OVERLAP): text text.strip() chunks [] start 0 while start len(text): end start size piece text[start:end] if len(piece.strip()) 50: chunks.append(piece.strip()) start end - overlap return chunks def submit(chunk, meta): payload { content: chunk, title: meta[title], source: meta[source], } for attempt in range(MAX_RETRY): try: resp requests.post( API_URL, jsonpayload, timeoutREQUEST_TIMEOUT ) if resp.status_code 200: return True except requests.RequestException: pass time.sleep(1.5 * (attempt 1)) return Falsesubmit里的重试间隔用1.5 * (attempt 1)做退避第一次等 1.5 秒第二次 3 秒避免失败后立刻重试又撞上限流。主流程把上面几块串起来就行遍历文件、查哈希、解析、分块、逐块提交、记录成功哈希、最后统一落盘。整个脚本不长但每一块都是经过实际踩坑才定下来的。4. 跑起来之后踩的坑和排查记录4.1 编码问题导致中文变乱码第一个坑是编码。我有些 Markdown 文件是在不同系统上编辑过的编码不统一有一批是 GBK。直接用 utf-8 读会抛异常或者读出乱码。解决办法是读文件时加errorsignore兜底更稳妥的做法是用chardet先探测编码再读。乱码的另一个来源是 PDF 里的特殊字体某些 PDF 内嵌字体没有正确映射提取出来是一堆方框。这种没有万能解法只能靠人工识别后在日志里标记别硬导。实操心得导入前一定要先抽样检查。我的做法是脚本加一个--dry-run开关只打印分块结果不提交先看几十条输出对不对确认没问题再真正跑。4.2 大文件把内存吃满第二个坑是我一次性把整个 PDF 读进内存做处理处理一个几百页的文件时内存直接飙上去了。改造思路是流式处理解析按页来分块后立刻提交不留大对象。哈希计算也改成分块读。这个问题其实说明一个道理脚本在测试集上跑得好不代表在生产数据上稳。样本要挑大的、怪的测试文件里一定要放几个极端案例。4.3 重复导入与中断恢复第三个坑是重跑导致的重复。我第一次跑的时候中途因为网络问题挂了重启脚本后又从头开始结果前一半内容进了两次。修复方式就是前面说的哈希记录文件。每次成功导入一个文件的所有块后把文件哈希写进记录脚本启动先加载记录已经处理过的直接跳过。另外我加了一个“按文件粒度记进度”的做法一个文件的所有块全部成功才记哈希中途失败就整份重来。这样虽然会重复提交几个块但能保证不会出现“半个文件进了库”的状态。4.4 常见问题速查表现象可能原因处理方式中文乱码文件编码不是 utf-8用编码探测库或统一转码后再处理PDF 读出来空白扫描版/图片型 PDF日志标记人工补录不要指望纯文本库接口 429提交太快触发限流加节流降低并发分批 sleep重复入库没做幂等用内容哈希 成功记录文件脚本中途卡死单次请求无超时所有网络请求都设 timeout表格内容丢失只读了段落流单独遍历表格并拼接这几类问题我基本都遇到过表格里列的是最直接的解法照着改一般都能解决。5. 后续还能怎么扩展跑顺之后我又陆续加了几样东西。一个是增量导入配合文件修改时间只处理上次运行之后变动的文件避免每次全量扫。用 shell 入口加 crontab 定时跑晚上自动处理新增资料第二天来看结果就行。另一个是分类标签自动填充。我根据文件所在目录名给内容打上来源标签这样检索时能按目录过滤。规则很简单但省去了手工打标签的功夫。说句实在话这个脚本本身技术含量不算高真正花时间的是边界情况的处理——各种奇怪编码、扫描版 PDF、断网重试、重复导入。这些在写之前想不到写出来一跑全冒出来了。AI 在这中间的作用是帮我把模板代码快速铺出来但判断哪里会出问题、怎么设计容错还是得自己来。用它写这两百来行代码我大概省了一半的敲键盘时间但调试和验证的时间一分没少。如果让我给同样想做批量导入的人一句建议那就是先别急着写代码拿一份最小的接口规范和一份最有代表性的文件样本把“一条数据从头到尾怎么进去”走通一遍再考虑批量。走通单条批量只是加个循环单条没走通批量只会把问题放大。 Python批量导入知识库：多格式解析、文本分块与幂等重试
1. 拆解需求批量导入知识库到底难在哪1.1 一个真实场景引出的痛点我手上有一批资料散在十几个文件夹里产品文档是 Word技术白皮书是 PDF日常笔记和部分规范是 Markdown还有一些从网页复制下来存成 txt 的零碎内容。过去往知识库里放东西基本靠手工——打开文件、复制正文、粘贴到录入框、填标题、选标签一条一条来。几十个文件还能忍上百个文件就是纯体力活而且人一累就会漏、会错标题和内容对不上号的低级错误我都犯过好几次。这就是我想写个脚本的起因。目标很朴素给脚本一个目录它自己把里面所有受支持的文件读出来、清洗、切分然后按知识库接口要求批量提交我只在旁边看日志就行。说白了就是把重复劳动一次性交给机器。这里需要先明确一点“批量导入知识库”不是一个单一动作它其实是一条流水线文件扫描 → 格式解析 → 文本清洗 → 内容分块 → 元数据组装 → 接口调用 → 结果回执 → 失败重试。任何一环没处理好整体都会出问题。下面我按这条流水线的顺序把每一段的设计和取舍讲清楚。1.2 为什么不用现成工具非要自己写先说清楚我为什么没直接找个现成的导入工具。原因有三个都是实际用下来才体会到的。第一是格式太杂。市面上不少工具对单一格式支持很好但对“Word PDF Markdown 混着来”的支持参差不齐经常出现某类文件解析出来一堆乱码。第二是分块策略不能自己定。知识库检索效果好不好很大程度取决于文本怎么切。现成工具往往固定一个 chunk 大小改不了。我这边有些政策条款必须整段保留有些技术文档又适合切细一点固定策略满足不了。第三是接口不透明。我想知道每一条到底提交成功没有、失败原因是什么、能不能只重试失败的那部分。自己写日志我想怎么打就怎么打。所以说自建脚本的本质不是“重复造轮子”而是把导入过程变成一个可控、可观测、可重跑的流程。这是后面所有设计的出发点。1.3 整体思路和方案选型整体方案我定成这样用 Python 做主脚本负责遍历、解析、分块、提交用 shell 脚本做一个简单的入口包装方便定时执行或手动触发。选 Python 而不是纯 shell理由很直接——文本解析生态好。读 docx 有 python-docx读 PDF 有 pdfplumber字符串处理、正则、JSON 组装都是现成的用 shell 去解析二进制文档基本是自找麻烦。选 shell 做入口是因为它启动快、写起来简单配合 crontab 做增量导入很顺手。注意格式解析库只负责“把内容尽量取出来”不保证 100% 还原排版。表格、图文混排、多栏 PDF 是最容易出问题的三类内容后面会专门讲怎么处理。参数上我一开始定了几条硬指标单次运行要能处理上千个文件、失败率控制在个位数、单文件解析超时要能跳过而不是卡死整个任务。这些约束直接决定了后面分块大小、超时设置和重试逻辑的写法。2. 脚本骨架拆解与关键参数怎么定2.1 目录扫描与文件类型过滤第一步是找出“要处理哪些文件”。最省事的做法是递归遍历目录把扩展名在白名单里的文件收进列表。我用一个集合来做白名单这样判断速度更快也方便后续扩展。这里有个容易忽略的点遍历顺序会带来隐性重复。如果目录里有软链接指向同一份文件或者同名文件出现在多个子目录直接按路径导入会重复。我的做法是对每个文件先算一个内容哈希把哈希放进集合做去重已经在集合里的直接跳过。另一个细节是隐藏文件和临时文件要过滤掉。像~$开头的 Word 临时文件、.DS_Store这类系统文件混进去会导致解析报错。过滤逻辑很简单但在实际跑的时候能省掉一大堆莫名其妙的报错。2.2 Word、PDF、Markdown 三种格式怎么解析这三类文件的解析思路不一样我分开说。Word.docx用 python-docx 读段落。要注意它读的是“段落对象”表格里的内容默认不在段落流里需要单独遍历表格单元格。我的处理是把段落和表格都读出来按出现顺序拼接虽然会丢一点原始排版但正文完整性有保障。PDF这是最麻烦的。PDF 本质是排版描述不是结构化文档所以解析出来经常是“一行被拆成好几段”。我的做法是用 pdfplumber 按页提取文字然后按行合并再根据空行和标点做二次分段。扫描版 PDF图片型纯文本库读不出来这种情况我会在日志里标记出来靠人工补不硬凑。Markdown最友好直接读文本就行。我会顺手把代码块、超链接的冗余语法清理掉但保留标题层级因为标题对分块很有用——它是天然的段落边界。实操心得解析库的版本要固定。我在不同机器上跑pdfplumber 换了个小版本同一份 PDF 出来的换行位置就变了。生产用的脚本依赖版本最好锁死别用“随最新版”。2.3 文本分块chunk 大小和重叠怎么定分块是决定检索质量的关键。切太大一段里混了好几个主题检索时命中不精准切太小一句话被拆散语义不完整。我的经验值是这样定的参数取值说明chunk_size500 字符中文场景下大约 300~400 字语义较完整chunk_overlap80 字符给相邻块留一点重叠避免边界信息丢失最小块长度50 字符低于这个长度的碎片直接丢弃或合并到上一块为什么要留重叠因为句子被硬切断时关键词可能正好落在切口上。留一段重叠检索时至少有一个块能包含完整语义。重叠太大又会导致冗余80 字符是我反复试下来比较平衡的值。切分边界上我优先按标题、空行、句号这类“硬边界”切实在没有边界才按字符数硬切。这个逻辑在代码里就是先找分隔符找不到再退化到长度切分。2.4 接口调用与失败重试提交环节要解决三个问题限流、超时、幂等。限流方面知识库接口一般都有速率限制短时间内大量请求会被拒。我在脚本里加了一个简单的节流每提交 N 条 sleep 一下或者用令牌桶控制速率。超时方面每条请求都要设超时时间网络抖动时不能让整个脚本挂住。超时后不直接判失败而是放进重试队列。幂等方面这条最重要。脚本可能会被重跑比如中途挂了再执行如果没有幂等控制同一份内容会重复入库。我的做法是给每个文件算哈希把“已成功导入的哈希”记到一个本地记录文件里重跑时先查记录已经成功的跳过。注意幂等靠的是“内容哈希 成功记录”不是靠文件名。文件改名后内容没变仍然应该被识别为已导入。3. 拿 AI 手搓脚本的真实过程3.1 怎么给 AI 描述需求才有效这是我这次最想分享的部分。让 AI 写代码效果好坏九成取决于你怎么描述。我踩过的坑是一开始只丢一句“帮我写个批量导入知识库的脚本”结果它给我的代码假设了一个根本不存在的接口格式还用了我看不懂的库。后来我改成“分块描述需求”效果好很多。具体是这样拆的先告诉它输入是什么一个目录里面有 docx、pdf、markdown 三类文件。再告诉它输出接口长什么样接口地址、请求方法、请求体的字段名、认证方式。然后告诉它关键约束chunk 大小、超时、去重、日志格式。最后要求它按函数拆分每个函数只干一件事不要写成一个巨型 main。把接口字段名、请求示例清楚地写进提示词AI 生成的代码可用率会大幅提升。因为它是照着你的规格在填空而不是凭空猜。3.2 分阶段生成别一次性要完整代码一次性要一整套代码出来的东西通常“看起来对、跑起来错”。我的做法是分阶段第一阶段只要文件遍历和解析让它先能打印出每个文件读到了多少字。跑通之后再进入下一阶段。第二阶段只要分块逻辑我拿几份典型文件验证切分结果合不合理。第三阶段才是接口提交和重试。这时候前面解析、分块已经稳定出问题就只可能出在网络和接口部分排查范围小很多。这个“小步验证”的思路是我觉得比代码本身更值钱的经验。AI 写代码快但你必须有能力分阶段验收否则就是在赌。3.3 完整脚本拆解与逐段注释下面是我最终的脚本骨架去掉业务细节后大概是这个结构import os import json import time import hashlib import requests from pathlib import Path SUPPORTED_EXT {.docx, .pdf, .md, .txt} CHUNK_SIZE 500 CHUNK_OVERLAP 80 RECORD_FILE imported_hashes.json API_URL https://your-kb-endpoint/api/documents REQUEST_TIMEOUT 15 MAX_RETRY 3 def load_imported(): if os.path.exists(RECORD_FILE): with open(RECORD_FILE, r, encodingutf-8) as f: return set(json.load(f)) return set() def save_imported(hashes): with open(RECORD_FILE, w, encodingutf-8) as f: json.dump(list(hashes), f, ensure_asciiFalse) def file_hash(path): h hashlib.md5() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() def iter_files(root): for dirpath, _, filenames in os.walk(root): for name in filenames: if name.startswith(~$) or name.startswith(.): continue if Path(name).suffix.lower() in SUPPORTED_EXT: yield os.path.join(dirpath, name)这段是“扫描 去重”的部分。file_hash用分块读取避免大文件一次性读进内存iter_files用生成器文件多的时候不会把整个列表撑在内存里。def parse_docx(path): from docx import Document doc Document(path) parts [p.text for p in doc.paragraphs if p.text.strip()] for table in doc.tables: for row in table.rows: cells [c.text.strip() for c in row.cells] if any(cells): parts.append( | .join(cells)) return \n.join(parts) def parse_pdf(path): import pdfplumber parts [] with pdfplumber.open(path) as pdf: for page in pdf.pages: text page.extract_text() or parts.append(text) return \n.join(parts) def parse_markdown(path): with open(path, r, encodingutf-8, errorsignore) as f: return f.read() def parse_file(path): ext Path(path).suffix.lower() if ext .docx: return parse_docx(path) if ext .pdf: return parse_pdf(path) if ext in (.md, .txt): return parse_markdown(path) return 解析部分的重点是每种格式单独一个函数出错时好定位。PDF 那里加or 是防止某页读到None直接抛异常。def split_text(text, sizeCHUNK_SIZE, overlapCHUNK_OVERLAP): text text.strip() chunks [] start 0 while start len(text): end start size piece text[start:end] if len(piece.strip()) 50: chunks.append(piece.strip()) start end - overlap return chunks def submit(chunk, meta): payload { content: chunk, title: meta[title], source: meta[source], } for attempt in range(MAX_RETRY): try: resp requests.post( API_URL, jsonpayload, timeoutREQUEST_TIMEOUT ) if resp.status_code 200: return True except requests.RequestException: pass time.sleep(1.5 * (attempt 1)) return Falsesubmit里的重试间隔用1.5 * (attempt 1)做退避第一次等 1.5 秒第二次 3 秒避免失败后立刻重试又撞上限流。主流程把上面几块串起来就行遍历文件、查哈希、解析、分块、逐块提交、记录成功哈希、最后统一落盘。整个脚本不长但每一块都是经过实际踩坑才定下来的。4. 跑起来之后踩的坑和排查记录4.1 编码问题导致中文变乱码第一个坑是编码。我有些 Markdown 文件是在不同系统上编辑过的编码不统一有一批是 GBK。直接用 utf-8 读会抛异常或者读出乱码。解决办法是读文件时加errorsignore兜底更稳妥的做法是用chardet先探测编码再读。乱码的另一个来源是 PDF 里的特殊字体某些 PDF 内嵌字体没有正确映射提取出来是一堆方框。这种没有万能解法只能靠人工识别后在日志里标记别硬导。实操心得导入前一定要先抽样检查。我的做法是脚本加一个--dry-run开关只打印分块结果不提交先看几十条输出对不对确认没问题再真正跑。4.2 大文件把内存吃满第二个坑是我一次性把整个 PDF 读进内存做处理处理一个几百页的文件时内存直接飙上去了。改造思路是流式处理解析按页来分块后立刻提交不留大对象。哈希计算也改成分块读。这个问题其实说明一个道理脚本在测试集上跑得好不代表在生产数据上稳。样本要挑大的、怪的测试文件里一定要放几个极端案例。4.3 重复导入与中断恢复第三个坑是重跑导致的重复。我第一次跑的时候中途因为网络问题挂了重启脚本后又从头开始结果前一半内容进了两次。修复方式就是前面说的哈希记录文件。每次成功导入一个文件的所有块后把文件哈希写进记录脚本启动先加载记录已经处理过的直接跳过。另外我加了一个“按文件粒度记进度”的做法一个文件的所有块全部成功才记哈希中途失败就整份重来。这样虽然会重复提交几个块但能保证不会出现“半个文件进了库”的状态。4.4 常见问题速查表现象可能原因处理方式中文乱码文件编码不是 utf-8用编码探测库或统一转码后再处理PDF 读出来空白扫描版/图片型 PDF日志标记人工补录不要指望纯文本库接口 429提交太快触发限流加节流降低并发分批 sleep重复入库没做幂等用内容哈希 成功记录文件脚本中途卡死单次请求无超时所有网络请求都设 timeout表格内容丢失只读了段落流单独遍历表格并拼接这几类问题我基本都遇到过表格里列的是最直接的解法照着改一般都能解决。5. 后续还能怎么扩展跑顺之后我又陆续加了几样东西。一个是增量导入配合文件修改时间只处理上次运行之后变动的文件避免每次全量扫。用 shell 入口加 crontab 定时跑晚上自动处理新增资料第二天来看结果就行。另一个是分类标签自动填充。我根据文件所在目录名给内容打上来源标签这样检索时能按目录过滤。规则很简单但省去了手工打标签的功夫。说句实在话这个脚本本身技术含量不算高真正花时间的是边界情况的处理——各种奇怪编码、扫描版 PDF、断网重试、重复导入。这些在写之前想不到写出来一跑全冒出来了。AI 在这中间的作用是帮我把模板代码快速铺出来但判断哪里会出问题、怎么设计容错还是得自己来。用它写这两百来行代码我大概省了一半的敲键盘时间但调试和验证的时间一分没少。如果让我给同样想做批量导入的人一句建议那就是先别急着写代码拿一份最小的接口规范和一份最有代表性的文件样本把“一条数据从头到尾怎么进去”走通一遍再考虑批量。走通单条批量只是加个循环单条没走通批量只会把问题放大。