# Janus-Pro-7B代码实例：Python调用Ollama API实现批量图片问答自动化 ## 1. 引言：让AI看懂你的图片世界 想象一下这样的场景：你手头有几百张产品图片，需要快速了解每张图片的内容、识别其中的关键元素、甚至生成详细的产品描述。传统方法需要人工一张张查看和描述，既耗时又容易出错。而现在，借助Janus-Pro-7B多模态模型，我们可以用Python代码实现批量图片的自动化问答处理。 Janus-Pro-7B是一个创新的多模态框架，它不仅能理解图片内容，还能生成自然语言回应。通过将视觉编码解耦为独立路径，同时使用统一的变压器架构处理，这个模型在理解和生成任务上都表现出色。简单来说，它既能"看懂"图片，又能"说出"图片的内容。 本文将带你一步步学习如何使用Python调用Ollama API，构建一个高效的批量图片问答自动化系统。无论你是需要处理电商产品图、监控截图还是设计素材，这个方案都能帮你节省大量时间和精力。 ## 2. 环境准备与Ollama部署 ### 2.1 安装必要依赖 在开始之前，我们需要准备Python环境和必要的库。建议使用Python 3.8或更高版本： ```bash pip install requests pillow python-dotenv tqdm ``` 这些库的作用分别是： - `requests`：用于发送HTTP请求到Ollama API - `pillow`：用于处理图片文件 - `python-dotenv`：用于管理环境变量 - `tqdm`：用于显示进度条 ### 2.2 部署Janus-Pro-7B模型 首先确保你已经通过Ollama部署了Janus-Pro-7B模型。在Ollama界面中： 1. 找到模型选择入口 2. 选择【Janus-Pro-7B:latest】模型 3. 模型加载完成后即可通过API进行调用 Ollama默认会在本地11434端口启动服务，我们可以通过http://localhost:11434访问API。 ## 3. 基础API调用方法 ### 3.1 单张图片问答示例 让我们先从最简单的单张图片问答开始，了解基本的API调用方式： ```python import requests import base64 from PIL import Image import io def analyze_single_image(image_path, question): """ 分析单张图片并回答问题 """ # 读取图片并转换为base64 with open(image_path, "rb") as image_file: image_data = base64.b64encode(image_file.read()).decode('utf-8') # 准备请求数据 payload = { "model": "janus-pro-7b:latest", "prompt": question, "images": [image_data], "stream": False } # 发送请求 response = requests.post( "http://localhost:11434/api/generate", json=payload, timeout=60 ) if response.status_code == 200: result = response.json() return result["response"] else: raise Exception(f"API请求失败: {response.status_code}") # 使用示例 if __name__ == "__main__": result = analyze_single_image("product.jpg", "这张图片中的产品是什么？描述它的主要特征。") print("分析结果:", result) ``` 这个基础示例展示了如何将图片转换为base64格式，然后通过API发送给Janus-Pro-7B模型进行分析。 ### 3.2 处理不同类型的图片问题 Janus-Pro-7B可以处理各种类型的图片相关问题，以下是一些常见的提问方式： ```python # 不同的提问模板 question_templates = { "object_detection": "识别图片中的主要物体是什么？", "description": "详细描述这张图片的内容和场景。", "count": "图片中有多少个{object}？", "color": "描述图片中主要物体的颜色。", "scene": "这是什么场景？在什么环境下拍摄的？", "text_extraction": "提取图片中的文字内容。" } def get_custom_question(question_type, **kwargs): """生成定制化的问题""" template = question_templates.get(question_type, "描述这张图片的内容。") return template.format(**kwargs) ``` ## 4. 批量图片处理系统 ### 4.1 完整的批量处理类 现在我们来构建一个完整的批量图片处理系统，支持并发处理和进度跟踪： ```python import os import glob import json import concurrent.futures from tqdm import tqdm from datetime import datetime class BatchImageAnalyzer: def __init__(self, model_name="janus-pro-7b:latest", api_url="http://localhost:11434/api/generate"): self.model_name = model_name self.api_url = api_url self.results = [] def process_image(self, image_path, question): """处理单张图片""" try: # 读取和编码图片 with open(image_path, "rb") as f: image_data = base64.b64encode(f.read()).decode('utf-8') # 准备请求 payload = { "model": self.model_name, "prompt": question, "images": [image_data], "stream": False } # 发送请求 response = requests.post(self.api_url, json=payload, timeout=120) response.raise_for_status() result_data = response.json() return { "image_path": image_path, "question": question, "answer": result_data["response"], "timestamp": datetime.now().isoformat(), "success": True } except Exception as e: return { "image_path": image_path, "question": question, "error": str(e), "timestamp": datetime.now().isoformat(), "success": False } def process_batch(self, image_folder, question, max_workers=4, file_pattern="*.jpg"): """ 批量处理图片文件夹 """ # 获取所有图片文件 image_pattern = os.path.join(image_folder, file_pattern) image_files = glob.glob(image_pattern) if not image_files: print("未找到匹配的图片文件") return [] print(f"找到 {len(image_files)} 张图片，开始处理...") # 使用线程池并发处理 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务 future_to_image = { executor.submit(self.process_image, img_path, question): img_path for img_path in image_files } # 使用进度条跟踪处理进度 results = [] with tqdm(total=len(image_files), desc="处理图片") as pbar: for future in concurrent.futures.as_completed(future_to_image): result = future.result() results.append(result) pbar.update(1) if result["success"]: pbar.set_postfix_str(" 成功") else: pbar.set_postfix_str(" 失败") self.results = results return results def save_results(self, output_file="results.json"): """保存处理结果到JSON文件""" with open(output_file, 'w', encoding='utf-8') as f: json.dump(self.results, f, ensure_ascii=False, indent=2) print(f"结果已保存到 {output_file}") def export_to_csv(self, output_file="results.csv"): """导出结果到CSV文件""" import csv successful_results = [r for r in self.results if r["success"]] with open(output_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(["图片路径", "问题", "回答", "处理时间"]) for result in successful_results: writer.writerow([ result["image_path"], result["question"], result["answer"], result["timestamp"] ]) print(f"CSV结果已导出到 {output_file}") ``` ### 4.2 使用示例 ```python # 使用批量处理系统 if __name__ == "__main__": # 初始化分析器 analyzer = BatchImageAnalyzer() # 处理整个文件夹的图片 results = analyzer.process_batch( image_folder="./product_images", question="描述图片中的产品特点和用途", max_workers=3, # 并发数，根据电脑性能调整 file_pattern="*.jpg" # 支持 *.jpg, *.png, *.jpeg 等 ) # 保存结果 analyzer.save_results("product_analysis.json") analyzer.export_to_csv("product_analysis.csv") # 打印统计信息 success_count = sum(1 for r in results if r["success"]) print(f"\n处理完成！成功: {success_count}/{len(results)}") ``` ## 5. 高级功能与优化 ### 5.1 支持多种图片格式 为了让系统更加通用，我们添加对多种图片格式的支持： ```python def get_supported_image_patterns(): """返回支持的图片格式模式""" return ["*.jpg", "*.jpeg", "*.png", "*.bmp", "*.tiff", "*.webp"] def find_all_images(folder_path): """查找文件夹中的所有图片文件""" patterns = get_supported_image_patterns() image_files = [] for pattern in patterns: image_files.extend(glob.glob(os.path.join(folder_path, pattern))) return sorted(image_files) # 按文件名排序 ``` ### 5.2 智能重试机制 网络请求可能会失败，添加重试机制提高稳定性： ```python def process_image_with_retry(self, image_path, question, max_retries=3): """带重试机制的图片处理""" for attempt in range(max_retries): try: return self.process_image(image_path, question) except requests.exceptions.RequestException as e: if attempt == max_retries - 1: return { "image_path": image_path, "question": question, "error": f"请求失败 after {max_retries} 次重试: {str(e)}", "timestamp": datetime.now().isoformat(), "success": False } print(f"第 {attempt + 1} 次尝试失败，正在重试...") time.sleep(2) # 等待2秒后重试 ``` ### 5.3 结果分析与统计 添加结果分析功能，生成处理报告： ```python def generate_report(self): """生成处理报告""" successful = [r for r in self.results if r["success"]] failed = [r for r in self.results if not r["success"]] report = { "total_images": len(self.results), "successful": len(successful), "failed": len(failed), "success_rate": len(successful) / len(self.results) * 100 if self.results else 0, "processing_time": { "start": min([r["timestamp"] for r in self.results]) if self.results else None, "end": max([r["timestamp"] for r in self.results]) if self.results else None }, "common_questions": set(r["question"] for r in successful), "error_summary": {} } # 错误统计 for error in failed: error_msg = error["error"] report["error_summary"][error_msg] = report["error_summary"].get(error_msg, 0) + 1 return report ``` ## 6. 实际应用案例 ### 6.1 电商产品图片分析 假设你有一个电商网站，需要为上千个产品图片生成自动描述： ```python # 电商产品分析专用配置 ecommerce_analyzer = BatchImageAnalyzer() # 针对不同产品类型使用不同的问题 product_questions = { "clothing": "描述这件服装的款式、颜色、适合的场合", "electronics": "这是什么电子产品？描述它的功能和特点", "furniture": "这是什么家具？描述它的材质、风格和用途", "default": "详细描述图片中的产品特点和用途" } def analyze_ecommerce_images(folder_path, product_type="default"): """分析电商产品图片""" question = product_questions.get(product_type, product_questions["default"]) results = ecommerce_analyzer.process_batch( folder_path, question, max_workers=4 ) # 生成产品描述文件 for result in results: if result["success"]: image_name = os.path.basename(result["image_path"]) product_id = image_name.split('.')[0] # 假设文件名是产品ID # 保存为单独的描述文件 desc_file = os.path.join(folder_path, f"{product_id}_description.txt") with open(desc_file, 'w', encoding='utf-8') as f: f.write(f"产品ID: {product_id}\n") f.write(f"生成时间: {result['timestamp']}\n") f.write(f"AI描述: {result['answer']}\n") return results ``` ### 6.2 监控图片分析 对于安防监控图片，可以分析异常情况： ```python def analyze_security_images(image_folder): """分析监控图片，检测异常情况""" question = "图片中是否有异常情况或可疑人物？描述任何不寻常的细节" results = analyzer.process_batch(image_folder, question) # 筛选出报告异常的图片 alerts = [] for result in results: if result["success"] and ("是" in result["answer"] or "异常" in result["answer"]): alerts.append({ "image_path": result["image_path"], "analysis": result["answer"], "timestamp": result["timestamp"] }) print(f"发现 {len(alerts)} 个异常情况需要关注") return alerts ``` ## 7. 性能优化建议 ### 7.1 调整并发数量 根据你的硬件配置调整并发数量，避免过度占用资源： ```python def get_optimal_workers(): """根据CPU核心数计算最优并发数""" import multiprocessing cpu_count = multiprocessing.cpu_count() return max(1, min(cpu_count - 1, 4)) # 保守估计，留出系统资源 ``` ### 7.2 图片预处理优化 对大图片进行预处理，减少传输数据量： ```python def optimize_image(image_path, max_size=1024): """优化图片大小，减少传输数据量""" from PIL import Image with Image.open(image_path) as img: # 保持宽高比调整大小 img.thumbnail((max_size, max_size)) # 保存为优化的临时文件 temp_path = f"/tmp/optimized_{os.path.basename(image_path)}" img.save(temp_path, format='JPEG', optimize=True, quality=85) return temp_path # 在process_image方法中使用优化后的图片 optimized_path = optimize_image(image_path) # 使用optimized_path而不是原始路径 ``` ## 8. 总结 通过本文的教程，我们构建了一个完整的批量图片问答自动化系统。这个系统利用Janus-Pro-7B多模态模型的能力，可以自动分析大量图片并回答相关问题。 **关键收获**： 1. **基础API调用**：学会了如何通过Python调用Ollama API进行图片问答 2. **批量处理能力**：构建了支持并发处理的批量图片分析系统 3. **实用功能**：添加了重试机制、进度显示、结果导出等实用功能 4. **应用场景**：了解了如何在电商、安防等实际场景中应用这个系统 **下一步建议**： - 尝试不同的提问方式，获得更精准的分析结果 - 根据具体需求定制问题模板 - 考虑将系统集成到现有的工作流程中 - 定期更新Ollama和模型版本，获得更好的性能 这个自动化系统可以显著提高处理图片相关任务的效率，让你从重复性的图片分析工作中解放出来，专注于更重要的决策和创新工作。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。