Início → Blog → Como montar um pipeline de transcrição para um acervo de áudio

Como montar um pipeline de transcrição para um acervo de áudio

Uma gravação é uma linha de código. Dez mil gravações são outro problema: alguma coisa vai falhar no meio, o processo vai ser morto, metade dos arquivos já estará pronta e você vai precisar saber qual metade. É este o formato de um pipeline em lote que sobrevive a tudo isso.

O que quebra ao ir de um arquivo para dez mil

Um laço sobre uma pasta funciona bem até o acervo crescer. Aí quatro coisas dão errado:

• Algo sempre falha. Em dez mil requisições, algumas vão pegar uma oscilação de rede ou um 5xx. Sem novas tentativas, o laço morre no arquivo 4312.
• O processo é morto. Um deploy, um OOM, a tampa do notebook. Se o progresso só existe na memória, você recomeça do zero e paga duas vezes pelos mesmos minutos.
• Sequencial é lento. Um arquivo leva alguns segundos; dez mil em sequência tomam quase um dia. Concorrência transforma isso em uma hora.
• A fatura é invisível. Cobrança por minuto significa que o custo é decidido pelo áudio que você envia, e você descobre depois.

Cada ponto abaixo existe para resolver um desses quatro.

Passo 1. Torne o progresso durável

A decisão mais importante: o registro do que já foi feito precisa sobreviver ao processo. Um diretório de arquivos de resultado basta — a presença da saída é o estado.

import pathlib

SRC = pathlib.Path("gravacoes")
OUT = pathlib.Path("transcricoes")
OUT.mkdir(exist_ok=True)

def pendentes():
    for p in sorted(SRC.glob("*.mp3")):
        if not (OUT / (p.stem + ".txt")).exists():
            yield p

Agora o trabalho é idempotente: rode, mate, rode de novo — ele retoma exatamente de onde parou e nunca paga duas vezes pelo mesmo arquivo. Para acervos maiores, uma tabela SQLite com coluna de status funciona igual e ainda dá consultas; o princípio não muda.

Escreva a saída de forma atômica, ou um processo morto no meio da escrita deixa um arquivo truncado com cara de pronto:

tmp = out.with_suffix(".part")
tmp.write_text(texto, encoding="utf-8")
tmp.rename(out)          # atômico: ou não há arquivo novo, ou ele está completo

Passo 2. Processe vários arquivos ao mesmo tempo

Transcrição é espera de rede, não trabalho de CPU local, então threads são a ferramenta certa e um pool basta:

import os
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI

client = OpenAI(base_url="https://voicesscribe.com/v1", api_key=os.environ["VS_KEY"], max_retries=0)

def transcrever(path):
    with path.open("rb") as f:
        return client.audio.transcriptions.create(model="whisper-1", file=f).text

with ThreadPoolExecutor(max_workers=4) as pool:
    for path, texto in zip(pendentes(), pool.map(transcrever, pendentes())):
        (OUT / (path.stem + ".txt")).write_text(texto, encoding="utf-8")

Escolha max_workers pelo limite de requisições por segundo do seu plano, não pelo número de núcleos. Ir além do limite não acelera nada — apenas converte requisições bem-sucedidas em 429.

Passo 3. Repita as falhas, não os sucessos

Duas classes de erro pedem tratamentos opostos. Um 5xx ou um timeout valem uma nova tentativa; um 400 num arquivo corrompido vai falhar igual para sempre.

import time
from openai import APIStatusError, APIConnectionError

def com_retry(path, tentativas=3):
    for n in range(tentativas):
        try:
            return transcrever(path)
        except APIConnectionError:
            pass                                  # oscilação de rede — repetir
        except APIStatusError as e:
            if e.status_code == 429 or e.status_code >= 500:
                pass                              # limite ou erro do servidor — repetir
            else:
                raise                             # 400/401 — repetir não resolve
        time.sleep(2 ** n)                        # 1s, 2s, 4s
    raise RuntimeError(f"falhou após {tentativas}: {path}")

