Traducción de audio y video en tiempo real - Qwen

Pruebe una demostración en línea con implementación con un clic mediante Function Compute .

​

Funcionalidades

• Soporte multilingüe: Traduce entre 60 idiomas — 29 con salida de audio y texto, 31 con salida solo de texto — incluyendo chino, inglés, francés, alemán, ruso, japonés, coreano, español, portugués y árabe.
• Mejora visual: Analiza señales visuales, como movimientos labiales, gestos y texto en pantalla, para mejorar la precisión de la traducción, especialmente en entornos ruidosos o para palabras ambiguas.
• Latencia de 2.3 segundos: Ofrece interpretación simultánea con una latencia tan baja como 2.3 segundos.
• Diarización de hablantes en tiempo real: Distingue a los diferentes hablantes y sus intervenciones cuando varias personas toman turnos para hablar, de modo que los oyentes puedan entender claramente quién dijo qué.
• Interpretación simultánea sin pérdidas: Predice unidades semánticas para resolver diferencias en el orden de las palabras entre idiomas, logrando una calidad comparable a la traducción sin conexión.
• Voz natural: Coincide automáticamente con la entonación y la emoción del audio de origen.
• Configuración de palabras clave: Las palabras clave configurables mejoran la precisión de la traducción para términos específicos.
• Clonación de voz: Clona la voz del hablante para la salida traducida. Admite clonación en tiempo real y perfiles de voz preclonados.

​

Procedimiento

​

1. Configurar la conexión

# 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.8-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, modalidad y voz

• Idioma de destino: Establezca session.translation.language, como en para inglés. Con qwen3.8-livetranslate-flash-realtime, el idioma de origen se detecta automáticamente.
  Con qwen3.5-livetranslate-flash-realtime, establezca session.input_audio_transcription.language para especificar el idioma de origen, u omítalo para la detección automática.
• Modalidades de salida: Con qwen3.8-livetranslate-flash-realtime, establezca session.output_modalities en ["text"] para solo texto o ["text", "audio"] para texto y audio (predeterminado).
  Con qwen3.5-livetranslate-flash-realtime, utilice session.modalities. Los valores admitidos y el valor predeterminado son los mismos.
• Transcripción de origen: Con qwen3.8-livetranslate-flash-realtime, la transcripción de origen siempre está habilitada y no se puede desactivar. El propio modelo LiveTranslate genera la transcripción en el idioma de origen. La información relevante de las imágenes puede ayudar a resolver ambigüedades en el audio de origen. Reciba incrementos mediante conversation.item.input_audio_transcription.delta y transcripciones completas mediante conversation.item.input_audio_transcription.completed.
  Con qwen3.5-livetranslate-flash-realtime, la transcripción de origen está habilitada de forma predeterminada. Un modelo ASR independiente transcribe el audio de entrada. Establezca session.input_audio_transcription.model en qwen3-asr-flash-realtime para habilitarla o en null para deshabilitarla. Reciba incrementos mediante conversation.item.input_audio_transcription.text y transcripciones completas mediante conversation.item.input_audio_transcription.completed.
• Audio y voz: Con qwen3.8-livetranslate-flash-realtime, los valores predeterminados son entrada PCM de 16000 Hz, salida PCM de 24000 Hz y la voz Tina. Para enviar audio Opus, establezca session.audio.input.format.type en opus. Consulte session.audio.output.voice en session.created para verificar la voz. Para conservar la voz del hablante, consulte Clonación de voz.
  Con qwen3.5-livetranslate-flash-realtime, configure el audio mediante session.sample_rate, session.input_audio_format y session.output_audio_format, y la voz mediante session.voice. La voz predeterminada también es Tina.
• Palabras clave: Configure session.translation.corpus.phrases para especificar traducciones preferidas. Consulte Configurar palabras clave.
  Se recomienda utilizar no más de 1,000 palabras clave.

​

3. Introducir audio e imágenes

Las imágenes pueden provenir de un archivo local o capturarse en tiempo real desde una transmisión de video.

​

4. Recibir la respuesta del modelo

