Alibaba Cloud Model Studio:Synthèse vocale en temps réel

Go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	// The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
	wsURL      = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"
	outputFile = "output.mp3"
)

func main() {
	// The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
	// If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: apiKey := "sk-xxx"
	apiKey := os.Getenv("DASHSCOPE_API_KEY")

	// Clear the output file
	os.Remove(outputFile)
	os.Create(outputFile)

	// Connect to 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("Connection failed, HTTP status code: %d\n", resp.StatusCode)
		}
		fmt.Println("Connection failed:", err)
		return
	}
	defer conn.Close()

	// Generate task ID
	taskID := uuid.New().String()
	fmt.Printf("Generated task ID: %s\n", taskID)

	// Sending run-task event
	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":      "qwen-audio-3.0-tts-flash",
			"parameters": map[string]interface{}{
				"text_type":   "PlainText",
				"voice":       "longanhuan_v3.6",
				"format":      "mp3",
				"sample_rate": 22050,
				"volume":      50,
				"rate":        1,
				"pitch":       1,
				// If enable_ssml is set to true, only one continue-task event can be sent; otherwise the error "Text request limit violated, expected 1." will be returned
				"enable_ssml": false,
			},
			"input": map[string]interface{}{},
		},
	}

	runTaskJSON, _ := json.Marshal(runTaskCmd)
	fmt.Printf("Sending run-task event: %s\n", string(runTaskJSON))

	err = conn.WriteMessage(websocket.TextMessage, runTaskJSON)
	if err != nil {
		fmt.Println("Failed to send run-task:", err)
		return
	}

	textSent := false

	// Process messages
	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			fmt.Println("Failed to read message:", err)
			break
		}

		// Process binary message
		if messageType == websocket.BinaryMessage {
			fmt.Printf("Received binary message, length: %d\n", len(message))
			file, _ := os.OpenFile(outputFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0644)
			file.Write(message)
			file.Close()
			continue
		}

		// Process text message
		messageStr := string(message)
		fmt.Printf("Received text message: %s\n", strings.ReplaceAll(messageStr, "\n", ""))

		// Parse JSON to get event type
		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("Event type: %s\n", event)

					switch event {
					case "task-started":
						fmt.Println("=== Received task-started event ===")

						if !textSent {
							// Sending continue-task event

							texts := []string{"Before my bed, moonlight shines bright, I suspect it's frost upon the ground.", "I raise my eyes to gaze at the bright moon, then bow my head, thinking of home."}

							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("Sending continue-task event: %s\n", string(continueTaskJSON))

								err = conn.WriteMessage(websocket.TextMessage, continueTaskJSON)
								if err != nil {
									fmt.Println("Failed to send continue-task:", err)
									return
								}
							}

							textSent = true

							// Delay before sending finish-task
							time.Sleep(500 * time.Millisecond)

							// Sending finish-task event
							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("Sending finish-task event: %s\n", string(finishTaskJSON))

							err = conn.WriteMessage(websocket.TextMessage, finishTaskJSON)
							if err != nil {
								fmt.Println("Failed to send finish-task:", err)
								return
							}
						}

					case "task-finished":
						fmt.Println("=== Task completed ===")
						return

					case "task-failed":
						fmt.Println("=== Task failed ===")
						if header["error_message"] != nil {
							fmt.Printf("Error message: %s\n", header["error_message"])
						}
						return

					case "result-generated":
						fmt.Println("Received result-generated event")
					}
				}
			}
		}
	}
}

C#

using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

class Program {
    // The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
    // If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: 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.");

    // The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
    private const string WebSocketUrl = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference";
    // Output file path
    private const string OutputFilePath = "output.mp3";

    // WebSocket client
    private static ClientWebSocket _webSocket = new ClientWebSocket();
    // Cancellation token source
    private static CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    // Task ID
    private static string? _taskId;
    // Whether the task has started
    private static TaskCompletionSource<bool> _taskStartedTcs = new TaskCompletionSource<bool>();

    static async Task Main(string[] args) {
        try {
            // Clear the output file
            ClearOutputFile(OutputFilePath);

            // Connect to the WebSocket service
            await ConnectToWebSocketAsync(WebSocketUrl);

            // Start the message receiving task
            Task receiveTask = ReceiveMessagesAsync();

            // Send the run-task event
            _taskId = GenerateTaskId();
            await SendRunTaskCommandAsync(_taskId);

            // Wait for the task-started event
            await _taskStartedTcs.Task;

            // Send continue-task events
            string[] texts = {
                "Before my bed, moonlight shines bright,",
                "I suspect it\'s frost upon the ground.",
                "I raise my eyes to gaze at the bright moon,",
                "then bow my head, thinking of home."
            };
            foreach (string text in texts) {
                await SendContinueTaskCommandAsync(text);
            }

            // Send the finish-task event
            await SendFinishTaskCommandAsync(_taskId);

            // Wait for the receiving task to complete
            await receiveTask;

            Console.WriteLine("Task completed, connection closed.");
        } catch (OperationCanceledException) {
            Console.WriteLine("Task cancelled.");
        } catch (Exception ex) {
            Console.WriteLine($"Error occurred: {ex.Message}");
        } finally {
            _cancellationTokenSource.Cancel();
            _webSocket.Dispose();
        }
    }

    private static void ClearOutputFile(string filePath) {
        if (File.Exists(filePath)) {
            File.WriteAllText(filePath, string.Empty);
            Console.WriteLine("Output file cleared.");
        } else {
            Console.WriteLine("Output file does not exist, no need to clear.");
        }
    }

