【语音相关ASR】OpenAI 兼容 Qwen_ASR 接口调用实战：SDK 与 HTTP 方式详解

CodeWang_NC

108人浏览 · 2026-04-08 22:57:59

CodeWang_NC · 2026-04-08 22:57:59 发布

OpenAI 兼容 ASR 接口调用实战：SDK 与 HTTP 方式详解

本文基于通义千问 ASR 模型（Qwen3-ASR），总结 OpenAI 兼容接口的多种调用方式，包括本地 vLLM 部署和阿里云官方接口，涵盖 SDK 和 HTTP两种协议，以及单音频识别、批量识别、热词支持等核心特性。

github： https://github.com/QwenLM/Qwen3-ASR

一、模型部署方式

1.1 本地 vLLM 部署

通义千问 ASR 模型可通过 vLLM 框架本地部署，提供 OpenAI 兼容接口：

# docker 启动命令
docker run --gpus all --name qwen3-asr_official \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -p 8000:80 \
    --mount type=bind,source=/opt/data/LLM/Qwen3-ASR-0.6B,target=/data/shared/Qwen3-ASR \
    --mount type=bind,source=/宿主机路径/start_asr.sh,target=/start_asr.sh \
    -d qwenllm/qwen3-asr:latest /bin/bash /start_asr.sh

# 在官方的docker内启动 /start_asr.sh 内容
CUDA_VISIBLE_DEVICES=3 python -m qwen_asr.serve.api_server \
/data/shared/Qwen3-ASR \
--gpu-memory-utilization 0.8 \
--host 0.0.0.0 \
--port 80

# 部署示例（具体参数根据服务器配置调整）

部署后可访问以下端点：

/v1/chat/completions - 对话式 ASR（支持批量、热词）
/v1/audio/transcriptions - 标准语音转写接口

1.2 阿里云百炼平台

阿里云百炼平台提供托管服务，无需本地部署：

部署方式	服务地址	模型名称
本地 vLLM	`http://{IP}:8872/v1`	`/data/shared/Qwen3-ASR`
阿里云北京	`https://dashscope.aliyuncs.com/compatible-mode/v1`	`qwen3-asr-flash`
阿里云新加坡	`https://dashscope-intl.aliyuncs.com/compatible-mode/v1`	`qwen3-asr-flash`

二、调用方式对比总览

2.1 本地部署接口对比

方式	接口	音频格式	Batch	长音频	热词	推荐场景
HTTP + completions	`/v1/chat/completions`	base64、url	✅	✅ 需要分片	✅	生产环境，灵活控制
SDK + completions	`chat.completions.create()`	base64	✅	❓	❓	Python 快速集成
HTTP + transcriptions	`/v1/audio/transcriptions`	wav 文件	❌	❓	❓	标准接口兼容
SDK + transcriptions	`audio.transcriptions.create()`	wav 文件	❌	❓	✅	标准接口兼容

2.2 阿里云官方接口对比

方式	接口	音频格式	Batch	长音频上限	热词	特点
OpenAI 兼容 SDK	`/compatible-mode/v1/chat/completions`	base64、url	❌	~3min	❓	速度较慢
OpenAI 兼容 HTTP	`/compatible-mode/v1/chat/completions`	base64、url	❌	~3min	❓	速度较快
异步转写接口	`/services/audio/asr/transcription`	url	❌	更长音频	✅	异步处理，支持文件转写

三、本地 vLLM 部署调用方式

3.1 HTTP 原生调用 - completions 接口

文件: openai_completions_http.py
长音频调用方法

import requests
import base64
import time
import json

