6. Pipelines Audio → Texto → LLM

Descripción

Los pipelines completos combinan transcripción (STT), procesamiento con LLM, y opcionalmente síntesis de voz (TTS). En esta cápsula implementarás 4 pipelines de producción: resumen de reuniones, Q&A sobre audio, generación de audio desde texto procesado, y el loop completo audio → texto → LLM → audio. Cada pipeline incluye tracking de costos y manejo de errores.

Por qué importa: STT y TTS por separado son componentes. Los pipelines son productos. Una empresa no necesita "transcripción" — necesita "resúmenes automáticos de reuniones con action items". Los pipelines son lo que convierte capacidades técnicas en valor de negocio.

Conexión con el proyecto: El Audio Pipeline de la cápsula 08 es una versión integrada de los pipelines que construyes aquí. Esta cápsula te da los bloques; la cápsula 08 los ensambla en un sistema completo con configuración, extensiones y testing.

Arquitectura de Pipelines

Visión general

Pipeline 1: Audio → Texto → Resumen
  reunion.mp3 → Whisper → transcript → GPT-4o-mini → resumen + action items

Pipeline 2: Audio → Texto → Q&A
  podcast.mp3 → Whisper → transcript → GPT-4o-mini + pregunta → respuesta

Pipeline 3: Texto → LLM → Audio
  articulo.txt → GPT-4o-mini → resumen → OpenAI TTS → resumen.mp3

Pipeline 4: Audio → Texto → LLM → Audio (loop completo)
  pregunta.mp3 → Whisper → texto → GPT-4o-mini → respuesta → TTS → respuesta.mp3

Componentes compartidos

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 → Transcripción → Resumen

El pipeline más común: transcribe una reunión y genera un resumen estructurado con puntos clave y 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": (
            "Resume el siguiente texto en 5-7 puntos clave. "
            "Usa viñetas. Sé conciso y específico.\n\n"
        ),
        "executive": (
            "Genera un resumen ejecutivo de máximo 3 párrafos. "
            "Incluye: contexto, decisiones tomadas, y próximos pasos.\n\n"
        ),
        "action_items": (
            "Extrae TODOS los action items (tareas pendientes) del siguiente texto. "
            "Formato: '- [ ] [Responsable]: [Tarea] (Fecha si se menciona)'\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
    )

Uso

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

if result.success:
    print(f"Resumen:\n{result.data['summary']}")
    print(f"Costo total: ${result.costs['total']:.4f}")
    print(f"Tiempo total: {result.timing['total_s']}s")

Pipeline 2: Audio → Transcripción → Q&A

Transcribe audio y permite hacer preguntas sobre el contenido. Ideal para buscar información específica en reuniones largas o entrevistas.

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": (
                        "Responde preguntas basándote SOLO en el contexto proporcionado. "
                        "Si la respuesta no está en el contexto, di 'No se menciona en el audio'."
                    )
                },
                {
                    "role": "user",
                    "content": f"Contexto (transcripción de audio):\n{context}\n\nPregunta: {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
    )

Uso

result = pipeline_audio_to_qa(
    "entrevista.mp3",
    questions=[
        "¿Cuál fue la decisión principal?",
        "¿Quién es responsable del seguimiento?",
        "¿Cuál es la fecha límite?"
    ]
)

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

Pipeline 3: Texto → LLM → Audio

Toma texto escrito, lo procesa con un LLM (resumir, reformular, traducir), y genera audio. Ideal para convertir artículos a podcasts o crear versiones de audio de documentos.

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": (
            "Resume el siguiente texto en un párrafo conciso, "
            "optimizado para ser escuchado en audio (usa lenguaje natural, "
            "evita listas con viñetas):\n\n"
        ),
        "simplify": (
            "Reescribe el siguiente texto en lenguaje simple y claro, "
            "como si lo explicaras a un público general. "
            "Optimiza para audio (lenguaje conversacional):\n\n"
        ),
        "translate_en": (
            "Traduce el siguiente texto al inglés. "
            "Mantén un tono natural y conversacional:\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
    )

Uso

article = """
La inteligencia artificial multimodal representa un cambio paradigmático
en cómo los sistemas de IA procesan información. A diferencia de los modelos
tradicionales limitados al texto, los modelos multimodales pueden procesar
imágenes, audio y video simultáneamente...
"""

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

Pipeline 4: Audio → Texto → LLM → Audio (Loop Completo)

El pipeline más poderoso: recibe una pregunta en audio, la transcribe, la procesa con el LLM, y genera la respuesta en audio. Es el core de un asistente de voz.

def pipeline_audio_full_loop(
    audio_path: str,
    system_prompt: str = "Eres un asistente útil. Responde de forma concisa y clara.",
    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
    )

