入口节点与 PDF 转 Markdown 节点

本文档详细介绍知识库导入流程的前两个业务节点：
入口节点（Entry）负责文件类型检测与路由
PDF 转 Markdown 节点负责将 PDF 文档转换为 Markdown 格式。

学习理念：入口节点和 PDF 转换是导入流程的"大门"——EntryNode 负责文件类型检测和路由分发，PdfToMdNode 负责将 PDF 解析为可处理的 Markdown 格式。理解这两个节点就掌握了导入流程的启动机制和外部工具调用模式。
海外对标：EntryNode 的条件路由设计对标 Amazon S3 的事件触发路由（根据文件后缀触发不同处理流程）；PdfToMdNode 的 MinerU 子进程调用模式对标 Google Document AI 的 OCR 服务封装。subprocess.Popen 的实时日志采集模式在 LangChain Inc 的文档解析 Pipeline 中也有类似实现。

本节 AI 替代率：~85% | 人工干预率：~15%

角色	能力范围
🤖 AI 擅长	生成 EntryNode 的文件类型检测逻辑、PdfToMdNode 的 MinerU 调用骨架、subprocess 配置模板、路径验证代码
👤 人类需理解	条件路由的工程决策（哪些文件类型需要区分）、MinerU 模型本地部署的路径配置、子进程超时和 OOM 的处理策略

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习目标明确，AI 可完整规划
🟢	§2 核心概念扫盲	~95%	~5%	pathlib/subprocess/MinerU 都是 API 级知识
🟡	§3 整体流程	~90%	~10%	理解 Entry → PdfToMd 的数据流转即可
🔴	§4.1 入口节点	~80%	~20%	路由逻辑的工程决策（哪些类型路由到哪）需人工确认
🟠	§4.2 PDF 转 Markdown	~70%	~30%	MinerU 子进程调用参数 + 输出路径拼接需人工调试
🟢	§5 测试运行	~95%	~5%	main 驱动模式固定，看预期输出即可
🟡	§6 总结	~90%	~10%	设计要点回顾

技术栈健康度标签体系

技术	健康度	建议
LangGraph	⏳ 成长期	本项目的核心编排框架，EntryNode + PdfToMdNode 作为其子类运行。API 仍在迭代中，建议锁定版本。
pathlib.Path	🔥 巅峰	Python 路径处理的行业标准，比 `os.path` 更优雅，推荐所有新项目使用。
subprocess	🔥 巅峰	Python 调用外部命令的标准方式。`Popen` + `PIPE` 模式在 AI 项目（调用模型推理脚本等）中广泛应用。
MinerU	⏳ 成长期	开源 PDF 解析工具，基于 PDF-Extract-Kit 和模型推理。适合学术文档解析，复杂排版、扫描件仍需人工校验。竞品有 Marker（🔥巅峰）、PyMuPDF4LLM（⏳成长期）。
BaseNode	🟢 稳定	项目自定义的节点基类，统一执行流程（日志 + 任务追踪 + 异常包装）。模式固定，按模板继承即可。

体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。

中英文对照表

English	中文	本质
Entry Node	入口节点	导入流程的起点，负责文件类型检测和路由分发
Conditional Edge	条件边	LangGraph 中根据状态值决定下一个节点的路由机制
subprocess	子进程	Python 启动外部命令行工具的模块
MinerU	PDF 解析工具	开源 PDF 转 Markdown 工具，基于深度学习模型
PIPE	管道	子进程的标准输出/输入通道，用于捕获命令输出
Stem	文件名主干	文件名中不含扩展名的部分（如 `doc.pdf` → `doc`）
ValidationError	验证错误	输入数据不符合预期时抛出的异常
FileProcessingError	文件处理错误	文件不存在或格式错误时抛出的异常

💡 程序员比喻
EntryNode 就像 Nginx 的反向代理——根据请求后缀（.pdf / .md）分发到不同的 upstream（PDF 转换 / MD 处理）。
PdfToMdNode 调用 MinerU 就像 CI/CD 中调用 docker build——启动一个外部进程，实时捕获日志，检查返回码。
subprocess.Popen(buffsize=1) 就像 kubectl logs -f——实时 tail 日志，不是等全部跑完才看到。
pathlib.Path / 运算符 就像 Go 的 path.Join——语法糖让路径拼接不再有 os.path.join(a, b, c) 的嵌套地狱。