def test_openai_completeions_http():
    """
    支持 base64 和 url格式的音频
    """
    url = "http://10.2.5.121:8872/v1/chat/completions"
    # url = "http://10.2.5.121:8871/v1/chat/completions"
    # url = "http://10.2.5.121:17800/v1/chat/completions"
    headers = {"Content-Type": "application/json"}

    # 读取本地音频文件并转为 base64
    audio_path = "audio_data/01.wav"
    with open(audio_path, "rb") as f:
        audio_base64 = base64.b64encode(f.read()).decode("utf-8")

    data = {
        "messages": [
            {
                "role": "system",
                "content": [
                    {
                        "type": "text",
                        "text": "asr上下文【这里可以放你想要识别的热词】 你是一个语音识别助手，擅长技术会议录音转写。"
                                "请特别关注以下术语与名称，识别时优先使用以下写法：\n"
                                "- 比值: 1:0.24\n"
                                "- 人名：王二、李四、张三\n"
                                "- 英文单词：matlab、modelsim、\n\n"
                                "- 数字: 1024, 2048, 9361 \n\n"
                                "遇到类似发音的词时，优先匹配上述术语，避免写成同音字。数字识别为阿拉伯数字"
                    }
                ],
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "audio_url",
                        "audio_url": {
                            "url": f"data:audio/wav;base64,{audio_base64}"
                            # # "url": f"http://10.2.5.121:17800/02.wav"
                        },
                    },
                ],
            },
        ],
        # "temperature": 0.1
    }

    start_time = time.time()
    # openai_调用兼容的热词方式
    response = requests.post(url, headers=headers, json=data, timeout=30)
    response.raise_for_status()
    content = response.json()['choices'][0]['message']['content']
    cost_time = time.time() - start_time
    print(content)
    print(f"cost_time: {cost_time:.2f}s")

def test_openai_completeions_http_batch():
    """
    支持 base64 和 url格式的音频
    content 可以放多段音频然后来识别 多个音频一起识别， 可以用于长音频拆分之后分块识别
    """
    url = "http://10.2.5.121:8872/v1/chat/completions"
    headers = {"Content-Type": "application/json"}
    # 读取本地音频文件并转为 base64
    audio_path = "audio_data/01.wav"
    with open(audio_path, "rb") as f:
        audio_base64 = base64.b64encode(f.read()).decode("utf-8")

    data = {
        "messages": [
            {
                "role": "system",
                "content": [
                    {
                        "type": "text",
                        "text": "上下文【这里可以放你想要识别的热词】"
                    }
                ],
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "audio_url",
                        "audio_url": {
                            "url": f"data:audio/wav;base64,{audio_base64}"
                            # # "url": f"http://10.2.5.121:17800/02.wav"
                        },
                    },
                    {
                        "type": "audio_url",
                        "audio_url": {
                            "url": f"data:audio/wav;base64,{audio_base64}"
                            # # "url": f"http://10.2.5.121:17800/02.wav"
                        },
                    },
                ],
            },
        ],
        # "temperature": 0.1
    }

    start_time = time.time()
    # openai_调用兼容的热词方式
    response = requests.post(url, headers=headers, json=data, timeout=30)
    response.raise_for_status()
    content = response.json()['choices'][0]['message']['content']
    cost_time = time.time() - start_time
    print(content)
    print(f"cost_time: {cost_time:.2f}s")

# 测试qwen_asr 通过vllm部署的openai的v1/chat/completions接口测试
# 支持base64 格式， 支持url格式 和minio的url格式
if __name__ == "__main__":
    test_openai_completeions_http()
    test_openai_completeions_http_batch()

特点：

✅ 支持 base64 编码和 URL 格式
✅ 支持热词（通过 system prompt）
✅ 支持批量识别（content 数组添加多个 audio_url）
✅ 支持长音频，单个最多支持两分钟左右的音频，时长比较长的音频可以通过分块批量传入

3.2 OpenAI SDK 调用 - completions 接口

文件: 04_openai_completions_sdk.py

import base64
import time
from openai import OpenAI


def test_openai_completeions_sdk():

    method_start_time = time.time()
    client = OpenAI(api_key="dummy", base_url="http://10.2.5.121:8872/v1")
    # 读取并转 Base64
    audio_path = "audio_data/01.wav"
    with open(audio_path, "rb") as f:
        audio_base64 = base64.b64encode(f.read()).decode("utf-8")

    # 用 chat 接口传 Base64（不是 audio.transcriptions）
    completion = client.chat.completions.create(
        model="/data/shared/Qwen3-ASR",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "转写这段音频"},
                {
                    "type": "input_audio",
                    "input_audio": {"data": audio_base64, "format": "wav"}
                }
            ]
        }],
        temperature=0
    )

    # 他这个就可以支持batch
    method_cost_time = time.time() - method_start_time
    print(completion.choices[0].message.content)
    print(f"消耗时间: {method_cost_time:.2f}s")

