DeepSeek-OCR-2开发者案例：集成至RAG系统实现图文混合检索增强

本文介绍了如何利用星图GPU平台自动化部署DeepSeek-OCR-2镜像，构建图文混合检索增强RAG系统。通过该平台，开发者可快速搭建环境，将DeepSeek-OCR-2集成至文档问答系统，高效识别PDF中的图文内容，实现基于图片和文字的混合检索，提升智能问答的准确性与实用性。

KX-EZ

296人浏览 · 2026-03-27 05:11:56

KX-EZ · 2026-03-27 05:11:56 发布

DeepSeek-OCR-2开发者案例：集成至RAG系统实现图文混合检索增强

1. 项目背景与需求

最近在做一个智能文档问答系统，客户的需求很明确：他们有很多PDF文档，里面既有文字又有图片，用户提问时，系统要能同时理解文字内容和图片信息。比如用户问“第三页那个图表说明了什么趋势”，传统的RAG系统只能检索文字，对图片内容完全无能为力。

这就是我们遇到的核心痛点——现有的OCR技术要么识别准确率不够，要么处理速度太慢，要么对复杂排版束手无策。直到DeepSeek-OCR-2的出现，让我们看到了解决这个问题的希望。

DeepSeek-OCR-2是DeepSeek在2026年初发布的开源模型，它采用了一种创新的DeepEncoder V2方法。简单来说，这个模型不是机械地从左到右扫描图片，而是能理解图像的含义，然后智能地重排图像的各个部分。这种设计让它在维持高数据压缩效率的同时，在多项基准测试中取得了显著突破。

最吸引我的是它的效率：模型仅需256到1120个视觉Token就能覆盖复杂的文档页面。在OmniDocBench v1.5评测中，它的综合得分达到了91.09%。这意味着什么？意味着我们可以在保持高质量识别的同时，大幅降低计算成本。

2. 技术架构设计

2.1 整体架构思路

我们的目标很明确：构建一个能够同时处理图文混合内容的RAG系统。整个架构分为三个核心模块：

OCR识别模块：使用DeepSeek-OCR-2从PDF中提取文字和图片内容
推理加速模块：通过vLLM对OCR识别进行加速，提升处理效率
前端展示模块：用Gradio构建用户界面，方便测试和演示

这三个模块协同工作，形成一个完整的图文混合检索增强系统。

2.2 为什么选择这些技术

选择DeepSeek-OCR-2的原因前面已经说了，它的识别准确率和效率都很出色。但光有好的OCR模型还不够，我们还需要考虑实际部署时的性能问题。

这就是我们引入vLLM的原因。vLLM是一个专门为大语言模型推理设计的服务框架，它通过PagedAttention等技术，可以显著提升推理速度，减少内存占用。对于需要处理大量文档的场景来说，这个优化至关重要。

至于前端选择Gradio，主要是考虑到它的易用性和快速原型能力。Gradio可以让我们在几小时内就搭建出一个可交互的演示界面，这对于向客户展示成果、收集反馈非常有帮助。

3. 环境搭建与快速部署

3.1 基础环境准备

首先，我们需要准备一个合适的运行环境。建议使用Python 3.9或更高版本，并确保有足够的GPU内存（至少8GB）。

# 创建虚拟环境
python -m venv ocr_rag_env
source ocr_rag_env/bin/activate  # Linux/Mac
# 或
ocr_rag_env\Scripts\activate  # Windows

# 安装基础依赖
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install transformers accelerate vllm gradio
pip install pdf2image pillow

3.2 DeepSeek-OCR-2模型下载

DeepSeek-OCR-2是开源模型，我们可以直接从Hugging Face下载：

from transformers import AutoModelForVision2Seq, AutoProcessor

# 加载模型和处理器
model = AutoModelForVision2Seq.from_pretrained(
    "deepseek-ai/DeepSeek-OCR-2",
    torch_dtype=torch.float16,
    device_map="auto"
)

processor = AutoProcessor.from_pretrained("deepseek-ai/DeepSeek-OCR-2")

3.3 vLLM服务配置

为了提升推理速度，我们使用vLLM来部署OCR模型：

from vllm import LLM, SamplingParams

# 初始化vLLM引擎
llm = LLM(
    model="deepseek-ai/DeepSeek-OCR-2",
    tensor_parallel_size=1,  # 根据GPU数量调整
    gpu_memory_utilization=0.9,
    max_model_len=4096
)

4. 核心功能实现

4.1 PDF文档处理模块

PDF文档的处理是整个系统的第一步。我们需要将PDF转换为图片，然后交给OCR模型识别。

import os
from pdf2image import convert_from_path
from PIL import Image
import tempfile

