Qwen-ASR-Realtime Python SDK - API リファレンス

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

# API キーは、シンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    # 以下はシンガポール (ap-southeast-1) リージョンの設定です。呼び出す際に、"{WorkspaceId}" を実際のワークスペース ID に置き換えてください。設定はリージョンによって異なります。
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

def setup_logging():
    """ログ出力を設定"""
    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):
    """サーバー側イベントを処理"""
    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):
    """サーバー側イベントを受信して処理"""
    for event in session.events():
        handle_event(event)

def read_audio_chunks(file_path, chunk_size=3200):
    """音声ファイルをチャンク単位で読み込む"""
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

def send_audio(session, file_path, delay=0.1):
    """音声データを送信"""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"音声ファイル {file_path} が存在しません。")

    print("Processing audio file... Press 'Ctrl+C' to stop.")
    for chunk in read_audio_chunks(file_path):
        session.append_audio(chunk)  # bytes は送信前に自動的に Base64 エンコードされます
        time.sleep(delay)

def main():
    setup_logging()

    audio_file_path = "./your_audio_file.pcm"

    # with コンテキストに入ると自動的に接続が確立されます
    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)

        # セッション設定を更新します：テキスト出力のみ、入力音声の文字起こしを有効化
        session.update_session(
            output_modalities=['text'],
            enable_input_audio_transcription=True,
            transcription_params={
                'language': 'zh',
                'sample_rate': 16000,
                'input_audio_format': 'pcm',
            },
        )
        # session.updated が返されるまで読み取り、設定が反映されたことを確認します
        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)
                # session.finish を送信し、finished の受信とクローズを待ちます
                session.end_session()
                receiver.result()
            finally:
                # close はレシーバーのブロックを解除するため、プールのコンテキスト内で実行する必要があります
                session.close()
        print("Audio processing completed.")

if __name__ == '__main__':
    main()