Qwen-Audio-TTS Python SDK

本文介绍通过DashScope Python SDK进行Qwen-Audio-TTS实时语音合成的类定义、请求参数和示例代码。

接口地址

SDK的接口地址由Client构造参数workspace/region决定（对应下方地址，包含WorkspaceId）。如需切换到其他地域，请修改Client构造参数为对应地域的值。

核心资源（client.audio.tts）

资源路径：client.audio.tts

功能：基于WebSocket的语音合成，提供一次性合成（synthesize）、流式合成（stream）与双向流式会话（connect）三种调用形态。

调用入口

# 一次性合成：返回完整音频 bytes
audio = client.audio.tts.synthesize(
    text,                      # text 必填（位置参数）
    model="qwen-audio-3.0-tts-flash", # 音色所属模型
    voice="longanhuan_v3.6",   # 音色
    format="mp3",              # 音频编码格式，默认 "mp3"
    sample_rate=22050,         # 采样率，默认 22050
)

# 流式合成：逐块产出音频 bytes（边收边写/播）
for chunk in client.audio.tts.stream(text, model=..., voice=...):
    ...

# 双向流式会话：逐段输入文本，进入 with 上下文时自动建立连接
with client.audio.tts.connect(model=..., voice=...) as session:
    session.send_text(text)    # 可多次调用追加文本
    session.stop()             # 通知服务端输入结束
    for chunk in session.audio_chunks():
        ...

synthesize() - 非流式调用

方法签名：

def synthesize(self, text: str, *, model: str,
               voice: str, format: str = "mp3",
               sample_rate: int = 22050, volume: int = None,
               rate: float = None, pitch: float = None,
               language_hints: list = None, enable_ssml: bool = None,
               parameters: Dict = None, ...) -> bytes

参数说明：

返回值：bytes，完整音频数据。

说明：非流式调用，阻塞等待并一次性返回完整音频数据。适用于短文本、对实时性无严格要求的场景。synthesize与stream每次调用自建会话，无需复用实例。

stream() - 流式调用

方法签名：

def stream(self, text: str, *, model: str, voice: str,
           format: str = "mp3", sample_rate: int = 22050, ...) -> Iterator[bytes]

参数说明：

返回值：Iterator[bytes]，逐块产出的音频数据。

说明：流式合成，一次性发送完整文本并逐块产出音频（Iterator[bytes]）。适用于对实时性要求高的短文本语音合成场景。需要分片提交文本时请使用双向流式会话（connect）。

send_text() - 流式输入（会话方法）

双向流式会话（client.audio.tts.connect(...)）中分片提交文本。

方法签名：

def send_text(self, text: str) -> None

参数说明：text为当前待合成的文本片段。可多次调用send_text以追加文本，单次不超过20000字符，累计不超过20万字符。发送文本与接收音频需并发进行（配合audio_chunks()/events()和stop使用）。

stop() - 结束流式输入（会话方法）

方法签名：

def stop(self) -> None

说明：通知服务端所有文本已发送完毕。调用后不阻塞当前线程，接收线程需继续消费audio_chunks()/events()直到剩余文本合成完成并返回所有音频数据。未调用此方法可能导致尾部文本无法转换为语音。

cancel() - 取消流式合成（会话方法）

方法签名：

def cancel(self) -> None

说明：取消当前轮次的流式语音合成任务。调用后，SDK 会立即结束当前任务。取消后当前会话不能再发送文本，如需继续合成请新建会话。

会话指标与响应（会话属性）

session.first_package_delay
session.usage
session.last_response

• first_package_delay：float，从发送文本到收到第一块音频数据的延迟时间（毫秒）。需在合成完成后读取。
• usage：dict，最近一次合成任务的用量信息。
• last_response：dict，最近一次合成任务的响应消息（含 header/payload 键）。最近一次请求的request_id可从session.last_response的header中获取，可用于问题排查和日志关联。

请求参数

以下参数通过client.audio.tts的synthesize/stream/connect方法设置，用于控制合成音频的模型、音色、格式和音频特征（model、voice、format、sample_rate、volume、rate、pitch、language_hints、enable_ssml为具名参数，其余参数通过parameters字典传入）。

audio = client.audio.tts.synthesize(
    "待合成文本",
    model="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6",
    parameters={"bit_rate": 128},
)

audio = client.audio.tts.stream(
    "待合成文本",
    model="qwen-audio-3.0-tts-flash",
    voice="your_voice",  # 支持字级别时间戳的系统音色或复刻音色
    parameters={"word_timestamp_enabled": True},
)

