折腾了大半年知识库，脚本终于跑通的那天晚上，我盯着终端里滚动的 1287 个文件名，心情有点复杂。这大概是我第一个真正意义上"纯 AI 手搓"的脚本程序——从目录遍历、文件解析到批量导入知识库的接口调用，代码里几乎每一行都出自 AI 之手，我做的只是拆需求、卡约束、看日志、改参数。放在两年前，这种"人不写代码只当监工"的玩法我是不信的，但现在它确实躺在我硬盘里，稳定跑完了两万多份文档的批量导入。这篇文章不聊虚的，就把这个纯 AI 手搓脚本的完整思路摊开讲：它解决了什么问题、架构为什么这么设计、每一段代码到底在干什么、踩了哪些坑、哪些地方 AI 给的答案我直接推翻重写了。如果你手里也堆着一大坨 Word、PDF、Markdown 想批量灌进知识库，或者你也想试试让 AI 帮你从零搓一个能用的自动化脚本，这篇应该能省你不少时间。
1. 需求拆解：为什么非得自己写一个批量导入脚本
1.1 手动拖拽的崩溃现场
先说清楚背景。我手上的资料大概分三类：一类是多年积累的 Markdown 笔记，散在十几个目录里，结构还算规整；一类是行业报告和合同模板，清一色 PDF，加起来四百多份；还有一类是同事交接过来的 Word 文档，命名风格堪称灾难，什么"最终版""最终版2""最终版_真的最终.docx"全都有。
最开始我的做法很原始：打开知识库后台，新建文档，复制粘贴，保存，等切片完成，再建下一个。前三份我还挺有耐心，到第十份的时候手已经开始机械化了，到第三十份我意识到一个问题——这么干，两万份文档我得干到明年。而且人工操作必然伴随遗漏和重复：同一份文件我可能因为手滑点了两次，有些文件在文件夹深处我根本没想起来，还有些文件名里的空格和中文括号会让手动输入变得极其难受。
这时候真正的痛点就浮出来了：知识库的价值在于"全"，而我手动搬的过程天然保证不了"全"。一个漏掉的目录，可能正好就是最关键的那批资料。
1.2 三种导入路径的对比与取舍
在动手写脚本之前，我把能想到的路子都过了一遍，做了个简单的对比：
| 方案 | 实现成本 | 可维护性 | 适用规模 | 主要问题 |
|---|---|---|---|---|
| 后台手动上传 | 极低 | 差 | 20 份以内 | 容易漏、容易重、无法追溯 |
| 知识库自带文件夹同步 | 低 | 中 | 中等 | 依赖部署形态，格式支持有限 |
| 调用 API 批量投递 | 中 | 高 | 上千份无压力 | 需要处理限流、断点、解析 |
| 直接操作底层向量库 | 高 | 差 | 特殊场景 | 耦合太深，升级即翻车 |
我最后选了第三条路：调用知识库暴露的 HTTP 接口，用脚本把本地文件一份一份喂进去。理由有三条，很实在。
第一，API 是有状态可记录的。每份文件投递成功还是失败、花了多久、返回了什么 ID，我都能落进日志里。手动上传是"黑箱"，出了问题只能重新翻一遍；脚本投递是"白盒"，出错能定位到具体文件。
第二，API 能把解析这一步交给我自己控制。很多知识库自带的文档解析不太理想，尤其是那种排版复杂的 PDF，切出来的段落七零八落。我自己在本地先把文本抽干净、把标题层级理清楚，再送进去，检索质量会明显好一截。
第三，脚本是可以反复跑的。今天导入一批，下周新增了三十份，我再跑一次就行——前提是脚本得支持增量识别，这一点后面会专门讲。
提示：选方案之前先确认你的知识库是否开放了文档写入接口，以及接口是否要求企业版授权。有些平台把批量写入放在付费档位里，动手前先确认，别写完了发现调不通。
1.3 什么样的知识库最适合接脚本
从我的经验看，能被脚本"喂"的知识库通常有三个特征：有稳定的文档创建接口、支持以纯文本或文件流的方式提交内容、返回结构化的文档 ID。RAG 类知识库基本都满足，比如 Dify 这类开源方案的知识库模块，接口文档写得相对清楚，创建文档、查询状态、删除文档都有对应端点，非常适合做批处理。
反过来，如果你的目标是像 Obsidian 这种纯本地 Markdown 库，那"导入"这个词的含义就变了——它本质上就是文件放进目录，不需要调接口，脚本只需要做格式清洗和目录归位。这两种场景的脚本写法差别很大，动手前一定要先想清楚你到底是在"投递"还是在"归位"。
2. 整体架构设计：AI 辅助写脚本的正确打开方式
2.1 三段式结构：扫描、解析、投递
整个脚本我拆成了三段，每一段职责单一，可以单独运行、单独调试：
扫描段负责把一个根目录下的所有目标文件找出来，过滤掉不该处理的目录和文件类型，输出一份"待办清单"。这一段不碰文件内容，只认路径和后缀。
解析段负责把清单里每个文件读成纯文本，同时抽取元数据——来源路径、文件名、最后修改时间、所属目录标签。解析失败的文件单独记录，不阻塞后面的流程。
投递段负责把文本和元数据通过接口送进知识库，处理限流、重试、失败登记，最后生成一份导入报告。
这么拆的最大好处是"可回滚"。扫描有问题，我只看清单就知道；解析出乱码，我单独跑解析段就能复现；接口挂了，前两步的产物已经落盘，等接口恢复直接重跑投递段，不用从头再来。
注意：千万不要把三段揉进一个函数里。我第一版图省事，一个 main 函数从头跑到尾，结果第十份文件解析报错，整个流程中断，前面九份白跑。拆开之后，任何一段崩了都不影响其他段的产物。
2.2 为什么是 Shell 加 Python 混着写
有朋友问我，既然都用 AI 写了，为什么不干脆全用 Python，或者全用 Shell？我当时的考虑是这样的。
Shell 适合做"调度和文件系统层面的粗活"：设置环境变量、切换工作目录、串联多个步骤、处理退出码、把日志重定向到带时间戳的文件里。这些东西用 Shell 写就是几行，用 Python 写反而啰嗦。
Python 适合做"内容层面的细活"：解析 docx、解析 PDF、算哈希、发 HTTP 请求、处理 JSON、做指数退避重试。这些用 Shell 写就是自我折磨。
所以最终形态是：一个run.sh做总调度，三个 Python 脚本做具体工作，中间用 JSON 文件当"接口"，Shell 完全不需要理解 JSON 里面是什么。
#!/usr/bin/env bash set -euo pipefail ROOT_DIR="${1:-./docs}" WORK_DIR="./.kb_import" mkdir -p "$WORK_DIR" echo "[1/3] 扫描文件..." python3 scan.py --root "$ROOT_DIR" --out "$WORK_DIR/manifest_scan.json" echo "[2/3] 解析文本..." python3 parse.py --in "$WORK_DIR/manifest_scan.json" --out "$WORK_DIR/manifest_parsed.jsonl" echo "[3/3] 批量投递..." python3 push.py --in "$WORK_DIR/manifest_parsed.jsonl" \ --state "$WORK_DIR/state.json" \ --log "$WORK_DIR/push.log" echo "全部完成，报告见 $WORK_DIR/push.log"set -euo pipefail这三个参数很重要，值得单独说一下。-e让脚本遇到非零退出码立刻停止；-u让引用未定义变量直接报错；-o pipefail让管道中任意一环失败都算失败。少了这几个，脚本会在出错后继续往下跑，最后给你一个"看起来成功了"的假象，这种坑最难查。
2.3 和 AI 协作的提示词拆分策略
这是我觉得最值得分享的部分。很多人用 AI 写脚本，习惯一次性把需求全丢过去："帮我写个脚本把文件夹里的文档批量导入知识库"。这样拿到的代码通常看着挺全，但一跑就废——因为它不知道你的目录结构、不知道你的接口长什么样、不知道你的文件命名习惯。
我的做法是把提示词拆成四层，逐层喂：
第一层给角色和边界。"你是一名 Python 工程师，写一个只做文件扫描的函数，输出文件绝对路径列表。不要做任何文件读取，不要发网络请求。" 边界越清楚，AI 越不容易自作主张加东西。
第二层给输入输出样例。直接告诉它输入是什么、输出长什么样。比如"输入是一个 Path 对象指向根目录，输出是一个生成器，逐个 yield Path 对象"。
第三层给约束和例外。"跳过 .git、node_modules、.trash 目录；跳过软链接；后缀只保留 .md/.txt/.docx/.pdf；目录名含空格和中文要能正确处理。" 这些例外才是真实世界和玩具代码的分界线。
第四层才是让它写实现。前三层对齐之后，实现部分基本一次就能过。
我实测下来，这种"分层喂"的方式，代码一次通过率能从三成提到八成以上。反过来，如果你把四层混成一段话丢过去，AI 会挑它最容易实现的那部分做，剩下的细节全部糊过去。
3. 核心细节解析：扫描、解析与元数据设计
3.1 目录遍历与文件类型白名单
扫描看着简单，其实细节不少。第一个坑是递归遍历时的性能问题：如果根目录下挂着一个巨大的node_modules或者.git，rglob("*")会把里面每一个文件都枚举一遍，几万个小文件能让脚本卡上几十秒。
我的处理是提前判断路径里有没有需要跳过的片段，一旦命中就直接剪枝：
from pathlib import Path ALLOW_SUFFIX = {".md", ".markdown", ".txt", ".docx", ".pdf"} SKIP_PARTS = {".git", ".obsidian", "node_modules", "__pycache__", ".trash", ".kb_import"} def walk(root: Path): root = root.resolve() for p in sorted(root.rglob("*")): if any(part in SKIP_PARTS for part in p.parts): continue if p.is_symlink(): continue if not p.is_file(): continue if p.suffix.lower() not in ALLOW_SUFFIX: continue yield p这里p.is_symlink()那行是我后来补的。起因是我有个目录用软链接指回了上层，结果脚本开始无限递归，文件数从一千多突然涨到几十万，跑了半天没停。加上这一行之后立刻正常。
第二个细节是排序。sorted()看起来多余，但它保证了每次运行的扫描顺序一致，配合后面的断点续传才能对得上。不排序的话，文件系统返回的顺序在不同机器上可能不同，状态文件就对不上了。
3.2 Word、PDF、Markdown 的文本提取差异
三种格式的解析难度完全不在一个量级上，我把它们分开处理。
Markdown 最简单，直接读文件就行。但有一点要注意：读的时候必须显式指定编码，encoding="utf-8"，而且最好加errors="replace"，否则遇到一个混了 GBK 的老文件，整个流程就中断了。
Word 中等难度。文本不能只从paragraphs里拿，因为很多文档的关键信息其实在表格里。表格如果不处理，导进去的就是残缺内容。我用的方案是把段落和表格按顺序都抽出来，表格用竖线拼成一行：
import docx def parse_docx(path: Path) -> str: d = docx.Document(str(path)) buf = [] for para in d.paragraphs: t = para.text.strip() if t: buf.append(t) for table in d.tables: for row in table.rows: cells = [c.text.strip().replace("\n", " ") for c in row.cells] line = " | ".join(c for c in cells if c) if line: buf.append(line) return "\n\n".join(buf)PDF 最麻烦，麻烦在于它分两种情况：带文字层的和纯扫描件的。带文字层的用pdfplumber或pypdf就能抽，扫描件抽出来是空字符串，必须走 OCR。我第一版脚本没做这个区分，结果四百多份 PDF 里有一百多份导进去是空的，检索时怎么都搜不到，排查了半天才发现是扫描件。
我的做法是在解析结果里加一个长度判断，如果抽出来的文本长度小于某个阈值，就在日志里打一个醒目的标记，让这些文件走人工通道：
def parse_pdf(path: Path) -> tuple[str, str]: import pdfplumber buf = [] with pdfplumber.open(str(path)) as pdf: for page in pdf.pages: txt = page.extract_text() or "" if txt.strip(): buf.append(txt) text = "\n".join(buf) if len(text.strip()) < 50: return text, "SUSPECT_SCANNED" return text, "OK"提示：不要指望脚本能自动化处理所有情况。遇到扫描件，与其硬上 OCR（识别率不稳定，还可能把数字和表格识别错），不如把这类文件单独列出来，人工确认一遍再用专门的工具处理。一百份文件人工过一遍是两小时的事，写一套稳定的 OCR 后处理流程可能是两天。
3.3 元数据怎么设计才方便后续检索
元数据这部分我返工过两次，值得多写几句。第一版我只存了文件名和文本，导进去之后发现检索结果全是文件名，完全看不出内容属于哪个项目、哪年写的，用起来很难受。
第二版我把元数据设计成了这样几个字段：
| 字段 | 来源 | 用途 |
|---|---|---|
| source_path | 文件的相对路径 | 定位原文，方便回查 |
| file_name | 文件名 | 展示用，去掉后缀 |
| dir_tags | 相对路径按斜杠拆开 | 充当天然的目录标签 |
| mtime | 文件最后修改时间 | 判断资料新旧 |
| fingerprint | 内容哈希 | 增量导入和幂等判断 |
| parse_status | 解析状态 | 标记扫描件、解析失败 |
其中dir_tags这个设计我觉得挺妙的。因为我的资料目录本身就是按"项目/年份/类型"组织的，相对路径拆开之后天然就是一组标签，不用再让 AI 去猜文件属于什么分类。比如一个文件路径是行业研究/2023/新能源/XX报告.pdf，拆出来的标签就是行业研究、2023、新能源，检索的时候按标签过滤非常准。
fingerprint是幂等的关键。我用文件内容的 SHA256 加上修改时间做一个哈希，投递前先查状态文件里有没有这个哈希，有就跳过。这样同一批文件重复跑一百次，也只有第一次会真正投递。
import hashlib def fingerprint(path: Path, text: str) -> str: h = hashlib.sha256() h.update(text.encode("utf-8", errors="replace")) h.update(str(path.stat().st_mtime_ns).encode()) return h.hexdigest()[:32]3.4 分块策略与 token 估算
文本切块这件事，直接影响检索质量，比很多人想的要重要。切得太碎，语义不完整，检索出来的片段读起来像断句；切得太粗，一个块里混了好几个主题，向量表示会被平均掉，检索精度下降。
我的策略是"优先按标题切，其次按段落切，最后才按长度硬切"。具体逻辑是：先找 Markdown 里的一级二级标题，以标题为边界切大块；如果某个大块还是超过上限，就再按空行切段落；段落还超，就按句子切。
长度上限我用的是 800 个字符，重叠 120 个字符。这个数字不是拍脑袋来的，我做过一轮小规模测试：拿 20 个已知答案的问题去检索，分别用 500、800、1200 三种块长跑，800 这一档的召回准确率最高，500 会因为上下文太短丢信息，1200 会因为块内主题太杂拉低相似度。当然这只是我的语料上的结果，你的资料风格不一样，建议也做一轮这样的对比。
如果你的知识库服务端自己会做切片，那本地这一步可以简化，只做超长文本的预切。但要注意，服务端的切片规则通常是固定长度的，对标题结构不敏感，遇到技术文档和报告容易切坏。我个人的偏好是本地切好，把每个块当成一份独立文档投递，元数据里带上块序号，检索时可以通过序号把相邻块拼回去看上下文。
4. 完整实操：从零把脚本跑起来
4.1 环境准备与依赖安装
环境这块没什么花活，Python 3.9 以上就行，依赖也就几个：
python3 -m venv .venv source .venv/bin/activate pip install requests pdfplumber python-docx pyyaml四个包的用途分别是：requests发 HTTP 请求，pdfplumber抽 PDF 文字，python-docx读 Word，pyyaml读配置文件。没有一个是重型依赖，装起来很快。
Windows 上要注意一点：虚拟环境激活脚本是\.venv\Scripts\activate，而且如果你在 PowerShell 里跑，可能会因为执行策略被拦下来。这时候不用去改系统策略，直接换成cmd或者用\.venv\Scripts\python.exe显式调用解释器就行，后者更省事。
注意：如果你是从别的机器上拷过来的脚本，第一次跑之前先确认换行符。Windows 上编辑过的
.sh文件在 Linux 上会因为\r\n报错，报错信息通常是bad interpreter或者$'\r': command not found。一行sed -i 's/\r$//' run.sh就能解决。
4.2 配置文件设计：把易变的东西全部抽出来
硬编码是脚本的头号敌人。接口地址、密钥、目录路径、并发数、重试次数这些东西，写死在代码里，换一个环境就得改代码，改完还容易漏。
我的做法是全部塞进一个config.yaml：
source: root: ./docs allow_suffix: [".md", ".txt", ".docx", ".pdf"] skip_parts: [".git", ".obsidian", "node_modules", ".trash"] chunk: max_chars: 800 overlap_chars: 120 target: base_url: "https://your-kb-host/v1" dataset_id: "your-dataset-id" api_key_env: "KB_API_KEY" timeout: 60 throttle: qps: 2 max_retry: 5 backoff_base: 1.5注意api_key_env这一项，存的是环境变量的名字，不是密钥本身。密钥通过环境变量注入，这样配置文件可以放心提交到仓库，不会泄密。这是个很小的习惯，但能避免很多麻烦。
export KB_API_KEY="你的密钥"4.3 投递模块：限流、重试和超时
投递是整个脚本里最容易出问题的部分，因为它依赖外部服务。我踩过的坑包括：接口限流返回 429、单个大文件超时、网络抖动导致连接中断、服务端偶发 5xx。
处理这些的标准做法是指数退避重试，只对可重试的错误码重试：
import time import requests import os RETRYABLE = {429, 500, 502, 503, 504} def push_one(cfg, name, text, meta, session): url = f"{cfg['target']['base_url']}/datasets/{cfg['target']['dataset_id']}/document/create-by-text" headers = { "Authorization": f"Bearer {os.environ[cfg['target']['api_key_env']]}", "Content-Type": "application/json", } payload = { "name": name, "text": text, "indexing_technique": "high_quality", "process_rule": {"mode": "custom"}, } last_err = None for attempt in range(cfg['throttle']['max_retry']): try: resp = session.post(url, json=payload, headers=headers, timeout=cfg['target']['timeout']) except requests.RequestException as e: last_err = e else: if resp.status_code < 300: return True, resp.json() if resp.status_code not in RETRYABLE: return False, resp.text last_err = f"HTTP {resp.status_code}" wait = cfg['throttle']['backoff_base'] ** attempt time.sleep(min(wait, 30)) return False, str(last_err)这段代码里有几个细节值得说。
session是复用的，不是每次请求都新建。TCP 连接复用能显著降低高频请求的失败率，尤其是在 QPS 稍微高一点的时候。
退避时间用1.5 ** attempt，也就是 1 秒、1.5 秒、2.25 秒这样递增，并且用min(wait, 30)封顶。为什么要封顶？如果不封顶，第五次重试要等 7.6 秒，看起来还好，但如果你的backoff_base设成 2，第五次就是 16 秒，中间一旦有几十个文件同时遇到限流，脚本会卡到你以为它死了。
最关键的判断是if resp.status_code not in RETRYABLE: return False。这一行把"不该重试的错误"和"该重试的错误"分开了。比如 401 是密钥错了，你重试一百次也没用；400 是请求体格式不对，重试同样没用。只有限流和服务端故障才值得重试。这个逻辑不加，脚本遇到密钥失效会硬扛五次重试，白白浪费几分钟。
4.4 断点续传：状态文件怎么存
脚本跑到一半被打断是常事——网络断了、我不小心关了终端、服务端临时维护。这时候如果没有断点续传，就得从头再来，两万份文件重跑一次动辄一两个小时。
我的做法是维护一个状态文件，记录每个文件哈希的投递结果：
{ "a1b2c3d4e5f6...": {"status": "ok", "doc_id": "doc-xxx", "ts": 1700000000}, "9f8e7d6c5b4a...": {"status": "failed", "reason": "HTTP 400", "ts": 1700000005} }每投递成功一份，立刻写一次状态文件。注意是"立刻"，不是"最后统一写"。我第一版是全部跑完才落盘，结果有一次跑到 80% 断电，状态全丢了。
写状态文件还有一个技巧：先写临时文件，再原子替换。否则刚好在写的过程中被打断，状态文件会变成一个残缺的 JSON，下次读的时候直接解析报错。
import json, os, tempfile def save_state(path, state): d = os.path.dirname(os.path.abspath(path)) fd, tmp = tempfile.mkstemp(dir=d) with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) os.replace(tmp, path)os.replace在同一个文件系统内是原子操作，这样不管什么时候中断，状态文件要么是旧的完整版本，要么是新的完整版本，不会出现半截内容。
4.5 限速参数怎么算出来
qps: 2这个值我是这么定的。先查接口文档，看它有没有写明速率限制，我这边文档里写的是每分钟 120 次，也就是每秒 2 次。但这是理论上限，实际跑的时候服务端还要做切片和向量化，压力比单纯接收请求大得多。
我的做法是在理论上限的基础上打七折，用 1.4 左右。然后跑一百份文件观察失败率，如果 429 出现次数在三次以内就往上加，反之就往下调。最后稳定在 1.5 左右，两万份文件跑了大约三个半小时，中间只有零星几次限流重试。
这个调参过程听着繁琐，但比"设个高并发然后被服务端拉黑"要省事得多。批量导入这种任务，本来就不追求实时性，稳比快重要。
5. 常见问题与排查技巧实录
5.1 中文乱码和文件名特殊字符
乱码问题我遇到过两次，原因不同。
第一次是 Markdown 文件读了半天全是问号。原因是老文件用了 GBK 编码，而我用 UTF-8 强读。解决办法是读的时候加errors="replace"，同时在日志里记录哪些文件出现了替换字符，事后单独处理。更稳妥的方案是用chardet之类的库先探测编码，但会增加依赖，我选择先用简单方案。
第二次是文件名导致的。有几个文件名里带空格和中文全角括号，我在 Shell 里用for f in $(ls)遍历，空格直接把文件名劈成了两半。正确的做法是不要用 Shell 遍历文件，全部交给 Python 的pathlib处理。如果你非要在 Shell 里遍历，用find . -print0 | while IFS= read -r -d '' f这种写法，-print0和read -d ''配对才能正确处理空格和换行。
提示：投递到知识库的文档名最好做一次清洗，把换行、制表符、连续空格全部替换掉，长度截断到 100 字符以内。有些接口对文档名长度有限制，超了会直接返回 400，而错误信息往往很含糊，查起来很费劲。
5.2 重复导入和幂等性
幂等这件事，我一开始没当回事，结果知识库里出现了同一个文件的七八个副本，检索的时候同一段内容重复出现，体验很差。
根因是只按文件名判重。不同目录下同名文件很常见，比如十几个目录里都有README.md；反过来，同一个文件被复制到别的目录，文件名一样但内容相同，也会被当成两份。
正确的判重维度是内容哈希。只要哈希一样，就认为是同一份内容，跳过。如果确实需要保留同一个内容在不同目录下的版本，那就在哈希里加上路径参与计算。我的选择是内容哈希加修改时间，理由是这样既能避免重复，又能在文件被修改后重新导入。
还有一个细节：如果投递请求发出去了但响应超时，你并不知道服务端到底建没建。这时候重试就会产生重复。处理办法是先查一次文档列表，按文档名匹配，匹配到就认为已经建好了。这个查询接口一般都有，值得花十分钟加上。
5.3 排查速查表
我把这段时间遇到的问题整理成了下面这张表，遇到类似现象可以对着查：
| 现象 | 最可能的原因 | 排查动作 |
|---|---|---|
| 脚本秒退，没有任何输出 | set -e遇到某个命令非零退出 | 临时去掉-e看真实报错 |
bad interpreter | 换行符是 CRLF | sed -i 's/\r$//' 脚本名 |
| 全部文件都投递失败 | 密钥错了或环境变量没导出 | echo $KB_API_KEY确认 |
| 大量 429 | 并发太高 | 调低qps，看退避是否生效 |
| PDF 导进去是空的 | 扫描件没有文字层 | 看日志里SUSPECT_SCANNED标记 |
| 中文变成问号 | 文件编码不是 UTF-8 | 检查日志里的替换字符计数 |
| 同一份文档重复出现 | 判重逻辑只按文件名 | 改成内容哈希判重 |
| 文件数远超预期 | 软链接导致循环遍历 | 加is_symlink()判断 |
| 跑到一半卡住不动 | 单文件超时且没有设 timeout | 给请求加显式 timeout |
| 状态文件解析报错 | 写入过程被打断 | 改成临时文件加原子替换 |
5.4 一个容易被忽略的检查点：投递后的切片状态
投递接口返回 200 不代表文档就能被检索到，它只是"收到并排队"。服务端还要做文本清洗、切片、向量化，这个过程对一份长文档可能要几十秒。
我一开始不知道这个差别，脚本跑完立刻去检索，发现什么都搜不到，以为是投递失败了，又重跑了一遍，结果造成大量重复。后来才明白要去查文档的状态字段，等它变成"已完成"再开始检索测试。
所以如果你的脚本是"投递完就结束"，最好在最后加一个轮询环节，把所有投递成功的文档 ID 收集起来，定期查一次状态，把还在处理中的数量打进日志。这不是必须的，但能让你对整批导入的进度心里有数，也方便判断什么时候可以进行检索效果验证。
6. 从纯 AI 手搓脚本这件事上，我攒下的几点体会
6.1 AI 写的代码，审查重点在哪里
两万多份文件跑下来，我对"AI 写的代码哪些地方必须自己盯"有了比较清晰的认识。
边界条件必须自己看。AI 写主流程很利索，但对空文件、零字节文件、超长单行、循环软链接这类边界情况的处理经常是缺失的。我的做法是专门写一个edge_cases目录，里面放各种畸形文件——空文件、只有一行的文件、编码混用的文件、超大文件——每次改完脚本先拿这个目录跑一遍。
错误处理必须自己看。AI 倾向于写try: ... except Exception: pass，这种写法在脚本里是灾难，因为它会把所有问题吞掉。我现在拿到 AI 生成的代码，第一件事就是搜except，看每一个捕获是不是做了最起码的日志记录。
资源释放必须自己看。文件句柄、HTTP 会话、临时文件，这些东西用完了该关的要关。批量处理几千个文件的时候，句柄泄漏会让你在某一个临界点上突然报"too many open files"，而且报错位置和真正的原因隔得很远。
6.2 后续可以怎么扩展
这套脚本目前只是个"一次性搬运工"，但它的骨架其实能撑起更多东西。
第一个扩展方向是定时增量同步。把脚本挂到系统定时任务里，每天凌晨跑一次，靠内容哈希自动判断哪些是新文件，只投递增量部分。这样知识库就变成了一个持续更新的状态，而不是一次性的快照。要注意的是定时任务里的环境变量和交互式终端里不一样，密钥一定要在任务脚本里显式导出。
第二个方向是投递前的质量预处理。比如自动去掉页眉页脚、自动合并被换行打断的段落、自动识别并跳过目录页和封面页。这些处理能明显提升检索质量，但需要针对你的语料特点来写，通用方案效果有限。
第三个方向是加一层导入后的效果验证。准备一批"问题—期望答案"的测试集，每次导入完成后自动跑一遍检索，看命中率有没有下降。如果某次导入之后命中率明显掉了，很可能是那批新文件里混进了大量低质量内容，把检索结果稀释了。这个环节我目前还在搭，但思路是清楚的：导入不是终点，能用起来才是。
最后说个实在话。这套东西从有想法到跑通，前后大概花了三个周末，其中真正写代码的时间不到三分之一，剩下的都花在调试参数、处理各种畸形文件、看日志找问题上。AI 帮你把键盘活干了，但"这份文件为什么解析出来是空的""这个 429 到底是并发高还是密钥限速"这类判断，还得你自己拿日志一点点啃。脚本能跑通的那一刻确实爽，但爽之前的那段排查，才是真正长本事的部分。
