Skip to main content
Speech-to-speech

Tradução de áudio e vídeo em tempo real - Qwen

O qwen3.5-livetranslate-flash-realtime é um modelo de tradução em tempo real com aprimoramento visual compatível com 60 idiomas (29 com áudio e texto, 31 apenas com texto). Ele processa entradas de áudio e imagem de fluxos de vídeo ou arquivos locais, usa o contexto visual para melhorar a precisão e gera texto e áudio traduzidos em tempo real.

Experimente uma demonstração online com one-click deployment using Function Compute .

Recursos

  • Suporte multilíngue: Traduz entre 60 idiomas — 29 com saída de áudio e texto, 31 apenas com saída de texto — incluindo chinês, inglês, francês, alemão, russo, japonês, coreano, espanhol, português e árabe.
  • Aprimoramento visual: Analisa pistas visuais, como movimentos labiais, gestos e textos na tela, para aumentar a precisão da tradução, especialmente em ambientes ruidosos ou para palavras ambíguas.
  • Latência de 2,8 segundos: Oferece interpretação simultânea com latência mínima de 2,8 segundos.
  • Interpretação simultânea sem perdas: Prevê unidades semânticas para resolver diferenças na ordem das palavras entre idiomas, alcançando qualidade comparável à tradução offline.
  • Voz natural: Reproduz automaticamente a entonação e a emoção do áudio original.
  • Configuração de hotwords: Hotwords configuráveis melhoram a precisão da tradução para termos específicos.
  • Clonagem de voz: Clona a voz do falante para a saída traduzida. Compatível com clonagem em tempo real no servidor e perfis de voz pré-clonados.
  • Ignorar saída no mesmo idioma: Quando os idiomas de origem e destino são iguais, o modelo pode ignorar a saída de texto, a saída de áudio ou ambas. Esse recurso entra em vigor apenas quando o idioma de destino é chinês (zh) ou inglês (en).

Procedimento

1. Configurar a conexão

O modelo se conecta via WebSocket com os seguintes parâmetros: Além do WebSocket, este modelo também é compatível com os protocolos AOQ e WebRTC. Para integrações no lado do cliente que priorizam latência estável, resiliência em redes instáveis e supressão de ruído e cancelamento de eco full-duplex integrados, recomenda-se o uso do AOQ. Para obter uma comparação dos protocolos, consulte Realtime API overview.
ParâmetroDescrição
endpointRegião China (Beijing): wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime. Substitua {WorkspaceId} pelo ID real do seu workspace.Região Singapore: wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime.Substitua {WorkspaceId} pelo seu workspace ID real.
query parameterDefina o query parameter do modelo com o nome do modelo. Exemplo: ?model=qwen3.5-livetranslate-flash-realtime
message headerUse um Bearer Token para autenticação: Authorization: Bearer DASHSCOPE_API_KEY
DASHSCOPE_API_KEY é a sua chave de API do Model Studio.
Exemplo de código de conexão (Python):
# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"Connected to server: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("Error:", error)

ws = websocket.WebSocketApp(
    API_URL,
    header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)

ws.run_forever()

2. Configurar idioma, modalidade e voz

