Todos os produtos
Search
Central de documentação

Platform For AI:CosyVoice 2.0 API

Última atualização: Sep 09, 2026

O CosyVoice 2.0 oferece APIs para gerenciar arquivos de áudio e criar síntese de fala. Este documento descreve os tipos de API disponíveis e como chamá-los.

Preparativos

  1. Implante o service WebUI do CosyVoice2 ou um service de alto desempenho com frontend e backend separados e monte o OSS ou outro service de armazenamento para guardar os arquivos de áudio enviados. Para obter instruções detalhadas, consulte Quick start: Deploy a WebUI service ou Quick start: Deploy a high-performance service with a separate frontend and backend.

  2. Obtenha o endpoint e o token do service.

    Importante
    • Para um service de alto desempenho com frontend e backend separados, faça as chamadas de API para o service de frontend.

    • Em testes de estresse, utilize o endereço da VPC. Ele oferece uma velocidade de processamento significativamente maior do que a rede pública.

    1. Clique no nome do seu service WebUI do CosyVoice2 ou service de frontend. Na página Overview, na seção Basic Information, clique em View Endpoint Information.

    2. Na aba Shared Gateway do painel Invocation Method, obtenha o endpoint do service (EAS_SERVICE_URL) e o token (EAS_TOKEN). Remova a barra / final do endpoint do service.

      Nota
      • Para usar o endpoint público, o cliente precisa ter acesso à rede pública.

      • Para usar o endpoint da VPC, o cliente deve estar na mesma VPC do service.

  3. Prepare o arquivo de áudio.

    Este guia utiliza o seguinte áudio de referência:

    • Arquivo WAV de referência: zero_shot_prompt.wav

    • Transcrição de referência: I hope you'll do even better than me.

Lista de APIs

Arquivo de áudio de referência

  • Chamada de API

    URL da solicitação

    <EAS_SERVICE_URL>/api/v1/audio/reference_audio

    Método da solicitação

    POST

    Cabeçalhos da solicitação

    Authorization: Bearer <EAS_TOKEN>

    Parâmetros da solicitação

    • file: Obrigatório. O arquivo de áudio a ser enviado. Formatos suportados: MP3, WAV e PCM. Tipo: file. Padrão: nenhum.

    • text: Obrigatório. O conteúdo de texto do arquivo de áudio. Tipo: string.

    Parâmetros de resposta

    Retorna um objeto de áudio de referência. Para mais detalhes, consulte Response parameter list.

    Lista de parâmetros de resposta

    Parâmetro

    Tipo

    Descrição

    id

    string

    O ID do arquivo de áudio.

    filename

    string

    O nome do arquivo de áudio.

    bytes

    integer

    O tamanho do arquivo em bytes.

    created_at

    integer

    O timestamp de criação do arquivo.

    text

    string

    O conteúdo de texto associado ao arquivo.

  • Exemplo de solicitação

    cURL

    # Replace <EAS_SERVICE_URL> with your service URL and <EAS_TOKEN> with your token. 
    curl -XPOST <EAS_SERVICE_URL>/api/v1/audio/reference_audio \
        --header 'Authorization: Bearer <EAS_TOKEN>' \
        --form 'file=@"/home/xxxx/zero_shot_prompt.wav"' \
        --form 'text="I hope you will do even better than me."'

    Python

    import requests
    response = requests.post(
        "<EAS_SERVICE_URL>/api/v1/audio/reference_audio",  # Replace <EAS_SERVICE_URL> with the service URL.
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with your token.
        },
        files={
            "file": open("./zero_shot_prompt.wav", "rb"),
        },
        data={
            "text": "I hope you will do even better than me."
        }
    )
    print(response.text)
    
  • Exemplo de resposta

    {
        "id": "50a5fdb9-c3ad-445a-adbb-3be32750****",
        "filename": "zero_shot_prompt.wav",
        "bytes": 111496,
        "created_at": 1748416005,
        "text": "I hope you will do even better than me."
    }