• Solo texto: Agregue los valores de delta de response.text.delta al final de los subtítulos en orden, sin sobrescribir el texto ya mostrado.
• Texto y audio: Concatene delta de response.audio_transcript.delta para obtener texto, y decodifique en Base64 el campo delta de response.audio.delta para obtener fragmentos de audio.

​

5. Finalizar la sesión

​

Modelos admitidos

​

Modelos recomendados

​

Introducción

​

Traducir la entrada del micrófono en tiempo real

1. Utilice Python 3.10 o posterior e instale pyaudio:
   • macOS
   • Debian/Ubuntu
   • CentOS
   • Windows

   Copy

   brew install portaudio && pip install pyaudio
   

   Copy

   sudo apt-get install python3-pyaudio
   

   Copy

   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
   

   Copy

   pip install pyaudio
   
   Instale la dependencia de WebSocket: pip install websocket-client.
2. Configure estas variables de entorno. La clave API y el espacio de trabajo deben pertenecer a la región indicada en la URL de conexión:
   • DASHSCOPE_API_KEY: Una clave API para la región y el espacio de trabajo. Consulte Obtener una clave API.
   • DASHSCOPE_WORKSPACE_ID: Su ID de espacio de trabajo. Consulte Espacios de trabajo y regiones.
3. Guarde el siguiente código como livetranslate_client.py, ejecute python livetranslate_client.py y hable al micrófono.

import base64
import json
import os
import signal
import threading
import time

import pyaudio
import websocket

model = "qwen3.8-livetranslate-flash-realtime"
workspace_id = os.environ["DASHSCOPE_WORKSPACE_ID"]
api_key = os.environ["DASHSCOPE_API_KEY"]
url = (
    f"wss://{workspace_id}.ap-southeast-1.maas.aliyuncs.com"
    f"/api-ws/v1/realtime?model={model}"
)
stop_recording = False
send_errors = []
ws = None
audio = None
input_stream = None
output_stream = None
sender = None
previous_sigint_handler = None


def receive():
    event = json.loads(ws.recv())
    if event.get("type") == "error" or event.get("code"):
        raise RuntimeError(event)
    return event


def stop_input(signum, frame):
    global stop_recording
    stop_recording = True


def send_audio():
    try:
        while not stop_recording:
            chunk = input_stream.read(1600)
            ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode("ascii"),
            }))
    except Exception as error:
        send_errors.append(error)
    finally:
        try:
            ws.send(json.dumps({"type": "session.finish"}))
        except Exception as error:
            send_errors.append(error)
            ws.close()


def receive_translation():
    while True:
        event = receive()
        event_type = event.get("type")
        if event_type in ("response.text.delta", "response.audio_transcript.delta"):
            print(event["delta"], end="", flush=True)
        elif event_type == "response.audio.delta":
            output_stream.write(base64.b64decode(event["delta"]))
        elif event_type == "response.done":
            if event["response"]["status"] != "completed":
                raise RuntimeError(event["response"])
            print()
        elif event_type == "session.finished":
            return


try:
    ws = websocket.create_connection(
        url, header={"Authorization": f"Bearer {api_key}"}, timeout=30
    )
    while receive().get("type") != "session.created":
        pass
    ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "output_modalities": ["text", "audio"],
            "translation": {"language": "en"},
        },
    }))
    while receive().get("type") != "session.updated":
        pass
    audio = pyaudio.PyAudio()
    input_stream = audio.open(
        format=pyaudio.paInt16, channels=1, rate=16000,
        input=True, frames_per_buffer=1600,
    )
    output_stream = audio.open(
        format=pyaudio.paInt16, channels=1, rate=24000, output=True,
    )
    print("Start speaking. Press Ctrl+C to stop recording and wait for the remaining translation.")
    previous_sigint_handler = signal.signal(signal.SIGINT, stop_input)
    sender = threading.Thread(target=send_audio, daemon=True)
    sender.start()
    receive_translation()
    stop_recording = True
    sender.join()
    if send_errors:
        raise send_errors[0]
except Exception:
    if send_errors:
        raise send_errors[0]
    raise
