当前位置：首页 > news >正文

Qwen2.5-VL-7B-Instruct代码实例：Python调用API实现批量图片问答处理

news 2026/4/14 15:45:18

Qwen2.5-VL-7B-Instruct代码实例：Python调用API实现批量图片问答处理

你是不是经常遇到一堆图片需要处理？比如，电商运营要分析商品主图，内容审核要识别违规图片，或者自媒体小编要给大量图片配文字。一张张看，一张张写，效率低还容易出错。

今天，我就带你用代码来解决这个问题。我们将使用一个强大的多模态模型——Qwen2.5-VL-7B-Instruct，它能“看懂”图片并回答你的问题。更重要的是，我会手把手教你如何用Python写一个脚本，让它自动批量处理成百上千张图片，把人工从重复劳动中解放出来。

读完这篇文章，你将掌握如何用几行代码，搭建一个属于自己的“图片智能分析流水线”。

1. 项目准备：认识你的AI助手

在写代码之前，我们先快速了解一下今天的主角。

Qwen2.5-VL-7B-Instruct是一个视觉-语言模型。简单说，它就像是一个同时具备“眼睛”和“大脑”的AI。它的“眼睛”可以识别图片里的内容（物体、场景、文字），它的“大脑”可以理解你的问题，并结合看到的内容给出回答。

比如，你给它一张街景图，问“图片里有多少辆车？”，它能数出来。你给它一张商品海报，问“这个产品的主要卖点是什么？”，它能根据图片上的文字和视觉元素进行分析。

我们这次要做的，就是通过编程，让这个AI助手为我们批量工作。整个流程的核心是调用模型提供的API接口。你可以把API想象成一个“服务窗口”，我们的Python脚本就是“客户”，我们把图片和问题打包发送给这个窗口，窗口背后的AI模型处理完后，再把答案返回给我们。

为了让你能跟着操作，你需要先确保Qwen2.5-VL-7B-Instruct服务已经在你本地或服务器上运行起来了。通常，运行后可以通过http://localhost:7860这样的地址在网页上访问。我们的代码将通过类似http://localhost:7860/api/chat这样的接口地址与它通信。

2. 环境搭建：安装必要的工具包

工欲善其事，必先利其器。我们只需要一个主要的Python库：requests。它专门用于发送HTTP请求，是我们和模型API沟通的桥梁。

打开你的终端或命令行，用pip安装它：

pip install requests

如果你的网络环境需要配置代理，可以顺便设置一下（根据实际情况调整或忽略）：

# 可选步骤，仅在需要时设置 export http_proxy=http://your_proxy:port export https_proxy=http://your_proxy:port

安装完成后，创建一个新的Python文件，比如命名为batch_image_qa.py，我们就可以开始写代码了。

3. 核心代码解析：从单张图片到批量处理

让我们把任务拆解，从最简单的“问一张图”开始，逐步升级到“问一堆图”。

3.1 基础单元：如何与AI模型对话

首先，我们写一个函数，负责发送单次问答请求。这是所有功能的基础。

import requests import base64 import json from typing import Optional, Dict, Any def ask_image_one_time(image_path: str, question: str, api_url: str = "http://localhost:7860/api/chat") -> Optional[str]: """ 向Qwen2.5-VL模型提问一张图片。 参数: image_path: 图片文件的路径。 question: 你想问的问题。 api_url: 模型API的地址，默认是本地7860端口。 返回: 模型返回的答案文本，如果出错则返回None。 """ # 1. 读取并编码图片 try: with open(image_path, "rb") as img_file: # 将图片二进制数据转换为base64字符串，方便通过网络传输 base64_image = base64.b64encode(img_file.read()).decode('utf-8') except FileNotFoundError: print(f"错误：找不到图片文件 {image_path}") return None except Exception as e: print(f"读取图片 {image_path} 时出错: {e}") return None # 2. 构造请求数据 # 模型期望的格式是一个包含对话历史的列表 payload = { "model": "Qwen2.5-VL-7B-Instruct-GPTQ", # 指定模型名称 "messages": [ { "role": "user", "content": [ {"type": "text", "text": question}, { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} # 这里告诉API，内容是一张base64格式的图片 } ] } ], "stream": False # 我们一次性获取完整回复，不要流式传输 } # 3. 发送请求 headers = {"Content-Type": "application/json"} try: response = requests.post(api_url, json=payload, headers=headers, timeout=60) response.raise_for_status() # 如果状态码不是200，抛出异常 except requests.exceptions.RequestException as e: print(f"请求API时出错: {e}") return None # 4. 解析回复 try: result = response.json() # 通常，答案在 response['choices'][0]['message']['content'] 里 answer = result.get('choices', [{}])[0].get('message', {}).get('content') if not answer: # 有些API格式可能不同，这里尝试另一种常见格式 answer = result.get('response', '') return answer.strip() if answer else "模型未返回有效答案。" except (json.JSONDecodeError, KeyError, IndexError) as e: print(f"解析模型回复时出错: {e}。原始回复: {response.text[:200]}...") return None # 试试这个函数 if __name__ == "__main__": # 假设你有一张名为 `test_cat.jpg` 的猫咪图片 test_answer = ask_image_one_time("test_cat.jpg", "图片里是什么动物？它是什么颜色的？") if test_answer: print("模型回答：", test_answer)