    private static async Task ConnectToWebSocketAsync(string url) {
        var uri = new Uri(url);
        if (_webSocket.State == WebSocketState.Connecting || _webSocket.State == WebSocketState.Open) {
            return;
        }

        // Set the WebSocket connection headers
        _webSocket.Options.SetRequestHeader("Authorization", $"bearer {ApiKey}");
        _webSocket.Options.SetRequestHeader("X-DashScope-DataInspection", "enable");

        try {
            await _webSocket.ConnectAsync(uri, _cancellationTokenSource.Token);
            Console.WriteLine("Successfully connected to the WebSocket service.");
        } catch (OperationCanceledException) {
            Console.WriteLine("WebSocket connection cancelled.");
        } catch (Exception ex) {
            Console.WriteLine($"WebSocket connection failed: {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 = "qwen-audio-3.0-tts-flash",
            parameters = new
            {
                text_type = "PlainText",
                voice = "longanhuan_v3.6",
                format = "mp3",
                sample_rate = 22050,
                volume = 50,
                rate = 1,
                pitch = 1,
                // If enable_ssml is set to true, only one continue-task event can be sent; otherwise the error "Text request limit violated, expected 1." will be returned
                enable_ssml = false
            },
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("run-task event sent.");
    }

    private static async Task SendContinueTaskCommandAsync(string text) {
        if (_taskId == null) {
            throw new InvalidOperationException("Task ID not initialized.");
        }

        var command = CreateCommand("continue-task", _taskId, "duplex", new {
            input = new {
                text
            }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("continue-task event sent.");
    }

    private static async Task SendFinishTaskCommandAsync(string taskId) {
        var command = CreateCommand("finish-task", taskId, "duplex", new {
            input = new { }
        });

        await SendJsonMessageAsync(command);
        Console.WriteLine("finish-task event sent.");
    }

    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("Message sending cancelled.");
        }
    }

    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("Task started.");
                        _taskStartedTcs.TrySetResult(true);
                        break;
                    case "task-finished":
                        Console.WriteLine("Task completed.");
                        _cancellationTokenSource.Cancel();
                        break;
                    case "task-failed":
                        Console.WriteLine("Task failed: " + response.RootElement.GetProperty("header").GetProperty("error_message").GetString());
                        _cancellationTokenSource.Cancel();
                        break;
                    default:
                        // result-generated can be handled here
                        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) {
                // Process binary data
                Console.WriteLine("Received binary data...");

                // Save the binary data to file
                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("Message receiving cancelled.");
            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);
    }
}

PHP

Structure du répertoire de code exemple :

my-php-project/

├── composer.json

├── vendor/

└── index.php

Contenu du fichier composer.json (adaptez les versions des dépendances selon vos besoins) :

{
    "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/"
        }
    }
}

Contenu du fichier index.php :

<?php

require __DIR__ . '/vendor/autoload.php';

use Ratchet\Client\Connector;
use React\EventLoop\Loop;
use React\Socket\Connector as SocketConnector;

// The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
// If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: $api_key = "sk-xxx"
$api_key = getenv("DASHSCOPE_API_KEY");
// The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
$websocket_url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'; // WebSocket server URL
$output_file = 'output.mp3'; // Output file path

$loop = Loop::get();

if (file_exists($output_file)) {
    // Clear file content
    file_put_contents($output_file, '');
}

// Create a custom connector
$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 "Connected to the WebSocket server\n";

    // Generate task ID
    $taskId = generateTaskId();

    // Send the run-task event
    sendRunTaskMessage($conn, $taskId);

    // Define the function to send continue-task events
    $sendContinueTask = function() use ($conn, $loop, $taskId) {
        // Text to be sent
        $texts = ["Before my bed, moonlight shines bright,", "I suspect it\'s frost upon the ground.", "I raise my eyes to gaze at the bright moon,", "then bow my head, thinking of home."];
        $continueTaskCount = 0;
        foreach ($texts as $text) {
            $continueTaskMessage = json_encode([
                "header" => [
                    "action" => "continue-task",
                    "task_id" => $taskId,
                    "streaming" => "duplex"
                ],
                "payload" => [
                    "input" => [
                        "text" => $text
                    ]
                ]
            ]);
            echo "Sending continue-task event: " . $continueTaskMessage . "\n";
            $conn->send($continueTaskMessage);
            $continueTaskCount++;
        }
        echo "Number of continue-task events sent: " . $continueTaskCount . "\n";

        // Send the finish-task event
        sendFinishTaskMessage($conn, $taskId);
    };

    // Flag for whether the task-started event was received
    $taskStarted = false;

    // Listen for messages
    $conn->on('message', function($msg) use ($conn, $sendContinueTask, $loop, &$taskStarted, $taskId, $output_file) {
        if ($msg->isBinary()) {
            // Write binary data to local file
            file_put_contents($output_file, $msg->getPayload(), FILE_APPEND);
        } else {
            // Process non-binary message
            $response = json_decode($msg, true);

            if (isset($response['header']['event'])) {
                handleEvent($conn, $response, $sendContinueTask, $loop, $taskId, $taskStarted);
            } else {
                echo "Unknown message format\n";
            }
        }
    });

    // Listen for connection close
    $conn->on('close', function($code = null, $reason = null) {
        echo "Connection closed\n";
        if ($code !== null) {
            echo "Close code: " . $code . "\n";
        }
        if ($reason !== null) {
            echo "Close reason: " . $reason . "\n";
        }
    });
}, function ($e) {
    echo "Unable to connect: {$e->getMessage()}\n";
});

$loop->run();

/**
 * Generate task ID
 * @return string
 */
function generateTaskId(): string {
    return bin2hex(random_bytes(16));
}

/**
 * Send the run-task event
 * @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" => "qwen-audio-3.0-tts-flash",
            "parameters" => [
                "text_type" => "PlainText",
                "voice" => "longanhuan_v3.6",
                "format" => "mp3",
                "sample_rate" => 22050,
                "volume" => 50,
                "rate" => 1,
                "pitch" => 1,
                // If enable_ssml is set to true, only one continue-task event can be sent; otherwise the error "Text request limit violated, expected 1." will be returned
                "enable_ssml" => false
            ],
            "input" => (object) []
        ]
    ]);
    echo "Sending run-task event: " . $runTaskMessage . "\n";
    $conn->send($runTaskMessage);
    echo "run-task event sent\n";
}

/**
 * Read audio file
 * @param string $filePath
 * @return bool|string
 */
function readAudioFile(string $filePath) {
    $voiceData = file_get_contents($filePath);
    if ($voiceData === false) {
        echo "Unable to read audio file\n";
    }
    return $voiceData;
}

/**
 * Split audio data
 * @param string $data
 * @param int $chunkSize
 * @return array
 */
