How to Build a Production-Ready Audio Transcription Pipeline in Python

Transcribing an audio file from Python looks simple in a demo:

1. Open a file.
2. Send it to an API.
3. Print the returned text.

Then real audio arrives.

A user uploads an MP3 instead of a WAV. A phone recording is narrowband. Two speakers interrupt each other. A 90-minute recording times out halfway through. The network returns a transient error. Your application needs timestamps rather than one giant string.

That is where transcription stops being an API-call problem and becomes a pipeline problem.

This guide builds that pipeline step by step. We will preprocess audio, send pre-recorded files through a speech-to-text API, handle hosted audio, work with structured transcription data, add speaker diarization, split long recordings, and introduce safer retry patterns.

For the implementation examples, we’ll use Smallest AI and its Pulse speech-to-text models.

Start with the full transcription pipeline

A production transcription workflow is more than “audio in, text out.”

A useful mental model is:

1. Receive or locate the audio.
2. Inspect and normalize it when necessary.
3. Choose the transcription mode and model.
4. Authenticate the request securely.
5. Send the audio bytes or hosted URL.
6. Receive structured JSON.
7. Extract transcript, timestamps, speakers, and metadata.
8. Store or post-process the result.
9. Handle failures, retries, duplicates, and long-running jobs.

That last part matters.

The actual HTTP request may only occupy a few lines of Python. Most of the engineering work happens around it.

A transcript might eventually power search, captions, meeting notes, call QA, downstream automation, analytics, or a voice application. Once that happens, timestamps, speaker boundaries, error handling, and repeatable input formats become as important as the transcript string itself.

For a more API-focused look at recorded audio ingestion, the programmatic audio-to-text API workflow covers the broader recorded-audio pattern.

What “transcribe” means at the API boundary

At the application layer, your responsibility is usually straightforward: provide valid audio and enough configuration for the transcription service to interpret it correctly.

Behind that API boundary, an ASR system has to map an acoustic signal into linguistic units, decode those units into likely text, and produce a usable output representation.

Depending on the system and configuration, that output can include:

• the complete transcript
• word-level timestamps
• utterance boundaries
• speaker labels
• language information
• confidence information
• processing metadata

The structured fields are what make transcription useful to software.

If you are building captions, for example, you need timing. If you are processing customer calls, you may need speaker labels. If transcripts trigger downstream actions, you may want to inspect confidence or other quality signals before trusting every token automatically.

This also explains why testing only on clean demo audio is risky. Acoustic conditions, microphones, codecs, accents, background noise, overlapping speech, and domain-specific terminology can all change what the model receives.

Set up the Python environment

Start with an isolated virtual environment:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

python -m venv stt-env

# macOS / Linux
source stt-env/bin/activate

# Windows
stt-env\Scripts\activate

pip install requests python-dotenv pydub tenacity

requests handles the HTTP calls, python-dotenv is useful during local development, pydub handles common audio preprocessing tasks, and tenacity gives us controlled retry behavior later.

If you use pydub with compressed formats such as MP3, make sure FFmpeg is installed on the machine running the application.

Normalize unpredictable audio before transcription

Tutorial audio is usually clean.

Production audio rarely is.

You may receive MP3, WAV, compressed call recordings, extracted video audio, stereo conversations, or recordings with inconsistent sample rates.

For a predictable baseline, converting incoming files to mono, 16 kHz WAV is useful. Smallest AI’s current Pulse documentation recommends a 16 kHz sample rate, and converting to a known format also removes one variable when you debug failures.

Do not interpret resampling as a way to recreate information that was never captured. Converting an 8 kHz telephone recording to 16 kHz does not restore frequencies lost during recording. The point is predictable input, not magic audio repair.

Here is a small preprocessing function:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

from pathlib import Path

from pydub import AudioSegment


def preprocess_audio(input_path: str, output_path: str) -> str:
    """Convert an audio file to mono, 16 kHz, 16-bit PCM WAV."""
    source = Path(input_path)

    if not source.is_file():
        raise FileNotFoundError(f"Audio file not found: {source}")

    audio = AudioSegment.from_file(source)
    audio = audio.set_channels(1)
    audio = audio.set_frame_rate(16_000)
    audio = audio.set_sample_width(2)

    audio.export(output_path, format="wav")
    return output_path


