6. Audio → Text → LLM Pipelines

Description

Complete pipelines combine transcription (STT), LLM processing, and optionally speech synthesis (TTS). In this capsule you'll implement 4 production pipelines: meeting summaries, Q&A over audio, audio generation from processed text, and the complete loop audio → text → LLM → audio. Each pipeline includes cost tracking and error handling.

Why it matters: STT and TTS on their own are components. Pipelines are products. A company doesn't need "transcription" — it needs "automatic meeting summaries with action items." Pipelines are what turn technical capabilities into business value.

Connection with the project: The Audio Pipeline in capsule 08 is an integrated version of the pipelines you build here. This capsule gives you the building blocks; capsule 08 assembles them into a complete system with configuration, extensions and testing.

Pipeline Architecture

Overview

Pipeline 1: Audio → Text → Summary
  meeting.mp3 → Whisper → transcript → GPT-4o-mini → summary + action items

Pipeline 2: Audio → Text → Q&A
  podcast.mp3 → Whisper → transcript → GPT-4o-mini + question → answer

Pipeline 3: Text → LLM → Audio
  article.txt → GPT-4o-mini → summary → OpenAI TTS → summary.mp3

Pipeline 4: Audio → Text → LLM → Audio (complete loop)
  question.mp3 → Whisper → text → GPT-4o-mini → answer → TTS → response.mp3

Shared components

from openai import OpenAI
from pydub import AudioSegment
from pathlib import Path
from dataclasses import dataclass, field
import tempfile
import time

client = OpenAI()

@dataclass
class PipelineResult:
    success: bool
    data: dict = field(default_factory=dict)
    costs: dict = field(default_factory=dict)
    timing: dict = field(default_factory=dict)
    error: str = None

def track_cost(costs: dict, service: str, amount: float):
    costs[service] = costs.get(service, 0) + amount
    costs["total"] = sum(v for k, v in costs.items() if k != "total")

Pipeline 1: Audio → Transcription → Summary

The most common pipeline: transcribe a meeting and generate a structured summary with key points and action items.

def pipeline_audio_to_summary(
    audio_path: str,
    language: str = "es",
    summary_format: str = "bullets"
) -> PipelineResult:
    costs = {}
    timing = {}

    t0 = time.time()
    audio = AudioSegment.from_file(audio_path)
    duration_min = len(audio) / 1000 / 60

    t1 = time.time()
    with open(audio_path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language=language
        ).text
    timing["transcription_s"] = round(time.time() - t1, 2)
    track_cost(costs, "whisper", duration_min * 0.006)

    prompts = {
        "bullets": (
            "Summarize the following text in 5-7 key points. "
            "Use bullet points. Be concise and specific.\n\n"
        ),
        "executive": (
            "Generate an executive summary of at most 3 paragraphs. "
            "Include: context, decisions made, and next steps.\n\n"
        ),
        "action_items": (
            "Extract ALL the action items (pending tasks) from the following text. "
            "Format: '- [ ] [Owner]: [Task] (Date if mentioned)'\n\n"
        ),
    }
    prompt = prompts.get(summary_format, prompts["bullets"])

    t2 = time.time()
    input_text = transcript[:12000]
    summary = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt + input_text}],
        max_tokens=800
    ).choices[0].message.content
    timing["llm_s"] = round(time.time() - t2, 2)
    track_cost(costs, "gpt-4o-mini", len(input_text) * 0.00000015 + 800 * 0.0000006)

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={
            "transcript": transcript,
            "summary": summary,
            "format": summary_format,
            "audio_duration_min": round(duration_min, 2),
            "transcript_words": len(transcript.split())
        },
        costs=costs,
        timing=timing
    )

Usage

result = pipeline_audio_to_summary("meeting.mp3", summary_format="action_items")

if result.success:
    print(f"Summary:\n{result.data['summary']}")
    print(f"Total cost: ${result.costs['total']:.4f}")
    print(f"Total time: {result.timing['total_s']}s")

Pipeline 2: Audio → Transcription → Q&A