1. 任务目标

1.1 本章目标

通过本章学习，你将掌握：

入口节点设计：理解文件类型检测与条件路由的实现方式
PDF 转换流程：掌握使用 MinerU 进行 PDF 解析的完整流程
子进程调用：学会使用 subprocess 模块调用外部命令行工具
异常处理：针对文件处理场景的异常设计
单元测试：为每个节点编写 if __name__ == "__main__" 测试

1.2 涉及模块

knowledge/processor/import_process/nodes/
├── __init__.py      # 节点模块导出
├── entry.py         # 入口节点
└── pdf_to_md.py     # PDF 转 Markdown 节点

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 pathlib.Path

Python 3.4+ 引入的面向对象路径处理库，比 os.path 更优雅：

🟢 【P2 后面可以查】 pathlib 是 Python 标准库，API 直观。理解 path.name / path.suffix / / 运算符即可，需要时查文档。

python

from pathlib import Path

# 创建 Path 对象
path = Path("D:/documents/万用表的使用.pdf")

# 常用属性
path.name       # "万用表的使用.pdf" - 完整文件名
path.stem       # "万用表的使用" - 不含扩展名的文件名
path.suffix     # ".pdf" - 扩展名
path.parent     # Path("D:/documents") - 父目录

# 常用方法
path.exists()   # True/False - 文件是否存在
path.is_file()  # True/False - 是否为文件
path.is_dir()   # True/False - 是否为目录

# 路径拼接（使用 / 运算符）
output_path = path.parent / "output" / "result.md"

2.2 subprocess 子进程调用

subprocess 模块用于调用外部命令行工具：

🟡 【P1 看注释就行】 subprocess 模板代码固定——Popen(cmd, stdout=PIPE, text=True, bufsize=1)。看懂 PIPE + STDOUT + bufsize=1 这三个参数的作用即可。AI 可完整生成此类模板。

python

import subprocess

# 基础用法：run() - 同步执行，等待完成
result = subprocess.run(
    ["mineru", "-p", "input.pdf", "-o", "output/"],
    capture_output=True,  # 捕获输出
    text=True,            # 文本模式
    check=True            # 非零返回码抛出异常
)
print(result.stdout)

# 高级用法：Popen() - 实时获取输出
proc = subprocess.Popen(
    ["mineru", "-p", "input.pdf"],
    stdout=subprocess.PIPE,    # 捕获标准输出
    stderr=subprocess.STDOUT,  # 合并错误到标准输出
    text=True,
    bufsize=1  # 行缓冲，实时输出
)

# 逐行读取输出
for line in proc.stdout:
    print(line.rstrip())

# 等待完成
return_code = proc.wait()

关键参数说明：

参数	说明
`stdout=subprocess.PIPE`	捕获标准输出
`stderr=subprocess.STDOUT`	将错误流合并到标准输出
`text=True`	文本模式（自动解码为字符串）
`encoding="utf-8"`	指定编码
`errors="replace"`	遇到无法解码的字符时替换（避免崩溃）
`env=os.environ.copy()`	传递当前环境变量
`bufsize=1`	行缓冲，每行立即输出

2.3 MinerU 工具介绍

MinerU 是一个开源的 PDF 解析工具，能够将 PDF 文档转换为 Markdown 格式，保留文档结构和图片。

安装：

bash

pip install mineru[all] -i https://pypi.tuna.tsinghua.edu.cn/simple

命令行用法：

bash

mineru -p input.pdf -o output_dir/ --source local

参数说明：

参数	说明
`-p`	输入 PDF 文件路径
`-o`	输出目录
`--source local`	使用本地已下载的模型（避免每次下载）

输出目录结构：

output_dir/
└── 文件名/
    └── hybrid_auto/
        ├── 文件名.md      # 转换后的 Markdown
        └── images/         # 提取的图片
            ├── image_0.png
            └── image_1.png

2.4 条件路由

LangGraph 中的条件路由允许根据状态决定下一个节点：

