Alibaba Cloud Model Studio:Python SDK

Parâmetros e interfaces do SDK Python para reconhecimento de fala em tempo real do Paraformer.

Guia do usuário: Para introdução aos modelos e recomendações de seleção, consulte Real-time speech recognition - Fun-ASR/Paraformer.

Pré-requisitos

Ative o serviço e Obtain an API key. Configure API key as an environment variable em vez de codificá-la diretamente no código para evitar riscos de segurança causados por vazamento de código.

• Install the latest DashScope SDK.

Lista de modelos

Primeiros passos

O recurso principal (client.audio.asr) fornece interfaces para chamadas sem streaming e com streaming bidirecional. Selecione o método apropriado conforme suas necessidades:

• Chamada sem streaming (recognize_file): reconhece um arquivo local. O SDK envia o áudio automaticamente em blocos e conduz a tarefa até o fim. Adequado para processamento de áudio pré-gravado.
• Chamada com streaming bidirecional: reconhece um fluxo de áudio diretamente e gera resultados em tempo real. O fluxo pode vir de um dispositivo externo, como um microfone, ou ser lido de um arquivo local. Ideal para cenários que exigem feedback imediato.

Chamada sem streaming

Este método envia uma tarefa de transcrição de fala em tempo real para um arquivo local. O processo bloqueia a execução até que o resultado completo da transcrição seja retornado.

Chame o método recognize_file do recurso principal (client.audio.asr) para vincular os parâmetros da solicitação e passar o caminho do arquivo local, iterando os eventos do servidor para obter o RecognitionResult.

import os
import dashscope

# A configuração abaixo refere-se à região China (Beijing). Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    # Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# Reconheça um arquivo de áudio local: recognize_file envia o áudio automaticamente em blocos e conduz a tarefa até o fim
for event in client.audio.asr.recognize_file(
        '{YOUR_AUDIO_FILE}',
        model='paraformer-realtime-v2',
        format='wav',
        sample_rate=16000,
        # O parâmetro "language_hints" só é suportado pelo modelo paraformer-realtime-v2
        parameters={'language_hints': ['zh', 'en']}):
    payload = event.get('payload') or {}
    sentence = (payload.get('output') or {}).get('sentence') or {}
    if sentence.get('text') and not sentence.get('heartbeat'):
        print('Resultado do reconhecimento: ', sentence['text'])

Chamada com streaming bidirecional

Este método envia uma tarefa de transcrição de fala em tempo real e retorna resultados de reconhecimento em tempo real por meio da iteração de eventos.

1. Inicie o reconhecimento de fala em streaming

   Chame o método recognize do recurso principal (client.audio.asr) para criar a sessão de reconhecimento e vincular os parâmetros de solicitação; entre no contexto with (que chama automaticamente o método start para estabelecer a conexão e enviar a tarefa) para iniciar o reconhecimento de fala em streaming.

2. Streaming

   Chame repetidamente o método send_audio_frame da sessão para enviar o fluxo de áudio binário de um arquivo local ou dispositivo (como um microfone) ao servidor em segmentos. O envio e o recebimento devem ocorrer de forma concorrente: a thread receptora itera os eventos do servidor por meio da iteração de eventos (events()/results()), que retorna os resultados de reconhecimento ao cliente em tempo real.

   Recomenda-se que a duração de cada segmento de áudio enviado seja de aproximadamente 100 milissegundos, com tamanho de dados entre 1 KB e 16 KB.

3. Encerramento do processamento

   Chame o método stop da sessão para interromper o reconhecimento de fala.

   Este método apenas notifica o servidor de que a entrada terminou; a thread receptora deve continuar consumindo events() até o evento task-finished para obter os últimos resultados de reconhecimento e as estatísticas. Os erros da camada WebSocket são lançados como exceções DashScopeError e podem ser capturados com try/except no envio/recebimento.

import os
import signal  # para tratar eventos de teclado (pressione "Ctrl+C" para encerrar a gravação)
import sys

import dashscope
import pyaudio
from concurrent.futures import ThreadPoolExecutor

# A configuração abaixo refere-se à região China (Beijing). Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    # Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

mic = None
stream = None

# Defina os parâmetros de gravação
sample_rate = 16000  # taxa de amostragem (Hz)
channels = 1  # canal mono
dtype = 'int16'  # tipo de dados
format_pcm = 'pcm'  # o formato dos dados de áudio
block_size = 3200  # número de quadros por buffer