Transcribe audio and let you ask questions about the content. Ideal for finding specific information in long meetings or interviews.

def pipeline_audio_to_qa(
    audio_path: str,
    questions: list[str],
    language: str = "es"
) -> PipelineResult:
    costs = {}
    timing = {}

    t0 = time.time()
    audio = AudioSegment.from_file(audio_path)
    duration_min = len(audio) / 1000 / 60

    t1 = time.time()
    with open(audio_path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language=language
        ).text
    timing["transcription_s"] = round(time.time() - t1, 2)
    track_cost(costs, "whisper", duration_min * 0.006)

    context = transcript[:10000]
    answers = []

    t2 = time.time()
    for question in questions:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Answer questions based ONLY on the provided context. "
                        "If the answer isn't in the context, say 'Not mentioned in the audio'."
                    )
                },
                {
                    "role": "user",
                    "content": f"Context (audio transcription):\n{context}\n\nQuestion: {question}"
                }
            ],
            max_tokens=300
        )
        answers.append({
            "question": question,
            "answer": response.choices[0].message.content
        })
        input_tokens = len(context.split()) + len(question.split())
        track_cost(costs, "gpt-4o-mini", input_tokens * 0.00000015 + 300 * 0.0000006)

    timing["qa_s"] = round(time.time() - t2, 2)
    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={
            "transcript": transcript,
            "qa": answers,
            "audio_duration_min": round(duration_min, 2)
        },
        costs=costs,
        timing=timing
    )

Usage

result = pipeline_audio_to_qa(
    "interview.mp3",
    questions=[
        "What was the main decision?",
        "Who is responsible for the follow-up?",
        "What is the deadline?"
    ]
)

for qa in result.data["qa"]:
    print(f"Q: {qa['question']}")
    print(f"A: {qa['answer']}\n")

Pipeline 3: Text → LLM → Audio

Takes written text, processes it with an LLM (summarize, rephrase, translate), and generates audio. Ideal for converting articles into podcasts or creating audio versions of documents.

def pipeline_text_to_audio(
    text: str,
    task: str = "summarize",
    voice: str = "nova",
    tts_model: str = "tts-1",
    output_path: str = "output_audio.mp3"
) -> PipelineResult:
    costs = {}
    timing = {}

    task_prompts = {
        "summarize": (
            "Summarize the following text in a concise paragraph, "
            "optimized to be heard as audio (use natural language, "
            "avoid bullet-point lists):\n\n"
        ),
        "simplify": (
            "Rewrite the following text in simple, clear language, "
            "as if you were explaining it to a general audience. "
            "Optimize for audio (conversational language):\n\n"
        ),
        "translate_en": (
            "Translate the following text to English. "
            "Keep a natural, conversational tone:\n\n"
        ),
    }

    prompt = task_prompts.get(task, task_prompts["summarize"])

    t0 = time.time()

    t1 = time.time()
    processed = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt + text[:10000]}],
        max_tokens=1000
    ).choices[0].message.content
    timing["llm_s"] = round(time.time() - t1, 2)
    track_cost(costs, "gpt-4o-mini", len(text[:10000].split()) * 0.00000015 + 1000 * 0.0000006)

    t2 = time.time()
    if len(processed) > 4000:
        from pydub import AudioSegment
        chunks = [processed[i:i+4000] for i in range(0, len(processed), 4000)]
        segments = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            for i, chunk in enumerate(chunks):
                chunk_path = str(Path(tmp_dir) / f"chunk_{i:03d}.mp3")
                response = client.audio.speech.create(
                    model=tts_model, voice=voice, input=chunk
                )
                response.stream_to_file(chunk_path)
                segments.append(AudioSegment.from_mp3(chunk_path))

            combined = segments[0]
            for seg in segments[1:]:
                combined += seg
            combined.export(output_path, format="mp3")
    else:
        response = client.audio.speech.create(
            model=tts_model, voice=voice, input=processed
        )
        response.stream_to_file(output_path)

    timing["tts_s"] = round(time.time() - t2, 2)
    cost_per_char = 0.000015 if tts_model == "tts-1" else 0.00003
    track_cost(costs, "tts", len(processed) * cost_per_char)

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={
            "processed_text": processed,
            "audio_path": output_path,
            "task": task,
            "voice": voice,
            "characters": len(processed)
        },
        costs=costs,
        timing=timing
    )

