MaaS Tokens 多模态接口调用示例

举报
AI在路上 发表于 2025/10/10 19:07:27 2025/10/10
【摘要】 MaaS Tokens 多模态接口Python脚本调用示例

一、MaaS tokens多模态模型

序号
服务名称
类型
计费方式
推理定价
模型限流
（默认值，可调）
1
Wan2.1-14B
视频生成
按token计费
输出:¥0.012 / 千tokens
TPM: --
RPM: 60
2
SDXL
图片生成
按生成耗时计费
¥0.02 / 秒
TPM: --
RPM: 144
3
Qwen2.5-VL-72B
图像理解
按token计费
输入:¥0.016 / 千tokens
输出:¥0.048 / 千tokens
TPM: 100,000
RPM: 60
4
Qwen_Image
图片生成
按张计费
¥0.25 / 张
TPM: --
RPM: 144
5
Wan2.2-T2V-A14B
视频生成
按token计费
输出:¥0.005 / 千tokens
TPM: --
RPM: 60
6
Wan2.2-I2V-A14B
视频生成
按token计费
输出:¥0.005 / 千tokens
TPM: --
RPM: 60
7
Qwen-Image
图片生成
按生成耗时计费
¥0.02 / 秒
TPM: --
RPM: 144
8
Qwen-Image-Edit
图片生成
按张计费
¥0.3 / 张
TPM: --
RPM: 144

二、模型调用示例

1、文生视频（以Python语言调用“wan2.1-t2v-14b”模型为例）

代码脚本：

import requests
import json
import time
import os
from typing import Optional, Dict