🟡 【P1 看注释就行】 条件路由是 LangGraph 的模式化代码——定义路由函数 → add_conditional_edges 绑定。理解 route_function 返回节点名称的映射逻辑即可。

python

def route_function(state):
    """路由函数：返回下一个节点名称"""
    if state.get("is_pdf"):
        return "pdf_node"
    elif state.get("is_md"):
        return "md_node"
    return END

# 添加条件边
workflow.add_conditional_edges(
    "entry",              # 源节点
    route_function,       # 路由函数
    {
        "pdf_node": "pdf_node",
        "md_node": "md_node",
        END: END
    }
)

3. 文件导入业务处理流程（总）

3.1 整体流程概述

3.2 状态流转

4. 文件导入业务处理流程（分）

4.1 入口节点

4.1.1 目标

检测输入文件的类型（PDF 或 MD）
设置相应的处理标志
提取文件标题
为后续路由提供判断依据

4.1.2 需求分析

输入：

import_file_path：用户上传的文件路径

输出：

is_pdf_read_enabled：是否为 PDF 文件
is_md_read_enabled：是否为 MD 文件
pdf_path 或 md_path：对应的文件路径
file_title：文件标题（不含扩展名）

边界条件：

场景	处理方式
`import_file_path` 为空	抛出 `ValidationError`
文件扩展名为 `.pdf`	设置 `is_pdf_read_enabled = True`
文件扩展名为 `.md`	设置 `is_md_read_enabled = True`
其他扩展名	记录警告，不设置任何标志

4.1.3 实现流程

4.1.3.1 实现流程图

4.1.3.2 具体实现步骤

步骤	操作	说明
Step 1	获取并验证路径	从 `state` 中获取 `import_file_path`，为空则抛出异常
Step 2	解析文件类型	使用 `pathlib.Path` 获取扩展名，转小写统一处理
Step 3	设置处理标志	根据扩展名设置 `is_pdf_read_enabled` 或 `is_md_read_enabled`
Step 4	提取文件标题	使用 `path.stem` 获取不含扩展名的文件名

4.1.4 代码实现

🔥 【P0 必须要学】 EntryNode 是导入流程的第一个节点，掌握文件类型检测 + 路由标志位设置 = 理解整个导入流程的启动机制。注意空路径异常处理和扩展名统一小写的细节。

python

# knowledge/processor/import_process/nodes/entry.py

"""
入口节点

检测文件类型并设置处理标志
"""

import json
from pathlib import Path

from knowledge.processor.import_process.base import BaseNode
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.exceptions import ValidationError


class EntryNode(BaseNode):
    """
    入口节点

    根据输入文件的扩展名设置相应的处理标志，
    决定后续流程走 PDF 转换分支还是直接处理 MD 分支。
    """

    name = "entry"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        """
        处理入口逻辑

        1. 获取输入文件路径
        2. 检测文件类型
        3. 设置相应的处理标志
        4. 提取文件标题

        Args:
            state: 图状态

        Returns:
            更新后的状态
        """
        # Step 1: 获取并验证路径
        self.log_step("step_1", "获取文件路径")
        file_path = state.get("import_file_path", "")

        if not file_path:
            raise ValidationError(
                "import_file_path 不能为空",
                node_name=self.name
            )

        # Step 2: 解析文件类型
        path = Path(file_path)
        suffix = path.suffix.lower()

        self.log_step("step_2", f"检测到文件类型: {suffix}")

        # Step 3: 设置处理标志
        if suffix == ".pdf":
            self.logger.info("启用 PDF 读取流程")
            state["is_pdf_read_enabled"] = True
            state["pdf_path"] = file_path
        elif suffix == ".md":
            self.logger.info("启用 MD 读取流程")
            state["is_md_read_enabled"] = True
            state["md_path"] = file_path
        else:
            self.logger.warning(f"不支持的文件类型: {suffix}")

        # Step 4: 提取文件标题（不含扩展名）
        state["file_title"] = path.stem
        self.log_step("step_3", f"文件标题: {state['file_title']}")

        return state


# ================================================================== #
#                        兼容 & 测试                                  #
# ================================================================== #

# 兼容原有调用方式
node_entry = EntryNode()