Usage

article = """
Multimodal artificial intelligence represents a paradigm shift
in how AI systems process information. Unlike traditional models
limited to text, multimodal models can process
images, audio and video simultaneously...
"""

result = pipeline_text_to_audio(article, task="summarize", voice="echo")
print(f"Audio generated: {result.data['audio_path']}")
print(f"Cost: ${result.costs['total']:.4f}")

Pipeline 4: Audio → Text → LLM → Audio (Complete Loop)

The most powerful pipeline: it receives a question in audio, transcribes it, processes it with the LLM, and generates the answer in audio. It's the core of a voice assistant.

def pipeline_audio_full_loop(
    audio_path: str,
    system_prompt: str = "You are a helpful assistant. Answer concisely and clearly.",
    voice: str = "nova",
    language: str = "es",
    output_path: str = "response.mp3"
) -> PipelineResult:
    costs = {}
    timing = {}

    t0 = time.time()
    audio = AudioSegment.from_file(audio_path)
    duration_min = len(audio) / 1000 / 60

    t1 = time.time()
    with open(audio_path, "rb") as f:
        user_text = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language=language
        ).text
    timing["stt_s"] = round(time.time() - t1, 2)
    track_cost(costs, "whisper", duration_min * 0.006)

    t2 = time.time()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_text}
        ],
        max_tokens=500
    )
    assistant_text = response.choices[0].message.content
    timing["llm_s"] = round(time.time() - t2, 2)
    track_cost(costs, "gpt-4o-mini", len(user_text.split()) * 0.00000015 + 500 * 0.0000006)

    t3 = time.time()
    tts_response = client.audio.speech.create(
        model="tts-1",
        voice=voice,
        input=assistant_text
    )
    tts_response.stream_to_file(output_path)
    timing["tts_s"] = round(time.time() - t3, 2)
    track_cost(costs, "tts", len(assistant_text) * 0.000015)

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={
            "user_input": user_text,
            "assistant_response": assistant_text,
            "audio_output": output_path,
            "audio_input_duration_min": round(duration_min, 2)
        },
        costs=costs,
        timing=timing
    )

Usage

result = pipeline_audio_full_loop(
    "question.mp3",
    system_prompt="You are an AI expert. Answer in Spanish, clearly and concisely.",
    voice="echo"
)

print(f"User said: {result.data['user_input']}")
print(f"Assistant replied: {result.data['assistant_response']}")
print(f"Audio response: {result.data['audio_output']}")

Pipeline with Long Transcription + Map-Reduce

For very long audio (>1 hour), use a map-reduce pattern: transcribe by chunks, summarize each chunk, and then generate a final summary.