Envie o evento de cliente session.update com os seguintes parâmetros:
  • Idioma
    • Idioma de origem: Configure por meio do parâmetro session.input_audio_transcription.language.
      Se não for especificado, o modelo detecta automaticamente o idioma de origem.
    • Idioma de destino: Configure por meio do parâmetro session.translation.language.
      O valor padrão é en (inglês).
    Consulte Supported languages.
  • Gerar resultados de reconhecimento do idioma de origem Defina session.input_audio_transcription.model como qwen3-asr-flash-realtime. Assim, o servidor retorna tanto a tradução quanto o resultado do reconhecimento de fala (texto original) do áudio de entrada. O servidor retorna estes eventos:
    • conversation.item.input_audio_transcription.text: Transmite os resultados de reconhecimento em fluxo contínuo.
    • conversation.item.input_audio_transcription.completed: Retorna o resultado final após a conclusão do reconhecimento.
    • conversation.item.input_audio_transcription.failed: Retorna informações de erro quando o reconhecimento falha.
  • Modalidade de saída Defina o parâmetro session.modalities como ["text"] (apenas texto) ou ["text","audio"] (texto e áudio).
  • Detecção de atividade de voz (VAD) e modo manual Configure a detecção dos limites de fala usando o parâmetro session.turn_detection:
    • Modo VAD (padrão): Defina turn_detection como um objeto de configuração. O servidor detecta automaticamente os limites de fala e aciona a tradução. Ideal para cenários em que o cliente envia fluxos de áudio continuamente.
    • Modo manual: Defina turn_detection como null. O cliente determina os limites de fala e envia um evento input_audio_buffer.commit para submeter o áudio após cada enunciado. Ideal para cenários de "pressionar para falar".
    Para obter as etapas completas de interação em ambos os modos, consulte 3. Input audio and images.
  • Voz Configure usando o parâmetro session.voice. Consulte Supported voices.
  • Hotword Configure hotwords usando o parâmetro session.translation.corpus.phrases. Hotwords são pares chave-valor que mapeiam termos de origem para traduções de destino, melhorando a precisão para termos específicos. Exemplo: Mapeie "artificial intelligence" para "Artificial Intelligence".
  • Clonagem de voz Configure usando os parâmetros session.enable_voice_clone, session.voice_clone_options.frequency e session.voice. Compatível com três modos: perfil de voz pré-clonado (frequency: never), clonagem única no servidor no início da sessão (once) ou clonagem em tempo real antes de cada resposta (always). Consulte Voice cloning.

3. Enviar áudio e imagens

Envie dados de áudio e imagem codificados em Base64 usando os eventos input_audio_buffer.append e input_image_buffer.append. A entrada de áudio é obrigatória; a entrada de imagem é opcional.
As imagens podem vir de um arquivo local ou ser capturadas em tempo real de um fluxo de vídeo.
A forma como o modelo determina que um enunciado foi concluído depende do modo VAD ou do modo manual configurado via parâmetro turn_detection:
  • Modo VAD (padrão): O cliente envia eventos input_audio_buffer.append continuamente. Quando o servidor detecta o início ou fim da fala, ele retorna os eventos input_audio_buffer.speech_started e input_audio_buffer.speech_stopped, respectivamente, confirma automaticamente o buffer de áudio e aciona a tradução. As respostas de tradução são geradas sincronizadamente com o fluxo de áudio e geralmente começam durante a entrada de áudio, sem esperar que a fala termine.
  • Modo manual: Defina session.turn_detection como null. Depois que o cliente termina de enviar um enunciado completo, ele envia um evento input_audio_buffer.commit para confirmar o buffer de áudio. Após o servidor retornar um evento input_audio_buffer.committed como confirmação, ele inicia automaticamente a geração da resposta de tradução; o cliente não precisa enviar nenhum outro evento para acionar a resposta. Para limpar áudios não confirmados antes da confirmação, envie um evento input_audio_buffer.clear.

4. Receber a resposta do modelo

As respostas de tradução são geradas sincronizadamente com o fluxo de áudio e normalmente não exigem espera até o fim da fala (consulte a descrição dos modos VAD/manual na seção anterior). O formato da resposta depende da modalidade de saída.
  • Saída apenas de texto O servidor transmite texto traduzido incrementalmente (incluindo texto confirmado e texto previsto provisoriamente) por meio de eventos response.text.text; após a conclusão, o texto traduzido completo é retornado em um evento response.text.done.
  • Saída de texto e áudio
O modelo de tradução em tempo real usa o evento response.text.text para entrega incremental de texto, o que difere do evento response.text.delta usado pelos modelos Omni (conversa de voz full-duplex). Esses eventos possuem estruturas de campos e semânticas diferentes — não os utilize de forma intercambiável.

5. Encerrar a sessão

Após enviar todo o áudio, envie um evento Client events e aguarde o servidor retornar um evento session.finished antes de fechar a conexão WebSocket. Se você fechar o WebSocket sem enviar session.finish, o VAD do servidor não conseguirá detectar o fim do último segmento de fala. Isso causa a perda total dos resultados de tradução desse segmento, e a conexão pode ficar travada indefinidamente. Sempre envie este evento antes de desconectar.

Modelos compatíveis

Modelos recomendados

