Alibaba Cloud Model Studio:Qwen-ASR-Realtime Python SDK - API リファレンス

WebSocket を介して音声を Qwen-ASR-Realtime にストリーミングし、DashScope Python SDK を使ってリアルタイムの文字起こし結果を受信します。

前提条件

作業を開始する前に、以下の項目が準備済みであることを確認してください。

• DashScope SDK 2.0.1 以降
• API キー
• インタラクションフロー の理解

完全な例

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

# 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()

リクエストパラメーター

client.audio.realtime.connect

以下のパラメーターは client.audio.realtime.connect を使用して設定します。

# 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",
)

# リアルタイム認識セッションを作成します：with コンテキストに入ると自動的に接続が確立されます
session = client.audio.realtime.connect(
    model='qwen3-asr-flash-realtime',
)

セッション構成

接続後、update_session を呼び出してセッションパラメーターを構成します。

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',
    },
)

transcription_params

update_session の transcription_params 辞書を使用して音声認識設定を構成します。

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

主要インターフェイス

RealtimeSession クラス

リアルタイム認識セッションは client.audio.realtime.connect(model=...) で作成されます：with コンテキストに入る（またはセッションの connect() メソッドを呼び出す）と接続が確立されます。

イベント反復（events()）

サーバーは、サーバー応答イベントとデータをイベントとしてクライアントに返します。受信スレッドでセッションの events() を反復処理し、event["type"] に基づいて振り分け処理してください（完全なイベントリストはサーバーイベントをご参照ください）。

def receive_events(session):
    for event in session.events():
        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']}")