# Receptor de eventos do reconhecimento de fala em tempo real
def receive_events(rec):
    for event in rec.events():
        payload = event.get('payload') or {}
        sentence = (payload.get('output') or {}).get('sentence') or {}
        if sentence.get('text') and not sentence.get('heartbeat'):
            print('Texto reconhecido: ', sentence['text'])
            if sentence.get('sentence_end'):
                print(
                    'Fim da frase, request_id:%s, usage:%s'
                    % (rec.request_id, rec.usage))

def signal_handler(sig, frame):
    print('Ctrl+C pressionado, interrompendo o reconhecimento ...')
    # Interrompa o reconhecimento
    recognition.stop()
    print('Reconhecimento interrompido.')
    print(
        '[Métrica] requestId: {}, atraso do primeiro pacote ms: {}, atraso do último pacote ms: {}'
        .format(
            recognition.request_id,
            recognition.first_package_delay,
            recognition.last_package_delay,
        ))
    # Force a saída do programa
    sys.exit(0)

# função principal
if __name__ == '__main__':
    print('Inicializando ...')

    signal.signal(signal.SIGINT, signal_handler)
    print("Pressione 'Ctrl+C' para interromper a gravação e o reconhecimento...")

    # Chame o service de reconhecimento; você pode personalizar os parâmetros de reconhecimento, como model, format,
    # sample_rate
    with client.audio.asr.recognize(
            model='paraformer-realtime-v2',
            format=format_pcm,
            # 'pcm', 'wav', 'opus', 'speex', 'aac' ou 'amr'; consulte os formatos suportados na documentação
            sample_rate=sample_rate,
            # 8000 ou 16000 são suportados
            parameters={'semantic_punctuation_enabled': False},
    ) as recognition:
        # Abra o microfone
        mic = pyaudio.PyAudio()
        stream = mic.open(format=pyaudio.paInt16,
                          channels=1,
                          rate=16000,
                          input=True)
        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events, recognition)
            try:
                # Envie o áudio do microfone até que "Ctrl+C" seja pressionado
                while stream:
                    data = stream.read(3200, exception_on_overflow=False)
                    recognition.send_audio_frame(data)
                recognition.stop()
                receiver.result()
            finally:
                # close deve ser executado dentro do contexto do pool para desbloquear o receptor
                recognition.close()
        stream.stop_stream()
        stream.close()
        mic.terminate()
        stream = None
        mic = None

import os
import time
import dashscope
from concurrent.futures import ThreadPoolExecutor

# A configuração abaixo refere-se à região China (Beijing). Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    # Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# Receptor de eventos do reconhecimento de fala em tempo real
def receive_events(rec):
    for event in rec.events():
        payload = event.get('payload') or {}
        sentence = (payload.get('output') or {}).get('sentence') or {}
        if sentence.get('text') and not sentence.get('heartbeat'):
            print(get_timestamp() + ' Texto reconhecido: ', sentence['text'])
            if sentence.get('sentence_end'):
                print(get_timestamp() +
                    'Fim da frase, request_id:%s, usage:%s'
                    % (rec.request_id, rec.usage))

with client.audio.asr.recognize(
        model='paraformer-realtime-v2',
        format='wav',
        sample_rate=16000,
        # O parâmetro "language_hints" só é suportado pelo modelo paraformer-realtime-v2
        parameters={'language_hints': ['zh', 'en']},
) as recognition:
    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_events, recognition)
        try:
            audio_data: bytes = None
            f = open("{YOUR_AUDIO_FILE}", 'rb')
            if os.path.getsize("{YOUR_AUDIO_FILE}"):
                while True:
                    audio_data = f.read(3200)
                    if not audio_data:
                        break
                    else:
                        recognition.send_audio_frame(audio_data)
                    time.sleep(0.1)
            else:
                raise Exception(
                    'The supplied file was empty (zero bytes long)')
            f.close()
            recognition.stop()
            receiver.result()
        finally:
            # close deve ser executado dentro do contexto do pool para desbloquear o receptor
            recognition.close()

print(
    '[Métrica] requestId: {}, atraso do primeiro pacote ms: {}, atraso do último pacote ms: {}'
    .format(
        recognition.request_id,
        recognition.first_package_delay,
        recognition.last_package_delay,
    ))

Chamadas concorrentes

No Python, devido ao Global Interpreter Lock (GIL), apenas uma thread pode executar código Python por vez (embora algumas bibliotecas focadas em desempenho possam remover essa limitação). Para aproveitar melhor os recursos computacionais de um computador multicore, recomendamos o uso de multiprocessing ou concurrent.futures.ProcessPoolExecutor. O multithreading pode aumentar significativamente a latência das chamadas do SDK sob alta concorrência.

