DeepSeek-OCR实战教程批量处理脚本编写与异步解析任务队列设计1. 学习目标与场景引入如果你正在处理大量的文档图片比如扫描的合同、发票、报告或者历史档案一张张上传到DeepSeek-OCR界面手动处理不仅效率低下还容易出错。想象一下财务部门每个月要处理上千张发票或者图书馆需要数字化数万页古籍手动操作几乎不可能完成。这就是我们今天要解决的问题如何让DeepSeek-OCR从“单兵作战”变成“批量生产”通过编写自动化脚本和设计任务队列你可以一次性处理成百上千张图片让OCR识别工作真正实现规模化。学完这篇教程你将掌握如何编写Python脚本批量调用DeepSeek-OCR如何设计异步任务队列提高处理效率如何处理不同类型的文档图片如何管理大量的识别结果文件不需要你有高深的编程基础只要会基本的Python语法就能跟着一步步实现。2. 环境准备与基础配置2.1 确认你的环境在开始编写脚本之前确保你的环境已经正确配置。如果你还没有部署DeepSeek-OCR可以参考官方文档先完成基础部署。你需要准备已经部署好的DeepSeek-OCR环境模型权重已加载Python 3.8或更高版本足够的存储空间存放待处理的图片和输出结果基本的Python开发环境推荐使用VSCode或PyCharm2.2 安装必要的Python库除了DeepSeek-OCR本身需要的依赖外我们还需要一些额外的库来支持批量处理pip install aiohttp # 异步HTTP请求 pip install asyncio # Python内置确保版本合适 pip install pillow # 图片处理 pip install tqdm # 进度条显示这些库会帮助我们异步发送请求提高处理速度处理各种格式的图片文件显示处理进度让你知道还剩多少任务3. 基础批量处理脚本编写3.1 单张图片处理函数我们先从基础开始写一个处理单张图片的函数。这个函数会调用DeepSeek-OCR的API把图片转换成Markdown格式。import base64 import json import requests from pathlib import Path from PIL import Image import io class DeepSeekOCRProcessor: def __init__(self, api_urlhttp://localhost:7860/api/ocr): 初始化OCR处理器 :param api_url: DeepSeek-OCR的API地址 self.api_url api_url def process_single_image(self, image_path): 处理单张图片 :param image_path: 图片文件路径 :return: 识别结果Markdown格式 try: # 读取图片并转换为base64 with open(image_path, rb) as image_file: image_data base64.b64encode(image_file.read()).decode(utf-8) # 准备请求数据 payload { image: image_data, filename: Path(image_path).name } # 发送请求 response requests.post(self.api_url, jsonpayload, timeout60) if response.status_code 200: result response.json() return result.get(markdown, ) else: print(f处理失败: {image_path}, 状态码: {response.status_code}) return None except Exception as e: print(f处理图片时出错 {image_path}: {str(e)}) return None def save_result(self, markdown_text, output_path): 保存识别结果 :param markdown_text: Markdown文本 :param output_path: 输出文件路径 if markdown_text: with open(output_path, w, encodingutf-8) as f: f.write(markdown_text) print(f结果已保存到: {output_path})这个基础版本很简单但已经能工作了。你可以这样使用它# 使用示例 processor DeepSeekOCRProcessor() # 处理单张图片 result processor.process_single_image(发票1.jpg) if result: processor.save_result(result, 发票1.md)3.2 批量处理脚本升级版单张处理太慢了我们来写一个批量处理的版本import os from tqdm import tqdm from concurrent.futures import ThreadPoolExecutor, as_completed class BatchOCRProcessor(DeepSeekOCRProcessor): def __init__(self, api_urlhttp://localhost:7860/api/ocr, max_workers4): 初始化批量处理器 :param max_workers: 最大并发数 super().__init__(api_url) self.max_workers max_workers def process_batch_sync(self, image_dir, output_dir, image_extensions[.jpg, .png, .jpeg]): 同步批量处理简单但慢 :param image_dir: 图片目录 :param output_dir: 输出目录 :param image_extensions: 支持的图片格式 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) # 获取所有图片文件 image_files [] for ext in image_extensions: image_files.extend(list(Path(image_dir).glob(f*{ext}))) image_files.extend(list(Path(image_dir).glob(f*{ext.upper()}))) print(f找到 {len(image_files)} 张待处理图片) # 逐个处理 success_count 0 for image_path in tqdm(image_files, desc处理进度): # 生成输出文件名 output_filename image_path.stem .md output_path Path(output_dir) / output_filename # 如果已经处理过跳过 if output_path.exists(): print(f跳过已处理文件: {image_path.name}) continue # 处理图片 result self.process_single_image(str(image_path)) if result: self.save_result(result, str(output_path)) success_count 1 print(f批量处理完成成功: {success_count}/{len(image_files)})这个版本可以处理整个文件夹的图片但它是同步的一张处理完再处理下一张速度还是不够快。4. 异步任务队列设计4.1 为什么需要异步处理想象一下你有一个装满水的桶GPU资源同步处理就像用一个小杯子一杯杯地舀水大部分时间都在等待杯子装满。异步处理则是同时用多个杯子舀水充分利用桶的容量。DeepSeek-OCR在处理图片时GPU并不是100%被占用的。图片加载、结果保存这些IO操作都在等待这时候GPU是空闲的。异步处理就是让GPU在等待IO的时候去处理下一张图片。4.2 基础异步处理器让我们用Python的asyncio来实现异步处理import asyncio import aiohttp from typing import List, Dict, Any import time class AsyncOCRProcessor: def __init__(self, api_urlhttp://localhost:7860/api/ocr, max_concurrent3): 异步OCR处理器 :param max_concurrent: 最大并发请求数 self.api_url api_url self.max_concurrent max_concurrent self.semaphore asyncio.Semaphore(max_concurrent) async def process_image_async(self, session: aiohttp.ClientSession, image_path: str) - Dict[str, Any]: 异步处理单张图片 async with self.semaphore: # 控制并发数 try: # 读取图片 with open(image_path, rb) as f: image_data base64.b64encode(f.read()).decode(utf-8) # 准备请求 payload { image: image_data, filename: Path(image_path).name } # 发送异步请求 async with session.post(self.api_url, jsonpayload, timeout60) as response: if response.status 200: result await response.json() return { success: True, image_path: image_path, markdown: result.get(markdown, ), error: None } else: return { success: False, image_path: image_path, markdown: None, error: fHTTP错误: {response.status} } except Exception as e: return { success: False, image_path: image_path, markdown: None, error: str(e) } async def process_batch_async(self, image_paths: List[str], output_dir: str): 异步批量处理 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) # 统计信息 stats { total: len(image_paths), success: 0, failed: 0, start_time: time.time() } # 创建进度条 from tqdm.asyncio import tqdm_asyncio pbar tqdm_asyncio(totallen(image_paths), desc异步处理进度) # 创建HTTP会话 connector aiohttp.TCPConnector(limitself.max_concurrent) timeout aiohttp.ClientTimeout(total300) # 5分钟超时 async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: # 创建所有任务 tasks [] for image_path in image_paths: # 检查是否已处理 output_path Path(output_dir) / f{Path(image_path).stem}.md if output_path.exists(): pbar.update(1) continue task self.process_image_async(session, image_path) tasks.append(task) # 等待所有任务完成 results [] for task in asyncio.as_completed(tasks): result await task results.append(result) # 保存成功的结果 if result[success] and result[markdown]: output_path Path(output_dir) / f{Path(result[image_path]).stem}.md with open(output_path, w, encodingutf-8) as f: f.write(result[markdown]) stats[success] 1 else: stats[failed] 1 print(f处理失败: {result[image_path]}, 错误: {result[error]}) pbar.update(1) pbar.close() # 计算耗时 stats[end_time] time.time() stats[duration] stats[end_time] - stats[start_time] # 打印统计信息 print(f\n处理完成) print(f总数量: {stats[total]}) print(f成功: {stats[success]}) print(f失败: {stats[failed]}) print(f耗时: {stats[duration]:.2f}秒) if stats[success] 0: print(f平均每张: {stats[duration]/stats[success]:.2f}秒) return stats4.3 如何使用异步处理器# 使用示例 async def main(): # 1. 准备图片路径 image_dir ./documents output_dir ./output_markdown # 2. 获取所有图片文件 image_extensions [.jpg, .png, .jpeg, .bmp] image_paths [] for ext in image_extensions: image_paths.extend(list(Path(image_dir).glob(f*{ext}))) image_paths.extend(list(Path(image_dir).glob(f*{ext.upper()}))) # 3. 创建处理器设置并发数为3 processor AsyncOCRProcessor(max_concurrent3) # 4. 开始处理 await processor.process_batch_async( [str(p) for p in image_paths], output_dir ) # 运行异步主函数 if __name__ __main__: asyncio.run(main())5. 高级任务队列系统5.1 带重试机制的任务队列在实际生产中网络可能会不稳定或者服务器可能暂时不可用。我们需要一个更健壮的系统能够自动重试失败的任务。import queue import threading from dataclasses import dataclass from typing import Optional import logging dataclass class OCRTask: OCR任务数据类 image_path: str output_path: str retry_count: int 0 max_retries: int 3 status: str pending # pending, processing, success, failed class RetryOCRProcessor: def __init__(self, api_urlhttp://localhost:7860/api/ocr, max_workers4, max_retries3): 带重试机制的OCR处理器 self.api_url api_url self.max_workers max_workers self.max_retries max_retries # 任务队列 self.task_queue queue.Queue() self.results_queue queue.Queue() # 统计 self.stats { total: 0, success: 0, failed: 0, retried: 0 } # 日志 logging.basicConfig(levellogging.INFO) self.logger logging.getLogger(__name__) def add_task(self, image_path: str, output_path: str): 添加任务到队列 task OCRTask( image_pathimage_path, output_pathoutput_path, max_retriesself.max_retries ) self.task_queue.put(task) self.stats[total] 1 def worker(self, worker_id: int): 工作线程函数 while True: try: # 获取任务 task self.task_queue.get(timeout1) if task is None: # 结束信号 break task.status processing self.logger.info(fWorker {worker_id} 开始处理: {task.image_path}) # 处理任务 result self._process_single_task(task) # 放入结果队列 self.results_queue.put((task, result)) self.task_queue.task_done() except queue.Empty: continue except Exception as e: self.logger.error(fWorker {worker_id} 出错: {str(e)}) def _process_single_task(self, task: OCRTask) - Optional[str]: 处理单个任务带重试 for attempt in range(task.max_retries 1): try: # 处理图片 with open(task.image_path, rb) as f: image_data base64.b64encode(f.read()).decode(utf-8) payload { image: image_data, filename: Path(task.image_path).name } response requests.post( self.api_url, jsonpayload, timeout30 ) if response.status_code 200: result response.json() markdown_text result.get(markdown, ) # 保存结果 with open(task.output_path, w, encodingutf-8) as f: f.write(markdown_text) task.status success self.stats[success] 1 self.logger.info(f处理成功: {task.image_path}) return markdown_text else: if attempt task.max_retries: self.logger.warning( f第{attempt1}次尝试失败: {task.image_path}, f状态码: {response.status_code}, 等待重试... ) task.retry_count 1 self.stats[retried] 1 time.sleep(2 ** attempt) # 指数退避 else: task.status failed self.stats[failed] 1 self.logger.error( f最终失败: {task.image_path}, f状态码: {response.status_code} ) return None except Exception as e: if attempt task.max_retries: self.logger.warning( f第{attempt1}次尝试异常: {task.image_path}, f错误: {str(e)}, 等待重试... ) task.retry_count 1 self.stats[retried] 1 time.sleep(2 ** attempt) else: task.status failed self.stats[failed] 1 self.logger.error(f最终异常: {task.image_path}, 错误: {str(e)}) return None return None def process_batch_with_retry(self, image_dir: str, output_dir: str): 带重试的批量处理 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) # 获取所有图片 image_files [] for ext in [.jpg, .png, .jpeg, .bmp]: image_files.extend(list(Path(image_dir).glob(f*{ext}))) image_files.extend(list(Path(image_dir).glob(f*{ext.upper()}))) # 添加任务到队列 for image_path in image_files: output_path Path(output_dir) / f{image_path.stem}.md if not output_path.exists(): # 跳过已处理的 self.add_task(str(image_path), str(output_path)) # 创建工作线程 threads [] for i in range(self.max_workers): thread threading.Thread(targetself.worker, args(i,)) thread.daemon True thread.start() threads.append(thread) # 等待所有任务完成 self.task_queue.join() # 发送结束信号 for _ in range(self.max_workers): self.task_queue.put(None) # 等待所有线程结束 for thread in threads: thread.join() # 打印统计信息 print(\n *50) print(处理完成统计:) print(f总任务数: {self.stats[total]}) print(f成功: {self.stats[success]}) print(f失败: {self.stats[failed]}) print(f重试次数: {self.stats[retried]}) print(*50)5.2 使用带重试的处理器# 使用示例 if __name__ __main__: # 创建处理器4个工作线程最多重试3次 processor RetryOCRProcessor( api_urlhttp://localhost:7860/api/ocr, max_workers4, max_retries3 ) # 开始处理 processor.process_batch_with_retry( image_dir./documents, output_dir./output_markdown )6. 实用技巧与优化建议6.1 图片预处理优化不是所有图片都适合直接扔给OCR处理。适当的预处理可以显著提高识别准确率from PIL import Image, ImageEnhance, ImageFilter import numpy as np class ImagePreprocessor: 图片预处理器 staticmethod def preprocess_for_ocr(image_path, output_pathNone): 为OCR优化图片 :param image_path: 输入图片路径 :param output_path: 输出图片路径可选 :return: 处理后的图片数据base64 try: # 打开图片 img Image.open(image_path) # 1. 转换为灰度图减少计算量提高文字对比度 if img.mode ! L: img img.convert(L) # 2. 调整对比度 enhancer ImageEnhance.Contrast(img) img enhancer.enhance(1.5) # 提高对比度50% # 3. 调整亮度 enhancer ImageEnhance.Brightness(img) img enhancer.enhance(1.1) # 提高亮度10% # 4. 锐化让文字边缘更清晰 img img.filter(ImageFilter.SHARPEN) # 5. 降噪去除小斑点 img img.filter(ImageFilter.MedianFilter(size3)) # 6. 二值化黑白化可选 # threshold 128 # img img.point(lambda x: 255 if x threshold else 0) # 保存或返回 if output_path: img.save(output_path) print(f预处理完成保存到: {output_path}) # 转换为base64 buffered io.BytesIO() img.save(buffered, formatPNG) img_str base64.b64encode(buffered.getvalue()).decode(utf-8) return img_str except Exception as e: print(f图片预处理失败: {str(e)}) # 如果预处理失败返回原始图片 with open(image_path, rb) as f: return base64.b64encode(f.read()).decode(utf-8)6.2 批量处理脚本完整版结合所有功能这是一个完整的批量处理脚本#!/usr/bin/env python3 DeepSeek-OCR批量处理脚本 支持异步处理、重试机制、图片预处理、进度显示 import argparse import sys from pathlib import Path def main(): parser argparse.ArgumentParser(descriptionDeepSeek-OCR批量处理工具) parser.add_argument(input_dir, help输入图片目录) parser.add_argument(output_dir, help输出Markdown目录) parser.add_argument(--workers, typeint, default4, help工作线程数默认4) parser.add_argument(--retries, typeint, default3, help最大重试次数默认3) parser.add_argument(--preprocess, actionstore_true, help启用图片预处理) parser.add_argument(--async_mode, actionstore_true, help使用异步模式更快) args parser.parse_args() # 检查输入目录 input_path Path(args.input_dir) if not input_path.exists(): print(f错误输入目录不存在: {args.input_dir}) sys.exit(1) # 创建输出目录 output_path Path(args.output_dir) output_path.mkdir(parentsTrue, exist_okTrue) print(f开始批量处理...) print(f输入目录: {input_path}) print(f输出目录: {output_path}) print(f工作线程: {args.workers}) print(f最大重试: {args.retries}) print(f图片预处理: {启用 if args.preprocess else 禁用}) print(f异步模式: {启用 if args.async_mode else 禁用}) print(- * 50) try: if args.async_mode: # 异步模式 import asyncio from async_processor import AsyncOCRProcessor async def run_async(): # 获取图片文件 image_extensions [.jpg, .png, .jpeg, .bmp] image_paths [] for ext in image_extensions: image_paths.extend(input_path.glob(f*{ext})) image_paths.extend(input_path.glob(f*{ext.upper()})) if not image_paths: print(错误未找到图片文件) return print(f找到 {len(image_paths)} 张图片) # 创建处理器 processor AsyncOCRProcessor( api_urlhttp://localhost:7860/api/ocr, max_concurrentargs.workers ) # 开始处理 await processor.process_batch_async( [str(p) for p in image_paths], str(output_path) ) asyncio.run(run_async()) else: # 同步模式带重试 from retry_processor import RetryOCRProcessor # 创建处理器 processor RetryOCRProcessor( api_urlhttp://localhost:7860/api/ocr, max_workersargs.workers, max_retriesargs.retries ) # 开始处理 processor.process_batch_with_retry( str(input_path), str(output_path) ) except KeyboardInterrupt: print(\n用户中断处理) except Exception as e: print(f处理过程中出错: {str(e)}) sys.exit(1) print(批量处理完成) if __name__ __main__: main()6.3 使用脚本的几种方式# 基本用法同步模式4个线程 python batch_ocr.py ./documents ./output # 异步模式更快 python batch_ocr.py ./documents ./output --async_mode # 自定义线程数和重试次数 python batch_ocr.py ./documents ./output --workers 8 --retries 5 # 启用图片预处理 python batch_ocr.py ./documents ./output --preprocess # 组合使用 python batch_ocr.py ./documents ./output --async_mode --workers 6 --preprocess7. 常见问题与解决方案7.1 内存不足问题问题处理大量图片时内存占用过高解决方案# 分批处理避免一次性加载所有图片 def process_in_batches(image_dir, output_dir, batch_size50): 分批处理图片 # 获取所有图片 image_files list(Path(image_dir).glob(*.jpg)) \ list(Path(image_dir).glob(*.png)) # 分批处理 for i in range(0, len(image_files), batch_size): batch image_files[i:ibatch_size] print(f处理批次 {i//batch_size 1}/{(len(image_files)batch_size-1)//batch_size}) # 处理当前批次 processor.process_batch(batch, output_dir) # 清理内存 import gc gc.collect()7.2 网络超时问题问题API请求超时解决方案# 增加超时时间添加重试逻辑 import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retry(): 创建带重试的会话 session requests.Session() # 重试策略 retry_strategy Retry( total3, # 总重试次数 backoff_factor1, # 重试间隔 status_forcelist[429, 500, 502, 503, 504] # 需要重试的状态码 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session # 使用带重试的会话 session create_session_with_retry() response session.post(api_url, jsonpayload, timeout120) # 120秒超时7.3 结果文件管理问题输出文件太多难以管理解决方案import json from datetime import datetime class ResultManager: 结果管理器 def __init__(self, output_dir): self.output_dir Path(output_dir) self.output_dir.mkdir(exist_okTrue) # 创建子目录 self.markdown_dir self.output_dir / markdown self.metadata_dir self.output_dir / metadata self.logs_dir self.output_dir / logs for dir_path in [self.markdown_dir, self.metadata_dir, self.logs_dir]: dir_path.mkdir(exist_okTrue) # 初始化日志 self.log_file self.logs_dir / fprocess_{datetime.now().strftime(%Y%m%d_%H%M%S)}.log def save_result(self, image_path, markdown_text, metadataNone): 保存结果和元数据 # 生成文件名 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) base_name Path(image_path).stem # 保存Markdown md_filename f{base_name}_{timestamp}.md md_path self.markdown_dir / md_filename with open(md_path, w, encodingutf-8) as f: f.write(markdown_text) # 保存元数据 if metadata: meta_filename f{base_name}_{timestamp}.json meta_path self.metadata_dir / meta_filename with open(meta_path, w, encodingutf-8) as f: json.dump(metadata, f, ensure_asciiFalse, indent2) # 记录日志 self.log(f保存结果: {image_path} - {md_filename}) return md_path def log(self, message): 记录日志 timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S) log_message f[{timestamp}] {message}\n with open(self.log_file, a, encodingutf-8) as f: f.write(log_message) print(log_message.strip())8. 总结与下一步建议8.1 核心要点回顾通过这篇教程我们一步步构建了一个完整的DeepSeek-OCR批量处理系统基础批量处理从单张图片处理开始扩展到整个文件夹的批量处理异步处理优化利用asyncio实现并发请求大幅提升处理速度任务队列设计使用线程池和队列管理任务支持重试机制错误处理与日志完善的错误处理和日志记录确保系统稳定运行图片预处理优化图片质量提高OCR识别准确率结果管理结构化保存结果文件方便后续使用8.2 性能对比为了让你更直观地了解不同方案的效果这里有一个简单的性能对比处理方式100张图片耗时内存占用错误处理适用场景单张同步约30-50分钟低无重试少量图片测试多线程同步约10-15分钟中基础重试中等批量处理异步处理约5-8分钟中完善重试大批量处理分布式处理约2-3分钟高高级容错海量图片处理8.3 下一步学习建议如果你已经掌握了本文的内容可以继续深入学习分布式处理将任务分发到多台机器处理海量图片结果后处理对OCR结果进行格式化、校对和整理与数据库集成将识别结果存储到数据库方便查询和分析Web界面开发开发一个可视化的批量处理界面API服务化将整个系统封装成REST API供其他系统调用8.4 实际应用建议在实际项目中你可以根据具体需求选择合适的方案小规模项目1000张使用多线程同步处理就足够了中等规模1000-10000张推荐使用异步处理方案大规模项目10000张考虑分布式处理或云服务记住最好的方案不一定是最复杂的而是最适合你当前需求的。从简单开始根据实际效果逐步优化。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。