Qwen3.5-27B图文理解API封装：Python requests调用+异常处理完整示例

1. 引言

Qwen3.5-27B是一款强大的视觉多模态理解模型，支持文本对话与图片理解功能。在实际应用中，我们通常需要通过编程方式调用其API接口，而不是仅仅使用Web界面。本文将详细介绍如何使用Python的requests库封装Qwen3.5-27B的图片理解API，并实现完善的异常处理机制。

通过本文，你将学会：

  • 如何准备API调用环境
  • 如何构建图片理解请求
  • 如何处理各种可能出现的异常情况
  • 如何优化API调用体验

2. 环境准备

2.1 安装必要库

首先确保你的Python环境已安装requests库，这是调用API的基础：

pip install requests

如果你需要处理图片，建议也安装Pillow库：

pip install pillow

2.2 获取API地址

根据你的部署环境，API地址可能有所不同。通常格式为：

BASE_URL = "http://127.0.0.1:7860"  # 本地部署
# 或
BASE_URL = "https://gpu-{实例ID}-7860.web.gpu.csdn.net/"  # CSDN星图镜像

3. 基础API调用

3.1 文本对话接口

我们先从简单的文本对话接口开始，了解基本的调用方式：

import requests
import json

def text_chat(prompt, max_new_tokens=128):
    url = f"{BASE_URL}/generate"
    headers = {"Content-Type": "application/json"}
    data = {
        "prompt": prompt,
        "max_new_tokens": max_new_tokens
    }
    
    try:
        response = requests.post(url, headers=headers, data=json.dumps(data))
        response.raise_for_status()  # 检查HTTP错误
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 使用示例
result = text_chat("请用中文介绍一下你自己")
if result:
    print(result.get("response", "无响应内容"))

3.2 图片理解接口

图片理解接口需要使用multipart/form-data格式发送数据：

def image_understanding(image_path, prompt="请描述这张图片的主要内容", max_new_tokens=128):
    url = f"{BASE_URL}/generate_with_image"
    
    try:
        with open(image_path, 'rb') as img_file:
            files = {
                'image': (image_path, img_file, 'image/png'),
                'prompt': (None, prompt),
                'max_new_tokens': (None, str(max_new_tokens))
            }
            response = requests.post(url, files=files)
            response.raise_for_status()
            return response.json()
    except FileNotFoundError:
        print(f"图片文件未找到: {image_path}")
        return None
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 使用示例
result = image_understanding("example.png")
if result:
    print(result.get("response", "无响应内容"))

4. 异常处理完善

4.1 常见异常类型

在实际应用中，我们需要处理多种可能的异常情况：

  1. 网络连接问题：服务器不可达、连接超时等
  2. 认证问题：如果API需要认证
  3. 请求限制：频率限制或配额不足
  4. 数据格式问题：图片格式不支持或损坏
  5. 服务器错误：API服务内部错误

4.2 增强版异常处理

下面是一个更完善的异常处理实现：

from PIL import Image
import io

def enhanced_image_understanding(image_path, prompt="请描述这张图片的主要内容", max_new_tokens=128):
    url = f"{BASE_URL}/generate_with_image"
    
    try:
        # 1. 检查并预处理图片
        try:
            with Image.open(image_path) as img:
                # 转换为RGB模式，确保兼容性
                if img.mode != 'RGB':
                    img = img.convert('RGB')
                
                # 将图片保存到内存中的字节流
                img_byte_arr = io.BytesIO()
                img.save(img_byte_arr, format='PNG')
                img_byte_arr = img_byte_arr.getvalue()
        except Exception as e:
            print(f"图片处理失败: {e}")
            return None
        
        # 2. 准备请求数据
        files = {
            'image': ('processed.png', img_byte_arr, 'image/png'),
            'prompt': (None, prompt),
            'max_new_tokens': (None, str(max_new_tokens))
        }
        
        # 3. 发送请求并处理响应
        try:
            response = requests.post(url, files=files, timeout=30)  # 设置30秒超时
            response.raise_for_status()
            
            # 检查响应内容
            if not response.content:
                print("空响应内容")
                return None
                
            return response.json()
            
        except requests.exceptions.Timeout:
            print("请求超时，请检查网络或稍后重试")
            return None
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                print("请求过于频繁，请稍后再试")
            elif e.response.status_code >= 500:
                print("服务器内部错误，请稍后重试")
            else:
                print(f"HTTP错误: {e}")
            return None
        except json.JSONDecodeError:
            print("响应不是有效的JSON格式")
            return None
            
    except Exception as e:
        print(f"未知错误: {e}")
        return None

# 使用示例
result = enhanced_image_understanding("example.jpg", "这张图片中有哪些主要物体？")
if result:
    print(result.get("response", "无响应内容"))