function splitAudioData(string $data, int $chunkSize): array {
    return str_split($data, $chunkSize);
}

/**
 * Send the finish-task event
 * @param $conn
 * @param $taskId
 */
function sendFinishTaskMessage($conn, $taskId) {
    $finishTaskMessage = json_encode([
        "header" => [
            "action" => "finish-task",
            "task_id" => $taskId,
            "streaming" => "duplex"
        ],
        "payload" => [
            "input" => (object) []
        ]
    ]);
    echo "Sending finish-task event: " . $finishTaskMessage . "\n";
    $conn->send($finishTaskMessage);
    echo "finish-task event sent\n";
}

/**
 * Handle events
 * @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 "Task started, sending continue-task events...\n";
            $taskStarted = true;
            // Send continue-task events
            $sendContinueTask();
            break;
        case 'result-generated':
            // Received result-generated event
            break;
        case 'task-finished':
            echo "Task completed\n";
            $conn->close();
            break;
        case 'task-failed':
            echo "Task failed\n";
            echo "Error code: " . $response['header']['error_code'] . "\n";
            echo "Error message: " . $response['header']['error_message'] . "\n";
            $conn->close();
            break;
        case 'error':
            echo "Error: " . $response['payload']['message'] . "\n";
            break;
        default:
            echo "Unknown event: " . $response['header']['event'] . "\n";
            break;
    }

    // If the task is completed, close the connection
    if ($response['header']['event'] == 'task-finished') {
        // Wait 1 second to ensure all data has been transmitted
        $loop->addTimer(1, function() use ($conn) {
            $conn->close();
            echo "Client closing connection\n";
        });
    }

    // If the task-started event has not been received, close the connection
    if (!$taskStarted && in_array($response['header']['event'], ['task-failed', 'error'])) {
        $conn->close();
    }
}

Node.js

Installez les dépendances requises :

npm install ws
npm install uuid

Code exemple :

const WebSocket = require('ws');
const fs = require('fs');
const uuid = require('uuid').v4;

// The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
// If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: const apiKey = "sk-xxx"
const apiKey = process.env.DASHSCOPE_API_KEY;
// The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
const url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference';
// Output file path
const outputFilePath = 'output.mp3';

// Clear output file
fs.writeFileSync(outputFilePath, '');

// Create WebSocket client
const ws = new WebSocket(url, {
  headers: {
    Authorization: `bearer ${apiKey}`,
    'X-DashScope-DataInspection': 'enable'
  }
});

let taskStarted = false;
let taskId = uuid();

ws.on('open', () => {
  console.log('Connected to the WebSocket server');

  // Send the run-task event
  const runTaskMessage = JSON.stringify({
    header: {
      action: 'run-task',
      task_id: taskId,
      streaming: 'duplex'
    },
    payload: {
      task_group: 'audio',
      task: 'tts',
      function: 'SpeechSynthesizer',
      model: 'qwen-audio-3.0-tts-flash',
      parameters: {
        text_type: 'PlainText',
        voice: 'longanhuan_v3.6', // Voice
        format: 'mp3', // Audio format
        sample_rate: 22050, // Sample rate
        volume: 50, // Volume
        rate: 1, // Speech rate
        pitch: 1, // Pitch
        enable_ssml: false // Whether to enable SSML. If enable_ssml is set to true, only one continue-task event can be sent; otherwise the error "Text request limit violated, expected 1." will be returned
      },
      input: {}
    }
  });
  ws.send(runTaskMessage);
  console.log('run-task message sent');
});

const fileStream = fs.createWriteStream(outputFilePath, { flags: 'a' });
ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // Write binary data to file
    fileStream.write(data);
  } else {
    const message = JSON.parse(data);

    switch (message.header.event) {
      case 'task-started':
        taskStarted = true;
        console.log('Task started');
        // Send continue-task events
        sendContinueTasks(ws);
        break;
      case 'task-finished':
        console.log('Task completed');
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      case 'task-failed':
        console.error('Task failed: ', message.header.error_message);
        ws.close();
        fileStream.end(() => {
          console.log('File stream closed');
        });
        break;
      default:
        // result-generated can be handled here
        break;
    }
  }
});

function sendContinueTasks(ws) {
  const texts = [
    'Before my bed, moonlight shines bright,',
    'I suspect it is frost upon the ground.',
    'I raise my eyes to gaze at the bright moon,',
    'then bow my head, thinking of home.'
  ];

  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 sent, text: ${text}`);
      }
    }, index * 1000); // Send one every second
  });

  // Send the finish-task event
  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 sent');
    }
  }, texts.length * 1000 + 1000); // Send 1 second after all continue-task events are sent
}

ws.on('close', () => {
  console.log('Disconnected from the WebSocket server');
});

Java

Nous vous recommandons d'utiliser le SDK DashScope Java pour vos développements. Consultez la rubrique SDK Java.

L'exemple ci-dessous illustre une connexion WebSocket directe en Java. Importez les dépendances suivantes avant l'exécution :

• Java-WebSocket
• jackson-databind

Gérez les dépendances avec Maven ou 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>

// Other code omitted
dependencies {
  // WebSocket Client
  implementation 'org.java-websocket:Java-WebSocket:1.5.3'
  // JSON Processing
  implementation 'com.fasterxml.jackson.core:jackson-databind:2.13.0'
}
// Other code omitted

Code 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("Connection established");

        // Send the run-task event
        // If enable_ssml is set to true, only one continue-task event can be sent; otherwise the error "Text request limit violated, expected 1." will be returned
        String runTaskCommand = "{ \"header\": { \"action\": \"run-task\", \"task_id\": \"" + taskId + "\", \"streaming\": \"duplex\" }, \"payload\": { \"task_group\": \"audio\", \"task\": \"tts\", \"function\": \"SpeechSynthesizer\", \"model\": \"qwen-audio-3.0-tts-flash\", \"parameters\": { \"text_type\": \"PlainText\", \"voice\": \"longanhuan_v3.6\", \"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("Received message from server: " + 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("Received task-started event from server");

                        List<String> texts = Arrays.asList(
                                "Before my bed, moonlight shines bright,I suspect it\'s frost upon the ground.",
                                "I raise my eyes to gaze at the bright moon,then bow my head, thinking of home."
                        );

                        for (String text : texts) {
                            // Send the continue-task event
                            sendContinueTask(text);
                        }

                        // Send the finish-task event
                        sendFinishTask();
                    } else if ("task-finished".equals(event)) {
                        System.out.println("Received task-finished event from server");
                        taskFinished = true;
                        closeConnection();
                    } else if ("task-failed".equals(event)) {
                        System.out.println("Task failed: " + message);
                        closeConnection();
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("Exception occurred: " + e.getMessage());
        }
    }

    @Override
    public void onMessage(ByteBuffer message) {
        System.out.println("Received binary audio data of size: " + message.remaining());

        try (FileOutputStream fos = new FileOutputStream(outputFile, true)) {
            byte[] buffer = new byte[message.remaining()];
            message.get(buffer);
            fos.write(buffer);
            System.out.println("Audio data written to local file " + outputFile);
        } catch (IOException e) {
            System.err.println("Failed to write audio data to local file: " + e.getMessage());
        }
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("Connection closed: " + reason + " (" + code + ")");
    }

    @Override
    public void onError(Exception ex) {
        System.err.println("Error: " + 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 {
            // The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
            // If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: String apiKey = "sk-xxx"
            String apiKey = System.getenv("DASHSCOPE_API_KEY");
            if (apiKey == null || apiKey.isEmpty()) {
                System.err.println("Please set the DASHSCOPE_API_KEY environment variable");
                return;
            }

            Map<String, String> headers = new HashMap<>();
            headers.put("Authorization", "bearer " + apiKey);
            // The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
            TTSWebSocketClient client = new TTSWebSocketClient(new URI("wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"), headers);

            client.connect();

            while (!client.isClosed() && !client.taskFinished) {
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            System.err.println("Failed to connect to WebSocket service: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Python

Nous recommandons d'utiliser le SDK DashScope pour Python lors du développement. Consultez SDK Python.

L'exemple ci-dessous illustre une connexion directe via WebSocket en Python. Installez les dépendances suivantes avant l'exécution :

pip uninstall websocket-client
pip uninstall websocket
pip install websocket-client

import websocket
import json
import uuid
import os
import time

class TTSClient:
    def __init__(self, api_key, uri):
        """
    Initialize TTSClient instance

    Parameters:
        api_key (str): API Key for authentication
        uri (str): WebSocket service URL
    """
        self.api_key = api_key  # Replace with your API Key
        self.uri = uri  # Replace with your WebSocket URL
        self.task_id = str(uuid.uuid4())  # Generate unique task ID
        self.output_file = f"output_{int(time.time())}.mp3"  # Output audio file path
        self.ws = None  # WebSocketApp instance
        self.task_started = False  # Whether task-started has been received
        self.task_finished = False  # Whether task-finished / task-failed has been received

    def on_open(self, ws):
        """
    Callback when WebSocket connection is established
    Sends run-task event to start speech synthesis task
    """
        print("WebSocket connected")

        # Construct run-task event
        run_task_cmd = {
            "header": {
                "action": "run-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "task_group": "audio",
                "task": "tts",
                "function": "SpeechSynthesizer",
                "model": "qwen-audio-3.0-tts-flash",
                "parameters": {
                    "text_type": "PlainText",
                    "voice": "longanhuan_v3.6",
                    "format": "mp3",
                    "sample_rate": 22050,
                    "volume": 50,
                    "rate": 1,
                    "pitch": 1,
                    # If enable_ssml is set to true, only one continue-task event can be sent; otherwise an error will be returned
                    "enable_ssml": False
                },
                "input": {}
            }
        }

        # Send run-task event
        ws.send(json.dumps(run_task_cmd))
        print("Sent run-task event")

    def on_message(self, ws, message):
        """
    Callback when a message is received
    Handles text and binary messages differently
    """
        if isinstance(message, str):
            # Handle JSON text messages
            try:
                msg_json = json.loads(message)
                print(f"Received JSON message: {msg_json}")

                if "header" in msg_json:
                    header = msg_json["header"]

                    if "event" in header:
                        event = header["event"]

                        if event == "task-started":
                            print("Task started")
                            self.task_started = True

                            # Send continue-task events
                            texts = [
                                "Before my bed, moonlight shines bright,I suspect it\'s frost upon the ground.",
                                "I raise my eyes to gaze at the bright moon,then bow my head, thinking of home."
                            ]

                            for text in texts:
                                self.send_continue_task(text)

                            # Send finish-task after all continue-task events are sent
                            self.send_finish_task()

                        elif event == "task-finished":
                            print("Task completed")
                            self.task_finished = True
                            self.close(ws)

                        elif event == "task-failed":
                            error_msg = msg_json.get("error_message", "Unknown error")
                            print(f"Task failed: {error_msg}")
                            self.task_finished = True
                            self.close(ws)

            except json.JSONDecodeError as e:
                print(f"JSON parsing failed: {e}")
        else:
            # Handle binary messages (audio data)
            print(f"Received binary message, size: {len(message)} bytes")
            with open(self.output_file, "ab") as f:
                f.write(message)
            print(f"Audio data written to local file {self.output_file}")

    def on_error(self, ws, error):
        """Callback when an error occurs"""
        print(f"WebSocket error: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        """Callback when connection is closed"""
        print(f"WebSocket closed: {close_msg} ({close_status_code})")

    def send_continue_task(self, text):
        """Send continue-task event with text content to be synthesized"""
        cmd = {
            "header": {
                "action": "continue-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {
                    "text": text
                }
            }
        }

        self.ws.send(json.dumps(cmd))
        print(f"Sent continue-task event, text content: {text}")

    def send_finish_task(self):
        """Send finish-task event to end the speech synthesis task"""
        cmd = {
            "header": {
                "action": "finish-task",
                "task_id": self.task_id,
                "streaming": "duplex"
            },
            "payload": {
                "input": {}
            }
        }

        self.ws.send(json.dumps(cmd))
        print("Sent finish-task event")

    def close(self, ws):
        """Actively close the connection"""
        if ws and ws.sock and ws.sock.connected:
            ws.close()
            print("Connection closed actively")

    def run(self):
        """Start the WebSocket client"""
        # Set request headers (authentication)
        header = {
            "Authorization": f"bearer {self.api_key}",
            "X-DashScope-DataInspection": "enable"
        }

        # Create WebSocketApp instance
        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("Listening for WebSocket messages...")
        self.ws.run_forever()  # Start persistent connection listener

# Example usage
if __name__ == "__main__":
    # The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
    # If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: API_KEY = "sk-xxx"
    API_KEY = os.environ.get("DASHSCOPE_API_KEY")
    # The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
    SERVER_URI = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference"  # Replace with your WebSocket URL

    client = TTSClient(API_KEY, SERVER_URI)
    client.run()

SDK Python

Le SDK Python utilise SpeechSynthesizerObjectPool pour gérer et réutiliser les objets SpeechSynthesizer.

Ce pool crée un nombre défini d'instances SpeechSynthesizer et établit les connexions WebSocket lors de son initialisation. Lorsque vous empruntez un objet, celui-ci est prêt à envoyer immédiatement des requêtes, ce qui réduit la latence du premier paquet. Une fois l'objet restitué, la connexion reste active pour la tâche suivante.

Étapes de mise en œuvre

1. Installez les dépendances : installez DashScope (pip install -U dashscope).

2. Créez et configurez le pool d'objets

   Fixez la taille du pool à 1.5x-2x le pic de concurrence, sans dépasser la limite QPS du compte.

   Créez un pool singleton global (l'établissement des connexions lors de l'initialisation prend un certain temps) :

from dashscope.audio.tts_v2 import SpeechSynthesizerObjectPool

synthesizer_object_pool = SpeechSynthesizerObjectPool(max_size=20)
import dashscope
# The following is the configuration for the China (Beijing) region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

3. Empruntez un objet SpeechSynthesizer au pool

   Si le nombre d'objets non restitués dépasse la capacité du pool, le système crée des objets supplémentaires.

   Ces objets additionnels doivent établir de nouvelles connexions et ne bénéficient pas des avantages du pool.

speech_synthesizer = connectionPool.borrow_synthesizer(
    model='cosyvoice-v3-flash',
    voice='longanyang',
    seed=12382,
    callback=synthesizer_callback
)

4. Effectuez la synthèse vocale

   Appelez la méthode call ou streaming_call de l'objet SpeechSynthesizer pour synthétiser la voix.

5. Restituez l'objet SpeechSynthesizer

   Renvoyez l'objet une fois la tâche terminée afin qu'il puisse être réutilisé.

   Ne restituez pas les objets dont la tâche a échoué ou n'est pas terminée.

connectionPool.return_synthesizer(speech_synthesizer)

Code complet

# !/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 = [
    'Sentence 1: Welcome to Alibaba speech synthesis service.',
    'Sentence 2: Welcome to Alibaba speech synthesis service.',
    'Sentence 3: Welcome to Alibaba speech synthesis service.',
]
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
    '''
    # The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
    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.
    for more information, please refer to https://www.alibabacloud.com/help/document_detail/2712523.html
    '''

    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__':
    # You must set dashscope.api_key and base_websocket_api_url before creating SpeechSynthesizerObjectPool.
    # The pool establishes WebSocket connections using the current global dashscope.api_key at initialization time.
    # Modifying dashscope.api_key after pool creation will not affect existing connections in the pool.
    # The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
    dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.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()

Gestion des ressources et des erreurs

• Réussite : après la fin normale de la synthèse, appelez connectionPool.return_synthesizer(speech_synthesizer) pour restituer l'objet SpeechSynthesizer au pool et le réutiliser.

  ImportantNe restituez pas les objets SpeechSynthesizer dont la tâche a échoué ou n'est pas terminée.

• Échec : si une erreur interne du SDK ou une exception métier interrompt la tâche, fermez la connexion WebSocket sous-jacente : speech_synthesizer.close()

• Lorsque toutes les synthèses sont terminées, arrêtez le pool : connectionPool.shutdown()

• Une erreur TaskFailed côté serveur ne nécessite aucun traitement supplémentaire.

SDK Java

Le SDK Java offre des performances optimales grâce à la coordination d'un pool de connexions intégré et d'un pool d'objets personnalisé.

• Pool de connexions : le pool OkHttp3 intégré au SDK gère et réutilise les connexions WebSocket, réduisant le coût des négociations réseau. Il est activé par défaut.
• Pool d'objets : fondé sur commons-pool2, il maintient des objets SpeechSynthesizer préconnectés. Leur emprunt élimine le délai d'établissement de connexion et réduit nettement la latence du premier paquet.

Étapes de mise en œuvre

1. Ajouter des dépendances

   Ajoutez dashscope-sdk-java et commons-pool2 à la configuration des dépendances de votre projet selon l'outil de build utilisé.

   Exemples pour Maven et Gradle :

   Maven

   1. Ouvrez le fichier pom.xml du projet Maven.
   2. Ajoutez les dépendances suivantes dans la balise <dependencies>.

   <dependency>
       <groupId>com.alibaba</groupId>
       <artifactId>dashscope-sdk-java</artifactId>
       <!-- Replace 'the-latest-version' with version 2.16.9 or later. Check available versions at: https://mvnrepository.com/artifact/com.alibaba/dashscope-sdk-java -->
       <version>the-latest-version</version>
   </dependency>
   
   <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-pool2</artifactId>
       <!-- Replace 'the-latest-version' with the latest version. Check available versions at: https://mvnrepository.com/artifact/org.apache.commons/commons-pool2 -->
       <version>the-latest-version</version>
   </dependency>
   

   1. Enregistrez le fichier pom.xml.
   2. Exécutez une commande Maven telle que mvn clean install ou mvn compile pour actualiser les dépendances.

   Gradle

   1. Ouvrez le fichier build.gradle du projet Gradle.
   2. Ajoutez les dépendances suivantes dans le bloc dependencies.

   dependencies {
       // Replace 'the-latest-version' with version 2.16.6 or later. Check available versions at: https://mvnrepository.com/artifact/com.alibaba/dashscope-sdk-java
       implementation group: 'com.alibaba', name: 'dashscope-sdk-java', version: 'the-latest-version'
   
       // Replace 'the-latest-version' with the latest version. Check available versions at: https://mvnrepository.com/artifact/org.apache.commons/commons-pool2
       implementation group: 'org.apache.commons', name: 'commons-pool2', version: 'the-latest-version'
   }
   

   3. Enregistrez le fichier build.gradle.
   4. Dans le terminal, placez-vous à la racine du projet et exécutez la commande Gradle suivante pour actualiser les dépendances.

   ./gradlew build --refresh-dependencies
   

   Sous Windows, utilisez plutôt la commande suivante :

   gradlew build --refresh-dependencies
   

2. Configurer le pool de connexions

   Définissez les paramètres clés du pool de connexions via des variables d'environnement :

   Variable d'environnement	Description
   DASHSCOPE_CONNECTION_POOL_SIZE	Taille du pool de connexions. Valeur recommandée: au moins 2x le pic de concurrence. Valeur par défaut: 32.
   DASHSCOPE_MAXIMUM_ASYNC_REQUESTS	Nombre maximal de requêtes asynchrones. Valeur recommandée: identique à DASHSCOPE_CONNECTION_POOL_SIZE. Valeur par défaut: 32.
   DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST	Nombre maximal de requêtes asynchrones par hôte. Valeur recommandée: identique à DASHSCOPE_CONNECTION_POOL_SIZE. Valeur par défaut: 32.

3. Configurer le pool d'objets

   Définissez la taille du pool d'objets via des variables d'environnement :

   Variable d'environnement	Description
   COSYVOICE_OBJECTPOOL_SIZE	Taille du pool d'objets. Valeur recommandée: 1.5x-2x le pic de concurrence. Valeur par défaut: 500.

   Important

   • La taille du pool d'objets (COSYVOICE_OBJECTPOOL_SIZE) doit être inférieure ou égale à celle du pool de connexions (DASHSCOPE_CONNECTION_POOL_SIZE). Sinon, lorsque ce dernier est saturé, les threads appelants se bloquent dans l'attente d'une connexion disponible.
   • La taille du pool d'objets ne doit pas dépasser la limite QPS (requêtes par seconde) du compte.

   Créez le pool d'objets à l'aide du code suivant :

class CosyvoiceObjectPool {
    // ... Other code is omitted here. For the complete example, see the full code.
    public static GenericObjectPool<SpeechSynthesizer> getInstance() {
        lock.lock();
        if (synthesizerPool == null) {
            // You can set the object pool size here, or set it in the environment variable COSYVOICE_OBJECTPOOL_SIZE.
            // It is recommended to set it to 1.5 to 2 times the maximum concurrent connections of the server.
            int objectPoolSize = getObjectivePoolSize();
            SpeechSynthesizerObjectFactory speechSynthesizerObjectFactory =
                    new SpeechSynthesizerObjectFactory();
            GenericObjectPoolConfig<SpeechSynthesizer> config =
                    new GenericObjectPoolConfig<>();
            config.setMaxTotal(objectPoolSize);
            config.setMaxIdle(objectPoolSize);
            config.setMinIdle(objectPoolSize);
            synthesizerPool =
                    new GenericObjectPool<>(speechSynthesizerObjectFactory, config);
        }
        lock.unlock();
        return synthesizerPool;
    }
}

4. Empruntez un objet SpeechSynthesizer au pool

   Si le nombre d'objets non restitués dépasse la capacité maximale du pool, le système crée un objet SpeechSynthesizer supplémentaire.

   Ces nouveaux objets nécessitent une réinitialisation ainsi qu'une nouvelle connexion WebSocket ; ils ne bénéficient donc pas des avantages du pooling.

synthesizer = CosyvoiceObjectPool.getInstance().borrowObject();

5. Effectuez la synthèse vocale

   Après avoir emprunté un objet SpeechSynthesizer depuis le pool, appelez updateParamAndCallback(param, callback) pour lier les paramètres et le callback de la tâche en cours, puis appelez streamingCall ou call pour synthétiser la voix.

   Important

   • Dans un pool d'objets, updateParamAndCallback est appelé plusieurs fois : à chaque emprunt, pour définir le callback et les paramètres de tâche tels que voice et format. La valeur de apiKey doit rester identique à chaque appel. updateParamAndCallback ne met à jour que les champs locaux de l'instance SpeechSynthesizer sans recréer la connexion WebSocket. Le SDK écrit apiKey dans l'en-tête Authorization uniquement pendant la négociation WebSocket. Les messages ultérieurs, tels que run-task, ne contiennent pas apiKey. Tant que la connexion réutilisée reste ouverte, une nouvelle valeur de apiKey n'est pas transmise au serveur : la requête utilise toujours apiKey de la négociation initiale, ce qui peut affecter l'identité, les quotas ou l'imputation des frais.
   • Pour utiliser plusieurs clés API, maintenez des pools d'objets distincts pour chaque clé.

6. Restituez l'objet SpeechSynthesizer

   Une fois la tâche de synthèse terminée, restituez l'objet SpeechSynthesizer afin que les tâches suivantes puissent le réutiliser.

   Ne restituez pas les objets dont la tâche a échoué ou n'est pas terminée.

CosyvoiceObjectPool.getInstance().returnObject(synthesizer);

Code complet

import com.alibaba.dashscope.audio.tts.SpeechSynthesisResult;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisAudioFormat;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisParam;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesizer;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.alibaba.dashscope.utils.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * You need to include the org.apache.commons.pool2 and DashScope packages in your project.
 *
 * DashScope SDK 2.16.6 and later versions are optimized for high-concurrency scenarios.
 * Versions earlier than DashScope SDK 2.16.6 are not recommended for high-concurrency use.
 *
 *
 * Before making high-concurrency calls to the TTS service,
 * please configure the connection pool parameters through the following environment variables.
 *
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS
 * DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST
 * DASHSCOPE_CONNECTION_POOL_SIZE
 *
 */

