跳转到主要内容
千问云 home page
开放平台文档
语音合成

实时语音合成

实时流式语音合成

实时语音合成基于 WebSocket 协议将文本实时转换为自然语音。千问云提供 CosyVoice、Qwen-TTS 和 Sambert 系列模型，支持流式输入输出，并提供声音复刻、声音设计及精细化音频控制能力，适用于语音助手、有声读物、智能客服等场景。

核心功能

  • 实时生成高保真语音，支持中英等多语种自然发声
  • 提供声音复刻与声音设计两种音色定制方式
  • 支持流式输入输出，首包延迟低，适用于实时对话场景
  • 可调节语速、语调、音量与码率，精细控制语音表现
  • 兼容主流音频格式（PCM、WAV、MP3、Opus），最高支持48kHz采样率输出
  • 支持指令控制，可通过自然语言指令控制语音表现力（仅Qwen-TTS Instruct系列及部分CosyVoice模型）

适用范围

支持的模型： 调用以下模型时，请使用 API Key：
  • CosyVoice: cosyvoice-v3.5-plus, cosyvoice-v3.5-flash, cosyvoice-v3-plus, cosyvoice-v3-flash, cosyvoice-v2, cosyvoice-v1
  • Qwen-TTS: qwen3-tts-flash-realtime, qwen3-tts-instruct-flash-realtime, qwen3-tts-vd-realtime, qwen3-tts-vc-realtime, qwen-tts-realtime
  • Sambert: 详情请参见 Sambert 模型列表
完整的模型列表和版本信息，请参见语音合成模型列表。

快速开始

在编写代码前，请根据业务场景选择合适的调用方式：
调用方式适用场景流式支持
非流式（同步）批量任务、短文本、生成完整音频文件否
流式输出（单向）对首包延迟敏感的实时应用是
流式输入+输出（双向，WebSocket）对话式AI、LLM语音输出、交互式语音助手是
如需最低延迟，推荐使用流式输出搭配 PCM 格式。PCM 无需编码开销，可直接送入音频设备播放。 下面是调用API的示例代码。更多常用场景的代码示例，请参见 GitHub。 获取 API Key 并设置为环境变量。如需使用 SDK，请先安装 SDK。
  • CosyVoice
  • Qwen-TTS-Realtime
cosyvoice-v3.5-plus 和 cosyvoice-v3.5-flash 模型专门用于声音设计和声音复刻场景（无系统音色）。在使用它们进行语音合成之前，请先参见CosyVoice声音复刻/设计API创建目标音色。创建完成后，只需将代码中的 voice 字段更新为您的音色 ID，并将 model 字段指定为对应模型，即可正常运行。
更多代码示例请参见 GitHub。
  • 使用系统音色进行语音合成
以下示例演示如何使用系统音色（参见CosyVoice音色列表）进行语音合成。如需非实时合成（发送完整文本，接收完整音频），请参见非实时语音合成。

将大模型生成的文本实时转为语音并播放

将 Qwen 模型（qwen3.5-flash）的输出文本实时合成语音，并在本地设备播放。
  • Python
  • Java
运行 Python 示例前，请通过 pip 安装第三方音频播放库。
# coding=utf-8
# pyaudio 安装说明：
# APPLE Mac OS X
#   brew install portaudio
#   pip install pyaudio
# Debian/Ubuntu
#   sudo apt-get install python-pyaudio python3-pyaudio
#   or
#   pip install pyaudio
# CentOS
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# Microsoft Windows
#   python -m pip install pyaudio

import os
import pyaudio
import dashscope
from dashscope.audio.tts_v2 import *


from http import HTTPStatus
from dashscope import Generation

# 如果未配置环境变量，请将下行替换为您的 API key：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope.aliyuncs.com/api-ws/v1/inference'

# cosyvoice-v3-flash/cosyvoice-v3-plus：可选用 longanyang 等音色。
# 每种音色支持的语言不同。合成日语、韩语等非中文语言时，请选择支持相应语言的音色。详见 CosyVoice 音色列表。
model = "cosyvoice-v3-flash"
voice = "longanyang"


class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    print("websocket is open.")
    self._player = pyaudio.PyAudio()
    self._stream = self._player.open(
      format=pyaudio.paInt16, channels=1, rate=22050, output=True
    )

  def on_complete(self):
    print("speech synthesis task complete successfully.")

  def on_error(self, message: str):
    print(f"speech synthesis task failed, {message}")

  def on_close(self):
    print("websocket is closed.")
    # 停止播放
    self._stream.stop_stream()
    self._stream.close()
    self._player.terminate()

  def on_event(self, message):
    print(f"recv speech synthsis message {message}")

  def on_data(self, data: bytes) -> None:
    print("audio result length:", len(data))
    self._stream.write(data)


def synthesizer_with_llm():
  callback = Callback()
  synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    format=AudioFormat.PCM_22050HZ_MONO_16BIT,
    callback=callback,
  )

  messages = [{"role": "user", "content": "Please introduce yourself"}]
  responses = Generation.call(
    model="qwen3.5-flash",
    messages=messages,
    result_format="message",  # 设置返回格式为 message
    stream=True,  # 启用流式输出
    incremental_output=True,  # 启用增量输出
  )
  for response in responses:
    if response.status_code == HTTPStatus.OK:
      print(response.output.choices[0]["message"]["content"], end="")
      synthesizer.streaming_call(response.output.choices[0]["message"]["content"])
    else:
      print(
        "Request id: %s, Status code: %s, error code: %s, error message: %s"
        % (
          response.request_id,
          response.status_code,
          response.code,
          response.message,
        )
      )
  synthesizer.streaming_complete()
  print('requestId: ', synthesizer.get_last_request_id())


if __name__ == "__main__":
  synthesizer_with_llm()

通过回调函数流式接收音频

发送完整文本，通过回调函数增量接收音频数据。适用于短文本场景，可在不阻塞主线程的情况下实现低延迟音频输出。
  • Python
  • Java
# coding=utf-8

import os
import dashscope
from dashscope.audio.tts_v2 import *

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：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope.aliyuncs.com/api-ws/v1/inference'

# 模型
model = "cosyvoice-v3-flash"
# 音色
voice = "longanyang"


# 定义回调接口
class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    self.file = open("output.mp3", "wb")
    print("连接已建立：" + get_timestamp())

  def on_complete(self):
    print("语音合成完成，已接收全部结果：" + get_timestamp())
    # 仅在 on_complete 触发后才可调用 get_first_package_delay
    # 首次请求的首包延迟包含 WebSocket 建连时间
    print('[Metric] requestId: {}, first-package delay: {} ms'.format(
      synthesizer.get_last_request_id(),
      synthesizer.get_first_package_delay()))

  def on_error(self, message: str):
    print(f"语音合成错误：{message}")

  def on_close(self):
    print("连接已关闭：" + get_timestamp())
    self.file.close()

  def on_event(self, message):
    pass

  def on_data(self, data: bytes) -> None:
    print(get_timestamp() + " 音频二进制数据长度：" + str(len(data)))
    self.file.write(data)


callback = Callback()

# 实例化 SpeechSynthesizer，在构造方法中传入 model、voice 等请求参数
synthesizer = SpeechSynthesizer(
  model=model,
  voice=voice,
  callback=callback,
)

# 发送待合成文本，通过回调接口的 on_data 方法实时获取二进制音频
synthesizer.call("How is the weather today?")

流式文本实时合成

增量发送文本片段，通过回调函数实时接收音频数据。这种双向流式方式适用于长文本或与大语言模型集成等文本分段到达的场景。
  • Python
  • Java
# 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
# Windows 系统：
#   python -m pip install pyaudio

import os
import time
import pyaudio
import dashscope
from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
from dashscope.audio.tts_v2 import *

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：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope.aliyuncs.com/api-ws/v1/inference'