5. 高级功能实现

5.1 流式响应处理

Qwen3.5-27B支持流式响应，我们可以通过以下方式处理：

def stream_chat(prompt, max_new_tokens=128):
    url = f"{BASE_URL}/chat_stream"
    headers = {"Content-Type": "application/json"}
    data = {
        "prompt": prompt,
        "max_new_tokens": max_new_tokens,
        "stream": True
    }
    
    try:
        with requests.post(url, headers=headers, json=data, stream=True) as response:
            response.raise_for_status()
            
            for chunk in response.iter_content(chunk_size=None):
                if chunk:
                    print(chunk.decode('utf-8'), end='', flush=True)
                    
    except requests.exceptions.RequestException as e:
        print(f"流式请求失败: {e}")

# 使用示例
stream_chat("请用中文介绍一下你自己")

5.2 上下文保持的多轮对话

要实现多轮对话，需要维护对话历史：

class QwenChatSession:
    def __init__(self):
        self.history = []
        
    def chat(self, prompt, max_new_tokens=128):
        url = f"{BASE_URL}/generate"
        headers = {"Content-Type": "application/json"}
        
        # 将历史对话加入prompt
        full_prompt = "\n".join([f"Q: {q}\nA: {a}" for q, a in self.history])
        full_prompt += f"\nQ: {prompt}\nA:"
        
        data = {
            "prompt": full_prompt,
            "max_new_tokens": max_new_tokens
        }
        
        try:
            response = requests.post(url, headers=headers, json=data)
            response.raise_for_status()
            result = response.json()
            
            if result and "response" in result:
                answer = result["response"]
                self.history.append((prompt, answer))
                return answer
            return None
                
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return None

# 使用示例
session = QwenChatSession()
print(session.chat("你好，你是谁？"))
print(session.chat("你能做什么？"))

6. 性能优化建议

6.1 连接池管理

使用requests.Session可以重用TCP连接，提高性能：

class QwenAPIClient:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = requests.Session()
        
    def image_understanding(self, image_path, prompt="请描述这张图片的主要内容", max_new_tokens=128):
        url = f"{self.base_url}/generate_with_image"
        
        try:
            with open(image_path, 'rb') as img_file:
                files = {
                    'image': (image_path, img_file, 'image/png'),
                    'prompt': (None, prompt),
                    'max_new_tokens': (None, str(max_new_tokens))
                }
                response = self.session.post(url, files=files, timeout=30)
                response.raise_for_status()
                return response.json()
        except Exception as e:
            print(f"请求失败: {e}")
            return None
            
    def close(self):
        self.session.close()

# 使用示例
client = QwenAPIClient("http://127.0.0.1:7860")
try:
    result = client.image_understanding("example.png")
    print(result)
finally:
    client.close()

6.2 异步调用

对于需要高并发的场景，可以使用aiohttp实现异步调用：

import aiohttp
import asyncio

async def async_image_understanding(image_path, prompt="请描述这张图片的主要内容", max_new_tokens=128):
    url = f"{BASE_URL}/generate_with_image"
    
    try:
        async with aiohttp.ClientSession() as session:
            with open(image_path, 'rb') as f:
                data = aiohttp.FormData()
                data.add_field('image', f, filename=image_path, content_type='image/png')
                data.add_field('prompt', prompt)
                data.add_field('max_new_tokens', str(max_new_tokens))
                
                async with session.post(url, data=data) as response:
                    if response.status == 200:
                        return await response.json()
                    else:
                        print(f"请求失败，状态码: {response.status}")
                        return None
    except Exception as e:
        print(f"异步请求失败: {e}")
        return None

# 使用示例
async def main():
    result = await async_image_understanding("example.png")
    print(result)

asyncio.run(main())

7. 总结

本文详细介绍了如何使用Python requests库调用Qwen3.5-27B的图片理解API，并实现了完善的异常处理机制。主要内容包括：

  1. 基础API调用：文本对话和图片理解接口的基本使用方法
  2. 异常处理：网络、数据、服务器等各种异常情况的处理方式
  3. 高级功能：流式响应、多轮对话等进阶用法
  4. 性能优化：连接池管理和异步调用等优化技巧

通过合理的封装和异常处理，可以大大提高API调用的稳定性和用户体验。在实际应用中，建议根据具体需求选择合适的调用方式，并做好日志记录和监控。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

DeepSeek RAG 热点文档加权：如何平衡实时性与检索质量

avatar DeepSeek技术社区
cover

多副本推理网关：路由规则该用代码还是配置？从 DeepSeek 生产环境看选型边界

avatar DeepSeek技术社区
cover

离线评测全绿上线被骂：DeepSeek-V4 模型切换的评测陷阱与影子流量实践

avatar DeepSeek技术社区