这个函数干了四件事：

读图编码：把你的图片变成一串文本（base64），方便在网络上传输。
打包问题：按照模型能理解的格式，把问题和图片打包成一个数据包（payload）。
发送请求：通过requests.post把这个数据包“寄”给模型的API。
拆包取答案：收到回信后，从里面提取出文字答案。

你可以修改api_url变量，如果模型服务运行在其他机器或端口上，比如http://192.168.1.100:7860/api/chat。

3.2 功能升级：批量处理与结果保存

单次问答有用，但批量处理才是效率的关键。接下来，我们升级脚本，让它能处理一个文件夹里的所有图片，并把结果保存下来，方便查看。

import os import csv from datetime import datetime def batch_process_images(image_folder: str, question: str, output_csv: str = "batch_results.csv", api_url: str = "http://localhost:7860/api/chat"): """ 批量处理一个文件夹内的所有图片，向每张图片提问同一个问题，并保存结果。 参数: image_folder: 存放图片的文件夹路径。 question: 要对每张图片提出的统一问题。 output_csv: 结果保存的CSV文件名。 api_url: 模型API地址。 """ # 支持的图片格式 valid_extensions = ('.png', '.jpg', '.jpeg', '.bmp', '.gif', '.webp') # 获取文件夹内所有图片文件 image_files = [] for file in os.listdir(image_folder): if file.lower().endswith(valid_extensions): image_files.append(os.path.join(image_folder, file)) if not image_files: print(f"在文件夹 {image_folder} 中未找到支持的图片文件。") return print(f"找到 {len(image_files)} 张待处理图片。开始批量处理...") print(f"统一问题：{question}") results = [] for idx, img_path in enumerate(image_files, 1): print(f"正在处理 [{idx}/{len(image_files)}]: {os.path.basename(img_path)}") answer = ask_image_one_time(img_path, question, api_url) if answer is None: answer = "处理失败" print(f" -> 处理失败") else: print(f" -> 完成") # 记录结果 results.append({ "序号": idx, "图片文件名": os.path.basename(img_path), "问题": question, "模型回答": answer, "处理时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) # 将结果保存到CSV文件 try: with open(output_csv, 'w', newline='', encoding='utf-8-sig') as csvfile: # utf-8-sig支持Excel中文 fieldnames = ["序号", "图片文件名", "问题", "模型回答", "处理时间"] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerows(results) print(f"\n批量处理完成！结果已保存至: {output_csv}") except Exception as e: print(f"保存结果到CSV文件时出错: {e}") # 使用批量处理函数 if __name__ == "__main__": # 指定你的图片文件夹路径 my_image_folder = "./product_images" # 指定你的问题 my_question = "这张图片展示的是什么产品？请简要描述其外观和可能的使用场景。" batch_process_images(my_image_folder, my_question, "product_analysis_results.csv")

这个批量处理函数做了几件聪明事：

自动扫描文件夹：你只需要把图片扔进一个文件夹，它自己会找出所有图片。
进度提示：处理时会显示当前进度，让你知道进行到哪了。
结果归档：把所有图片的文件名、问题、答案和处理时间，整整齐齐地记录到一个CSV表格文件里。你可以用Excel或WPS直接打开查看，一目了然。

3.3 高级技巧：为每张图定制不同问题

有时候，我们可能想对不同的图片问不同的问题。比如，A图片问“这是什么品牌？”，B图片问“价格是多少？”。我们可以进一步优化脚本，让它根据一个“问题清单”来工作。