def test_openai_completeions_sdk_batch():
    
    method_start_time = time.time()
    client = OpenAI(api_key="dummy", base_url="http://10.2.5.121:8872/v1")
    # 读取并转 Base64
    audio_path = "audio_data/1min.wav"
    
    with open(audio_path, "rb") as f:
        audio_base64 = base64.b64encode(f.read()).decode("utf-8")

    # 用 chat 接口传 Base64（不是 audio.transcriptions）
    completion = client.chat.completions.create(
        model="/data/shared/Qwen3-ASR",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "转写这段音频"},
                {
                    "type": "input_audio",
                    "input_audio": {"data": audio_base64, "format": "wav"}
                },
                {
                    "type": "input_audio",
                    "input_audio": {"data": audio_base64, "format": "wav"}
                }
            ]
        }],
        temperature=0
    )

    # 他这个就可以支持batch
    method_cost_time = time.time() - method_start_time
    print(completion.choices[0].message.content)
    print(f"消耗时间: {method_cost_time:.2f}s")

if __name__ == "__main__":
    test_openai_completeions_sdk()
    test_openai_completeions_sdk_batch()

3.3 标准语音转写接口

文件: 05_openai_asr_transcription_http.py

import time
import requests
from openai import OpenAI

def test_openai_asr_transcription_sdk():
    method_start_time = time.time()
    client = OpenAI(
        api_key="dummy",                      # 本地服务随便填
        base_url="http://10.2.5.121:8872/v1"  # 改成你自己的地址
    )

    # 直接打开文件调用
    audio_path = "audio_data/2min.wav"
    f = open(audio_path, "rb")

    transcription = client.audio.transcriptions.create(
        model="/data/shared/Qwen3-ASR",
        file=f,
        language="zh",
        prompt="专业术语：AI、大模型、ASR"
    )

    f.close()

    method_cost_time = time.time() - method_start_time
    print(transcription.text)
    print(f"asr耗时: {method_cost_time:.2f}s")


def test_openai_asr_transcription_http():
    method_start_time = time.time()
    url = "http://10.2.5.121:8872/v1/audio/transcriptions"
    API_BASE = "http://10.2.5.121:8872"
    API_KEY = "sk-xxx"
    # AUDIO_FILE = "test.wav"
    AUDIO_FILE = "audio_data/1min.wav"
    try:
        response = requests.post(
            url=f"{API_BASE}/v1/audio/transcriptions",
            headers={"Authorization": f"Bearer {API_KEY}"} if API_KEY else {},
            files={
                "file": open(AUDIO_FILE, "rb"),
                "model": (None, "/data/shared/Qwen3-ASR"),
            },
            data={
                "language": "zh",
                "response_format": "json",
            },
            timeout=30,
        )
        method_cost_time = time.time() - method_start_time
        print(f"asr耗时: {method_cost_time:.2f}s")
        if response.ok:
            print("识别结果：", response.json()["text"])
        else:
            print(f"错误 {response.status_code}: {response.text}")

    except Exception as e:
        print("请求异常：", e)

if __name__ == "__main__":
    test_openai_asr_transcription_http()
    test_openai_asr_transcription_sdk()

四、阿里云百炼平台Qwen_ASR调用方式

4.1 OpenAI 兼容接口 - SDK

文件: 01_openai_asr_officical_aliyun.py

import os
import base64
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

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

# 本地音频转 base64
AUDIO_FILE = "audio_data/1min.wav"
with open(AUDIO_FILE, "rb") as f:
    base64_data = base64.b64encode(f.read()).decode("utf-8")
base64_audio_data = f"data:audio/wav;base64,{base64_data}"

# SDK 调用（也支持直接传 URL）
completion = client.chat.completions.create(
    model="qwen3-asr-flash",
    messages=[
        {
            "content": [
                {
                    "type": "input_audio",
                    "input_audio": {
                        "data": base64_audio_data,  # 或直接传 URL
                        "format": "wav"
                    }
                }
            ],
            "role": "user"
        }
    ],
    stream=False,
    extra_body={
        "asr_options": {
            "enable_itn": False  # 是否启用逆文本标准化
        }
    }
)

print(completion.choices[0].message.content)

注意事项：

⚠️ content 中只支持单个音频，不支持 batch
⚠️ 音频时长上限约 3 分钟，4 分钟以上无法识别
✅ 支持 base64 和 URL 两种传入方式

4.2 OpenAI 兼容接口 - HTTP

import os
import requests
import base64

DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
headers = {
    "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
    "Content-Type": "application/json"
}

# 本地音频转 base64
with open("audio_data/3min.wav", "rb") as f:
    base64_data = base64.b64encode(f.read()).decode("utf-8")