ModeloVersãoJanela de contextoEntrada máximaSaída máxima
(tokens)
qwen3.5-livetranslate-flash-realtime
Alias para qwen3.5-livetranslate-flash-realtime-2026-05-19
Estável53.24849.1524.096
qwen3.5-livetranslate-flash-realtime-2026-05-19Snapshot

Modelos legados

O modelo a seguir ainda está disponível, mas não é mais a escolha recomendada. Para novos casos de uso, utilize o modelo mais recente acima para obter melhor qualidade de tradução e custo-benefício.
ModeloVersãoJanela de contextoEntrada máximaSaída máxima
(tokens)
qwen3-livetranslate-flash-realtime
Alias para qwen3-livetranslate-flash-realtime-2025-09-22
Estável53.24849.1524.096
qwen3-livetranslate-flash-realtime-2025-09-22Snapshot

Primeiros passos

  1. Preparar o ambiente Requer Python 3.10 ou superior. Primeiro, instale o pyaudio.
    macOS
    brew install portaudio && pip install pyaudio
    
    Em seguida, instale as dependências do WebSocket:
pip install websocket-client==1.8.0 websockets
  1. Criar o cliente Crie um arquivo chamado livetranslate_client.py com o seguinte código:
    import os
    import time
    import base64
    import asyncio
    import json
    import websockets
    import pyaudio
    import queue
    import threading
    import traceback
    
    class LiveTranslateClient:
        def __init__(self, api_key: str, target_language: str = "en", *, audio_enabled: bool = True):
            if not api_key:
                raise ValueError("API key cannot be empty.")
    
            self.api_key = api_key
            self.target_language = target_language
            self.audio_enabled = audio_enabled
            self.ws = None
            self.api_url = "wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3.5-livetranslate-flash-realtime"
    
            # Audio input configuration (from microphone)
            self.input_rate = 16000
            self.input_chunk = 1600
            self.input_format = pyaudio.paInt16
            self.input_channels = 1
    
            # Audio output configuration (for playback)
            self.output_rate = 24000
            self.output_chunk = 2400
            self.output_format = pyaudio.paInt16
            self.output_channels = 1
    
            # State management
            self.is_connected = False
            self.audio_player_thread = None
            self.audio_playback_queue = queue.Queue()
            self.pyaudio_instance = pyaudio.PyAudio()
            self.session_finished_event = asyncio.Event()
    
        async def connect(self):
            """Establish a WebSocket connection to the translation service."""
            headers = {"Authorization": f"Bearer {self.api_key}"}
            try:
                self.ws = await websockets.connect(self.api_url, additional_headers=headers)
                self.is_connected = True
                print(f"Successfully connected to the server: {self.api_url}")
                await self.configure_session()
            except Exception as e:
                print(f"Connection failed: {e}")
                self.is_connected = False
                raise
    
        async def configure_session(self):
            """Configure the translation session, setting the target language, voice, etc."""
            config = {
                "event_id": f"event_{int(time.time() * 1000)}",
                "type": "session.update",
                "session": {
                    # 'modalities' controls the output type.
                    # ["text", "audio"]: Returns both translated text and synthesized audio (recommended).
                    # ["text"]: Returns only the translated text.
                    "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
                    "input_audio_format": "pcm",
                    "output_audio_format": "pcm",
                    # 'input_audio_transcription' configures source language recognition.
                    # Set 'model' to 'qwen3-asr-flash-realtime' to also output the source language recognition result.
                    # "input_audio_transcription": {
                    #     "model": "qwen3-asr-flash-realtime",
                    #     "language": "zh"  # source language, default 'en'
                    # },
                    "translation": {
                        "language": self.target_language,
                        # 'corpus' configures hotwords to improve the translation accuracy of specific terms.
                        # "corpus": {
                        #     "phrases": {
                        #         "Artificial Intelligence": "Artificial Intelligence",
                        #         "Machine Learning": "Machine Learning"
                        #     }
                        # }
                    }
                }
            }
            print(f"Sending session configuration: {json.dumps(config, indent=2, ensure_ascii=False)}")
            await self.ws.send(json.dumps(config))
    
        async def send_audio_chunk(self, audio_data: bytes):
            """Encode and send an audio chunk to the server."""
            if not self.is_connected:
                return
    
            event = {
                "event_id": f"event_{int(time.time() * 1000)}",
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_data).decode()
            }
            await self.ws.send(json.dumps(event))
    
        async def send_image_frame(self, image_bytes: bytes, *, event_id: str | None = None):
            # Send an image frame to the server.
            if not self.is_connected:
                return
    
            if not image_bytes:
                raise ValueError("image_bytes cannot be empty.")
    
            # Encode to Base64
            image_b64 = base64.b64encode(image_bytes).decode()
    
            event = {
                "event_id": event_id or f"event_{int(time.time() * 1000)}",
                "type": "input_image_buffer.append",
                "image": image_b64,
            }
    
            await self.ws.send(json.dumps(event))
    
        def _audio_player_task(self):
            stream = self.pyaudio_instance.open(
                format=self.output_format,
                channels=self.output_channels,
                rate=self.output_rate,
                output=True,
                frames_per_buffer=self.output_chunk,
            )
            try:
                while self.is_connected or not self.audio_playback_queue.empty():
                    try:
                        audio_chunk = self.audio_playback_queue.get(timeout=0.1)
                        if audio_chunk is None: # Termination signal
                            break
                        stream.write(audio_chunk)
                        self.audio_playback_queue.task_done()
                    except queue.Empty:
                        continue
            finally:
                stream.stop_stream()
                stream.close()
    
        def start_audio_player(self):
            """Start the audio player thread (only when audio output is enabled)."""
            if not self.audio_enabled:
                return
            if self.audio_player_thread is None or not self.audio_player_thread.is_alive():
                self.audio_player_thread = threading.Thread(target=self._audio_player_task, daemon=True)
                self.audio_player_thread.start()
    
        async def handle_server_messages(self, on_text_received):
            """Handle incoming messages from the server in a loop."""
            try:
                async for message in self.ws:
                    event = json.loads(message)
                    event_type = event.get("type")
                    if event_type == "response.audio.delta" and self.audio_enabled:
                        audio_b64 = event.get("delta", "")
                        if audio_b64:
                            audio_data = base64.b64decode(audio_b64)
                            self.audio_playback_queue.put(audio_data)
    
                    elif event_type == "response.done":
                        print("\n[INFO] Response round complete.")
                        usage = event.get("response", {}).get("usage", {})
                        if usage:
                            print(f"[INFO] token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}")
                    elif event_type == "session.finished":
                        print("[INFO] Session finished.")
                        self.session_finished_event.set()
                    # Process source language recognition results (requires enabling input_audio_transcription.model)
                    # elif event_type == "conversation.item.input_audio_transcription.text":
                    #     stash = event.get("stash", "")  # Pending recognition text
                    #     print(f"[Recognizing] {stash}")
                    # elif event_type == "conversation.item.input_audio_transcription.completed":
                    #     transcript = event.get("transcript", "")  # Complete recognition result
                    #     print(f"[Source language] {transcript}")
                    elif event_type == "response.text.text":
                        # Streaming translated text in text-only modality
                        text = event.get("text", "")
                        stash = event.get("stash", "")
                        print(f"\r[Translating] {text}{stash}", end="", flush=True)
                    elif event_type == "response.audio_transcript.done":
                        print("\n[INFO] Translation complete.")
                        text = event.get("transcript", "")
                        if text:
                            print(f"[INFO] Translated text: {text}")
                    elif event_type == "response.text.done":
                        print("\n[INFO] Translation complete.")
                        text = event.get("text", "")
                        if text:
                            print(f"[INFO] Translated text: {text}")
    
            except websockets.exceptions.ConnectionClosed as e:
                print(f"[WARNING] Connection closed: {e}")
                self.is_connected = False
            except Exception as e:
                print(f"[ERROR] An unexpected error occurred while processing messages: {e}")
                traceback.print_exc()
                self.is_connected = False
    
        async def start_microphone_streaming(self):
            """Capture audio from the microphone and stream it to the server."""
            stream = self.pyaudio_instance.open(
                format=self.input_format,
                channels=self.input_channels,
                rate=self.input_rate,
                input=True,
                frames_per_buffer=self.input_chunk
            )
            print("Microphone is on. Start speaking...")
            try:
                while self.is_connected:
                    audio_chunk = await asyncio.get_event_loop().run_in_executor(
                        None, stream.read, self.input_chunk
                    )
                    await self.send_audio_chunk(audio_chunk)
            finally:
                stream.stop_stream()
                stream.close()
    
        async def close(self):
            """Gracefully close the connection and release resources."""
            # Send session.finish to ensure the server completes translation of the final speech segment
            if self.is_connected and self.ws:
                finish_event = {
                    "event_id": f"event_{int(time.time() * 1000)}",
                    "type": "session.finish",
                }
                await self.ws.send(json.dumps(finish_event))
                print("Sent session.finish, waiting for server to finish processing...")
                try:
                    await asyncio.wait_for(self.session_finished_event.wait(), timeout=15)
                    print("Server processing complete.")
                except asyncio.TimeoutError:
                    print("Timed out waiting for session.finished.")
            self.is_connected = False
            if self.ws:
                await self.ws.close()
                print("WebSocket connection closed.")
    
            if self.audio_player_thread:
                self.audio_playback_queue.put(None) # Send termination signal
                self.audio_player_thread.join(timeout=1)
                print("Audio player thread stopped.")
    
            self.pyaudio_instance.terminate()
            print("PyAudio instance released.")
    
  2. Interagir com o modelo No mesmo diretório, crie um arquivo chamado main.py com o seguinte código:
    import os
    import asyncio
    from livetranslate_client import LiveTranslateClient
    
    def print_banner():
        print("=" * 60)
        print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime")
        print("=" * 60 + "\n")
    
    def get_user_config():
        """Get user configuration."""
        print("Select a mode:")
        print("1. Voice + Text [Default] | 2. Text Only")
        mode_choice = input("Enter your choice (press Enter for Voice + Text): ").strip()
        audio_enabled = (mode_choice != "2")
    
        if audio_enabled:
            lang_map = {
                "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt",
                "7": "es", "8": "it", "9": "ko", "10": "ja", "11": "yue"
            }
            print("Select the target language (Voice + Text mode):")
            print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Korean | 10. Japanese | 11. Cantonese")
        else:
            lang_map = {
                "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt", "7": "es", "8": "it",
                "9": "id", "10": "ko", "11": "ja", "12": "vi", "13": "th", "14": "ar",
                "15": "yue", "16": "hi", "17": "el", "18": "tr"
            }
            print("Select the target language (Text Only mode):")
            print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Indonesian | 10. Korean | 11. Japanese | 12. Vietnamese | 13. Thai | 14. Arabic | 15. Cantonese | 16. Hindi | 17. Greek | 18. Turkish")
    
        choice = input("Enter your choice (defaults to the first option): ").strip()
        target_language = lang_map.get(choice, next(iter(lang_map.values())))
    
        return target_language, audio_enabled
    
    async def main():
        """Main program entry point."""
        print_banner()
    
        api_key = os.environ.get("DASHSCOPE_API_KEY")
        if not api_key:
            print("[ERROR] Please set the DASHSCOPE_API_KEY environment variable.")
            print("  For example: export DASHSCOPE_API_KEY='your_api_key_here'")
            return
    
        target_language, audio_enabled = get_user_config()
        print("\nConfiguration complete:")
        print(f"  - Target language: {target_language}")
        if not audio_enabled:
            print("  - Output mode: Text Only")
    
        client = LiveTranslateClient(api_key=api_key, target_language=target_language, audio_enabled=audio_enabled)
    
        # Define the callback function.
        def on_translation_text(text):
            print(text, end="", flush=True)
    
        try:
            print("Connecting to the translation service...")
            await client.connect()
    
            # Start audio playback based on the mode.
            client.start_audio_player()
    
            print("\n" + "-" * 60)
            print("Connection successful! Speak into the microphone.")
            print("The program will translate your speech in real time and play the translated audio. Press Ctrl+C to exit.")
            print("-" * 60 + "\n")
    
            # Run message handling and microphone recording concurrently.
            message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
            tasks = [message_handler]
            # Capture audio from the microphone for translation, regardless of whether audio output is enabled.
            microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
            tasks.append(microphone_streamer)
    
            await asyncio.gather(*tasks)
    
        except KeyboardInterrupt:
            print("\n\nUser interrupted. Exiting...")
        except Exception as e:
            print(f"\nA critical error occurred: {e}")
        finally:
            print("\nCleaning up resources...")
            await client.close()
            print("Program exited.")
    
    if __name__ == "__main__":
        asyncio.run(main())
    
    Execute main.py e fale ao microfone. O modelo traduz sua fala e gera áudio e texto em tempo real.