Parâmetros de solicitação

Os parâmetros de solicitação são definidos nos métodos recognize/recognize_file do recurso principal (client.audio.asr): model, format, sample_rate, vocabulary_id e disfluency_removal_enabled são parâmetros nomeados; os demais parâmetros (como language_hints e semantic_punctuation_enabled) são passados por meio do dicionário parameters.

Interfaces principais

Recurso principal (client.audio.asr)

O reconhecimento de fala em tempo real é montado no recurso client.audio.asr: recognize cria uma sessão de reconhecimento com streaming bidirecional (Recognition) e recognize_file faz o reconhecimento de um arquivo local em uma única chamada.

def recognize(self,
              *,
              model: str,
              format: str = "pcm",
              sample_rate: int = 16000,
              vocabulary_id: str = None,
              disfluency_removal_enabled: bool = None,
              diarization_enabled: bool = None,
              parameters: Dict = None,
              input: Dict = None,
              phrase_id: str = None,
              ...) -> Recognition

def recognize_file(self,
                   path: str,
                   *,
                   model: str,
                   format: str = "pcm",
                   sample_rate: int = 16000,
                   ...) -> Iterator[Dict]

def start(self)

def send_audio_frame(self, buffer: bytes)

def stop(self)

def events(self) -> Iterator[Dict]
def results(self) -> Iterator[RecognitionResult]

def close(self)

rec.request_id
rec.first_package_delay
rec.last_package_delay

Iteração de eventos (events()/results())

Durante uma chamada com streaming bidirecional, o servidor retorna informações e dados importantes do processo ao cliente por meio de eventos. Itere os events() da sessão (eventos dict brutos) ou results() (frames tipados de resultado de reconhecimento) na thread receptora, para processar as informações e dados retornados.

# Itere os eventos brutos: o texto reconhecido está em payload.output.sentence
def receive_events(rec):
    for event in rec.events():
        payload = event.get('payload') or {}
        sentence = (payload.get('output') or {}).get('sentence') or {}
        if sentence.get('text') and not sentence.get('heartbeat'):
            print('Texto reconhecido: ', sentence['text'])


# Ou itere apenas os frames de resultado de reconhecimento (retorna RecognitionResult, frames de heartbeat filtrados)
def receive_results(rec):
    for result in rec.results():
        sentence = result.get_sentence()
        if sentence and 'text' in sentence:
            print('Texto reconhecido: ', sentence['text'])

event["header"]["event"] == "task-started"

def events(self) -> Iterator[Dict]

def results(self) -> Iterator[RecognitionResult]

try:
    ...
except DashScopeError as e:
    print(e.code, e.message, e.request_id)

event["header"]["event"] == "task-finished"

Resultados da resposta

Resultado de reconhecimento (RecognitionResult)

RecognitionResult representa um frame de resultado de reconhecimento produzido pela iteração de results() da sessão (utilizável tanto na chamada com streaming bidirecional quanto na chamada sem streaming; events() produz eventos dict brutos — use results() se precisar dessa visão tipada).

def get_sentence(self) -> Union[Dict[str, Any], List[Any]]

result.request_id

def is_sentence_end(self) -> bool

Frase (Sentence)

Os membros da classe Sentence são os seguintes:

Informações de carimbo de data/hora por palavra (Word)

Os membros da classe Word são os seguintes:

Códigos de erro

Se encontrar erros, consulte Error codes para solução de problemas.

Se o problema persistir, junte-se à comunidade de desenvolvedores para relatar seu problema e forneça o Request ID para investigação adicional.

Mais exemplos

Para mais exemplos, consulte o GitHub.

FAQ

Recursos

P: Como manter uma conexão longa com o servidor durante silêncio prolongado?

Defina o parâmetro de solicitação heartbeat como true e envie continuamente áudio silencioso ao servidor.

Áudio silencioso refere-se a arquivos de áudio ou fluxos de dados que não contêm sinal sonoro. Áudio silencioso pode ser gerado por vários métodos, como softwares de edição de áudio (Audacity ou Adobe Audition) ou ferramentas de linha de comando como FFmpeg.

P: Como converter áudio para um formato suportado?

Utilize a ferramenta FFmpeg. Para mais informações de uso, consulte o site oficial do FFmpeg.

# Basic conversion command (universal template)
# -i: Input file path. Example: audio.wav
# -c:a: Audio codec. Example: aac, libmp3lame, pcm_s16le
# -b:a: Bitrate (quality control). Example: 192k, 320k
# -ar: Sample rate. Example: 44100 (CD), 48000, 16000
# -ac: Number of channels. Example: 1 (mono), 2 (stereo)
# -y: Overwrite existing file (no value needed)
ffmpeg -i input_audio.ext -c:a codec_name -b:a bitrate -ar sample_rate -ac channels output.ext