base64_audio_data = f"data:audio/wav;base64,{base64_data}"

payload = {
    "model": "qwen3-asr-flash",
    "messages": [
        {
            "content": [
                {
                    "type": "input_audio",
                    "input_audio": {"data": base64_audio_data}
                }
            ],
            "role": "user"
        }
    ],
    "stream": False,
    "asr_options": {"enable_itn": False}
}

response = requests.post(url, headers=headers, json=payload)
print(response.json()["choices"][0]["message"]["content"])

注意事项：

⚠️ content 中只支持单个音频，不支持 batch
⚠️ 音频时长上限约 3 分钟，4 分钟以上无法识别
✅ 支持 base64 和 URL 两种传入方式

4.3 阿里云官方异步转写接口

文件: 02_qwen_asr_official_aliyun_transcription.py

适用于较长音频的异步处理，支持文件转写模型 qwen3-asr-flash-filetrans。

import os
import json
import requests

from dotenv         import load_dotenv
load_dotenv()

def get_asr_transcription_aliyun():
    """
    调用这个接口 阿里云官方的qwen_asr语音识别接口, audio/asr/transcription
    https://help.aliyun.com/zh/model-studio/qwen-asr-api-reference?spm=a2c4g.11186623.help-menu-2400256.d_2_7_7.7f0f707f9WrsaZ&scm=20140722.H_2986952._.OR_help-T_cn~zh-V_1#3e5fac1af9j1v
    """
    # 这个没有用
    # 以下为北京地域url，若使用新加坡地域的模型，需将url替换为：https://dashscope-intl.aliyuncs.com/api/v1/services/audio/asr/transcription
    url = "https://dashscope.aliyuncs.com/api/v1/services/audio/asr/transcription"

    # 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
    # 若没有配置环境变量，请用百炼API Key将下行替换为：DASHSCOPE_API_KEY = "sk-xxx"
    DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")

    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json",
        "X-DashScope-Async": "enable"
    }

    payload = {
        "model": "qwen3-asr-flash-filetrans",
        "input": {
            # "file_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3"
            "file_url": "audio_data/01.wav"
        },
        "parameters": {
            "channel_id": [0],
            # "language": "zh",
            "enable_itn": False
            # "corpus": {
            #     "text": ""
            # }
        }
    }

    response = requests.post(url, headers=headers, data=json.dumps(payload))
    if response.status_code == 200:
        task_id = response.json()["output"]["task_id"]
        print(f"task_id: {task_id}")
        return task_id
    else:
        print("task failed!")
        print(response.json())

def get_asr_text_by_task_id(task_id):
    """
    当你获取task_id之后你需要在调用一个url才能获取到。 json地址
    再调用一次json的url地址获取asr_text 识别内容
    """
    # 新加坡地域和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
    # 若没有配置环境变量，请用百炼API Key将下行替换为：DASHSCOPE_API_KEY = "sk-xxx"
    DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")

    # 替换为实际的task_id
    # 以下为北京地域url，若使用新加坡地域的模型，需将url替换为：https://dashscope-intl.aliyuncs.com/api/v1/tasks/{task_id}，注意，将{task_id}替换为待查询任务ID
    url = f"https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}"

    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "X-DashScope-Async": "enable",
        "Content-Type": "application/json"
    }

    response = requests.get(url, headers=headers)
    download_url = response.json()["output"]["result"]["transcription_url"]
    print(f"download_url: {download_url}")

    # 获取具体的json
    # case1: 保存到本地
    import urllib.request
    urllib.request.urlretrieve(download_url, "result.json")

    # case2: 直接读取 后续用
    response = requests.get(download_url)
    result_dict = response.json()  # ✅ 这就是你要的字典变量！
    asr_text = result_dict["transcripts"][0]["text"]
    print(f"完成 asr识别文本: {asr_text}")
    
    return asr_text

# 调用阿里云官方提供的接口 qwen_asr 异步调用接口 asr
"""
1. 先通过传入音频url文件获取task_id
2. 通过task_id 获取json对应的url
3. 获取json的到最终的asr识别文本
"""
if __name__ == "__main__":
    task_id =  get_asr_transcription_aliyun()
    asr_text = get_asr_text_by_task_id(task_id)
    print(f"完成 asr识别文本: {asr_text}")

特点：

