用AI写Python脚本:RAG知识库批量导入与断点续传实战
2026/9/18 22:37:50 网站建设 项目流程

上个月我把攒了三年的技术笔记做了一次彻底归档,一千两百多份 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(encoding="utf-8", errors="ignore") 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, retries=3): 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": f"Bearer {cfg['api_key']}", "Content-Type": "application/json", } delay = 2.0 for attempt in range(1, retries + 1): try: resp = session.post(url, json=payload, headers=headers, timeout=60) 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(f"HTTP {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(parents=True, exist_ok=True) tmp = STATE_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-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都吃掉,你按 Ctrl+C 都停不下来,必须改成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 错误或路径拼错对照平台文档核对接口路径与 ID
HTTP 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 写脚本这件事真正降低的门槛不是「写代码」,而是「把想法翻译成可执行逻辑」这一步。以前我脑子里有清晰的需求,但卡在不知道用什么库、怎么写语法,现在这个环节被抹平了,我只需要把设计想清楚。整个脚本从提需求到跑通,我花了大概两个晚上,其中真正的编码时间不到三分之一,剩下全是在想清楚每个环节的边界条件。如果你手里也有一堆文档等着进知识库,别急着打开浏览器拖文件,花一个晚上把脚本搭出来,后面省下的时间会远超这个投入。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询