6. Integración de Audio

Descripción

El AudioModule agrega la dimensión sonora al Document Analyzer. Después de procesar un documento, extraer datos y generar un resumen, el sistema puede convertir ese resumen en audio — útil para accesibilidad, consumo en movimiento, o simplemente como formato alternativo de entrega. Opcionalmente, también puede aceptar preguntas por audio (STT → Q&A → TTS), creando una interfaz conversacional completa.

Por qué importa: No todos los usuarios quieren leer. Un ejecutivo que recibe 20 documentos al día prefiere escuchar los resúmenes mientras maneja. Una persona con discapacidad visual necesita audio como output primario. Agregar TTS transforma el Document Analyzer de una herramienta de texto a una herramienta verdaderamente multimodal en su output.

Conexión con el módulo: En el Módulo 5 aprendiste a usar la API de TTS de OpenAI para convertir texto en habla, y Whisper para transcribir audio. Aquí integras ambas capacidades en el Document Analyzer: TTS para generar resúmenes hablados, y opcionalmente STT para recibir preguntas por audio.

Arquitectura del AudioModule

Responsabilidades

AudioModule
├── generate_summary_audio()     → Convertir resumen textual en archivo de audio
├── generate_long_audio()        → Manejar resúmenes que exceden el límite de TTS
├── transcribe_question()        → (Opcional) Convertir pregunta de audio a texto
└── audio_qa_pipeline()          → (Opcional) Audio pregunta → texto → Q&A → audio respuesta

Pipeline de audio

Resumen textual
    │
    ├── ¿Menos de 4096 chars? ── Sí ──→ TTS directo → archivo .mp3
    │
    └── ¿Más de 4096 chars? ─── Sí ──→ Dividir en chunks
                                           │
                                           ▼
                                      TTS por chunk → múltiples .mp3
                                           │
                                           ▼
                                      Concatenar con pydub → archivo .mp3 final

Pipeline de audio Q&A (opcional)

Audio pregunta (.mp3/.wav)
    │
    ▼
Whisper STT → Texto de la pregunta
    │
    ▼
RAGModule.query() → Respuesta textual
    │
    ▼
TTS → Audio respuesta (.mp3)

Implementación: AudioModule

Clase completa

import logging
import os
import tempfile
from pathlib import Path
from typing import Optional

from openai import OpenAI

logger = logging.getLogger(__name__)

TTS_CHAR_LIMIT = 4096
SUPPORTED_VOICES = {"alloy", "echo", "fable", "onyx", "nova", "shimmer"}
TTS_MODEL = "tts-1"
TTS_MODEL_HD = "tts-1-hd"
STT_MODEL = "whisper-1"