✅ 支持更长的音频文件
✅ 异步处理，不阻塞主线程
✅ 支持热词（corpus 参数）
⚠️ 需要音频文件已上传到 OSS 或提供可访问的 URL

五、统一封装客户端代码

文件: 07_opanai_client.py

以下封装类支持本地 vLLM 部署的多种调用方式：

import requests
import base64
import time
import json
from openai import OpenAI


class QwenASRClient:
    """
    通义千问 ASR 统一调用类
    支持：HTTP原生请求 / OpenAI SDK / 标准语音转写接口
    支持：单音频识别、批量音频识别、Base64/URL音频
    """

    def __init__(self, base_url="http://x.x.x.x:8000/v1", api_key="dummy", model="/data/shared/Qwen3-ASR"):
        """
        初始化ASR客户端
        :param base_url: 服务地址
        :param api_key: API密钥（本地服务可填dummy）
        :param model: 模型路径
        """
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        # 拼接基础HTTP接口地址
        self.chat_http_url = f"{base_url}/chat/completions"
        self.trans_http_url = f"{base_url}/audio/transcriptions"
        # 初始化OpenAI客户端
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def _audio_to_base64(self, audio_path):
        """工具方法：本地音频文件转base64"""
        with open(audio_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")

    # ==================== 1. HTTP原生 - 单音频识别 ====================
    def completions_http_single_asr(self, audio_path, system_prompt=None):
        """
        HTTP方式： /chat/completions 单音频ASR识别（支持热词）
        1. 支持batch, 
        2. 支持url, base64
        3. 长音频需要分片，单个audio_url最多支持2min
        4. 支持热词，放在text在中
        """
        # 默认系统提示词
        if system_prompt is None:
            system_prompt = (
                "asr上下文【热词】你是一个语音识别助手，擅长技术会议录音转写。"
                "请特别关注以下术语与名称，识别时优先使用以下写法：\n"
                "- 比值: 1:0.24\n"
                "- 人名：王二、李四、张三\n"
                "- 英文单词：matlab、modelsim\n"
                "- 数字: 1024, 2048, 9361 \n"
                "遇到类似发音的词时，优先匹配上述术语，避免写成同音字。数字识别为阿拉伯数字"
            )

        audio_base64 = self._audio_to_base64(audio_path)
        headers = {"Content-Type": "application/json"}
        data = {
            "messages": [
                {
                    "role": "system",
                    "content": [{"type": "text", "text": system_prompt}]
                },
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "audio_url",
                            "audio_url": {"url": f"data:audio/wav;base64,{audio_base64}"}
                        }
                    ]
                }
            ]
        }
        start_time = time.time()
        response = requests.post(self.chat_http_url, headers=headers, json=data, timeout=60)
        response.raise_for_status()
        result = response.json()['choices'][0]['message']['content']
        cost = round(time.time() - start_time, 2)

        print("=== HTTP 单音频识别结果 ===")
        print(result)
        print(f"耗时：{cost}s\n")
        return result, cost

    # ==================== 2. HTTP原生 - 批量音频识别 ====================
    def completions_http_batch_asr(self, audio_path, batch_count=2):
        """
        HTTP方式：/chat/completions 批量音频识别（同一音频重复识别演示） 
        1. 支持batch, 
        2. 支持url, base64
        3. 长音频需要分片，单个audio_url最多支持2min
        4. 支持热词，放在text在中
        """
        audio_base64 = self._audio_to_base64(audio_path)
        headers = {"Content-Type": "application/json"}

        # 批量构造音频内容
        data = {
            "messages": [
                {
                    "role": "system",
                    "content": [{"type": "text", "text": "上下文【这里可以放你想要识别的热词】"}]
                },
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "audio_url",
                            "audio_url": {
                                "url": f"data:audio/wav;base64,{audio_base64}"
                                # # "url": f"http://10.2.5.121:17800/02.wav"
                            },
                        },
                        {
                            "type": "audio_url",
                            "audio_url": {
                                "url": f"data:audio/wav;base64,{audio_base64}"
                                # # "url": f"http://10.2.5.121:17800/02.wav"
                            },
                        },
                    ],
                }
            ]
        }

        start_time = time.time()
        response = requests.post(self.chat_http_url, headers=headers, json=data, timeout=60)
        response.raise_for_status()
        result = response.json()['choices'][0]['message']['content']
        cost = round(time.time() - start_time, 2)

        print("=== HTTP 批量音频识别结果 ===")
        print(result)
        print(f"耗时：{cost}s\n")
        return result, cost

    # ==================== 3. OpenAI SDK - 单音频识别 ====================
    def completions_sdk_single_asr(self, audio_path, prompt="转写这段音频"):
        """
        OpenAI SDK：单音频识别 sdk格式
        1. 支持音频格式wav
        2. 支持长音频。    (不确定)
        3. 支持batch
        4. 是否支持热词调用(不确定)
        """
        start_time = time.time()
        audio_base64 = self._audio_to_base64(audio_path)

        completion = self.client.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "input_audio", "input_audio": {"data": audio_base64, "format": "wav"}}
                ]
            }],
            temperature=0
        )

        result = completion.choices[0].message.content
        cost = round(time.time() - start_time, 2)

        print("=== SDK 单音频识别结果 ===")
        print(result)
        print(f"耗时：{cost}s\n")
        return result, cost

    # ==================== 4. OpenAI SDK - 批量音频识别 ====================
    def completions_sdk_batch_asr(self, audio_path, batch_count=2, prompt="转写这段音频"):
        """
        OpenAI SDK：批量音频识别
        1. 支持音频格式wav
        2. 是否支持长音频   (不确定)
        3. 支持batch
        4. 是否支持热词调用 (不确定)
        """
        start_time = time.time()
        audio_base64 = self._audio_to_base64(audio_path)

        # 批量音频
        completion = self.client.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "input_audio", 
                        "input_audio": {"data": audio_base64, "format": "wav"}
                    },
                    {
                        "type": "input_audio", 
                        "input_audio": {"data": audio_base64, "format": "wav"}
                    }
                ]
            }],
            temperature=0
        )

        result = completion.choices[0].message.content
        cost = round(time.time() - start_time, 2)

        print("=== SDK 批量音频识别结果 ===")
        print(result)
        print(f"耗时：{cost}s\n")
        return result, cost

    # ==================== 5. 标准转写接口 - SDK ====================
    def transcription_sdk(self, audio_path, language="zh", prompt="专业术语：AI、大模型、ASR"):
        """
        标准语音转写 API（SDK方式）
        1. 支持识别的语音格式 wav
        2. 不支持batch
        3. 是否支持长音频   (不确定)
        4. 是否支持热词调用 (不确定)
        """
        start_time = time.time()
        audio_path = "audio_data/2min.wav"
        file_path = open(audio_path, "rb")
        # with open(audio_path, "rb") as f:
        transcription = self.client.audio.transcriptions.create(
            model=self.model,
            file=file_path,
            language=language,
            prompt=prompt
        )
        result = transcription.text
        cost = round(time.time() - start_time, 2)

        print("=== 标准转写 SDK 结果 ===")
        print(result)
        print(f"耗时：{cost}s\n")
        return result, cost

    # ==================== 6. 标准转写接口 - HTTP ====================
    def transcription_http(self, audio_path, language="zh"):
        """
        标准语音转写 API（HTTP方式）
        1. 支持识别的语音格式 wav
        2. 不支持batch
        3. 是否支持长音频   (不确定)
        4. 是否支持热词调用 (不确定)
        """
        start_time = time.time()
        try:
            # with open(audio_path, "rb") as f:
            file_path = open(audio_path, "rb")
            response = requests.post(
                url=self.trans_http_url,
                headers={"Authorization": f"Bearer {self.api_key}"} if self.api_key else {},
                files={
                    "file": file_path,
                    "model": (None, self.model),
                },
                data={
                    "language": language,
                    "response_format": "json",
                },
                timeout=60
            )

            cost = round(time.time() - start_time, 2)
            if response.ok:
                result = response.json()["text"]
                print("=== 标准转写 HTTP 结果 ===")
                print(result)
                print(f"耗时：{cost}s\n")
                return result, cost
            else:
                print(f"请求失败：{response.status_code} - {response.text}")
                return None, cost

        except Exception as e:
            print(f"请求异常：{e}")
            return None, 0