O recuo exponencial importa especialmente no 429: repetir imediatamente mantém você acima do limite. Note o max_retries=0 no cliente acima — sem isso o SDK também repete e as duas políticas brigam entre si.

Não deixe um arquivo ruim parar a execução. Registre, siga em frente e trate a lista de falhas no fim — normalmente são dois ou três arquivos em milhares.

Passo 4. Prepare o áudio antes de enviar

É aqui que a fatura é decidida de verdade. Duas regras fazem quase todo o trabalho:

• Converta uma vez, no começo. Mono a 16 kHz é o que o modelo usa internamente e sobe muito mais rápido que o wav original.
• Pule o que não tem fala. Silêncio custa minutos do mesmo jeito.

# converte o acervo inteiro numa passada
find gravacoes -name '*.wav' -print0 |
  xargs -0 -P4 -I{} ffmpeg -loglevel error -i {} -ac 1 -ar 16000 -b:a 48k {}.mp3

Arquivos acima de 25 MB precisam ser divididos de qualquer jeito; com mono a 48 kbit/s esse limiar fica em torno de uma hora de áudio, então a maioria dos acervos deixa de precisar de divisão depois da conversão.

Passo 5. Saiba o custo antes de começar

A cobrança é por minuto de áudio, então o total é conhecível de antemão — meça o acervo em vez de chutar:

ffprobe -v error -show_entries format=duration -of csv=p=0 arquivo.mp3

import subprocess

def minutos(path):
    out = subprocess.run(["ffprobe", "-v", "error", "-show_entries",
                          "format=duration", "-of", "csv=p=0", str(path)],
                         capture_output=True, text=True).stdout
    return float(out) / 60

total = sum(minutos(p) for p in SRC.glob("*.mp3"))
print(f"{total:.0f} min, {total * TARIFA:.2f} na tarifa atual")

Rode isso antes do pipeline, não depois. Leva um minuto, não custa nada e é a diferença entre um gasto planejado e uma surpresa.

Passo 6. Acompanhe a execução

Um lote longo precisa de três números visíveis: prontos, falhos e ritmo. Um contador impresso a cada cem arquivos já basta:

prontos = falhos = 0
for path in pendentes():
    try:
        salvar(path, com_retry(path))
        prontos += 1
    except Exception as e:
        falhos += 1
        print("FALHA", path.name, e)
    if (prontos + falhos) % 100 == 0:
        print(f"{prontos} prontos, {falhos} falhos")

A área do cliente mostra a mesma execução pelo outro lado — minutos consumidos, requisições e o histórico por requisição — que é a forma mais rápida de confirmar que o pipeline está fazendo o que você imagina.

O que não construir

• Um broker de mensagens para dez mil arquivos. Uma pasta e um pool de threads dão conta dessa escala. Fila é para trabalho que chega continuamente, não para um acervo pontual.
• Um limitador de taxa próprio. Deixe o pool abaixo do limite do plano e você nunca esbarra em um.
• Um contador de posição para retomar. Guardar «parei no arquivo 4312» quebra assim que a ordem de entrada muda. Derive o estado do que existe em disco.

Perguntas frequentes

Leitura relacionada

• Como transcrever a gravação de uma chamada — Guia passo a passo para transformar a gravação de uma ligação em texto pela API em poucos minutos. Com exemplos em Python e C# e a lista de erros mais comuns.
• Preço da transcrição: pelo que você realmente paga — Como funciona a cobrança da transcrição, quais partes do áudio custam dinheiro sem você notar, quando uma GPU própria sai mais barata e como estimar o gasto mensal.
• Como melhorar a precisão da transcrição: oito ajustes práticos — Oito mudanças que realmente movem a qualidade da transcrição: preparo do áudio, idioma explícito, o parâmetro prompt, corte em pausas, formatos e como medir o erro.