Deepseek 多模态来解析图片，结合上下文分析pdf文档

打造你的文档小助手：PDF智能解析系统揭秘！

嘿，朋友们！你有没有为翻阅一堆PDF文件头晕眼花过？有没有为从文档中提取关键信息熬夜到怀疑人生？别担心！今天我要隆重介绍一款“文档界的卷王” ——PDF智能解析系统！它不仅能帮你轻松搞定PDF，还能生成酷炫的分析报告，简直是懒人福音！

这货到底有啥用？

简单来说，这套系统能让你从“手动狗头”进化到“智能AI狗头”。它的核心功能包括：

1. 1. PDF上传：轻轻一点，文件上传，so easy！
2. 2. 智能分析：AI出马，帮你挖掘文档中的隐藏宝藏。
3. 3. 实时反馈：进度条实时更新，告别“等到天荒地老”。
4. 4. 报告生成：自动输出Markdown格式报告，简洁又高效。
5. 5. 预览下载：预览先过目，满意再下载，完美！

它是怎么运作的？

整个系统架构采用了“前后端分离”，听着高大上，其实就是“分工明确，各干各的”：

• • 前端：HTML/CSS/JavaScript三剑客在线打工，再加上marked.js负责Markdown渲染，用户体验感直接拉满！
• • 后端：用Python，负责文件处理和分析逻辑，API接口设计得贼精致。

亮点功能大揭秘

1. 1. 任务状态轮询：上传文件后，系统会给你一个任务ID，然后开始疯狂“打探消息”。“任务完成了吗？”、“进度多少了？”系统比你还着急！
2. 2. Markdown渲染：用marked.js把分析结果变成赏心悦目的Markdown格式，你甚至可以假装自己是个写作高手！
3. 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格式的分析报告

系统架构

本系统使用以下技术栈：

• • FastAPI：构建高性能的Web API
• • PyMuPDF：提取PDF文件中的文本和图像
• • Qwen-VL-Max：处理和分析图像内容
• • DeepSeek-Chat：分析和理解文本内容
• • Pydantic：数据验证和模型定义
• • Pillow：图像处理

安装与设置

环境要求

• • Python 3.11

安装步骤

1.. 创建并激活虚拟环境（可选）

python -m venv venv
# 在Windows上
venv\Scripts\activate
# 在macOS/Linux上
source venv/bin/activate

1. 3. 安装依赖

pip install -r requirements.txt 
pip install -r requirements.txt - i  https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple

1. 4. 配置API密钥

创建.env文件并添加以下内容：

QIANWEN_API_KEY=你的千问API密钥
DEEPSEEK_API_KEY=你的DeepSeek API密钥

运行服务

uvicorn main:app --reload

默认情况下，服务将在 http://localhost:8000 上运行。

API接口说明

上传PDF文件进行分析

• • 端点：POST /upload
• • 请求类型：multipart/form-data
• • 参数：
• • file：PDF文件

查询任务状态

• • 端点：GET /task/{task_id}
• • 参数：
• • task_id：任务ID

下载分析结果

• • 端点：GET /download/{task_id}
• • 参数：
• • task_id：任务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