Clonagem de voz

O modelo clona a voz do falante a partir do áudio de entrada e a utiliza na saída traduzida. Use um perfil de voz pré-clonado ou permita que o servidor faça a clonagem em tempo real. Útil para interpretação de conferências, transmissões ao vivo e dublagem de vídeos. Defina os seguintes parâmetros em session.update para ativar a clonagem de voz:
  • session.enable_voice_clone: Defina como true para ativar a clonagem de voz.
  • session.voice_clone_options.frequency: Controla quando a clonagem de voz ocorre. Valores aceitos:
    • never: Não realiza clonagem no servidor. Em vez disso, usa um perfil de voz pré-clonado. Defina session.voice com o ID da sua voz clonada personalizada.
    • once: Clona a voz do áudio de entrada uma única vez no início da sessão e a reutiliza para todas as saídas subsequentes. Ideal para cenários com um único falante. Defina session.voice como default.
    • always: Clona a voz antes de cada resposta, adaptando-se dinamicamente às mudanças de falante. Mais indicado para conversas com múltiplos falantes. Defina session.voice como default.
  • session.voice: Especifica a voz de saída. O valor depende da configuração de frequency:
    • Defina como default: Use com frequency definido como once ou always. O servidor clona a voz do falante a partir do áudio de entrada. Uma voz padrão é usada até que a clonagem seja concluída.
    • Defina como um ID de voz clonada personalizada (por exemplo, qwen-translate-vc-xxx-yyy-zzz): Use com frequency definido como never. Você deve preparar a voz antecipadamente usando Voice Cloning API com targetModel definido como qwen3.5-livetranslate-flash-realtime.
