用 RAGFlow API 自动建库、上传文档并跟踪解析

用网页一个个上传文件很慢，也不好复现。 这篇做 rf_ingest.py：用官方 HTTP API 一条命令完成“建知识库 → 上传文件 → 开始解析 → 等待完成 → 打印每个文件的块数或失败原因”，同时兼容 RAGFlow 1.0 和之前的版本。

先看清成品

• 对 demo.ragflow.io 真实运行：第一步的只读探测请求就因无效 Key 停止（HTTP 401），没有执行任何写操作。
• 根据官方文档处理了两处容易出错的地方：建库字段 1.0 改为 parser_id（旧版是 chunk_method）；文档状态 run 有 "UNSTART"、"1"、整数等多种写法。
• 6 项 pytest 通过（两种后端的建库字段、各种 run 写法、多文件上传）。

合规提醒：RAGFlow 是 Apache 2.0 开源项目。本文只用官方 HTTP API 和你自己实例的 API Key（在 RAGFlow 网页的 API 设置中创建）。Key 只放服务端。demo.ragflow.io 是官方的公开演示站，我只用无效 Key 对它发了几个请求来确认错误格式，没有注册、上传或压测；不要把它当成自己的生产服务，也不要上传敏感文档。

运行说明：RAGFlow 官方推荐至少 4 核 CPU、16 GB 内存、50 GB 磁盘，并用 Docker 部署。box 没有 Docker 权限，所以没有部署 RAGFlow。真实跑过的是：用无效 Key 调用 demo.ragflow.io（2026-10-21 21:31–21:33，UTC+8，返回原样贴在文中），以及全部本地代码。正常返回来自我写的 mock，字段照官方 HTTP API 参考。版本提醒：2026-10-16 发布了 v1.0.0-rc2，main 分支的 API 文档已经按 1.0（Go 后端）写，有几个旧接口被移除；demo 站的响应没有 1.0 才有的 X-API-Source: go 头。先确认你的版本，再看对应版本的文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

84-ragflow-B-src/
├── rf_ingest.py
├── docs/
│   ├── IT指南.md
│   └── 员工手册.md
└── test_rf_ingest.py

第 1 步：脚本

"""用 RAGFlow HTTP API 建知识库 → 上传文件 → 解析 → 等待完成，并打印每个文件的结果。

兼容两代后端：RAGFlow 1.0（Go，响应头 X-API-Source: go）建库用 parser_id="general"；
之前的 Python 后端用 chunk_method="naive"。先发一个只读请求探测后端，再决定字段。"""
import argparse, os, pathlib, sys, time
import requests

UA = "binbho-tutorial/1.0"
RUN = {"0": "UNSTART", "1": "RUNNING", "2": "CANCEL", "3": "DONE", "4": "FAIL", "5": "SCHEDULE"}
FINAL = {"DONE", "FAIL", "CANCEL"}


def norm_run(v):
    """官方文档里 run 既有 "UNSTART" 这种文字，也有 "1" 这种数字字符串，还可能是整数。"""
    s = str(v).strip().upper()
    return RUN.get(s, s)


class Client:
    def __init__(self, base, key, session=None):
        if not key.strip():
            raise SystemExit("缺少 RAGFLOW_API_KEY")
        self.base, self.s = base.rstrip("/"), session or requests.Session()
        self.s.headers.update({"Authorization": f"Bearer {key.strip()}", "User-Agent": UA})
        self.go = None

    def _do(self, method, path, **kw):
        r = self.s.request(method, self.base + path, timeout=120, **kw)
        if self.go is None:
            self.go = r.headers.get("X-API-Source") == "go"
        d = r.json()
        if r.status_code != 200 or d.get("code") != 0:
            raise SystemExit(f"{method} {path} 失败：HTTP {r.status_code} code={d.get('code')} {d.get('message')}")
        return d.get("data")

    def detect(self):
        self._do("GET", "/api/v1/datasets", params={"page": 1, "page_size": 1})
        return "go" if self.go else "python"

    def create_dataset(self, name):
        body = {"name": name, **({"parser_id": "general"} if self.go else {"chunk_method": "naive"})}
        return self._do("POST", "/api/v1/datasets", json=body)["id"]

    def upload(self, ds, paths):
        files = [("file", (p.name, p.read_bytes(), "text/markdown")) for p in paths]
        return [d["id"] for d in self._do("POST", f"/api/v1/datasets/{ds}/documents", files=files)]

    def parse(self, ds, doc_ids):
        self._do("POST", f"/api/v1/datasets/{ds}/chunks", json={"document_ids": doc_ids})

    def docs(self, ds):
        return self._do("GET", f"/api/v1/datasets/{ds}/documents", params={"page": 1, "page_size": 100})["docs"]


def wait(client, ds, doc_ids, interval=5, max_wait=900, sleep=time.sleep):
    waited = 0
    while True:
        docs = {d["id"]: d for d in client.docs(ds) if d["id"] in doc_ids}
        states = {i: norm_run(docs[i]["run"]) for i in doc_ids if i in docs}
        if len(states) == len(doc_ids) and all(s in FINAL for s in states.values()):
            return docs
        if waited >= max_wait:
            raise SystemExit(f"{max_wait}s 未完成：{states}")
        sleep(interval); waited += interval


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("doc_dir"); ap.add_argument("--name", default="员工制度")
    a = ap.parse_args(argv)
    c = Client(os.environ.get("RAGFLOW_BASE_URL", "http://localhost"), os.environ.get("RAGFLOW_API_KEY", ""))
    print("后端：", c.detect())
    paths = sorted(pathlib.Path(a.doc_dir).glob("*.md"))
    ds = c.create_dataset(a.name); print("知识库：", ds)
    ids = c.upload(ds, paths); print("已上传：", len(ids), "个文件")
    c.parse(ds, ids); print("开始解析…")
    bad = 0
    for d in wait(c, ds, ids).values():
        st = norm_run(d["run"]); bad += st != "DONE"
        print(f"  {d['name']}: {st}，{d.get('chunk_count', 0)} 块" + ("" if st == "DONE" else f"，{d.get('progress_msg')}"))
    return 1 if bad else 0


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

