Alibaba Cloud Model Studio:Qwen-ASR-Realtime Python SDK - Referência da API

Transmita áudio para o Qwen-ASR-Realtime via WebSocket e receba resultados de transcrição em tempo real pelo DashScope Python SDK.

Pré-requisitos

Antes de começar, verifique se você tem:

• DashScope SDK 2.0.1 ou posterior
• Uma chave de API
• Conhecimento do fluxo de interação

Exemplo completo

ffmpeg -i your_audio.mp3 -ar 16000 -ac 1 -f s16le your_audio_file.pcm

import logging
import os
import signal
import sys
import time
import dashscope
from concurrent.futures import ThreadPoolExecutor

# A API Key da região Singapura é diferente da da região Pequim. Obtenha sua API Key: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Alibaba Cloud Model Studio: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    # A configuração abaixo refere-se à região Singapura (ap-southeast-1). Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

def setup_logging():
    """Configura a saída de log"""
    logger = logging.getLogger('dashscope')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False
    return logger

def handle_event(event):
    """Trata um evento do servidor"""
    kind = event.get('type')
    if kind == 'session.created':
        print(f"Sessão iniciada: {event['session']['id']}")
    elif kind == 'conversation.item.input_audio_transcription.completed':
        print(f"Texto final reconhecido: {event['transcript']}")
    elif kind == 'conversation.item.input_audio_transcription.text':
        print(f"Resultado da transcrição recebido: {event['text'] + event['stash']}")
    elif kind == 'input_audio_buffer.speech_started':
        print('======Início da fala======')
    elif kind == 'input_audio_buffer.speech_stopped':
        print('======Fim da fala======')

def receive_events(session):
    """Recebe e processa eventos do servidor"""
    for event in session.events():
        handle_event(event)

def read_audio_chunks(file_path, chunk_size=3200):
    """Lê o arquivo de áudio em blocos"""
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

def send_audio(session, file_path, delay=0.1):
    """Envia dados de áudio"""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"O arquivo de áudio {file_path} não existe.")

    print("Processando o arquivo de áudio... Pressione 'Ctrl+C' para interromper.")
    for chunk in read_audio_chunks(file_path):
        session.append_audio(chunk)  # os bytes são codificados automaticamente em Base64 antes do envio
        time.sleep(delay)

def main():
    setup_logging()

    audio_file_path = "./your_audio_file.pcm"

    # Entrar no contexto with estabelece a conexão automaticamente
    with client.audio.realtime.connect(
        model='qwen3-asr-flash-realtime',
    ) as session:
        def handle_exit(sig, frame):
            print('Ctrl+C pressionado, encerrando...')
            session.close()
            sys.exit(0)

        signal.signal(signal.SIGINT, handle_exit)

        # Atualiza a configuração da sessão: apenas saída de texto, com transcrição do áudio de entrada ativada
        session.update_session(
            output_modalities=['text'],
            enable_input_audio_transcription=True,
            transcription_params={
                'language': 'zh',
                'sample_rate': 16000,
                'input_audio_format': 'pcm',
            },
        )
        # Lê até session.updated para confirmar que a configuração entrou em vigor
        for event in session.events():
            if event.get('type') == 'session.updated':
                break

        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events, session)
            try:
                send_audio(session, audio_file_path)
                # envia session.finish e aguarda finished e o fechamento
                session.end_session()
                receiver.result()
            finally:
                # close deve ser executado dentro do contexto do pool para desbloquear o receptor
                session.close()
        print("Processamento de áudio concluído.")

if __name__ == '__main__':
    main()

Parâmetros da solicitação

Criação da sessão (client.audio.realtime.connect)

Os parâmetros a seguir são definidos por meio do método client.audio.realtime.connect.

# A API Key da região Singapura é diferente da da região Pequim. Obtenha sua API Key: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Alibaba Cloud Model Studio: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    # A configuração abaixo refere-se à região Singapura (ap-southeast-1). Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Cria a sessão de reconhecimento em tempo real: entrar no contexto with estabelece a conexão automaticamente
session = client.audio.realtime.connect(
    model='qwen3-asr-flash-realtime',
)

Configuração da sessão

Após a conexão ser estabelecida, chame o método update_session da sessão (RealtimeSession) para configurar os seguintes parâmetros.

session.update_session(
    output_modalities=['text'],
    enable_turn_detection=True,
    turn_detection_type="server_vad",
    turn_detection_threshold=0.0,
    turn_detection_silence_duration_ms=400,
    enable_input_audio_transcription=True,
    transcription_params={
        'language': 'zh',
        'sample_rate': 16000,
        'input_audio_format': 'pcm',
    },
)

Parâmetros de transcription_params

Configure as definições de reconhecimento de fala por meio do dicionário transcription_params do método update_session.

session.update_session(
    output_modalities=['text'],
    enable_input_audio_transcription=True,
    transcription_params={
        'language': 'zh',
        'sample_rate': 16000,
        'input_audio_format': 'pcm',
    },
)

Interfaces principais

Sessão RealtimeSession (client.audio.realtime.connect)

A sessão de reconhecimento em tempo real é criada por meio de client.audio.realtime.connect(model=...): a conexão é estabelecida ao entrar no contexto with (ou ao chamar o método connect() da sessão).

Iteração de eventos (events())

O servidor retorna os eventos e dados de resposta ao cliente por meio de eventos. Você precisa iterar os events() da sessão na thread de recepção e fazer o despacho por event["type"] (a lista completa de eventos está em eventos do servidor).

def receive_events(session):
    for event in session.events():
        kind = event.get('type')
        if kind == 'session.created':
            print(f"Sessão iniciada: {event['session']['id']}")
        elif kind == 'conversation.item.input_audio_transcription.completed':
            print(f"Texto final reconhecido: {event['transcript']}")
        elif kind == 'conversation.item.input_audio_transcription.text':
            print(f"Resultado da transcrição recebido: {event['text'] + event['stash']}")