掌柜智库项目 Day 02｜主体搭建与 PDF 文档处理

在项目骨架上定义 ImportGraphState 状态与 12 个属性，梳理节点与业务分层、搭建 LangGraph 主图，并实现 node_entry 与 node_pdf_to_md 两个节点。

本篇要点

• 定义导入图的 state：ImportGraphState(TypedDict) 与 12 个属性
• 分清”节点层”与”业务层”的职责，掌握节点统一写法
• 搭建 LangGraph 主图结构（builder -> add_node -> 条件边 -> compile）
• 实现任务分发节点 node_entry
• 实现 PDF 解析节点 node_pdf_to_md，顺带回顾 HTTP 协议与 requests

内容回顾

1、环境搭建

• 使用 uv 管理项目：安装 uv、配置清华镜像、uv init --bare 初始化项目、uv venv --python=3.11 创建 .venv、pycharm 手动指定解释器。
• 用 uv sync 同步依赖（pyproject.toml + uv.lock），再用 uv tree 校验依赖树。

2、项目骨架

• 搭好了 app/ 下的 utils/shared、process、rag、resources、api、infra、tests 包结构。
• 导入了公共类代码：.env、logger（node_log / step_log）、配置加载类、InfraConfig 等。
• 用 .ignore 插件创建 .gitignore，项目接入 git 并完成第一次提交。

今天内容

1、定义 state

1.1 分析 state 属性

本次分析了 state 需要的属性，一共 12 个。

1.2 定义导入图的状态

process / import_ / agent / state
  -> ImportGraphState(TypedDict)   # 12 个属性

• ImportGraphState 继承 TypedDict，是整个导入图在节点之间流转的共享状态。
• 属性的作用分两类：一类是节点之间要传递的数据（如 task_id、各类文件路径），一类是控制流程走向的开关（如 is_md_read_enabled、is_pdf_read_enabled）。

1.3 方便使用，定义默认对象

state: ImportGraphState = {}

create_default_state(**kwargs)   # 深 copy | dict.update
get_default_state()

• create_default_state 用深 copy 配合 dict.update 生成一份默认状态，避免多个任务共用同一个 dict 造成相互污染。
• get_default_state() 直接拿到默认状态，省去每次手写初始化。

2、节点和业务导入

2.1 节点和业务关系

process / import_ / agent / nodes   -> langgraph 节点 (Node)

节点专门负责：

• 节点和图流程任务编排
• 专门处理 state 数据（获取和返回）
• 也会调用自己的对应业务（rag…）
• 专门打印节点对应的日志流程 @node_log
• 专门记录节点状态 add_running_task(task_id, "节点名")——注意不要随意写，要根据 dict 字典转化

节点的统一写法：

@node_log      # 节点的日志输出
def node_函数(state):
    add_running_task(state['task_id'], "节点名")
    state = 调用业务逻辑函数(state)
    add_done_task(state['task_id'], "节点名")
    return state

业务层的位置：

rag / import_ / xxx_service

分层结论： nodes 只做流程编排、state 读写和日志 / 任务状态记录，真正的业务逻辑下沉到 rag 下的 service 里。

2.2 创建主图结构

位置：process / import_ / agent / main_graph（import_graph_builder、import_app）。

创建构建对象，指定全局 state
添加节点 ("", 函数)  (函数)
添加起始节点  set_entry_point(node_entry) | add_edge(START, node_entry)
添加条件边 (.., {} ..) 和声明条件路由（添加适当的日志）
添加静态边
编译图对象 app = compile

按顺序理解这 6 步：

1. 创建构建对象并指定全局 state，告诉图”用哪个 state 类型流转”。
2. 逐个添加节点，节点名与节点函数一一对应。
3. 添加起始节点：set_entry_point(node_entry)，也可以写成 add_edge(START, node_entry)。
4. 添加条件边和声明条件路由，这里要加上适当的日志，方便排查走错分支的问题。
5. 添加静态边，把固定顺序的节点串起来。
6. 编译图对象 app = compile，得到可执行的图。

2.3 测试

• 动态测试：执行起来看节点日志。
• 静态检查：看图结构。

3、任务分发节点（node_entry）

节点作用： 根据文件类型进行 state 调整，进而影响节点调用流程。

注意： 后续如果有文件类型扩展，需要改写该节点！

节点参数： state

• 使用：task_id、local_file_path
• 生成：md_path、is_md_read_enabled、pdf_path、is_pdf_read_enabled、file_title

实现步骤：