class AudioModule:
    def __init__(self, output_dir: str = "/tmp/audio"):
        self.client = OpenAI()
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def generate_summary_audio(
        self,
        text: str,
        filename: str = "resumen.mp3",
        voice: str = "nova",
        hd: bool = False
    ) -> str:
        if voice not in SUPPORTED_VOICES:
            logger.warning(f"Voz '{voice}' no reconocida, usando 'nova'")
            voice = "nova"

        if len(text) > TTS_CHAR_LIMIT:
            return self.generate_long_audio(text, filename, voice, hd)

        output_path = os.path.join(self.output_dir, filename)
        model = TTS_MODEL_HD if hd else TTS_MODEL

        response = self.client.audio.speech.create(
            model=model,
            voice=voice,
            input=text
        )
        response.stream_to_file(output_path)

        file_size = os.path.getsize(output_path)
        logger.info(
            f"Audio generado: {output_path} "
            f"({file_size / 1024:.1f} KB, {len(text)} chars)"
        )

        return output_path

    def generate_long_audio(
        self,
        text: str,
        filename: str = "resumen.mp3",
        voice: str = "nova",
        hd: bool = False
    ) -> str:
        try:
            from pydub import AudioSegment
        except ImportError:
            logger.warning(
                "pydub no instalado. Truncando resumen a 4096 chars."
            )
            return self.generate_summary_audio(
                text[:TTS_CHAR_LIMIT], filename, voice, hd
            )

        chunks = self._split_text_for_tts(text)
        model = TTS_MODEL_HD if hd else TTS_MODEL
        temp_files = []

        try:
            for i, chunk in enumerate(chunks):
                temp_path = os.path.join(self.output_dir, f"_temp_part_{i}.mp3")
                response = self.client.audio.speech.create(
                    model=model,
                    voice=voice,
                    input=chunk
                )
                response.stream_to_file(temp_path)
                temp_files.append(temp_path)
                logger.info(f"Parte {i+1}/{len(chunks)} generada ({len(chunk)} chars)")

            segments = [AudioSegment.from_mp3(f) for f in temp_files]

            pause = AudioSegment.silent(duration=500)
            combined = segments[0]
            for seg in segments[1:]:
                combined = combined + pause + seg

            output_path = os.path.join(self.output_dir, filename)
            combined.export(output_path, format="mp3")

            total_duration = len(combined) / 1000
            logger.info(
                f"Audio largo generado: {output_path} "
                f"({len(chunks)} partes, {total_duration:.1f}s)"
            )

            return output_path

        finally:
            for f in temp_files:
                try:
                    os.remove(f)
                except OSError:
                    pass

    def transcribe_question(self, audio_path: str) -> str:
        with open(audio_path, "rb") as f:
            transcript = self.client.audio.transcriptions.create(
                model=STT_MODEL,
                file=f,
                language="es"
            )
        logger.info(f"Transcripción: '{transcript.text[:100]}...'")
        return transcript.text

    def audio_qa_pipeline(
        self,
        audio_question_path: str,
        rag_module,
        doc_id: Optional[str] = None,
        voice: str = "nova"
    ) -> dict:
        question_text = self.transcribe_question(audio_question_path)

        qa_result = rag_module.query(question=question_text, doc_id=doc_id)

        answer_audio_path = self.generate_summary_audio(
            text=qa_result.answer,
            filename=f"qa_answer_{doc_id or 'all'}.mp3",
            voice=voice
        )

        return {
            "question_text": question_text,
            "answer_text": qa_result.answer,
            "answer_audio_path": answer_audio_path,
            "sources": qa_result.sources,
            "confidence": qa_result.confidence
        }

    def _split_text_for_tts(self, text: str) -> list[str]:
        if len(text) <= TTS_CHAR_LIMIT:
            return [text]

        chunks = []
        sentences = text.replace(". ", ".\n").split("\n")
        current_chunk = ""

        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue

            if len(current_chunk) + len(sentence) + 1 <= TTS_CHAR_LIMIT:
                current_chunk += (" " + sentence if current_chunk else sentence)
            else:
                if current_chunk:
                    chunks.append(current_chunk)
                if len(sentence) > TTS_CHAR_LIMIT:
                    for i in range(0, len(sentence), TTS_CHAR_LIMIT):
                        chunks.append(sentence[i:i + TTS_CHAR_LIMIT])
                    current_chunk = ""
                else:
                    current_chunk = sentence

        if current_chunk:
            chunks.append(current_chunk)

        return chunks

    def estimate_cost(self, text: str) -> dict:
        char_count = len(text)
        cost_per_char = 15.0 / 1_000_000  # $15 por 1M chars para tts-1
        cost_per_char_hd = 30.0 / 1_000_000  # $30 por 1M chars para tts-1-hd

        return {
            "characters": char_count,
            "chunks_needed": max(1, (char_count + TTS_CHAR_LIMIT - 1) // TTS_CHAR_LIMIT),
            "cost_tts1": round(char_count * cost_per_char, 4),
            "cost_tts1_hd": round(char_count * cost_per_char_hd, 4)
        }

Configuración de Voces

Voces disponibles en OpenAI TTS

Selección de voz por tipo de documento

VOICE_BY_DOCTYPE = {
    "factura": "alloy",      # neutra para datos financieros
    "contrato": "onyx",      # formal para documentos legales
    "manual": "fable",       # cálida para instrucciones
    "informe": "nova",       # amigable para resúmenes
}


def get_voice_for_document(doc_type: str, user_preference: Optional[str] = None) -> str:
    if user_preference and user_preference in SUPPORTED_VOICES:
        return user_preference
    return VOICE_BY_DOCTYPE.get(doc_type, "nova")

Modelos TTS: Estándar vs HD

Comparación

Cuándo usar HD

def should_use_hd(doc_type: str, text_length: int) -> bool:
    if doc_type in ("contrato", "informe") and text_length < 2000:
        return True
    return False

Para el Document Analyzer, tts-1 es suficiente en la mayoría de casos. Usa tts-1-hd solo para documentos importantes donde la calidad de audio es crítica.

Integración con el Document Analyzer

Flujo completo con audio

def analyze_with_audio(
    processor: DocumentProcessor,
    analyzer: VisionAnalyzer,
    summarizer,
    audio: AudioModule,
    file_path: str,
    voice: str = "nova",
    hd: bool = False
) -> dict:
    content = processor.process(file_path)
    doc_type = analyzer.classify(content)
    summary = summarizer.summarize(content)

    cost_estimate = audio.estimate_cost(summary)
    logger.info(f"Costo estimado TTS: ${cost_estimate['cost_tts1']:.4f}")

    audio_path = audio.generate_summary_audio(
        text=summary,
        filename=f"{Path(file_path).stem}_resumen.mp3",
        voice=voice,
        hd=hd
    )

    return {
        "summary": summary,
        "audio_path": audio_path,
        "audio_cost": cost_estimate,
        "document_type": doc_type
    }

Servir Audio desde FastAPI

Endpoint para audio

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()


@app.get("/audio/{filename}")
async def serve_audio(filename: str):
    audio_dir = "/tmp/audio"
    file_path = os.path.join(audio_dir, filename)

    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="Audio no encontrado")

    if not filename.endswith((".mp3", ".wav")):
        raise HTTPException(status_code=400, detail="Formato de audio no soportado")

    return FileResponse(
        file_path,
        media_type="audio/mpeg",
        filename=filename
    )