if __name__ == '__main__':
    """
    入口节点测试

    测试不同文件类型的处理逻辑
    """
    from knowledge.processor.import_process.base import setup_logging

    # 配置日志
    setup_logging()

    print("=" * 60)
    print("Entry 节点测试")
    print("=" * 60)

    # 测试用例 1: PDF 文件
    print("\n--- 测试用例 1: PDF 文件 ---")
    entry_node = EntryNode()
    state_pdf = {
        "import_file_path": "D:/docs/万用表的使用.pdf"
    }
    result_pdf = entry_node.process(state_pdf)
    print(json.dumps(result_pdf, indent=4, ensure_ascii=False))

    # 测试用例 2: MD 文件
    print("\n--- 测试用例 2: MD 文件 ---")
    state_md = {
        "import_file_path": "D:/docs/使用指南.md"
    }
    result_md = entry_node.process(state_md)
    print(json.dumps(result_md, indent=4, ensure_ascii=False))

    # 测试用例 3: 空路径（预期抛出异常）
    print("\n--- 测试用例 3: 空路径 ---")
    try:
        state_empty = {"import_file_path": ""}
        entry_node.process(state_empty)
    except ValidationError as e:
        print(f"捕获到预期异常: {e}")

    # 测试用例 4: 不支持的文件类型
    print("\n--- 测试用例 4: 不支持的文件类型 ---")
    state_other = {
        "import_file_path": "D:/docs/document.docx"
    }
    result_other = entry_node.process(state_other)
    print(json.dumps(result_other, indent=4, ensure_ascii=False))
    print(f"is_pdf_read_enabled: {result_other.get('is_pdf_read_enabled', False)}")
    print(f"is_md_read_enabled: {result_other.get('is_md_read_enabled', False)}")

关键设计点：

路径验证优先
- 空路径直接抛出异常，避免后续处理
扩展名统一小写
- suffix.lower() 确保 .PDF 和 .pdf 处理一致
标志位设计
- 使用布尔标志位而非枚举，便于路由判断
- 同时保存原始路径到 pdf_path 或 md_path
文件标题提取
- 使用 path.stem 提取不含扩展名的文件名
- 后续用于商品名识别等场景

4.2 PDF 转 Markdown 节点

4.2.1 目标

将 PDF 文档转换为 Markdown 格式
调用 MinerU 命令行工具
实时输出转换日志
处理转换异常

4.2.2 需求分析

输入：

pdf_path：PDF 文件路径
file_dir：输出目录（可选，默认为 PDF 所在目录）

输出：

md_path：转换后的 Markdown 文件路径

依赖：

MinerU 命令行工具已安装
模型文件已下载到本地

边界条件：

场景	处理方式
`pdf_path` 为空	抛出 `FileProcessingError`
PDF 文件不存在	抛出 `FileProcessingError`
MinerU 转换失败	抛出 `PdfConversionError`
输出目录不存在	MinerU 会自动创建

4.2.3 实现流程

4.2.3.1 实现流程图

4.2.3.2 具体实现步骤

步骤	方法	操作	说明
Step 1	`_validate_paths()`	验证输入路径	确保 PDF 路径有效且文件存在
Step 2	`_execute_mineru()`	执行转换命令	调用 MinerU 并实时输出日志
Step 3	`_get_output_path()`	计算输出路径	根据 MinerU 输出结构拼接 MD 路径

MinerU 命令参数：

bash

mineru -p <pdf_path> -o <output_dir> --source local

参数	说明
`-p`	输入 PDF 文件路径
`-o`	输出目录
`--source local`	使用本地模型（避免每次下载）

subprocess.Popen 参数详解：

🔥 【P0 必须要学】 PdfToMdNode 展示了"子进程调用外部工具"的标准模式——验证路径 → 构造命令 → Popen 实时日志 → 检查返回码 → 计算输出路径。这种模式在 AI 项目中调用模型推理脚本时反复出现。

python

proc = subprocess.Popen(
    cmd,
    stdout=subprocess.PIPE,     # 捕获标准输出
    stderr=subprocess.STDOUT,   # 合并错误到标准输出
    text=True,                  # 文本模式（字符串）
    encoding="utf-8",           # 编码
    errors="replace",           # 遇到乱码时替换
    env=os.environ.copy(),      # 传递环境变量
    bufsize=1,                  # 行缓冲（实时输出）
)

