音视频文件理解

import os
import base64
import soundfile as sf
import numpy as np
from openai import OpenAI

# 1. 初始化客户端
client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),  # 确保环境变量已配置
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

# 2. 发起请求
try:
  completion = client.chat.completions.create(
    model="qwen3.5-omni-plus",
    messages=[{"role": "user", "content": "你是谁？"}],
    modalities=["text", "audio"],  # 指定文本和音频输出
    audio={"voice": "Tina", "format": "wav"},
    stream=True,  # 必须设为 True
    stream_options={"include_usage": True},
  )

  # 3. 处理流式响应并解码音频
  print("模型回复：")
  audio_base64_string = ""
  for chunk in completion:
    # 处理文本部分
    if chunk.choices and chunk.choices[0].delta.content:
      print(chunk.choices[0].delta.content, end="")

    # 收集音频部分
    if chunk.choices and hasattr(chunk.choices[0].delta, "audio") and chunk.choices[0].delta.audio:
      audio_base64_string += chunk.choices[0].delta.audio.get("data", "")

  # 4. 保存音频文件
  if audio_base64_string:
    wav_bytes = base64.b64decode(audio_base64_string)
    audio_np = np.frombuffer(wav_bytes, dtype=np.int16)
    sf.write("audio_assistant.wav", audio_np, samplerate=24000)
    print("\n音频文件已保存至: audio_assistant.wav")

except Exception as e:
  print(f"请求失败: {e}")

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

try:
  completion = client.chat.completions.create(
    model="qwen3.5-omni-plus",
    messages=[{
      "role": "user",
      "content": "请查询今天的日期和星期几，并告诉我今天有哪些重要节日。"
    }],
    stream=True,
    stream_options={"include_usage": True},
    extra_body={
      "enable_search": True,
      "search_options": {
        "search_strategy": "agent"
      }
    }
  )

  print("模型回复（包含实时信息）：")
  for chunk in completion:
    if chunk.choices and chunk.choices[0].delta.content:
      print(chunk.choices[0].delta.content, end="")
  print()

except Exception as e:
  print(f"请求失败: {e}")

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[
    {
      "role": "user",
      "content": [
        {
          "type": "video_url",
          "video_url": {
            "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241115/cqqkru/1.mp4"
          },
        },
        {"type": "text", "text": "这个视频讲了什么？"},
      ],
    },
  ],
  modalities=["text", "audio"],
  audio={"voice": "Tina", "format": "wav"},
  stream=True,
  stream_options={"include_usage": True},
)

for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta)
  else:
    print(chunk.usage)

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[
    {
      "role": "user",
      "content": [
        {
          "type": "video",
          "video": [
            "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241108/xzsgiz/football1.jpg",
            "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241108/tdescd/football2.jpg",
            "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241108/zefdja/football3.jpg",
            "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241108/aedbqh/football4.jpg",
          ],
        },
        {"type": "text", "text": "描述这个视频中展示的过程"},
      ],
    }
  ],
  modalities=["text", "audio"],
  audio={"voice": "Tina", "format": "wav"},
  stream=True,
  stream_options={"include_usage": True},
)

for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta)
  else:
    print(chunk.usage)

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[
    {
      "role": "user",
      "content": [
        {
          "type": "input_audio",
          "input_audio": {
            "data": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250211/tixcef/cherry.wav",
            "format": "wav",
          },
        },
        {"type": "text", "text": "这段音频讲了什么"},
      ],
    },
  ],
  modalities=["text", "audio"],
  audio={"voice": "Tina", "format": "wav"},
  stream=True,
  stream_options={"include_usage": True},
)

for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta)
  else:
    print(chunk.usage)

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[
    {
      "role": "user",
      "content": [
        {
          "type": "image_url",
          "image_url": {
            "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"
          },
        },
        {"type": "text", "text": "图片中描绘了什么场景？"},
      ],
    },
  ],
  # 设置输出数据模态。当前支持两种：["text","audio"] 和 ["text"]。
  modalities=["text", "audio"],
  audio={"voice": "Tina", "format": "wav"},
  # stream 必须设置为 True，否则会报错。
  stream=True,
  stream_options={
    "include_usage": True
  }
)

for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta)
  else:
    print(chunk.usage)

import os
from openai import OpenAI

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[
    {
      "role": "user",
      "content": [
        {
          "type": "input_audio",
          "input_audio": {
            "data": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3",
            "format": "mp3",
          },
        },
        {"type": "text", "text": "这段音频讲了什么"},
      ],
    },
    {
      "role": "assistant",
      "content": [{"type": "text", "text": "这段音频的内容是：欢迎来到千问AI平台"}],
    },
    {
      "role": "user",
      "content": [{"type": "text", "text": "你能介绍一下这家公司吗？"}],
    },
  ],
  # 设置输出数据模态。当前支持两种：["text","audio"] 和 ["text"]。
  modalities=["text"],
  # stream 必须设置为 True，否则会报错。
  stream=True,
  stream_options={"include_usage": True},
)

for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta)
  else:
    print(chunk.usage)

# pyaudio 安装说明:
# APPLE Mac OS X
#   brew install portaudio
#   pip install pyaudio
# Debian/Ubuntu
#   sudo apt-get install python-pyaudio python3-pyaudio
#   或
#   pip install pyaudio
# CentOS
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# Microsoft Windows
#   python -m pip install pyaudio

import os
from openai import OpenAI
import base64
import numpy as np
import soundfile as sf

client = OpenAI(
  api_key=os.getenv("DASHSCOPE_API_KEY"),
  base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
  model="qwen3.5-omni-plus",
  messages=[{"role": "user", "content": "你是谁"}],
  # 设置输出数据模态。当前支持两种：["text","audio"] 和 ["text"]。
  modalities=["text", "audio"],
  audio={"voice": "Tina", "format": "wav"},
  # stream 必须设置为 True，否则会报错。
  stream=True,
  stream_options={"include_usage": True},
)

# 方法一：生成完成后解码
audio_string = ""
for chunk in completion:
  if chunk.choices:
    if hasattr(chunk.choices[0].delta, "audio"):
      try:
        audio_string += chunk.choices[0].delta.audio["data"]
      except Exception as e:
        print(chunk.choices[0].delta.content)
  else:
    print(chunk.usage)

wav_bytes = base64.b64decode(audio_string)
audio_np = np.frombuffer(wav_bytes, dtype=np.int16)
sf.write("audio_assistant_py.wav", audio_np, samplerate=24000)

# 方法二：边生成边解码（使用方法二时请注释掉方法一的代码）
# # 初始化 PyAudio
# import pyaudio
# import time
# p = pyaudio.PyAudio()
# # 创建音频流
# stream = p.open(format=pyaudio.paInt16,
#                 channels=1,
#                 rate=24000,
#                 output=True)

# for chunk in completion:
#     if chunk.choices:
#         if hasattr(chunk.choices[0].delta, "audio"):
#             try:
#                 audio_string = chunk.choices[0].delta.audio["data"]
#                 wav_bytes = base64.b64decode(audio_string)
#                 audio_np = np.frombuffer(wav_bytes, dtype=np.int16)
#                 # 直接播放音频数据
#                 stream.write(audio_np.tobytes())
#             except Exception as e:
#                 print(chunk.choices[0].delta.content)

# time.sleep(0.8)
# # 清理资源
# stream.stop_stream()
# stream.close()
# p.terminate()