DEV Community

Harpreet Singh Seehra
Harpreet Singh Seehra

Posted on

Build an AI Video Dubbing Pipeline with STT, LLM, and TTS on One Network

Imagine uploading a podcast clip in English and getting back the same conversation dubbed in Spanish — same speakers, same pacing, same tone, but in a different language. No stitching together Deepgram for transcription, OpenAI for translation, and ElevenLabs for synthesis. No juggling three API keys, three billing accounts, three SDKs. One pipeline, three API calls, one network.

This is the AI Video Dubbing Pipeline — a ~280-line Python app built on Telnyx that runs the full Speech-to-Text → AI Inference → Text-to-Speech chain. Upload any audio file, pick from 15 target languages, and download the dubbed mp3 a few seconds later. The pipeline labels speakers, translates the dialogue, and synthesizes each speaker with a matched voice from a KokoroTTS voice pool.

In this walkthrough, you'll build it from scratch. Clone the repo, add your API key, and dub your first clip in minutes.

What You'll Build

An HTTP API that dubs audio files into another language:

  • Upload audio — POST /dub with a multipart audio file and a target language
  • STT transcribes — Whisper-large-v3-turbo turns the audio into timestamped segments
  • LLM labels + translates — a single chat-completions call assigns speaker labels (SPEAKER_0, SPEAKER_1, ...) and translates each segment to the target language
  • TTS synthesizes — each translated segment is rendered with a voice assigned per speaker from a KokoroTTS voice pool
  • Download the dub — GET /dub/<job_id>/audio streams the concatenated mp3
  • Side-by-side transcript — GET /dub/<job_id>/transcript returns the original and translated text per segment

The whole pipeline runs asynchronously: POST /dub returns 202 with a job_id, and the client polls GET /dub/<job_id> until status == "complete". No webhooks, no phone numbers, no ngrok — this is a pure HTTP API.

Why This Is Interesting

Most dubbing pipelines are stitched together from three separate vendors. That means three API keys, three billing relationships, three sets of rate limits, and three network round trips — the STT transcription, the LLM translation, and the TTS synthesis each cross the internet to a different provider. Latency adds up. Authentication adds up. So does the operational overhead of keeping three SDKs current.

This pipeline runs all three primitives on Telnyx. The STT, chat, and TTS endpoints share one API key, one billing account, and one network path. The STT and chat endpoints are OpenAI-compatible — drop-in for the OpenAI Python or JS SDK with the base URL changed to https://api.telnyx.com/v2. The TTS endpoint takes a Telnyx voice id (Telnyx.KokoroTTS.am_onyx, af_nova, etc.) and renders in the target language you pass in.

The interesting technical move: Telnyx STT does not perform speaker diarization. The transcription API returns segments with start, end, and text — no speaker field. Most pipelines handle this by adding a fourth vendor (a diarization service) or by accepting that all segments come back as one speaker. This app takes a different path: it sends all the transcribed segments to the LLM in a single chat call and asks the model to assign speaker labels AND translate in the same response. One model, two jobs, no extra vendor.

Prerequisites

That's it. No phone number, no Call Control Application, no ngrok, no webhook configuration. This is a pure HTTP API — you upload audio and poll for results.

The Architecture 

Upload audio (POST /dub, multipart)
        │
        ▼
  ┌──────────────────┐
  │ STT (Whisper)     │ ── POST /v2/ai/audio/transcriptions
  │                   │    returns segments: [{start, end, text}, ...]
  └────────┬──────────┘
           │ segments
           ▼
  ┌──────────────────┐
  │ AI Inference      │ ── POST /v2/ai/chat/completions
  │ (single call)     │    returns JSON: [{id, speaker, translated}, ...]
  │ label + translate │    (LLM assigns SPEAKER_0/1/... AND translates)
  └────────┬──────────┘
           │ translated segments with speaker labels
           ▼
  ┌──────────────────┐
  │ TTS (Kokoro)      │ ── POST /v2/text-to-speech/speech  (per segment)
  │ per-segment render│    voice assigned per speaker from a 5-voice pool
  └────────┬──────────┘
           │
           ├──► GET /dub/<job_id>         (status + transcript)
           ├──► GET /dub/<job_id>/audio    (download mp3)
           └──► GET /dub/<job_id>/transcript (side-by-side text)
Enter fullscreen mode Exit fullscreen mode