# 模型
model = "cosyvoice-v3-flash"
# 音色
voice = "longanyang"


# 定义回调接口
class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    print("连接已建立：" + get_timestamp())
    self._player = pyaudio.PyAudio()
    self._stream = self._player.open(
      format=pyaudio.paInt16, channels=1, rate=22050, output=True
    )

  def on_complete(self):
    print("语音合成完成，已接收全部结果：" + get_timestamp())

  def on_error(self, message: str):
    print(f"语音合成错误：{message}")

  def on_close(self):
    print("连接已关闭：" + get_timestamp())
    # 停止播放器
    self._stream.stop_stream()
    self._stream.close()
    self._player.terminate()

  def on_event(self, message):
    pass

  def on_data(self, data: bytes) -> None:
    print(get_timestamp() + " 音频二进制数据长度：" + str(len(data)))
    self._stream.write(data)


callback = Callback()

test_text = [
  "流式文本语音合成 SDK，",
  "可以将输入文本",
  "转换为二进制音频数据。",
  "相较于非流式语音合成，",
  "流式合成具有更优的实时性能。",
  "用户在输入的同时即可听到近乎同步的音频输出，",
  "大幅提升交互体验",
  "并减少等待时间。",
  "非常适合与大语言模型（LLM）集成，",
  "将文本流式传输进行语音合成。",
]

# 实例化 SpeechSynthesizer，在构造方法中传入 model、voice 等请求参数
synthesizer = SpeechSynthesizer(
  model=model,
  voice=voice,
  format=AudioFormat.PCM_22050HZ_MONO_16BIT,
  callback=callback,
)


# 流式发送文本进行合成，通过回调接口的 on_data 方法实时获取二进制音频
for text in test_text:
  synthesizer.streaming_call(text)
  time.sleep(0.1)
# 结束流式语音合成
synthesizer.streaming_complete()

# 首次请求的首包延迟包含 WebSocket 建连时间
print('[Metric] requestId: {}, first-package delay: {} ms'.format(
  synthesizer.get_last_request_id(),
  synthesizer.get_first_package_delay()))

进阶功能

Qwen-TTS 交互模式

Qwen-TTS Realtime API 提供两种 WebSocket 交互模式，通过 session.mode 参数切换：
  • server_commit 模式：服务端智能处理文本分段和合成时机，适合大段文本的连续合成场景。客户端只需持续追加文本，无需关注切分和提交。
  • commit 模式：客户端主动提交文本缓冲区以触发合成，适合需要精确控制合成时机的场景（如对话式 AI 逐轮合成）。
详细的 WebSocket 事件生命周期和连接复用方式，请参见实时语音合成-千问API参考。

交互流程

  • CosyVoice
  • Qwen-TTS-Realtime
CosyVoice 使用基于 WebSocket 的流式协议。协议详情请参见 CosyVoice WebSocket API 参考。

指令控制

  • CosyVoice
  • Qwen-TTS-Realtime
支持的模型：cosyvoice-v3.5-plus、cosyvoice-v3.5-flash、cosyvoice-v3-flash
  • cosyvoice-v3.5-plus、cosyvoice-v3.5-flash：无系统音色，仅支持使用声音设计或声音复刻音色，可输入任意指令控制合成效果（如情感、语速等）。
  • cosyvoice-v3-flash 的声音设计或声音复刻音色：可输入任意指令控制合成效果。
  • cosyvoice-v3-flash 的系统音色：指令必须使用固定格式和内容，详情请参见CosyVoice音色列表。
支持语言：
  • cosyvoice-v3.5-plus、cosyvoice-v3.5-flash：中文、英文、法语、德语、日语、韩语、俄语、葡萄牙语、泰语、印尼语、越南语
  • cosyvoice-v3-flash：中文、英文、法语、德语、日语、韩语、俄语
长度限制：100 字符。汉字（包括简体/繁体汉字、日文汉字和韩文汉字）按 2 个字符计算，其他所有字符（如标点符号、字母、数字、日韩文假名/谚文等）均按 1 个字符计算。

方言

让模型用中文方言（如河南话、四川话、粤语等）输出语音。不同模型和音色类型的设置方式不同。
  • CosyVoice
  • Qwen-TTS
  • 系统音色：在CosyVoice音色列表中选择以下任一种音色：
    • 支持方言的系统音色（例如 longshange_v3），无需额外设置即可输出对应方言。
    • 支持指令控制且可指定方言的音色（例如 longanhuan_v3），通过指令文本指定方言。
  • 声音复刻音色：通过指令控制功能设置，例如指令文本写 请用河南话表达。
  • 声音设计音色：暂不支持方言。
具体支持哪些方言：参见语音合成模型列表中各 CosyVoice 模型“支持的语言”。示例：以 cosyvoice-v3-flash + longanhuan_v3 音色，通过指令文本 "请用河南话表达。" 输出河南话语音。
# coding=utf-8
import os
import dashscope
from dashscope.audio.tts_v2 import *
# 获取API Key：https://platform.qianwenai.com/home/api-keys
# 若没有配置环境变量，请将下行替换为：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')
dashscope.base_websocket_api_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference'
# 模型
# 不同模型版本需要使用对应版本的音色：
# cosyvoice-v3-flash/cosyvoice-v3-plus：使用longanyang等音色。
# cosyvoice-v2：使用longxiaochun_v2等音色。
# 不同语言选择对应音色
model = "cosyvoice-v3-flash"
# 音色
voice = "longanhuan_v3"
# 实例化SpeechSynthesizer，并在构造方法中传入模型（model）、音色（voice）等请求参数
synthesizer = SpeechSynthesizer(model=model, voice=voice, instruction="请用河南话表达。")
# 发送待合成文本，获取二进制音频
audio = synthesizer.call("叫你去买盐，你买回来一袋面，这不是弄啥嘞吗！")
# 首次发送文本时需建立 WebSocket 连接，因此首包延迟会包含连接建立的耗时
print('[Metric] requestId为：{}，首包延迟为：{}毫秒'.format(
  synthesizer.get_last_request_id(),
  synthesizer.get_first_package_delay()))
# 将音频保存至本地
with open('output.mp3', 'wb') as f:
  f.write(audio)

WebSocket 原始协议调用

以下示例展示如何通过 WebSocket 原始协议直连服务端，适用于不使用 DashScope SDK 的场景。此为最小可运行实现，WebSocket 协议请参见各模型的 API 参考。
  • CosyVoice
  • Sambert
