用 FastGPT 构建可审查的知识库导入流程：Markdown 切块与重复检查

知识库问答好不好，很大程度取决于导入的文档怎么切块。 这篇做 kb_import.py：先在本地按标题把 Markdown 切块，每块都带上所属标题，表格整块保留；检查重复片段和过长片段，生成 import_plan.json 给人审；确认后再调用官方接口创建文本集合。

先看清成品

• 在示例文档上真实运行：2 个文件各 4 块，查出 1 个重复片段（两份文档都写了远程办公规定）和 1 个过长的表格块。
• 发现自己的真实问题：第一版有警告也照样导入，已改为有警告时拒绝，除非加 --force。
• 用无效 Key 真实调用云服务的导入接口，返回同样是 HTTP 500 + 514。6 项 pytest 通过。

合规提醒：只用 FastGPT 官方 OpenAPI 和你自己账号的 API Key。官方文档说明 API Key“相当于当前账号在当前团队下的访问凭证”，所以只能放在服务端，不要写进网页或 App。开源版是 Apache 2.0 加附加条件：可以商用，但做类似 FastGPT 的多租户 SaaS、或去掉控制台的 LOGO/版权信息，需要另外取得商业授权（以仓库 LICENSE 为准）。上传到知识库的文档要有权使用，不放个人隐私。

运行说明：box 不能用 Docker，也没有 FastGPT 账号，所以没有部署或调用真实的 FastGPT 应用。真实跑过的是：用无效 Key 调用 FastGPT 云服务（cloud.fastgpt.cn 和 cloud.fastgpt.io，2026-10-21 21:26–21:28，UTC+8），返回原样贴在文中；以及全部本地代码。导入成功的返回来自我写的 mock，结构照官方文档 detail=true, stream=false 的响应示例；官方也说明不同版本 responseData 的字段可能不同。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

83-fastgpt-B-src/
├── kb_import.py
├── docs/
│   ├── IT指南.md
│   └── 员工手册.md
└── test_kb_import.py

第 1 步：示例文档

<!-- docs/员工手册.md -->
# 员工手册（示例，内容虚构）

## 年假

入职满一年后，每年享有 10 天带薪年假。年假需提前 3 个工作日在系统中申请，经直属主管批准后生效。
当年未休完的年假可顺延至次年 3 月 31 日，逾期作废。

## 报销

| 类别 | 标准 | 需要的材料 |
| :--- | :--- | :--- |
| 市内交通 | 实报实销 | 电子发票 |
| 出差住宿 | 一线城市每晚 500 元以内，其他城市 350 元以内 | 酒店发票、水单 |
| 出差餐补 | 每天 100 元 | 无需发票 |
| 办公用品 | 单笔 200 元以内 | 电子发票、购买截图 |

报销需在费用发生后 30 天内提交，超过 30 天需部门负责人特批。

## 远程办公

每周最多 2 天远程办公，需提前一天在团队日历登记。远程办公期间需保持即时通讯在线。
<!-- docs/IT指南.md -->
# IT 指南（示例，内容虚构）

## 账号

新员工入职当天由 IT 发放账号，首次登录必须修改密码并开启两步验证。

## 年假

年假相关问题请查看员工手册，IT 不处理年假申请。

## 电脑

笔记本电脑使用 3 年后可申请更换。丢失电脑须在 24 小时内报告 IT 和直属主管。

## 远程办公

每周最多 2 天远程办公，需提前一天在团队日历登记。远程办公期间需保持即时通讯在线。

第 2 步：导入脚本

"""把本地 Markdown 文档按标题切块，检查后导入 FastGPT 知识库（集合 = 文件）。

默认只做 dry-run，生成 import_plan.json 给人看；加 --apply 才会调用
POST /api/core/dataset/collection/create/text 。
"""
import argparse, hashlib, json, os, pathlib, re, sys
import requests

UA = "binbho-tutorial/1.0"


def split_markdown(text, max_chars=400):
    """按 ## 标题切段；段落太长再按空行切；表格整块保留，并在每块前面带上标题。"""
    chunks, title, buf = [], "", []

    def flush():
        body = "\n".join(buf).strip()
        if body:
            for piece in pack(body, max_chars):
                chunks.append(f"{title}\n{piece}".strip() if title else piece)
        buf.clear()

    for line in text.splitlines():
        if line.startswith("## "):
            flush(); title = line.strip()
        elif line.startswith("# "):
            continue          # 文档大标题不进正文
        else:
            buf.append(line)
    flush()
    return chunks