# Example: WAV -> MP3 (preserve original quality)
ffmpeg -i input.wav -c:a libmp3lame -q:a 0 output.mp3
# Example: MP3 -> WAV (16-bit PCM standard format)
ffmpeg -i input.mp3 -c:a pcm_s16le -ar 44100 -ac 2 output.wav
# Example: M4A -> AAC (extract/convert Apple audio)
ffmpeg -i input.m4a -c:a copy output.aac  # Direct extraction without re-encoding
ffmpeg -i input.m4a -c:a aac -b:a 256k output.aac  # Re-encode for higher quality
# Example: FLAC lossless -> Opus (high compression)
ffmpeg -i input.flac -c:a libopus -b:a 128k -vbr on output.opus

P: Há suporte para visualizar o intervalo de tempo de cada frase?

Sim. Os resultados do reconhecimento de fala incluem os carimbos de data/hora de início e fim de cada frase, que podem ser usados para determinar o intervalo de tempo de cada frase.

P: Como reconheço um arquivo local (arquivo de áudio gravado)?

Existem duas maneiras de reconhecer um arquivo local:

• Passar diretamente o caminho do arquivo local: Este método retorna o resultado completo do reconhecimento após o arquivo ser totalmente processado. Não é adequado para cenários que exigem feedback imediato.

  Passe o caminho do arquivo para o método recognize_file para reconhecer diretamente o arquivo de áudio. Para mais informações, consulte Chamada sem streaming.

• Converter o arquivo local em um fluxo binário para reconhecimento: Este método retorna resultados de reconhecimento como um fluxo enquanto o arquivo está sendo processado. É adequado para cenários que exigem feedback imediato.

  Use o método send_audio_frame da sessão de reconhecimento para enviar um fluxo binário ao servidor para reconhecimento. Para mais informações, consulte Chamada com streaming bidirecional.

Solução de problemas

P: O que causa falha no reconhecimento de fala (sem resultados de reconhecimento)?

1. Verifique se o formato de áudio (format) e a taxa de amostragem (sampleRate/sample_rate) nos parâmetros de solicitação estão definidos corretamente e cumprem as restrições de parâmetros. Veja abaixo exemplos comuns de erros:

   • A extensão do arquivo de áudio é .wav, mas o formato real é MP3, e o parâmetro de solicitação format está definido como wav (configuração incorreta de parâmetro).
   • A taxa de amostragem do áudio é 3600 Hz, mas o parâmetro de solicitação sampleRate/sample_rate está definido como 48000 (configuração incorreta de parâmetro).

   Use a ferramenta ffprobe para obter informações sobre container, codec, taxa de amostragem, canal e outros dados do áudio:

ffprobe -v error -show_entries format=format_name -show_entries stream=codec_name,sample_rate,channels -of default=noprint_wrappers=1 input.xxx

2. Ao usar o modelo paraformer-realtime-v2, verifique se o idioma definido em language_hints corresponde ao idioma real do áudio.

   Por exemplo: O áudio está realmente em chinês, mas language_hints está definido como en (inglês).

3. Se todas as verificações acima forem aprovadas, utilize palavras-chave personalizadas para melhorar a precisão do reconhecimento de palavras específicas.

Q: Como corrigir a ausência de resultados de reconhecimento causada por um parâmetro de formato de áudio incorreto?

format é obrigatório e deve corresponder ao formato real do áudio. Por exemplo, para um arquivo de áudio em formato MP3 com a extensão .wav, defina format='mp3'.

1. Verifique o formato real do áudio: Use ffprobe para consultar o contêiner, o codec, a taxa de amostragem e os canais do áudio.

   ffprobe -v error -show_entries format=format_name -show_entries stream=codec_name,sample_rate,channels -of default=noprint_wrappers=1 your_audio_file
   

2. Defina os parâmetros da solicitação: Defina format como o formato real do áudio. Defina sample_rate como a taxa de amostragem real e verifique se ela atende aos requisitos do modelo utilizado.

3. Converta o formato do áudio se necessário: Se precisar de áudio WAV, use o comando a seguir para gerar um arquivo WAV com codificação PCM, taxa de amostragem de 16 kHz e um canal.

   ffmpeg -i input_file -c:a pcm_s16le -ar 16000 -ac 1 output.wav
   

   Após a conversão, defina format='wav' e sample_rate=16000.