Respuesta del endpoint /analyze con audio

Cuando generate_audio_summary=True, la respuesta incluye la URL del audio:

{
    "success": true,
    "summary": "Factura FAC-2025-0042...",
    "audio_summary_url": "/audio/factura_marzo_resumen.mp3",
    "metadata": {
        "audio_duration_estimate_seconds": 15.5,
        "audio_cost_usd": 0.0075
    }
}

Manejo de Resúmenes Largos

El problema

OpenAI TTS tiene un límite de 4096 caracteres por request. Un resumen de un informe de 30 páginas puede tener 6000+ caracteres.

Estrategia de división

La función _split_text_for_tts() divide por oraciones, no por caracteres arbitrarios:

Texto de 6000 chars
    │
    ├── Dividir por oraciones (". " como separador)
    │
    ├── Agrupar oraciones hasta 4096 chars por chunk
    │   ├── Chunk 1: oraciones 1-15 (3800 chars)
    │   └── Chunk 2: oraciones 16-22 (2200 chars)
    │
    ├── Generar audio por chunk
    │   ├── Chunk 1 → part_0.mp3
    │   └── Chunk 2 → part_1.mp3
    │
    └── Concatenar con pausa de 500ms entre chunks
        └── resumen_final.mp3

Por qué dividir por oraciones

Dividir a mitad de oración produce audio con cortes abruptos. Dividir por oraciones asegura que cada chunk es una unidad semántica completa:

MAL:  "...el total de la factura es de $1,74"  |  "0.00 pesos mexicanos..."
BIEN: "...el total de la factura es de $1,740.00 pesos mexicanos."  |  "Los items incluyen..."

Estimación de Duración de Audio

Fórmula aproximada