class SpeechSynthesizerObjectFactory
        extends BasePooledObjectFactory<SpeechSynthesizer> {
    public SpeechSynthesizerObjectFactory() {
        super();
    }
    @Override
    public SpeechSynthesizer create() throws Exception {
        return new SpeechSynthesizer();
    }

    @Override
    public PooledObject<SpeechSynthesizer> wrap(SpeechSynthesizer obj) {
        return new DefaultPooledObject<>(obj);
    }
}

class CosyvoiceObjectPool {
    public static GenericObjectPool<SpeechSynthesizer> synthesizerPool;
    public static String COSYVOICE_OBJECTPOOL_SIZE_ENV = "COSYVOICE_OBJECTPOOL_SIZE";
    public static int DEFAULT_OBJECT_POOL_SIZE = 500;
    private static Lock lock = new java.util.concurrent.locks.ReentrantLock();
    public static int getObjectivePoolSize() {
        try {
            Integer n = Integer.parseInt(System.getenv(COSYVOICE_OBJECTPOOL_SIZE_ENV));
            System.out.println("Using Object Pool Size In Env: "+ n);
            return n;
        } catch (NumberFormatException e) {
            System.out.println("Using Default Object Pool Size: "+ DEFAULT_OBJECT_POOL_SIZE);
            return DEFAULT_OBJECT_POOL_SIZE;
        }
    }
    public static GenericObjectPool<SpeechSynthesizer> getInstance() {
        lock.lock();
        if (synthesizerPool == null) {
            // You can set the object pool size here or in the COSYVOICE_OBJECTPOOL_SIZE environment variable.
            // It is recommended to set it to 1.5 to 2 times your server's maximum concurrent connections.
            int objectPoolSize = getObjectivePoolSize();
            SpeechSynthesizerObjectFactory speechSynthesizerObjectFactory =
                    new SpeechSynthesizerObjectFactory();
            GenericObjectPoolConfig<SpeechSynthesizer> config =
                    new GenericObjectPoolConfig<>();
            config.setMaxTotal(objectPoolSize);
            config.setMaxIdle(objectPoolSize);
            config.setMinIdle(objectPoolSize);
            synthesizerPool =
                    new GenericObjectPool<>(speechSynthesizerObjectFactory, config);
        }
        lock.unlock();
        return synthesizerPool;
    }
}

