打造你的文档小助手：PDF智能解析系统揭秘！
嘿，朋友们！你有没有为翻阅一堆PDF文件头晕眼花过？有没有为从文档中提取关键信息熬夜到怀疑人生？别担心！今天我要隆重介绍一款“文档界的卷王” ——PDF智能解析系统！它不仅能帮你轻松搞定PDF，还能生成酷炫的分析报告，简直是懒人福音！
这货到底有啥用？
简单来说，这套系统能让你从“手动狗头”进化到“智能AI狗头”。它的核心功能包括：
- 1. PDF上传：轻轻一点，文件上传，so easy！
- 2. 智能分析：AI出马，帮你挖掘文档中的隐藏宝藏。
- 3. 实时反馈：进度条实时更新，告别“等到天荒地老”。
- 4. 报告生成：自动输出Markdown格式报告，简洁又高效。
它是怎么运作的？
整个系统架构采用了“前后端分离”，听着高大上，其实就是“分工明确，各干各的”：
- • 前端：HTML/CSS/JavaScript三剑客在线打工，再加上marked.js负责Markdown渲染，用户体验感直接拉满！
- • 后端：用Python，负责文件处理和分析逻辑，API接口设计得贼精致。
亮点功能大揭秘
- 1. 任务状态轮询：上传文件后，系统会给你一个任务ID，然后开始疯狂“打探消息”。“任务完成了吗？”、“进度多少了？”系统比你还着急！
- 2. Markdown渲染：用marked.js把分析结果变成赏心悦目的Markdown格式，你甚至可以假装自己是个写作高手！
- 3. 报告下载：一键下载报告，拿去交差都不是问题。
背后的技术黑科技
- • PDF处理：用Python的PyPDF2或者pdfminer提取文本，再用NLP库（比如spaCy）分析内容。听起来很硬核，其实就是让AI替你干活。
- • 异步任务管理：Celery或者Bull出马，确保任务处理又快又稳。
- • 存储方案：小型应用用本地存储，大型应用直接上云。
结尾彩蛋
总的来说，这款PDF智能解析系统就是你的“文档拯救者”，让你从繁琐的文件处理中解脱出来。再也不用为报表熬夜到秃头了！还在等什么？赶紧试试吧！
解析效果
文档智能解析系统
基于FastAPI + PyMuPDF + Qwen-VL + DeepSeek构建的文档智能解析系统，可自动分析PDF文档中的文本和图像内容，生成结构化的Markdown报告。
功能特点
- • PDF处理：使用PyMuPDF提取PDF文件中的文本和图片
- • 图像分析：基于Qwen-VL-Max模型分析图片内容
- • 文本分析：使用DeepSeek-Chat模型分析文本内容
- • 结构化输出：自动生成Markdown格式的分析报告
系统架构
本系统使用以下技术栈：
- • DeepSeek-Chat：分析和理解文本内容
安装与设置
环境要求
安装步骤
1.. 创建并激活虚拟环境（可选）
python -m venv venv
# 在Windows上
venv\Scripts\activate
# 在macOS/Linux上
source venv/bin/activate
pip install -r requirements.txt
pip install -r requirements.txt - i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
创建.env文件并添加以下内容：
QIANWEN_API_KEY=你的千问API密钥
DEEPSEEK_API_KEY=你的DeepSeek API密钥
运行服务
uvicorn main:app --reload
默认情况下，服务将在 http://localhost:8000 上运行。
API接口说明
上传PDF文件进行分析
- • 请求类型：
multipart/form-data
查询任务状态
下载分析结果
- • 端点：
GET /download/{task_id}
使用示例
- • 打开浏览器，上传 pdf 文件
目录结构
文档智能解析系统/
├── main.py # FastAPI应用主文件
├── pdf_processor.py # PDF处理和分析模块
├── prompts.py # 提示词模板模块
├── llms.py # AI模型客户端配置
├── requirements.txt # 项目依赖
├── .env # 环境变量配置
├── README.md # 项目说明
├── uploads/ # 上传的PDF文件存储目录
├── results/ # 生成的分析报告存储目录
└── static/ # 静态文件目录
前端核心逻辑分析
1. 文件上传流程
用户选择文件 → 前端验证文件类型 → 上传至服务器 → 服务器返回任务ID → 前端开始轮询任务状态
关键代码
// 上传文件
uploadBtn.addEventListener('click', async function() {
// ...验证文件
const formData = new FormData();
formData.append('file', file);
// 发送上传请求
const response = await fetch('/upload', {
method: 'POST',
body: formData
});
const data = await response.json();
taskId = data.task_id;
// 开始轮询任务状态
pollTaskStatus();
});
2. 任务状态轮询机制
获取任务ID → 定期请求任务状态 → 更新进度条和状态信息 → 任务完成或失败时更新界面
关键代码
async function pollTaskStatus() {
if (!taskId) return;
const response = await fetch(`/task/${taskId}`);
const data = await response.json();
// 更新进度条
const progress = data.progress || 0;
progressFill.style.width = `${progress}%`;
progressText.textContent = `${progress}%`;
// 任务完成或失败时的处理逻辑
if (data.status === 'completed') {
// 处理完成逻辑
} else if (data.status === 'failed') {
// 处理失败逻辑
} else {
// 继续轮询
setTimeout(pollTaskStatus, 2000);
}
}
3. 报告预览和下载
获取任务ID → 请求预览内容 → 使用marked.js渲染Markdown → 显示在预览区域
关键代码
async function loadPreview() {
// 获取Markdown内容
const response = await fetch(`/preview/${taskId}`);
const data = await response.json();
// 使用marked.js渲染
if (typeof marked.parse === 'function') {
markdownContent.innerHTML = marked.parse(data.content);
} else if (typeof marked === 'function') {
markdownContent.innerHTML = marked(data.content);
}
}
后端核心逻辑分析
LLM 大模型链接
# 千问
try:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
return await response.json()
else:
error_text = await response.text()
print(f"千问API调用失败: {response.status} - {error_text}")
return {"error": f"API调用失败: {response.status}", "details": error_text}
except Exception as e:
print(f"千问API请求异常: {str(e)}")
return {"error": f"API请求异常: {str(e)}"}
# deepseek
try:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
return await response.json()
else:
error_text = await response.text()
print(f"DeepSeek API调用失败: {response.status} - {error_text}")
return {"error": f"API调用失败: {response.status}", "details": error_text}
except Exception as e:
print(f"DeepSeek API请求异常: {str(e)}")
return {"error": f"API请求异常: {str(e)}"}
提取 pdf 文字与 图片
# 打开PDF文件
pdf_doc = fitz.open(file_path)
content_by_page = {}
# 遍历每一页
for page_idx in range(len(pdf_doc)):
page = pdf_doc[page_idx]
page_content = {
"page_num": page_idx + 1, # 页码从1开始
"text": page.get_text(),
"images": []
}
# 提取图片
image_list = page.get_images(full=True)
# 处理页面上的每张图片
for img_idx, img_info in enumerate(image_list):
try:
xref = img_info[0] # 图片的xref
base_image = pdf_doc.extract_image(xref)
image_bytes = base_image["image"]
image_ext = base_image["ext"]
# 创建PIL图像对象
image = Image.open(io.BytesIO(image_bytes))
# 将图像转换为base64编码
buffered = io.BytesIO()
image.save(buffered, format=image.format if image.format else "PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
# 存储图片信息
page_content["images"].append({
"index": img_idx + 1, # 图片索引从1开始
"format": image_ext,
"width": image.width,
"height": image.height,
"base64": img_base64
})
except Exception as e:
print(f"处理图片时出错 (页面 {page_idx+1}, 图片 {img_idx+1}): {str(e)}")
# 将页面内容添加到结果中
content_by_page[page_idx + 1] = page_content
# 关闭PDF文件
pdf_doc.close()
return content_by_page
提取图片跟文字后，传给 后端上传文件接口，去实现解析图片跟文字，最后输出 Markdown 文档下载
async def process_pdf_analysis(task_id: str, file_path: str, file_name: str):
"""
处理PDF分析的后台任务
"""
try:
# 调用PDF分析函数
result_path = await analyze_pdf(
file_path=file_path,
file_name=file_name,
task_id=task_id,
status_callback=lambda progress: update_task_progress(task_id, progress)
)
# 更新任务状态为已完成
active_tasks[task_id]["status"] = "completed"
active_tasks[task_id]["result_path"] = result_path
active_tasks[task_id]["progress"] = 100
# 将结果文件复制到静态目录
os.makedirs("static/results", exist_ok=True)
with open(result_path, "r", encoding="utf-8") as src_file:
with open(f"static/results/{task_id}.md", "w", encoding="utf-8") as dst_file:
dst_file.write(src_file.read())
except Exception as e:
# 发生错误时，更新任务状态
active_tasks[task_id]["status"] = "failed"
active_tasks[task_id]["error"] = str(e)
print(f"处理文件时出错: {str(e)}")
# 记录详细的错误信息
import traceback
print(traceback.format_exc())
最后即可实现如下效果
实现效果
原始PDF
粤ICP备14082021号
免费POC，
零成本试错
