Beranda → Blog → Cara membangun pipeline transkripsi untuk arsip audio

Cara membangun pipeline transkripsi untuk arsip audio

Satu rekaman adalah satu baris kode. Sepuluh ribu rekaman adalah persoalan lain: sesuatu akan gagal di tengah jalan, prosesnya akan dimatikan, separuh berkas sudah selesai, dan Anda perlu tahu separuh yang mana. Beginilah bentuk pipeline massal yang selamat dari semua itu.

Apa yang rusak saat naik dari satu berkas ke sepuluh ribu

Perulangan atas satu folder bekerja baik sampai arsipnya membesar. Setelah itu empat hal mulai salah:

• Selalu ada yang gagal. Dari sepuluh ribu permintaan, beberapa akan kena gangguan jaringan atau 5xx. Tanpa percobaan ulang, perulangan mati di berkas ke-4312.
• Prosesnya dimatikan. Deploy, kehabisan memori, tutup laptop. Kalau kemajuan hanya hidup di memori, Anda mulai dari nol dan membayar dua kali untuk menit yang sama.
• Berurutan itu lambat. Satu berkas makan beberapa detik; sepuluh ribu berurutan makan hampir sehari. Konkurensi mengubahnya jadi satu jam.
• Tagihannya tak terlihat. Penagihan per menit berarti biayanya ditentukan oleh audio yang Anda kirim, dan Anda baru tahu belakangan.

Setiap butir di bawah ada untuk memperbaiki salah satu dari keempatnya.

Langkah 1. Buat kemajuan yang tahan mati

Keputusan terpenting: catatan tentang apa yang sudah selesai harus selamat melewati prosesnya. Satu direktori berisi berkas hasil sudah cukup — keberadaan keluaran adalah keadaannya.

import pathlib

SRC = pathlib.Path("rekaman")
OUT = pathlib.Path("transkrip")
OUT.mkdir(exist_ok=True)

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

Sekarang pekerjaannya idempoten: jalankan, matikan, jalankan lagi — ia menyambung persis dari tempat berhenti dan tidak pernah membayar dua kali untuk berkas yang sama. Untuk arsip lebih besar, tabel SQLite dengan kolom status bekerja sama saja dan memberi Anda kueri; prinsipnya tidak berubah.

Tulis keluarannya secara atomik, atau proses yang mati di tengah penulisan meninggalkan berkas terpotong yang tampak selesai:

tmp = out.with_suffix(".part")
tmp.write_text(teks, encoding="utf-8")
tmp.rename(out)          # atomik: tidak ada berkas baru, atau berkas yang utuh

Langkah 2. Proses beberapa berkas sekaligus