Una voz estándar en español habla ~150 palabras por minuto. Un carácter promedio en español tiene ~5 caracteres por palabra.

def estimate_audio_duration(text: str) -> float:
    word_count = len(text.split())
    words_per_minute = 150
    return round(word_count / words_per_minute * 60, 1)

Tabla de referencia

Troubleshooting

"El audio suena cortado o con glitches"

Causa probable: El texto tiene caracteres especiales que TTS no maneja bien (emojis, caracteres Unicode extraños, URLs largas).

Solución: Limpiar el texto antes de TTS:

import re


def clean_text_for_tts(text: str) -> str:
    text = re.sub(r'https?://\S+', 'enlace web', text)
    text = re.sub(r'[^\w\s.,;:!?¿¡()\-\'"áéíóúüñÁÉÍÓÚÜÑ$%]', '', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

"Error al concatenar audios con pydub"

Causa probable: ffmpeg no está instalado (requerido por pydub).

Solución:

# Mac
brew install ffmpeg

# Ubuntu/Debian
apt-get install ffmpeg

# Docker (agregar al Dockerfile)
RUN apt-get update && apt-get install -y ffmpeg

"El audio es demasiado largo (>5 minutos)"

Solución: Resumir más agresivamente antes de TTS:

def compact_summary_for_tts(summary: str, max_chars: int = 2000) -> str:
    if len(summary) <= max_chars:
        return summary

    from openai import OpenAI
    client = OpenAI()

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Compacta este resumen en máximo {max_chars} caracteres "
                f"manteniendo los puntos clave:\n\n{summary}"
            )
        }],
        max_tokens=500
    )
    return r.choices[0].message.content

"Costo alto de TTS en producción"

Solución: Implementar cache de audio:

import hashlib


class AudioCache:
    def __init__(self, cache_dir: str = "/tmp/audio_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def get_or_generate(
        self,
        text: str,
        audio_module: AudioModule,
        voice: str = "nova"
    ) -> str:
        cache_key = hashlib.md5(f"{text}_{voice}".encode()).hexdigest()
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.mp3")

        if os.path.exists(cache_path):
            logger.info(f"Audio cache hit: {cache_key}")
            return cache_path

        return audio_module.generate_summary_audio(
            text=text,
            filename=f"{cache_key}.mp3",
            voice=voice
        )

Uso del AudioModule

Ejemplo completo

audio = AudioModule(output_dir="./audio_output")

summary = """
Factura FAC-2025-0042 de Tech Solutions S.A. por $1,740.00 MXN.
Incluye: Licencia de software por $1,200.00 y soporte técnico por $300.00.
Subtotal: $1,500.00. IVA 16%: $240.00. Total: $1,740.00.
Fecha de emisión: 15 de marzo de 2025.
Método de pago: Transferencia bancaria.
"""

cost = audio.estimate_cost(summary)
print(f"Costo estimado: ${cost['cost_tts1']:.4f}")
print(f"Chunks necesarios: {cost['chunks_needed']}")

audio_path = audio.generate_summary_audio(
    text=summary,
    filename="factura_resumen.mp3",
    voice="nova"
)
print(f"Audio generado: {audio_path}")

duration = estimate_audio_duration(summary)
print(f"Duración estimada: {duration}s")

Output esperado

Costo estimado: $0.0042
Chunks necesarios: 1
Audio generado: ./audio_output/factura_resumen.mp3
Duración estimada: 24.0s

Ejercicios

Ejercicio 1: Audio con múltiples secciones y pausas

