实时语音识别（Qwen-ASR-Realtime）Python SDK-API参考

本文档介绍如何使用 DashScope Python SDK 调用实时语音识别（Qwen-ASR-Realtime）模型。

前提条件

1. 安装SDK，确保DashScope SDK版本不低于2.0.1。
2. 获取与配置 API Key。
3. 了解WebSocket 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 Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
# 若没有配置环境变量，请用阿里云百炼API Key将下行替换为：api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    # 以下为华北2（北京）地域的配置，调用时请将"{WorkspaceId}"替换为真实的业务空间ID，各地域的配置不同。
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

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"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)  # 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)
                # send session.finish and wait for finished and close
                session.end_session()
                receiver.result()
            finally:
                # close must run inside the pool context to unblock the receiver
                session.close()
        print("Audio processing completed.")

if __name__ == '__main__':
    main()

请求参数

• 以下参数通过client.audio.realtime.connect方法设置。

  点击查看示例代码

  # 新加坡和北京地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
  # 若没有配置环境变量，请用阿里云百炼API Key将下行替换为：api_key="sk-xxx"
  client = dashscope.Client(
      api_key=os.environ.get('DASHSCOPE_API_KEY'),
      # 以下为华北2（北京）地域的配置，调用时请将"{WorkspaceId}"替换为真实的业务空间ID，各地域的配置不同。
      workspace="{WorkspaceId}",
      region="cn-beijing",
  )
  
  # 创建实时识别会话：进入 with 上下文自动建立连接
  session = client.audio.realtime.connect(
      model='qwen3-asr-flash-realtime',
  )
  

  参数	类型	是否必须	说明
  model	str	是	指定要使用的模型名称。
  ws_url	str	否	完整WebSocket端点，未指定时按Client的workspace与region自动选择端点： 华北2（北京）地域：wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime。调用时请将{WorkspaceId}替换为真实的Workspace ID。 新加坡地域：wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime。调用时请将{WorkspaceId}替换为真实的业务空间ID。

• 以下参数通过RealtimeSession的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',
      },
  )
  

  参数	类型	是否必须	说明
  output_modalities	List[str]	是	模型输出模态，固定为["text"]。
  enable_turn_detection	bool	否	是否开启服务端语音活动检测（VAD）。关闭后，需手动调用commit()方法触发识别。 默认值：True。 取值范围： True：开启 False：关闭
  turn_detection_type	str	否	服务端VAD类型，固定为 server_vad。
  turn_detection_threshold	float	否	VAD检测阈值。推荐将该值设为0.0。 默认值：0.5。 取值范围：[-1, 1]。 较低的阈值会提高 VAD 的灵敏度，可能将背景噪音误判为语音。较高的阈值则降低灵敏度，有助于在嘈杂环境中减少误触发。
  turn_detection_silence_duration_ms	int	否	VAD断句检测阈值（ms）。静音持续时长超过该阈值将被认为是语句结束。推荐将该值设为400。 默认值：800。 取值范围：[200, 6000]。 较低的值（如 300ms）可使模型更快响应，但可能导致在自然停顿处发生不合理的断句。较高的值（如 1200ms）可更好地处理长句内的停顿，但会增加整体响应延迟。
  transcription_params	dict	否	语音识别相关配置。

• 以下参数通过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',
      },
  )
  

  参数	类型	是否必须	说明
  language	str	否	音频源语言。 zh：中文（普通话、四川话、闽南语、吴语） yue：粤语 en：英文 ja：日语 de：德语 ko：韩语 ru：俄语 fr：法语 pt：葡萄牙语 ar：阿拉伯语 it：意大利语 es：西班牙语 hi：印地语 id：印尼语 th：泰语 tr：土耳其语 uk：乌克兰语 vi：越南语 cs：捷克语 da：丹麦语 fil：菲律宾语 fi：芬兰语 is：冰岛语 ms：马来语 no：挪威语 pl：波兰语 sv：瑞典语
  sample_rate	int	否	音频采样率（Hz）。支持16000和8000。 默认值：16000。 设置为 8000 时，服务端会先升采样到16000Hz再进行识别，可能引入微小延迟。建议仅在源音频为8000Hz（如电话线路）时使用。
  input_audio_format	str	否	音频格式。支持pcm和opus。 默认值：pcm。

关键接口

RealtimeSession（client.audio.realtime.connect）

实时识别会话通过client.audio.realtime.connect(model=...)创建：进入with上下文（或调用会话的connect()方法）后建立连接。

def connect(self) -> RealtimeSession

def update_session(self,
                   output_modalities: List[str] = ...,
                   *,
                   voice: str = ...,
                   input_audio_format: str = ...,
                   output_audio_format: str = ...,
                   enable_input_audio_transcription: bool = ...,
                   input_audio_transcription_model: str = ...,
                   enable_turn_detection: bool = ...,
                   turn_detection_type: str = ...,
                   prefix_padding_ms: int = ...,
                   turn_detection_threshold: float = ...,
                   turn_detection_silence_duration_ms: int = ...,
                   turn_detection_param: dict = ...,
                   translation_params: dict = ...,
                   transcription_params: dict = ...,
                   ...) -> None

def append_audio(self, audio_b64: Union[str, bytes]) -> None

def commit(self) -> None

def end_session(self) -> None

def close(self) -> None

def events(self) -> Iterator[Dict]

session.session_id
session.last_response_id

事件迭代（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']}")

def events(self) -> Iterator[Dict]

def recv_event(self) -> Optional[Dict]