当前位置：首页 > news >正文

MaxKB+MinerU：通过API实现PDF文档解析并存储至知识库

news 2025/7/31 10:49:42

MinerU是一款开源的高质量数据提取工具，能够将PDF文档转换为Markdown和JSON格式。2025年6月13日，MinerU发布了v2.0版本，相较于v1.0版本实现了架构和功能的全面重构与升级。在优化代码结构和交互方式的同时，v2.0版本还集成了小参数量、高性能多模态文档解析模型，能够实现端到端的高速、高精度文档理解。实际测试表明，新版本对复杂图表的解析效果较上一版本有明显提升，目前已经能够满足90%以上的复杂文档解析需求。

值得一提的是，MinerU出色的PDF文档解析能力特别适合与MaxKB开源项目配合使用。通过"MinerU+MaxKB"的组合方案，用户不仅能够获得高质量的文档解析效果，还能显著提升知识库问答系统的性能。为方便用户集成，MinerU项目现已提供API对接服务（https://mineru.net/apiManage）。接下来，本文将详细介绍如何通过MinerU在线API实现与MaxKB的对接。

一、实现方法

当用户提供文件地址后，系统会将该地址赋值给file_url变量，并作为参数传递给MinerU文件解析服务。MinerU在完成文件解析后，会返回一个任务ID（task_id）。系统会将其传入MinerU的查询接口，当检测到任务处理完成时，自动获取结果文件的下载链接（full_url）。随后，系统执行文件下载操作，将结果文件保存到MaxKB容器的/opt/maxkb/download目录下。最后，系统会自动完成文件上传和智能分段处理，将内容存储到知识库中。
在这里插入图片描述

二、MaxKB函数创建

我们需要在MaxKB的函数库中创建四个核心功能函数，其用途分别为：

1.调用MinerU单个文件解析；

2.从MinerU获取任务结果；

3.通过URL链接下载文件至服务器；

4.将解析后的ZIP文件上传至知识库。

相关的代码说明如下：

■ MinerU单个文件解析函数：负责调用MinerU的单文件解析服务，通过传入PDF文档的在线地址来创建解析任务，并返回对应的task_id；

import requestsdef create_task(file_url):url = 'https://mineru.net/api/v4/extract/task'token = '自己申请的 Token'header = {'Content-Type': 'application/json','Authorization': f'Bearer {token}'}data = {'url': file_url,'is_ocr': True,  #是否启动 ocr 功能，默认 false'enable_formula': True,  #是否开启公式识别，默认 true'enable_table': True,    #是否开启表格识别，默认 true'language': "ch",    #指定文档语言，默认 ch，可以设置为auto'model_version': "v2",  #mineru模型版本，两个选项:v1、v2，默认v1。}res = requests.post(url,headers=header,json=data,timeout=5)res_data = res.json()task_id_data = res_data["data"]["task_id"]return task_id_data

在这里插入图片描述

■ MinerU获取任务结果函数：用于查询任务状态，通过传入task_id获取解析结果，成功后将返回ZIP格式解析文件的下载地址；

import time
import requests
def querybyid(task_id,max_retries=100,retry_interval=5):url = f'https://mineru.net/api/v4/extract/task/{task_id}'token = '申请的Token'header = {'Content-Type': 'application/json','Authorization': f'Bearer {token}'}retries = 0while retries < max_retries:try:res = requests.get(url, headers=header, timeout=5)res.raise_for_status()  # 检查请求是否成功data = res.json()if "data" in data and "full_zip_url" in data["data"] and data["data"]["full_zip_url"]:return data["data"]["full_zip_url"]else:print(f"full_zip_url 为空，正在等待任务完成。已重试 {retries + 1} 次，共 {max_retries} 次。")time.sleep(retry_interval)retries += 1except requests.exceptions.RequestException as e:print(f"请求失败，错误信息：{e}。正在重试...")time.sleep(retry_interval)retries += 1raise Exception(f"在 {max_retries} 次重试后，仍未获取到有效的 full_zip_url。")

在这里插入图片描述

■ 文件下载函数：根据提供的ZIP文件下载链接，将文件保存至容器内的/opt/maxkb/download目录。需要注意的是，MaxKB默认使用sandbox用户运行，需确保该用户对/opt/maxkb/download目录有读写权限；

import os
import requests
from urllib.parse import urlparsedef download_file(download_url, save_dir='/opt/maxkb/download'):os.makedirs(save_dir, exist_ok=True)# 获取文件名parsed_url = urlparse(download_url)filename = os.path.basename(parsed_url.path)save_path = os.path.join(save_dir, filename) # 文件下载后保存的目录，需要默认用户对此目录有读写权限# 下载文件try:response = requests.get(download_url, stream=True)response.raise_for_status()  # 检查请求是否成功total_size = int(response.headers.get('content-length', 0))block_size = 1024  # 1KBprogress = 0print(f"开始下载 {filename} 到 {save_dir}")with open(save_path, 'wb') as f:for data in response.iter_content(block_size):f.write(data)progress += len(data)# 打印下载进度print(f"下载进度: {progress / total_size * 100:.2f}%", end='\r')print(f"\n下载完成: {save_path}")return save_pathexcept requests.exceptions.RequestException as e:print(f"下载失败: {e}")return None

在这里插入图片描述

■ ZIP文件上传至知识库：通过MaxKB API将服务器上的ZIP解析文件上传至知识库存储。

import json
import logging
import requests
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')def initialize(file_path):config = {# MaxKB API密钥'authorization_apikey': 'user-ac86ec515de17969f2f8a9c8ab21e52f',    # 文件分段处理的API地址'split_url': 'http://10.1.11.58:8080/api/dataset/document/split',# 目标知识库的API地址'upload_url': 'http://10.1.11.58:8080/api/dataset/3d1d5d4e-5576-11f0-bc5c-0242ac120003/document/_bach','file_path': rf'{file_path}','file_name': '函数库上传文档分段'}return configdef upload_file(config):headers = {'accept': 'application/json','AUTHORIZATION': f'{config["authorization_apikey"]}'}try:files = {'file': open(config["file_path"], 'rb')}response = requests.post(config["split_url"], headers=headers, files=files)response.raise_for_status()response_data = response.json()map_content = {}for item in response_data.get("data", []):for content_item in item.get("content", []):map_content[content_item.get("title", "")] = content_item.get("content", "")return map_contentexcept requests.exceptions.RequestException as e:logging.error(f"文件分段上传失败: {e}")return {}except Exception as e:logging.error(f"处理文件内容时出错: {e}")return {}def send_post_request(config, map_content):headers = {"Content-Type": "application/json","Authorization": f'{config["authorization_apikey"]}'}paragraphs = [{"title": key, "content": value} for key, value in map_content.items()]document_wrapper = {"name": config["file_name"],"paragraphs": paragraphs}json_body = json.dumps([document_wrapper])try:response = requests.post(config["upload_url"], headers=headers, data=json_body)response.raise_for_status()logging.info(f"上传文件响应: {response.text}")return Trueexcept requests.exceptions.RequestException as e:logging.error(f"上传文件失败: {e}")return Falsedef main(file_path):config = initialize(file_path)map_content = upload_file(config)if not map_content:logging.error("文件分段上传失败或内容为空，程序终止")return Falseif not send_post_request(config, map_content):logging.error("文件上传失败，程序终止")return Falsereturn "文件已上传成功，并保持在知识库中"