package main
import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)
const (
	wsURL      = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
	outputFile = "output.mp3"
)
func main() {
	// 获取API Key：https://platform.qianwenai.com/home/api-keys
	// 若没有配置环境变量，请将下行替换为：apiKey := "sk-xxx"
	apiKey := os.Getenv("DASHSCOPE_API_KEY")
	// 清空输出文件
	os.Remove(outputFile)
	os.Create(outputFile)
	// 连接WebSocket
	header := make(http.Header)
	header.Add("X-DashScope-DataInspection", "enable")
	header.Add("Authorization", fmt.Sprintf("bearer %s", apiKey))
	conn, resp, err := websocket.DefaultDialer.Dial(wsURL, header)
	if err != nil {
		if resp != nil {
			fmt.Printf("连接失败 HTTP状态码: %d\n", resp.StatusCode)
		}
		fmt.Println("连接失败:", err)
		return
	}
	defer conn.Close()
	// 生成任务ID
	taskID := uuid.New().String()
	fmt.Printf("生成任务ID: %s\n", taskID)
	// 发送run-task事件
	runTaskCmd := map[string]interface{}{
		"header": map[string]interface{}{
			"action":    "run-task",
			"task_id":   taskID,
			"streaming": "duplex",
		},
		"payload": map[string]interface{}{
			"task_group": "audio",
			"task":       "tts",
			"function":   "SpeechSynthesizer",
			"model":      "cosyvoice-v3-flash",
			"parameters": map[string]interface{}{
				"text_type":   "PlainText",
				"voice":       "longanyang",
				"format":      "mp3",
				"sample_rate": 22050,
				"volume":      50,
				"rate":        1,
				"pitch":       1,
				// 如果enable_ssml设为true，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
				"enable_ssml": false,
			},
			"input": map[string]interface{}{},
		},
	}
	runTaskJSON, _ := json.Marshal(runTaskCmd)
	fmt.Printf("发送run-task事件: %s\n", string(runTaskJSON))
	err = conn.WriteMessage(websocket.TextMessage, runTaskJSON)
	if err != nil {
		fmt.Println("发送run-task失败:", err)
		return
	}
	textSent := false
	// 处理消息
	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			fmt.Println("读取消息失败:", err)
			break
		}
		// 处理二进制消息
		if messageType == websocket.BinaryMessage {
			fmt.Printf("收到二进制消息，长度: %d\n", len(message))
			file, _ := os.OpenFile(outputFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0644)
			file.Write(message)
			file.Close()
			continue
		}
		// 处理文本消息
		messageStr := string(message)
		fmt.Printf("收到文本消息: %s\n", strings.ReplaceAll(messageStr, "\n", ""))
		// 简单解析JSON获取event类型
		var msgMap map[string]interface{}
		if json.Unmarshal(message, &msgMap) == nil {
			if header, ok := msgMap["header"].(map[string]interface{}); ok {
				if event, ok := header["event"].(string); ok {
					fmt.Printf("事件类型: %s\n", event)
					switch event {
					case "task-started":
						fmt.Println("=== 收到task-started事件 ===")
						if !textSent {
							// 发送continue-task事件
							texts := []string{"床前明月光，疑是地上霜。", "举头望明月，低头思故乡。"}
							for _, text := range texts {
								continueTaskCmd := map[string]interface{}{
									"header": map[string]interface{}{
										"action":    "continue-task",
										"task_id":   taskID,
										"streaming": "duplex",
									},
									"payload": map[string]interface{}{
										"input": map[string]interface{}{
											"text": text,
										},
									},
								}
								continueTaskJSON, _ := json.Marshal(continueTaskCmd)
								fmt.Printf("发送continue-task事件: %s\n", string(continueTaskJSON))
								err = conn.WriteMessage(websocket.TextMessage, continueTaskJSON)
								if err != nil {
									fmt.Println("发送continue-task失败:", err)
									return
								}
							}
							textSent = true
							// 延迟发送finish-task
							time.Sleep(500 * time.Millisecond)
							// 发送finish-task事件
							finishTaskCmd := map[string]interface{}{
								"header": map[string]interface{}{
									"action":    "finish-task",
									"task_id":   taskID,
									"streaming": "duplex",
								},
								"payload": map[string]interface{}{
									"input": map[string]interface{}{},
								},
							}
							finishTaskJSON, _ := json.Marshal(finishTaskCmd)
							fmt.Printf("发送finish-task事件: %s\n", string(finishTaskJSON))
							err = conn.WriteMessage(websocket.TextMessage, finishTaskJSON)
							if err != nil {
								fmt.Println("发送finish-task失败:", err)
								return
							}
						}
					case "task-finished":
						fmt.Println("=== 任务完成 ===")
						return
					case "task-failed":
						fmt.Println("=== 任务失败 ===")
						if header["error_message"] != nil {
							fmt.Printf("错误信息: %s\n", header["error_message"])
						}
						return
					case "result-generated":
						fmt.Println("收到result-generated事件")
					}
				}
			}
		}
	}
}

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
class Program {
    // 获取API Key：https://platform.qianwenai.com/home/api-keys
    // 若没有配置环境变量，请将下行替换为：private static readonly string ApiKey = "sk-xxx"
    private static readonly string ApiKey = Environment.GetEnvironmentVariable("DASHSCOPE_API_KEY") ?? throw new InvalidOperationException("DASHSCOPE_API_KEY environment variable is not set.");
    private const string WebSocketUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/";
    // 输出文件路径
    private const string OutputFilePath = "output.mp3";
    // WebSocket客户端
    private static ClientWebSocket _webSocket = new ClientWebSocket();
    // 取消令牌源
    private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    // 任务ID
    private static string? _taskId;
    // 任务是否已启动
    private static TaskCompletionSource<bool> _taskStartedTcs = new TaskCompletionSource<bool>();
    static async Task Main(string[] args) {
        try {
            // 清空输出文件
            ClearOutputFile(OutputFilePath);
            // 连接WebSocket服务
            await ConnectToWebSocketAsync(WebSocketUrl);
            // 启动接收消息的任务
            Task receiveTask = ReceiveMessagesAsync();
            // 发送run-task事件
            _taskId = GenerateTaskId();
            await SendRunTaskCommandAsync(_taskId);
            // 等待task-started事件
            await _taskStartedTcs.Task;
            // 持续发送continue-task事件
            string[] texts = {
                "床前明月光",
                "疑是地上霜",
                "举头望明月",
                "低头思故乡"
            };
            foreach (string text in texts) {
                await SendContinueTaskCommandAsync(text);
            }
            // 发送finish-task事件
            await SendFinishTaskCommandAsync(_taskId);
            // 等待接收任务完成
            await receiveTask;
            Console.WriteLine("任务完成，连接已关闭。");
        } catch (OperationCanceledException) {
            Console.WriteLine("任务被取消。");
        } catch (Exception ex) {
            Console.WriteLine($"发生错误：{ex.Message}");
        } finally {
            _cancellationTokenSource.Cancel();
            _webSocket.Dispose();
        }
    }
    private static void ClearOutputFile(string filePath) {
        if (File.Exists(filePath)) {
            File.WriteAllText(filePath, string.Empty);
            Console.WriteLine("输出文件已清空。");
        } else {
            Console.WriteLine("输出文件不存在，无需清空。");
        }
    }
    private static async Task ConnectToWebSocketAsync(string url) {
        var uri = new Uri(url);
        if (_webSocket.State == WebSocketState.Connecting || _webSocket.State == WebSocketState.Open) {
            return;
        }
        // 设置WebSocket连接的头部信息
        _webSocket.Options.SetRequestHeader("Authorization", $"bearer {ApiKey}");
        _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");
        try {
            await _webSocket.ConnectAsync(uri, _cancellationTokenSource.Token);
            Console.WriteLine("已成功连接到WebSocket服务。");
        } catch (OperationCanceledException) {
            Console.WriteLine("WebSocket连接被取消。");
        } catch (Exception ex) {
            Console.WriteLine($"WebSocket连接失败: {ex.Message}");
            throw;
        }
    }
    private static async Task SendRunTaskCommandAsync(string taskId) {
        var command = CreateCommand("run-task", taskId, "duplex", new {
            task_group = "audio",
            task = "tts",
            function = "SpeechSynthesizer",
            model = "cosyvoice-v3-flash",
            parameters = new
            {
                text_type = "PlainText",
                voice = "longanyang",
                format = "mp3",
                sample_rate = 22050,
                volume = 50,
                rate = 1,
                pitch = 1,
                // 如果enable_ssml设为true，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
                enable_ssml = false
            },
            input = new { }
        });
        await SendJsonMessageAsync(command);
        Console.WriteLine("已发送run-task事件。");
    }
    private static async Task SendContinueTaskCommandAsync(string text) {
        if (_taskId == null) {
            throw new InvalidOperationException("任务ID未初始化。");
        }
        var command = CreateCommand("continue-task", _taskId, "duplex", new {
            input = new {
                text
            }
        });
        await SendJsonMessageAsync(command);
        Console.WriteLine("已发送continue-task事件。");
    }
    private static async Task SendFinishTaskCommandAsync(string taskId) {
        var command = CreateCommand("finish-task", taskId, "duplex", new {
            input = new { }
        });
        await SendJsonMessageAsync(command);
        Console.WriteLine("已发送finish-task事件。");
    }
    private static async Task SendJsonMessageAsync(string message) {
        var buffer = Encoding.UTF8.GetBytes(message);
        try {
            await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, _cancellationTokenSource.Token);
        } catch (OperationCanceledException) {
            Console.WriteLine("消息发送被取消。");
        }
    }
    private static async Task ReceiveMessagesAsync() {
        while (_webSocket.State == WebSocketState.Open) {
            var response = await ReceiveMessageAsync();
            if (response != null) {
                var eventStr = response.RootElement.GetProperty("header").GetProperty("event").GetString();
                switch (eventStr) {
                    case "task-started":
                        Console.WriteLine("任务已启动。");
                        _taskStartedTcs.TrySetResult(true);
                        break;
                    case "task-finished":
                        Console.WriteLine("任务已完成。");
                        _cancellationTokenSource.Cancel();
                        break;
                    case "task-failed":
                        Console.WriteLine("任务失败：" + response.RootElement.GetProperty("header").GetProperty("error_message").GetString());
                        _cancellationTokenSource.Cancel();
                        break;
                    default:
                        // result-generated可在此处理
                        break;
                }
            }
        }
    }
    private static async Task<JsonDocument?> ReceiveMessageAsync() {
        var buffer = new byte[1024 * 4];
        var segment = new ArraySegment<byte>(buffer);
        try {
            WebSocketReceiveResult result = await _webSocket.ReceiveAsync(segment, _cancellationTokenSource.Token);
            if (result.MessageType == WebSocketMessageType.Close) {
                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Closing", _cancellationTokenSource.Token);
                return null;
            }
            if (result.MessageType == WebSocketMessageType.Binary) {
                // 处理二进制数据
                Console.WriteLine("接收到二进制数据...");
                // 将二进制数据保存到文件
                using (var fileStream = new FileStream(OutputFilePath, FileMode.Append)) {
                    fileStream.Write(buffer, 0, result.Count);
                }
                return null;
            }
            string message = Encoding.UTF8.GetString(buffer, 0, result.Count);
            return JsonDocument.Parse(message);
        } catch (OperationCanceledException) {
            Console.WriteLine("消息接收被取消。");
            return null;
        }
    }
    private static string GenerateTaskId() {
        return Guid.NewGuid().ToString("N").Substring(0, 32);
    }
    private static string CreateCommand(string action, string taskId, string streaming, object payload) {
        var command = new {
            header = new {
                action,
                task_id = taskId,
                streaming
            },
            payload
        };
        return JsonSerializer.Serialize(command);
    }
}
示例代码目录结构为：
my-php-project/
├── composer.json
├── vendor/
└── index.php
composer.json内容如下，相关依赖的版本号请根据实际情况自行决定：
{
    "require": {
        "react/event-loop": "^1.3",
        "react/socket": "^1.11",
        "react/stream": "^1.2",
        "react/http": "^1.1",
        "ratchet/pawl": "^0.4"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    }
}
index.php内容如下：
<?php
require __DIR__ . '/vendor/autoload.php';
use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;
// 获取API Key：https://platform.qianwenai.com/home/api-keys
// 若没有配置环境变量，请将下行替换为：$api_key = "sk-xxx"
$api_key = getenv("DASHSCOPE_API_KEY");
$websocket_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/'; // WebSocket服务器地址
$output_file = 'output.mp3'; // 输出文件路径
$loop = Loop::get();
if (file_exists($output_file)) {
    // 清空文件内容
    file_put_contents($output_file, '');
}
// 创建自定义的连接器
$socketConnector = new SocketConnector($loop, [
    'tcp' => [
        'bindto' => '0.0.0.0:0',
    ],
    'tls' => [
        'verify_peer' => false,
        'verify_peer_name' => false,
    ],
]);
$connector = new Connector($loop, $socketConnector);
$headers = [
    'Authorization' => 'bearer ' . $api_key,
    'X-DashScope-DataInspection' => 'enable'
];
$connector($websocket_url, [], $headers)->then(function ($conn) use ($loop, $output_file) {
    echo "连接到WebSocket服务器\n";
    // 生成任务ID
    $taskId = generateTaskId();
    // 发送 run-task 事件
    sendRunTaskMessage($conn, $taskId);
    // 定义发送 continue-task 事件的函数
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
        // 待发送的文本
        $texts = ["床前明月光", "疑是地上霜", "举头望明月", "低头思故乡"];
        $continueTaskCount = 0;
        foreach ($texts as $text) {
            $continueTaskMessage = json_encode([
                "header" => [
                    "action" => "continue-task",
                    "task_id" => $taskId,
                    "streaming" => "duplex"
                ],
                "payload" => [
                    "input" => [
                        "text" => $text
                    ]
                ]
            ]);
            echo "准备发送continue-task事件: " . $continueTaskMessage . "\n";
            $conn->send($continueTaskMessage);
            $continueTaskCount++;
        }
        echo "发送的continue-task事件个数为：" . $continueTaskCount . "\n";
        // 发送 finish-task 事件
        sendFinishTaskMessage($conn, $taskId);
    };
    // 标记是否收到 task-started 事件
    $taskStarted = false;
    // 监听消息
    $conn->on('message', function($msg) use ($conn, $sendContinueTask, $loop, &$taskStarted, $taskId, $output_file) {
        if ($msg->isBinary()) {
            // 写入二进制数据到本地文件
            file_put_contents($output_file, $msg->getPayload(), FILE_APPEND);
        } else {
            // 处理非二进制消息
            $response = json_decode($msg, true);
            if (isset($response['header']['event'])) {
                handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, $taskStarted);
            } else {
                echo "未知的消息格式\n";
            }
        }
    });
    // 监听连接关闭
    $conn->on('close', function($code = null, $reason = null) {
        echo "连接已关闭\n";
        if ($code !== null) {
            echo "关闭代码: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "关闭原因：" . $reason . "\n";
        }
    });
}, function ($e) {
    echo "无法连接：{$e->getMessage()}\n";
});
$loop->run();
/**
 * 生成任务ID
 * @return string
 */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}