4.2.4 代码实现

python

# knowledge/processor/import_process/nodes/pdf_to_md.py

"""
PDF 转 Markdown 节点

使用 MinerU 将 PDF 文档转换为 Markdown 格式
"""

import json
import subprocess
import os
import time
from pathlib import Path

from knowledge.processor.import_process.base import BaseNode, setup_logging
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.exceptions import PdfConversionError, FileProcessingError


class PdfToMdNode(BaseNode):
    """
    PDF 转 Markdown 节点

    调用 MinerU 命令行工具将 PDF 转换为 Markdown，
    支持实时输出转换日志。
    """

    name = "pdf_to_md"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        """
        执行 PDF 转换

        1. 验证 PDF 路径
        2. 执行 MinerU 转换
        3. 处理转换结果

        Args:
            state: 图状态

        Returns:
            更新后的状态（包含 md_path）
        """
        # Step 1: 验证路径
        pdf_path_obj, output_dir_obj = self._validate_paths(state)

        # Step 2: 执行转换
        return_code = self._execute_mineru(pdf_path_obj, output_dir_obj)

        if return_code != 0:
            raise PdfConversionError(
                "MinerU 转换失败，请检查 mineru 日志",
                node_name=self.name
            )

        # Step 3: 获取结果路径
        state["md_path"] = self._get_output_path(pdf_path_obj, output_dir_obj)
        self.log_step("step_3", f"输出路径: {state['md_path']}")

        return state

    def _validate_paths(self, state: ImportGraphState) -> tuple:
        """
        验证 PDF 路径和输出目录

        Args:
            state: 图状态

        Returns:
            (pdf_path_obj, output_dir_obj) 元组

        Raises:
            FileProcessingError: 路径无效时抛出
        """
        self.log_step("step_1", "验证路径")

        # 获取 PDF 路径
        pdf_path = state.get("pdf_path", "")
        if not pdf_path:
            raise FileProcessingError("pdf_path 为空", node_name=self.name)

        pdf_path_obj = Path(pdf_path)
        if not pdf_path_obj.exists():
            raise FileProcessingError(
                f"PDF 文件不存在: {pdf_path}",
                node_name=self.name
            )

        # 获取输出目录
        output_dir = state.get("file_dir", "")
        if not output_dir:
            # 默认输出到 PDF 所在目录
            output_dir = str(pdf_path_obj.parent)

        output_dir_obj = Path(output_dir)
        self.logger.info(f"处理 PDF: {pdf_path_obj.name}")

        return pdf_path_obj, output_dir_obj

    def _execute_mineru(self, pdf_path_obj: Path, output_dir_obj: Path) -> int:
        """
        执行 MinerU 命令

        Args:
            pdf_path_obj: PDF 文件 Path 对象
            output_dir_obj: 输出目录 Path 对象

        Returns:
            命令返回码（0 表示成功）
        """
        self.log_step("step_2", "执行 MinerU 转换")

        # 1. 构建命令
        cmd = [
            "mineru",
            "-p", str(pdf_path_obj),
            "-o", str(output_dir_obj),
            "--source", "local"  # 使用本地模型
        ]

        self.logger.info(f"执行命令: {' '.join(cmd)}")
        start_ts = time.time()

        # 2. 调用命令行工具
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,     # 捕获标准输出
            stderr=subprocess.STDOUT,   # 合并错误到 stdout
            text=True,                  # 文本模式
            encoding="utf-8",
            errors="replace",           # 遇到乱码时替换
            env=os.environ.copy(),      # 传递环境变量
            bufsize=1,                  # 行缓冲（实时输出）
        )

        # 3. 实时输出日志
        for line in proc.stdout:
            self.logger.debug(f"[mineru] {line.rstrip()}")

        # 4. 等待命令完成
        return_code = proc.wait()

        elapsed = time.time() - start_ts
        if return_code == 0:
            self.logger.info(f"转换完成，耗时: {elapsed:.2f} 秒")
        else:
            self.logger.error("转换失败")

        return return_code

    def _get_output_path(self, pdf_path_obj: Path, output_dir_obj: Path) -> str:
        """
        获取转换结果路径

        MinerU 输出目录结构:
        output_dir/
          └── 文件名/
               └── hybrid_auto/
                    ├── 文件名.md
                    └── images/

        Args:
            pdf_path_obj: PDF 文件 Path 对象
            output_dir_obj: 输出目录 Path 对象

        Returns:
            Markdown 文件路径字符串
        """
        file_stem = pdf_path_obj.stem
        md_path = output_dir_obj / file_stem / "hybrid_auto" / f"{file_stem}.md"
        return str(md_path)