Quando frequency estiver definido como once ou always , o parâmetro voice deve ser definido como default . Qualquer outro valor fará com que o servidor retorne um erro.

Exemplos de configuração de clonagem de voz

Perfil de voz pré-clonado (qualidade consistente; recomendado quando uma identidade de voz estável é necessária):
{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "qwen-translate-vc-xxx-yyy-zzz",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "never"
        }
    }
}
Clonagem no servidor, uma vez por sessão (ideal para cenários com um único falante):
{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "once"
        }
    }
}
Clonagem no servidor, a cada resposta (ideal para conversas com múltiplos falantes):
{
    "type": "session.update",
    "session": {
        "modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "always"
        }
    }
}

Melhorar a tradução com imagens

A entrada de imagens ajuda a desambiguar homônimos e a reconhecer substantivos próprios incomuns durante a tradução. Envie no máximo 2 imagens por segundo. Baixe as seguintes imagens de exemplo: medical mask.png, masquerade mask.png Baixe o código abaixo para o mesmo diretório onde está livetranslate_client.py e execute-o. Diga "What is mask?" ao microfone. O modelo usa a imagem para desambiguar: medical mask.png resulta em "What is a medical mask?" e masquerade mask.png resulta em "What is a masquerade mask?".
import os
import time
import json
import asyncio
import contextlib
import functools

