GLM-OCR实战教程：批量图片识别脚本编写（循环调用client.predict）

GLM-OCR实战教程批量图片识别脚本编写循环调用client.predict1. 项目概述与环境准备GLM-OCR是一个强大的多模态OCR识别模型专门处理复杂文档理解任务。它不仅支持常规的文字识别还能准确识别表格结构和数学公式在处理各种文档场景时表现出色。在实际工作中我们经常需要处理大量图片文件比如扫描的文档、拍摄的表格或者截图内容。手动一张张上传识别效率太低这时候就需要一个批量处理的自动化脚本。1.1 环境检查与依赖安装在开始编写脚本前确保你的环境已经正确配置# 检查Python版本 python --version # 应该是Python 3.10.19 # 检查必要的库 pip list | grep -E gradio-client|requests|Pillow如果缺少相关依赖可以使用以下命令安装pip install gradio-client requests Pillow1.2 服务状态确认确保GLM-OCR服务已经正常启动# 检查服务是否运行 ps aux | grep serve_gradio.py # 检查端口是否监听 netstat -tlnp | grep 7860 # 简单的连通性测试 curl -I http://localhost:78602. 基础单张图片识别在编写批量脚本之前我们先从最简单的单张图片识别开始理解基本的调用流程。2.1 基本调用示例from gradio_client import Client import time def single_image_ocr(image_path): 单张图片OCR识别函数 try: # 创建客户端连接 client Client(http://localhost:7860) # 调用预测接口 result client.predict( image_pathimage_path, promptText Recognition:, # 文本识别任务 api_name/predict ) return result except Exception as e: print(f识别失败: {str(e)}) return None # 使用示例 if __name__ __main__: result single_image_ocr(/path/to/your/image.png) if result: print(识别结果:, result)2.2 不同任务类型的调用GLM-OCR支持多种识别任务只需要修改prompt参数def recognize_with_different_tasks(image_path, task_typetext): 支持多种识别任务 task_prompts { text: Text Recognition:, table: Table Recognition:, formula: Formula Recognition: } if task_type not in task_prompts: print(不支持的任务类型使用默认文本识别) task_type text client Client(http://localhost:7860) result client.predict( image_pathimage_path, prompttask_prompts[task_type], api_name/predict ) return result3. 批量图片处理脚本编写现在进入核心部分编写能够循环处理多张图片的批量脚本。3.1 基础批量处理框架import os import glob from gradio_client import Client import json from datetime import datetime class GLMOCRBatchProcessor: def __init__(self, server_urlhttp://localhost:7860): self.client Client(server_url) self.results [] def process_single_image(self, image_path, task_typetext): 处理单张图片 try: prompt_map { text: Text Recognition:, table: Table Recognition:, formula: Formula Recognition: } prompt prompt_map.get(task_type, Text Recognition:) print(f正在处理: {os.path.basename(image_path)}) start_time time.time() result self.client.predict( image_pathimage_path, promptprompt, api_name/predict ) processing_time time.time() - start_time return { filename: os.path.basename(image_path), filepath: image_path, result: result, task_type: task_type, processing_time: round(processing_time, 2), timestamp: datetime.now().isoformat(), status: success } except Exception as e: print(f处理失败 {image_path}: {str(e)}) return { filename: os.path.basename(image_path), filepath: image_path, result: None, error: str(e), timestamp: datetime.now().isoformat(), status: failed }3.2 完整的批量处理循环def process_batch_images(self, image_folder, file_patternsNone, task_typetext, output_fileocr_results.json): 批量处理文件夹中的图片 if file_patterns is None: file_patterns [*.png, *.jpg, *.jpeg, *.webp] # 收集所有图片文件 image_files [] for pattern in file_patterns: image_files.extend(glob.glob(os.path.join(image_folder, pattern))) print(f找到 {len(image_files)} 个图片文件) results [] successful 0 failed 0 for i, image_path in enumerate(image_files, 1): print(f\n处理进度: {i}/{len(image_files)}) result self.process_single_image(image_path, task_type) results.append(result) if result[status] success: successful 1 print(f✓ 成功: {result[filename]} ({result[processing_time]}s)) else: failed 1 print(f✗ 失败: {result[filename]}) # 每处理10张图片保存一次中间结果 if i % 10 0: self._save_results(results, output_file) # 最终保存结果 self._save_results(results, output_file) print(f\n处理完成! 成功: {successful}, 失败: {failed}) return results def _save_results(self, results, output_file): 保存结果到JSON文件 with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2)3.3 使用示例# 使用批量处理类 if __name__ __main__: processor GLMOCRBatchProcessor() # 处理一个文件夹中的所有图片 results processor.process_batch_images( image_folder/path/to/your/images, task_typetext, # 可以是 text, table, formula output_filebatch_ocr_results.json ) print(批量处理完成结果已保存到 batch_ocr_results.json)4. 高级功能与优化4.1 支持多种文件格式def get_supported_image_files(folder_path): 获取文件夹中所有支持的图片文件 supported_extensions [.png, .jpg, .jpeg, .webp, .bmp, .tiff] all_files [] for ext in supported_extensions: all_files.extend(glob.glob(os.path.join(folder_path, f*{ext}))) all_files.extend(glob.glob(os.path.join(folder_path, f*{ext.upper()}))) return sorted(all_files)4.2 并发处理优化对于大量图片可以使用多线程提高处理速度import concurrent.futures def process_batch_concurrently(self, image_paths, task_typetext, max_workers3): 使用线程池并发处理图片 results [] with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 创建任务映射 future_to_image { executor.submit(self.process_single_image, path, task_type): path for path in image_paths } # 收集结果 for future in concurrent.futures.as_completed(future_to_image): image_path future_to_image[future] try: result future.result() results.append(result) print(f完成: {os.path.basename(image_path)}) except Exception as e: print(f处理失败 {image_path}: {str(e)}) results.append({ filename: os.path.basename(image_path), filepath: image_path, status: failed, error: str(e) }) return results4.3 结果后处理与导出def export_to_text(self, results, output_fileocr_results.txt): 将结果导出为纯文本文件 with open(output_file, w, encodingutf-8) as f: for result in results: if result[status] success: f.write(f {result[filename]} \n) f.write(f处理时间: {result[processing_time]}s\n) f.write(f识别结果:\n{result[result]}\n) f.write(\n *50 \n\n) def export_to_csv(self, results, output_fileocr_results.csv): 将结果导出为CSV文件 import csv with open(output_file, w, encodingutf-8, newline) as f: writer csv.writer(f) writer.writerow([文件名, 处理状态, 处理时间(s), 识别结果摘要]) for result in results: if result[status] success: # 截取结果的前100个字符作为摘要 result_preview result[result][:100] ... if len(result[result]) 100 else result[result] writer.writerow([ result[filename], 成功, result[processing_time], result_preview ]) else: writer.writerow([ result[filename], 失败, , result.get(error, 未知错误) ])5. 错误处理与重试机制5.1 增强的错误处理def robust_process_image(self, image_path, task_typetext, max_retries3): 带重试机制的图片处理 for attempt in range(max_retries): try: result self.process_single_image(image_path, task_type) if result[status] success: return result else: print(f第{attempt1}次尝试失败等待重试...) time.sleep(2) # 等待2秒后重试 except Exception as e: print(f第{attempt1}次尝试异常: {str(e)}) if attempt max_retries - 1: return { filename: os.path.basename(image_path), filepath: image_path, status: failed, error: f经过{max_retries}次尝试后失败: {str(e)} } time.sleep(2) return None5.2 连接健康检查def check_server_health(self, timeout10): 检查服务器连接状态 try: response requests.get(http://localhost:7860, timeouttimeout) return response.status_code 200 except: return False def wait_for_server(self, max_wait300, check_interval10): 等待服务器就绪 start_time time.time() print(等待OCR服务启动...) while time.time() - start_time max_wait: if self.check_server_health(): print(服务已就绪!) return True print(服务未就绪等待中...) time.sleep(check_interval) print(等待超时服务可能未正常启动) return False6. 完整实战示例6.1 完整的批量处理脚本#!/usr/bin/env python3 GLM-OCR批量图片识别脚本 支持文本、表格、公式识别自动重试和结果导出 import os import glob import time import json import argparse from datetime import datetime from gradio_client import Client import requests class GLMOCRBatchProcessor: def __init__(self, server_urlhttp://localhost:7860): self.client Client(server_url) self.results [] def process_batch(self, input_path, task_typetext, output_formatjson, max_workers1, retries3): 完整的批量处理流程 # 确定输入是文件还是文件夹 if os.path.isfile(input_path): image_paths [input_path] elif os.path.isdir(input_path): image_paths self.get_supported_image_files(input_path) else: raise ValueError(输入路径必须是文件或文件夹) print(f开始处理 {len(image_paths)} 个文件...) # 处理图片 if max_workers 1: results self.process_batch_concurrently(image_paths, task_type, max_workers) else: results [] for image_path in image_paths: result self.robust_process_image(image_path, task_type, retries) results.append(result) # 导出结果 output_file self.export_results(results, output_format) print(f处理完成! 结果已保存到: {output_file}) return results def export_results(self, results, format_typejson): 根据格式导出结果 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) if format_type json: output_file focr_results_{timestamp}.json with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) elif format_type text: output_file focr_results_{timestamp}.txt self.export_to_text(results, output_file) elif format_type csv: output_file focr_results_{timestamp}.csv self.export_to_csv(results, output_file) return output_file def main(): parser argparse.ArgumentParser(descriptionGLM-OCR批量图片识别工具) parser.add_argument(input, help输入文件或文件夹路径) parser.add_argument(--task, choices[text, table, formula], defaulttext, help识别任务类型) parser.add_argument(--output, choices[json, text, csv], defaultjson, help输出格式) parser.add_argument(--workers, typeint, default1, help并发工作线程数) parser.add_argument(--retries, typeint, default3, help失败重试次数) args parser.parse_args() # 创建处理器并执行 processor GLMOCRBatchProcessor() # 检查服务状态 if not processor.check_server_health(): print(警告: OCR服务可能未启动请先启动服务) if input(是否等待服务启动? (y/n): ).lower() y: processor.wait_for_server() # 执行批量处理 processor.process_batch( input_pathargs.input, task_typeargs.task, output_formatargs.output, max_workersargs.workers, retriesargs.retries ) if __name__ __main__: main()6.2 使用方式示例# 处理单个图片 python batch_ocr.py document.png --task text # 处理整个文件夹 python batch_ocr.py ./scanned_docs/ --task table --output csv # 使用多线程处理 python batch_ocr.py ./images/ --workers 4 --retries 2 # 识别公式 python batch_ocr.py math_images/ --task formula --output text7. 总结与最佳实践通过本教程我们学习了如何使用GLM-OCR的Python API编写批量图片识别脚本。关键要点包括7.1 核心知识点总结基础调用使用gradio_client库连接GLM-OCR服务批量处理通过循环调用client.predict实现多图片处理错误处理添加重试机制和健壮的错误处理结果导出支持JSON、文本、CSV多种输出格式性能优化使用多线程提高处理效率7.2 实用建议合理设置并发数根据服务器性能调整工作线程数量避免过度负载定期保存结果处理大量文件时定期保存中间结果防止数据丢失监控服务状态批量处理前检查服务健康状态处理中监控资源使用结果验证对重要文档建议抽样检查识别结果准确性文件管理处理好文件路径问题特别是相对路径和绝对路径的使用7.3 扩展思路这个基础脚本还可以进一步扩展添加进度条显示支持更多输出格式如PDF、Word集成图像预处理功能添加结果后处理和质量评估支持分布式处理现在你已经掌握了GLM-OCR批量处理的核心技能可以高效地处理大量图片识别任务了。根据实际需求调整和优化脚本让它更好地为你服务。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。