finally:
    stop_recording = True
    if ws is not None:
        ws.close()
    if sender is not None:
        sender.join()
    for stream in (input_stream, output_stream):
        if stream is not None:
            stream.stop_stream()
            stream.close()
    if audio is not None:
        audio.terminate()
    if previous_sigint_handler is not None:
        signal.signal(signal.SIGINT, previous_sigint_handler)

​

Configurar palabras clave

• Redacción fija: El eslogan de Alibaba Cloud, “计算，为了无法计算的价值。”, tiene la redacción acordada en inglés “Computing, for the value beyond computation.”. La redacción y el orden de las palabras deben conservarse.
• Nombres de marcas que suenan similar: Qoder suena como la palabra común coder. Cuando el hablante se refiere al producto Qoder, la traducción necesita conservar la ortografía de la marca.

​

Configuración

{
  "event_id": "configure-hotwords",
  "type": "session.update",
  "session": {
    "translation": {
      "language": "en",
      "corpus": {
        "phrases": {
          "计算，为了无法计算的价值。": "Computing, for the value beyond computation.",
          "Qoder": "Qoder"
        }
      }
    }
  }
}

​

Identificar hablantes

{"type": "session.update", "session": {"audio": {"input": {"turn_detection": {"type": "speaker_detection", "threshold": 0.5}}}}}

1. Registre el speaker_id y el item_id de origen de input_audio_buffer.speech_started.
2. Lea el texto de origen para ese item_id de conversation.item.input_audio_transcription.delta.
3. En el evento conversation.item.created de la traducción, asocie item.id con el ID de origen en el campo de nivel superior previous_item_id del evento.
4. Utilice el item_id de cada incremento de traducción para encontrar su texto de origen y hablante, y luego actualice el subtítulo.

speaker_by_source = {}
source_by_translation = {}
source_text = {}
translation_text = {}


def show_subtitle(translation_id):
    source_id = source_by_translation.get(translation_id)
    if source_id not in speaker_by_source or translation_id not in translation_text:
        return
    print({
        "speaker_id": speaker_by_source[source_id],
        "source": source_text.get(source_id, ""),
        "translation": translation_text[translation_id],
    })


def handle_speaker_event(event):
    event_type = event.get("type")
    if event_type in ("response.text.delta", "response.audio_transcript.delta"):
        item_id = event["item_id"]
        translation_text[item_id] = translation_text.get(item_id, "") + event["delta"]
        show_subtitle(item_id)
        return
    if event_type == "conversation.item.created":
        item = event["item"]
        if any(part.get("type") == "input_audio" for part in item.get("content", [])):
            return
        source_id = event.get("previous_item_id")
        if item.get("role") == "assistant" and source_id is not None:
            source_by_translation[item["id"]] = source_id
            show_subtitle(item["id"])
        return
    if event_type == "input_audio_buffer.speech_started" and "speaker_id" in event:
        source_id = event["item_id"]
        speaker_by_source[source_id] = event["speaker_id"]
    elif event_type == "conversation.item.input_audio_transcription.delta":
        source_id = event["item_id"]
        source_text[source_id] = source_text.get(source_id, "") + event["delta"]
    elif event_type == "conversation.item.input_audio_transcription.completed":
        source_id = event["item_id"]
        source_text[source_id] = event["transcript"]
    else:
        return
    for translation_id, mapped_source in source_by_translation.items():
        if mapped_source == source_id:
            show_subtitle(translation_id)

​

Mejorar la traducción con imágenes

1. Instale Pillow: pip install Pillow.
2. Establezca translation.language en zh para la salida en chino y luego diga "What is mask?" al micrófono cuando ejecute el ejemplo.
3. Coloque el código de carga de imágenes a continuación antes de def send_audio(): en el ejemplo del micrófono, y reemplace el send_audio() original con esta versión. Utiliza los elementos existentes input_stream, stop_recording, ws, send_errors e importaciones.

import io
from PIL import Image

IMAGE_PATH = "medical-mask.png"
# IMAGE_PATH = "masquerade-mask.png"
with Image.open(IMAGE_PATH) as image:
    buffer = io.BytesIO()
    image.convert("RGB").save(buffer, format="JPEG")