audio = client.audio.tts.synthesize(
    "待合成文本",
    model="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6",
    parameters={
        "enable_aigc_tag": True,
        "aigc_propagator": "your_propagator",
        "aigc_propagate_id": "your_propagate_id"
    },
)

audio = client.audio.tts.synthesize(
    "待合成文本",
    model="qwen-audio-3.0-tts-flash",
    voice="longanhuan_v3.6", # 音色
    parameters={
        "hot_fix": {
            "pronunciation": [{"天气": "tian1 qi4"}],
            "replace": [{"今天": "金天"}]
        }
    },
)

会话与事件迭代

流式合成不再使用回调接口：音频与服务端事件通过会话的迭代方法接收。stream()/synthesize()无需会话管理；双向流式会话（client.audio.tts.connect(...)）的生命周期为：进入with上下文自动建立连接 → 多次send_text() → stop() → 读完全部结果 → 退出上下文自动关闭。

audio_chunks() - 迭代音频数据

方法签名：

def audio_chunks(self) -> Iterator[bytes]

返回值：Iterator[bytes]，逐块产出的音频二进制数据片段，格式由format参数指定。

说明：在接收线程中迭代，逐块接收音频数据，合成过程中会产出多块。可将数据写入文件或送入播放设备。同一会话只允许一个events()或audio_chunks()读取者。

events() - 迭代服务端事件

方法签名：

def events(self) -> Generator[Union[Dict, bytes], None, None]

返回值：同时产出服务端响应事件（dict，包含header（请求信息）和payload（输出信息））与音频bytes。其中payload.output包含事件类型、原始文本等信息，详见events()事件中的output字段。

说明：在接收线程中迭代服务端回复，包含合成事件的输出信息（事件类型、原始文本、句子信息等），可直接访问payload.output获取详细信息。

错误处理

合成过程中发生错误时，ws层会以DashScopeError异常抛出，异常触发后连接将自动关闭。可在发送/接收处使用try/except捕获并记录错误日志以便排查问题：

from dashscope.v2 import DashScopeError

try:
    ...
except DashScopeError as e:
    print(e.code, e.message, e.request_id)

会话生命周期

• 进入with上下文时自动建立连接（可在此前初始化音频输出流或打开文件等资源）。
• 所有文本合成完成且音频数据已全部产出后，events()/audio_chunks()迭代结束；此后可读取first_package_delay等性能指标。
• 退出with上下文时自动关闭连接（可在此后释放音频播放设备等资源）。

events()事件中的output字段

会话events()迭代产出的事件dict中，payload.output包含合成事件的输出信息，可用于跟踪合成进度和获取逐句信息。以下为output字段的结构说明：

消息示例：

{
  "header": {
    "task_id": "xxx",
    "event": "result-generated",
    "attributes": {}
  },
  "payload": {
    "output": {
      "type": "sentence-begin",
      "original_text": "今天天气怎么样？",
      "sentence": {
        "index": 0,
        "words": []
      }
    }
  }
}

解析示例：

def handle_event(event):
    output = (event.get('payload') or {}).get('output') or {}
    event_type = output.get('type', '')
    original_text = output.get('original_text', '')
    if event_type:
        print(f'事件类型: {event_type}, 原始文本: {original_text}')

示例代码

SDK提供了语音合成的关键接口，支持以下几种调用方式：

• 非流式调用：阻塞式，一次性发送完整文本，直接返回完整音频。适合短文本语音合成场景。
• 单向流式调用：非阻塞式，一次性发送完整文本，通过迭代stream返回的音频分片接收数据。适用于对实时性要求高的短文本语音合成场景。
• 双向流式调用：非阻塞式，可分多次发送文本片段，通过会话的audio_chunks()/events()迭代实时接收增量合成的音频流。适合实时性要求高的长文本语音合成场景。

更多示例，请参见GitHub。

非流式调用

单次调用发送的文本长度不得超过20000字符，超出限制将返回错误。

重要synthesize与stream每次调用自建会话，无需复用实例。

# coding=utf-8

import os

import dashscope

# 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
# 以下为华北2（北京）地域的配置，调用时请将"{WorkspaceId}"替换为真实的业务空间ID，各地域的配置不同。
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# 模型
model = "qwen-audio-3.0-tts-flash"
# 音色
voice = "longanhuan_v3.6"

