上个月我把攒了三年的技术笔记做了一次彻底归档一千两百多份 Markdown、PDF 和 Word 摊在硬盘里想全部塞进自建的知识库做语义检索。手动拖拽了十几份之后我就放弃了——重复劳动、进度不可见、中途断一次还得从头再来最要命的是我连自己拖到哪一份都记不清。于是我干了件之前一直没敢干的事逻辑自己定代码一行不写全程让 AI 把脚本吐出来。这就是标题里说的「第一个纯 AI 手搓的脚本程序」它只做一件事批量导入知识库。这篇东西写给两类人一类是手里躺着几百上千份文档、想搭 RAG 知识库却被导入环节卡住的人另一类是想用 AI 写脚本但不知道该怎么开口、怎么验收的人。我会把完整的目录结构、参数计算、代码实现、断点续传思路还有调试时踩过的坑全部摊开讲你可以直接照着复现。1. 需求拆解与整体设计思路1.1 手动导入到底卡在哪先说清楚我为什么非要写脚本。知识库平台一般都会提供网页端的拖拽上传传十个八个文件确实方便但文件量一上来问题就成倍放大。我整理了一份自己的痛点清单你可以对照看看是不是也中招。第一是重复动作消耗注意力。每传一份都要点选文件、等解析、看状态、确认成功动作本身不难但重复两百次之后人的判断力会明显下降很容易漏传或者重复传。第二是进度不可回溯。网页端那个上传列表刷新一下就没了你不知道哪几份成功了、哪几份失败了只能靠记忆而记忆在几百个文件面前基本等于零。第三是解析失败没有反馈。扫描版 PDF 提取不出文字、加密文档读不出内容、编码不对导致乱码这些问题在网页端往往只是一句笼统的失败提示你需要逐个排查。第四是无法增量更新。我每周都会往笔记目录里加新文件也可能修改老文件网页端做不到「只上传变化的那些」每次都得全量重来。这些问题归结起来其实就一句话批量导入知识库这件事本质上是一个可自动化、可记录、可重试的数据管道任务而不是一个人工操作任务。但凡一个流程具备「输入确定、步骤固定、结果可校验」这三个特征就值得写成脚本。这也是我下定决心动手的根本原因。1.2 方案选型为什么是脚本而不是别的东西动手之前我评估过三条路。第一条路是直接用知识库平台自带的批量上传功能有些平台确实支持一次选多个文件但它们的共性问题是不支持自定义分块规则、不支持增量、失败重试也要手动点。第二条路是用现成的开源同步工具比如一些笔记软件自带的同步插件或者社区写好的导入器好处是开箱即用坏处是约束太死——你的目录结构、文件命名、元数据格式必须完全贴合它的假设一旦不匹配就得改自己的习惯这个代价我不愿意付。第三条路就是自己写脚本我最后选了它理由有三条。完全掌控分块策略。不同文档类型的最优切分方式差异很大结构化笔记适合按标题切产品文档适合按段落切PDF 转出来的文本则必须先做清洗再切通用工具很难覆盖这些细节。天然支持增量与断点续传。脚本可以把每个文件的内容哈希记下来下次运行时只处理变化的文件这是网页端永远做不到的。可复用、可迁移。脚本写一次换个知识库平台只需要改一个 API 地址和参数格式逻辑骨架完全不动边际成本极低。至于「用 AI 写」这个选择其实是因为我对 Python 的熟练度只到能看懂、能改错的水平从零手写一个带重试和状态管理的完整脚本我大概要花两三个晚上而且中间会卡在细节上。让 AI 出第一版我负责审和调这个分工效率高得多。1.3 整体数据流长什么样在写第一行代码之前我强制自己先用自然语言把整个流程描述了一遍。这一步非常关键因为AI 拿到的需求描述越接近伪代码产出的代码质量就越高。我当时写给自己看的需求是这样的扫描指定目录下所有 Markdown、TXT、PDF、DOCX 文件对每个文件计算内容哈希如果这个哈希在上次运行记录里已经存在就跳过。没处理过的文件先解析成纯文本做基础清洗判断内容是否过短然后按分块规则切分或直接整篇提交给知识库接口。接口调用失败要自动重试三次采用指数退避。每个文件的处理结果写入状态文件包括文件名、哈希、返回的文档 ID、时间戳。整个过程要有日志日志同时输出到控制台和文件。把这段话拆开其实是五个模块文件扫描器、内容解析器、清洗与分块器、上传客户端、状态管理器。这五个模块之间是流水线关系前一个的输出是后一个的输入中间任何一环失败都不能影响其他文件的处理。这个「模块化 单文件隔离」的设计决策非常关键它意味着一个文件解析崩溃不会导致整个批次中断你只需要看日志定位到那个文件单独排查。很多新手写批处理脚本习惯把逻辑写成一大坨 for 循环中间一个异常就全盘退出这是最常见的坑。2. 环境准备与关键依赖2.1 Python 版本与依赖清单我用的 Python 3.10实际上 3.8 以上都能跑。依赖库只装了四个全部是轻量级的没有引入任何重型框架。选依赖的原则是能用标准库就用标准库必须装第三方库就选维护活跃、依赖少的。pip install requests pyyaml pypdf python-docx逐个说明为什么要装它们。requests负责发 HTTP 请求比标准库的 urllib 好用太多尤其是连接复用和超时控制。pyyaml用来读配置文件把 API 密钥、目录路径这些容易变动的东西从代码里剥出来避免硬编码。pypdf负责解析 PDF注意它只能提取文本型 PDF扫描件提取不出来这一点后面会专门讲。python-docx负责解析 Word 文档它读的是.docx格式老的.doc读不了。提示装依赖的时候建议用虚拟环境不要直接装在系统 Python 里。我见过太多人因为系统环境被污染导致别的项目跑不起来排查半天才发现是版本冲突。如果你的文档里还有.doc、.pptx、.xlsx这些格式我的建议是先手动转换成支持的格式再导入不要为了省事在脚本里装一堆转换库。转换本身涉及办公软件的格式兼容问题坑比导入环节多得多不值得把复杂度堆在一个脚本里。2.2 知识库接口的准备与鉴权不同知识库平台的接口长得不一样但套路基本一致一个数据集 ID 加一个 API Key通过 Bearer 方式鉴权POST 一个 JSON 上去。我这边用的是支持「按文本创建文档」接口的 RAG 知识库接口路径形如/v1/datasets/{dataset_id}/document/create-by-text。这里有个容易忽略的点API Key 的权限范围要和操作匹配。有些平台的密钥分「应用密钥」和「数据集密钥」前者是用来调用对话接口的后者才有权限往知识库里写数据。我第一次调试时拿错了密钥接口一直返回 401查了半小时才反应过来。所以拿到密钥之后先不要写代码用curl手动打一发请求验证通路这五分钟的投入能省你一小时的排查。curl -X POST https://your-kb-host/v1/datasets/YOUR_DATASET_ID/document/create-by-text \ -H Authorization: Bearer YOUR_DATASET_KEY \ -H Content-Type: application/json \ -d {name:联通测试,text:这是一段测试内容。,indexing_technique:high_quality}这条命令能返回 200 并且看到文档 ID才说明鉴权和路径都没问题。返回 404 一般是数据集 ID 写错了返回 401 是密钥问题返回 400 多半是请求体字段名不对。先用 curl 打通再用代码封装这是我调试所有 HTTP 接口的固定顺序。2.3 目录结构与配置文件设计目录结构我按「代码、配置、数据、状态、日志」五分离的原则来组织这样后期维护和迁移都清爽。kb-importer/ ├── importer.py # 主程序 ├── config.yaml # 配置接口地址、密钥、源目录 ├── requirements.txt # 依赖清单 ├── docs/ # 待导入的文档源目录 │ ├── 网络/ │ ├── 数据库/ │ └── 杂项/ ├── state/ │ └── uploaded.json # 处理状态记录 └── run.log # 运行日志配置文件把易变项全部抽离出来代码里不出现任何魔法值。base_url: https://your-kb-host/v1 dataset_id: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx api_key: dataset-xxxxxxxxxxxxxxxx source_dir: ./docs interval: 1.2 # 每个文件之间的间隔秒数 batch_size: 5 # 预留并发批大小interval这个参数看着不起眼实际很重要。知识库在做文档索引时要调用嵌入模型这是一个重计算的过程如果你以每秒几十个文件的速度狂发请求平台侧很容易排队超时或者直接给你限流。我实测下来 1.2 秒的间隔是比较稳的一万字左右的文档每次索引大约需要 1 到 2 秒这个节奏基本能跑满而不会触发 429。3. 核心环节的完整实现3.1 文件扫描别小看遍历这件事扫描看起来是最简单的一步其实藏着几个必须处理的边界情况。第一是隐藏文件和临时文件比如 Office 打开文档时会生成~$xxx.docx这样的锁文件扫进去会解析失败。第二是递归层级用rglob可以递归所有子目录但要防止符号链接造成无限递归。第三是排序稳定性一定要排序否则每次运行顺序都不一样日志没法对照。SUPPORTED {.md, .txt, .pdf, .docx} def scan_files(root: Path): root root.resolve() for p in sorted(root.rglob(*)): if not p.is_file(): continue if p.suffix.lower() not in SUPPORTED: continue if p.name.startswith(~$) or p.name.startswith(.): continue yield p这段代码里sorted()的作用比看起来大。它保证了处理顺序稳定当你在日志里看到第 158 个文件失败时下次重跑还能定位到同一个位置。另外p.suffix.lower()是必要的Windows 上文件的扩展名大小写经常不统一.PDF和.pdf都要能识别。3.2 内容解析不同格式的取文本姿势解析环节是整个脚本里最容易翻车的地方因为文档格式的坑太多了。我按格式逐个说明。Markdown 和 TXT 最简单直接读就行但必须指定编码并且允许错误忽略因为很多从网页复制来的文本会混入奇怪的字符。def read_text(path: Path) - str: suffix path.suffix.lower() if suffix in (.md, .txt): return path.read_text(encodingutf-8, errorsignore) if suffix .pdf: from pypdf import PdfReader reader PdfReader(str(path)) return \n.join(page.extract_text() or for page in reader.pages) if suffix .docx: from docx import Document doc Document(str(path)) return \n.join(p.text for p in doc.paragraphs) raise ValueError(f不支持的类型: {suffix})PDF 解析这块要特别提醒pypdf提取的是 PDF 里内嵌的文本层如果你的 PDF 是扫描件或者图片导出的提取结果会是空字符串。判断方法很简单解析完之后看文本长度如果短于 50 个字符基本可以判定是扫描件这种情况只能上 OCR而我个人建议是单独用 OCR 工具转成 Markdown 再放进源目录不要把这个环节塞进脚本。Word 文档解析还有个隐藏问题python-docx只读段落读不到表格里的内容。如果你的文档里有大量表格数据直接导入会丢信息。解决办法是同时遍历 tables 对象把表格拼成文本。def read_docx_full(path: Path) - str: from docx import Document doc Document(str(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)这个细节是我实际踩坑后才补上的。最早导入的一批产品文档表格里的参数配置全部丢失检索的时候怎么都查不到回头一查才发现是这个原因。3.3 清洗与分块决定检索质量的胜负手清洗的力度需要拿捏。洗得太轻乱码和多余空白会污染向量洗得太重会误删有效内容。我的做法是只做三件确定性的事把全角空格和不换行空格替换成普通空格去掉行尾多余空白把三个以上连续换行压缩成两个。import re RE_TRAIL_WS re.compile(r[ \t]\n) RE_MULTI_BLANK re.compile(r\n{3,}) def clean(text: str) - str: text text.replace(\u3000, ).replace(\xa0, ) text RE_TRAIL_WS.sub(\n, text) text RE_MULTI_BLANK.sub(\n\n, text) return text.strip()注意我没有删除 URL。很多教程默认会把链接过滤掉理由是链接没有语义价值。但我的文档里大量存在「参考资料https://xxx」这种引用链接本身就是信息的一部分删掉反而让上下文断裂。这个取舍取决于你的文档类型做技术文档的就保留做通用问答的可以删。分块参数是决定影响检索效果的核心我把当时的计算过程记录一下。假设你的文档平均 8000 字中文知识库的分块上限是 500 个 token。中文里 1 个汉字大约对应 1 个 token那么 8000 字会被切成大约 16 到 20 块。块与块之间要设置重叠重叠量一般取分块大小的 10% 到 20%也就是 50 到 100 个 token目的是避免一句话被硬生生截断导致语义不完整。def estimate_tokens(text: str) - int: cjk sum(1 for ch in text if \u4e00 ch \u9fff) others len(text) - cjk return int(cjk * 1.0 others / 4)分块策略我采用的是「先按标题切超长了再按段落切」的两级方案。理由很实在技术笔记天然带有标题层级一个##或者###下的内容通常就是一个完整的知识单元按这个边界切出来的块语义最完整检索命中率明显高于机械按字数切。def split_by_headers(text: str, max_tokens: int 500): blocks re.split(r\n(?#{1,4}\s), text) chunks [] for block in blocks: if estimate_tokens(block) max_tokens: if block.strip(): chunks.append(block.strip()) continue buf for para in [p for p in block.split(\n\n) if p.strip()]: if estimate_tokens(buf para) max_tokens and buf: chunks.append(buf.strip()) buf para else: buf buf \n\n para if buf else para if buf.strip(): chunks.append(buf.strip()) return [c for c in chunks if len(c) 20]最后那个len(c) 20是过滤规则把「相关阅读」「更新日志」这种只有几个字的碎片丢掉。实测下来这些碎片块不仅浪费索引额度还会在检索时被误召回拉低结果质量。注意如果你的知识库平台自己支持服务端分块很多平台在接口里提供 segmentation 参数那就没必要在本地再切一遍两边都切会导致语义被破坏两次。我当时的选择是本地只做解析和清洗把干净的全文传上去分块交给平台按参数执行。3.4 上传客户端重试机制是必需品接口调用这块最核心的设计是重试。批量任务里出现偶发网络抖动几乎是必然的如果一次失败就放弃你会在日志里看到大量无意义的失败记录。我的重试策略是指数退避第一次失败等 2 秒第二次等 4 秒第三次等 8 秒。def upload_text(session, cfg, name, text, retries3): url f{cfg[base_url].rstrip(/)}/datasets/{cfg[dataset_id]}/document/create-by-text payload { name: name, text: text, indexing_technique: high_quality, process_rule: { mode: custom, rules: { pre_processing_rules: [ {id: remove_extra_spaces, enabled: True}, {id: remove_urls_emails, enabled: False}, ], segmentation: { separator: \n\n, max_tokens: 500, chunk_overlap: 50, }, }, }, } headers { Authorization: fBearer {cfg[api_key]}, Content-Type: application/json, } delay 2.0 for attempt in range(1, retries 1): try: resp session.post(url, jsonpayload, headersheaders, timeout60) if resp.status_code in (200, 201): return resp.json() if resp.status_code in (429, 500, 502, 503, 504): time.sleep(delay) delay * 2 continue raise RuntimeError(fHTTP {resp.status_code}: {resp.text[:200]}) except requests.RequestException as exc: time.sleep(delay) delay * 2 raise RuntimeError(f上传失败: {name})这段代码里有几个设计决策值得展开说。只对特定状态码重试429 是限流、5xx 是服务端问题这些是重试有意义的而 400 参数错误、401 鉴权错误重试一百次也没用直接抛出更快。超时设为 60 秒因为大文档的服务端索引时间可能长设太短会频繁触发假失败。用 Session 复用连接比每次新建连接快不少还能减少 TCP 握手开销。name字段的处理也有讲究。我用的是相对路径转换来的名字比如网络_HTTP缓存机制.md因为知识库列表里只显示一个名字如果把所有文件都叫index.md你根本分不清哪个是哪个。3.5 断点续传让脚本可以随时中断状态管理是我认为这个脚本最有价值的部分也是新手最容易忽略的部分。核心思路很简单用文件内容的哈希值作为指纹记录哪些文件已经处理过。下次运行时只要指纹没变就跳过指纹变了就重新上传。def fingerprint(path: Path) - str: digest hashlib.sha256() with open(path, rb) as f: for block in iter(lambda: f.read(1 20), b): digest.update(block) return digest.hexdigest()[:16]这里用sha256分块读取而不是一次性读完整个文件是为了处理大文件时不占内存。截取前 16 位足够了碰撞概率可以忽略。状态文件我用「写临时文件再原子替换」的方式保存避免写到一半程序被杀导致文件损坏。def save_state(state): STATE_FILE.parent.mkdir(parentsTrue, exist_okTrue) tmp STATE_FILE.with_suffix(.tmp) tmp.write_text(json.dumps(state, ensure_asciiFalse, indent2), encodingutf-8) tmp.replace(STATE_FILE)主循环把上面所有模块串起来每个文件的处理都包在 try 里无论成功失败都保存状态这样即使中途强制退出下次也能接上。def main(): cfg load_config() state load_state() session requests.Session() source Path(cfg[source_dir]).expanduser().resolve() files list(scan_files(source)) ok skip fail 0 for idx, path in enumerate(files, 1): rel str(path.relative_to(source)) fp fingerprint(path) record state[done].get(rel) if record and record.get(fp) fp: skip 1 continue try: text clean(read_text(path)) if estimate_tokens(text) 30: skip 1 continue result upload_text(session, cfg, rel.replace(os.sep, _), text) doc_id (result.get(document) or {}).get(id, ) state[done][rel] {fp: fp, doc_id: doc_id, ts: int(time.time())} ok 1 log.info([%d/%d] 成功: %s - %s, idx, len(files), rel, doc_id) except Exception as exc: fail 1 log.error([%d/%d] 失败: %s | %s, idx, len(files), rel, exc) finally: save_state(state) time.sleep(float(cfg.get(interval, 1.0))) log.info(完成: 成功 %d, 跳过 %d, 失败 %d, ok, skip, fail)跑完之后你会看到这样的输出一眼就能看出整体情况2025-06-12 21:14:03 | INFO | 共扫描到 1284 个文件 2025-06-12 21:14:03 | INFO | [1/1284] 成功: 网络_HTTP缓存机制.md - 9f3c... 2025-06-12 21:14:06 | INFO | [2/1284] 成功: 网络_TCP三次握手.md - 71ab... 2025-06-12 21:14:08 | WARN | 临时错误 429第 1 次重试 2025-06-12 21:14:19 | INFO | [3/1284] 成功: 数据库_索引原理.pdf - 3e2d... 2025-06-12 21:31:47 | INFO | 完成: 成功 1247, 跳过 32, 失败 5失败的那 5 个我逐个看了日志3 个是扫描版 PDF 提取不出文本2 个是超长文档超过了平台的单文档大小限制。前者手动 OCR 后重跑后者拆成两篇再传半小时内全部解决。4. 用 AI 写脚本的实操方法论4.1 怎么给 AI 描述需求才有效这次经历让我最大的收获不是脚本本身而是搞明白了一件事AI 写代码的质量八成取决于你描述需求的质量而不是模型本身有多强。我第一次提需求的时候只说了一句「帮我写个 Python 脚本把文件夹里的文档批量上传到知识库」AI 给我的版本一坨没有任何异常处理没有状态记录连文件类型判断都是硬编码的。后来我改了策略把需求拆成三段来描述。第一段描述数据流也就是从哪来、经过什么处理、到哪去。第二段描述约束条件包括只支持哪几种格式、遇到什么情况跳过、失败重试几次、用什么策略退避。第三段描述输出产物包括要生成什么文件、日志格式长什么样、状态文件里存哪些字段。我实际用的提示词大致是这样组织的我需要一个批量导入知识库的脚本。数据源是本地目录递归扫描.md、.txt、.pdf、.docx四类文件跳过隐藏文件和~$开头的临时文件。处理流程是读取内容、清洗文本、判断长度、通过 HTTP POST 提交到接口。约束条件每个文件用 sha256 哈希做去重已处理且未修改的跳过网络失败或 429、5xx 状态码重试三次采用指数退避初始 2 秒每个文件处理完都要落盘状态支持随时中断。输出要求控制台和文件双写日志状态用 JSON 存包含文件名、哈希、返回的文档 ID 和时间戳。请先给出模块划分再逐个模块给代码。用这种方式提需求AI 给出的第一版就已经能用七八成剩下的都是细节打磨。关键区别在于前者是让 AI 猜你要什么后者是你已经把设计做完了、只让 AI 负责翻译成代码。这个分工对新手尤其友好因为设计能力是可以凭空想象的而语法细节才是真正需要查资料的部分。4.2 让 AI 帮你调试而不是重写代码跑起来一定会报错这时候很多人习惯把整个报错贴回去说「跑不通帮我改」AI 会给你一版重写的代码但改了什么你完全不知道改完可能引入新的问题。我的做法是只贴报错和相关的那一段代码并且明确要求解释原因再给修复方案。比如我当时遇到的FileHandler报父目录不存在的问题我没有直接让它改而是这样问的运行时报错FileNotFoundError: [Errno 2] No such file or directory: run.log。我的目录结构里日志文件在脚本同级目录代码是logging.FileHandler(LOG_FILE)。请解释为什么找不到并给出最小修改方案不要重写整个日志配置。它给出的解释是 FileHandler 不会自动创建父目录给出的修复是在配置前加一句mkdir。修改只有一行我一眼就看懂了也知道以后遇到类似问题该怎么处理。这种「解释优先」的提问方式长期收益远大于直接要答案因为你是在积累调试直觉而不是收集代码片段。4.3 AI 生成的代码必须审哪些地方我对 AI 生成的代码有一个固定审查清单每次都会过一遍因为这个清单里的问题几乎每次都中招。第一项是状态码判断AI 特别喜欢只判断status_code 200但创建接口常常返回 201这一条不补上会导致明明成功却记为失败。第二项是异常捕获范围AI 有时候用裸except:这会连KeyboardInterrupt都吃掉你按 CtrlC 都停不下来必须改成except Exception。第三项是资源释放打开文件、创建 Session 的地方有没有正确关闭。第四项是边界条件空文件、超大文件、特殊字符文件名这些 AI 默认不会处理得主动要求。最后一项也是最重要的API 密钥绝对不能硬编码在代码里。AI 给的第一版示例经常是api_key sk-xxx这种形式方便演示但极其危险一旦你把代码传到任何公开地方密钥就泄露了。我的做法是全部走配置文件并且把config.yaml加进.gitignore仓库里只保留一个config.example.yaml。5. 常见问题排查与优化经验5.1 高频报错速查表跑这一千多个文件的过程中我记录下了遇到的所有问题整理成表你遇到类似症状可以直接对照。报错或症状根本原因解决办法HTTP 401密钥类型不对或已失效用 curl 单独验证确认使用的是数据集密钥HTTP 404数据集 ID 错误或路径拼错对照平台文档核对接口路径与 IDHTTP 400请求体字段名不匹配打印 payload 逐字段核对文档HTTP 429请求频率过高被限流增大 interval 参数检查是否有并发解析结果为空扫描版 PDF 无文本层单独 OCR 转 Markdown 后再导入中文变问号文件编码是 GBK读取时显式指定或做编码探测文档过大被拒超过平台单文档上限按标题拆成多篇分别导入状态文件损坏写入中途被中断改为临时文件原子替换表格内容丢失docx 段落读不到表格额外遍历 tables 对象拼接这张表里的每一条我都是真金白银踩出来的。尤其是「中文变问号」那条我有一批早期的博客备份是 GBK 编码直接按 UTF-8 读出来全是乱码但程序不报错只是把乱码传进了知识库直到检索时发现结果全是问号才反应过来。编码问题最阴险的地方就在于它静默失败所以我在清洗环节加了一个检查如果文本中连续出现大量\ufffd替换字符就记一条警告日志。5.2 几个反直觉的实操心得第一个心得是不要追求一次全量跑完。我最开始想一口气把 1284 个文件全传上去结果跑到第 300 个的时候平台开始大量返回 429整个任务效率反而更低。后来我改成先跑 50 个试试水确认接口稳定、参数合理之后再放开跑。先小批量验证再全量执行的节奏在任何批量任务里都成立。第二个心得是日志比调试器有用。脚本跑一两个小时你不可能盯着调试器看日志是你唯一的观测窗口。我在日志里记录的信息包括序号、总数、文件名、文档 ID这几个字段能让我随时知道进度在哪、哪个文件出问题。如果你发现某个文件反复失败直接 grep 文件名就能看到它每次的报错内容。第三个心得是分块大小要跟着文档类型调不要一刀切。我一开始所有文档都用 500 token 的分块结果发现 API 文档这种「一个接口一段」的内容被切得七零八落检索时经常只召回半段说明。后来我给这类文档单独设了 800 token效果明显好转。如果你的知识库平台支持按数据集设置分块规则最好的做法是按内容类型建多个数据集每个用不同的分块参数。第四个心得是导入完成不等于检索可用。文档上传成功后平台还需要时间做向量化索引尤其是文档多的时候可能要等几分钟到十几分钟。我一开始传完立刻去检索结果什么都搜不到以为是导入失败白排查了半天。判断索引是否完成要看平台的状态字段通常在文档列表里会显示「索引中」和「可用」两种状态。5.3 性能优化与后续扩展方向性能这块我做了两处优化。一是把文件哈希计算和上传解耦扫描阶段先把所有文件的指纹算出来这样即使中途中断重启时也不需要重新计算全部哈希。二是状态文件改为批量落盘原来的实现是每处理一个文件就写一次状态文件多的时候磁盘 IO 反而成了瓶颈改成每 10 个文件落一次性能提升明显代价是最坏情况下丢失 10 条记录重跑一遍也就几分钟。接下来我准备做的扩展有这么几个。第一是元数据注入把文件的目录路径、修改时间、标签一起写进文档描述里这样检索时可以做条件过滤比如只在「数据库」目录下找。第二是多知识库分发同一批文档按分类自动路由到不同的数据集技术类进技术库产品类进产品库避免混在一起互相干扰。第三是定时增量同步用系统的计划任务每天跑一次只处理有变动的文件这样就彻底告别手动导入了。我在实际使用中的体会是AI 写脚本这件事真正降低的门槛不是「写代码」而是「把想法翻译成可执行逻辑」这一步。以前我脑子里有清晰的需求但卡在不知道用什么库、怎么写语法现在这个环节被抹平了我只需要把设计想清楚。整个脚本从提需求到跑通我花了大概两个晚上其中真正的编码时间不到三分之一剩下全是在想清楚每个环节的边界条件。如果你手里也有一堆文档等着进知识库别急着打开浏览器拖文件花一个晚上把脚本搭出来后面省下的时间会远超这个投入。
用AI写Python脚本：RAG知识库批量导入与断点续传实战
上个月我把攒了三年的技术笔记做了一次彻底归档一千两百多份 Markdown、PDF 和 Word 摊在硬盘里想全部塞进自建的知识库做语义检索。手动拖拽了十几份之后我就放弃了——重复劳动、进度不可见、中途断一次还得从头再来最要命的是我连自己拖到哪一份都记不清。于是我干了件之前一直没敢干的事逻辑自己定代码一行不写全程让 AI 把脚本吐出来。这就是标题里说的「第一个纯 AI 手搓的脚本程序」它只做一件事批量导入知识库。这篇东西写给两类人一类是手里躺着几百上千份文档、想搭 RAG 知识库却被导入环节卡住的人另一类是想用 AI 写脚本但不知道该怎么开口、怎么验收的人。我会把完整的目录结构、参数计算、代码实现、断点续传思路还有调试时踩过的坑全部摊开讲你可以直接照着复现。1. 需求拆解与整体设计思路1.1 手动导入到底卡在哪先说清楚我为什么非要写脚本。知识库平台一般都会提供网页端的拖拽上传传十个八个文件确实方便但文件量一上来问题就成倍放大。我整理了一份自己的痛点清单你可以对照看看是不是也中招。第一是重复动作消耗注意力。每传一份都要点选文件、等解析、看状态、确认成功动作本身不难但重复两百次之后人的判断力会明显下降很容易漏传或者重复传。第二是进度不可回溯。网页端那个上传列表刷新一下就没了你不知道哪几份成功了、哪几份失败了只能靠记忆而记忆在几百个文件面前基本等于零。第三是解析失败没有反馈。扫描版 PDF 提取不出文字、加密文档读不出内容、编码不对导致乱码这些问题在网页端往往只是一句笼统的失败提示你需要逐个排查。第四是无法增量更新。我每周都会往笔记目录里加新文件也可能修改老文件网页端做不到「只上传变化的那些」每次都得全量重来。这些问题归结起来其实就一句话批量导入知识库这件事本质上是一个可自动化、可记录、可重试的数据管道任务而不是一个人工操作任务。但凡一个流程具备「输入确定、步骤固定、结果可校验」这三个特征就值得写成脚本。这也是我下定决心动手的根本原因。1.2 方案选型为什么是脚本而不是别的东西动手之前我评估过三条路。第一条路是直接用知识库平台自带的批量上传功能有些平台确实支持一次选多个文件但它们的共性问题是不支持自定义分块规则、不支持增量、失败重试也要手动点。第二条路是用现成的开源同步工具比如一些笔记软件自带的同步插件或者社区写好的导入器好处是开箱即用坏处是约束太死——你的目录结构、文件命名、元数据格式必须完全贴合它的假设一旦不匹配就得改自己的习惯这个代价我不愿意付。第三条路就是自己写脚本我最后选了它理由有三条。完全掌控分块策略。不同文档类型的最优切分方式差异很大结构化笔记适合按标题切产品文档适合按段落切PDF 转出来的文本则必须先做清洗再切通用工具很难覆盖这些细节。天然支持增量与断点续传。脚本可以把每个文件的内容哈希记下来下次运行时只处理变化的文件这是网页端永远做不到的。可复用、可迁移。脚本写一次换个知识库平台只需要改一个 API 地址和参数格式逻辑骨架完全不动边际成本极低。至于「用 AI 写」这个选择其实是因为我对 Python 的熟练度只到能看懂、能改错的水平从零手写一个带重试和状态管理的完整脚本我大概要花两三个晚上而且中间会卡在细节上。让 AI 出第一版我负责审和调这个分工效率高得多。1.3 整体数据流长什么样在写第一行代码之前我强制自己先用自然语言把整个流程描述了一遍。这一步非常关键因为AI 拿到的需求描述越接近伪代码产出的代码质量就越高。我当时写给自己看的需求是这样的扫描指定目录下所有 Markdown、TXT、PDF、DOCX 文件对每个文件计算内容哈希如果这个哈希在上次运行记录里已经存在就跳过。没处理过的文件先解析成纯文本做基础清洗判断内容是否过短然后按分块规则切分或直接整篇提交给知识库接口。接口调用失败要自动重试三次采用指数退避。每个文件的处理结果写入状态文件包括文件名、哈希、返回的文档 ID、时间戳。整个过程要有日志日志同时输出到控制台和文件。把这段话拆开其实是五个模块文件扫描器、内容解析器、清洗与分块器、上传客户端、状态管理器。这五个模块之间是流水线关系前一个的输出是后一个的输入中间任何一环失败都不能影响其他文件的处理。这个「模块化 单文件隔离」的设计决策非常关键它意味着一个文件解析崩溃不会导致整个批次中断你只需要看日志定位到那个文件单独排查。很多新手写批处理脚本习惯把逻辑写成一大坨 for 循环中间一个异常就全盘退出这是最常见的坑。2. 环境准备与关键依赖2.1 Python 版本与依赖清单我用的 Python 3.10实际上 3.8 以上都能跑。依赖库只装了四个全部是轻量级的没有引入任何重型框架。选依赖的原则是能用标准库就用标准库必须装第三方库就选维护活跃、依赖少的。pip install requests pyyaml pypdf python-docx逐个说明为什么要装它们。requests负责发 HTTP 请求比标准库的 urllib 好用太多尤其是连接复用和超时控制。pyyaml用来读配置文件把 API 密钥、目录路径这些容易变动的东西从代码里剥出来避免硬编码。pypdf负责解析 PDF注意它只能提取文本型 PDF扫描件提取不出来这一点后面会专门讲。python-docx负责解析 Word 文档它读的是.docx格式老的.doc读不了。提示装依赖的时候建议用虚拟环境不要直接装在系统 Python 里。我见过太多人因为系统环境被污染导致别的项目跑不起来排查半天才发现是版本冲突。如果你的文档里还有.doc、.pptx、.xlsx这些格式我的建议是先手动转换成支持的格式再导入不要为了省事在脚本里装一堆转换库。转换本身涉及办公软件的格式兼容问题坑比导入环节多得多不值得把复杂度堆在一个脚本里。2.2 知识库接口的准备与鉴权不同知识库平台的接口长得不一样但套路基本一致一个数据集 ID 加一个 API Key通过 Bearer 方式鉴权POST 一个 JSON 上去。我这边用的是支持「按文本创建文档」接口的 RAG 知识库接口路径形如/v1/datasets/{dataset_id}/document/create-by-text。这里有个容易忽略的点API Key 的权限范围要和操作匹配。有些平台的密钥分「应用密钥」和「数据集密钥」前者是用来调用对话接口的后者才有权限往知识库里写数据。我第一次调试时拿错了密钥接口一直返回 401查了半小时才反应过来。所以拿到密钥之后先不要写代码用curl手动打一发请求验证通路这五分钟的投入能省你一小时的排查。curl -X POST https://your-kb-host/v1/datasets/YOUR_DATASET_ID/document/create-by-text \ -H Authorization: Bearer YOUR_DATASET_KEY \ -H Content-Type: application/json \ -d {name:联通测试,text:这是一段测试内容。,indexing_technique:high_quality}这条命令能返回 200 并且看到文档 ID才说明鉴权和路径都没问题。返回 404 一般是数据集 ID 写错了返回 401 是密钥问题返回 400 多半是请求体字段名不对。先用 curl 打通再用代码封装这是我调试所有 HTTP 接口的固定顺序。2.3 目录结构与配置文件设计目录结构我按「代码、配置、数据、状态、日志」五分离的原则来组织这样后期维护和迁移都清爽。kb-importer/ ├── importer.py # 主程序 ├── config.yaml # 配置接口地址、密钥、源目录 ├── requirements.txt # 依赖清单 ├── docs/ # 待导入的文档源目录 │ ├── 网络/ │ ├── 数据库/ │ └── 杂项/ ├── state/ │ └── uploaded.json # 处理状态记录 └── run.log # 运行日志配置文件把易变项全部抽离出来代码里不出现任何魔法值。base_url: https://your-kb-host/v1 dataset_id: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx api_key: dataset-xxxxxxxxxxxxxxxx source_dir: ./docs interval: 1.2 # 每个文件之间的间隔秒数 batch_size: 5 # 预留并发批大小interval这个参数看着不起眼实际很重要。知识库在做文档索引时要调用嵌入模型这是一个重计算的过程如果你以每秒几十个文件的速度狂发请求平台侧很容易排队超时或者直接给你限流。我实测下来 1.2 秒的间隔是比较稳的一万字左右的文档每次索引大约需要 1 到 2 秒这个节奏基本能跑满而不会触发 429。3. 核心环节的完整实现3.1 文件扫描别小看遍历这件事扫描看起来是最简单的一步其实藏着几个必须处理的边界情况。第一是隐藏文件和临时文件比如 Office 打开文档时会生成~$xxx.docx这样的锁文件扫进去会解析失败。第二是递归层级用rglob可以递归所有子目录但要防止符号链接造成无限递归。第三是排序稳定性一定要排序否则每次运行顺序都不一样日志没法对照。SUPPORTED {.md, .txt, .pdf, .docx} def scan_files(root: Path): root root.resolve() for p in sorted(root.rglob(*)): if not p.is_file(): continue if p.suffix.lower() not in SUPPORTED: continue if p.name.startswith(~$) or p.name.startswith(.): continue yield p这段代码里sorted()的作用比看起来大。它保证了处理顺序稳定当你在日志里看到第 158 个文件失败时下次重跑还能定位到同一个位置。另外p.suffix.lower()是必要的Windows 上文件的扩展名大小写经常不统一.PDF和.pdf都要能识别。3.2 内容解析不同格式的取文本姿势解析环节是整个脚本里最容易翻车的地方因为文档格式的坑太多了。我按格式逐个说明。Markdown 和 TXT 最简单直接读就行但必须指定编码并且允许错误忽略因为很多从网页复制来的文本会混入奇怪的字符。def read_text(path: Path) - str: suffix path.suffix.lower() if suffix in (.md, .txt): return path.read_text(encodingutf-8, errorsignore) if suffix .pdf: from pypdf import PdfReader reader PdfReader(str(path)) return \n.join(page.extract_text() or for page in reader.pages) if suffix .docx: from docx import Document doc Document(str(path)) return \n.join(p.text for p in doc.paragraphs) raise ValueError(f不支持的类型: {suffix})PDF 解析这块要特别提醒pypdf提取的是 PDF 里内嵌的文本层如果你的 PDF 是扫描件或者图片导出的提取结果会是空字符串。判断方法很简单解析完之后看文本长度如果短于 50 个字符基本可以判定是扫描件这种情况只能上 OCR而我个人建议是单独用 OCR 工具转成 Markdown 再放进源目录不要把这个环节塞进脚本。Word 文档解析还有个隐藏问题python-docx只读段落读不到表格里的内容。如果你的文档里有大量表格数据直接导入会丢信息。解决办法是同时遍历 tables 对象把表格拼成文本。def read_docx_full(path: Path) - str: from docx import Document doc Document(str(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)这个细节是我实际踩坑后才补上的。最早导入的一批产品文档表格里的参数配置全部丢失检索的时候怎么都查不到回头一查才发现是这个原因。3.3 清洗与分块决定检索质量的胜负手清洗的力度需要拿捏。洗得太轻乱码和多余空白会污染向量洗得太重会误删有效内容。我的做法是只做三件确定性的事把全角空格和不换行空格替换成普通空格去掉行尾多余空白把三个以上连续换行压缩成两个。import re RE_TRAIL_WS re.compile(r[ \t]\n) RE_MULTI_BLANK re.compile(r\n{3,}) def clean(text: str) - str: text text.replace(\u3000, ).replace(\xa0, ) text RE_TRAIL_WS.sub(\n, text) text RE_MULTI_BLANK.sub(\n\n, text) return text.strip()注意我没有删除 URL。很多教程默认会把链接过滤掉理由是链接没有语义价值。但我的文档里大量存在「参考资料https://xxx」这种引用链接本身就是信息的一部分删掉反而让上下文断裂。这个取舍取决于你的文档类型做技术文档的就保留做通用问答的可以删。分块参数是决定影响检索效果的核心我把当时的计算过程记录一下。假设你的文档平均 8000 字中文知识库的分块上限是 500 个 token。中文里 1 个汉字大约对应 1 个 token那么 8000 字会被切成大约 16 到 20 块。块与块之间要设置重叠重叠量一般取分块大小的 10% 到 20%也就是 50 到 100 个 token目的是避免一句话被硬生生截断导致语义不完整。def estimate_tokens(text: str) - int: cjk sum(1 for ch in text if \u4e00 ch \u9fff) others len(text) - cjk return int(cjk * 1.0 others / 4)分块策略我采用的是「先按标题切超长了再按段落切」的两级方案。理由很实在技术笔记天然带有标题层级一个##或者###下的内容通常就是一个完整的知识单元按这个边界切出来的块语义最完整检索命中率明显高于机械按字数切。def split_by_headers(text: str, max_tokens: int 500): blocks re.split(r\n(?#{1,4}\s), text) chunks [] for block in blocks: if estimate_tokens(block) max_tokens: if block.strip(): chunks.append(block.strip()) continue buf for para in [p for p in block.split(\n\n) if p.strip()]: if estimate_tokens(buf para) max_tokens and buf: chunks.append(buf.strip()) buf para else: buf buf \n\n para if buf else para if buf.strip(): chunks.append(buf.strip()) return [c for c in chunks if len(c) 20]最后那个len(c) 20是过滤规则把「相关阅读」「更新日志」这种只有几个字的碎片丢掉。实测下来这些碎片块不仅浪费索引额度还会在检索时被误召回拉低结果质量。注意如果你的知识库平台自己支持服务端分块很多平台在接口里提供 segmentation 参数那就没必要在本地再切一遍两边都切会导致语义被破坏两次。我当时的选择是本地只做解析和清洗把干净的全文传上去分块交给平台按参数执行。3.4 上传客户端重试机制是必需品接口调用这块最核心的设计是重试。批量任务里出现偶发网络抖动几乎是必然的如果一次失败就放弃你会在日志里看到大量无意义的失败记录。我的重试策略是指数退避第一次失败等 2 秒第二次等 4 秒第三次等 8 秒。def upload_text(session, cfg, name, text, retries3): url f{cfg[base_url].rstrip(/)}/datasets/{cfg[dataset_id]}/document/create-by-text payload { name: name, text: text, indexing_technique: high_quality, process_rule: { mode: custom, rules: { pre_processing_rules: [ {id: remove_extra_spaces, enabled: True}, {id: remove_urls_emails, enabled: False}, ], segmentation: { separator: \n\n, max_tokens: 500, chunk_overlap: 50, }, }, }, } headers { Authorization: fBearer {cfg[api_key]}, Content-Type: application/json, } delay 2.0 for attempt in range(1, retries 1): try: resp session.post(url, jsonpayload, headersheaders, timeout60) if resp.status_code in (200, 201): return resp.json() if resp.status_code in (429, 500, 502, 503, 504): time.sleep(delay) delay * 2 continue raise RuntimeError(fHTTP {resp.status_code}: {resp.text[:200]}) except requests.RequestException as exc: time.sleep(delay) delay * 2 raise RuntimeError(f上传失败: {name})这段代码里有几个设计决策值得展开说。只对特定状态码重试429 是限流、5xx 是服务端问题这些是重试有意义的而 400 参数错误、401 鉴权错误重试一百次也没用直接抛出更快。超时设为 60 秒因为大文档的服务端索引时间可能长设太短会频繁触发假失败。用 Session 复用连接比每次新建连接快不少还能减少 TCP 握手开销。name字段的处理也有讲究。我用的是相对路径转换来的名字比如网络_HTTP缓存机制.md因为知识库列表里只显示一个名字如果把所有文件都叫index.md你根本分不清哪个是哪个。3.5 断点续传让脚本可以随时中断状态管理是我认为这个脚本最有价值的部分也是新手最容易忽略的部分。核心思路很简单用文件内容的哈希值作为指纹记录哪些文件已经处理过。下次运行时只要指纹没变就跳过指纹变了就重新上传。def fingerprint(path: Path) - str: digest hashlib.sha256() with open(path, rb) as f: for block in iter(lambda: f.read(1 20), b): digest.update(block) return digest.hexdigest()[:16]这里用sha256分块读取而不是一次性读完整个文件是为了处理大文件时不占内存。截取前 16 位足够了碰撞概率可以忽略。状态文件我用「写临时文件再原子替换」的方式保存避免写到一半程序被杀导致文件损坏。def save_state(state): STATE_FILE.parent.mkdir(parentsTrue, exist_okTrue) tmp STATE_FILE.with_suffix(.tmp) tmp.write_text(json.dumps(state, ensure_asciiFalse, indent2), encodingutf-8) tmp.replace(STATE_FILE)主循环把上面所有模块串起来每个文件的处理都包在 try 里无论成功失败都保存状态这样即使中途强制退出下次也能接上。def main(): cfg load_config() state load_state() session requests.Session() source Path(cfg[source_dir]).expanduser().resolve() files list(scan_files(source)) ok skip fail 0 for idx, path in enumerate(files, 1): rel str(path.relative_to(source)) fp fingerprint(path) record state[done].get(rel) if record and record.get(fp) fp: skip 1 continue try: text clean(read_text(path)) if estimate_tokens(text) 30: skip 1 continue result upload_text(session, cfg, rel.replace(os.sep, _), text) doc_id (result.get(document) or {}).get(id, ) state[done][rel] {fp: fp, doc_id: doc_id, ts: int(time.time())} ok 1 log.info([%d/%d] 成功: %s - %s, idx, len(files), rel, doc_id) except Exception as exc: fail 1 log.error([%d/%d] 失败: %s | %s, idx, len(files), rel, exc) finally: save_state(state) time.sleep(float(cfg.get(interval, 1.0))) log.info(完成: 成功 %d, 跳过 %d, 失败 %d, ok, skip, fail)跑完之后你会看到这样的输出一眼就能看出整体情况2025-06-12 21:14:03 | INFO | 共扫描到 1284 个文件 2025-06-12 21:14:03 | INFO | [1/1284] 成功: 网络_HTTP缓存机制.md - 9f3c... 2025-06-12 21:14:06 | INFO | [2/1284] 成功: 网络_TCP三次握手.md - 71ab... 2025-06-12 21:14:08 | WARN | 临时错误 429第 1 次重试 2025-06-12 21:14:19 | INFO | [3/1284] 成功: 数据库_索引原理.pdf - 3e2d... 2025-06-12 21:31:47 | INFO | 完成: 成功 1247, 跳过 32, 失败 5失败的那 5 个我逐个看了日志3 个是扫描版 PDF 提取不出文本2 个是超长文档超过了平台的单文档大小限制。前者手动 OCR 后重跑后者拆成两篇再传半小时内全部解决。4. 用 AI 写脚本的实操方法论4.1 怎么给 AI 描述需求才有效这次经历让我最大的收获不是脚本本身而是搞明白了一件事AI 写代码的质量八成取决于你描述需求的质量而不是模型本身有多强。我第一次提需求的时候只说了一句「帮我写个 Python 脚本把文件夹里的文档批量上传到知识库」AI 给我的版本一坨没有任何异常处理没有状态记录连文件类型判断都是硬编码的。后来我改了策略把需求拆成三段来描述。第一段描述数据流也就是从哪来、经过什么处理、到哪去。第二段描述约束条件包括只支持哪几种格式、遇到什么情况跳过、失败重试几次、用什么策略退避。第三段描述输出产物包括要生成什么文件、日志格式长什么样、状态文件里存哪些字段。我实际用的提示词大致是这样组织的我需要一个批量导入知识库的脚本。数据源是本地目录递归扫描.md、.txt、.pdf、.docx四类文件跳过隐藏文件和~$开头的临时文件。处理流程是读取内容、清洗文本、判断长度、通过 HTTP POST 提交到接口。约束条件每个文件用 sha256 哈希做去重已处理且未修改的跳过网络失败或 429、5xx 状态码重试三次采用指数退避初始 2 秒每个文件处理完都要落盘状态支持随时中断。输出要求控制台和文件双写日志状态用 JSON 存包含文件名、哈希、返回的文档 ID 和时间戳。请先给出模块划分再逐个模块给代码。用这种方式提需求AI 给出的第一版就已经能用七八成剩下的都是细节打磨。关键区别在于前者是让 AI 猜你要什么后者是你已经把设计做完了、只让 AI 负责翻译成代码。这个分工对新手尤其友好因为设计能力是可以凭空想象的而语法细节才是真正需要查资料的部分。4.2 让 AI 帮你调试而不是重写代码跑起来一定会报错这时候很多人习惯把整个报错贴回去说「跑不通帮我改」AI 会给你一版重写的代码但改了什么你完全不知道改完可能引入新的问题。我的做法是只贴报错和相关的那一段代码并且明确要求解释原因再给修复方案。比如我当时遇到的FileHandler报父目录不存在的问题我没有直接让它改而是这样问的运行时报错FileNotFoundError: [Errno 2] No such file or directory: run.log。我的目录结构里日志文件在脚本同级目录代码是logging.FileHandler(LOG_FILE)。请解释为什么找不到并给出最小修改方案不要重写整个日志配置。它给出的解释是 FileHandler 不会自动创建父目录给出的修复是在配置前加一句mkdir。修改只有一行我一眼就看懂了也知道以后遇到类似问题该怎么处理。这种「解释优先」的提问方式长期收益远大于直接要答案因为你是在积累调试直觉而不是收集代码片段。4.3 AI 生成的代码必须审哪些地方我对 AI 生成的代码有一个固定审查清单每次都会过一遍因为这个清单里的问题几乎每次都中招。第一项是状态码判断AI 特别喜欢只判断status_code 200但创建接口常常返回 201这一条不补上会导致明明成功却记为失败。第二项是异常捕获范围AI 有时候用裸except:这会连KeyboardInterrupt都吃掉你按 CtrlC 都停不下来必须改成except Exception。第三项是资源释放打开文件、创建 Session 的地方有没有正确关闭。第四项是边界条件空文件、超大文件、特殊字符文件名这些 AI 默认不会处理得主动要求。最后一项也是最重要的API 密钥绝对不能硬编码在代码里。AI 给的第一版示例经常是api_key sk-xxx这种形式方便演示但极其危险一旦你把代码传到任何公开地方密钥就泄露了。我的做法是全部走配置文件并且把config.yaml加进.gitignore仓库里只保留一个config.example.yaml。5. 常见问题排查与优化经验5.1 高频报错速查表跑这一千多个文件的过程中我记录下了遇到的所有问题整理成表你遇到类似症状可以直接对照。报错或症状根本原因解决办法HTTP 401密钥类型不对或已失效用 curl 单独验证确认使用的是数据集密钥HTTP 404数据集 ID 错误或路径拼错对照平台文档核对接口路径与 IDHTTP 400请求体字段名不匹配打印 payload 逐字段核对文档HTTP 429请求频率过高被限流增大 interval 参数检查是否有并发解析结果为空扫描版 PDF 无文本层单独 OCR 转 Markdown 后再导入中文变问号文件编码是 GBK读取时显式指定或做编码探测文档过大被拒超过平台单文档上限按标题拆成多篇分别导入状态文件损坏写入中途被中断改为临时文件原子替换表格内容丢失docx 段落读不到表格额外遍历 tables 对象拼接这张表里的每一条我都是真金白银踩出来的。尤其是「中文变问号」那条我有一批早期的博客备份是 GBK 编码直接按 UTF-8 读出来全是乱码但程序不报错只是把乱码传进了知识库直到检索时发现结果全是问号才反应过来。编码问题最阴险的地方就在于它静默失败所以我在清洗环节加了一个检查如果文本中连续出现大量\ufffd替换字符就记一条警告日志。5.2 几个反直觉的实操心得第一个心得是不要追求一次全量跑完。我最开始想一口气把 1284 个文件全传上去结果跑到第 300 个的时候平台开始大量返回 429整个任务效率反而更低。后来我改成先跑 50 个试试水确认接口稳定、参数合理之后再放开跑。先小批量验证再全量执行的节奏在任何批量任务里都成立。第二个心得是日志比调试器有用。脚本跑一两个小时你不可能盯着调试器看日志是你唯一的观测窗口。我在日志里记录的信息包括序号、总数、文件名、文档 ID这几个字段能让我随时知道进度在哪、哪个文件出问题。如果你发现某个文件反复失败直接 grep 文件名就能看到它每次的报错内容。第三个心得是分块大小要跟着文档类型调不要一刀切。我一开始所有文档都用 500 token 的分块结果发现 API 文档这种「一个接口一段」的内容被切得七零八落检索时经常只召回半段说明。后来我给这类文档单独设了 800 token效果明显好转。如果你的知识库平台支持按数据集设置分块规则最好的做法是按内容类型建多个数据集每个用不同的分块参数。第四个心得是导入完成不等于检索可用。文档上传成功后平台还需要时间做向量化索引尤其是文档多的时候可能要等几分钟到十几分钟。我一开始传完立刻去检索结果什么都搜不到以为是导入失败白排查了半天。判断索引是否完成要看平台的状态字段通常在文档列表里会显示「索引中」和「可用」两种状态。5.3 性能优化与后续扩展方向性能这块我做了两处优化。一是把文件哈希计算和上传解耦扫描阶段先把所有文件的指纹算出来这样即使中途中断重启时也不需要重新计算全部哈希。二是状态文件改为批量落盘原来的实现是每处理一个文件就写一次状态文件多的时候磁盘 IO 反而成了瓶颈改成每 10 个文件落一次性能提升明显代价是最坏情况下丢失 10 条记录重跑一遍也就几分钟。接下来我准备做的扩展有这么几个。第一是元数据注入把文件的目录路径、修改时间、标签一起写进文档描述里这样检索时可以做条件过滤比如只在「数据库」目录下找。第二是多知识库分发同一批文档按分类自动路由到不同的数据集技术类进技术库产品类进产品库避免混在一起互相干扰。第三是定时增量同步用系统的计划任务每天跑一次只处理有变动的文件这样就彻底告别手动导入了。我在实际使用中的体会是AI 写脚本这件事真正降低的门槛不是「写代码」而是「把想法翻译成可执行逻辑」这一步。以前我脑子里有清晰的需求但卡在不知道用什么库、怎么写语法现在这个环节被抹平了我只需要把设计想清楚。整个脚本从提需求到跑通我花了大概两个晚上其中真正的编码时间不到三分之一剩下全是在想清楚每个环节的边界条件。如果你手里也有一堆文档等着进知识库别急着打开浏览器拖文件花一个晚上把脚本搭出来后面省下的时间会远超这个投入。