# ================================================================== #
#                        兼容 & 测试                                  #
# ================================================================== #

# 兼容原有调用方式
node_pdf_to_md = PdfToMdNode()


if __name__ == '__main__':
    """
    PDF 转 Markdown 节点测试

    注意：需要确保 MinerU 已安装且模型已下载
    """
    # 配置日志
    setup_logging()

    print("=" * 60)
    print("PDF to MD 节点测试")
    print("=" * 60)

    # 实例化节点
    pdf_to_md_node = PdfToMdNode()

    # 测试用例 1: 正常转换
    print("\n--- 测试用例 1: 正常 PDF 转换 ---")

    # 请修改为实际存在的 PDF 文件路径
    test_pdf_path = r"D:\develop\workspace\knowledge\test_data\万用表的使用.pdf"
    test_output_dir = r"D:\develop\workspace\knowledge\test_data\output"

    # 检查测试文件是否存在
    if not Path(test_pdf_path).exists():
        print(f"警告: 测试文件不存在: {test_pdf_path}")
        print("请修改 test_pdf_path 为有效的 PDF 文件路径")
    else:
        state = {
            "pdf_path": test_pdf_path,
            "file_dir": test_output_dir
        }

        try:
            result = pdf_to_md_node.process(state)
            print("转换成功!")
            print(json.dumps(result, indent=4, ensure_ascii=False))

            # 检查输出文件是否存在
            md_path = Path(result["md_path"])
            if md_path.exists():
                print(f"\n输出文件已生成: {md_path}")
                print(f"文件大小: {md_path.stat().st_size} 字节")
            else:
                print(f"警告: 输出文件不存在: {md_path}")

        except PdfConversionError as e:
            print(f"转换失败: {e}")
        except FileProcessingError as e:
            print(f"文件处理错误: {e}")

    # 测试用例 2: PDF 路径为空
    print("\n--- 测试用例 2: PDF 路径为空 ---")
    try:
        state_empty = {"pdf_path": ""}
        pdf_to_md_node.process(state_empty)
    except FileProcessingError as e:
        print(f"捕获到预期异常: {e}")

    # 测试用例 3: PDF 文件不存在
    print("\n--- 测试用例 3: PDF 文件不存在 ---")
    try:
        state_not_exist = {"pdf_path": "D:/not_exist/file.pdf"}
        pdf_to_md_node.process(state_not_exist)
    except FileProcessingError as e:
        print(f"捕获到预期异常: {e}")

    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251

关键设计点：

职责分离
- _validate_paths()：验证逻辑
- _execute_mineru()：执行逻辑
- _get_output_path()：路径计算
实时日志输出
- 使用 bufsize=1 行缓冲
- 逐行读取 proc.stdout 并打印
环境变量传递
- env=os.environ.copy() 确保子进程能访问 Python 环境
错误处理
- 返回码非零时抛出 PdfConversionError
- 文件不存在时抛出 FileProcessingError

5. 测试运行

5.1 运行 Entry 节点测试

bash

# 进入项目目录
cd knowledge

# 激活虚拟环境
.venv\Scripts\activate

# 运行测试
python -m knowledge.processor.import_process.nodes.entry

预期输出：

============================================================
Entry 节点测试
============================================================