class SynthesizeTaskWithCallback implements Runnable {
    String[] textArray;
    String requestId;
    long timeCost;
    public SynthesizeTaskWithCallback(String[] textArray) {
        this.textArray = textArray;
    }
    @Override
    public void run() {
        SpeechSynthesizer synthesizer = null;
        long startTime = System.currentTimeMillis();
        // if recv onError
        final boolean[] hasError = {false};
        try {
            class ReactCallback extends ResultCallback<SpeechSynthesisResult> {
                ReactCallback() {}

                @Override
                public void onEvent(SpeechSynthesisResult message) {
                    if (message.getAudioFrame() != null) {
                        try {
                            byte[] bytesArray = message.getAudioFrame().array();
                            System.out.println("Received audio, audio stream length: " + bytesArray.length);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                    }
                }

                @Override
                public void onComplete() {}

                @Override
                public void onError(Exception e) {
                    System.out.println(e.getMessage());
                    e.printStackTrace();
                    hasError[0] = true;
                }
            }

            SpeechSynthesisParam param =
                    SpeechSynthesisParam.builder()
                            .model("cosyvoice-v3-flash")
                            .voice("longanyang")
                            // The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
                            // If you have not configured the environment variable, replace the next line with your Chinese Model Studio API Key: .apiKey("sk-xxx")
                            .apiKey(System.getenv("DASHSCOPE_API_KEY"))
                            .format(SpeechSynthesisAudioFormat
                                    .MP3_22050HZ_MONO_256KBPS) // Use PCM or MP3 for streaming synthesis
                            .build();

            try {
                synthesizer = CosyvoiceObjectPool.getInstance().borrowObject();
                // Note: In an object pool scenario, the apiKey passed in multiple calls to updateParamAndCallback must always be the same. The SDK does not update the apiKey for an established connection, so passing a different apiKey will not take effect. See the important notes in the "Perform speech synthesis" step above.
                synthesizer.updateParamAndCallback(param, new ReactCallback());
                for (String text : textArray) {
                    synthesizer.streamingCall(text);
                }
                Thread.sleep(20);
                synthesizer.streamingComplete(60000);
                requestId = synthesizer.getLastRequestId();
            } catch (Exception e) {
                System.out.println("Exception e: " + e.toString());
                hasError[0] = true;
            }
        } catch (Exception e) {
            hasError[0] = true;
            throw new RuntimeException(e);
        }
        if (synthesizer != null) {
            try {
                if (hasError[0] == true) {
                    // If an exception occurs, close the connection and invalidate the object in the pool.
                    synthesizer.getDuplexApi().close(1000, "bye");
                    CosyvoiceObjectPool.getInstance().invalidateObject(synthesizer);
                } else {
                    // If the task completes normally, return the object to the pool.
                    CosyvoiceObjectPool.getInstance().returnObject(synthesizer);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            long endTime = System.currentTimeMillis();
            timeCost = endTime - startTime;
            System.out.println("[Thread " + Thread.currentThread() + "] Speech synthesis task completed. Time cost: " + timeCost + " ms, RequestId " + requestId);
        }
    }
}

@Slf4j
public class SynthesizeTextToSpeechWithCallbackConcurrently {
    public static void checkoutEnv(String envName, int defaultSize) {
        if (System.getenv(envName) != null) {
            System.out.println("[ENV CHECK]: " + envName + " "
                    + System.getenv(envName));
        } else {
            System.out.println("[ENV CHECK]: " + envName
                    + " Using Default which is " + defaultSize);
        }
    }

    public static void main(String[] args)
            throws InterruptedException, NoApiKeyException {
        // The following is the configuration for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
        Constants.baseWebsocketApiUrl = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference";
        // Check for connection pool env
        checkoutEnv("DASHSCOPE_CONNECTION_POOL_SIZE", 32);
        checkoutEnv("DASHSCOPE_MAXIMUM_ASYNC_REQUESTS", 32);
        checkoutEnv("DASHSCOPE_MAXIMUM_ASYNC_REQUESTS_PER_HOST", 32);
        checkoutEnv(CosyvoiceObjectPool.COSYVOICE_OBJECTPOOL_SIZE_ENV, CosyvoiceObjectPool.DEFAULT_OBJECT_POOL_SIZE);

        int runTimes = 3;
        // Create the pool of SpeechSynthesis objects
        ExecutorService executorService = Executors.newFixedThreadPool(runTimes);

        for (int i = 0; i < runTimes; i++) {
            // Record the task submission time
            LocalDateTime submissionTime = LocalDateTime.now();
            executorService.submit(new SynthesizeTaskWithCallback(new String[] {
                    "Before my bed, moonlight shines bright,", "I wonder if it is frost on the ground,", "I raise my eyes to gaze at the bright moon,", "then bow my head, thinking of home."}));
        }

        // Shut down the ExecutorService and wait for all tasks to complete
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.MINUTES);
        System.exit(0);
    }
}

Configuration recommandée

La configuration suivante repose sur des tests effectués en exécutant uniquement le service de synthèse vocale Qwen-Audio-TTS/CosyVoice sur des instances ECS Alibaba Cloud aux spécifications indiquées. Une concurrence excessive risque d'accroître la latence de traitement des tâches.

La « concurrence par machine » correspond au nombre de tâches de synthèse Qwen-Audio-TTS/CosyVoice exécutées simultanément, soit l'équivalent du nombre de threads de travail.

Gestion des ressources et des erreurs

• Réussite : après la fin normale de la synthèse, appelez la méthode returnObject de GenericObjectPool pour restituer l'objet SpeechSynthesizer au pool et le réutiliser.

  Dans l'exemple de code, cela correspond à CosyvoiceObjectPool.getInstance().returnObject(synthesizer).

  ImportantNe restituez pas les objets SpeechSynthesizer dont la tâche a échoué ou n'est pas terminée.

• Échec : si une erreur interne du SDK ou une exception métier interrompt la tâche :

  1. Fermez la connexion WebSocket sous-jacente.
  2. Invalidez l'objet dans le pool pour empêcher sa réutilisation.

// In the current code, the corresponding content is as follows
// Close the connection
synthesizer.getDuplexApi().close(1000, "bye");
// Invalidate the synthesizer in the object pool when an exception occurs
CosyvoiceObjectPool.getInstance().invalidateObject(synthesizer);

• Une erreur TaskFailed côté serveur ne nécessite aucun traitement supplémentaire.

Préchauffage et mesure de la latence

Pour évaluer correctement la latence et les performances du SDK Java DashScope en contexte concurrentiel, exécutez d'abord un nombre suffisant d'opérations de préchauffage. Ainsi, vos mesures refléteront les performances en régime permanent plutôt que la surcharge initiale liée à l'établissement des connexions.

Mécanisme de réutilisation des connexions

Le SDK Java DashScope utilise un pool de connexions global singleton pour gérer et réutiliser efficacement les connexions WebSocket. Cette approche réduit la surcharge associée à la création et à la fermeture fréquentes de connexions lors de charges de travail fortement concurrentes.

Fonctionnement de ce mécanisme :

• Création à la demande : le SDK ne précrée pas les connexions au démarrage. Elles sont établies au premier appel.

• Réutilisation limitée dans le temps : après une requête, la connexion reste disponible dans le pool pendant au plus 60 secondes.

  • Si une nouvelle requête arrive dans les 60 secondes, elle réutilise la connexion sans nouvelle négociation.
  • Après plus de 60 secondes d'inactivité, la connexion se ferme automatiquement pour libérer les ressources.

Importance du préchauffage

Dans les situations suivantes, le pool peut ne contenir aucune connexion active réutilisable, obligeant ainsi le système à créer de nouvelles connexions :

• L'application vient de démarrer et n'a encore effectué aucun appel.
• Le service est resté inactif plus de 60 secondes ; les connexions du pool ont expiré et ont été fermées.

Dans ces cas de figure, la première requête doit effectuer un handshake WebSocket complet (handshake TCP, négociation TLS, mise à niveau du protocole), ce qui engendre une latence nettement supérieure à celle des requêtes suivantes qui réutilisent les connexions. Sans préchauffage préalable, les résultats des tests de performance seront faussés par cette surcharge initiale de connexion.

Latence signalée par le SDK et latence réelle du premier paquet

La latence du premier paquet indiquée par le SDK (par exemple, la valeur retournée par get_first_package_delay()) inclut le temps d'établissement de la connexion WebSocket et de transmission réseau. Elle ne correspond donc pas à la latence réelle du premier paquet traitée par le service de modèle.

La latence réelle du premier paquet représente l'intervalle entre la réception de l'instruction run-task par le serveur et le renvoi du premier événement result-generated. Vous trouverez cette valeur dans les journaux côté serveur.

En contexte de forte concurrence, la création simultanée de connexions et la planification des ressources peuvent entraîner une latence rapportée par le SDK bien supérieure à la latence réelle du premier paquet côté serveur. Si vous constatez une latence élevée du premier paquet dans les rapports du SDK :

• Comparez la latence du premier paquet dans les journaux du serveur, de run-task au premier result-generated, pour vérifier les performances d'inférence.
• Préchauffez le service avec le pool d'objets ou de connexions décrit ci-dessus. Cela élimine le coût de négociation WebSocket et rapproche la latence du SDK de celle du premier paquet.

Pratiques recommandées

Pour obtenir des données de performance fiables, suivez ces étapes de préchauffage avant de lancer des tests de charge ou de collecter des métriques de latence :

1. Simulez le niveau de concurrence visé et envoyez des requêtes de préchauffage, par exemple pendant 1-2 minutes, pour remplir le pool de connexions.
2. Vérifiez que le pool a établi et maintient suffisamment de connexions actives avant de mesurer les performances.

Un préchauffage adéquat amène le pool de connexions du SDK à un état stable de réutilisation, produisant ainsi des métriques de latence qui reflètent fidèlement les performances en régime permanent de production.

Exceptions courantes du SDK Java