The pipeline runs in a background thread. POST /dub validates the request, stores the job, kicks off the thread, and returns 202 immediately. The thread runs STT → Inference → TTS, updates the job status at each step, and stores the concatenated audio on the job when done.

Step 1: Clone and Configure 

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-video-dubbing-pipeline-python
cp .env.example .env
pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode

Edit .env with your Telnyx API key:

TELNYX_API_KEY=your_telnyx_api_key_here
AI_MODEL=moonshotai/Kimi-K2.6
STT_MODEL=openai/whisper-large-v3-turbo
HOST=127.0.0.1
PORT=5000
Enter fullscreen mode Exit fullscreen mode

The defaults are solid. whisper-large-v3-turbo is multilingual — it transcribes any of the 16 supported source languages. If you only ever dub English source audio, you can switch to distil-whisper/distil-large-v2 for a faster, cheaper, English-only pass.

Step 2: Understand the Code

Everything lives in app.py (~280 lines). Here's what each piece does.

The Voice Pool

The app cycles through five Telnyx KokoroTTS voices, one per speaker:

VOICE_MAP = {
    "male_low": "Telnyx.KokoroTTS.am_onyx",
    "male_mid": "Telnyx.KokoroTTS.am_echo",
    "female_mid": "Telnyx.KokoroTTS.af_nova",
    "female_high": "Telnyx.KokoroTTS.af_heart",
    "neutral": "Telnyx.KokoroTTS.af_alloy",
}
Enter fullscreen mode Exit fullscreen mode

When the pipeline hits a new speaker label, it assigns the next voice from the pool. A two-speaker conversation gets am_onyx and am_echo. A five-speaker panel gets all five. A sixth speaker wraps back to am_onyx. The voice ids are real — you can list every available voice with GET /v2/text-to-speech/voices on the Telnyx API.

The STT Helper 

def transcribe_audio(audio_bytes, source_language="en"):
    bcp47 = STT_LANGUAGE_MAP.get(source_language, "en-US")
    resp = requests.post(f"{API}/ai/audio/transcriptions", headers={
        "Authorization": f"Bearer {TELNYX_API_KEY}"
    }, files={
        "file": ("audio.mp3", audio_bytes, "audio/mpeg")
    }, data={
        "model": STT_MODEL,
        "language": bcp47,
        "response_format": "verbose_json",
        "timestamp_granularities[]": "segment"
    }, timeout=120)
    resp.raise_for_status()
    return resp.json()
Enter fullscreen mode Exit fullscreen mode

OpenAI-compatible transcription. The verbose_json response format with timestamp_granularities[]=segment returns segments with start, end, and text — exactly what the next stage needs. Note: no diarize parameter. Telnyx STT doesn't diarize. The LLM does that downstream.

The LLM Labels Speakers AND Translates in One Call

This is the move that replaces a fourth vendor:

def label_and_translate_segments(segments, source_lang, target_lang_name):
    segment_texts = [
        {"id": i, "text": s.get("text", "").strip(),
         "start": s.get("start", 0), "end": s.get("end", 0)}
        for i, s in enumerate(segments)
    ]
    prompt = (
        f"You are a dubbing assistant. Below are {len(segment_texts)} transcribed audio segments "
        f"from a conversation in {source_lang}. For EACH segment:\n"
        f"1. Assign a speaker label (SPEAKER_0, SPEAKER_1, etc.) based on conversational context.\n"
        f"2. Translate the text to {target_lang_name}, preserving tone and meaning.\n\n"
        f"Return ONLY a JSON array (no markdown fences, no explanation) where each element is:\n"
        f'{{"id": <int>, "speaker": "SPEAKER_N", "translated": "<text in {target_lang_name}>"}}\n\n'
        f"Segments:\n{json.dumps(segment_texts, ensure_ascii=False)}"
    )
    raw = inference([
        {"role": "system", "content": "You are a translation and speaker-diarization engine. You always return valid JSON arrays, never prose."},
        {"role": "user", "content": prompt}
    ], max_tokens=4000)
    # ... parse JSON, fall back to per-segment translation if the model returns non-JSON ...
Enter fullscreen mode Exit fullscreen mode

One chat call. The LLM reads every segment, figures out who's speaking based on conversational context (turn-taking, topic shifts, register), assigns SPEAKER_0, SPEAKER_1, ..., and translates each segment to the target language. Returns a JSON array. If the model wraps the JSON in markdown fences or returns prose, the parser strips the fences; if JSON parsing still fails, the function falls back to translating each segment individually and labeling everything as SPEAKER_0.