1. 先获取 local_file_path 参数（state）
2. 对 local_file_path 进行非空校验 -> 为空 -> 直接抛出异常 FileNotFound...
3. 判断是不是 md -> md_path、is_md_read_enabled，同时把 is_pdf_read_enabled = False
4. 判断是不是 pdf -> is_md_read_enabled = False、pdf_path、is_pdf_read_enabled
5. 都不是，做好警告提示，is_md_read_enabled、pdf_path、is_pdf_read_enabled = False，提前结束
6. 获取 file_title 参数，同步更新 state
7. 返回处理后的 state

4、PDF 解析节点（node_pdf_to_md）

节点作用： pdf 文件格式解析，保存解析后的 md 文件，并且更新 state。

节点参数： state

• 使用：task_id、pdf_path、local_dir【md 文件，下载到本地】
• 生成：md_path、md_content（优化，不读取了）

4.1 节点准备：使用第三方 MinerU

.env：

MINERU_API_TOKEN=你的_mineru_token     # [你的 token]
MINERU_BASE_URL=https://mineru.net/api/v4

config.py：

MINERU_MODEL_VERSION = "vlm"

# MinerU 任务轮询最大超时时间（单位：秒），超过则判定任务失败
# 600 -> 一个 pdf 约等于 1 秒
MINERU_POLL_TIMEOUT_SECONDS = 600

# MinerU 任务轮询间隔时间（单位：秒），每隔多久查询一次任务状态
MINERU_POLL_INTERVAL_SECONDS = 3

# MinerU 文件下载超时时间（单位：秒），下载文件超过此时长则中断
MINERU_DOWNLOAD_TIMEOUT_SECONDS = 30

重点数字： 轮询最大超时 600 秒（600 -> 一个 pdf 约等于 1 秒）、轮询间隔 3 秒、下载超时 30 秒。

4.2 参数获取和校验 validate_pdf_paths

def validate_pdf_paths(state: ImportGraphState) -> tuple[Path, Path]:
    """(state -> str) -> (pdf_path: Path, local_dir: Path)"""

步骤：

1. 1 state 获取 pdf_path 和 local_dir：str
2. 1.2 进行 pdf_path 非空校验：空 -> 打印日志 error -> 异常 ValueError
3. 1.3 进行 local_path 非空校验：空 -> 打印日志 warning -> 给与默认值 项目根路径 / output
4. 1.4 将 pdf_path、local_dir 转成 Path
5. 1.5 pdf_path_obj: Path 判断是否存在：不存在，打印日志 error -> 异常 FileNotFoundError
6. 1.6 local_dir_obj: Path 是不是目录（is_dir()）：不是（不存在，或者不是目录）-> 打印日志 warning -> 创建 local_dir_obj 对应的文件夹
7. 1.7 返回结果 return pdf_path_obj, local_dir_obj

Path 使用要点（顺带回顾）：

Path(str | Path)
exists()   是否存在        存在 -> true   不存在 -> false
is_dir()   是不是有效的文件夹  是 -> true   不是文件夹或者不存在 -> false
地址拼接兼容不同系统  Path / "xx"

4.3 向 MinerU 上传文件并且获取解析结果 upload_pdf_and_poll

def upload_pdf_and_poll(pdf_path_obj: Path) -> str:   # -> zip_url

步骤：

1. 2.1 向 MinerU 服务器发送请求申请上传地址（batch_id / url）
2. 2.2 向指定的 url 地址发起网络请求并且上传 pdf 文件
3. 2.3 轮询向 MinerU 获取 batch_id 解析状态 zip_url
4. 2.4 返回 zip_url …

回顾：网络请求 requests 包的使用

import requests

• 作用：在我们代码里可以调用外部的 api 接口，例如访问 MinerU -> url；requests 可以发起 http 协议的网络请求。
• 介绍：hyper text transfer protocol 协议，用于客户端和服务器的通信网络协议。
  • 超文本： 超越文本类型的数据传递（不仅仅能传字符，更能传递字节，如图片、压缩包…）
  • 传输： 客户端 - 服务端
  • 协议： 规定传递数据格式

http 协议的数据格式（请求报文 | 响应报文）：

                     ------请求报文[请求行[请求地址 请求方式 协议版本] 请求头[key=value] 请求空行[\n\n] 请求体]----->
客户端     （一次性）                                                                              服务端
                     <-----响应报文[状态行 协议版本 状态码200 状态值 响应头 响应空行 响应体]--------

请求方式：

• 字节数据、隐私数据、大于 2kb 的字符（浏览器对 url 的长度有限制）-> post / put
• url、method：GET / POST / PUT / DELETE … 代表客户端对服务端的数据操作动作行为！
  • 查询 -> get
  • 保存 -> post（restful 基于 http 协议”最”标准的使用方案，提出 restful 的人就是 http 协议的主要发明人）
  • 更新 -> put
  • 删除 -> delete