if __name__ == "__main__":
    preprocess_audio("input_audio.mp3", "preprocessed_audio.wav")

This is local preprocessing only, so no API credentials are involved.

You also should not preprocess blindly.

If your input is already in a supported, appropriate format, another lossy conversion can do more harm than good. Inspect the audio first and normalize when your pipeline actually needs it.

Choose the right transcription mode

Smallest AI currently exposes Pulse and Pulse Pro through the same pre-recorded speech-to-text endpoint.

The important distinction for this workflow is:

• pulse-pro is intended for pre-recorded English transcription.
• pulse supports multilingual transcription and is also used when you need capabilities such as audio-by-URL, streaming, or speaker diarization.

Both use the unified pre-recorded endpoint:

https://api.smallest.ai/waves/v1/stt/

The model is selected through the model query parameter.

For developers implementing the pipeline, the Smallest AI speech-to-text API provides the programmatic entry point used by these examples.

Create and store the API key

Keep the API key in an environment variable rather than hard-coding it into the application.

Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.

export SMALLEST_API_KEY="your-api-key-here"

Every authenticated request sends the value through the Authorization header:

Authorization: Bearer <SMALLEST_API_KEY value>

Keep the key on your server. Do not expose it in browser JavaScript, mobile application code, public repositories, screenshots, or client-side logs.

For production deployments, store it in a server-side secrets manager provided by your cloud or infrastructure platform rather than committing a .env file to the repository.

Make the first transcription request from Python

For a pre-recorded English file, we can send the raw file bytes using Pulse Pro.

The request uses application/octet-stream and asks for word timestamps so the response can carry more structure than plain text.

Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.

import os
from pathlib import Path

import requests


ENDPOINT = "https://api.smallest.ai/waves/v1/stt/"


def transcribe_audio(file_path: str) -> dict:
    """Transcribe a pre-recorded English audio file with Pulse Pro."""
    audio_path = Path(file_path)

    if not audio_path.is_file():
        raise FileNotFoundError(f"Audio file not found: {audio_path}")

    api_key = os.environ["SMALLEST_API_KEY"]

    params = {
        "model": "pulse-pro",
        "language": "en",
        "word_timestamps": "true",
    }

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/octet-stream",
    }

    with audio_path.open("rb") as audio_file:
        response = requests.post(
            ENDPOINT,
            params=params,
            headers=headers,
            data=audio_file,
            timeout=120,
        )

    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    result = transcribe_audio("preprocessed_audio.wav")
    print(result.get("transcription", ""))

raise_for_status() deserves to stay in even the smallest example.

Without it, your code can accidentally treat an authentication failure, rate limit, or server error as if it were a successful transcription with missing data.

Word timestamps also become useful surprisingly quickly. They let you build searchable audio, synchronize captions, highlight matching sections of a recording, or connect transcript spans back to the original media.

Transcribe audio from a hosted URL

Sometimes the file does not live on the transcription worker.

It may already be stored in object storage behind a public or signed URL.

For URL-based pre-recorded transcription, use Pulse and send a JSON payload containing the URL.

The URL must be reachable by the transcription service. If you use private object storage, prefer a short-lived signed URL rather than making the object permanently public.

Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.

import os

import requests


ENDPOINT = "https://api.smallest.ai/waves/v1/stt/"


def transcribe_audio_url(audio_url: str) -> dict:
    """Transcribe hosted audio using the Pulse model."""
    api_key = os.environ["SMALLEST_API_KEY"]

    params = {
        "model": "pulse",
        "language": "en",
        "word_timestamps": "true",
    }

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "url": audio_url,
    }

    response = requests.post(
        ENDPOINT,
        params=params,
        headers=headers,
        json=payload,
        timeout=120,
    )

    response.raise_for_status()
    return response.json()

Notice that the model choice changed.

Pulse Pro accepts raw pre-recorded audio, while URL input belongs to the Pulse workflow. Treat model selection as part of the request contract rather than a cosmetic option.

Treat the response as structured data

Do not throw away everything except the transcript string.

For a response with word timestamps enabled, you may have timing and confidence information that can be valuable elsewhere in the product.

A defensive parser should also assume optional fields can be absent:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