image_bytes = buffer.getvalue()
if len(image_bytes) > 500_000:
    raise ValueError("JPEG > 500 KB")
image_b64 = base64.b64encode(image_bytes).decode("ascii")


def send_audio():
    try:
        last_image_time = None
        while not stop_recording:
            chunk = input_stream.read(1600)
            ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode("ascii"),
            }))
            now = time.monotonic()
            if last_image_time is None or now - last_image_time >= 0.5:
                ws.send(json.dumps({
                    "type": "input_image_buffer.append",
                    "image": image_b64,
                }))
                last_image_time = now
    except Exception as error:
        send_errors.append(error)
    finally:
        try:
            ws.send(json.dumps({"type": "session.finish"}))
        except Exception as error:
            send_errors.append(error)
            ws.close()

​

Clonación de voz

• session.enable_voice_clone: Establezca este valor en true para habilitar la clonación de voz.
• session.voice_clone_options.frequency: Controla cuándo ocurre la clonación de voz. Valores aceptados:
  • never: Utilice una voz creada previamente sin extraer una nueva voz durante esta sesión. Establezca session.voice en su ID de voz clonada.
  • once: Extrae la voz una vez del audio de entrada al inicio de la sesión y la reutiliza para traducciones posteriores en esta sesión. “Una vez” describe la extracción, no una única salida de audio. Adecuado para presentaciones de un solo hablante; establezca session.voice en default.
  • always: Extrae la voz antes de cada respuesta de audio traducido, de modo que la salida siga los cambios de hablante. Adecuado para múltiples hablantes; establezca session.voice en default.
• session.voice: Especifica la voz de salida. El valor depende de la configuración de frequency:
  • Establecer en default: Úselo con frequency establecido en once o always. Clona la voz del hablante a partir del audio de entrada. Se utiliza una voz predeterminada hasta que finaliza la clonación.
  • Establezca un ID de voz clonada personalizada, por ejemplo qwen-translate-vc-xxx-yyy-zzz, con frequency en never. Prepare la voz previamente mediante la API de clonación de voz, con target_model establecido en el modelo de traducción que use.

Cuando frequency se establece en once o always , el parámetro voice debe establecerse en default . Cualquier otro valor hace que el servidor devuelva un error.

​

Ejemplos de configuración de clonación de voz

{
    "type": "session.update",
    "session": {
        "output_modalities": ["text","audio"],
        "voice": "qwen-translate-vc-xxx-yyy-zzz",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "never"
        }
    }
}

{
    "type": "session.update",
    "session": {
        "output_modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "once"
        }
    }
}

{
    "type": "session.update",
    "session": {
        "output_modalities": ["text","audio"],
        "voice": "default",
        "translation": {
            "language": "en"
        },
        "enable_voice_clone": true,
        "voice_clone_options": {
            "frequency": "always"
        }
    }
}

​

Despliegue con un clic de Function Compute

