# LightOnOCR-2-1B实战教程：Python调用API实现批量图片文字提取代码实例 ## 1. 项目介绍与准备工作 LightOnOCR-2-1B是一个强大的多语言OCR识别模型，拥有10亿参数规模，专门用于从图片中提取文字内容。这个模型支持11种语言，包括中文、英文、日文、法文、德文、西班牙文、意大利文、荷兰文、葡萄牙文、瑞典文和丹麦文，能够满足大多数国际化场景的文字识别需求。 在实际应用中，我们经常需要处理大量图片文件，比如扫描文档、照片文字、表格数据等。手动一个个处理既费时又容易出错，通过编程方式批量调用OCR API就能大大提高工作效率。 ### 1.1 环境准备 在开始编写代码前，我们需要准备以下环境： - Python 3.8或更高版本 - requests库用于API调用 - Pillow库用于图片处理 - 已部署LightOnOCR-2-1B服务的服务器（IP地址替换为实际地址） 安装所需依赖： ```bash pip install requests Pillow ``` ### 1.2 服务确认 确保OCR服务正常运行，可以通过以下命令检查服务状态： ```bash # 检查服务端口是否正常监听 curl -I http://<服务器IP>:8000/v1/chat/completions ``` 如果返回HTTP 200状态码，说明服务正常运行。 ## 2. 单张图片文字提取实战 我们先从最简单的单张图片识别开始，了解基本的API调用方法。 ### 2.1 图片预处理函数 在处理图片前，我们需要将图片转换为Base64编码格式，这是API接收图片的标准方式： ```python import base64 from PIL import Image import io def image_to_base64(image_path, max_size=1540): """ 将图片转换为Base64编码，并自动调整大小 参数: image_path: 图片文件路径 max_size: 图片最长边的最大像素（推荐1540） 返回: Base64编码的图片字符串 """ # 打开图片并调整大小 with Image.open(image_path) as img: # 获取原始尺寸 width, height = img.size # 计算调整比例 if max(width, height) > max_size: scale = max_size / max(width, height) new_width = int(width * scale) new_height = int(height * scale) img = img.resize((new_width, new_height), Image.Resampling.LANCZOS) # 转换为RGB模式（处理PNG透明背景） if img.mode != 'RGB': img = img.convert('RGB') # 保存到内存缓冲区 buffer = io.BytesIO() img.save(buffer, format='JPEG', quality=95) # 转换为Base64 img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') return f"data:image/jpeg;base64,{img_base64}" ``` ### 2.2 基础API调用 下面是调用LightOnOCR API的核心函数： ```python import requests import json def extract_text_from_image(image_path, server_ip): """ 从单张图片中提取文字 参数: image_path: 图片文件路径 server_ip: 服务器IP地址 返回: 识别出的文字内容 """ # 将图片转换为Base64 image_data = image_to_base64(image_path) # 构建API请求 api_url = f"http://{server_ip}:8000/v1/chat/completions" headers = { "Content-Type": "application/json" } payload = { "model": "/root/ai-models/lightonai/LightOnOCR-2-1B", "messages": [{ "role": "user", "content": [{ "type": "image_url", "image_url": {"url": image_data} }] }], "max_tokens": 4096 } try: # 发送请求 response = requests.post(api_url, headers=headers, json=payload, timeout=30) response.raise_for_status() # 解析响应 result = response.json() extracted_text = result['choices'][0]['message']['content'] return extracted_text except requests.exceptions.RequestException as e: print(f"API请求失败: {e}") return None except KeyError as e: print(f"响应解析失败: {e}") return None ``` ### 2.3 简单测试 让我们测试一下单张图片的识别效果： ```python # 使用示例 if __name__ == "__main__": server_ip = "你的服务器IP" # 替换为实际IP image_path = "test_image.jpg" # 测试图片路径 text = extract_text_from_image(image_path, server_ip) if text: print("识别结果:") print(text) else: print("识别失败") ``` ## 3. 批量图片处理实现 现在我们来实现批量处理功能，这是实际工作中最常用的场景。 ### 3.1 批量处理核心类 创建一个专门处理批量OCR的类，包含进度跟踪和错误处理： ```python import os from pathlib import Path import time class BatchOCRProcessor: def __init__(self, server_ip, output_dir="output"): self.server_ip = server_ip self.output_dir = output_dir self.supported_formats = ['.jpg', '.jpeg', '.png', '.bmp', '.tiff'] # 创建输出目录 os.makedirs(output_dir, exist_ok=True) def get_image_files(self, input_path): """ 获取指定路径下的所有图片文件 参数: input_path: 输入路径（文件或目录） 返回: 图片文件路径列表 """ if os.path.isfile(input_path): # 单个文件 if Path(input_path).suffix.lower() in self.supported_formats: return [input_path] else: return [] elif os.path.isdir(input_path): # 目录下的所有图片文件 image_files = [] for format in self.supported_formats: image_files.extend(Path(input_path).glob(f"*{format}")) image_files.extend(Path(input_path).glob(f"*{format.upper()}")) return [str(f) for f in image_files] else: return [] def process_batch(self, image_files, batch_size=5, delay=1): """ 批量处理图片文件 参数: image_files: 图片文件列表 batch_size: 每批处理数量 delay: 批处理间隔（秒） 返回: 处理结果列表 """ results = [] total_files = len(image_files) print(f"开始处理 {total_files} 个文件...") for i in range(0, total_files, batch_size): batch = image_files[i:i+batch_size] batch_results = [] for j, image_path in enumerate(batch): file_num = i + j + 1 print(f"处理中 [{file_num}/{total_files}]: {os.path.basename(image_path)}") try: text = extract_text_from_image(image_path, self.server_ip) if text: # 保存结果 result = { 'file_path': image_path, 'file_name': os.path.basename(image_path), 'extracted_text': text, 'status': 'success' } self.save_result(result) batch_results.append(result) else: batch_results.append({ 'file_path': image_path, 'status': 'failed', 'error': '识别失败' }) except Exception as e: batch_results.append({ 'file_path': image_path, 'status': 'error', 'error': str(e) }) results.extend(batch_results) # 添加延迟，避免服务器过载 if i + batch_size < total_files: time.sleep(delay) return results def save_result(self, result): """ 保存识别结果到文件 参数: result: 识别结果字典 """ base_name = Path(result['file_path']).stem output_file = os.path.join(self.output_dir, f"{base_name}.txt") with open(output_file, 'w', encoding='utf-8') as f: f.write(result['extracted_text']) # 同时保存元数据 meta_file = os.path.join(self.output_dir, f"{base_name}_meta.json") import json with open(meta_file, 'w', encoding='utf-8') as f: json.dump({ 'file_name': result['file_name'], 'original_path': result['file_path'], 'process_time': time.strftime("%Y-%m-%d %H:%M:%S") }, f, ensure_ascii=False, indent=2) ``` ### 3.2 完整的批量处理脚本 下面是一个完整的批量处理示例，包含命令行参数解析： ```python import argparse def main(): parser = argparse.ArgumentParser(description='LightOnOCR批量图片文字提取工具') parser.add_argument('input', help='输入文件或目录路径') parser.add_argument('--server', required=True, help='服务器IP地址') parser.add_argument('--output', default='output', help='输出目录') parser.add_argument('--batch-size', type=int, default=5, help='每批处理数量') parser.add_argument('--delay', type=float, default=1, help='批处理间隔（秒）') args = parser.parse_args() # 创建处理器 processor = BatchOCRProcessor(args.server, args.output) # 获取图片文件 image_files = processor.get_image_files(args.input) if not image_files: print("未找到支持的图片文件") return print(f"找到 {len(image_files)} 个图片文件") # 批量处理 results = processor.process_batch(image_files, args.batch_size, args.delay) # 统计结果 success_count = sum(1 for r in results if r['status'] == 'success') failed_count = sum(1 for r in results if r['status'] == 'failed') error_count = sum(1 for r in results if r['status'] == 'error') print(f"\n处理完成!") print(f"成功: {success_count}, 失败: {failed_count}, 错误: {error_count}") print(f"结果保存在: {args.output}") if __name__ == "__main__": main() ``` ## 4. 高级功能与最佳实践 ### 4.1 并发处理优化 对于大量图片，我们可以使用多线程来提高处理速度： ```python import concurrent.futures class ConcurrentOCRProcessor(BatchOCRProcessor): def process_concurrent(self, image_files, max_workers=4): """ 使用多线程并发处理 参数: image_files: 图片文件列表 max_workers: 最大线程数 返回: 处理结果列表 """ results = [] def process_single(image_path): try: text = extract_text_from_image(image_path, self.server_ip) if text: result = { 'file_path': image_path, 'file_name': os.path.basename(image_path), 'extracted_text': text, 'status': 'success' } self.save_result(result) return result else: return { 'file_path': image_path, 'status': 'failed', 'error': '识别失败' } except Exception as e: return { 'file_path': image_path, 'status': 'error', 'error': str(e) } with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_image = { executor.submit(process_single, img_path): img_path for img_path in image_files } for future in concurrent.futures.as_completed(future_to_image): results.append(future.result()) return results ``` ### 4.2 结果后处理与导出 识别后的文字可能需要进一步处理和导出： ```python class OCRResultExporter: @staticmethod def export_to_csv(results, output_file): """导出结果到CSV文件""" import csv with open(output_file, 'w', newline='', encoding='utf-8') as csvfile: fieldnames = ['file_name', 'status', 'text_preview', 'process_time'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for result in results: text_preview = result.get('extracted_text', '')[:100] + '...' if result.get('extracted_text') else '' writer.writerow({ 'file_name': result['file_name'], 'status': result['status'], 'text_preview': text_preview, 'process_time': time.strftime("%Y-%m-%d %H:%M:%S") }) @staticmethod def export_to_json(results, output_file): """导出结果到JSON文件""" import json with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) ``` ### 4.3 错误处理与重试机制 增强的错误处理确保批量处理的稳定性： ```python def robust_extract_text(image_path, server_ip, max_retries=3): """ 带重试机制的文本提取函数 参数: image_path: 图片路径 server_ip: 服务器IP max_retries: 最大重试次数 返回: 识别结果或None """ for attempt in range(max_retries): try: text = extract_text_from_image(image_path, server_ip) if text: return text else: print(f"第{attempt+1}次尝试失败，准备重试...") time.sleep(2) # 等待2秒后重试 except requests.exceptions.ConnectionError: print(f"连接错误，第{attempt+1}次尝试失败") time.sleep(5) # 连接错误等待更久 except Exception as e: print(f"第{attempt+1}次尝试失败: {e}") time.sleep(2) print(f"图片 {os.path.basename(image_path)} 处理失败，已达最大重试次数") return None ``` ## 5. 实战案例与性能优化 ### 5.1 实际应用场景 以下是一些实际的应用场景示例： ```python # 场景1：处理扫描文档目录 def process_scanned_documents(docs_folder, server_ip): processor = BatchOCRProcessor(server_ip, "scanned_docs_output") image_files = processor.get_image_files(docs_folder) results = processor.process_batch(image_files) return results # 场景2：提取照片中的文字 def extract_text_from_photos(photos_folder, server_ip): processor = ConcurrentOCRProcessor(server_ip, "photos_output") image_files = processor.get_image_files(photos_folder) results = processor.process_concurrent(image_files, max_workers=3) return results # 场景3：批量处理表格图片 def process_table_images(tables_folder, server_ip): # 表格识别可能需要特殊处理 processor = BatchOCRProcessor(server_ip, "tables_output") image_files = processor.get_image_files(tables_folder) # 可以在这里添加表格检测和预处理逻辑 results = processor.process_batch(image_files, batch_size=3, delay=2) return results ``` ### 5.2 性能优化建议 根据实际使用经验，以下优化建议可以提高处理效率： 1. **图片预处理**：确保图片分辨率适当（最长边1540像素效果最佳） 2. **批量大小**：根据服务器性能调整批量大小，一般5-10张为宜 3. **并发控制**：避免过多并发请求导致服务器过载 4. **网络优化**：确保客户端与服务器之间的网络连接稳定 5. **错误处理**：实现合理的重试机制和错误日志记录 ### 5.3 资源监控与管理 在处理大量图片时，监控资源使用情况很重要： ```python def monitor_processing(image_files, processor, check_interval=10): """ 监控处理进度和资源使用 参数: image_files: 要处理的文件列表 processor: OCR处理器实例 check_interval: 检查间隔（秒） """ start_time = time.time() processed_count = 0 total_count = len(image_files) # 这里可以添加系统资源监控逻辑 # 例如：内存使用、CPU负载、网络状态等 def print_progress(): elapsed = time.time() - start_time progress = (processed_count / total_count) * 100 print(f"进度: {progress:.1f}% ({processed_count}/{total_count})") print(f"已用时间: {elapsed:.1f}秒") if processed_count > 0: estimated_total = elapsed * total_count / processed_count remaining = estimated_total - elapsed print(f"预计剩余时间: {remaining:.1f}秒") # 在实际处理循环中调用进度监控 # 这里只是示例框架 ``` ## 6. 总结 通过本教程，我们学习了如何使用Python调用LightOnOCR-2-1B的API实现批量图片文字提取。从单张图片处理到批量并发处理，从基础功能到高级优化，我们覆盖了实际应用中的各种场景。 **关键要点回顾**： 1. **环境准备**：确保Python环境和依赖库正确安装 2. **图片预处理**：调整图片大小和格式以提高识别效果 3. **API调用**：正确构建请求和处理响应 4. **批量处理**：使用批处理和并发提高效率 5. **错误处理**：实现重试机制和健壮的错误处理 6. **结果导出**：多种格式保存识别结果 **实用建议**： - 对于大量图片处理，建议使用并发处理但控制并发数 - 定期监控处理进度和系统资源使用情况 - 根据实际需求调整图片预处理参数 - 保存详细的处理日志便于排查问题 LightOnOCR-2-1B作为一个支持多语言的高精度OCR模型，在各种文字提取场景中都能发挥出色表现。通过编程方式批量调用，可以极大提高工作效率，特别适合处理文档数字化、数据提取、内容分析等任务。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。