def pipeline_long_audio_summary(
    audio_path: str,
    language: str = "es",
    chunk_duration_ms: int = 10 * 60 * 1000,
    output_audio_path: str = None
) -> PipelineResult:
    costs = {}
    timing = {}
    t0 = time.time()

    audio = AudioSegment.from_file(audio_path)
    total_duration_min = len(audio) / 1000 / 60
    chunks = [
        audio[i:i + chunk_duration_ms]
        for i in range(0, len(audio), chunk_duration_ms)
    ]

    t1 = time.time()
    transcripts = []
    with tempfile.TemporaryDirectory() as tmp_dir:
        for i, chunk in enumerate(chunks):
            chunk_path = str(Path(tmp_dir) / f"chunk_{i:03d}.mp3")
            chunk.export(chunk_path, format="mp3", bitrate="128k")
            with open(chunk_path, "rb") as f:
                text = client.audio.transcriptions.create(
                    model="whisper-1", file=f, language=language
                ).text
            transcripts.append(text)

    timing["transcription_s"] = round(time.time() - t1, 2)
    track_cost(costs, "whisper", total_duration_min * 0.006)

    t2 = time.time()
    chunk_summaries = []
    for i, transcript in enumerate(transcripts):
        summary = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"Summarize this segment ({i+1}/{len(transcripts)}) "
                    f"in 3-5 key sentences:\n\n{transcript[:8000]}"
                )
            }],
            max_tokens=300
        ).choices[0].message.content
        chunk_summaries.append(summary)

    combined_summaries = "\n\n".join(
        f"Segment {i+1}:\n{s}" for i, s in enumerate(chunk_summaries)
    )

    final_summary = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                "Generate a complete executive summary from these partial summaries. "
                "Include: key points, decisions, and action items.\n\n"
                + combined_summaries
            )
        }],
        max_tokens=800
    ).choices[0].message.content

    timing["llm_s"] = round(time.time() - t2, 2)

    result_data = {
        "full_transcript": " ".join(transcripts),
        "chunk_summaries": chunk_summaries,
        "final_summary": final_summary,
        "chunks_processed": len(chunks),
        "total_duration_min": round(total_duration_min, 2)
    }

    if output_audio_path:
        t3 = time.time()
        tts_response = client.audio.speech.create(
            model="tts-1", voice="nova", input=final_summary[:4000]
        )
        tts_response.stream_to_file(output_audio_path)
        timing["tts_s"] = round(time.time() - t3, 2)
        track_cost(costs, "tts", len(final_summary[:4000]) * 0.000015)
        result_data["audio_summary_path"] = output_audio_path

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data=result_data,
        costs=costs,
        timing=timing
    )

Cost Tracking

Report function

def generate_cost_report(results: list[PipelineResult]) -> dict:
    total_costs = {}
    total_time = 0

    for result in results:
        for service, cost in result.costs.items():
            if service != "total":
                total_costs[service] = total_costs.get(service, 0) + cost
        total_time += result.timing.get("total_s", 0)

    total = sum(total_costs.values())

    return {
        "breakdown": {k: round(v, 6) for k, v in total_costs.items()},
        "total_usd": round(total, 4),
        "total_time_s": round(total_time, 2),
        "runs": len(results),
        "avg_cost_per_run": round(total / len(results), 6) if results else 0
    }

Cost reference table per pipeline

Troubleshooting

Problem 1: Truncated transcript in the pipeline

Symptom: The summary only covers part of the audio.

Solution: Use pipeline_long_audio_summary() with map-reduce for long audio. The LLM has a context limit (~12K tokens for gpt-4o-mini).

Problem 2: Generic or vague summary

Symptom: The LLM generates summaries that don't capture the specific points.

Solution:

specific_prompt = (
    "Summarize this text from a team meeting. "
    "Include NAMES of people and DATES mentioned. "
    "List concrete decisions, not generalities.\n\n"
)

Problem 3: TTS fails with long text

Symptom: Error when generating audio from long summaries.

Solution: Truncate or split the text before TTS:

tts_text = summary[:4000]

Problem 4: Unexpectedly high cost

Symptom: The pipeline costs more than expected.

Solution: Check costs with generate_cost_report(). The common culprits:

• Very long audio → high Whisper cost
• Multiple LLM calls → token accumulation
• TTS with tts-1-hd → double the cost vs tts-1

Exercises

Exercise 1: Structured meeting-notes pipeline

Create a pipeline that transcribes audio and generates a structured document with: attendees (if mentioned), topics discussed, decisions, action items, and next meeting.