def pack(body, max_chars):
    blocks = re.split(r"\n\s*\n", body)
    out, cur = [], ""
    for b in blocks:
        if cur and len(cur) + len(b) + 2 > max_chars:
            out.append(cur); cur = b
        else:
            cur = f"{cur}\n\n{b}" if cur else b
    if cur:
        out.append(cur)
    return out


def plan(doc_dir, max_chars):
    files, seen, warnings = [], {}, []
    for p in sorted(pathlib.Path(doc_dir).glob("*.md")):
        chunks = split_markdown(p.read_text(encoding="utf-8"), max_chars)
        for c in chunks:
            h = hashlib.sha1(c.split("\n", 1)[-1].encode()).hexdigest()[:10]
            if h in seen:
                warnings.append(f"{p.name} 与 {seen[h]} 有完全相同的片段：{c[:30]!r}")
            seen.setdefault(h, p.name)
            if len(c) > max_chars * 1.5:
                warnings.append(f"{p.name} 片段过长（{len(c)} 字），检查是否有超大表格：{c[:30]!r}")
        files.append({"name": p.name, "chunks": chunks, "text": "\n\n".join(chunks)})
    return files, warnings


def apply(files, base, key, dataset_id, max_chars):
    for f in files:
        r = requests.post(base.rstrip("/") + "/core/dataset/collection/create/text", timeout=120,
                          headers={"Authorization": f"Bearer {key.strip()}", "User-Agent": UA},
                          json={"text": f["text"], "datasetId": dataset_id, "parentId": None, "name": f["name"],
                                "trainingType": "chunk", "chunkSettingMode": "auto", "metadata": {}})
        d = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        if r.status_code != 200 or d.get("code") != 200:
            raise SystemExit(f"{f['name']} 导入失败：HTTP {r.status_code} {d.get('code')} {d.get('statusText')}")
        print(f"已导入 {f['name']} → collectionId={d['data']['collectionId']}")


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("doc_dir"); ap.add_argument("--max-chars", type=int, default=400)
    ap.add_argument("--apply", action="store_true")
    ap.add_argument("--force", action="store_true", help="有警告也导入")
    a = ap.parse_args(argv)
    files, warnings = plan(a.doc_dir, a.max_chars)
    pathlib.Path("import_plan.json").write_text(json.dumps(
        [{"name": f["name"], "chunks": f["chunks"]} for f in files], ensure_ascii=False, indent=2), encoding="utf-8")
    for f in files:
        print(f"{f['name']}: {len(f['chunks'])} 块，最长 {max(map(len, f['chunks']))} 字")
    for w in warnings:
        print("⚠", w)
    if not a.apply:
        print("dry-run：已写 import_plan.json。确认无误后加 --apply 导入。"); return 0
    if warnings and not a.force:
        print("有警告，未导入。处理后重跑，或确认无妨后加 --force。", file=sys.stderr); return 2
    apply(files, os.environ.get("FASTGPT_BASE_URL", "http://localhost:3000/api"),
          os.environ["FASTGPT_API_KEY"], os.environ["FASTGPT_DATASET_ID"], a.max_chars)
    return 0


if __name__ == "__main__":
    sys.exit(main())

第 3 步：测试

import json
import pytest
import kb_import as k

DOC = "# 标题\n\n## A\n\n第一段。\n\n第二段。\n\n## 表\n\n| a | b |\n| :- | :- |\n| 1 | 2 |\n| 3 | 4 |\n"


def test_split_keeps_heading_and_table():
    cs = k.split_markdown(DOC, max_chars=10)
    assert cs[0].startswith("## A") and all(not c.startswith("# 标题") for c in cs)
    table = [c for c in cs if "| a | b |" in c]
    assert len(table) == 1 and "| 3 | 4 |" in table[0] and table[0].startswith("## 表")


def test_pack_merges_small_blocks():
    assert k.pack("a\n\nb\n\nc", 100) == ["a\n\nb\n\nc"]
    assert k.pack("aaaa\n\nbbbb", 5) == ["aaaa", "bbbb"]


def test_plan_on_sample_docs():
    files, warnings = k.plan("docs", 120)
    assert [f["name"] for f in files] == ["IT指南.md", "员工手册.md"]
    assert any("完全相同" in w for w in warnings) and any("过长" in w for w in warnings)


class R:
    def __init__(self, code, body):
        self.status_code, self._b, self.headers = code, body, {"content-type": "application/json; charset=utf-8"}
    def json(self):
        return self._b