The TTS Helper 

def tts_generate(text, voice, language):
    resp = requests.post(f"{API}/text-to-speech/speech", headers=HEADERS, json={
        "text": text,
        "voice": voice,
        "language": language,
        "output_type": "binary_output",
        "provider": "telnyx"
    }, timeout=30)
    resp.raise_for_status()
    return resp.content
Enter fullscreen mode Exit fullscreen mode

One TTS call per segment. output_type: "binary_output" returns the raw mp3 bytes. provider: "telnyx" selects the Telnyx KokoroTTS engine. The language parameter controls pronunciation for the target language — you pass es for Spanish, ja for Japanese, etc.

The Async Pipeline 

@app.route("/dub", methods=["POST"])
def start_dubbing():
    # ... validate, build job record ...
    thread = threading.Thread(
        target=_run_pipeline,
        args=(job_id, audio_bytes, source_lang, target_lang),
        daemon=True
    )
    thread.start()
    return jsonify({"job_id": job_id, "status": "queued", ...}), 202
Enter fullscreen mode Exit fullscreen mode

The handler returns immediately. The background thread runs STT → Inference → TTS, updates the job status at each step (transcribing → translating → synthesizing → complete), and stores the concatenated audio bytes on the job. On any exception, the job moves to failed with the error message.

The Audio Download Endpoint 

@app.route("/dub/<job_id>/audio", methods=["GET"])
def get_audio(job_id):
    # ... fetch job, check status == "complete" ...
    return Response(
        io.BytesIO(audio),
        mimetype="audio/mpeg",
        headers={"Content-Disposition": f"attachment; filename={job_id}.mp3"}
    )
Enter fullscreen mode Exit fullscreen mode

The dubbed audio is the concatenated mp3 chunks from each TTS call. Byte-level concatenation is fine for a demo — for production you'd mux with ffmpeg's concat demuxer so segment boundaries are sample-accurate.

Step 3: Run the App 

python app.py
Enter fullscreen mode Exit fullscreen mode

Server starts on http://127.0.0.1:5000. No ngrok, no webhook configuration — this is a pure HTTP API.

Step 4: Dub a Clip

Health check first:

curl http://localhost:5000/health
# {"status":"ok","total_jobs":0,"active":0,"supported_languages":15,"version":"1.0.0"}
Enter fullscreen mode Exit fullscreen mode

List the supported target languages:

curl http://localhost:5000/languages
# {"languages":{"es":"Spanish","fr":"French","de":"German","pt":"Portuguese",
#  "it":"Italian","ja":"Japanese","ko":"Korean","zh":"Chinese",
#  "ar":"Arabic","hi":"Hindi","ru":"Russian","nl":"Dutch",
#  "sv":"Swedish","pl":"Polish","tr":"Turkish"}}
Enter fullscreen mode Exit fullscreen mode

Start a dubbing job — upload a short mp3 and pick a target language:

curl -X POST http://localhost:5000/dub \
  -F audio=@episode.mp3 \
  -F target_language=es \
  -F source_language=en
# {"job_id":"dub-a1b2c3d4","status":"queued","source_language":"en",
#  "target_language":"es (Spanish)",
#  "message":"Pipeline started. Poll GET /dub/<job_id> for status."}
Enter fullscreen mode Exit fullscreen mode

Poll for status:

curl http://localhost:5000/dub/dub-a1b2c3d4 | python3 -m json.tool
# status: queued → transcribing → translating → synthesizing → complete
Enter fullscreen mode Exit fullscreen mode

Once status == "complete", download the dubbed mp3:

curl http://localhost:5000/dub/dub-a1b2c3d4/audio --output dubbed.mp3
Enter fullscreen mode Exit fullscreen mode

And get the side-by-side transcript:

curl http://localhost:5000/dub/dub-a1b2c3d4/transcript | python3 -m json.tool
# {
#   "job_id": "dub-a1b2c3d4",
#   "source": "en",
#   "target": "es",
#   "segments": [
#     {
#       "speaker": "SPEAKER_0",
#       "original": "Hello and welcome to the show.",
#       "translated": "Hola y bienvenidos al programa.",
#       "timestamp": "0.0s - 2.5s"
#     }
#   ]
# }
Enter fullscreen mode Exit fullscreen mode

Customizing the Pipeline