def parse_transcript(response: dict) -> None:
    """Print transcript text and available word metadata."""
    full_text = response.get("transcription", "")
    print(f"Transcript: {full_text}")

    for word in response.get("words", []):
        text = word.get("word", "")
        start = word.get("start")
        end = word.get("end")
        confidence = word.get("confidence")

        start_text = (
            f"{start:.2f}s"
            if isinstance(start, (int, float))
            else "?"
        )

        end_text = (
            f"{end:.2f}s"
            if isinstance(end, (int, float))
            else "?"
        )

        confidence_text = (
            f"{confidence:.2f}"
            if isinstance(confidence, (int, float))
            else "n/a"
        )

        print(
            f"[{start_text} - {end_text}] "
            f"{text} "
            f"(confidence: {confidence_text})"
        )

    language = response.get("language", "unknown")
    print(f"Language: {language}")

Confidence values are better treated as signals than universal truth.

Avoid assuming that a single threshold such as 0.7 works for every model, language, microphone, or domain. If confidence will route transcripts to human review, calibrate that threshold against your own labeled audio.

Add speaker diarization when “who said it” matters

For meetings, interviews, podcasts, and customer calls, plain transcription may not be enough.

You also need to know which speaker produced each segment.

That is speaker diarization.

With the current Pulse pre-recorded API, diarization is enabled by using the Pulse model and passing:

model=pulse, language=en, and diarize=true.

You can combine diarization with word timestamps when you need both timing and speaker structure.

The response can contain speaker information at the word and utterance levels. A local formatter can then rebuild a readable conversation:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

def format_diarized_transcript(response: dict) -> str:
    """Convert diarized utterances into readable speaker turns."""
    lines = []

    for utterance in response.get("utterances", []):
        speaker = utterance.get("speaker", "unknown_speaker")
        text = utterance.get("text", "").strip()
        start = utterance.get("start")

        if not text:
            continue

        start_text = (
            f"{start:.1f}s"
            if isinstance(start, (int, float))
            else "?"
        )

        lines.append(f"[{start_text}] {speaker}: {text}")

    return "\n".join(lines)

Diarization introduces its own failure modes.

Overlapping speech is particularly difficult because two voices can occupy the same time interval. Recording separate channels upstream, when your telephony or conferencing stack makes that possible, can simplify later processing.

Also remember that diarization is not the same as identity recognition. A label such as speaker_0 tells you that the segment belongs to one detected speaker; it does not automatically tell you that the person is “Alice.”

If multi-speaker transcription is central to your application, the speaker diarization API guide goes deeper into those tradeoffs.

Handle long recordings deliberately

Long audio creates a different reliability problem.

A single large request gives you a large failure domain. If the request times out late in processing, you may have to repeat substantial work.

There are two useful approaches.

For long Pulse Pro transcription, an asynchronous webhook workflow can avoid holding one HTTP connection open for the entire job.

Application-level chunking is another option when you want smaller independent work units.

A practical chunking strategy is:

• split the recording into manageable segments
• include a small overlap between adjacent segments
• transcribe segments independently
• preserve each segment’s original time offset
• reconcile duplicate words created by the overlap

Here is the local splitting step:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

from pathlib import Path

from pydub import AudioSegment


def split_audio_with_overlap(
    input_path: str,
    output_dir: str,
    chunk_length_ms: int = 60_000,
    overlap_ms: int = 2_000,
) -> list[str]:
    """Split audio into overlapping WAV chunks."""
    audio = AudioSegment.from_file(input_path)

    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    chunk_paths = []
    start = 0
    chunk_index = 0

    while start < len(audio):
        end = min(start + chunk_length_ms, len(audio))
        chunk = audio[start:end]

        chunk_path = output / f"chunk_{chunk_index:04d}.wav"
        chunk.export(chunk_path, format="wav")
        chunk_paths.append(str(chunk_path))

        if end == len(audio):
            break

        start = max(0, end - overlap_ms)
        chunk_index += 1

    return chunk_paths

The overlap protects the boundary.

Without it, a word beginning near the end of one chunk and finishing at the beginning of the next can be truncated. A small overlap gives both requests enough surrounding audio to decode the boundary more reliably.

But overlapping chunks also create duplicate transcript content.

Do not merge them by simply concatenating strings.

Track each chunk’s original start time, offset the returned timestamps accordingly, and reconcile the overlapping region when building the final transcript.