def create_video_generation_task(api_key: str, prompt: str, model: str = "wan2.1-t2v-14b",
                                 size: str = "720*1280", fps: int = 16, duration: int = 5,
                                 seed: int = 0) -> Optional[str]:
    """
    第一步：创建视频生成任务并获取task_id

    Args:
        api_key: API认证密钥
        prompt: 生成视频的文本描述
        model: 使用的模型名称
        size: 视频分辨率
        fps: 帧率
        duration: 视频时长(秒)
        seed: 随机种子

    Returns:
        任务ID(task_id)，失败返回None
    """
    url = "https://api.modelarts-maas.com/v1/video/generations"

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    data = {
        "model": model,
        "input": {
            "prompt": prompt,
        },
        "parameters": {
            "size": size,
            "fps": fps,
            "duration": duration,
            "seed": seed
        }
    }

    try:
        response = requests.post(url, headers=headers, data=json.dumps(data))

        if response.status_code == 200:
            result = response.json()
            task_id = result.get('task_id')
            if task_id:
                print(f"✅ 任务创建成功，task_id: {task_id}")
                return task_id
            else:
                print("❌ 响应中未找到task_id")
                print(f"完整响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
                return None
        else:
            print(f"❌ 任务创建失败，状态码: {response.status_code}")
            print(f"错误信息: {response.text}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"❌ JSON解析异常: {e}")
        print(f"原始响应: {response.text}")
        return None


def check_video_generation_status(task_id: str, api_key: str) -> Optional[Dict]:
    """
    第二步：根据task_id查询视频生成任务状态

    Args:
        task_id: 视频生成任务ID
        api_key: API认证密钥

    Returns:
        包含任务状态和视频URL的响应数据，失败返回None
    """
    url = f"https://api.modelarts-maas.com/v1/video/generations/{task_id}"

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    try:
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            return response.json()
        else:
            print(f"❌ 状态查询失败，状态码: {response.status_code}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"❌ JSON解析异常: {e}")
        return None


def download_video(video_url: str, save_path: str) -> bool:
    """
    下载视频文件到本地

    Args:
        video_url: 视频文件的URL
        save_path: 本地保存路径

    Returns:
        下载是否成功
    """
    try:
        # 设置流式下载，适合大文件[6](@ref)
        response = requests.get(video_url, stream=True,)

        if response.status_code == 200:
            # 获取文件大小信息（如果服务器提供）
            total_size = int(response.headers.get('content-length', 0))
            downloaded_size = 0

            with open(save_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
                        downloaded_size += len(chunk)

                        # 显示下载进度（如果有文件大小信息）
                        if total_size > 0:
                            percent = (downloaded_size / total_size) * 100
                            print(f"\r📥 下载进度: {percent:.1f}% ({downloaded_size}/{total_size} bytes)", end='')

            print(f"\n✅ 视频已成功保存到: {os.path.abspath(save_path)}")
            return True
        else:
            print(f"❌ 视频下载失败，状态码: {response.status_code}")
            return False

    except Exception as e:
        print(f"❌ 下载过程中发生错误: {e}")
        return False


def wait_for_video_completion(task_id: str, api_key: str, max_wait_time: int = 600,
                              check_interval: int = 10) -> Optional[str]:
    """
    等待视频生成完成并返回视频URL

    Args:
        task_id: 任务ID
        api_key: API密钥
        max_wait_time: 最大等待时间(秒)
        check_interval: 检查间隔(秒)

    Returns:
        视频URL，超时或失败返回None
    """
    start_time = time.time()
    attempts = 0

    print("⏳ 等待视频生成完成...")
    print("💡 提示：视频生成通常需要30秒到几分钟，请耐心等待")

    while time.time() - start_time < max_wait_time:
        attempts += 1
        print(f"\r🔄 第 {attempts} 次检查任务状态...", end='')

        result = check_video_generation_status(task_id, api_key)

        if not result:
            print("\n❌ 获取任务状态失败")
            return None

        status = result.get('status', 'unknown')

        if status == 'succeeded':
            result_url = result.get('content', {}).get('result_url', '')
            if result_url:
                elapsed_time = time.time() - start_time
                print(f"\n✅ 视频生成成功！耗时: {elapsed_time:.1f}秒")
                return result_url
            else:
                print("\n❌ 任务成功但未找到视频URL")
                return None

        elif status == 'failed':
            error_info = result.get('error', {})
            print(f"\n❌ 任务失败，错误代码: {error_info.get('code', '未知')}")
            print(f"错误信息: {error_info.get('message', '未知错误')}")
            return None

        elif status in ['queued', 'running','moderating']:
            # 显示预估等待时间
            elapsed = time.time() - start_time
            remaining = max_wait_time - elapsed
            print(f" 当前状态: {status}, 已等待: {elapsed:.0f}秒, 剩余时间: {remaining:.0f}秒", end='')

            if status == 'running':
                # 处理中状态可以稍微缩短检查间隔
                time.sleep(max(5, check_interval))
            else:
                time.sleep(check_interval)

        else:
            print(f"\n❓ 未知状态: {status}")
            return None

    print(f"\n⏰ 超过最大等待时间({max_wait_time}秒)，任务仍未完成")
    return None


def main():
    # 配置参数
    api_key = "YOU API KEY"
   # api_key = "your_api_key_here"  # 替换为你的有效API密钥
    prompt = "小猫在散步"  # 替换为你的提示词

    # 第一步：创建视频生成任务
    print("🚀 正在创建视频生成任务...")
    task_id = create_video_generation_task(api_key, prompt)

    if not task_id:
        print("❌ 任务创建失败，程序退出")
        return

    # 第二步：等待视频生成完成并获取URL
    video_url = wait_for_video_completion(task_id, api_key, max_wait_time=300, check_interval=10)

    if not video_url:
        print("❌ 无法获取视频URL，程序退出")
        return

    # 第三步：下载视频到本地
    print(f"🌐 获取到视频URL: {video_url}")

    # 生成带时间戳的文件名，避免覆盖
    timestamp = int(time.time())
    save_path = f"generated_video_{timestamp}.mp4"

    print("📥 开始下载视频...")
    if download_video(video_url, save_path):
        print("🎉 视频处理流程完成！")
    else:
        print("❌ 视频下载失败")


if __name__ == '__main__':
    main()

调用结果：

🚀 正在创建视频生成任务...
✅ 任务创建成功，task_id: c64628525b7b46c7cdc3e7b631fc6ad5
⏳ 等待视频生成完成...
💡 提示：视频生成通常需要30秒到几分钟
🔄 第 17 次检查状态...
✅ 视频生成成功！耗时: 168.2秒
🌐 获取到视频URL: https://modelarts-pri-inference-async.obs.cn-southwest-2.myhuaweicloud.com:443/a837b47d-82d6-4578-a410-6eaf586fa703/default/c64628525b7b46c7cdc3e7b631fc6ad5.mp4?AWSAccessKeyId=HPUAJ5GBCRMVJDQW14D6&Expires=1758794153&Signature=e3HGMQz0%2FtQMrrt%2FJ%2BU7Fh9ZHvU%3D
📥 开始下载视频...
📥 下载进度: 100.0%
✅ 视频已保存到: C:\Users\Maggie\PycharmProjects\PythonProject3\generated_video_1758707921.mp4

2、图生视频（以Python语言调用“wan2.1-i2v-14b-720p”模型为例）

import requests
import json
import time
import os
import base64
import urllib3
from typing import Optional, Dict, Union
from pathlib import Path

# 禁用SSL警告（生产环境应配置正确的证书验证）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def image_to_base64(image_path: Union[str, Path]) -> Optional[str]:
    """
    将本地图片转换为Base64编码字符串

    Args:
        image_path: 本地图片路径

    Returns:
        Base64编码的图片字符串，失败返回None
    """
    try:
        with open(image_path, 'rb') as img_file:
            img_data = img_file.read()
            return f"data:image/{Path(image_path).suffix[1:]};base64,{base64.b64encode(img_data).decode('utf-8')}"
    except Exception as e:
        print(f"❌ 图片转换失败: {e}")
        return None


def create_video_from_image(
        api_key: str,
        image_input: Union[str, Path],  # 可以是网络URL或本地路径
        prompt: str = "",
        model: str = "wan2.1-i2v-14b-720p",
        size: str = "720*1280",
        fps: int = 16,
        duration: int = 5,
        seed: int = 0
) -> Optional[str]:
    """
    创建图片生成视频任务

    Args:
        api_key: API认证密钥
        image_input: 图片输入（网络URL或本地路径）
        prompt: 视频描述文本
        model: 使用的模型名称
        size: 视频分辨率
        fps: 帧率
        duration: 视频时长(秒)
        seed: 随机种子

    Returns:
        任务ID(task_id)，失败返回None
    """
    url = "https://api.modelarts-maas.com/v1/video/generations"

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    # 判断输入是网络URL还是本地文件
    if isinstance(image_input, str) and image_input.startswith(('http://', 'https://')):
        image_data = image_input  # 直接使用网络URL
    else:
        image_data = image_to_base64(image_input)  # 本地文件转为Base64
        if not image_data:
            return None

    data = {
        "model": model,
        "input": {
            "prompt": prompt,
            "img_url": image_data
        },
        "parameters": {
            "size": size,
            "fps": fps,
            "duration": duration,
            "seed": seed
        }
    }

    try:
        response = requests.post(url, headers=headers, data=json.dumps(data))

        if response.status_code == 200:
            result = response.json()
            task_id = result.get('task_id')
            if task_id:
                print(f"✅ 任务创建成功，task_id: {task_id}")
                return task_id
            else:
                print("❌ 响应中未找到task_id")
                print(f"完整响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
                return None
        else:
            print(f"❌ 任务创建失败，状态码: {response.status_code}")
            print(f"错误信息: {response.text}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None


def check_video_status(task_id: str, api_key: str) -> Optional[Dict]:
    """
    查询视频生成任务状态

    Args:
        task_id: 视频生成任务ID
        api_key: API认证密钥

    Returns:
        包含任务状态和视频URL的响应数据，失败返回None
    """
    url = f"https://api.modelarts-maas.com/v1/video/generations/{task_id}"

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    try:
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            return response.json()
        else:
            print(f"❌ 状态查询失败，状态码: {response.status_code}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None


def download_video(video_url: str, save_path: Union[str, Path]) -> bool:
    """
    下载视频文件到本地

    Args:
        video_url: 视频文件的URL
        save_path: 本地保存路径

    Returns:
        下载是否成功
    """
    try:
        response = requests.get(video_url, stream=True)

        if response.status_code == 200:
            total_size = int(response.headers.get('content-length', 0))
            downloaded_size = 0

            with open(save_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
                        downloaded_size += len(chunk)
                        if total_size > 0:
                            percent = (downloaded_size / total_size) * 100
                            print(f"\r📥 下载进度: {percent:.1f}%", end='')

            print(f"\n✅ 视频已保存到: {os.path.abspath(save_path)}")
            return True
        else:
            print(f"❌ 视频下载失败，状态码: {response.status_code}")
            return False

    except Exception as e:
        print(f"❌ 下载过程中发生错误: {e}")
        return False


def wait_for_video_completion(
        task_id: str,
        api_key: str,
        max_wait_time: int = 600,
        check_interval: int = 10
) -> Optional[str]:
    """
    等待视频生成完成

    Args:
        task_id: 任务ID
        api_key: API密钥
        max_wait_time: 最大等待时间(秒)
        check_interval: 检查间隔(秒)

    Returns:
        视频URL，超时或失败返回None
    """
    start_time = time.time()
    attempts = 0

    print("⏳ 等待视频生成完成...")
    print("💡 提示：视频生成通常需要30秒到几分钟")

    while time.time() - start_time < max_wait_time:
        attempts += 1
        print(f"\r🔄 第 {attempts} 次检查状态...", end='')

        result = check_video_status(task_id, api_key)

        if not result:
            print("\n❌ 获取任务状态失败")
            return None

        status = result.get('status', 'unknown')

        if status == 'succeeded':
            video_url = result.get('content', {}).get('result_url', '')
            if video_url:
                elapsed = time.time() - start_time
                print(f"\n✅ 视频生成成功！耗时: {elapsed:.1f}秒")
                return video_url
            else:
                print("\n❌ 任务成功但未找到视频URL")
                return None

        elif status == 'failed':
            error_info = result.get('error', {})
            print(f"\n❌ 任务失败，错误代码: {error_info.get('code', '未知')}")
            print(f"错误信息: {error_info.get('message', '未知错误')}")
            return None

        elif status in ['queued', 'running', 'moderating']:
            # 显示预估等待时间
            elapsed = time.time() - start_time
            remaining = max_wait_time - elapsed
            print(f" 当前状态: {status}, 已等待: {elapsed:.0f}秒, 剩余时间: {remaining:.0f}秒", end='')

            if status == 'running':
                # 处理中状态可以稍微缩短检查间隔
                time.sleep(max(5, check_interval))
            else:
                time.sleep(check_interval)
    print(f"\n⏰ 超过最大等待时间({max_wait_time}秒)")
    return None


def main():
    # 配置参数
    api_key = "YOU API KEY"
    # api_key = "your_api_key_here"  # 替换为你的API密钥
    image_input = "C:\\Users\\Maggie\\Desktop\\cat.jpg"
    # image_input = "https://example.com/cat.jpg"  # 可以是网络URL或本地路径如"./cat.jpg"
    prompt = "小猫在草地上玩耍"  # 视频描述（可选）

    # 第一步：创建视频生成任务
    print("🚀 正在创建视频生成任务...")
    task_id = create_video_from_image(api_key, image_input, prompt)

    if not task_id:
        print("❌ 任务创建失败，程序退出")
        return

    # 第二步：等待视频生成完成
    video_url = wait_for_video_completion(task_id, api_key)

    if not video_url:
        print("❌ 无法获取视频URL，程序退出")
        return

    # 第三步：下载视频
    print(f"🌐 获取到视频URL: {video_url}")

    # 生成带时间戳的文件名
    timestamp = int(time.time())
    save_path = f"generated_video_{timestamp}.mp4"

    print("📥 开始下载视频...")
    if download_video(video_url, save_path):
        print("🎉 视频处理流程完成！")
    else:
        print("❌ 视频下载失败")


if __name__ == '__main__':
    main()

调用结果：

🚀 正在创建视频生成任务...
✅ 任务创建成功，task_id: 4c0197fc641a964a7f59c54f8c667c7c
⏳ 等待视频生成完成...
💡 提示：视频生成通常需要30秒到几分钟
🔄 第 16 次检查状态...
✅ 视频生成成功！耗时: 151.6秒
🌐 获取到视频URL: https://modelarts-pri-inference-async.obs.cn-southwest-2.myhuaweicloud.com:443/a837b47d-82d6-4578-a410-6eaf586fa703/default/4c0197fc641a964a7f59c54f8c667c7c.mp4?AWSAccessKeyId=HPUAJ5GBCRMVJDQW14D6&Expires=1760180459&Signature=kUtMjKduFdPsZv6%2FEPQ9h0cbfr4%3D
📥 开始下载视频...
📥 下载进度: 100.0%
✅ 视频已保存到: D:\Python_develop\Pythonproject\projectfile\maas-test\多模态Demo\多模态Demo\generated_video_1760094221.mp4
🎉 视频处理流程完成！

3、图片生成（以Python语言调用“qwen-image”模型为例）

import requests
import json
import base64
import os
import re
from typing import Union, Optional
from pathlib import Path


def fix_base64_padding(base64_string: str) -> str:
    """
    修复Base64字符串的填充问题

    Args:
        base64_string: 需要修复的Base64字符串

    Returns:
        修复后的Base64字符串
    """
    # 移除可能存在的data URI前缀
    base64_string = re.sub(r"^data:image/\w+;base64,", "", base64_string)

    # 检查长度是否为4的倍数，如果不是则补充等号
    missing_padding = 4 - len(base64_string) % 4
    if missing_padding != 4:  # 如果不是正好为4的倍数
        base64_string += '=' * missing_padding

    return base64_string


def text_to_image(
        api_key: str,
        prompt: str,
        save_path: Union[str, Path],
        model: str = "qwen-image",
        size: str = "1024x1024",
        response_format: str = "b64_json",
        seed: int = 1
) -> Optional[str]:
    """
    文生图功能，并将生成的图像保存到本地

    Args:
        api_key: MAAS平台API密钥
        prompt: 生成图像的描述文本
        save_path: 图像保存路径
        model: 使用的模型名称
        size: 生成图像尺寸 (格式: "widthxheight")
        response_format: 返回格式 (目前仅支持"b64_json")
        seed: 随机种子 (0-2147483648)
        verify_ssl: 是否验证SSL证书

    Returns:
        保存的图像文件路径，失败返回None
    """
    url = "https://api.modelarts-maas.com/v1/images/generations"

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    data = {
        "model": model,
        "prompt": prompt,
        "size": size,
        "response_format": response_format,
        "seed": seed
    }

    try:
        print("🚀 正在发送文生图请求...")
        response = requests.post(
            url,
            headers=headers,
            data=json.dumps(data),
        )

        print(f"📊 响应状态码: {response.status_code}")

        if response.status_code == 200:
            result = response.json()
            print("✅ API请求成功")

            # 获取Base64数据
            b64_data_list = result.get('data', [])
            if not b64_data_list:
                print("❌ 响应中未找到图像数据")
                return None

            b64_data = b64_data_list[0].get('b64_json', '')
            if not b64_data:
                print("❌ Base64数据为空")
                return None

            print(f"📏 原始Base64数据长度: {len(b64_data)}")

            # 修复Base64填充问题
            fixed_b64_data = fix_base64_padding(b64_data)
            print(f"🔧 修复后Base64数据长度: {len(fixed_b64_data)}")

            # 确保保存目录存在
            save_path = Path(save_path)
            save_path.parent.mkdir(parents=True, exist_ok=True)

            try:
                # 解码并保存图像
                image_data = base64.b64decode(fixed_b64_data)
                with open(save_path, 'wb') as f:
                    f.write(image_data)

                file_size = os.path.getsize(save_path)
                print(f"✅ 图像已保存到: {os.path.abspath(save_path)}")
                print(f"💾 文件大小: {file_size} 字节")
                return str(save_path)

            except Exception as e:
                print(f"❌ Base64解码或文件保存失败: {e}")
                return None

        else:
            print(f"❌ 文生图失败，状态码: {response.status_code}")
            print(f"错误信息: {response.text}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None
    except Exception as e:
        print(f"❌ 发生未知错误: {e}")
        return None


def validate_base64_string(base64_string: str) -> bool:
    """
    验证Base64字符串是否有效

    Args:
        base64_string: 要验证的Base64字符串

    Returns:
        是否有效
    """
    try:
        # 检查长度是否为4的倍数
        if len(base64_string) % 4 != 0:
            return False

        # 尝试解码验证
        base64.b64decode(base64_string)
        return True
    except Exception:
        return False


def main():
    # 配置参数
    api_key = "YOU API KEY"
    prompt = "一只在草地上奔跑的猫"
    save_path = "generated_cat.png"

    print("🎨 开始文生图流程...")
    print(f"📝 提示词: {prompt}")
    print(f"💾 保存路径: {save_path}")

    # 调用文生图函数
    result_path = text_to_image(
        api_key=api_key,
        prompt=prompt,
        save_path=save_path,
        size="1024x1024"
    )

    if result_path:
        print(f"🎉 文生图成功！图像保存在: {result_path}")
    else:
        print("❌ 文生图失败，请检查以上错误信息")


if __name__ == '__main__':
    main()

调用结果：

🎨 开始文生图流程...
📝 提示词: 一只在草地上奔跑的猫
💾 保存路径: generated_cat.png
🚀 正在发送文生图请求...
📊 响应状态码: 200
✅ API请求成功
📏 原始Base64数据长度: 2250902
🔧 修复后Base64数据长度: 2250880
✅ 图像已保存到: D:\Python_develop\Pythonproject\projectfile\maas-test\多模态Demo\多模态Demo\generated_cat.png
💾 文件大小: 1688158 字节
🎉 文生图成功！图像保存在: generated_cat.png

4、图像理解（以Python语言调用“qwen2.5-vl-72b”模型为例）

import requests
import json
import base64
import os
from pathlib import Path
from typing import Union, Optional

def main(
        api_url: str = "https://api.modelarts-maas.com/v1/chat/completions",
        api_key: str = "YOU API KEY",
        image_path: str = "generated_cat.png",
        model: str = "qwen2.5-vl-72b",
        prompt: str = "请详细描述这张图片的内容",
        output_file: str = "image_understanding_result.txt"
):
    """
    主函数 - 图片理解演示

    Args:
        api_url: API地址
        api_key: API密钥
        image_path: 图片路径
        model: 模型名称
        prompt: 提示词
        output_file: 输出文件路径
    """
    print("=== 华为云MAAS图片理解 ===")
    print(f"🌐 API地址: {api_url}")
    print(f"🤖 模型: {model}")
    print(f"🖼️ 图片: {image_path}")
    print(f"❓ 提示: {prompt}")

    # 调用图片理解函数
    result = understand_image(api_url, api_key, image_path, model, prompt)

    if result:
        print("\n📝 理解结果:")
        print("=" * 50)
        print(result)
        print("=" * 50)

        # 保存结果
        if save_result(result, output_file):
            print(f"💾 结果已保存: {output_file}")
    else:
        print("❌ 图片理解失败")


def encode_image(image_path: Union[str, Path]) -> Optional[str]:
    """
    将本地图片转换为Base64编码字符串

    Args:
        image_path: 本地图片路径

    Returns:
        Base64编码的图片字符串，失败返回None
    """
    try:
        path = Path(image_path)
        if not path.exists():
            print(f"❌ 图片文件不存在: {path}")
            return None

        with open(path, "rb") as image_file:
            image_data = image_file.read()
            # 根据文件扩展名确定图片格式
            ext = path.suffix.lower()[1:] if path.suffix else 'png'
            base64_str = base64.b64encode(image_data).decode("utf-8")
            return f"data:image/{ext};base64,{base64_str}"
    except Exception as e:
        print(f"❌ 图片编码失败: {e}")
        return None


def understand_image(
        api_url: str,
        api_key: str,
        image_path: Union[str, Path],
        model: str = "qwen2.5-vl-72b",
        prompt: str = "请描述这张图片的内容"
) -> Optional[str]:
    """
    图片理解功能（图生文）

    Args:
        api_url: API地址URL
        api_key: API认证密钥
        image_path: 图片文件路径
        model: 使用的模型名称
        prompt: 对图片的提问或指令

    Returns:
        图片理解结果文本，失败返回None
    """
    # 编码图片为Base64
    base64_image = encode_image(image_path)
    if not base64_image:
        return None

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {api_key}'
    }

    data = {
        "model": model,
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": base64_image
                        }
                    }
                ]
            }
        ]
    }

    try:
        print(f"🚀 发送图片理解请求到: {api_url}")
        response = requests.post(api_url, headers=headers, json=data, timeout=60)

        print(f"📊 响应状态码: {response.status_code}")

        if response.status_code == 200:
            result = response.json()
            # 提取回复内容
            if 'choices' in result and result['choices']:
                content = result['choices'][0].get('message', {}).get('content', '')
                if content:
                    return content
            print("❌ 响应中未找到有效内容")
        else:
            print(f"❌ 请求失败: {response.text}")

        return None

    except requests.exceptions.RequestException as e:
        print(f"❌ 网络请求异常: {e}")
        return None
    except Exception as e:
        print(f"❌ 发生未知错误: {e}")
        return None


def save_result(text: str, save_path: Union[str, Path]) -> bool:
    """
    保存理解结果到文件

    Args:
        text: 文本内容
        save_path: 保存路径

    Returns:
        保存是否成功
    """
    try:
        path = Path(save_path)
        path.parent.mkdir(parents=True, exist_ok=True)

        with open(path, 'w', encoding='utf-8') as f:
            f.write(text)

        print(f"✅ 结果已保存到: {path.resolve()}")
        return True
    except Exception as e:
        print(f"❌ 保存失败: {e}")
        return False




# 简化调用函数
def quick_understand(api_key: str, image_path: str, prompt: str = "描述图片") -> Optional[str]:
    """
    快速图片理解（简化调用）

    Args:
        api_key: API密钥
        image_path: 图片路径
        prompt: 提示词

    Returns:
        理解结果
    """
    api_url = "https://api.modelarts-maas.com/v1/chat/completions"
    return understand_image(api_url, api_key, image_path, "qwen2.5-vl-72b", prompt)


if __name__ == '__main__':
    # 使用默认参数调用
    main()

调用结果：

=== 华为云MAAS图片理解 ===
🌐 API地址: https://api.modelarts-maas.com/v1/chat/completions
🤖 模型: qwen2.5-vl-72b
🖼️ 图片: generated_cat.png
❓ 提示: 请详细描述这张图片的内容
🚀 发送图片理解请求到: https://api.modelarts-maas.com/v1/chat/completions
📊 响应状态码: 200

📝 理解结果:
==================================================
这张图片展示了一只正在奔跑的猫。猫的身体呈现出动态的姿态，前爪离地，后腿用力蹬地，显示出它正在快速移动。猫的毛色是典型的虎斑纹，身体上有深色和浅色相间的条纹，脸部也有明显的条纹图案。猫的眼睛大而明亮，呈现出绿色，目光直视前方，显得非常专注。

背景是一片绿色的草地，草地上有一些小花点缀，显得生机勃勃。远处的背景有些模糊，但可以看出是一些树木和灌木丛，营造出一种自然和谐的氛围。阳光从上方照射下来，给整个场景增添了一种温暖的感觉。整体画面充满了活力和动感，展现了猫在自然环境中自由奔跑的美好瞬间。
==================================================
✅ 结果已保存到: C:\Users\Maggie\PycharmProjects\PythonProject3\image_understanding_result.txt
💾 结果已保存: image_understanding_result.txt


【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
ModelArts Studio 人工智能
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入