build_dify_import_pack.py 4.7 KB
#!/usr/bin/env python3
"""
生成一套更适合直接导入 Dify / 通用 RAG 平台的中颗粒度知识包。

设计原则:
1. 保留公共文件。
2. 每个模块尽量保持为一个主文件。
3. 交给 Dify 在导入时继续做内部 chunk,而不是预先切成大量零散片段。

输入:
  dist/usable_kb/*.md
  inputs/priority_refs/*.md

输出:
  dist/dify_import/
"""

from __future__ import annotations

from pathlib import Path


BASE_DIR = Path(__file__).parent.parent
SRC_DIR = BASE_DIR / "dist" / "usable_kb"
PRIORITY_REF_DIR = BASE_DIR / "inputs" / "priority_refs"
OUT_DIR = BASE_DIR / "dist" / "dify_import"

COMMON_FILES = [
    "01_知识库使用规则.md",
    "02_版本变更总览.md",
    "03_需求预评审执行指南.md",
    "04_后台实现导读.md",
]

MODULE_FILES = [
    "10_AUTH_认证.md",
    "11_INCOME_收入提现.md",
    "12_INQUIRY_问诊.md",
    "13_CLINIC_门诊.md",
    "14_PATIENT_患者.md",
    "15_NOTIFICATION_通知.md",
    "16_BACKSTAGE_后台.md",
    "17_GENERAL_通用.md",
]

RENAME_MAP = {
    "01_知识库使用规则.md": "01_使用规则.md",
    "10_AUTH_认证.md": "10_AUTH_主知识库.md",
    "11_INCOME_收入提现.md": "11_INCOME_主知识库.md",
    "12_INQUIRY_问诊.md": "12_INQUIRY_主知识库.md",
    "13_CLINIC_门诊.md": "13_CLINIC_主知识库.md",
    "14_PATIENT_患者.md": "14_PATIENT_主知识库.md",
    "15_NOTIFICATION_通知.md": "15_NOTIFICATION_主知识库.md",
    "16_BACKSTAGE_后台.md": "16_BACKSTAGE_主知识库.md",
    "17_GENERAL_通用.md": "17_GENERAL_主知识库.md",
}


def write_text(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8")


def copy_file(name: str) -> str | None:
    src = SRC_DIR / name
    if not src.exists():
        return None
    dest_name = RENAME_MAP.get(name, name)
    write_text(OUT_DIR / dest_name, src.read_text(encoding="utf-8"))
    return dest_name


def copy_priority_refs() -> list[str]:
    copied = []
    if not PRIORITY_REF_DIR.exists():
        return copied
    for src in sorted(PRIORITY_REF_DIR.glob("*.md")):
        write_text(OUT_DIR / src.name, src.read_text(encoding="utf-8"))
        copied.append(src.name)
    return copied


def build_manifest(common: list[str], modules: list[str], priority_refs: list[str]) -> str:
    lines = [
        "# Dify 导入包说明",
        "",
        "这套文件从 `dist/usable_kb` 重新整理而来,采用中颗粒度组织:",
        "- 公共规则、版本总览、预评审指南单独成文件",
        "- 每个模块尽量保留为一个主文件",
        "- 不再预先切成大量零散主题片段",
        "- 由 Dify 在导入时继续做内部分段",
        "",
        "推荐导入顺序:",
        "1. `00_导入说明.md`",
    ]
    for idx, name in enumerate(common, start=2):
        lines.append(f"{idx}. `{name}`")
    start = len(common) + 2
    for offset, name in enumerate(modules, start=start):
        lines.append(f"{offset}. `{name}`")
    start = len(common) + len(modules) + 2
    for offset, name in enumerate(priority_refs, start=start):
        lines.append(f"{offset}. `{name}`")
    lines.extend(
        [
            "",
            "使用建议:",
            "- 先导入模块主文件和高优先参考文件,跑通召回。",
            "- 如果后续发现问答和预评审互相干扰,再拆成多个知识库。",
            "- 模块主文件已经带产品规则、交互补充、测试边界和后台实现线索。",
            "- 高优先参考文件适合放入对应产品知识库,作为人工整理后的强参考。",
            "",
            f"- 公共文件数:{len(common)}",
            f"- 模块主文件数:{len(modules)}",
            f"- 高优先参考文件数:{len(priority_refs)}",
            f"- 总文件数:{1 + len(common) + len(modules) + len(priority_refs)}",
        ]
    )
    return "\n".join(lines) + "\n"


def main() -> None:
    if not SRC_DIR.exists():
        raise SystemExit(f"src_not_found: {SRC_DIR}")

    OUT_DIR.mkdir(parents=True, exist_ok=True)
    for old in OUT_DIR.glob("*"):
        if old.is_file():
            old.unlink()

    copied_common = [name for name in (copy_file(item) for item in COMMON_FILES) if name]
    copied_modules = [name for name in (copy_file(item) for item in MODULE_FILES) if name]
    copied_priority_refs = copy_priority_refs()
    write_text(OUT_DIR / "00_导入说明.md", build_manifest(copied_common, copied_modules, copied_priority_refs))

    print(f"output={OUT_DIR.relative_to(BASE_DIR)}")
    print(f"common_files={len(copied_common)}")
    print(f"module_files={len(copied_modules)}")
    print(f"priority_ref_files={len(copied_priority_refs)}")


if __name__ == "__main__":
    main()