Transkripsi adalah menunggu jaringan, bukan kerja CPU lokal, jadi utas adalah alat yang tepat dan satu kolam utas sudah cukup:

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 transkrip(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, teks in zip(tertunda(), pool.map(transkrip, tertunda())):
        (OUT / (path.stem + ".txt")).write_text(teks, encoding="utf-8")

Pilih max_workers menurut batas permintaan per detik pada paket Anda, bukan menurut jumlah inti prosesor. Melebihi batas tidak mempercepat apa pun — ia hanya mengubah permintaan yang berhasil menjadi 429.

Langkah 3. Ulangi yang gagal, bukan yang berhasil

Dua kelas galat butuh penanganan yang berlawanan. 5xx atau kehabisan waktu layak diulang; 400 pada berkas rusak akan gagal dengan cara yang sama selamanya.

import time
from openai import APIStatusError, APIConnectionError

def dengan_ulang(path, percobaan=3):
    for n in range(percobaan):
        try:
            return transkrip(path)
        except APIConnectionError:
            pass                                  # gangguan jaringan — ulangi
        except APIStatusError as e:
            if e.status_code == 429 or e.status_code >= 500:
                pass                              # limit atau galat server — ulangi
            else:
                raise                             # 400/401 — mengulang tidak menolong
        time.sleep(2 ** n)                        # 1 dtk, 2 dtk, 4 dtk
    raise RuntimeError(f"gagal setelah {percobaan}: {path}")

Jeda yang membesar penting khusus untuk 429: mengulang seketika membuat Anda tetap di atas batas. Perhatikan max_retries=0 pada klien di atas — tanpa itu SDK juga mengulang dan kedua kebijakan saling bertabrakan.

Jangan biarkan satu berkas buruk menghentikan seluruh jalannya. Catat, lanjutkan, dan urus daftar kegagalannya di akhir — biasanya dua atau tiga berkas dari ribuan.

Langkah 4. Siapkan audionya sebelum dikirim

Di sinilah tagihan sebenarnya ditentukan. Dua aturan mengerjakan sebagian besarnya:

• Konversi sekali, di awal. Mono 16 kHz adalah yang dipakai model di dalamnya, dan ia terunggah jauh lebih cepat daripada wav aslinya.
• Lewati yang tidak berisi ucapan. Keheningan tetap memakan menit.

# mengonversi seluruh arsip dalam satu jalan
find rekaman -name '*.wav' -print0 |
  xargs -0 -P4 -I{} ffmpeg -loglevel error -i {} -ac 1 -ar 16000 -b:a 48k {}.mp3

Berkas di atas 25 MB tetap harus dipotong; dengan mono 48 kbit/dtk ambang itu setara sekitar satu jam audio, jadi setelah konversi kebanyakan arsip berhenti membutuhkan pemotongan sama sekali.

Langkah 5. Ketahui biayanya sebelum mulai

Penagihan per menit audio, jadi totalnya bisa diketahui di muka — ukur arsipnya alih-alih menebak:

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

import subprocess

def menit(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(menit(p) for p in SRC.glob("*.mp3"))
print(f"{total:.0f} menit, {total * TARIF:.2f} pada tarif saat ini")

Jalankan ini sebelum pipeline, bukan sesudahnya. Butuh satu menit, tidak berbiaya, dan itulah beda antara pengeluaran yang direncanakan dan kejutan.

Langkah 6. Awasi jalannya

Proses massal yang panjang butuh tiga angka yang terlihat: selesai, gagal, dan laju. Satu penghitung yang dicetak tiap seratus berkas sudah cukup:

selesai = gagal = 0
for path in tertunda():
    try:
        simpan(path, dengan_ulang(path))
        selesai += 1
    except Exception as e:
        gagal += 1
        print("GAGAL", path.name, e)
    if (selesai + gagal) % 100 == 0:
        print(f"{selesai} selesai, {gagal} gagal")

Area klien menampilkan jalannya yang sama dari sisi lain — menit terpakai, jumlah permintaan, dan riwayat per permintaan — dan itu cara tercepat memastikan pipeline melakukan persis seperti yang Anda kira.

Yang tidak perlu dibangun

• Broker pesan untuk sepuluh ribu berkas. Satu folder plus kolam utas sanggup pada skala itu. Antrean baru masuk akal saat pekerjaan datang terus-menerus, bukan saat ini arsip sekali jalan.
• Pembatas laju buatan sendiri. Setel ukuran kolam di bawah batas paket dan Anda tidak akan pernah menyentuhnya.
• Penghitung posisi untuk melanjutkan. Menyimpan «berhenti di berkas 4312» rusak begitu urutan masukan berubah. Turunkan keadaannya dari apa yang ada di disk.

Pertanyaan yang sering diajukan

Bacaan terkait

• Cara mengubah rekaman telepon menjadi teks — Panduan langkah demi langkah mengubah rekaman percakapan telepon menjadi teks lewat API dalam hitungan menit. Lengkap dengan contoh Python, C#, dan daftar galat.
• Harga transkripsi: sebenarnya Anda membayar untuk apa — Bagaimana penagihan transkripsi bekerja, bagian audio mana yang diam-diam memakan biaya, kapan GPU sendiri lebih murah daripada API, dan cara memperkirakan biaya bulanan.
• Cara meningkatkan akurasi transkripsi: delapan langkah praktis — Delapan perubahan yang benar-benar menggerakkan kualitas transkripsi: menyiapkan audio, menyebut bahasa, parameter prompt, memotong di jeda, format, dan cara mengukur galat.