def pipeline_meeting_notes(audio_path: str) -> PipelineResult:
    costs = {}
    timing = {}
    t0 = time.time()

    with open(audio_path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1", file=f, language="es"
        ).text

    audio = AudioSegment.from_file(audio_path)
    track_cost(costs, "whisper", (len(audio) / 1000 / 60) * 0.006)

    structured_prompt = """Analyze this meeting transcription and generate a structured document with EXACTLY these sections:

## Attendees
(List the people mentioned by name. If no names are mentioned, write "Not identified")

## Topics Discussed
(Numbered list of the main topics)

## Decisions Made
(List of concrete decisions. If there are no clear decisions, write "No explicit decisions")

## Action Items
(Format: - [ ] [Owner if known]: [Task] [Date if mentioned])

## Next Meeting
(Date/time if mentioned, otherwise "Not specified")

Transcription:
"""

    notes = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": structured_prompt + transcript[:10000]}],
        max_tokens=1000
    ).choices[0].message.content

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={"transcript": transcript, "meeting_notes": notes},
        costs=costs,
        timing=timing
    )

Exercise 2: Multilingual pipeline

Create a pipeline that transcribes audio in any language, detects the language, translates it to Spanish if necessary, and generates a summary in Spanish.

def pipeline_multilingual_summary(
    audio_path: str,
    target_language: str = "es"
) -> PipelineResult:
    costs = {}
    timing = {}
    t0 = time.time()

    with open(audio_path, "rb") as f:
        verbose = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            response_format="verbose_json"
        )

    detected_lang = verbose.language
    transcript = verbose.text

    audio = AudioSegment.from_file(audio_path)
    track_cost(costs, "whisper", (len(audio) / 1000 / 60) * 0.006)

    needs_translation = detected_lang != target_language

    if needs_translation:
        translated = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"Translate the following text from {detected_lang} "
                    f"to {target_language}. Keep the original meaning:\n\n"
                    + transcript[:10000]
                )
            }],
            max_tokens=2000
        ).choices[0].message.content
        text_for_summary = translated
    else:
        text_for_summary = transcript

    summary = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Summarize in 5 key points (in {target_language}):\n\n{text_for_summary[:10000]}"
        }],
        max_tokens=500
    ).choices[0].message.content

    timing["total_s"] = round(time.time() - t0, 2)

    return PipelineResult(
        success=True,
        data={
            "original_language": detected_lang,
            "transcript": transcript,
            "translated": needs_translation,
            "text_for_summary": text_for_summary if needs_translation else None,
            "summary": summary,
        },
        costs=costs,
        timing=timing
    )

Exercise 3: Pipeline with transcription cache

Create a wrapper that saves transcriptions to disk to avoid re-transcribing the same file. Use the file's hash as the key.

import hashlib
import json

CACHE_DIR = Path("transcript_cache")

def get_file_hash(file_path: str) -> str:
    h = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()

def cached_transcribe(
    audio_path: str,
    language: str = "es"
) -> dict:
    CACHE_DIR.mkdir(exist_ok=True)
    file_hash = get_file_hash(audio_path)
    cache_file = CACHE_DIR / f"{file_hash}.json"

    if cache_file.exists():
        cached = json.loads(cache_file.read_text())
        cached["from_cache"] = True
        return cached

    with open(audio_path, "rb") as f:
        response = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language=language,
            response_format="verbose_json"
        )

    result = {
        "text": response.text,
        "language": response.language,
        "duration": response.duration,
        "file_hash": file_hash,
        "source_file": Path(audio_path).name,
        "from_cache": False
    }

    cache_file.write_text(json.dumps(result, ensure_ascii=False, indent=2))
    return result

Summary

• Pipeline 1 (Summary): Audio → Whisper → transcript → LLM → summary/action items.
• Pipeline 2 (Q&A): Audio → Whisper → transcript → LLM + questions → contextual answers.
• Pipeline 3 (Text to Audio): Text → LLM (processes/summarizes/translates) → TTS → audio.
• Pipeline 4 (Complete loop): Audio → Whisper → LLM → TTS → audio (voice assistant).
• Map-reduce for long audio: transcribe by chunks → summarize each chunk → final summary.
• Cost tracking built into each pipeline with PipelineResult.
• The typical cost of a complete pipeline for 30 min of audio is ~$0.20 USD.

Additional Resources

1. Whisper API — Transcription
2. OpenAI TTS — Speech synthesis
3. OpenAI Chat API — LLM processing
4. pydub — Audio manipulation in Python