The VOICE_MAP and SUPPORTED_LANGUAGES dicts are the control surfaces. Small changes produce very different dubs.

Use different voices per language — pick voices that match the target language's accent:

LANGUAGE_VOICE_OVERRIDE = {
    "es": {"male": "Telnyx.KokoroTTS.em_alex", "female": "Telnyx.KokoroTTS.ef_dora"},
    "fr": {"male": "Telnyx.KokoroTTS.bm_daniel", "female": "Telnyx.KokoroTTS.ff_siwis"},
    "pt": {"male": "Telnyx.KokoroTTS.pm_alex", "female": "Telnyx.KokoroTTS.pf_dora"},
}
Enter fullscreen mode Exit fullscreen mode

Add more languages — extend SUPPORTED_LANGUAGES and STT_LANGUAGE_MAP:

SUPPORTED_LANGUAGES["vi"] = "Vietnamese"
STT_LANGUAGE_MAP["vi"] = "vi-VN"
Enter fullscreen mode Exit fullscreen mode

Use a stronger diarization prompt — for panel discussions with 5+ speakers, give the LLM more context:

prompt = (
    f"You are a dubbing assistant. Below are {len(segment_texts)} transcribed audio segments "
    f"from a panel discussion in {source_lang}. There are likely 3-5 speakers. "
    f"For EACH segment: assign a speaker label and translate to {target_lang_name}. ..."
)
Enter fullscreen mode Exit fullscreen mode

Going to Production

This example uses in-memory storage and byte-level MP3 concatenation for simplicity. For production:

  • Database — replace the in-memory dict with PostgreSQL or Redis so jobs survive restarts
  • Audio muxing — replace byte concat with ffmpeg's concat demuxer for sample-accurate boundaries
  • Object storage — write finished audio to Telnyx Storage and return a signed URL instead of streaming from memory
  • Authentication — add API key validation on your endpoints
  • Error recovery — retry transient STT/TTS failures with exponential backoff
  • Prompt engineering — tune the speaker-labeling prompt for your domain (2-person interview vs. podcast with N speakers)
  • Rate limiting — protect /dub from abuse
  • WebSockets — for real-time dubbing, swap the polling model for the Telnyx streaming STT and TTS WebSocket endpoints

Frequently Asked Questions

Can I use a different LLM for the labeling + translation call?
Yes. Telnyx AI Inference is OpenAI-compatible. Set AI_MODEL in .env to any model available on the platform — moonshotai/Kimi-K2.6 is the default, but meta-llama/Llama-3.3-70B-Instruct or openai/gpt-4o work too. The JSON-output prompt is model-agnostic.

Why does the LLM do speaker labeling instead of the STT API?
Telnyx STT (like OpenAI's Whisper transcription API) does not return speaker labels. Rather than adding a separate diarization vendor, the app sends all segments to the LLM in one chat call and asks it to assign speaker labels based on conversational context — turn-taking, topic shifts, register. The LLM is already in the pipeline for translation, so the marginal cost is a slightly longer prompt.

How much does a dub cost?
A 1-minute clip with ~120 words of dialogue costs roughly: STT ~$0.006 (Whisper per minute), Inference ~$0.001 (one short chat call), TTS ~$0.015 (Kokoro per 1K characters). Total: ~2 cents per minute of dubbed audio. The 15-language voice pool is the same price.

What audio formats can I upload?
The Telnyx STT endpoint accepts mp3, wav, m4a, ogg, and webm. The example uploads as audio/mpeg (mp3) — change the files= tuple's mime type if you upload a different format.

Can I dub a 30-minute file?
Yes. The async pipeline means the HTTP request returns immediately. A 30-minute file takes ~30-60 seconds end-to-end (STT is the slow part). The in-memory dict will hold the full audio — for files longer than an hour, swap to object storage.

What happens if the LLM returns non-JSON?
The parser strips markdown fences if present. If JSON parsing still fails, the function falls back to translating each segment individually with a per-segment chat call and labeling everything as SPEAKER_0. The job completes with a single-speaker dub — degraded but not broken.

Can I stream the dubbed audio back instead of downloading?
Yes — Telnyx has streaming STT (GET /v2/speech-to-text/transcription, WebSocket) and streaming TTS (GET /v2/text-to-speech/speech, WebSocket) endpoints. For real-time dubbing, swap the batch endpoints for the streaming ones and pipe STT segments to the LLM as they arrive.

Resources

Top comments (0)