from livetranslate_client import LiveTranslateClient

IMAGE_PATH = "medical mask.png"
# IMAGE_PATH = "masquerade mask.png"

def print_banner():
    print("=" * 60)
    print("  Powered by Qwen qwen3.5-livetranslate-flash-realtime — single-turn interaction example (mask)")
    print("=" * 60 + "\n")

async def stream_microphone_once(client: LiveTranslateClient, image_bytes: bytes):
    pa = client.pyaudio_instance
    stream = pa.open(
        format=client.input_format,
        channels=client.input_channels,
        rate=client.input_rate,
        input=True,
        frames_per_buffer=client.input_chunk,
    )
    print(f"[INFO] Recording started. Please speak...")
    loop = asyncio.get_event_loop()
    last_img_time = 0.0
    frame_interval = 0.5  # 2 fps
    try:
        while client.is_connected:
            data = await loop.run_in_executor(None, stream.read, client.input_chunk)
            await client.send_audio_chunk(data)

            # Append an image frame every 0.5 seconds
            now = time.time()
            if now - last_img_time >= frame_interval:
                await client.send_image_frame(image_bytes)
                last_img_time = now
    finally:
        stream.stop_stream()
        stream.close()

async def main():
    print_banner()
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("[ERROR] Please set the DASHSCOPE_API_KEY environment variable.")
        return

    client = LiveTranslateClient(api_key=api_key, target_language="zh", audio_enabled=True)

    def on_text(text: str):
        print(text, end="", flush=True)

    try:
        await client.connect()
        client.start_audio_player()
        message_task = asyncio.create_task(client.handle_server_messages(on_text))
        with open(IMAGE_PATH, "rb") as f:
            img_bytes = f.read()
        await stream_microphone_once(client, img_bytes)
        await asyncio.sleep(15)
    finally:
        await client.close()
        if not message_task.done():
            message_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await message_task