Uso

result = pipeline_audio_full_loop(
    "pregunta.mp3",
    system_prompt="Eres un experto en IA. Responde en español, de forma clara y concisa.",
    voice="echo"
)

print(f"Usuario dijo: {result.data['user_input']}")
print(f"Asistente respondió: {result.data['assistant_response']}")
print(f"Audio respuesta: {result.data['audio_output']}")

Pipeline con Transcripción Larga + Map-Reduce

Para audio muy largo (>1 hora), usa un patrón map-reduce: transcribe por chunks, resume cada chunk, y luego genera un resumen final.

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"Resume este segmento ({i+1}/{len(transcripts)}) "
                    f"en 3-5 oraciones clave:\n\n{transcript[:8000]}"
                )
            }],
            max_tokens=300
        ).choices[0].message.content
        chunk_summaries.append(summary)

    combined_summaries = "\n\n".join(
        f"Segmento {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": (
                "Genera un resumen ejecutivo completo a partir de estos resúmenes parciales. "
                "Incluye: puntos clave, decisiones, y 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
    )

Tracking de Costos

Función de reporte

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
    }

Tabla de referencia de costos por pipeline

Troubleshooting

Problema 1: Transcript truncado en pipeline

Síntoma: El resumen solo cubre parte del audio.

Solución: Usar pipeline_long_audio_summary() con map-reduce para audio largo. El LLM tiene límite de contexto (~12K tokens para gpt-4o-mini).

Problema 2: Resumen genérico o vago

Síntoma: El LLM genera resúmenes que no capturan los puntos específicos.

Solución:

specific_prompt = (
    "Resume este texto de una reunión de equipo. "
    "Incluye NOMBRES de personas y FECHAS mencionadas. "
    "Lista decisiones concretas, no generalidades.\n\n"
)

Problema 3: TTS falla con texto largo

Síntoma: Error al generar audio de resúmenes extensos.

Solución: Truncar o dividir el texto antes de TTS:

tts_text = summary[:4000]

Problema 4: Costo inesperadamente alto

Síntoma: El pipeline cuesta más de lo esperado.

Solución: Verificar costos con generate_cost_report(). Los culpables comunes:

• Audio muy largo → alto costo de Whisper
• Múltiples llamadas al LLM → acumulación de tokens
• TTS con tts-1-hd → doble costo vs tts-1

Ejercicios

Ejercicio 1: Pipeline de notas de reunión estructuradas

Crea un pipeline que transcriba audio y genere un documento estructurado con: asistentes (si se mencionan), temas discutidos, decisiones, action items, y próxima reunión.

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 = """Analiza esta transcripción de reunión y genera un documento estructurado con EXACTAMENTE estas secciones:

## Asistentes
(Lista las personas mencionadas por nombre. Si no se mencionan nombres, escribe "No identificados")

## Temas Discutidos
(Lista numerada de los temas principales)

## Decisiones Tomadas
(Lista de decisiones concretas. Si no hay decisiones claras, escribe "Sin decisiones explícitas")

## Action Items
(Formato: - [ ] [Responsable si se conoce]: [Tarea] [Fecha si se menciona])

## Próxima Reunión
(Fecha/hora si se menciona, sino "No especificada")

Transcripción:
"""

    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
    )

Ejercicio 2: Pipeline multi-idioma

Crea un pipeline que transcriba audio en cualquier idioma, detecte el idioma, traduzca al español si es necesario, y genere un resumen en español.

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"Traduce el siguiente texto de {detected_lang} "
                    f"a {target_language}. Mantén el significado original:\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"Resume en 5 puntos clave (en {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
    )

Ejercicio 3: Pipeline con caché de transcripción

Crea un wrapper que guarde transcripciones en disco para no re-transcribir el mismo archivo. Usa hash del archivo como 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

Resumen

• Pipeline 1 (Resumen): Audio → Whisper → transcript → LLM → resumen/action items.
• Pipeline 2 (Q&A): Audio → Whisper → transcript → LLM + preguntas → respuestas contextuales.
• Pipeline 3 (Texto a Audio): Texto → LLM (procesa/resume/traduce) → TTS → audio.
• Pipeline 4 (Loop completo): Audio → Whisper → LLM → TTS → audio (asistente de voz).
• Map-reduce para audio largo: transcribir por chunks → resumir cada chunk → resumen final.
• Tracking de costos integrado en cada pipeline con PipelineResult.
• El costo típico de un pipeline completo para 30 min de audio es ~$0.20 USD.

Recursos Adicionales

1. Whisper API — Transcripción
2. OpenAI TTS — Síntesis de voz
3. OpenAI Chat API — Procesamiento con LLM
4. pydub — Manipulación de audio en Python