def batch_process_with_question_list(image_question_list: list, output_csv: str = "custom_batch_results.csv", api_url: str = "http://localhost:7860/api/chat"): """ 根据提供的列表，为每张图片处理定制化的问题。 参数: image_question_list: 一个列表，每个元素是字典，格式为： [{"image_path": "路径1", "question": "问题1"}, ...] output_csv: 结果保存的CSV文件名。 api_url: 模型API地址。 """ if not image_question_list: print("问题列表为空。") return print(f"开始处理 {len(image_question_list)} 个定制化任务...") results = [] for idx, task in enumerate(image_question_list, 1): img_path = task.get("image_path") question = task.get("question") if not img_path or not question: print(f"任务 {idx} 数据不完整，已跳过。") continue print(f"正在处理 [{idx}/{len(image_question_list)}]: {os.path.basename(img_path)}") print(f" 问题：{question}") answer = ask_image_one_time(img_path, question, api_url) if answer is None: answer = "处理失败" print(f" -> 处理失败") else: print(f" -> 完成") results.append({ "序号": idx, "图片文件名": os.path.basename(img_path), "问题": question, "模型回答": answer, "处理时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) # 保存结果 try: with open(output_csv, 'w', newline='', encoding='utf-8-sig') as csvfile: fieldnames = ["序号", "图片文件名", "问题", "模型回答", "处理时间"] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerows(results) print(f"\n定制化批量处理完成！结果已保存至: {output_csv}") except Exception as e: print(f"保存结果到CSV文件时出错: {e}") # 使用定制化批量处理 if __name__ == "__main__": # 准备你的任务列表 my_task_list = [ {"image_path": "./ads/ad1.jpg", "question": "这张广告海报的主题是什么？主要推广什么产品？"}, {"image_path": "./ads/ad2.png", "question": "海报上的联系电话或网址是什么？"}, {"image_path": "./screenshots/ui_design.png", "question": "这个界面设计有哪些主要功能区域？"}, ] batch_process_with_question_list(my_task_list, "custom_analysis_results.csv")

这个函数给了你最大的灵活性。你可以提前准备好一个任务列表，脚本就会按顺序执行，非常适合处理复杂、多样的分析需求。

4. 实战演练：几个真实的应用场景

光看代码可能有点抽象，我们来看看这些代码能用在哪些地方。

场景一：电商商品图分析假设你有一个网店，上传了100件新商品的图片。你可以运行脚本，问：“图片中的商品是什么？它的主要颜色和材质是什么？” 脚本运行完毕后，你会得到一个CSV文件，里面是AI对每张商品图的描述。你可以快速检查描述是否准确，或者直接把这些描述用作商品详情页的初稿。

场景二：社交媒体内容审核如果你运营一个社区，需要审核用户上传的图片是否合规。你可以用脚本批量问：“这张图片是否包含暴力、色情或令人不适的内容？” AI会给出初步判断，帮你快速过滤出需要人工重点复核的图片，大大提升审核效率。

场景三：设计素材管理设计师电脑里存了成千上万的图片素材，找起来很麻烦。你可以用脚本为每张图片生成一段文字描述，比如：“这张图片的风格是什么？包含哪些元素？（例如：极简、蓝色、城市、夜晚）”。然后，你可以用文本搜索的方式来查找图片，比如搜索“蓝色夜晚城市”，就能快速找到相关的设计素材。

场景四：教育或研究资料处理在做研究或备课，收集了很多图表、示意图。你可以用脚本批量提问：“这张图表展示了什么数据趋势？” 或者 “这张示意图解释了什么原理？”。AI生成的摘要能帮你快速回顾和整理资料。

5. 让脚本更健壮：错误处理与性能考虑

在实际使用中，网络可能会波动，图片格式可能不支持，API可能暂时无响应。一个健壮的脚本需要处理好这些问题。

网络重试：我们可以在ask_image_one_time函数的请求部分增加重试逻辑。

import time def ask_image_one_time_with_retry(image_path, question, api_url, max_retries=3): for attempt in range(max_retries): answer = ask_image_one_time(image_path, question, api_url) if answer is not None and "处理失败" not in answer: return answer else: wait_time = (attempt + 1) * 2 # 重试等待时间递增 print(f"第{attempt+1}次尝试失败，{wait_time}秒后重试...") time.sleep(wait_time) return "经过多次尝试后处理失败。" # 然后在批量函数中调用这个带重试的函数

控制并发（可选）：如果你有成千上万张图，一张张处理太慢，可以考虑使用多线程或异步IO来并发请求。但要注意，这会给模型服务器带来较大压力，需要根据服务器性能酌情使用。对于初学者，顺序处理更稳定。
日志记录：除了保存结果到CSV，还可以将运行过程中的重要信息（如错误、开始结束时间）写入一个日志文件，方便后期排查问题。

6. 总结

好了，我们来回顾一下今天搭建的“图片智能分析流水线”：

核心原理：我们利用Qwen2.5-VL-7B-Instruct模型的API，通过Python的requests库发送包含图片和问题的请求，并接收文本答案。
核心函数：ask_image_one_time函数是基石，负责单次对话。基于它，我们构建了batch_process_images函数来处理文件夹，以及batch_process_with_question_list函数来处理定制化任务列表。
输出成果：所有处理结果都会清晰、结构化地保存在CSV文件中，便于后续分析和使用。
应用广泛：从电商分析、内容审核到素材管理、研究辅助，这个脚本可以适配多种需要批量理解图片内容的场景。

这个脚本的价值在于将强大的多模态AI能力，封装成了简单、可重复、可批量的自动化工具。你不需要每次都打开网页上传图片、输入问题，只需要运行一次脚本，泡杯咖啡的功夫，结果就整齐地摆在表格里了。

你可以根据实际需求，随意修改和扩展这个脚本。比如，从CSV读取任务，将结果自动导入数据库，或者与你的其他工作流（如自动生成报告）结合起来。代码的世界里，想象力是唯一的限制。