For especially long files, compare chunking against asynchronous transcription before automatically deciding that one large synchronous HTTP request is the right architecture.

Add retry logic without retrying everything

Production networks fail.

You should expect:

• connection errors
• timeouts
• rate limits
• temporary upstream failures

Retries help, but only when they are selective.

Retrying an invalid request five times does not make it valid. Authentication errors and malformed payloads generally need intervention rather than exponential backoff.

A better pattern is to retry connection failures, timeouts, rate limiting, and transient server responses.

Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.

import os
from pathlib import Path

import requests
from tenacity import (
    retry,
    retry_if_exception,
    stop_after_attempt,
    wait_exponential,
)


ENDPOINT = "https://api.smallest.ai/waves/v1/stt/"
TRANSIENT_STATUS_CODES = {429, 500, 502, 503, 504}


def should_retry(error: BaseException) -> bool:
    if isinstance(error, (requests.Timeout, requests.ConnectionError)):
        return True

    if isinstance(error, requests.HTTPError):
        response = error.response
        return (
            response is not None
            and response.status_code in TRANSIENT_STATUS_CODES
        )

    return False


@retry(
    retry=retry_if_exception(should_retry),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
def transcribe_with_retries(file_path: str) -> dict:
    """Transcribe audio and retry only transient failures."""
    audio_path = Path(file_path)

    if not audio_path.is_file():
        raise FileNotFoundError(f"Audio file not found: {audio_path}")

    api_key = os.environ["SMALLEST_API_KEY"]

    with audio_path.open("rb") as audio_file:
        response = requests.post(
            ENDPOINT,
            params={
                "model": "pulse-pro",
                "language": "en",
                "word_timestamps": "true",
            },
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/octet-stream",
            },
            data=audio_file,
            timeout=120,
        )

    response.raise_for_status()
    return response.json()

In a larger system, retries should also be observable.

Record the request ID when the API provides one, track retry counts, distinguish permanent from transient failures, and make sure repeated jobs do not create duplicate downstream records.

Prevent duplicate transcription work

Reliability and cost control often point to the same design.

If the same audio file can be submitted more than once, calculate a deterministic content hash and use it as an idempotency or cache key in your own application.

For example:

(Before running the snippet, create a Smallest.ai API key in the dashboard and store it in the SMALLEST_API_KEY environment variable.)

import hashlib
from pathlib import Path


def sha256_file(file_path: str) -> str:
    digest = hashlib.sha256()

    with Path(file_path).open("rb") as file:
        for block in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(block)

    return digest.hexdigest()

Before creating a new transcription job, check whether that hash already has a completed result.

Other useful controls include checking recording duration before submission, setting application-level limits, storing completed responses durably, and separating transcription workers from the request path when jobs can take significant time.

Accuracy is an application-level measurement

Word Error Rate, or WER, is useful because it forces you to quantify transcription errors.

At 5% WER, a 500-word transcript corresponds to roughly 25 word-level errors.

Whether that is acceptable depends entirely on what happens next.

A rough meeting summary might tolerate mistakes that would be unacceptable in a workflow where transcript content automatically updates records, triggers transactions, or feeds a compliance process.

The important lesson is not to choose a universal “good” WER.

Test the system on your own audio.

Start with recordings from the actual environment:

• the microphones users really have
• the codecs used in production
• the accents and languages you expect
• real background noise
• overlapping speakers
• actual product names and domain terminology

A short representative sample can expose obvious problems during prototyping, but production evaluation should grow into a larger labeled dataset.

This is also why a clean demo clip is a weak benchmark. You are not deploying the benchmark. You are deploying your own acoustic environment.

From prototype to production

A dependable Python transcription pipeline usually comes back to a few engineering disciplines.

Normalize inconsistent input when necessary.

Treat the API response as structured data rather than one transcript string.

Choose the transcription model according to the actual input and features you need.

Keep credentials server-side.

Handle long recordings deliberately instead of assuming one synchronous request will always succeed.

Retry transient failures, not permanent ones.

Cache completed work when the same recording can be submitted more than once.

And above all, evaluate accuracy using the audio your application will actually receive.

The first successful transcription request proves that the API works.

Everything around that request determines whether your application works.

If you want to test the pipeline against your own recording, start building with the Smallest AI API, create an API key, and run the pre-recorded Python example with representative audio from your application.