# 发送待合成文本，获取二进制音频
audio = client.audio.tts.synthesize("今天天气怎么样？", model=model, voice=voice)

# 将音频保存至本地
with open('output.mp3', 'wb') as f:
    f.write(audio)

dashscope CLI

export DASHSCOPE_API_KEY="your-api-key"
# 将 {WorkspaceId} 替换为业务空间ID，cn-beijing 替换为对应地域（新加坡: ap-southeast-1, 美东: us-east-1）
export DASHSCOPE_HTTP_BASE_URL="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"
dashscope speech-synthesis create -m qwen-audio-3.0-tts-flash -t "你好世界" --voice longanhuan_v3.6

重要SDK Expert 交互式助手可按自然语言完成同样的开发与排障，见

DashScope SDK Expert。

完整地域表见 Base URL 总览。

单向流式调用

单次调用发送的文本长度不得超过20000字符，超出限制将返回错误。

重要synthesize与stream每次调用自建会话，无需复用实例。

# coding=utf-8

import os

from datetime import datetime

import dashscope

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
# 以下为华北2（北京）地域的配置，调用时请将"{WorkspaceId}"替换为真实的业务空间ID，各地域的配置不同。
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# 模型
model = "qwen-audio-3.0-tts-flash"
# 音色
voice = "longanhuan_v3.6"

# 一次性发送待合成文本，逐块接收二进制音频
with open("output.mp3", "wb") as f:
    for chunk in client.audio.tts.stream("今天天气怎么样？", model=model, voice=voice):
        print(get_timestamp() + " 二进制音频长度为：" + str(len(chunk)))
        f.write(chunk)
print("语音合成完成，所有合成结果已被接收：" + get_timestamp())

双向流式调用

单次发送文本长度不得超过 20000 字符，且累计发送文本总长度不得超过 20 万字符。

• 流式输入时可多次调用会话的send_text按顺序提交文本片段。服务端接收文本片段后自动进行分句：

  • 完整语句立即合成
  • 不完整语句缓存至完整后合成

  调用 stop 时，服务端会强制合成所有已接收但未处理的文本片段（包括未完成的句子）。

• 发送文本片段的间隔不得超过23秒，否则触发“request timeout after 23 seconds”异常。

  若无待发送文本，需及时调用 stop结束任务。

  重要请务必确保调用stop方法，否则可能会导致结尾部分的文本无法成功转换为语音。

  服务端强制设定23秒超时机制，客户端无法修改该配置。

# coding=utf-8
#
# pyaudio安装说明：
# 如果是macOS操作系统，执行如下命令：
#   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
import time
import pyaudio
from concurrent.futures import ThreadPoolExecutor

import dashscope

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
# 以下为华北2（北京）地域的配置，调用时请将"{WorkspaceId}"替换为真实的业务空间ID，各地域的配置不同。
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# 模型
model = "qwen-audio-3.0-tts-flash"
# 音色
voice = "longanhuan_v3.6"

test_text = [
    "流式文本语音合成SDK，",
    "可以将输入的文本",
    "合成为语音二进制数据，",
    "相比于非流式语音合成，",
    "流式合成的优势在于实时性",
    "更强。用户在输入文本的同时",
    "可以听到接近同步的语音输出，",
    "极大地提升了交互体验，",
    "减少了用户等待时间。",
    "适用于调用大规模",
    "语言模型（LLM），以",
    "流式输入文本的方式",
    "进行语音合成的场景。",
]

# 创建双向流式合成会话：进入 with 上下文时自动建立连接
with client.audio.tts.connect(
        model=model,
        voice=voice,
        format="pcm",
        sample_rate=22050,
) as session:
    # 初始化播放器
    player = pyaudio.PyAudio()
    stream = player.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)

    # 接收线程：迭代音频数据，实时播放
    def receive_audio():
        for chunk in session.audio_chunks():
            print(get_timestamp() + " 二进制音频长度为：" + str(len(chunk)))
            stream.write(chunk)

    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_audio)
        # 流式发送待合成文本
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # 结束流式语音合成
        session.stop()
        receiver.result()

    print("语音合成完成，所有合成结果已被接收：" + get_timestamp())
    # 停止播放器
    stream.stop_stream()
    stream.close()
    player.terminate()

    # 首次发送文本时需建立 WebSocket 连接，因此首包延迟会包含连接建立的耗时
    print('[Metric] 首包延迟为：{}毫秒'.format(session.first_package_delay))