第 2 步：测试（mock）

"""假 RAGFlow（我写的，字段照官方 HTTP API 参考）。"""
import pathlib
import pytest
import rf_ingest as g


class R:
    def __init__(self, data, go=False, code=0, http=200):
        self.status_code, self.headers = http, ({"X-API-Source": "go"} if go else {})
        self._b = {"code": code, "data": data} if code == 0 else {"code": code, "message": "err"}
    def json(self):
        return self._b


class S:
    def __init__(self, script):
        self.script, self.calls, self.headers = list(script), [], {}
    def request(self, method, url, timeout=None, **kw):
        self.calls.append((method, url, kw)); return self.script.pop(0)


def test_norm_run_all_forms():
    assert [g.norm_run(x) for x in ("UNSTART", "1", 3, "done", " 4 ", "SCHEDULE")] == ["UNSTART", "RUNNING", "DONE", "DONE", "FAIL", "SCHEDULE"]


@pytest.mark.parametrize("go,field,value", [(True, "parser_id", "general"), (False, "chunk_method", "naive")])
def test_create_uses_right_field(go, field, value):
    s = S([R([], go=go), R({"id": "ds1"}, go=go)])
    c = g.Client("http://h", "k", session=s)
    assert c.detect() == ("go" if go else "python")
    assert c.create_dataset("x") == "ds1"
    body = s.calls[1][2]["json"]
    assert body[field] == value and len(body) == 2


def test_wait_handles_numeric_string_run():
    docs1 = {"docs": [{"id": "a", "name": "a.md", "run": "1"}, {"id": "b", "name": "b.md", "run": "UNSTART"}]}
    docs2 = {"docs": [{"id": "a", "name": "a.md", "run": "3", "chunk_count": 4}, {"id": "b", "name": "b.md", "run": "FAIL", "progress_msg": "parse error"}]}
    c = g.Client("http://h", "k", session=S([R(docs1), R(docs2)]))
    out = g.wait(c, "ds", ["a", "b"], sleep=lambda x: None)
    assert g.norm_run(out["a"]["run"]) == "DONE" and g.norm_run(out["b"]["run"]) == "FAIL"


def test_upload_multipart():
    s = S([R([{"id": "d1"}, {"id": "d2"}])])
    c = g.Client("http://h", "k", session=s)
    ids = c.upload("ds", sorted(pathlib.Path("docs").glob("*.md")))
    assert ids == ["d1", "d2"] and [f[1][0] for f in s.calls[0][2]["files"]] == ["IT指南.md", "员工手册.md"]


def test_business_error_exits():
    c = g.Client("http://h", "k", session=S([R(None, code=102, http=200)]))
    with pytest.raises(SystemExit, match="code=102"):
        c.detect()

第 3 步：运行

$ RAGFLOW_BASE_URL=https://demo.ragflow.io RAGFLOW_API_KEY=ragflow-invalid-demo python rf_ingest.py docs   # 2026-10-21 21:33 UTC+8
GET /api/v1/datasets 失败：HTTP 401 code=401 <Unauthorized '401: Unauthorized'>
$ echo $?
1
$ python -m pytest -q
6 passed

# 在自己的实例上预期的输出格式（我没跑过）：
后端： go
知识库： <dataset_id>
已上传： 2 个文件
开始解析…
  IT指南.md: DONE，<n> 块
  员工手册.md: DONE，<n> 块

怎么检查结果？

• 解析完成后，在 RAGFlow 网页打开知识库，逐个文件看切块结果。
• FAIL 的文件看 progress_msg；常见原因是格式不支持或嵌入模型没配置好。
• 接着用 A 篇的 rf_retrieve.py 问几个问题，确认能检索到。

调试示例

官方文档里的两处不一致。 写这个脚本时我对照的是 main 分支的 HTTP API 参考，发现两处如果照抄会出错：

1. 建库字段。 “Create dataset”一节的请求体仍然列着 chunk_method，响应示例也是 "chunk_method": "naive"；但同一份文档开头的“Migrate to 1.0”表格写着：1.0 建库和更新用 parser_id，通用解析器传 "general"，“Replace chunk_method with parser_id”。所以脚本先发一个只读的 GET /api/v1/datasets 探测后端（看 X-API-Source: go），再决定用哪个字段。实测中这一步就因 401 停下了，没有执行任何写操作。

2. run 的写法。 List documents 的示例返回 "run": "UNSTART"，Ingest documents 的示例却是 "run": "1"，状态说明里又同时列出数字和文字两套写法。如果只判断 run == "DONE"，遇到 "3" 就会一直等到超时。norm_run() 把数字、数字字符串、大小写不同的文字统一成 UNSTART/RUNNING/…，测试 test_norm_run_all_forms 覆盖了这些写法。

常见问题

模板规则

• 先只读探测，再写。
• 状态值要先归一化。
• 每个文件的结果都打印出来。
• 对照自己版本的文档。

项目资源

能用一条命令重建知识库，调参数时就不怕推倒重来。