class PDFProcessor:
    def __init__(self, dpi=300):
        self.dpi = dpi
        
    def pdf_to_images(self, pdf_path):
        """将PDF转换为图片列表"""
        images = convert_from_path(pdf_path, dpi=self.dpi)
        return images
    
    def process_pdf(self, pdf_path, output_dir=None):
        """处理PDF文档，返回识别结果"""
        if output_dir is None:
            output_dir = tempfile.mkdtemp()
            
        images = self.pdf_to_images(pdf_path)
        results = []
        
        for i, image in enumerate(images):
            # 保存临时图片
            temp_path = os.path.join(output_dir, f"page_{i+1}.png")
            image.save(temp_path, "PNG")
            
            # OCR识别
            ocr_result = self.ocr_recognize(image)
            results.append({
                "page": i + 1,
                "image_path": temp_path,
                "text": ocr_result["text"],
                "bounding_boxes": ocr_result.get("bounding_boxes", [])
            })
            
        return results

4.2 OCR识别与vLLM加速

这是系统的核心部分，我们使用DeepSeek-OCR-2进行识别，并通过vLLM进行加速。

class OCRService:
    def __init__(self, use_vllm=True):
        self.use_vllm = use_vllm
        
        if use_vllm:
            # 使用vLLM加速
            self.llm = LLM(
                model="deepseek-ai/DeepSeek-OCR-2",
                tensor_parallel_size=1,
                max_model_len=4096
            )
        else:
            # 使用标准transformers
            self.model = AutoModelForVision2Seq.from_pretrained(
                "deepseek-ai/DeepSeek-OCR-2",
                torch_dtype=torch.float16,
                device_map="auto"
            )
            self.processor = AutoProcessor.from_pretrained("deepseek-ai/DeepSeek-OCR-2")
    
    def recognize_image(self, image):
        """识别单张图片"""
        if self.use_vllm:
            return self._recognize_with_vllm(image)
        else:
            return self._recognize_with_transformers(image)
    
    def _recognize_with_vllm(self, image):
        """使用vLLM加速识别"""
        # 将图片转换为模型输入格式
        # 这里需要根据实际模型输入格式进行调整
        prompts = self._prepare_prompts(image)
        
        sampling_params = SamplingParams(
            temperature=0.1,
            top_p=0.9,
            max_tokens=1024
        )
        
        outputs = self.llm.generate(prompts, sampling_params)
        
        # 解析输出结果
        result = self._parse_output(outputs[0].outputs[0].text)
        return result
    
    def _recognize_with_transformers(self, image):
        """使用标准transformers识别"""
        inputs = self.processor(
            images=image,
            return_tensors="pt"
        ).to(self.model.device)
        
        generated_ids = self.model.generate(
            **inputs,
            max_new_tokens=1024
        )
        
        generated_text = self.processor.batch_decode(
            generated_ids,
            skip_special_tokens=True
        )[0]
        
        return {"text": generated_text}
    
    def _prepare_prompts(self, image):
        """准备vLLM输入提示"""
        # 这里需要根据DeepSeek-OCR-2的具体输入格式来编写
        # 示例格式，实际使用时需要调整
        prompt = f"请识别以下图片中的文字内容：\n[IMAGE_PLACEHOLDER]"
        return [prompt]
    
    def _parse_output(self, text):
        """解析模型输出"""
        # 根据模型输出格式进行解析
        # 这里可以提取文字内容、位置信息等
        return {"text": text.strip()}

4.3 RAG系统集成

将OCR识别结果集成到RAG系统中，实现图文混合检索。

class HybridRAGSystem:
    def __init__(self, ocr_service, embedding_model="text-embedding-3-small"):
        self.ocr_service = ocr_service
        self.embedding_model = embedding_model
        self.documents = []  # 存储文档信息
        self.embeddings = []  # 存储向量
        
    def add_document(self, pdf_path):
        """添加PDF文档到系统"""
        # 处理PDF
        processor = PDFProcessor()
        pages = processor.process_pdf(pdf_path)
        
        for page in pages:
            # 提取文本内容
            text_content = page["text"]
            
            # 如果有图片，添加图片描述
            if page.get("image_path"):
                # 这里可以添加图片特征提取
                image_features = self._extract_image_features(page["image_path"])
                text_content += f"\n[图片特征: {image_features}]"
            
            # 生成向量
            embedding = self._get_embedding(text_content)
            
            # 存储文档信息
            doc_info = {
                "source": pdf_path,
                "page": page["page"],
                "content": text_content,
                "embedding": embedding,
                "has_image": page.get("image_path") is not None
            }
            
            self.documents.append(doc_info)
            self.embeddings.append(embedding)
    
    def query(self, question, top_k=5):
        """查询相关文档"""
        # 生成问题向量
        question_embedding = self._get_embedding(question)
        
        # 计算相似度
        similarities = self._calculate_similarities(question_embedding)
        
        # 获取最相关的文档
        top_indices = similarities.argsort()[-top_k:][::-1]
        results = []
        
        for idx in top_indices:
            results.append({
                "document": self.documents[idx],
                "similarity": similarities[idx]
            })
        
        return results
    
    def _get_embedding(self, text):
        """获取文本向量"""
        # 这里可以使用OpenAI Embedding、本地模型等
        # 示例使用简单的TF-IDF，实际项目中建议使用更好的嵌入模型
        return self._simple_embedding(text)
    
    def _simple_embedding(self, text):
        """简单的文本向量化方法（示例）"""
        # 实际项目中应该使用专业的嵌入模型
        words = text.lower().split()
        word_set = set(words)
        return list(word_set)  # 简化示例
    
    def _calculate_similarities(self, query_embedding):
        """计算相似度"""
        # 简化示例，实际应该使用余弦相似度等
        similarities = []
        for doc_embedding in self.embeddings:
            # 计算Jaccard相似度（示例）
            intersection = len(set(query_embedding) & set(doc_embedding))
            union = len(set(query_embedding) | set(doc_embedding))
            similarity = intersection / union if union > 0 else 0
            similarities.append(similarity)
        
        return similarities
    
    def _extract_image_features(self, image_path):
        """提取图片特征"""
        # 这里可以集成图片特征提取模型
        # 示例返回简单描述
        return "包含图表或图形的页面"