Arquivos de áudio de referência

  • Referência da API

    Endpoint

    <EAS_SERVICE_URL>/api/v1/audio/reference_audio

    Método da solicitação

    GET

    Cabeçalhos da solicitação

    Authorization: Bearer <EAS_TOKEN>

    Parâmetros da solicitação

    • limit: Opcional. O número máximo de arquivos a retornar. Tipo: integer. Padrão: 100.

    • order: Opcional. Tipo: string. Especifica a ordem de classificação dos objetos com base no timestamp created_at. Os valores válidos são:

      • asc: Ascendente

      • desc (padrão): Descendente

    Corpo da resposta

    Contém um array de objetos de áudio de referência. Para mais detalhes, consulte Response parameters for uploading audio.

  • Exemplo de solicitação

    cURL

    # Replace <EAS_SERVICE_URL> with the service URL and <EAS_TOKEN> with your token.
    curl -XGET <EAS_SERVICE_URL>/api/v1/audio/reference_audio?limit=10&order=desc \
          --header 'Authorization: Bearer <EAS_TOKEN>'

    Python

    import requests
    response = requests.get(
        "<EAS_SERVICE_URL>/api/v1/audio/reference_audio",  # Replace <EAS_SERVICE_URL> with the service URL. 
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with your token.
        }
    )
    print(response.text)
  • Exemplo de resposta

    [
        {
            "id": "50a5fdb9-c3ad-445a-adbb-3be32750****",
            "filename": "zero_shot_prompt.wav",
            "bytes": 111496,
            "created_at": 1748416005,
            "text": "I hope you'll do even better than me."
        }
    ]

Arquivo de áudio de referência

  • Chamada de API

    URL da solicitação

    <EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id>

    Método da solicitação

    GET

    Cabeçalhos da solicitação

    Authorization: Bearer <EAS_TOKEN>

    Parâmetros de caminho

    reference_audio_id: Obrigatório. O ID do áudio de referência. Consulte List reference audio files. Tipo: String. Padrão: nenhum.

    Parâmetros de resposta

    Um objeto de áudio de referência. Para mais detalhes, consulte response parameter list.

  • Exemplo de solicitação

    cURL

    # Replace <EAS_SERVICE_URL>, <EAS_TOKEN>, and <reference_audio_id> with your service URL, token, and reference audio ID.
    curl -XGET <EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id> \
          --header 'Authorization: Bearer <EAS_TOKEN>'

    Python

    import requests
    response = requests.get(
        "<EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id>",  # Replace <EAS_SERVICE_URL> with your service URL and <reference_audio_id> with your reference audio ID.
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with your token.
        }
    )
    print(response.text)
    
  • Exemplo de resposta

    {
        "id": "50a5fdb9-c3ad-445a-adbb-3be32750****",
        "filename": "zero_shot_prompt.wav",
        "bytes": 111496,
        "created_at": 1748416005,
        "text": "I hope you'll do even better than me."
    }

Excluir arquivo de áudio de referência

  • Solicitação

    Endpoint

    <EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id>

    Método da solicitação

    DELETE

    Cabeçalhos da solicitação

    Authorization: Bearer <EAS_TOKEN>

    Parâmetros de caminho

    reference_audio_id: Obrigatório. Especifica o identificador exclusivo do áudio de referência. Para encontrar esse ID, consulte List reference audio files. Tipo: String. Padrão: nenhum.

    Parâmetros de resposta

    Retorna um objeto de áudio de referência.

  • Exemplo de solicitação

    cURL

    # Replace <EAS_SERVICE_URL> with the service endpoint, <EAS_TOKEN> with the service token, and <reference_audio_id> with the reference audio ID.
    curl -XDELETE <EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id> \
          --header 'Authorization: Bearer <EAS_TOKEN>'

    Python

    import requests
    response = requests.delete(
        "<EAS_SERVICE_URL>/api/v1/audio/reference_audio/<reference_audio_id>",  # Replace <EAS_SERVICE_URL> with the service endpoint and <reference_audio_id> with the reference audio ID.
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with the service token.
        }
    )
    print(response.text)
    
  • Exemplo de resposta

    {
        "code": "OK",
        "message": "reference audio: c0939ce0-308e-4073-918f-91ac88e3**** deleted.",
        "data": {}
    }