if __name__ == "__main__":
    asyncio.run(main())

Implantação com um clique no Function Compute

Para implantar o aplicativo:
  1. Acesse o modelo do Function Compute, insira sua chave de API e clique em Create and Deploy Default Environment para testar o aplicativo.
  2. Aguarde cerca de um minuto. Em Environment Details > Environment Context, obtenha o endpoint, altere o protocolo de http para https (por exemplo, https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/) e abra a URL em um navegador para interagir com o modelo.
    Este endpoint usa um certificado autoassinado e destina-se apenas a testes temporários. Seu navegador exibirá um aviso de segurança na primeira visita. Esse comportamento é esperado. Não use este endpoint em um ambiente de produção. Para prosseguir, siga as instruções na tela (por exemplo, clique em Advanced → Proceed to (unsafe)).
Se for solicitado que você configure permissões do Resource Access Management, siga as instruções na tela.
Para visualizar o código-fonte do projeto, acesse Resource Information > Function Resources .
Tanto o Function Compute quanto o Model Studio oferecem uma cota gratuita para novos usuários, suficiente para depuração básica. Após o esgotamento da cota gratuita, aplica-se o faturamento de pagamento conforme o uso.

Fluxo de interação

A tradução utiliza um modelo WebSocket orientado a eventos. A forma como os limites de fala são determinados depende do modo VAD ou do modo manual (consulte 3. Input audio and images). A tabela abaixo baseia-se no modo VAD (padrão) e anota os diferentes eventos do servidor no modo manual.
Ciclo de vidaEvento do clienteEvento do servidor
Inicialização da sessãosession.update
Configuração da sessão
session.created
Sessão criada
session.updated
Configuração da sessão atualizada
Entrada de áudio do usuárioinput_audio_buffer.append
Adicionar áudio ao buffer
input_image_buffer.append
Adicionar imagem ao buffer
input_audio_buffer.commit
(Apenas modo manual) Confirmar o buffer de áudio
Modo VAD:input_audio_buffer.speech_started
Início de fala detectado
input_audio_buffer.speech_stopped
Fim de fala detectado; o servidor confirma automaticamente o buffer de áudio
Modo manual:input_audio_buffer.committed
Retornado após o cliente enviar input_audio_buffer.commit, confirmando que o buffer de áudio foi confirmado
Saída de áudio do servidorNenhumresponse.created
Indica que o servidor começou a gerar uma resposta.
response.output_item.added
Indica que um novo item de saída está disponível.
conversation.item.created
Um novo item de mensagem é criado na conversa.
response.content_part.added
Indica que uma nova parte de conteúdo foi adicionada à mensagem do assistente.
response.text.text
Texto traduzido incremental na modalidade apenas texto
response.audio_transcript.text
Texto traduzido incremental na modalidade áudio+texto
response.audio.delta
Contém um trecho incremental do áudio sintetizado.
response.text.done
Texto de tradução concluído na modalidade apenas texto
response.audio_transcript.done
Texto de tradução concluído na modalidade áudio+texto
response.audio.done
Indica que o áudio sintetizado está completo.
response.content_part.done
Indica que uma parte de conteúdo de texto ou áudio da mensagem do assistente está completa.
response.output_item.done
Indica que todo o item de saída da mensagem do assistente está completo.
response.done
Indica que toda a resposta está completa.
Encerramento da sessãosession.finish
Notifica o servidor de que a entrada de áudio foi concluída
session.finished
Processamento do servidor concluído; sessão encerrada
Após enviar todo o áudio, envie um evento session.finish e aguarde session.finished antes de fechar o WebSocket. Se você fechar a conexão sem enviar session.finish, o servidor não saberá que a entrada de áudio terminou, e os resultados de reconhecimento e tradução do último segmento de fala serão perdidos.