--- 测试用例 1: PDF 文件 ---
2026-02-23 10:00:00 - import.entry - INFO - --- entry 开始 ---
2026-02-23 10:00:00 - import.entry - INFO - [step_1] 获取文件路径
2026-02-23 10:00:00 - import.entry - INFO - [step_2] 检测到文件类型: .pdf
2026-02-23 10:00:00 - import.entry - INFO - 启用 PDF 读取流程
2026-02-23 10:00:00 - import.entry - INFO - [step_3] 文件标题: 万用表的使用
2026-02-23 10:00:00 - import.entry - INFO - --- entry 完成 ---
{
    "import_file_path": "D:/docs/万用表的使用.pdf",
    "is_pdf_read_enabled": true,
    "pdf_path": "D:/docs/万用表的使用.pdf",
    "file_title": "万用表的使用"
}

--- 测试用例 2: MD 文件 ---
...

--- 测试用例 3: 空路径 ---
捕获到预期异常: [entry] import_file_path 不能为空

--- 测试用例 4: 不支持的文件类型 ---
...

5.2 运行 PDF to MD 节点测试

bash

# 运行测试
python -m knowledge.processor.import_process.nodes.pdf_to_md

预期输出：

============================================================
PDF to MD 节点测试
============================================================

--- 测试用例 1: 正常 PDF 转换 ---
2026-02-23 10:00:00 - import.pdf_to_md - INFO - --- pdf_to_md 开始 ---
2026-02-23 10:00:00 - import.pdf_to_md - INFO - [step_1] 验证路径
2026-02-23 10:00:00 - import.pdf_to_md - INFO - 处理 PDF: 万用表的使用.pdf
2026-02-23 10:00:00 - import.pdf_to_md - INFO - [step_2] 执行 MinerU 转换
2026-02-23 10:00:00 - import.pdf_to_md - INFO - 执行命令: mineru -p D:\...\万用表的使用.pdf -o D:\...\output --source local
2026-02-23 10:00:05 - import.pdf_to_md - DEBUG - [mineru] Processing page 1/10...
2026-02-23 10:00:10 - import.pdf_to_md - DEBUG - [mineru] Processing page 2/10...
...
2026-02-23 10:00:30 - import.pdf_to_md - INFO - 转换完成，耗时: 30.15 秒
2026-02-23 10:00:30 - import.pdf_to_md - INFO - [step_3] 输出路径: D:\...\output\万用表的使用\hybrid_auto\万用表的使用.md
2026-02-23 10:00:30 - import.pdf_to_md - INFO - --- pdf_to_md 完成 ---
转换成功!
{
    "pdf_path": "D:\\...\\万用表的使用.pdf",
    "file_dir": "D:\\...\\output",
    "md_path": "D:\\...\\output\\万用表的使用\\hybrid_auto\\万用表的使用.md"
}

输出文件已生成: D:\...\output\万用表的使用\hybrid_auto\万用表的使用.md
文件大小: 15234 字节

--- 测试用例 2: PDF 路径为空 ---
捕获到预期异常: [pdf_to_md] pdf_path 为空

--- 测试用例 3: PDF 文件不存在 ---
捕获到预期异常: [pdf_to_md] PDF 文件不存在: D:/not_exist/file.pdf

============================================================
测试完成
============================================================

6. 总结

6.1 两个节点对比

特性	Entry 节点	PDF to MD 节点
职责	文件类型检测	PDF 转换
复杂度	简单	中等
外部依赖	无	MinerU 命令行工具
异常类型	`ValidationError`	`FileProcessingError`, `PdfConversionError`
输出	标志位 + 路径	md_path

6.2 设计要点

单一职责
- 每个节点只做一件事
- Entry 只检测类型，不做转换
- PDF to MD 只做转换，不做后续处理
防御性编程
- 所有输入都要验证
- 文件路径要检查存在性
- 外部命令要检查返回码
可观测性
- 详细的日志输出
- 实时日志（bufsize=1）
- 步骤标记（log_step）
可测试性
- 每个节点都有独立的测试入口
- 多种边界条件测试
- 异常捕获验证

企业痛点映射