1. Abra la plantilla de Function Compute, ingrese su clave API y haga clic en Create and Deploy Default Environment para probar la aplicación.
2. Espere aproximadamente un minuto. En Detalles del entorno > Contexto del entorno, obtenga el endpoint, cambie el protocolo de http a https (por ejemplo, https://qwen-livetranslate-flash-realtime-intl.fcv3.xxx.ap-southeast-1.fc.devsapp.net/) y abra la URL en un navegador para interactuar con el modelo.
   Este endpoint utiliza un certificado autofirmado y es solo para pruebas temporales. Su navegador mostrará una advertencia de seguridad en su primera visita. Este es un comportamiento esperado. No utilice este endpoint en un entorno de producción. Para continuar, siga las instrucciones en pantalla (por ejemplo, haga clic en Advanced → Continuar a (no seguro)).

Si se le solicita configurar permisos de Resource Access Management, siga las instrucciones en pantalla.

Para ver el código fuente del proyecto, vaya a Información de recursos > Recursos de funciones .

Tanto Function Compute como Model Studio proporcionan una cuota gratuita para nuevos usuarios, suficiente para la depuración básica. Una vez agotada la cuota gratuita, se aplica la facturación de pago por uso.

​

Flujo de interacción

​

API

• Traducción de audio y video en tiempo real (Qwen-LiveTranslate-Realtime).
• Conexión mediante WebRTC

​

Facturación

• Audio: 7 tokens por segundo de audio de entrada; 12.5 tokens por segundo de audio de salida. Con la detección de voz del servidor, la duración del audio de entrada se calcula sobre el audio que llega al modelo después del filtrado. Los silencios eliminados por ese filtrado no se incluyen. Consulte la cantidad real de tokens de audio de entrada en response.usage.input_tokens_details.audio_tokens del evento response.done.
• Imagen: Cada 32×32 píxeles consume 0.5 tokens. Para conocer los precios de tokens de cada modelo, consulte Precios y facturación de modelos.

​

Límites de tasa

​

Idiomas admitidos

Algunos idiomas de destino solo admiten texto.

​

Voces admitidas

​

Preguntas frecuentes

​

¿Se cobra mientras la conexión está abierta, pero nadie habla?

• Detección de voz del servidor: El silencio filtrado no se cuenta como entrada de audio.
• Modo Manual: El silencio que el cliente confirma explícitamente se cuenta como entrada de audio.

​

¿Cómo migro de Gummy a LiveTranslate?

import os
import sys
import time
import wave
from concurrent.futures import ThreadPoolExecutor

import dashscope


def send_audio(session, recording):
    while chunk := recording.readframes(1600):
        session.append_audio(chunk)
        time.sleep(len(chunk) / 32000)
    session.end_session()


def translate(path):
    client = dashscope.Client(api_key=os.environ["DASHSCOPE_API_KEY"],
                              workspace=os.environ["DASHSCOPE_WORKSPACE"],
                              region="cn-beijing")
    with wave.open(path, "rb") as recording:
        if (recording.getframerate(), recording.getnchannels(), recording.getsampwidth()) != (16000, 1, 2):
            raise ValueError("Expected a 16 kHz mono 16-bit PCM WAV file")
        with client.audio.realtime.connect(model="qwen3.8-livetranslate-flash-realtime", timeout=30) as session:
            events = session.events()
            for event in events:
                if event["type"] == "session.created":
                    session.update_session(output_modalities=["text"], translation_params={"language": "en"})
                elif event["type"] == "session.updated":
                    break
                elif event["type"] == "error":
                    raise RuntimeError(event["error"])
            else:
                raise RuntimeError("Session closed before configuration completed")

            with ThreadPoolExecutor(max_workers=1) as pool:
                sender = pool.submit(send_audio, session, recording)
                try:
                    for event in events:
                        kind = event["type"]
                        if kind == "response.text.delta":
                            print(event["delta"], end="", flush=True)
                        elif kind == "response.text.done":
                            print(flush=True)
                        elif kind == "response.done" and event["response"]["status"] != "completed":
                            raise RuntimeError(event["response"].get("status_details"))
                        elif kind == "error":
                            raise RuntimeError(event["error"])
                        elif kind == "session.finished":
                            break
                    else:
                        raise RuntimeError("Session closed before session.finished")
                except BaseException:
                    session.close()
                    raise
                sender.result()


if __name__ == "__main__":
    translate(sys.argv[1])

• Dirección: cambia /api-ws/v1/inference por /api-ws/v1/realtime?model=qwen3.8-livetranslate-flash-realtime.
• Audio: sustituye las tramas binarias por eventos JSON input_audio_buffer.append y envía el PCM codificado en Base64 en el campo audio.
• Traducción: sustituye result-generated por response.text.delta y añade cada fragmento delta al subtítulo. Cuando termines de enviar el audio, envía session.finish y espera a session.finished antes de cerrar.

​

¿Cómo reanudar la traducción después de una desconexión?

1. Deje de enviar audio a la conexión anterior.
2. Vuelva a conectarse y envíe session.update para restablecer el idioma de destino, las modalidades de salida y los demás ajustes.
3. Espere a recibir session.updated y luego reanude el envío de audio.