Criar Text-to-Speech

  • Chamada de API

    URL da solicitação

    <EAS_SERVICE_URL>/api/v1/audio/speech

    Método da solicitação

    POST

    Cabeçalhos da solicitação

    • Authorization: Bearer <EAS_TOKEN>

    • Content-Type: application/json

    Parâmetros da solicitação

    • model: Obrigatório. O nome do modelo. Atualmente, apenas CosyVoice2-0.5B é suportado. Tipo: string. Padrão: nenhum.

    • input: Obrigatório. Um objeto contendo os parâmetros de entrada. Tipo: object. Padrão: nenhum. Inclui os seguintes parâmetros:

      • mode: Obrigatório. O modo de síntese de fala. Tipo: string. Os valores válidos incluem:

        • fast_replication: replicação rápida

        • cross_lingual_replication: replicação multilíngue

        • natural_language_replication: replicação em linguagem natural

      • text: Obrigatório. O texto a ser sintetizado. Tipo: string. Padrão: nenhum.

      • reference_audio_id: Obrigatório. O ID do áudio de referência. Para obter esse ID, consulte List Reference Audio Files. Tipo: string. Padrão: nenhum.

      • instruct: Opcional. Instruções para ajustar dinamicamente o estilo da voz (por exemplo, tom, emoção ou velocidade de fala). Este parâmetro se aplica apenas quando mode está definido como natural_language_replication. Tipo: string. Padrão: nenhum.

      • sample_rate: Opcional. A taxa de amostragem. Padrão: 24000.

      • bit_rate: Opcional. A taxa de bits. Tipo: string. Padrão: 192k. Valores suportados: 16k, 32k, 48k, 64k, 128k, 192k, 256k, 320k, 384k.

      • volume: Opcional. O volume. Tipo: float. Padrão: 1.0. Por exemplo, 3.0 significa 3 vezes o volume, e 0.8 significa 80% do volume.

      • speed: Opcional. A velocidade da fala. O valor deve estar no intervalo [0.5, 2.0]. Tipo: float. Padrão: 1.0.

      • output_format: Opcional. O formato do áudio de saída. Formatos suportados: wav, mp3, pcm. Padrão: wav.

    • stream: Opcional. Se deve usar saída em streaming. Tipo: boolean. Padrão: true.

    Parâmetros de resposta

    A resposta da API. Para detalhes sobre os parâmetros de resposta, consulte List of Response Parameters.

    Lista de parâmetros de resposta

    Parâmetro

    Tipo

    Descrição

    request_id

    string

    O ID da solicitação.

    output

    string

    O conteúdo de saída.

    audio

    object

    O conteúdo de áudio.

    audio.id

    string

    O ID do áudio.

    audio.data

    string

    O fluxo de bytes WAV como uma string codificada em Base64.

    finish_reason

    string

    • Saída sem streaming: Retorna null em caso de sucesso. Em caso de falha, retorna o motivo.

    • Saída com streaming: Retorna null durante a geração do áudio. Retorna "stop" quando o modelo termina naturalmente ou atende a uma condição stop dos parâmetros de entrada.

    usage

    integer

    O tamanho do arquivo em bytes.

  • Exemplos de solicitação

    Chamada sem streaming

    cURL

    • Substitua pelo endpoint do service e pelo seu token.

    • Substitua pelo ID do áudio de referência.

    # Replace <EAS_SERVICE_URL> with the service endpoint and <EAS_TOKEN> with your token.
    # Replace <reference_audio_id> with the reference audio ID. 
    curl -XPOST <EAS_SERVICE_URL>/api/v1/audio/speech \
    --header 'Authorization: Bearer <EAS_TOKEN>' \
    --header 'Content-Type: application/json' \
    --data '{
        "model": "CosyVoice2-0.5B",
        "input": {
            "mode": "natural_language_replication",
            "reference_audio_id": "<reference_audio_id>",
            "text": "Receiving a birthday gift from a friend far away filled my heart with joy. The surprise and warm wishes made me smile.",
            "speed": 1.0,
            "output_format": "mp3",
            "sample_rate": 32000,
            "bit_rate": "48k",
            "volume": 2.0,
            "instruct": "Speak in Sichuan dialect."
        },
        "stream": false
    }'

    A API retorna um objeto JSON com os dados de áudio codificados em Base64:

    {"output":{"finish_reason":null,"audio":{"data":"DNgB9djax9su3Ba...."}},"request_id": "f90a65be-f47b-46b5-9ddc-70bae550****"}

    Python

    Instale as seguintes dependências:

    pip install requests==2.32.3 packaging==24.2
    import json
    import base64
    import requests
    from packaging import version
    required_version = "2.32.3"
    if version.parse(requests.__version__) < version.parse(required_version):
        raise RuntimeError(f"requests version must >= {required_version}")
    with requests.post(
        "<EAS_SERVICE_URL>/api/v1/audio/speech",    # Replace <EAS_SERVICE_URL> with the service endpoint.
                                                    # Example: "http://cosyvoice-frontend-test.1534081855183999.cn-hangzhou.pai-eas.aliyuncs.com/api/v1/audio/speech"
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with the service token.
            "Content-Type": "application/json",
        },
        json={
            "model": "CosyVoice2-0.5B",
            "input": {
                "mode": "natural_language_replication",
                "reference_audio_id": "<reference_audio_id>",  # Replace <reference_audio_id> with the reference audio ID.
                "text": "Receiving a birthday gift from a friend far away filled my heart with joy. The surprise and warm wishes made me smile.",
                "output_format": "mp3",
                "sample_rate": 24000,
                "speed": 1.0,
                "bit_rate": "48k",
                "volume": 2.0,
                "instruct": "Speak in Sichuan dialect."
            },
            "stream": False
        },
        timeout=10
    ) as response:
        if response.status_code != 200:
            print(response.text)
            exit()
        data = json.loads(response.content)
        encode_buffer = data['output']['audio']['data']
        decode_buffer = base64.b64decode(encode_buffer)
        with open('./http_non_stream.mp3', 'wb') as f:
            f.write(decode_buffer)
    

    Chamada com streaming

    cURL

    • Substitua pelo endpoint do service e pelo seu token.

    • Substitua pelo ID do áudio de referência.

    # Replace <EAS_SERVICE_URL> with the service endpoint and <EAS_TOKEN> with your token.
    # Replace <reference_audio_id> with the reference audio ID. 
    curl -XPOST <EAS_SERVICE_URL>/api/v1/audio/speech \
    --header 'Authorization: Bearer <EAS_TOKEN>' \
    --header 'Content-Type: application/json' \
    --data '{
        "model": "CosyVoice2-0.5B",
        "input": {
            "mode": "natural_language_replication",
            "reference_audio_id": "<reference_audio_id>",
            "text": "Receiving a birthday gift from a friend far away filled my heart with joy. The surprise and warm wishes made me smile.",
            "speed": 1.0,
            "output_format": "mp3",
            "sample_rate": 32000,
            "bit_rate": "48k",
            "volume": 2.0,
            "instruct": "Speak in Sichuan dialect."
        },
        "stream": true
    }'

    A resposta é um fluxo de Server-Sent Events (SSE) contendo blocos de áudio codificados em Base64:

    data: {"output":{"finish_reason":null,"audio":{"data":"DNgB9djax9su3Ba...."}},"request_id": "f90a65be-f47b-46b5-9ddc-70bae550****"}
    data: {"output":{"finish_reason":null,"audio":{"data":"DNgB9djax9su3Ba...."}},"request_id": "f90a65be-f47b-46b5-9ddc-70bae550****"}
    data: {"output":{"finish_reason":null,"audio":{"data":"DNgB9djax9su3Ba...."}},"request_id": "f90a65be-f47b-46b5-9ddc-70bae550****"}
    data: {"output":{"finish_reason":null,"audio":{"data":"DNgB9djax9su3Ba...."}},"request_id": "f90a65be-f47b-46b5-9ddc-70bae550****"}

    Python

    Instale o cliente SSE do Python:

    pip install requests==2.32.3 packaging==24.2 sseclient-py==1.8.0 -i http://mirrors.cloud.aliyuncs.com/pypi/simple --trusted-host mirrors.cloud.aliyuncs.com
    import io
    import json
    import base64
    import wave
    import requests
    from sseclient import SSEClient			# pip install sseclient-py
    from packaging import version
    required_version = "2.32.3"
    if version.parse(requests.__version__) < version.parse(required_version):
        raise RuntimeError(f"requests version must >= {required_version}")
    with requests.post(
        "<EAS_SERVICE_URL>/api/v1/audio/speech",    # Replace <EAS_SERVICE_URL> with the service endpoint.
                                                    # Example: "http://cosyvoice-frontend-test.1534081855183999.cn-hangzhou.pai-eas.aliyuncs.com/api/v1/audio/speech"
        headers={
            "Authorization": "Bearer <EAS_TOKEN>",  # Replace <EAS_TOKEN> with the service token.
            "Content-Type": "application/json",
        },
        json={
            "model": "CosyVoice2-0.5B",
            "input": {
                "mode": "natural_language_replication",
                "reference_audio_id": "<reference_audio_id>",  # Replace <reference_audio_id> with the reference audio ID. 
                "text": "Receiving a birthday gift from a friend far away filled my heart with joy. The surprise and warm wishes made me smile.",
                "output_format": "mp3",
                "sample_rate": 24000,
                "speed": 1.0,
                "bit_rate": "48k",
                "volume": 2.0,
                "instruct": "Speak in Sichuan dialect.",
                "debug": True
            },
            "stream": True
        },
        stream=True,
        timeout=10
    ) as response:
        if response.status_code != 200:
            print(response.text)
            exit()
        messages = SSEClient(response)
        with open('./http_stream.mp3', 'wb') as f:
            for i, msg in enumerate(messages.events()):
                print(f"Event: {msg.event}, Data: {msg.data}")
                data = json.loads(msg.data)
                if data['error'] is not None:
                    print(data['error'])
                    break
                metrics = data['metrics']
                print(f"{metrics=}")
                encode_buffer = data['output']['audio']['data']
                decode_buffer = base64.b64decode(encode_buffer)
                f.write(decode_buffer)
    

    API WebSocket

    Instale a seguinte dependência:

    pip install websocket-client==1.8.0 -i http://mirrors.cloud.aliyuncs.com/pypi/simple --trusted-host mirrors.cloud.aliyuncs.com
    #!/usr/bin/python
    # -*- coding: utf-8 -*-
    import base64
    import json
    import logging
    import sys
    import time
    import uuid
    import traceback
    import websocket
    class TTSClient:
        def __init__(self, api_key, uri, params, log_level='INFO'):
            """
        Initialize a TTSClient instance.
        Parameters:
            api_key (str): The API key used for authentication.
            uri (str): The WebSocket service endpoint.
        """
            self._api_key = api_key
            self._uri = uri  # Replace with your WebSocket endpoint.
            self._task_id = str(uuid.uuid4())  # Generate a unique task ID.
            self._ws = None  # WebSocketApp instance.
            self._task_started = False  # Whether task-started is received.
            self._task_finished = False  # Whether task-finished / task-failed is received.
            self._check_params(params)
            self._params = params
            self._chunk_metrics = []
            self._metrics = {}
            self._first_package_time = None
            self._last_time = None
            self._init_log(log_level)
            self.audio_data = b''
        def _init_log(self, log_level):
            self._log = logging.getLogger("ws_client")
            log_formatter = logging.Formatter('%(asctime)s - Process(%(process)s) - %(levelname)s - %(message)s')
            stream_handler = logging.StreamHandler(stream=sys.stdout)
            stream_handler.setFormatter(log_formatter)
            self._log.addHandler(stream_handler)
            self._log.setLevel(log_level)
        def get_metrics(self):
            """Gets the performance metrics of the synthesis task."""
            return self._metrics
        def _check_params(self, params):
            assert 'mode' in params and params['mode'] in ['fast_replication', 'cross_lingual_replication', 'natural_language_replication']
            assert 'reference_audio_id' in params
            assert 'output_format' in params and params['output_format'] in ['wav', 'mp3', 'pcm']
            if params['mode'] == 'natural_language_replication':
                assert 'instruct' in params and params['instruct']
            else:
                if 'instruct' in params:
                    del params['instruct']
        def on_open(self, ws):
            """
        Callback function for when the WebSocket connection is established.
        Sends a 'run-task' command to start the speech synthesis task.
        """
            self._log.debug("WebSocket connection established.")
            # Build the run-task command.
            run_task_cmd = {
                "header": {
                    "action": "run-task",
                    "task_id": self._task_id,
                    "streaming": "duplex"
                },
                "payload": {
                    "task_group": "audio",
                    "task": "tts",
                    "function": "SpeechSynthesizer",
                    "model": "cosyvoice-v2",
                    "parameters": {
                        "mode": self._params['mode'],
                        "reference_audio_id": self._params['reference_audio_id'],
                        "output_format": self._params.get('output_format', 'wav'),
                        "sample_rate": self._params.get('sample_rate', 24000),
                        "bit_rate": self._params.get('bit_rate', '192k'),
                        "volume": self._params.get('volume', 1.0),
                        "instruct": self._params.get('instruct', ''),
                        "speed": self._params.get('speed', 1.0),
                        "debug": True,
                    },
                    "input": {}
                }
            }
            # Send the run-task command.
            ws.send(json.dumps(run_task_cmd))
            self._log.debug("Sent the run-task command.")
        def on_message(self, ws, message):
            """
        Callback function for when a message is received.
        Handles text and binary messages.
        """
            try:
                msg_json = json.loads(message)
                # self._log.debug(f"Received JSON message: {msg_json}")
                self._log.debug(f"Received JSON message: {msg_json['header']['event']}")
                if "header" in msg_json:
                    header = msg_json["header"]
                    if "event" in header:
                        event = header["event"]
                        if event == "task-started":
                            self._log.debug("Task started.")
                            self._task_started = True
                            # Send continue-task commands.
                            for text in self._params['texts']:
                                self.send_continue_task(text)
                            # Send finish-task after all continue-task commands are sent.
                            self.send_finish_task()
                            self._last_time = time.time()
                        elif event == "result-generated":
                            metrics = msg_json['payload']['metrics']
                            cur_time = time.time()
                            metrics['client_cost_time'] = cur_time - self._last_time
                            self._last_time = cur_time
                            encode_data = msg_json["payload"]["output"]["audio"]["data"]
                            decode_data = base64.b64decode(encode_data)
                            self._log.debug(f"Received audio data. Size: {len(decode_data)} bytes.")
                            self.audio_data += decode_data
                            metrics['client_rtf'] = metrics['client_cost_time'] / metrics['speech_len']
                            self._chunk_metrics.append(metrics)
                        elif event == "task-finished":
                            self._metrics = {
                                'client_first_package_time': self._chunk_metrics[0]['client_cost_time'],
                                "client_rtf": sum([m["client_cost_time"] for m in self._chunk_metrics]) / sum([m["speech_len"] for m in self._chunk_metrics]),
                                'client_cost_time': sum([m["client_cost_time"] for m in self._chunk_metrics]),
                                'speech_len': sum([m["speech_len"] for m in self._chunk_metrics]),
                                'server_first_package_time': self._chunk_metrics[0]['server_cost_time'],
                                'server_rtf': sum([m["server_cost_time"] for m in self._chunk_metrics]) / sum([m["speech_len"] for m in self._chunk_metrics]),
                                'server_cost_time': sum([m["server_cost_time"] for m in self._chunk_metrics]),
                                "generate_time": sum([m["generate_time"] for m in self._chunk_metrics])
                            }
                            self._log.debug(f"Task finished. Request performance metrics: client_first_package_time: {self._metrics['client_first_package_time']:.3f}, client_rtf: {self._metrics['client_rtf']:.3f}, client_cost_time: {self._metrics['client_cost_time']:.3f}, speech_len: {self._metrics['speech_len']:.3f}, server_cost_time: {self._metrics['server_cost_time']:.3f}, generate_time: {self._metrics['generate_time']:.3f}")
                            self._task_finished = True
                            self.close(ws)
                        elif event == "task-failed":
                            self._log.error(f"Task failed: {msg_json}")
                            self._task_finished = True
                            self.close(ws)
            except json.JSONDecodeError as e:
                self._log.error(f"JSON parsing failed: {str(e)}\t{traceback.format_exc()}")
        def on_error(self, ws, error):
            """Callback for when an error occurs."""
            self._log.error(f"WebSocket error: {error}\t{traceback.format_exc()}")
            self._metrics = {'error': error}
        def on_close(self, ws, close_status_code, close_msg):
            """Callback for when the connection is closed."""
            self._log.debug(f"WebSocket closed: {close_msg} ({close_status_code})")
        def send_continue_task(self, text):
            """Sends a 'continue-task' command with the text 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))
            self._log.debug(f"Sent 'continue-task' command with text: {text}")
        def send_finish_task(self):
            """Sends a 'finish-task' command 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))
            self._log.debug("Sent 'finish-task' command.")
        def close(self, ws):
            """Actively closes the connection."""
            if ws and ws.sock and ws.sock.connected:
                ws.close()
                self._log.debug("Connection actively closed.")
        def run(self):
            """Starts the WebSocket client."""
            # Set request headers for authentication.
            header = {
                "Authorization": f"Bearer {self._api_key}",
            }
            # Create a 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
            )
            self._log.debug("Listening for WebSocket messages...")
            self._ws.run_forever()  # Start the persistent connection.
    # Example usage
    if __name__ == "__main__":
        API_KEY = "<EAS_TOKEN>"                                      # Replace <EAS_TOKEN> with the service token.
        SERVER_URI = "ws://<EAS_SERVICE_URL>/api-ws/v1/audio/speech" # Replace <EAS_SERVICE_URL> with the service endpoint.
                                                                     # Example: "ws://cosyvoice-frontend-test.1534081855183999.cn-hangzhou.pai-eas.aliyuncs.com/api-ws/v1/audio/speech"
        texts = [
            "Receiving a birthday gift from a friend far away filled my heart with joy. The surprise and warm wishes made me smile."
        ]
        params = {
            "mode": "natural_language_replication",
            "texts": texts,
            "reference_audio_id": "<reference_audio_id>",
            "speed": 1.0,
            "output_format": "mp3",
            "sample_rate": 24000,
            "bit_rate": "48k",
            "volume": 2.0,
            "instruct": "Speak in a calm tone."
        }
        client = TTSClient(API_KEY, SERVER_URI, params, log_level='DEBUG')
        client.run()
        with open('./websocket_stream.mp3', 'wb') as wfile:
            wfile.write(client.audio_data)
        metrics = client.get_metrics()
        print(f"{metrics=}")

Instruções

  1. Leia o conteúdo em inglês para entender O QUE precisa ser comunicado

  2. Escreva o português brasileiro DO ZERO — esqueça a estrutura das frases em inglês

  3. Preserve toda a formatação markdown, blocos de código, links e imagens exatamente como estão

  4. Copie os placeholders xref ({XREF_N}) literalmente, NÃO os traduza ou modifique

  5. Aplique todas as regras específicas de idioma rigorosamente

  6. Aplique as regras de stopwords com tolerância zero

  7. Use o modo imperativo em etapas numeradas e listas de procedimentos

  8. Garanta a consistência terminológica — mesmo termo = mesma tradução em todo o documento

  9. Varie os inícios de frase em listas/tabelas — nenhum início repetido mais de 3 vezes

  10. Retorne APENAS o documento markdown em português brasileiro, sem explicações