Qwen3.5-35B-A3B-AWQ-4bit实战教程：API接口封装+Python调用完整示例
你是不是也遇到过这样的问题：好不容易部署了一个强大的多模态AI模型，比如Qwen3.5-35B-A3B-AWQ-4bit，它看图说话的能力确实惊艳，但每次都要打开网页、上传图片、输入问题，才能得到结果。想把它集成到自己的应用里，却发现无从下手？
别担心，今天我就带你彻底解决这个问题。我将手把手教你如何为这个强大的图文对话模型封装一个简洁的API接口，并用Python轻松调用。学完这篇教程，你就能在自己的代码里直接调用这个“看图说话”的AI能力了。
1. 为什么需要API封装？
你可能已经体验过Qwen3.5-35B-A3B-AWQ-4bit的Web界面，上传一张图片，问几个问题，它就能给出相当不错的回答。但如果你想把这种能力集成到自己的项目中，比如：
- 开发一个智能客服系统，自动分析用户上传的产品图片
- 创建一个内容审核工具，识别图片中的敏感信息
- 搭建一个教育应用，帮助学生理解复杂的图表
- 做一个电商工具，自动生成商品图片的描述
这时候，每次都打开网页手动操作就不现实了。我们需要的是程序化的调用方式——这就是API的价值所在。
API封装能给你带来三个核心好处：
接下来，我会从环境准备开始，一步步带你完成API的封装和调用。即使你之前没接触过API开发，也能跟着做出来。
2. 环境准备与快速部署
在开始封装API之前，我们需要确保Qwen3.5-35B-A3B-AWQ-4bit服务已经正常运行。如果你还没有部署，可以按照以下步骤快速启动。
2.1 检查服务状态
首先，通过SSH连接到你的服务器，检查服务是否已经启动：
# 查看后端服务状态
supervisorctl status qwen35awq-backend
# 查看Web服务状态
supervisorctl status qwen35awq-web
# 预期输出应该是 RUNNING 状态
# qwen35awq-backend RUNNING pid 1234, uptime 1:23:45
# qwen35awq-web RUNNING pid 1235, uptime 1:23:45
如果服务没有运行，可以使用以下命令启动：
# 启动服务
supervisorctl start qwen35awq-backend
supervisorctl start qwen35awq-web
2.2 验证Web服务可访问
服务启动后，验证Web界面是否可以正常访问。如果你有公网地址，直接在浏览器打开即可。如果没有，可以通过SSH隧道访问：
# 建立SSH隧道（端口和地址根据你的实际情况调整）
ssh -L 7860:127.0.0.1:7860 -p 32468 root@gpu-kktv84d3pq.ssh.gpu.csdn.net
然后在本地浏览器打开 http://127.0.0.1:7860，应该能看到图文对话的界面。
2.3 了解服务架构
在开始封装API之前，先简单了解一下当前服务的架构：
- 后端服务：运行在8000端口，使用vLLM + compressed-tensors提供模型推理能力
- Web服务：运行在7860端口，提供用户交互界面
- 通信方式：Web界面通过HTTP请求与后端服务交互
我们的目标就是直接与后端服务（8000端口）通信，绕过Web界面，实现程序化调用。
3. API接口封装实战
现在进入核心部分——为Qwen3.5-35B-A3B-AWQ-4bit封装一个简洁的Python API。我会提供一个完整的、可直接使用的封装类。
3.1 创建API封装类
首先，创建一个Python文件，比如 qwen_multimodal_api.py，然后添加以下代码：
import requests
import base64
import json
from typing import Optional, Dict, Any
from pathlib import Path
import mimetypes
class QwenMultimodalAPI:
"""
Qwen3.5-35B-A3B-AWQ-4bit 多模态模型API封装类
支持图片理解、图文问答、视觉描述等功能
"""
def __init__(self, base_url: str = "http://localhost:8000"):
"""
初始化API客户端
Args:
base_url: 后端服务地址，默认为本地8000端口
"""
self.base_url = base_url.rstrip('/')
self.chat_url = f"{self.base_url}/v1/chat/completions"
# 设置请求头
self.headers = {
"Content-Type": "application/json",
"User-Agent": "QwenMultimodalAPI/1.0"
}
def _image_to_base64(self, image_path: str) -> str:
"""
将图片文件转换为base64编码
Args:
image_path: 图片文件路径
Returns:
base64编码的图片字符串
"""
with open(image_path, "rb") as image_file:
image_data = image_file.read()
base64_encoded = base64.b64encode(image_data).decode('utf-8')
# 获取图片MIME类型
mime_type, _ = mimetypes.guess_type(image_path)
if mime_type is None:
mime_type = "image/jpeg" # 默认类型
return f"data:{mime_type};base64,{base64_encoded}"
def ask_image(self,
image_path: str,
question: str,
max_tokens: int = 1024,
temperature: float = 0.7,
stream: bool = False) -> Dict[str, Any]:
"""
向图片提问
Args:
image_path: 图片文件路径
question: 问题文本
max_tokens: 最大生成token数
temperature: 温度参数，控制随机性
stream: 是否使用流式响应
Returns:
包含模型响应的字典
"""
# 将图片转换为base64
image_base64 = self._image_to_base64(image_path)
# 构建请求数据
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{"type": "image_url", "image_url": {"url": image_base64}}
]
}
]
payload = {
"model": "Qwen/Qwen2.5-VL-7B-Instruct", # 模型名称，根据实际部署调整
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": stream
}
try:
response = requests.post(
self.chat_url,
headers=self.headers,
json=payload,
timeout=300 # 5分钟超时，处理大图片可能需要较长时间
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"响应状态码: {e.response.status_code}")
print(f"响应内容: {e.response.text}")
raise
def ask_image_simple(self, image_path: str, question: str) -> str:
"""
简化版的图片提问方法，直接返回回答文本
Args:
image_path: 图片文件路径
question: 问题文本
Returns:
模型的回答文本
"""
result = self.ask_image(image_path, question)
# 提取回答文本
if "choices" in result and len(result["choices"]) > 0:
return result["choices"][0]["message"]["content"]
else:
raise ValueError("无法从响应中提取回答内容")
def batch_ask_images(self,
image_questions: list,
max_tokens: int = 1024,
temperature: float = 0.7) -> list:
"""
批量处理多张图片和问题
Args:
image_questions: 列表，每个元素是 (image_path, question) 元组
max_tokens: 最大生成token数
temperature: 温度参数
Returns:
包含所有回答的列表
"""
results = []
for image_path, question in image_questions:
try:
answer = self.ask_image_simple(image_path, question)
results.append({
"image": image_path,
"question": question,
"answer": answer,
"status": "success"
})
except Exception as e:
results.append({
"image": image_path,
"question": question,
"answer": str(e),
"status": "error"
})
return results
这个封装类提供了三个核心方法：
3.2 添加错误处理和重试机制
在实际使用中，网络波动或服务暂时不可用是常见情况。让我们增强一下错误处理：
import time
from typing import Optional
class QwenMultimodalAPIEnhanced(QwenMultimodalAPI):
"""
增强版的API封装类，添加重试机制和更多实用功能
"""
def __init__(self,
base_url: str = "http://localhost:8000",
max_retries: int = 3,
retry_delay: float = 2.0):
"""
初始化增强版API客户端
Args:
base_url: 后端服务地址
max_retries: 最大重试次数
retry_delay: 重试延迟（秒）
"""
super().__init__(base_url)
self.max_retries = max_retries
self.retry_delay = retry_delay
def ask_image_with_retry(self,
image_path: str,
question: str,
max_tokens: int = 1024,
temperature: float = 0.7) -> Dict[str, Any]:
"""
带重试机制的图片提问
Args:
image_path: 图片文件路径
question: 问题文本
max_tokens: 最大生成token数
temperature: 温度参数
Returns:
包含模型响应的字典
"""
last_exception = None
for attempt in range(self.max_retries):
try:
return self.ask_image(image_path, question, max_tokens, temperature)
except requests.exceptions.ConnectionError as e:
last_exception = e
print(f"连接失败，第 {attempt + 1} 次重试…")
if attempt < self.max_retries – 1:
time.sleep(self.retry_delay * (attempt + 1)) # 指数退避
continue
except requests.exceptions.Timeout as e:
last_exception = e
print(f"请求超时，第 {attempt + 1} 次重试…")
if attempt < self.max_retries – 1:
time.sleep(self.retry_delay)
continue
except Exception as e:
# 其他错误不重试
raise e
# 所有重试都失败
raise last_exception or Exception("所有重试尝试都失败了")
def get_model_info(self) -> Optional[Dict[str, Any]]:
"""
获取模型信息
Returns:
模型信息字典，如果失败则返回None
"""
try:
response = requests.get(
f"{self.base_url}/v1/models",
headers=self.headers,
timeout=10
)
response.raise_for_status()
return response.json()
except:
return None
def check_health(self) -> bool:
"""
检查服务健康状态
Returns:
服务是否健康
"""
try:
info = self.get_model_info()
return info is not None
except:
return False
3.3 添加异步支持（可选）
如果你的应用需要高并发处理，可以添加异步支持：
import aiohttp
import asyncio
from typing import List, Tuple
class AsyncQwenMultimodalAPI:
"""
异步版本的API封装类
"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url.rstrip('/')
self.chat_url = f"{self.base_url}/v1/chat/completions"
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def ask_image_async(self,
image_path: str,
question: str,
max_tokens: int = 1024,
temperature: float = 0.7) -> Dict[str, Any]:
"""
异步向图片提问
Args:
image_path: 图片文件路径
question: 问题文本
max_tokens: 最大生成token数
temperature: 温度参数
Returns:
包含模型响应的字典
"""
# 同步方法转换图片为base64
image_base64 = self._image_to_base64_sync(image_path)
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{"type": "image_url", "image_url": {"url": image_base64}}
]
}
]
payload = {
"model": "Qwen/Qwen2.5-VL-7B-Instruct",
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": False
}
async with self.session.post(
self.chat_url,
json=payload,
timeout=aiohttp.ClientTimeout(total=300)
) as response:
response.raise_for_status()
return await response.json()
def _image_to_base64_sync(self, image_path: str) -> str:
"""同步的图片转base64方法，供异步方法调用"""
with open(image_path, "rb") as image_file:
image_data = image_file.read()
base64_encoded = base64.b64encode(image_data).decode('utf-8')
mime_type, _ = mimetypes.guess_type(image_path)
if mime_type is None:
mime_type = "image/jpeg"
return f"data:{mime_type};base64,{base64_encoded}"
async def batch_ask_images_async(self,
image_questions: List[Tuple[str, str]],
max_tokens: int = 1024,
temperature: float = 0.7) -> List[Dict]:
"""
异步批量处理图片
Args:
image_questions: 图片和问题列表
max_tokens: 最大生成token数
temperature: 温度参数
Returns:
处理结果列表
"""
tasks = []
for image_path, question in image_questions:
task = self.ask_image_async(image_path, question, max_tokens, temperature)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
processed_results = []
for (image_path, question), result in zip(image_questions, results):
if isinstance(result, Exception):
processed_results.append({
"image": image_path,
"question": question,
"answer": str(result),
"status": "error"
})
else:
answer = result["choices"][0]["message"]["content"]
processed_results.append({
"image": image_path,
"question": question,
"answer": answer,
"status": "success"
})
return processed_results
4. Python调用完整示例
现在，让我们看看如何在实际项目中使用这个API封装。我会提供几个完整的示例，覆盖不同的使用场景。
4.1 基础使用示例
首先，创建一个简单的测试脚本 test_basic.py：
#!/usr/bin/env python3
"""
Qwen3.5-35B-A3B-AWQ-4bit API基础使用示例
"""
from qwen_multimodal_api import QwenMultimodalAPIEnhanced
import os
def test_basic_usage():
"""基础使用示例"""
# 初始化API客户端
# 如果你的服务运行在其他地址，修改这里的base_url
api = QwenMultimodalAPIEnhanced(base_url="http://localhost:8000")
# 检查服务状态
print("检查服务状态…")
if api.check_health():
print("✅ 服务运行正常")
else:
print("❌ 服务不可用，请检查服务是否启动")
return
# 准备测试图片和问题
# 这里假设有一张测试图片，你可以替换成自己的图片路径
test_image = "test_image.jpg"
# 如果测试图片不存在，创建一个简单的示例
if not os.path.exists(test_image):
print(f"⚠️ 测试图片 {test_image} 不存在")
print("请准备一张测试图片，或修改代码中的图片路径")
return
# 测试问题列表
test_questions = [
"描述这张图片的内容",
"图片中有哪些物体？",
"图片中的人在做什么？",
"这张图片是什么场景？"
]
# 逐个提问
for i, question in enumerate(test_questions, 1):
print(f"\\n{'='*50}")
print(f"问题 {i}: {question}")
print(f"{'='*50}")
try:
# 调用API
answer = api.ask_image_simple(test_image, question)
print(f"回答: {answer}")
except Exception as e:
print(f"❌ 请求失败: {e}")
def test_single_image_analysis():
"""单张图片深度分析示例"""
api = QwenMultimodalAPIEnhanced()
# 使用带重试的方法
image_path = "product_photo.jpg"
if not os.path.exists(image_path):
print(f"图片 {image_path} 不存在")
return
# 构建一个多轮对话
questions = [
"请详细描述这张图片",
"图片中的产品有什么特点？",
"这个产品适合什么人群使用？",
"从营销角度，你会如何描述这个产品？"
]
print(f"分析图片: {image_path}")
print(f"{'='*60}")
for question in questions:
print(f"\\nQ: {question}")
try:
# 使用带重试的方法
result = api.ask_image_with_retry(image_path, question)
answer = result["choices"][0]["message"]["content"]
print(f"A: {answer}")
print(f"{'-'*40}")
except Exception as e:
print(f"❌ 错误: {e}")
break
if __name__ == "__main__":
print("Qwen多模态API测试")
print("=" * 60)
# 运行基础测试
test_basic_usage()
print("\\n" + "=" * 60)
print("单张图片深度分析测试")
print("=" * 60)
# 运行深度分析测试
test_single_image_analysis()
4.2 批量处理示例
如果你需要处理大量图片，可以使用批量处理方法。创建 test_batch.py：
#!/usr/bin/env python3
"""
批量处理图片示例
"""
from qwen_multimodal_api import QwenMultimodalAPIEnhanced
import os
import json
from datetime import datetime
def batch_process_images():
"""批量处理多张图片"""
api = QwenMultimodalAPIEnhanced()
# 准备图片和问题列表
# 这里假设有一个图片目录
image_dir = "images"
if not os.path.exists(image_dir):
print(f"图片目录 {image_dir} 不存在")
print("创建示例目录和文件…")
os.makedirs(image_dir, exist_ok=True)
# 这里可以添加创建示例图片的代码
print("请在实际使用前准备好图片文件")
return
# 获取所有图片文件
image_files = []
for ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
image_files.extend([f for f in os.listdir(image_dir) if f.lower().endswith(ext)])
if not image_files:
print("没有找到图片文件")
return
print(f"找到 {len(image_files)} 张图片")
# 为每张图片准备问题
image_questions = []
for image_file in image_files[:5]: # 限制前5张，避免处理时间过长
image_path = os.path.join(image_dir, image_file)
# 根据图片文件名生成问题（实际使用中可以根据需要定制）
base_name = os.path.splitext(image_file)[0]
questions = [
f"描述图片内容",
f"这张图片可能是什么场景？",
f"图片中有哪些主要元素？"
]
for question in questions:
image_questions.append((image_path, question))
print(f"准备处理 {len(image_questions)} 个问题")
# 批量处理
print("开始批量处理…")
start_time = datetime.now()
results = api.batch_ask_images(image_questions)
end_time = datetime.now()
processing_time = (end_time – start_time).total_seconds()
# 统计结果
success_count = sum(1 for r in results if r["status"] == "success")
error_count = len(results) – success_count
print(f"\\n处理完成!")
print(f"总耗时: {processing_time:.2f}秒")
print(f"平均每个问题: {processing_time/len(results):.2f}秒")
print(f"成功: {success_count}, 失败: {error_count}")
# 保存结果到文件
output_file = f"batch_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"结果已保存到: {output_file}")
# 显示部分结果
print("\\n前3个结果示例:")
for i, result in enumerate(results[:3]):
print(f"\\n[{i+1}] 图片: {os.path.basename(result['image'])}")
print(f" 问题: {result['question']}")
print(f" 回答: {result['answer'][:100]}…") # 只显示前100字符
print(f" 状态: {result['status']}")
def process_with_custom_questions():
"""使用自定义问题处理图片"""
api = QwenMultimodalAPIEnhanced()
# 定义具体的处理任务
tasks = [
{
"image": "product1.jpg",
"questions": [
"这是什么产品？",
"产品的颜色是什么？",
"适合什么场合使用？"
]
},
{
"image": "scene1.jpg",
"questions": [
"这是什么地方？",
"天气怎么样？",
"图片中有多少人？"
]
}
]
all_results = []
for task in tasks:
image_path = task["image"]
if not os.path.exists(image_path):
print(f"图片不存在: {image_path}")
continue
print(f"\\n处理图片: {image_path}")
for question in task["questions"]:
print(f" 问题: {question}")
try:
answer = api.ask_image_simple(image_path, question)
print(f" 回答: {answer[:80]}…") # 显示前80字符
all_results.append({
"image": image_path,
"question": question,
"answer": answer
})
except Exception as e:
print(f" ❌ 错误: {e}")
all_results.append({
"image": image_path,
"question": question,
"answer": f"错误: {str(e)}",
"status": "error"
})
# 保存结果
if all_results:
output_file = "custom_questions_results.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(all_results, f, ensure_ascii=False, indent=2)
print(f"\\n结果已保存到: {output_file}")
if __name__ == "__main__":
print("批量处理示例")
print("=" * 60)
# 运行批量处理
batch_process_images()
print("\\n" + "=" * 60)
print("自定义问题处理示例")
print("=" * 60)
# 运行自定义问题处理
process_with_custom_questions()
4.3 异步处理示例（高性能场景）
如果你的应用需要高并发处理大量图片，可以使用异步版本。创建 test_async.py：
#!/usr/bin/env python3
"""
异步处理示例 – 适合高并发场景
"""
import asyncio
import aiofiles
from qwen_multimodal_api import AsyncQwenMultimodalAPI
import os
import json
async def async_batch_process():
"""异步批量处理"""
# 使用异步上下文管理器
async with AsyncQwenMultimodalAPI() as api:
# 准备测试数据
image_questions = [
("image1.jpg", "描述这张图片"),
("image2.jpg", "图片中有哪些颜色"),
("image3.jpg", "这是什么场景"),
# 可以添加更多…
]
# 过滤掉不存在的图片
valid_questions = []
for img_path, question in image_questions:
if os.path.exists(img_path):
valid_questions.append((img_path, question))
else:
print(f"图片不存在: {img_path}")
if not valid_questions:
print("没有有效的图片可处理")
return
print(f"开始异步处理 {len(valid_questions)} 个任务…")
# 批量处理
results = await api.batch_ask_images_async(valid_questions)
# 输出结果
print(f"\\n处理完成，共 {len(results)} 个结果:")
for i, result in enumerate(results):
status_icon = "✅" if result["status"] == "success" else "❌"
print(f"{status_icon} [{i+1}] {os.path.basename(result['image'])}: {result['question']}")
if result["status"] == "success":
print(f" 回答: {result['answer'][:50]}…")
# 保存结果
output_file = "async_results.json"
async with aiofiles.open(output_file, 'w', encoding='utf-8') as f:
await f.write(json.dumps(results, ensure_ascii=False, indent=2))
print(f"\\n结果已保存到: {output_file}")
async def concurrent_processing():
"""并发处理示例"""
async with AsyncQwenMultimodalAPI() as api:
# 创建多个并发任务
tasks = []
# 任务1：分析产品图片
tasks.append(
api.ask_image_async("product.jpg", "这是什么产品？有什么特点？")
)
# 任务2：分析场景图片
tasks.append(
api.ask_image_async("scene.jpg", "描述这个场景，天气如何？")
)
# 任务3：分析图表
tasks.append(
api.ask_image_async("chart.jpg", "这个图表展示了什么数据？")
)
print("开始并发处理…")
# 同时执行所有任务
results = await asyncio.gather(*tasks, return_exceptions=True)
print("\\n并发处理完成:")
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"❌ 任务{i+1}失败: {result}")
else:
answer = result["choices"][0]["message"]["content"]
print(f"✅ 任务{i+1}成功: {answer[:80]}…")
def main():
"""主函数"""
print("异步处理示例")
print("=" * 60)
# 运行异步批量处理
asyncio.run(async_batch_process())
print("\\n" + "=" * 60)
print("并发处理示例")
print("=" * 60)
# 运行并发处理
asyncio.run(concurrent_processing())
if __name__ == "__main__":
main()
5. 实际应用场景示例
现在，让我们看几个实际的应用场景，看看如何将API集成到具体的项目中。
5.1 电商商品图片分析
假设你有一个电商平台，需要自动分析商品图片并生成描述：
#!/usr/bin/env python3
"""
电商商品图片分析示例
"""
from qwen_multimodal_api import QwenMultimodalAPIEnhanced
import json
class EcommerceImageAnalyzer:
"""电商图片分析器"""
def __init__(self, api_base_url="http://localhost:8000"):
self.api = QwenMultimodalAPIEnhanced(base_url=api_base_url)
def analyze_product_image(self, image_path: str) -> dict:
"""
分析商品图片，提取关键信息
Args:
image_path: 商品图片路径
Returns:
包含商品信息的字典
"""
# 定义分析问题
analysis_questions = [
"这是什么类型的商品？",
"商品的主要颜色是什么？",
"商品有哪些显著特征？",
"商品的使用场景是什么？",
"从图片看，这个商品的质量如何？"
]
analysis_results = {}
print(f"分析商品图片: {image_path}")
print("-" * 50)
for question in analysis_questions:
try:
answer = self.api.ask_image_simple(image_path, question)
analysis_results[question] = answer
print(f"Q: {question}")
print(f"A: {answer[:100]}…") # 显示前100字符
print()
except Exception as e:
print(f"分析失败: {e}")
analysis_results[question] = f"分析失败: {str(e)}"
# 生成商品描述
try:
description_prompt = "根据以上分析，为这个商品写一段吸引人的商品描述（100字以内）"
description = self.api.ask_image_simple(image_path, description_prompt)
analysis_results["商品描述"] = description
print(f"商品描述: {description}")
except Exception as e:
analysis_results["商品描述"] = f"生成失败: {str(e)}"
return analysis_results
def batch_analyze_products(self, product_images: list) -> list:
"""
批量分析多个商品图片
Args:
product_images: 商品图片路径列表
Returns:
分析结果列表
"""
all_results = []
for i, image_path in enumerate(product_images, 1):
print(f"\\n[{i}/{len(product_images)}] 分析商品: {image_path}")
try:
result = self.analyze_product_image(image_path)
result["image_path"] = image_path
result["status"] = "success"
all_results.append(result)
except Exception as e:
print(f"❌ 分析失败: {e}")
all_results.append({
"image_path": image_path,
"status": "error",
"error": str(e)
})
return all_results
def main():
"""主函数"""
analyzer = EcommerceImageAnalyzer()
# 检查服务状态
if not analyzer.api.check_health():
print("服务不可用，请先启动Qwen多模态服务")
return
# 测试商品图片分析
test_products = [
"product_shoes.jpg",
"product_dress.jpg",
"product_electronics.jpg"
]
# 实际使用时，替换成你的商品图片路径
import os
existing_products = [p for p in test_products if os.path.exists(p)]
if not existing_products:
print("没有找到商品图片，请准备测试图片")
print("创建示例图片或修改代码中的图片路径")
return
print("开始商品图片分析…")
print("=" * 60)
# 批量分析
results = analyzer.batch_analyze_products(existing_products[:2]) # 限制前2个
# 保存结果
output_file = "product_analysis_results.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"\\n分析完成! 结果已保存到: {output_file}")
# 显示摘要
print("\\n分析摘要:")
for result in results:
if result["status"] == "success":
product_type = result.get("这是什么类型的商品？", "未知")
print(f"商品: {os.path.basename(result['image_path'])}")
print(f" 类型: {product_type[:50]}…")
print(f" 描述: {result.get('商品描述', '无')[:80]}…")
print()
if __name__ == "__main__":
main()
5.2 内容审核系统
创建一个简单的内容审核工具，自动识别图片中的敏感内容：
#!/usr/bin/env python3
"""
内容审核系统示例
"""
from qwen_multimodal_api import QwenMultimodalAPIEnhanced
from typing import Dict, List, Tuple
class ContentModerationSystem:
"""内容审核系统"""
def __init__(self, api_base_url="http://localhost:8000"):
self.api = QwenMultimodalAPIEnhanced(base_url=api_base_url)
# 定义审核规则（可以根据需要扩展）
self.moderation_rules = {
"violence": ["暴力", "武器", "打架", "伤害", "血腥"],
"nudity": ["裸露", "色情", "不雅", "敏感部位"],
"drugs": ["毒品", "吸毒", "违禁药品", "大麻"],
"hate": ["仇恨", "歧视", "侮辱", "攻击性"],
"sensitive": ["政治", "敏感人物", "争议话题"]
}
def check_image_content(self, image_path: str) -> Dict:
"""
检查图片内容
Args:
image_path: 图片路径
Returns:
审核结果
"""
# 第一步：让AI描述图片内容
try:
description = self.api.ask_image_simple(
image_path,
"请详细描述这张图片的内容，包括人物、场景、动作、物品等所有细节"
)
except Exception as e:
return {
"status": "error",
"error": str(e),
"description": "",
"violations": [],
"risk_level": "unknown"
}
# 第二步：分析是否包含敏感内容
violations = self._analyze_violations(description)
# 第三步：评估风险等级
risk_level = self._assess_risk_level(violations)
return {
"status": "success",
"description": description,
"violations": violations,
"risk_level": risk_level,
"suggestion": self._get_suggestion(risk_level)
}
def _analyze_violations(self, description: str) -> List[str]:
"""分析描述中的违规内容"""
violations = []
description_lower = description.lower()
for category, keywords in self.moderation_rules.items():
for keyword in keywords:
if keyword in description_lower:
violations.append(f"{category}: {keyword}")
break # 每个类别只记录一次
return violations
def _assess_risk_level(self, violations: List[str]) -> str:
"""评估风险等级"""
if not violations:
return "safe"
elif len(violations) <= 2:
return "low"
elif len(violations) <= 4:
return "medium"
else:
return "high"
def _get_suggestion(self, risk_level: str) -> str:
"""根据风险等级给出建议"""
suggestions = {
"safe": "内容安全，可以发布",
"low": "内容基本安全，建议人工复核",
"medium": "内容存在风险，建议修改或删除",
"high": "内容高风险，建议立即删除",
"unknown": "审核失败，需要人工检查"
}
return suggestions.get(risk_level, "需要人工检查")
def batch_moderate(self, image_paths: List[str]) -> List[Dict]:
"""批量审核图片"""
results = []
for image_path in image_paths:
print(f"审核图片: {image_path}")
result = self.check_image_content(image_path)
result["image_path"] = image_path
# 输出结果
status_icon = "✅" if result["status"] == "success" else "❌"
risk_icon = {
"safe": "🟢",
"low": "🟡",
"medium": "🟠",
"high": "🔴",
"unknown": "⚫"
}.get(result["risk_level"], "⚫")
print(f" {status_icon} 状态: {result['status']}")
print(f" {risk_icon} 风险等级: {result['risk_level']}")
print(f" 违规项: {', '.join(result['violations']) if result['violations'] else '无'}")
print(f" 建议: {result['suggestion']}")
print(f" 描述摘要: {result['description'][:100]}…")
print()
results.append(result)
return results
def main():
"""主函数"""
moderator = ContentModerationSystem()
# 检查服务
if not moderator.api.check_health():
print("审核服务不可用")
return
# 测试图片（实际使用时替换为你的图片）
test_images = [
"test_image1.jpg",
"test_image2.jpg",
"test_image3.jpg"
]
# 过滤存在的图片
import os
existing_images = [img for img in test_images if os.path.exists(img)]
if not existing_images:
print("没有找到测试图片")
return
print("开始内容审核…")
print("=" * 60)
# 批量审核
results = moderator.batch_moderate(existing_images)
# 统计结果
safe_count = sum(1 for r in results if r["risk_level"] == "safe")
risky_count = len(results) – safe_count
print("=" * 60)
print(f"审核完成! 共审核 {len(results)} 张图片")
print(f"安全图片: {safe_count} 张")
print(f"风险图片: {risky_count} 张")
# 保存详细结果
output_file = "moderation_results.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"详细结果已保存到: {output_file}")
if __name__ == "__main__":
main()
5.3 教育辅助工具
创建一个帮助学生学习的教育工具：
#!/usr/bin/env python3
"""
教育辅助工具示例 – 帮助理解图表和图示
"""
from qwen_multimodal_api import QwenMultimodalAPIEnhanced
import json
class EducationAssistant:
"""教育辅助工具"""
def __init__(self, api_base_url="http://localhost:8000"):
self.api = QwenMultimodalAPIEnhanced(base_url=api_base_url)
def explain_chart(self, chart_image_path: str, subject: str = "数学") -> Dict:
"""
解释图表
Args:
chart_image_path: 图表图片路径
subject: 学科类型
Returns:
图表解释结果
"""
questions = [
f"这是一张{subject}图表，请描述图表的主要内容",
"图表中的横坐标和纵坐标分别代表什么？",
"图表展示了什么趋势或规律？",
"从图表中能得出什么结论？",
"这个图表在实际生活或学习中有什么应用？"
]
explanations = {}
print(f"分析{subject}图表: {chart_image_path}")
print("-" * 60)
for question in questions:
try:
answer = self.api.ask_image_simple(chart_image_path, question)
explanations[question] = answer
print(f"Q: {question}")
print(f"A: {answer[:120]}…")
print()
except Exception as e:
print(f"解释失败: {e}")
explanations[question] = f"解释失败: {str(e)}"
return explanations
def analyze_science_image(self, image_path: str, topic: str = "科学") -> Dict:
"""
分析科学图片
Args:
image_path: 科学图片路径
topic: 主题
Returns:
分析结果
"""
questions = [
f"这是一张{topic}相关的图片，请描述图片内容",
"图片中展示了什么科学原理或现象？",
"这个原理或现象在现实生活中有哪些应用？",
"相关的科学知识有哪些？",
"你能提出一个与图片相关的问题吗？"
]
analysis = {}
print(f"分析{topic}图片: {image_path}")
print("-" * 60)
for question in questions:
try:
answer = self.api.ask_image_simple(image_path, question)
analysis[question] = answer
print(f"Q: {question}")
print(f"A: {answer[:120]}…")
print()
except Exception as e:
print(f"分析失败: {e}")
analysis[question] = f"分析失败: {str(e)}"
return analysis
def create_quiz_from_image(self, image_path: str, difficulty: str = "中等") -> Dict:
"""
根据图片创建测验问题
Args:
image_path: 图片路径
difficulty: 难度等级
Returns:
测验问题
"""
prompt = f"""
根据这张图片，创建{difficulty}难度的测验问题。
请提供：
1. 3个选择题（每个问题4个选项）
2. 2个简答题
3. 所有问题的参考答案
格式要求：
选择题格式：
Q1: [问题]
A) [选项A]
B) [选项B]
C) [选项C]
D) [选项D]
答案: [正确选项]
简答题格式：
Q4: [问题]
参考答案: [答案]
"""
try:
quiz_content = self.api.ask_image_simple(image_path, prompt)
# 解析quiz内容
quiz = {
"image": image_path,
"difficulty": difficulty,
"content": quiz_content,
"questions": self._parse_quiz_content(quiz_content)
}
print(f"✅ 成功创建{difficulty}难度测验")
print(f"内容预览: {quiz_content[:200]}…")
return quiz
except Exception as e:
print(f"❌ 创建测验失败: {e}")
return {
"image": image_path,
"difficulty": difficulty,
"error": str(e)
}
def _parse_quiz_content(self, content: str) -> Dict:
"""解析测验内容（简化版）"""
# 这里可以添加更复杂的解析逻辑
return {"raw_content": content}
def main():
"""主函数"""
assistant = EducationAssistant()
# 检查服务
if not assistant.api.check_health():
print("教育辅助服务不可用")
return
print("教育辅助工具")
print("=" * 60)
# 测试图表解释
chart_image = "math_chart.jpg"
import os
if os.path.exists(chart_image):
print("1. 图表解释功能测试")
print("-" * 40)
chart_explanation = assistant.explain_chart(chart_image, "数学")
# 保存结果
with open("chart_explanation.json", 'w', encoding='utf-8') as f:
json.dump(chart_explanation, f, ensure_ascii=False, indent=2)
print("图表解释已保存到: chart_explanation.json")
# 测试科学图片分析
science_image = "science_diagram.jpg"
if os.path.exists(science_image):
print("\\n2. 科学图片分析测试")
print("-" * 40)
science_analysis = assistant.analyze_science_image(science_image, "物理")
with open("science_analysis.json", 'w', encoding='utf-8') as f:
json.dump(science_analysis, f, ensure_ascii=False, indent=2)
print("科学分析已保存到: science_analysis.json")
# 测试测验创建
quiz_image = "history_image.jpg"
if os.path.exists(quiz_image):
print("\\n3. 测验创建功能测试")
print("-" * 40)
quiz = assistant.create_quiz_from_image(quiz_image, "中等")
with open("quiz_content.json", 'w', encoding='utf-8') as f:
json.dump(quiz, f, ensure_ascii=False, indent=2)
print("测验内容已保存到: quiz_content.json")
print("\\n" + "=" * 60)
print("教育辅助工具测试完成!")
print("=" * 60)
if __name__ == "__main__":
main()
6. 总结与最佳实践
通过上面的教程和示例，你已经掌握了如何为Qwen3.5-35B-A3B-AWQ-4bit封装API接口，并用Python进行调用。让我们回顾一下关键要点，并分享一些最佳实践。
6.1 关键要点回顾
6.2 最佳实践建议
1. 错误处理与重试
# 总是添加适当的错误处理
try:
response = api.ask_image(image_path, question)
except requests.exceptions.ConnectionError:
# 处理连接错误
pass
except requests.exceptions.Timeout:
# 处理超时
pass
except Exception as e:
# 处理其他错误
print(f"未知错误: {e}")
2. 性能优化
- 对于批量处理，使用异步版本提高并发性能
- 合理设置超时时间，避免长时间等待
- 考虑使用连接池复用HTTP连接
3. 资源管理
# 使用上下文管理器确保资源正确释放
async with AsyncQwenMultimodalAPI() as api:
results = await api.batch_ask_images_async(tasks)
4. 日志记录
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)
logger = logging.getLogger(__name__)
# 在关键位置添加日志
logger.info(f"开始处理图片: {image_path}")
logger.error(f"处理失败: {error}")
5. 配置管理
import os
from dataclasses import dataclass
@dataclass
class APIConfig:
base_url: str = os.getenv("QWEN_API_URL", "http://localhost:8000")
timeout: int = int(os.getenv("QWEN_TIMEOUT", "300"))
max_retries: int = int(os.getenv("QWEN_MAX_RETRIES", "3"))
retry_delay: float = float(os.getenv("QWEN_RETRY_DELAY", "2.0"))
# 使用配置
config = APIConfig()
api = QwenMultimodalAPIEnhanced(
base_url=config.base_url,
max_retries=config.max_retries,
retry_delay=config.retry_delay
)
6.3 常见问题解决
问题1：服务连接失败
- 检查服务是否运行：supervisorctl status qwen35awq-backend
- 检查端口是否监听：ss -ltnp | grep 8000
- 确认防火墙设置
问题2：响应时间过长
- 减少图片尺寸（API封装中可以考虑自动压缩）
- 调整max_tokens参数，减少生成长度
- 使用异步处理避免阻塞
问题3：内存不足
- 减少并发请求数量
- 优化图片处理，使用缩略图
- 增加服务端资源
问题4：回答质量不佳
- 优化问题表述，更清晰具体
- 调整temperature参数（0.7通常是个好起点）
- 提供更详细的上下文信息
6.4 下一步学习建议
通过这篇教程，你已经掌握了Qwen3.5-35B-A3B-AWQ-4bit API封装的核心技能。现在你可以将这个强大的多模态AI能力集成到自己的项目中，无论是开发智能应用、自动化工具，还是构建复杂的AI系统，都有了坚实的基础。
记住，技术的学习在于实践。建议你从简单的示例开始，逐步尝试更复杂的应用场景。如果在实践中遇到问题，可以回顾本文中的代码示例，或者根据你的具体需求进行调整和扩展。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
