Dify 官方提供了网页拖拽上传，但一次只Neng处理几百条记录。当你的知识库Yi经突破千级别，就需要借助 API 批量写入。本节代码Yi经经过实测，可直接拷贝到项目根目录并执行。

import os, json, time, requests
# ------------------- 参数开始 -------------------
BASE_URL   = "http://localhost/v1"                 # Dify API 根地址，默认就是这个
API_KEY     = "YOUR_DIFY_API_KEY"                  # 在 Dify → Settings → API Keys 中生成的新钥匙
DATASET_ID = "YOUR_DATASET_UUID"                   # 浏览器打开某个 Dataset 时 URL 中出现的那串 UUID
LOCAL_FOLDER = r"C:\Users\Me\Documents\MyKnowledge"   # 本地存放文档的根目录，请使用原始字符串或双反斜杠转义
# ------------------- 参数结束 -------------------
headers = {"Authorization": f"Bearer {API_KEY}"}

def fetch_existing_names:
    """查询当前 Dataset Yi经有哪些文件，用来避免重复"""
    names = set
    page = 1          # 第几页，从 1 开始计数
    limit = 50        # 每页返回多少条记录，可自行调大
    print
    while True:
        try:
            url = f"{BASE_URL}/datasets/{DATASET_ID}/documents?page={page}&limit={limit}"
            resp = requests.get
            if resp.status_code != 200:
                print
                break
            data = resp.json
            docs = data.get
            if not docs:
                break
            for d in docs:
                names.add
            if not data.get:
                break
            page += 1
            print
        except Exception as e:
            print
            break
    print} 条记录")
------
return names
def upload_one:
    file_name = os.path.basename
    url = f"{BASE_URL}/datasets/{DATASET_ID}/document/create-by-file"
    payload = {
        "indexing_technique": "high_quality",   # 可切换为 economy 来加速但牺牲精度
        "process_rule": {"mode": "automatic"} # 自动分段，无需手工标记章节标题等
    }
    with open as f:
        files = {
            "file": ,
            "data": )
        }
        resp = requests.post
    if resp.status_code == 200:
        print
        return True
    else:
        print
        return False
------

def main:
    existing = fetch_existing_names
    for root, _, files in os.walk:
        for fn in files:
            # 排除隐藏文件和非文本格式，仅保留常见办公文档类型。
            if fn.startswith or not fn.lower.endswith):
                continue
            if fn in existing:
                print
                continue
            full_path = os.path.join
            upload_one
            # 为防止短时间内请求过多导致 Ollama 卡顿，加一点点延迟。
            time.sleep
if __name__ == "__main__":
    main
------

* 小提示：Ru果你希望一次性导入上万篇文档，Ke以把上述脚本 成多线程模式，只要注意不要让 Docker 内部的 CPU 被压垮即可。