• url -> 指向服务端的资源 -> 名词，url 配合请求方式决定对资源的动作：
  • 根据 id 删除用户：错误 /user/remove?id=1，正确 /user/1 + delete
  • 根据 id 查询用户：错误 /user/find?id=1，正确 /user/1 + get

注意： 请求方式一定不要选错！不能选择无法传递数据的方式类型！

• get 客户端想要获取、delete 想要删除 -> 请求行 url?key=value&..、请求头、没有请求体
• post 想要上传、put 想要更新 -> 请求行 url?key=value&..、请求头、请求空行、请求体（字符 / 字节…）

对比举例：

• 客户端 -> 上传头像（图片、字节、流）-> 服务端：post | put
• 客户端 -> 登录账号（账号密码、字符）-> 服务端：post | put，不会选择 get / delete
• 删除 id=1：user/delete?id=1...

响应报文结构：

响应状态行   协议版本 200 -> 状态码
响应头
空行
响应体       -> 返回的数据

使用流程：1. 获取状态码 == 成功；2. 获取响应体的数据 …

响应状态码：

• 1xx 中继状态：101、102 … 请求还在继续，没有完成
• 2xx 成功状态：200 完整请求成功；202 断点请求成功 -> 王者荣耀（20% 继续下载：读取本地下载的进度 -> 请求头 -> 服务端 -> 服务端只返回剩余的部分）
• 3xx 重定向：304 我们访问目标服务器 -> 服务器将请求转到其他的服务器进行处理 -> 对于第一次请求已经是成功！间接成功
• 4xx 客户端访问资源出现问题：404 资源未找到（url 地址错误）、405 请求方式异常、400 参数错误【对于客户端来说致命，不需要尝试】
• 5xx 服务器内部异常：500 -> 代码异常抛出，又没有捕捉到 -> 客户端

requests 发起请求：

response = requests.get / post / delete / put(
    url=...,       # 请求地址 -> 不优雅 -> url?name=张三
    headers=...,   # 请求头
    params=...,    # -> ? 参数 -> dict -> 优雅地传递路径后的 ? 参数  {name=张三}
    json=...,      # -> 请求体使用字符存储 json 字符串 -> dict -> {name=张三}
    data=...,      # -> 请求体使用字节存储 文件 pdf 压缩包...
    timeout=...,   # 超时时间 单位秒 下载的时候...
)

requests 接收结果：

response.status_code   # 返回状态码 200
response.json()        # 获取响应体的数据(json 字符串)，转成 python dict
response.text          # 获取响应体的字符串 "{'code':0,'data':{}}" -> dict = json.loads(字符串)
response.content       # 获取响应体的字节数据  minerU -> zipurl -> zip 的压缩包 -> response.content

• json() -> 一套解析业务，可以定制：json(参数...)
• text / content 是 @property，读取请求体数据是一套业务代码，用属性方式使用方法，这个方法没有参数。

MinerU 的使用步骤：

1. 申请上传文件地址（token、请求体、文件名）
2. 向第一次返回的文件地址上传文件
3. 轮询根据 batch_id 向 MinerU 获取解析结果
4. 使用 zip_url 下载和解压文件

4.4 下载 zip 内容并且解压到 local_dir

def download_and_extract(zip_url, local_dir, pdf_path.stem) -> md_path: Path

3.1 向 zip 地址发起请求，获取返回的文件字节数据

response = requests.get(url=zip_url, headers=..., params=..., json=..., data=..., timeout=30)
# response -> status_code
zip_path_obj: Path = local_dir_obj / 文件名.zip
zip_path_obj.write_bytes(response.content)

3.2 进行 zip 文件的解压操作

• 创建对应的文件夹：

zip_extract_dir: Path = local_dir_obj / 文件名
# 检查是否存在文件夹
if zip_extract_dir.is_dir():
    # 清空文件夹内容
    ...
zip_extract_dir.mkdir(parents=True, exist_ok=True)
# 进行 zip 文件的解压...

3.3 文件的重命名

MinerU 产出的 md 文件名并不固定，可能是 文件名.md、full.md、随机字符串.md：

list[md] = zip_extract_dir.rglob("*.md")   # 或 glob("*.md")

循环检查文件：

• md -> 文件名 -> return md_path
• md -> full -> 记录
• md -> 随机取第一个 -> 记录

记录 md_path_obj，进行重命名即可：

md_path_obj.rename(...)
return md_path_obj

坑： MinerU 返回的 md 文件名不固定，要按 文件名.md > full.md > 随机第一个 的优先级挑选，再 rename 成统一的名字。

4.5 更新 state 参数

state["md_path"] = md_path      # md_path: Path

4. 更新 state 参数 md_path -> md_path: Path，供后续节点使用。