/**
 * 发送 run-task 事件
 * @param $conn
 * @param $taskId
 */
function sendRunTaskMessage($conn, $taskId) {
    $runTaskMessage = json_encode([
        "header" => [
            "action" => "run-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "task_group" => "audio",
            "task" => "tts",
            "function" => "SpeechSynthesizer",
            "model" => "cosyvoice-v3-flash",
            "parameters" => [
                "text_type" => "PlainText",
                "voice" => "longanyang",
                "format" => "mp3",
                "sample_rate" => 22050,
                "volume" => 50,
                "rate" => 1,
                "pitch" => 1,
                // 如果enable_ssml设为true，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
                "enable_ssml" => false
            ],
            "input" => (object) []
        ]
    ]);
    echo "准备发送run-task事件: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task事件已发送\n";
}
/**
 * 读取音频文件
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "无法读取音频文件\n";
    }
    return $voiceData;
}
/**
 * 分割音频数据
 * @param string $data
 * @param int $chunkSize
 * @return array
 */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}
/**
 * 发送 finish-task 事件
 * @param $conn
 * @param $taskId
 */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
        "header" => [
            "action" => "finish-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "input" => (object) []
        ]
    ]);
    echo "准备发送finish-task事件: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task事件已发送\n";
}
/**
 * 处理事件
 * @param $conn
 * @param $response
 * @param $sendContinueTask
 * @param $loop
 * @param $taskId
 * @param $taskStarted
 */
function handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, &$taskStarted) {
    switch ($response['header']['event']) {
        case 'task-started':
            echo "任务开始，发送continue-task事件...\n";
            $taskStarted = true;
            // 发送 continue-task 事件
            $sendContinueTask();
            break;
        case 'result-generated':
            // 收到result-generated事件
            break;
        case 'task-finished':
            echo "任务完成\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "任务失败\n";
            echo "错误代码：" . $response['header']['error_code'] . "\n";
            echo "错误信息：" . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "错误：" . $response['payload']['message'] . "\n";
            break;
        default:
            echo "未知事件：" . $response['header']['event'] . "\n";
            break;
    }
    // 如果任务已完成，关闭连接
    if ($response['header']['event'] == 'task-finished') {
        // 等待1秒以确保所有数据都已传输完毕
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "客户端关闭连接\n";
        });
    }
    // 如果没有收到 task-started 事件，关闭连接
    if (!$taskStarted && in_array($response['header']['event'], ['task-failed', 'error'])) {
        $conn->close();
    }
}
需安装相关依赖：
npm install ws
npm install uuid
示例代码如下：
const WebSocket = require('ws');
const fs = require('fs');
const uuid = require('uuid').v4;
// 获取API Key：https://platform.qianwenai.com/home/api-keys
// 若没有配置环境变量，请将下行替换为：const apiKey = "sk-xxx"
const apiKey = process.env.DASHSCOPE_API_KEY;
const url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/';
// 输出文件路径
const outputFilePath = 'output.mp3';
// 清空输出文件
fs.writeFileSync(outputFilePath, '');
// 创建WebSocket客户端
const ws = new WebSocket(url, {
  headers: {
    Authorization: `bearer ${apiKey}`,
    'X-DashScope-DataInspection': 'enable'
  }
});
let taskStarted = false;
let taskId = uuid();
ws.on('open', () => {
  console.log('已连接到WebSocket服务器');
  // 发送run-task事件
  const runTaskMessage = JSON.stringify({
    header: {
      action: 'run-task',
      task_id: taskId,
      streaming: 'duplex'
    },
    payload: {
      task_group: 'audio',
      task: 'tts',
      function: 'SpeechSynthesizer',
      model: 'cosyvoice-v3-flash',
      parameters: {
        text_type: 'PlainText',
        voice: 'longanyang', // 音色
        format: 'mp3', // 音频格式
        sample_rate: 22050, // 采样率
        volume: 50, // 音量
        rate: 1, // 语速
        pitch: 1, // 音调
        enable_ssml: false // 是否开启SSML功能。如果enable_ssml设为true，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
      },
      input: {}
    }
  });
  ws.send(runTaskMessage);
  console.log('已发送run-task消息');
});
const fileStream = fs.createWriteStream(outputFilePath, { flags: 'a' });
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // 写入二进制数据到文件
    fileStream.write(data);
  } else {
    const message = JSON.parse(data);
    switch (message.header.event) {
      case 'task-started':
        taskStarted = true;
        console.log('任务已开始');
        // 发送continue-task事件
        sendContinueTasks(ws);
        break;
      case 'task-finished':
        console.log('任务已完成');
        ws.close();
        fileStream.end(() => {
          console.log('文件流已关闭');
        });
        break;
      case 'task-failed':
        console.error('任务失败：', message.header.error_message);
        ws.close();
        fileStream.end(() => {
          console.log('文件流已关闭');
        });
        break;
      default:
        // 可以在这里处理result-generated
        break;
    }
  }
});
function sendContinueTasks(ws) {
  const texts = [
    '床前明月光，',
    '疑是地上霜。',
    '举头望明月，',
    '低头思故乡。'
  ];
  texts.forEach((text, index) => {
    setTimeout(() => {
      if (taskStarted) {
        const continueTaskMessage = JSON.stringify({
          header: {
            action: 'continue-task',
            task_id: taskId,
            streaming: 'duplex'
          },
          payload: {
            input: {
              text: text
            }
          }
        });
        ws.send(continueTaskMessage);
        console.log(`已发送continue-task，文本：${text}`);
      }
    }, index * 1000); // 每隔1秒发送一次
  });
  // 发送finish-task事件
  setTimeout(() => {
    if (taskStarted) {
      const finishTaskMessage = JSON.stringify({
        header: {
          action: 'finish-task',
          task_id: taskId,
          streaming: 'duplex'
        },
        payload: {
          input: {}
        }
      });
      ws.send(finishTaskMessage);
      console.log('已发送finish-task');
    }
  }, texts.length * 1000 + 1000); // 在所有continue-task事件发送完毕后1秒发送
}
ws.on('close', () => {
  console.log('已断开与WebSocket服务器的连接');
});
建议使用 Java DashScope SDK 进行开发，请参见 Java SDK。以下是 Java WebSocket 直连示例，运行前请导入以下依赖：
  • Java-WebSocket
  • jackson-databind
推荐使用Maven或Gradle管理依赖包，其配置如下：
<dependencies>
    <!-- WebSocket Client -->
    <dependency>
        <groupId>org.java-websocket</groupId>
        <artifactId>Java-WebSocket</artifactId>
        <version>1.5.3</version>
    </dependency>
    <!-- JSON Processing -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.13.0</version>
    </dependency>