痛点	传统方案	AI Agent 导入流程方案	效率提升
PDF 文档手工转 Markdown 耗时	人工复制粘贴，每份文档 15~30min	MinerU 自动转换 + subprocess 实时日志	~95% 时间节省，每份文档降至 1~2min
多格式文件需不同处理流程	写大量 if-else 硬编码判断	EntryNode 条件路由 + 标志位设计	新增文件类型只需加一条路由规则
外部命令调用无日志追踪	命令行手动执行，出错无法回溯	subprocess.Popen + 实时日志 + 返回码校验	排查时间降低 ~70%（预估）
PDF 解析结果路径不可预期	手动在输出目录翻找	`_get_output_path` 根据 MinerU 目录结构自动拼接	零人工查找成本
空路径/文件不存在未校验	运行时才报错，错误信息模糊	EntryNode 前置 ValidationError 校验	错误提前暴露，排错提速 ~80%

Remote & Agent 应用场景价值

Remote 场景价值：EntryNode + PdfToMdNode 的职责清晰分离，远程团队可并行开发两个节点，通过 ImportGraphState 的 pdf_path / md_path / file_title 字段契约协作。_execute_mineru() 的实时日志机制适合在远程 CI/CD 环境中追踪 PDF 转换进度。
Agent 落地场景：PdfToMdNode 可以直接封装为"文档转换 Agent"——Agent 监听文件上传事件 → 自动调用 MinerU → 校验转换质量 → 输出 Markdown 路径。EntryNode 的条件路由逻辑可扩展为"文件类型分类 Agent"，自动检测文件类型并路由到对应的处理 Pipeline。

Git Commit 对应

本节入口节点和 PDF 转换节点对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议在项目仓库中搜索 "entry.py" / "pdf_to_md.py" 相关提交>

bash

cd shopkeeper_brain
# 查看节点相关代码
git log --oneline --all -- knowledge/processor/import_process/nodes/

入口节点与 PDF 转 Markdown 节点 ​

阅读指引 ​

技术栈健康度标签体系 ​

中英文对照表 ​

1. 任务目标 ​

1.1 本章目标 ​

1.2 涉及模块 ​

1.3 节点在流程中的位置 ​

2. 核心概念扫盲 ​

2.1 pathlib.Path ​

2.2 subprocess 子进程调用 ​

2.3 MinerU 工具介绍 ​

2.4 条件路由 ​

3. 文件导入业务处理流程（总） ​

3.1 整体流程概述 ​

3.2 状态流转 ​

4. 文件导入业务处理流程（分） ​

4.1 入口节点 ​

4.1.1 目标 ​

4.1.2 需求分析 ​

4.1.3 实现流程 ​

4.1.3.1 实现流程图 ​

4.1.3.2 具体实现步骤 ​

4.1.4 代码实现 ​

4.2 PDF 转 Markdown 节点 ​

4.2.1 目标 ​

4.2.2 需求分析 ​

4.2.3 实现流程 ​

4.2.3.1 实现流程图 ​

4.2.3.2 具体实现步骤 ​

4.2.4 代码实现 ​

5. 测试运行 ​

5.1 运行 Entry 节点测试 ​

5.2 运行 PDF to MD 节点测试 ​

6. 总结 ​

6.1 两个节点对比 ​

6.2 设计要点 ​

企业痛点映射 ​

Remote & Agent 应用场景价值 ​

Git Commit 对应 ​

入口节点与 PDF 转 Markdown 节点

阅读指引

技术栈健康度标签体系

中英文对照表

1. 任务目标

1.1 本章目标

1.2 涉及模块

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 pathlib.Path

2.2 subprocess 子进程调用

2.3 MinerU 工具介绍

2.4 条件路由

3. 文件导入业务处理流程（总）

3.1 整体流程概述

3.2 状态流转

4. 文件导入业务处理流程（分）

4.1 入口节点

4.1.1 目标

4.1.2 需求分析

4.1.3 实现流程

4.1.3.1 实现流程图

4.1.3.2 具体实现步骤

4.1.4 代码实现

4.2 PDF 转 Markdown 节点

4.2.1 目标

4.2.2 需求分析

4.2.3 实现流程

4.2.3.1 实现流程图

4.2.3.2 具体实现步骤

4.2.4 代码实现

5. 测试运行

5.1 运行 Entry 节点测试

5.2 运行 PDF to MD 节点测试

6. 总结

6.1 两个节点对比

6.2 设计要点

企业痛点映射

Remote & Agent 应用场景价值

Git Commit 对应