【多模态处理】利用GPT逐一读取本地图片并生成描述并保存，支持崩溃后从最新进度恢复

代码功能：

读取本地图片文件，并使用GPT模型生成图像的元数据描述。生成的结果会保存到一个JSON文件中。代码还包含了检查点机制，以便在处理过程中程序崩溃时能够从最新的位置继续生成。

核心功能

1. 读取文件并设置变量：
   • 从JSON文件中读取图像路径、宽度和高度等变量。
   • 根据读取的变量设置prompt，调用GPT模型。
2. 调用GPT模型：
   • 使用openai.ChatCompletion.create方法调用GPT模型，生成图像的元数据描述。
   • 将生成的结果保存到JSON文件中。
3. 保存输出到JSON：
   • 每处理一张图片，就将结果追加到JSON文件中。
4. 使用检查点机制：
   • 每处理一张图片后，保存当前处理的位置。
   • 如果处理过程中出现错误，程序可以从上次保存的位置继续处理。
5. 处理本地图片文件：
   • 从本地文件夹读取图片文件，并对每张图片进行处理。

最后碎碎念

提供一个模板，方便大家理解其思想，使用的时候，可以和openai最基本的代码对比着看

代码（使用中转平台url）：

使用中转平台（需要设置中转平台url）：

from PIL import Image
import os
import base64
import openai
import pickle
import json# 设置API密钥和中转平台URL
API_SECRET_KEY = "your_api_secret_key"
BASE_URL = "https://api.your_base_url.com/v1"# 图像文件夹路径
image_directory_path = 'your_image_directory_path'# 设置要处理的图像数量
number_of_images_to_process = 50# 输出文件路径
output_file_path = "output_results.json"# 初始化计数器
image_counter = 0# 读取 JSON 数据文件
data_file = 'your_data_file.json'
with open(data_file, 'r') as f:data = json.load(f)def encode_image(image_path):with open(image_path, "rb") as image_file:return base64.b64encode(image_file.read()).decode('utf-8')def get_image_details(image_path):"""获取图像的详细信息，包括图像ID和尺寸。参数:image_path (str): 图像文件的路径。返回:tuple: 包含图像ID（文件名，不包括扩展名）和图像尺寸（宽度，高度）的元组。示例:get_image_details('path/to/image.jpg')  -> ('image', (800, 600))"""image_filename = os.path.basename(image_path)image_id = os.path.splitext(image_filename)[0]with Image.open(image_path) as img:image_size = img.sizereturn image_id, image_sizedef chat_completions(image_path, width, height):base64_image = encode_image(image_path)image_id, image_size = get_image_details(image_path)client = OpenAI(api_key=API_SECRET_KEY, base_url=BASE_URL)response = openai.ChatCompletion.create(model="gpt-4",messages=[{"role": "system", "content": "You are an assistant that provides metadata information about images."},{"role": "user", "content": f"Image ID: {image_id}, Width: {width}, Height: {height}"}],max_tokens=3000,timeout=999,)return response# 初始化结果字典
results_dict = {}# 检查是否存在检查点文件
checkpoint_file = "checkpoint.pkl"
if os.path.exists(checkpoint_file):with open(checkpoint_file, "rb") as f:start_index = pickle.load(f)
else:start_index = 0# 处理图像文件
for i, image in enumerate(data[start_index:], start=start_index):image_name = image['image_path']image_file = os.path.join(image_directory_path, image_name)image_width = image['width']image_height = image['height']if image_name.lower().endswith(('.png', '.jpg', '.jpeg', '.tiff', '.bmp', '.gif')):try:response = chat_completions(image_file, image_width, image_height)result = {image_name: response.choices[0].message['content']}with open(output_file_path, "a") as output_file:output_file.write(json.dumps(result) + "\n")except Exception as e:print(f"Error processing image {image_name}: {e}")continueimage_counter += 1if image_counter >= number_of_images_to_process:breakwith open(checkpoint_file, "wb") as f:pickle.dump(i+1, f)

代码（直接使用openai的key）

from PIL import Image
import os
import base64
import openai
import pickle
import json# 设置API密钥
API_SECRET_KEY = "your_api_secret_key"# 图像文件夹路径
image_directory_path = 'your_image_directory_path'# 设置要处理的图像数量
number_of_images_to_process = 50# 输出文件路径
output_file_path = "output_results.json"# 初始化计数器
image_counter = 0# 读取 JSON 数据文件
data_file = 'your_data_file.json'
with open(data_file, 'r') as f:data = json.load(f)def encode_image(image_path):with open(image_path, "rb") as image_file:return base64.b64encode(image_file.read()).decode('utf-8')def get_image_details(image_path):"""获取图像的详细信息，包括图像ID和尺寸。参数:image_path (str): 图像文件的路径。返回:tuple: 包含图像ID（文件名，不包括扩展名）和图像尺寸（宽度，高度）的元组。示例:get_image_details('path/to/image.jpg')  -> ('image', (800, 600))"""image_filename = os.path.basename(image_path)image_id = os.path.splitext(image_filename)[0]with Image.open(image_path) as img:image_size = img.sizereturn image_id, image_sizedef chat_completions(image_path, width, height):base64_image = encode_image(image_path)image_id, image_size = get_image_details(image_path)openai.api_key = API_SECRET_KEYresponse = openai.ChatCompletion.create(model="gpt-4",messages=[{"role": "system", "content": "You are an assistant that provides metadata information about images."},{"role": "user", "content": f"Image ID: {image_id}, Width: {width}, Height: {height}"}],max_tokens=3000,timeout=999,)return response# 初始化结果字典
results_dict = {}# 检查是否存在检查点文件
checkpoint_file = "checkpoint.pkl"
if os.path.exists(checkpoint_file):with open(checkpoint_file, "rb") as f:start_index = pickle.load(f)
else:start_index = 0# 处理图像文件
for i, image in enumerate(data[start_index:], start=start_index):image_name = image['image_path']image_file = os.path.join(image_directory_path, image_name)image_width = image['width']image_height = image['height']if image_name.lower().endswith(('.png', '.jpg', '.jpeg', '.tiff', '.bmp', '.gif')):try:response = chat_completions(image_file, image_width, image_height)result = {image_name: response.choices[0].message['content']}with open(output_file_path, "a") as output_file:output_file.write(json.dumps(result) + "\n")except Exception as e:print(f"Error processing image {image_name}: {e}")continueimage_counter += 1if image_counter >= number_of_images_to_process:breakwith open(checkpoint_file, "wb") as f:pickle.dump(i+1, f)

注意

上面的代码，最后四行：（先判断处理图像数量是否大于规定处理图像数量，再保存checkpoint）

if image_counter >= number_of_images_to_process:break
with open(checkpoint_file, "wb") as f:pickle.dump(i+1, f)

有时候要替换逻辑为这样（先保存checkpoint，再判断处理图像数量是否大于规定处理图像数量）

with open(checkpoint_file, "wb") as f:pickle.dump(i+1, f)
if image_counter >= number_of_images_to_process:break

然后，每次程序运行结束时，比如5.jpg处理完，第二次运行程序，不是再处理一遍5.jpg，而是从6.jpg开始。但有的时候不用替换，也仍然从6.jpg开始，不知道为什么。

但确实下方替换后更好一点，因为有的时候break完后直接跳出循环，导致最后一次的i+1没有更新。