def test_apply_payload_and_error(monkeypatch):
    sent = []
    monkeypatch.setattr(k.requests, "post", lambda url, **kw: sent.append((url, kw)) or R(200, {"code": 200, "statusText": "", "message": "", "data": {"collectionId": "c1", "results": {"insertLen": 3}}}))
    k.apply([{"name": "a.md", "text": "t"}], "http://h/api/", "key\n", "ds1", 400)
    url, kw = sent[0]
    assert url == "http://h/api/core/dataset/collection/create/text"
    assert kw["json"]["trainingType"] == "chunk" and kw["json"]["datasetId"] == "ds1"
    assert kw["headers"]["Authorization"] == "Bearer key"
    monkeypatch.setattr(k.requests, "post", lambda url, **kw: R(500, {"code": 514, "statusText": "unAuthApiKey"}))
    with pytest.raises(SystemExit, match="514"):
        k.apply([{"name": "a.md", "text": "t"}], "http://h/api", "bad", "ds1", 400)


def test_apply_refused_when_warnings(monkeypatch, tmp_path):
    monkeypatch.chdir(tmp_path)
    (tmp_path / "d").mkdir()
    (tmp_path / "d" / "a.md").write_text("## X\n\n同一段\n", encoding="utf-8")
    (tmp_path / "d" / "b.md").write_text("## X\n\n同一段\n", encoding="utf-8")
    monkeypatch.setattr(k.requests, "post", lambda *a, **kw: pytest.fail("不应调用 API"))
    assert k.main([str(tmp_path / "d"), "--apply"]) == 2

第 4 步：运行

$ python kb_import.py docs --max-chars 120
IT指南.md: 4 块，最长 50 字
员工手册.md: 4 块，最长 181 字
⚠ 员工手册.md 片段过长（181 字），检查是否有超大表格：'## 报销\n| 类别 | 标准 | 需要的材料 |\n| :-'
⚠ 员工手册.md 与 IT指南.md 有完全相同的片段：'## 远程办公\n每周最多 2 天远程办公，需提前一天在团队日'
dry-run：已写 import_plan.json。确认无误后加 --apply 导入。

# 有警告时 --apply 会被拒绝（21:28 UTC+8）
$ python kb_import.py docs --max-chars 120 --apply
…
有警告，未导入。处理后重跑，或确认无妨后加 --force。
$ echo $?
2

# 加 --force，用无效 Key 调用云服务
$ FASTGPT_BASE_URL=https://cloud.fastgpt.cn/api FASTGPT_API_KEY=fastgpt-invalid-demo FASTGPT_DATASET_ID=000000000000000000000000 python kb_import.py docs --max-chars 120 --apply --force
…
IT指南.md 导入失败：HTTP 500 514 unAuthApiKey
$ python -m pytest -q
6 passed

怎么检查结果？

• 打开 import_plan.json 逐块读：每块单独拿出来，是否还能看懂（有没有带标题）？
• 重复片段：决定保留哪一份。两份文档说法不一致时，问答结果会互相矛盾。
• 导入后在 FastGPT 里用“搜索测试”问几个问题，看命中的是不是你期望的块（C 篇自动化这一步）。
• --max-chars 只控制本地切块；用 chunk 模式导入时 FastGPT 还会再按自己的设置分割。

调试示例

1. 有警告也照样导入。 第一版的流程是“打印警告 → 导入”。用 --apply 跑时，屏幕上刚出现“完全相同的片段”警告，紧接着就开始调用接口（这次因为 Key 无效失败了，否则重复内容已经进了知识库）。警告混在输出里很容易被忽略，所以改成：有警告时退出码 2、不调用接口，确认无妨后显式加 --force。测试 test_apply_refused_when_warnings 会在调用 API 时直接让测试失败。

2. 表格块为什么超长。 报销表格 181 字，超过 --max-chars 120 的 1.5 倍。这是故意的：表格按行切开后，“出差住宿 500 元”这一行就失去了表头，检索到也看不懂。所以表格整块保留、只提示，由人决定要不要拆成几个小表。

3. 文件名里的空格。 我第一次创建示例文件时用了 cat > docs/IT 指南.md，shell 把它拆成两个参数，生成了一个叫 IT 的文件，并报 指南.md: No such file or directory。改成 IT指南.md；文件名会作为集合名显示在 FastGPT 里，不带空格也更好处理。

常见问题

模板规则

• 先 dry-run，人看过再导入。
• 每块带上所属标题。
• 表格整块保留。
• 有警告就停。

项目资源

切块是知识库质量的地基，花时间看一遍 import_plan.json 很值。