</dependencies>
Java代码如下：
import com.fasterxml.jackson.databind.ObjectMapper;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.*;
public class TTSWebSocketClient extends WebSocketClient {
    private final String taskId = UUID.randomUUID().toString();
    private final String outputFile = "output_" + System.currentTimeMillis() + ".mp3";
    private boolean taskFinished = false;
    public TTSWebSocketClient(URI serverUri, Map<String, String> headers) {
        super(serverUri, headers);
    }
    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        System.out.println("连接成功");
        // 发送run-task事件
        // 如果enable_ssml设为true，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
        String runTaskCommand = "{ \"header\": { \"action\": \"run-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"task_group\": \"audio\", \"task\": \"tts\", \"function\": \"SpeechSynthesizer\", \"model\": \"cosyvoice-v3-flash\", \"parameters\": { \"text_type\": \"PlainText\", \"voice\": \"longanyang\", \"format\": \"mp3\", \"sample_rate\": 22050, \"volume\": 50, \"rate\": 1, \"pitch\": 1, \"enable_ssml\": false }, \"input\": {} }}";
        send(runTaskCommand);
    }
    @Override
    public void onMessage(String message) {
        System.out.println("收到服务端返回的消息：" + message);
        try {
            // Parse JSON message
            Map<String, Object> messageMap = new ObjectMapper().readValue(message, Map.class);
            if (messageMap.containsKey("header")) {
                Map<String, Object> header = (Map<String, Object>) messageMap.get("header");
                if (header.containsKey("event")) {
                    String event = (String) header.get("event");
                    if ("task-started".equals(event)) {
                        System.out.println("收到服务端返回的task-started事件");
                        List<String> texts = Arrays.asList(
                                "床前明月光，疑是地上霜",
                                "举头望明月，低头思故乡"
                        );
                        for (String text : texts) {
                            // 发送continue-task事件
                            sendContinueTask(text);
                        }
                        // 发送finish-task事件
                        sendFinishTask();
                    } else if ("task-finished".equals(event)) {
                        System.out.println("收到服务端返回的task-finished事件");
                        taskFinished = true;
                        closeConnection();
                    } else if ("task-failed".equals(event)) {
                        System.out.println("任务失败：" + message);
                        closeConnection();
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("出现异常：" + e.getMessage());
        }
    }
    @Override
    public void onMessage(ByteBuffer message) {
        System.out.println("收到的二进制音频数据大小为：" + message.remaining());
        try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
            byte[] buffer = new byte[message.remaining()];
            message.get(buffer);
            fos.write(buffer);
            System.out.println("音频数据已写入本地文件" + outputFile + "中");
        } catch (IOException e) {
            System.err.println("音频数据写入本地文件失败：" + e.getMessage());
        }
    }
    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("连接关闭：" + reason + " (" + code + ")");
    }
    @Override
    public void onError(Exception ex) {
        System.err.println("报错：" + ex.getMessage());
        ex.printStackTrace();
    }
    private void sendContinueTask(String text) {
        String command = "{ \"header\": { \"action\": \"continue-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": { \"text\": \"" + text + "\" } }}";
        send(command);
    }
    private void sendFinishTask() {
        String command = "{ \"header\": { \"action\": \"finish-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"input\": {} }}";
        send(command);
    }
    private void closeConnection() {
        if (!isClosed()) {
            close();
        }
    }
    public static void main(String[] args) {
        try {
            // 获取API Key：https://platform.qianwenai.com/home/api-keys
            // 若没有配置环境变量，请将下行替换为：String apiKey = "sk-xxx"
            String apiKey = System.getenv("DASHSCOPE_API_KEY");
            if (apiKey == null || apiKey.isEmpty()) {
                System.err.println("请设置 DASHSCOPE_API_KEY 环境变量");
                return;
            }
            Map<String, String> headers = new HashMap<>();
            headers.put("Authorization", "bearer " + apiKey);
            TTSWebSocketClient client = new TTSWebSocketClient(new URI("wss://dashscope.aliyuncs.com/api-ws/v1/inference/"), headers);
            client.connect();
            while (!client.isClosed() && !client.taskFinished) {
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            System.err.println("连接WebSocket服务失败：" + e.getMessage());
            e.printStackTrace();
        }
    }
}
建议使用 Python DashScope SDK 进行开发，请参见 Python SDK。以下是 Python WebSocket 直连示例，运行前请导入以下依赖：
pip uninstall websocket-client
pip uninstall websocket
pip install websocket-client
请不要将运行示例代码的Python文件命名为“websocket.py”，否则会报错（AttributeError: module 'websocket' has no attribute 'WebSocketApp'. Did you mean: 'WebSocket'?）。
import websocket
import json
import uuid
import os
import time
class TTSClient:
  def __init__(self, api_key, uri):
    """
  初始化 TTSClient 实例
  参数:
    api_key (str): 鉴权用的 API Key
    uri (str): WebSocket 服务地址
  """
    self.api_key = api_key  # 替换为你的 API Key
    self.uri = uri  # 替换为你的 WebSocket 地址
    self.task_id = str(uuid.uuid4())  # 生成唯一任务 ID
    self.output_file = f"output_{int(time.time())}.mp3"  # 输出音频文件路径
    self.ws = None  # WebSocketApp 实例
    self.task_started = False  # 是否收到 task-started
    self.task_finished = False  # 是否收到 task-finished / task-failed
  def on_open(self, ws):
    """
  WebSocket 连接建立时回调函数
  发送 run-task 事件开启语音合成任务
  """
    print("WebSocket 已连接")
    # 构造 run-task 事件
    run_task_cmd = {
      "header": {
        "action": "run-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "task_group": "audio",
        "task": "tts",
        "function": "SpeechSynthesizer",
        "model": "cosyvoice-v3-flash",
        "parameters": {
          "text_type": "PlainText",
          "voice": "longanyang",
          "format": "mp3",
          "sample_rate": 22050,
          "volume": 50,
          "rate": 1,
          "pitch": 1,
          # 如果enable_ssml设为True，只允许发送一次continue-task事件，否则会报错“Text request limit violated, expected 1.”
          "enable_ssml": False
        },
        "input": {}
      }
    }
    # 发送 run-task 事件
    ws.send(json.dumps(run_task_cmd))
    print("已发送 run-task 事件")
  def on_message(self, ws, message):
    """
  接收到消息时的回调函数
  区分文本和二进制消息处理
  """
    if isinstance(message, str):
      # 处理 JSON 文本消息
      try:
        msg_json = json.loads(message)
        print(f"收到 JSON 消息: {msg_json}")
        if "header" in msg_json:
          header = msg_json["header"]
          if "event" in header:
            event = header["event"]
            if event == "task-started":
              print("任务已启动")
              self.task_started = True
              # 发送 continue-task 事件
              texts = [
                "床前明月光，疑是地上霜",
                "举头望明月，低头思故乡"
              ]
              for text in texts:
                self.send_continue_task(text)
              # 所有 continue-task 发送完成后发送 finish-task
              self.send_finish_task()
            elif event == "task-finished":
              print("任务已完成")
              self.task_finished = True
              self.close(ws)
            elif event == "task-failed":
              error_msg = msg_json.get("error_message", "未知错误")
              print(f"任务失败: {error_msg}")
              self.task_finished = True
              self.close(ws)
      except json.JSONDecodeError as e:
        print(f"JSON 解析失败: {e}")
    else:
      # 处理二进制消息（音频数据）
      print(f"收到二进制消息，大小: {len(message)} 字节")
      with open(self.output_file, "ab") as f:
        f.write(message)
      print(f"已将音频数据写入本地文件{self.output_file}中")
  def on_error(self, ws, error):
    """发生错误时的回调"""
    print(f"WebSocket 出错: {error}")
  def on_close(self, ws, close_status_code, close_msg):
    """连接关闭时的回调"""
    print(f"WebSocket 已关闭: {close_msg} ({close_status_code})")
  def send_continue_task(self, text):
    """发送 continue-task 事件，附带要合成的文本内容"""
    cmd = {
      "header": {
        "action": "continue-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "input": {
          "text": text
        }
      }
    }
    self.ws.send(json.dumps(cmd))
    print(f"已发送 continue-task 事件，文本内容: {text}")
  def send_finish_task(self):
    """发送 finish-task 事件，结束语音合成任务"""
    cmd = {
      "header": {
        "action": "finish-task",
        "task_id": self.task_id,
        "streaming": "duplex"
      },
      "payload": {
        "input": {}
      }
    }
    self.ws.send(json.dumps(cmd))
    print("已发送 finish-task 事件")
  def close(self, ws):
    """主动关闭连接"""
    if ws and ws.sock and ws.sock.connected:
      ws.close()
      print("已主动关闭连接")
  def run(self):
    """启动 WebSocket 客户端"""
    # 设置请求头部（鉴权）
    header = {
      "Authorization": f"bearer {self.api_key}",
      "X-DashScope-DataInspection": "enable"
    }
    # 创建 WebSocketApp 实例
    self.ws = websocket.WebSocketApp(
      self.uri,
      header=header,
      on_open=self.on_open,
      on_message=self.on_message,
      on_error=self.on_error,
      on_close=self.on_close
    )
    print("正在监听 WebSocket 消息...")
    self.ws.run_forever()  # 启动长连接监听
# 示例使用方式
if __name__ == "__main__":
  # 获取API Key：https://platform.qianwenai.com/home/api-keys
  # 若没有配置环境变量，请将下行替换为：API_KEY = "sk-xxx"
  API_KEY = os.environ.get("DASHSCOPE_API_KEY")
  SERVER_URI = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"  # 替换为你的 WebSocket 地址
  client = TTSClient(API_KEY, SERVER_URI)
  client.run()

声音定制

  • CosyVoice
  • Qwen-TTS-Realtime

声音复刻：输入音频格式要求

高质量的输入音频是实现优秀复刻效果的基础。
项目要求
支持格式WAV（16-bit）、MP3、M4A
音频时长推荐：10~20秒。最长：60秒。
文件大小≤ 10 MB
采样率≥ 16 kHz
声道单声道或立体声。立体声音频仅处理第一声道，请确保第一声道包含清晰的人声。
内容音频必须包含至少5秒的连续、清晰人声，不含背景音。其余部分仅允许短暂停顿（≤ 2秒）。整段音频应无背景音乐、噪音或其他人声，以确保核心语音内容的高质量。请使用正常说话的音频作为输入，不要上传歌曲或演唱音频，以确保复刻效果的准确性和可用性。

声音设计：编写高质量的声音描述

限制条件

编写声音描述（voice_prompt）时，请遵循以下技术约束：
  • 长度限制：voice_prompt 的内容不得超过500个字符。
  • 支持语言：描述文本仅支持中文和英文。

核心原则

voice_prompt 用于引导模型生成具有特定特征的声音。编写声音描述时，请遵循以下核心原则：
  • 具体而非模糊：使用能够描绘具体声音特质的词语，如"低沉"、"清脆"、"语速偏快"。避免使用"好听"、"普通"等主观且缺乏信息量的词汇。
  • 多维而非单一：优秀的描述通常结合多个维度（如性别、年龄、情感等）。单一维度的描述（如仅"女声"）过于宽泛，难以生成特色鲜明的效果。
  • 客观而非主观：专注于声音本身的物理和感知特征，而不是个人喜好。例如，用"音调偏高，带有活力"代替"我最喜欢的声音"。
  • 原创而非模仿：请描述声音的特质，而不是要求模仿特定人物（如名人、演员）。此类请求涉及版权风险，且模型不支持直接模仿。
  • 简洁而非冗余：确保每个词都有其意义。避免重复使用同义词或无意义的强调词（如"非常非常棒的声音"）。

描述维度参考

维度示例
性别男性、女性、中性
年龄儿童（5-12岁）、青少年（13-18岁）、青年（19-35岁）、中年（36-55岁）、老年（55岁以上）
音调高、中、低、偏高、偏低
语速快、中、慢、偏快、偏慢
情感欢快、沉稳、温柔、严肃、活泼、冷酷、舒缓
特质磁性、清脆、沙哑、浑厚、甜美、浓郁、有力
用途新闻播报、广告配音、有声读物、动画角色、语音助手、纪录片解说

示例对比

好的案例：
  • "年轻活泼的女声，语速较快，带有明显的上扬语调，适合介绍时尚产品。"
    • 分析：该描述结合了年龄、性格、语速和语调，并指定了使用场景，形成了清晰的声音画像。
  • "沉稳的中年男声，语速偏慢，低沉而富有磁性，适合新闻播报或纪录片解说。"
    • 分析：该描述清晰定义了性别、年龄段、语速、音质和用途。
  • "可爱的童声，约8岁女孩，说话略带稚气，适合动画角色配音。"
    • 分析：该描述精准定位了年龄和声音特质（稚气），且有明确用途。
  • "温柔知性的女性，约30岁，语气平和，适合有声读物朗读。"
    • 分析：该描述通过"知性"、"平和"等词有效传达了声音的情感和风格。
不好的案例及改进建议：
不好的案例主要问题改进建议
"好听的声音"描述过于模糊和主观，缺乏可操作的细节。添加具体维度，如"音色清亮的年轻女声，语调轻柔"。
"像某明星的声音"涉及版权风险，模型不支持直接模仿。提取声音特征进行描述，如"成熟、磁性、语速沉稳的男声"。
"非常非常非常好听的女声"描述冗余，重复用词无法帮助定义声音。去除重复，添加有效描述，如"20~24岁的女声，音色轻快，语调活泼，音质甜美"。
123456无效输入，无法解析为声音特征。请提供有意义的文字描述，参见上方推荐示例。

连接复用（WebSocket）

WebSocket 连接支持复用：一个合成任务结束后，无需重新建立连接即可开启下一个任务。 复用流程：
  • CosyVoice / Sambert：客户端发送 finish-task，服务端返回 task-finished 后，可重新发送 run-task 开启新任务。
  • Qwen-TTS：客户端发送 session.finish，服务端返回 session.finished 后，可建立新会话开启下一个任务。
  • 必须等服务端返回结束事件（task-finished 或 session.finished）后才可发起新任务。
  • CosyVoice 和 Sambert 在复用连接中的不同任务需要使用不同的 task_id。
  • 任务失败时服务端返回错误事件并关闭连接，该连接不可复用。
  • 任务结束后 60 秒无新任务，连接自动断开。
各模型事件说明请参见对应的 API 参考。

高并发最佳实践

DashScope SDK 内置池化机制，可复用 WebSocket 连接和合成对象，避免频繁创建销毁带来的开销。
  • CosyVoice
  • Sambert
前提条件：
  • 获取API Key
  • 已安装符合版本要求的 DashScope SDK，建议安装最新版：
    • Python SDK：版本 ≥ 1.25.2
    • Java SDK：版本 ≥ 2.16.6
  • Python SDK
  • Java SDK
Python SDK 通过 SpeechSynthesizerObjectPool 管理和复用 SpeechSynthesizer 对象。对象池在初始化时即创建指定数量的 SpeechSynthesizer 实例并建立 WebSocket 连接，获取对象时可直接发起请求，降低首包延迟。归还后连接保持活跃，等待下次复用。

实现步骤

  1. 安装依赖：安装 DashScope 依赖（pip install -U dashscope）。
  2. 创建并配置对象池。 对象池大小推荐设为峰值并发数的 1.5~2 倍，且不应超过账户的 QPS 限制。 创建全局单例对象池（初始化时建立连接，有一定耗时）： 
from dashscope.audio.tts_v2 import SpeechSynthesizerObjectPool
synthesizer_object_pool = SpeechSynthesizerObjectPool(max_size=20)
import dashscope
dashscope.base_http_api_url = "https://dashscope.aliyuncs.com/api/v1"
  • 在对象池场景中，SpeechSynthesizerObjectPool 在初始化时即按当前全局 dashscope.api_key 与服务端建立 WebSocket 连接。apiKey 仅在 WebSocket 建连握手时写入 Authorization 请求头用于鉴权，后续任务消息（如 run-task）本身不携带 apiKey。池创建后修改 dashscope.api_key 不会影响池内已建连接——borrow_synthesizer 取出的对象（包括归还后再次复用的对象）仍使用握手时的 apiKey，新值会被静默忽略，可能导致身份、配额或计费归属与预期不一致。注意：borrow_synthesizer 也不支持通过参数指定 apiKey。
  • 如确需使用多个不同的 API Key，请为每个 API Key 维护独立的 SpeechSynthesizerObjectPool 实例。
  1. 从对象池中获取 SpeechSynthesizer 对象。 如果当前未归还的对象数已超过池容量，系统会额外创建新对象。此类对象需重新建立连接，不具备复用效果。 
speech_synthesizer = connectionPool.borrow_synthesizer(
  model='cosyvoice-v3-flash',
  voice='longanyang',
  seed=12382,
  callback=synthesizer_callback
)
  1. 进行语音合成。调用 SpeechSynthesizer 对象的 call 或 streaming_call 方法进行语音合成。
  2. 归还 SpeechSynthesizer 对象。任务结束后归还对象以供复用。不要归还未完成任务或任务失败的对象。
connectionPool.return_synthesizer(speech_synthesizer)
复制使用前请注意：SpeechSynthesizerObjectPool 在初始化时即按当前全局 dashscope.api_key 与服务端建立 WebSocket 连接并完成鉴权；池创建后再修改 dashscope.api_key 不会影响池内已建连接，新值会被静默忽略。多 API Key 场景请为每个 API Key 维护独立的池实例。详见上文重要说明。
# !/usr/bin/env python3
# Copyright (C) Alibaba Group. All Rights Reserved.
# MIT License (https://opensource.org/licenses/MIT)
import os
import time
import threading
import dashscope
from dashscope.audio.tts_v2 import *
USE_CONNECTION_POOL = True
text_to_synthesize = [
  '第一句、欢迎使用阿里巴巴语音合成服务。',
  '第二句、欢迎使用阿里巴巴语音合成服务。',
  '第三句、欢迎使用阿里巴巴语音合成服务。',
]
connectionPool = None
def init_dashscope_api_key():
  '''
  Set your DashScope API-key. More information:
  https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
  '''
  if 'DASHSCOPE_API_KEY' in os.environ:
    dashscope.api_key = os.environ[
      'DASHSCOPE_API_KEY']  # load API-key from environment variable DASHSCOPE_API_KEY
  else:
    dashscope.api_key = '<your-dashscope-api-key>'  # set API-key manually
def synthesis_text_to_speech_and_play_by_streaming_mode(text, task_id):
  global USE_CONNECTION_POOL, connectionPool
  '''
  Synthesize speech with given text by streaming mode, async call and play the synthesized audio in real-time.
  '''
  complete_event = threading.Event()
  # Define a callback to handle the result
  class Callback(ResultCallback):
    def on_open(self):
      # when using object pool, on_open will be called after task start
      self.file = open(f'result_{task_id}.mp3', 'wb')
      print(f'[task_{task_id}] start')
    def on_complete(self):
      print(f'[task_{task_id}] speech synthesis task complete successfully.')
      complete_event.set()
    def on_error(self, message: str):
      print(f'[task_{task_id}] speech synthesis task failed, {message}')
    def on_close(self):
      # when using object pool, on_open will be called after task finished
      print(f'[task_{task_id}] finished')
    def on_event(self, message):
      # print(f'recv speech synthsis message {message}')
      pass
    def on_data(self, data: bytes) -> None:
      # send to player
      # save audio to file
      self.file.write(data)
  # Call the speech synthesizer callback
  synthesizer_callback = Callback()
  # Initialize the speech synthesizer
  # you can customize the synthesis parameters, like voice, format, sample_rate or other parameters
  if USE_CONNECTION_POOL:
    speech_synthesizer = connectionPool.borrow_synthesizer(
      model='cosyvoice-v3-flash',
      voice='longanyang',
      seed=12382,
      callback=synthesizer_callback
    )
  else:
    speech_synthesizer = SpeechSynthesizer(model='cosyvoice-v3-flash',
                       voice='longanyang',
                       seed=12382,
                       callback=synthesizer_callback)
  try:
    speech_synthesizer.call(text)
  except Exception as e:
    print(f'[task_{task_id}] speech synthesis task failed, {e}')
    if USE_CONNECTION_POOL:
      # close the synthesizer connection manually if task failed when using connection pool.
      speech_synthesizer.close()
    return
  print('[task_{}] Synthesized text: {}'.format(task_id, text))
  complete_event.wait()
  print('[task_{}][Metric] requestId: {}, first package delay ms: {}'.format(
    task_id,
    speech_synthesizer.get_last_request_id(),
    speech_synthesizer.get_first_package_delay()))
  if USE_CONNECTION_POOL:
    connectionPool.return_synthesizer(speech_synthesizer)
# main function
if __name__ == '__main__':
  # 必须先设置 dashscope.api_key 和 base_websocket_api_url，再创建 SpeechSynthesizerObjectPool。
  # 池在初始化时即按当前全局 dashscope.api_key 建立 WebSocket 连接，
  # 池创建后再修改 dashscope.api_key 不会影响池内已建连接。
  dashscope.base_websocket_api_url='wss://dashscope.aliyuncs.com/api-ws/v1/inference'
  init_dashscope_api_key()
  if USE_CONNECTION_POOL:
    print('creating connection pool')
    start_time = time.time() * 1000
    connectionPool = SpeechSynthesizerObjectPool(max_size=3)
    end_time = time.time() * 1000
    print('connection pool created, cost: {} ms'.format(end_time - start_time))
  task_thread_list = []
  for task_id in range(3):
    thread = threading.Thread(
      target=synthesis_text_to_speech_and_play_by_streaming_mode,
      args=(text_to_synthesize[task_id], task_id))
    task_thread_list.append(thread)
  for task_thread in task_thread_list:
    task_thread.start()
  for task_thread in task_thread_list:
    task_thread.join()
  if USE_CONNECTION_POOL:
    connectionPool.shutdown()

资源管理与异常处理

  • 任务成功：当语音合成任务正常完成时，必须调用 connectionPool.return_synthesizer(speech_synthesizer) 将 SpeechSynthesizer 对象归还到池中，以便复用。
    不要归还未完成任务或任务失败的 SpeechSynthesizer 对象。
  • 任务失败：当 SDK 内部或业务逻辑抛出异常导致任务中断时，主动关闭底层的 WebSocket 连接：speech_synthesizer.close()。
  • 在所有语音合成任务完成后，要通过如下方式关闭对象池：connectionPool.shutdown()。
  • 在服务出现 TaskFailed 报错时，不需要额外处理。

API 参考

  • CosyVoice
  • Qwen-TTS-Realtime

系统音色

  • CosyVoice
  • Qwen-TTS-Realtime

常见问题

  • 将多音字替换为同音的其他汉字，快速解决发音问题。
  • 使用 SSML 标记语言控制发音：Sambert 和 CosyVoice 均支持 SSML。
  1. 确认音色状态：调用 CosyVoice 声音复刻/设计 API 接口，确认音色的 status 是否为 OK。
  2. 检查模型版本一致性：确保复刻音色时使用的 target_model 参数与语音合成时的 model 参数完全一致。例如复刻时使用 cosyvoice-v3-plus，合成时也必须使用 cosyvoice-v3-plus。
  3. 验证源音频质量：检查复刻音色时使用的源音频是否符合 CosyVoice 声音复刻/设计 API 的音频要求（音频时长 10-20 秒、音质清晰、无背景噪音）。
  4. 检查请求参数：确认语音合成请求中的 voice 参数已设置为复刻音色的 ID。
如果复刻音色后合成的语音出现以下问题：
  • 语音播放不完整，只读出部分文字
  • 合成效果不稳定，时好时坏
  • 语音中包含异常停顿或静音段
可能原因：源音频质量不符合要求。解决方案：请检查源音频是否符合 CosyVoice 声音复刻/设计 API 中的音频要求，建议重新录制。
语音合成采用流式机制，边合成边返回数据，因此保存的 WAV 文件头中的时长是预估值，存在一定误差。如需精确时长，可将 format 设置为 pcm，待获取完整合成结果后自行添加 WAV 文件头信息。
请按以下场景逐一排查：
  • 音频保存为完整文件（如 xx.mp3）的情况
    • 音频格式一致性：请求参数中的音频格式须与文件后缀一致（如参数为 wav 则文件须为 .wav）。
    • 播放器兼容性：确认播放器支持该音频的格式和采样率。
  • 流式播放音频的情况
    • 将音频流保存为完整文件，尝试用播放器播放。如果文件无法播放，请参考场景 1 的排查方法。
    • 如果文件可正常播放，则问题在流式播放实现。请确认播放器支持流式播放（如 ffmpeg、pyaudio、AudioFormat、MediaSource 等）。
请按以下步骤逐一排查：
  • 检查文本发送速度：确保发送间隔合理，避免上段音频播完后下段文本尚未到达。
  • 检查回调函数性能：
    • 确认回调函数中无阻塞性业务逻辑。
    • 回调运行在 WebSocket 线程，阻塞会影响数据接收。建议将音频数据写入独立缓冲区，在其他线程中处理。
  • 检查网络稳定性：网络波动可能导致音频传输中断或延迟。
请按以下步骤排查：
  • 检查输入间隔：如果是流式合成，确认文本发送间隔是否过长，过长会导致合成总时长增加。
  • 分析性能指标：
    • 首包延迟：正常约 500ms。
    • RTF（实时率 = 合成总耗时 / 音频时长）：正常应小于 1.0。
通过新建业务空间并仅授权特定模型，可限制 API Key 的使用范围。请参见业务空间。
默认业务空间下，所有模型均可调用。子业务空间下，需要为 API Key 对应的子业务空间进行模型授权。请参见业务空间。