5. Gradio前端界面

为了让用户能够方便地使用系统，我们使用Gradio构建了一个简单的前端界面。

import gradio as gr
import tempfile

class GradioInterface:
    def __init__(self, rag_system):
        self.rag_system = rag_system
        
    def create_interface(self):
        """创建Gradio界面"""
        with gr.Blocks(title="图文混合RAG系统") as demo:
            gr.Markdown("# 📄 图文混合RAG系统")
            gr.Markdown("上传PDF文档，系统会自动识别文字和图片内容，支持混合检索。")
            
            with gr.Tab("文档上传"):
                with gr.Row():
                    pdf_input = gr.File(label="上传PDF文档", file_types=[".pdf"])
                    upload_btn = gr.Button("上传并处理", variant="primary")
                
                upload_output = gr.JSON(label="处理结果")
                
                upload_btn.click(
                    fn=self.process_upload,
                    inputs=pdf_input,
                    outputs=upload_output
                )
            
            with gr.Tab("文档检索"):
                with gr.Row():
                    query_input = gr.Textbox(
                        label="输入问题",
                        placeholder="例如：第三页的图表说明了什么？"
                    )
                    search_btn = gr.Button("搜索", variant="primary")
                
                search_output = gr.JSON(label="检索结果")
                
                search_btn.click(
                    fn=self.process_query,
                    inputs=query_input,
                    outputs=search_output
                )
            
            with gr.Tab("系统状态"):
                status_output = gr.JSON(label="系统状态")
                refresh_btn = gr.Button("刷新状态")
                
                refresh_btn.click(
                    fn=self.get_status,
                    inputs=[],
                    outputs=status_output
                )
        
        return demo
    
    def process_upload(self, pdf_file):
        """处理上传的PDF文件"""
        if pdf_file is None:
            return {"error": "请先上传PDF文件"}
        
        try:
            # 保存临时文件
            temp_dir = tempfile.mkdtemp()
            pdf_path = os.path.join(temp_dir, "uploaded.pdf")
            
            # 保存上传的文件
            with open(pdf_path, "wb") as f:
                f.write(pdf_file)
            
            # 添加到RAG系统
            self.rag_system.add_document(pdf_path)
            
            return {
                "status": "success",
                "message": f"文档处理完成，已添加到检索系统",
                "file_size": os.path.getsize(pdf_path),
                "pages_processed": len([d for d in self.rag_system.documents if d["source"] == pdf_path])
            }
            
        except Exception as e:
            return {"error": str(e)}
    
    def process_query(self, query):
        """处理用户查询"""
        if not query.strip():
            return {"error": "请输入查询内容"}
        
        try:
            results = self.rag_system.query(query, top_k=3)
            
            formatted_results = []
            for result in results:
                doc = result["document"]
                formatted_results.append({
                    "来源": doc["source"],
                    "页码": doc["page"],
                    "相关度": round(result["similarity"], 3),
                    "包含图片": doc["has_image"],
                    "内容摘要": doc["content"][:200] + "..." if len(doc["content"]) > 200 else doc["content"]
                })
            
            return {
                "query": query,
                "results": formatted_results,
                "total_documents": len(self.rag_system.documents)
            }
            
        except Exception as e:
            return {"error": str(e)}
    
    def get_status(self):
        """获取系统状态"""
        return {
            "总文档数": len(self.rag_system.documents),
            "包含图片的文档": len([d for d in self.rag_system.documents if d["has_image"]]),
            "系统状态": "运行正常",
            "OCR模型": "DeepSeek-OCR-2",
            "加速引擎": "vLLM"
        }

# 启动Gradio界面
def launch_interface():
    # 初始化服务
    ocr_service = OCRService(use_vllm=True)
    rag_system = HybridRAGSystem(ocr_service)
    interface = GradioInterface(rag_system)
    
    demo = interface.create_interface()
    demo.launch(server_name="0.0.0.0", server_port=7860)

if __name__ == "__main__":
    launch_interface()