# ==================== 使用示例 ====================
if __name__ == "__main__":
    # 1. 初始化客户端
    asr_client = QwenASRClient(
        base_url="http://x.x.x.x:8000/v1",
        api_key="dummy",
        model="/data/shared/Qwen3-ASR"
    )

    # 音频文件路径（根据自己的路径修改）
    AUDIO_1 = "audio_data/01.wav"
    AUDIO_2 = "audio_data/1min.wav"
    AUDIO_3 = "audio_data/2min.wav"

    # 2. 调用所有识别方法
    asr_client.completions_http_single_asr(AUDIO_1)          # HTTP单音频
    asr_client.completions_http_batch_asr(AUDIO_1)           # HTTP批量
    asr_client.completions_sdk_single_asr(AUDIO_1)           # SDK单音频
    asr_client.completions_sdk_batch_asr(AUDIO_2)            # SDK批量
    asr_client.transcription_sdk(AUDIO_2)                    # 标准转写SDK
    asr_client.transcription_http(AUDIO_2)                   # 标准转写HTTP

六、核心特性总结

6.1 支持的音频格式

格式	本地 completions	本地 transcriptions	阿里云 OpenAI 兼容	阿里云异步转写
base64 编码	✅	❌	✅	❌
WAV 文件	❌	✅	❌	❌
URL 形式	✅	❌	✅	✅（必须）
PCM 数据	❓	❓	❓	❌