Modifica generate_long_audio() para que, además de la pausa entre chunks, agregue una pausa más larga (1.5 segundos) cuando detecte un cambio de sección en el texto (indicado por saltos de línea dobles o headers con #).

def generate_sectioned_audio(
    self,
    text: str,
    filename: str = "resumen.mp3",
    voice: str = "nova"
) -> str:
    from pydub import AudioSegment

    sections = re.split(r'\n\n+|(?=^#{1,3}\s)', text, flags=re.MULTILINE)
    sections = [s.strip() for s in sections if s.strip()]

    temp_files = []
    try:
        for i, section in enumerate(sections):
            sub_chunks = self._split_text_for_tts(section)
            for j, chunk in enumerate(sub_chunks):
                temp_path = os.path.join(self.output_dir, f"_sec_{i}_part_{j}.mp3")
                response = self.client.audio.speech.create(
                    model=TTS_MODEL, voice=voice, input=chunk
                )
                response.stream_to_file(temp_path)
                temp_files.append({"path": temp_path, "section_end": j == len(sub_chunks) - 1})

        short_pause = AudioSegment.silent(duration=500)
        long_pause = AudioSegment.silent(duration=1500)

        combined = AudioSegment.from_mp3(temp_files[0]["path"])
        for item in temp_files[1:]:
            pause = long_pause if item.get("section_end") else short_pause
            combined = combined + pause + AudioSegment.from_mp3(item["path"])

        output_path = os.path.join(self.output_dir, filename)
        combined.export(output_path, format="mp3")
        return output_path

    finally:
        for item in temp_files:
            try:
                os.remove(item["path"])
            except OSError:
                pass

Ejercicio 2: Pipeline completo audio Q&A

Implementa el flujo completo: el usuario envía un archivo de audio con su pregunta, el sistema la transcribe, busca en el RAG, genera la respuesta, y la convierte a audio. Retorna tanto el texto como el audio de la respuesta.

def full_audio_qa(
    audio_module: AudioModule,
    rag_module,
    audio_question_path: str,
    doc_id: Optional[str] = None,
    voice: str = "nova"
) -> dict:
    question_text = audio_module.transcribe_question(audio_question_path)
    logger.info(f"Pregunta transcrita: {question_text}")

    qa_result = rag_module.query(question=question_text, doc_id=doc_id)
    logger.info(f"Respuesta generada: {qa_result.answer[:100]}...")

    clean_answer = clean_text_for_tts(qa_result.answer)
    answer_filename = f"qa_answer_{doc_id or 'all'}.mp3"
    answer_audio = audio_module.generate_summary_audio(
        text=clean_answer,
        filename=answer_filename,
        voice=voice
    )

    return {
        "question": {
            "audio_path": audio_question_path,
            "text": question_text
        },
        "answer": {
            "text": qa_result.answer,
            "audio_path": answer_audio,
            "sources": qa_result.sources,
            "confidence": qa_result.confidence
        },
        "costs": {
            "stt": 0.006,  # ~$0.006/min para Whisper
            "rag_query": 0.01,
            "tts": audio_module.estimate_cost(clean_answer)["cost_tts1"]
        }
    }


audio = AudioModule()
rag = RAGModule(persist_directory="./chroma_data")

result = full_audio_qa(audio, rag, "pregunta.mp3", doc_id="factura_001")
print(f"Pregunta: {result['question']['text']}")
print(f"Respuesta: {result['answer']['text']}")
print(f"Audio: {result['answer']['audio_path']}")

Resumen

• El AudioModule convierte resúmenes textuales en audio con OpenAI TTS.
• Maneja resúmenes largos dividiéndolos en chunks y concatenando con pydub.
• 6 voces disponibles: alloy, echo, fable, onyx, nova, shimmer — seleccionables por tipo de documento.
• Dos modelos: tts-1 (estándar, económico) y tts-1-hd (alta calidad, doble costo).
• Pipeline audio Q&A opcional: pregunta en audio → transcripción → RAG → respuesta en audio.
• Cache de audio evita regenerar el mismo contenido, reduciendo costos en producción.
• Integración con FastAPI para servir archivos de audio como endpoints.

Recursos Adicionales

1. OpenAI TTS API — Documentación oficial
2. OpenAI Whisper API — Transcripción
3. pydub Documentation — Manipulación de audio
4. Módulo 5 de esta guía — Base de procesamiento de audio