API

Faturamento

Qwen3.5-LiveTranslate-Flash-Realtime
  • Áudio: 7 tokens por segundo de áudio de entrada; 12,5 tokens por segundo de áudio de saída.
  • Imagem: Cada 32×32 pixels consome 0,5 tokens.
  • Texto: Quando o reconhecimento de fala no idioma de origem está ativado, o service retorna uma transcrição do áudio de entrada além da tradução. Essa transcrição é cobrada como tokens de texto de saída.
Qwen3-LiveTranslate-Flash-Realtime
  • Áudio: Cada segundo de áudio de entrada ou saída consome 12,5 tokens.
  • Imagem: Cada 28×28 pixels consome 0,5 tokens.
  • Texto: Quando o reconhecimento de fala no idioma de origem está ativado, o service retorna uma transcrição do áudio de entrada além da tradução. Essa transcrição é cobrada como tokens de texto de saída.
Preços: Model list.

Limites de taxa

Para obter informações sobre os limites de taxa do modelo, consulte Rate limiting.

Idiomas compatíveis

Use os seguintes códigos de idioma para especificar os idiomas de origem e de destino.
Alguns idiomas de destino suportam apenas texto. O modelo legado qwen3-livetranslate-flash-realtime suporta apenas os seguintes 18 idiomas: en, zh, ru, fr, de, pt, es, it, id, ko, ja, vi, th, ar, yue, hi, el, tr.

Código do idioma

Idioma

Saída

zh

Chinês

Áudio + texto

en

Inglês

Áudio + texto

ar

Árabe

Áudio + texto

de

Alemão

Áudio + texto

fr

Francês

Áudio + texto

es

Espanhol

Áudio + texto

pt

Português

Áudio + texto

id

Indonésio

Áudio + texto

it

Italiano

Áudio + texto

ko

Coreano

Áudio + texto

ru

Russo

Áudio + texto

th

Tailandês

Áudio + texto

vi

Vietnamita

Áudio + texto

ja

Japonês

Áudio + texto

tr

Turco

Áudio + texto

hi

Híndi

Áudio + texto

ms

Malaio

Áudio + texto

nl

Holandês

Áudio + texto

ur

Urdu

Áudio + texto

nb

Norueguês Bokmål

Áudio + texto

sv

Sueco

Áudio + texto

da

Dinamarquês

Áudio + texto

he

Hebraico

Áudio + texto

fi

Finlandês

Áudio + texto

pl

Polonês

Áudio + texto

is

Islandês

Áudio + texto

cs

Tcheco

Áudio + texto

fil

Filipino

Áudio + texto

fa

Persa

Áudio + texto

yue

Cantonês

Texto

el

Grego

Texto

af

Africâner

Texto

ast

Asturiano

Texto

be

Bielorrusso

Texto

bg

Búlgaro

Texto

bn

Bengali

Texto

bs

Bósnio

Texto

ca

Catalão

Texto

ceb

Cebuano

Texto

et

Estoniano

Texto

gl

Galego

Texto

gu

Guzerate

Texto

hr

Croata

Texto

hu

Húngaro

Texto

jv

Javanês

Texto

kk

Cazaque

Texto

kn

Canarês

Texto

ky

Quirguiz

Texto

lv

Letão

Texto

mk

Macedônio

Texto

ml

Malaiala

Texto

mr

Marata

Texto

pa

Panjabi

Texto

ro

Romeno

Texto

sk

Eslovaco

Texto

sl

Esloveno

Texto

sw

Suaíli

Texto

tg

Tadjique

Texto

az

Azeri

Texto

uk

Ucraniano

Texto

Vozes compatíveis

Para obter as vozes compatíveis e os valores correspondentes do parâmetro voice, consulte Voice list.
Plano de Tokens
Playground de Modelos
Inferência do Modelo
Avaliação
Compressão de Modelos
Estatísticas e Monitoramento
Suporte