说明：

completions 接口需要 base64 编码或 URL
transcriptions 接口直接传文件对象
阿里云异步转写必须提供 OSS URL

6.2 Batch 支持

方式	是否支持	说明
本地 HTTP completions	✅	content 数组添加多个 audio_url
本地 SDK completions	✅	content 数组添加多个 input_audio
本地 transcriptions	❌	单文件接口
阿里云 OpenAI 兼容	❌	content 只支持单个音频
阿里云异步转写	❌	单文件处理

应用场景：长音频切分后批量识别，减少网络往返开销（仅本地部署支持）。

6.3 长音频支持

方式	上限	说明
本地 completions	~2min	单个音频建议 ≤ 2 分钟，更长需分片
本地 transcriptions	❓	待验证上限
阿里云 OpenAI 兼容	~3min	实测 3min 可识别，4min 失败
阿里云异步转写	更长	支持文件转写模型，适合长音频

6.4 热词支持

方式	热词方式	示例
本地 HTTP completions	✅ system prompt	`{"type": "text", "text": "专业术语：matlab"}`
本地 SDK completions	❓	待验证
本地 transcriptions	✅ prompt 参数	`prompt="专业术语：AI"`
阿里云 OpenAI 兼容	❓	待验证（无显式参数）
阿里云异步转写	✅ corpus 参数	`parameters.corpus.text`

七、选型建议

场景	推荐方式	理由
本地部署生产环境	HTTP completions	灵活控制，支持热词和批量
本地 Python 快速集成	SDK completions	代码简洁，SDK 封装完善
无本地 GPU 资源	阿里云 OpenAI 兼容	无需部署，API 调用
长音频转写（>3min）	阿里云异步转写	支持更长音频，异步处理
OpenAI 兼容迁移	transcriptions 接口	标准接口，零修改迁移

八、注意事项

音频时长限制：不同接口上限不同，超长音频需预处理或使用异步转写
Base64 编码：completions 接口需要编码，增加约 33% 体积
URL 格式：需要服务端能访问该 URL，阿里云异步转写必须用 OSS URL
超时设置：长音频识别耗时较长，建议设置 60s+ 超时
模型路径：本地部署需传入完整模型路径，阿里云使用模型名称
API Key：阿里云需配置 DASHSCOPE_API_KEY 环境变量

九、相关文件

文件	说明
`01_openai_asr_officical_aliyun.py`	阿里云 OpenAI 兼容接口（SDK/HTTP）
`02_qwen_asr_official_aliyun_transcription.py`	阿里云官方异步转写接口
`03_openai_completions_http.py`	本地 HTTP completions 接口
`04_openai_completions_sdk.py`	本地 SDK completions 接口
`05_openai_asr_transcription_http.py`	本地标准转写接口
`07_opanai_client.py`	统一封装客户端

声明: 本人自用笔记如有错误，请指正交流

参考链接：

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

软件工程原则在多智能体系统中的应用：分层与解耦

ChatGPT 发布之后，AI 智能体的概念就一直牵动着整个行业的想象力。它描绘的场景很诱人：给 AI 系统一个目标，让它自行拆解问题、调用工具、收集信息，最终综合出结果。围绕这个概念的框架生态已经相当拥挤了：LangChain、CrewAI、AutoGen、Semantic Kernel、Agent Framework……新框架层出不穷，个个声称能简化智能应用的构建。但大多数还停留在 hello