SDK de Python de Qwen-ASR-Realtime - Referencia de API

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

# Las API Key de las regiones de Singapur y Pekín son diferentes. Obtenga su API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Si no se ha configurado la variable de entorno, sustituya la siguiente línea por su API Key de Model Studio de Alibaba Cloud: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    # La siguiente es la configuración de la región de Singapur (ap-southeast-1). Al realizar la llamada, sustituya "{WorkspaceId}" por el ID real de su espacio de trabajo. La configuración varía según la región.
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

def setup_logging():
    """Configure log output"""
    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):
    """Handle a server-side event"""
    kind = event.get('type')
    if kind == 'session.created':
        print(f"Start session: {event['session']['id']}")
    elif kind == 'conversation.item.input_audio_transcription.completed':
        print(f"Final recognized text: {event['transcript']}")
    elif kind == 'conversation.item.input_audio_transcription.text':
        print(f"Got transcription result: {event['text'] + event['stash']}")
    elif kind == 'input_audio_buffer.speech_started':
        print('======Speech Start======')
    elif kind == 'input_audio_buffer.speech_stopped':
        print('======Speech Stop======')

def receive_events(session):
    """Receive and process server-side events"""
    for event in session.events():
        handle_event(event)

def read_audio_chunks(file_path, chunk_size=3200):
    """Read the audio file in chunks"""
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

def send_audio(session, file_path, delay=0.1):
    """Send audio data"""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Audio file {file_path} does not exist.")

    print("Processing audio file... Press 'Ctrl+C' to stop.")
    for chunk in read_audio_chunks(file_path):
        session.append_audio(chunk)  # los bytes se codifican automáticamente en Base64 antes de enviarse
        time.sleep(delay)

def main():
    setup_logging()

    audio_file_path = "./your_audio_file.pcm"

    # Al entrar en el contexto with se establece la conexión automáticamente
    with client.audio.realtime.connect(
        model='qwen3-asr-flash-realtime',
    ) as session:
        def handle_exit(sig, frame):
            print('Ctrl+C pressed, exiting...')
            session.close()
            sys.exit(0)

        signal.signal(signal.SIGINT, handle_exit)

        # Actualice la configuración de la sesión: solo salida de texto, con la transcripción del audio de entrada habilitada
        session.update_session(
            output_modalities=['text'],
            enable_input_audio_transcription=True,
            transcription_params={
                'language': 'zh',
                'sample_rate': 16000,
                'input_audio_format': 'pcm',
            },
        )
        # Lea hasta session.updated para confirmar que la configuración se ha aplicado
        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)
                # enviar session.finish y esperar el evento finished y el cierre
                session.end_session()
                receiver.result()
            finally:
                # close debe ejecutarse dentro del contexto del pool para desbloquear el receptor
                session.close()
        print("Audio processing completed.")

if __name__ == '__main__':
    main()