Módulo 5: Procesamiento de Audio
8. Proyecto: Audio Pipeline
Descripción
Este proyecto integra todo lo aprendido en el Módulo 5 en un Audio Pipeline completo y configurable. El sistema recibe un archivo de audio, lo valida y preprocesa, lo transcribe con Whisper, procesa el transcript con un LLM (resumen, action items, Q&A), y opcionalmente genera audio del resultado con TTS. Es un sistema de producción con configuración flexible, tracking de costos, manejo de errores, y extensiones para batch processing y diarización.
Por qué importa: Este pipeline es el componente de audio del Document Analyzer del Módulo 8. Cuando el sistema final reciba un archivo de audio, delegará a exactamente este pipeline. Además, es un sistema útil por sí mismo: cualquier equipo que grabe reuniones puede usarlo para generar resúmenes automáticos.
Especificaciones
Input
| Parámetro | Tipo | Requerido | Descripción |
|---|---|---|---|
audio_path | str | Sí | Ruta al archivo de audio |
task | str | No | "summarize", "action_items", "qa", "full" |
questions | list[str] | Solo si task="qa" | Preguntas sobre el audio |
generate_audio | bool | No | Si True, genera TTS del resultado |
language | str | No | Código ISO del idioma (default: "es") |
voice | str | No | Voz para TTS (default: "nova") |
vocabulary | list[str] | No | Términos técnicos para mejorar transcripción |
Output
{
"status": "success",
"transcript": "Texto completo transcrito...",
"result": {
"summary": "Resumen en puntos clave...",
"action_items": ["Tarea 1", "Tarea 2"],
},
"audio_output_path": "resumen.mp3",
"metadata": {
"audio_duration_min": 15.3,
"transcript_words": 2340,
"language_detected": "es",
"chunks_processed": 1
},
"costs": {
"whisper": 0.0918,
"llm": 0.0012,
"tts": 0.0045,
"total": 0.0975
},
"timing": {
"preprocessing_s": 0.5,
"transcription_s": 8.2,
"llm_s": 1.3,
"tts_s": 2.1,
"total_s": 12.1
}
}Paso 1: Configuración
from openai import OpenAI
from pydub import AudioSegment
from pathlib import Path
from dataclasses import dataclass, field
import tempfile
import time
import json
client = OpenAI()
WHISPER_FORMATS = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm", ".flac", ".ogg", ".oga"}
WHISPER_MAX_SIZE = 25 * 1024 * 1024
@dataclass
class PipelineConfig:
task: str = "summarize"
language: str = "es"
voice: str = "nova"
tts_model: str = "tts-1"
llm_model: str = "gpt-4o-mini"
generate_audio: bool = False
questions: list[str] = field(default_factory=list)
vocabulary: list[str] = field(default_factory=list)
max_transcript_chars: int = 12000
max_tts_chars: int = 4000
output_dir: str = "pipeline_output"
auto_preprocess: bool = True
@dataclass
class PipelineMetrics:
costs: dict = field(default_factory=dict)
timing: dict = field(default_factory=dict)
def add_cost(self, service: str, amount: float):
self.costs[service] = self.costs.get(service, 0) + amount
self.costs["total"] = sum(v for k, v in self.costs.items() if k != "total")
def start_timer(self, name: str):
self.timing[f"_start_{name}"] = time.time()
def stop_timer(self, name: str):
start_key = f"_start_{name}"
if start_key in self.timing:
self.timing[name] = round(time.time() - self.timing.pop(start_key), 2)Paso 2: Validación y Preprocesamiento
def validate_audio(audio_path: str) -> dict:
path = Path(audio_path)
if not path.exists():
return {"valid": False, "error": "Archivo no encontrado", "path": audio_path}
suffix = path.suffix.lower()
size_bytes = path.stat().st_size
size_mb = size_bytes / (1024 * 1024)
issues = []
if suffix not in WHISPER_FORMATS:
issues.append({
"type": "unsupported_format",
"detail": f"Formato {suffix} no soportado",
"auto_fixable": True
})
if size_bytes > WHISPER_MAX_SIZE:
issues.append({
"type": "file_too_large",
"detail": f"Archivo de {size_mb:.1f}MB excede 25MB",
"auto_fixable": True
})
try:
audio = AudioSegment.from_file(audio_path)
duration_s = len(audio) / 1000
if audio.dBFS < -35:
issues.append({
"type": "low_volume",
"detail": f"Volumen bajo ({audio.dBFS:.1f} dBFS)",
"auto_fixable": True
})
if audio.frame_rate < 16000:
issues.append({
"type": "low_sample_rate",
"detail": f"Sample rate bajo ({audio.frame_rate} Hz)",
"auto_fixable": True
})
except Exception as e:
return {"valid": False, "error": f"No se pudo leer el audio: {str(e)}"}
return {
"valid": len([i for i in issues if not i["auto_fixable"]]) == 0,
"size_mb": round(size_mb, 2),
"duration_s": round(duration_s, 2),
"format": suffix,
"issues": issues,
"needs_preprocessing": len(issues) > 0
}
def preprocess_audio(audio_path: str, output_dir: str) -> dict:
Path(output_dir).mkdir(parents=True, exist_ok=True)
path = Path(audio_path)
steps = []
current = audio_path
if path.suffix.lower() not in WHISPER_FORMATS:
converted = str(Path(output_dir) / f"{path.stem}.mp3")
AudioSegment.from_file(current).export(converted, format="mp3", bitrate="128k")
current = converted
steps.append("format_conversion")
audio = AudioSegment.from_file(current)
needs_normalize = (
audio.dBFS < -30
or audio.channels > 1
or audio.frame_rate < 16000
)
if needs_normalize:
normalized = str(Path(output_dir) / f"{path.stem}_normalized.mp3")
audio = audio.set_frame_rate(16000).set_channels(1).normalize()
audio.export(normalized, format="mp3", bitrate="64k")
current = normalized
steps.append("normalization")
size_mb = Path(current).stat().st_size / (1024 * 1024)
if size_mb > 24:
compressed = str(Path(output_dir) / f"{path.stem}_compressed.mp3")
audio = AudioSegment.from_file(current)
audio = audio.set_channels(1).set_frame_rate(16000)
audio.export(compressed, format="mp3", bitrate="48k")
current = compressed
steps.append("compression")
return {
"output_path": current,
"steps": steps,
"final_size_mb": round(Path(current).stat().st_size / (1024 * 1024), 2)
}Paso 3: Transcripción con Whisper
def transcribe_audio(
audio_path: str,
config: PipelineConfig,
metrics: PipelineMetrics
) -> dict:
metrics.start_timer("transcription")
path = Path(audio_path)
size_mb = path.stat().st_size / (1024 * 1024)
audio = AudioSegment.from_file(audio_path)
duration_min = len(audio) / 1000 / 60
if size_mb > 24:
result = _transcribe_chunked(audio, config, metrics)
else:
result = _transcribe_single(audio_path, config)
metrics.stop_timer("transcription")
metrics.add_cost("whisper", duration_min * 0.006)
return {
"text": result["text"],
"language": result.get("language", config.language),
"duration_min": round(duration_min, 2),
"word_count": len(result["text"].split()),
"chunks": result.get("chunks", 1)
}
def _transcribe_single(audio_path: str, config: PipelineConfig) -> dict:
kwargs = {
"model": "whisper-1",
"response_format": "verbose_json",
}
if config.language:
kwargs["language"] = config.language
if config.vocabulary:
kwargs["prompt"] = ", ".join(config.vocabulary)
with open(audio_path, "rb") as f:
kwargs["file"] = f
response = client.audio.transcriptions.create(**kwargs)
return {
"text": response.text,
"language": response.language,
"duration": response.duration,
"chunks": 1
}
def _transcribe_chunked(
audio: AudioSegment,
config: PipelineConfig,
metrics: PipelineMetrics,
chunk_duration_ms: int = 10 * 60 * 1000
) -> dict:
chunks = [
audio[i:i + chunk_duration_ms]
for i in range(0, len(audio), chunk_duration_ms)
]
transcripts = []
detected_language = None
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")
result = _transcribe_single(chunk_path, config)
transcripts.append(result["text"])
if detected_language is None:
detected_language = result.get("language")
return {
"text": " ".join(transcripts),
"language": detected_language,
"chunks": len(chunks)
}Paso 4: Procesamiento con LLM
def process_with_llm(
transcript: str,
config: PipelineConfig,
metrics: PipelineMetrics
) -> dict:
metrics.start_timer("llm")
truncated = transcript[:config.max_transcript_chars]
if config.task == "summarize":
result = _summarize(truncated, config)
elif config.task == "action_items":
result = _extract_action_items(truncated, config)
elif config.task == "qa":
result = _answer_questions(truncated, config)
elif config.task == "full":
result = _full_analysis(truncated, config)
else:
result = _summarize(truncated, config)
input_tokens = len(truncated.split())
metrics.add_cost("llm", input_tokens * 0.00000015 + 1000 * 0.0000006)
metrics.stop_timer("llm")
return result
def _summarize(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Resume el siguiente texto en 5-7 puntos clave. "
"Sé específico: incluye nombres, fechas y datos mencionados.\n\n"
+ transcript
)
}],
max_tokens=800
)
return {"summary": response.choices[0].message.content}
def _extract_action_items(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Extrae TODOS los action items del siguiente texto.\n"
"Formato por cada uno:\n"
"- [ ] [Responsable si se menciona]: [Tarea concreta] "
"[Fecha si se menciona]\n\n"
"Si no hay action items claros, indica 'No se identificaron "
"action items explícitos' y sugiere posibles tareas implícitas.\n\n"
+ transcript
)
}],
max_tokens=600
)
return {"action_items": response.choices[0].message.content}
def _answer_questions(transcript: str, config: PipelineConfig) -> dict:
answers = []
for question in config.questions:
response = client.chat.completions.create(
model=config.llm_model,
messages=[
{
"role": "system",
"content": (
"Responde basándote SOLO en el contexto proporcionado. "
"Si la información no está disponible, di "
"'No se menciona en el audio'."
)
},
{
"role": "user",
"content": f"Contexto:\n{transcript}\n\nPregunta: {question}"
}
],
max_tokens=300
)
answers.append({
"question": question,
"answer": response.choices[0].message.content
})
return {"qa": answers}
def _full_analysis(transcript: str, config: PipelineConfig) -> dict:
response = client.chat.completions.create(
model=config.llm_model,
messages=[{
"role": "user",
"content": (
"Analiza el siguiente texto y genera un reporte completo con "
"EXACTAMENTE estas secciones:\n\n"
"## Resumen Ejecutivo\n"
"(3-5 oraciones con los puntos más importantes)\n\n"
"## Temas Principales\n"
"(Lista numerada)\n\n"
"## Decisiones Tomadas\n"
"(Lista. Si no hay, indicar 'Sin decisiones explícitas')\n\n"
"## Action Items\n"
"(Formato: - [ ] [Responsable]: [Tarea])\n\n"
"## Datos y Cifras Mencionados\n"
"(Fechas, montos, porcentajes, etc.)\n\n"
+ transcript
)
}],
max_tokens=1200
)
return {"full_analysis": response.choices[0].message.content}Paso 5: Generación de Audio (TTS)
def generate_audio_output(
text: str,
config: PipelineConfig,
metrics: PipelineMetrics,
output_path: str = None
) -> dict:
metrics.start_timer("tts")
if output_path is None:
output_path = str(Path(config.output_dir) / "audio_output.mp3")
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
tts_text = text[:config.max_tts_chars]
tts_text = _prepare_text_for_tts(tts_text)
if len(tts_text) > 4000:
_generate_long_audio(tts_text, config, output_path)
else:
response = client.audio.speech.create(
model=config.tts_model,
voice=config.voice,
input=tts_text
)
response.stream_to_file(output_path)
cost_per_char = 0.000015 if config.tts_model == "tts-1" else 0.00003
metrics.add_cost("tts", len(tts_text) * cost_per_char)
metrics.stop_timer("tts")
return {
"audio_path": output_path,
"characters": len(tts_text),
"voice": config.voice,
"model": config.tts_model
}
def _prepare_text_for_tts(text: str) -> str:
text = text.strip()
text = text.replace("- [ ]", ".")
text = text.replace("##", ".")
text = text.replace("**", "")
text = text.replace("*", "")
text = text.replace("\n\n", ". ")
text = text.replace("\n", " ")
import re
text = re.sub(r'\s+', ' ', text)
if not text.endswith((".", "!", "?")):
text += "."
return text
def _generate_long_audio(text: str, config: PipelineConfig, output_path: str):
sentences = text.replace(". ", ".\n").split("\n")
sentences = [s.strip() for s in sentences if s.strip()]
chunks = []
current = ""
for sentence in sentences:
if len(current) + len(sentence) + 1 <= 4000:
current = f"{current} {sentence}".strip()
else:
if current:
chunks.append(current)
current = sentence
if current:
chunks.append(current)
segments = []
with tempfile.TemporaryDirectory() as tmp_dir:
for i, chunk in enumerate(chunks):
chunk_path = str(Path(tmp_dir) / f"tts_chunk_{i:03d}.mp3")
response = client.audio.speech.create(
model=config.tts_model,
voice=config.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")Paso 6: Pipeline Integrado
class AudioPipeline:
def __init__(self, config: PipelineConfig = None):
self.config = config or PipelineConfig()
Path(self.config.output_dir).mkdir(parents=True, exist_ok=True)
def run(self, audio_path: str) -> dict:
metrics = PipelineMetrics()
metrics.start_timer("total")
try:
validation = validate_audio(audio_path)
if not validation["valid"] and not self.config.auto_preprocess:
return self._error_result(
f"Audio inválido: {validation.get('error', validation.get('issues'))}",
metrics
)
if validation.get("needs_preprocessing") and self.config.auto_preprocess:
metrics.start_timer("preprocessing")
prep = preprocess_audio(audio_path, self.config.output_dir)
working_path = prep["output_path"]
metrics.stop_timer("preprocessing")
else:
working_path = audio_path
transcript_data = transcribe_audio(working_path, self.config, metrics)
llm_result = process_with_llm(
transcript_data["text"], self.config, metrics
)
audio_output = None
if self.config.generate_audio:
result_text = self._get_text_for_tts(llm_result)
if result_text:
audio_output = generate_audio_output(
result_text, self.config, metrics
)
metrics.stop_timer("total")
return {
"status": "success",
"transcript": transcript_data["text"],
"result": llm_result,
"audio_output_path": audio_output["audio_path"] if audio_output else None,
"metadata": {
"audio_duration_min": transcript_data["duration_min"],
"transcript_words": transcript_data["word_count"],
"language_detected": transcript_data["language"],
"chunks_processed": transcript_data["chunks"],
"task": self.config.task
},
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}
except Exception as e:
metrics.stop_timer("total")
return self._error_result(str(e), metrics)
def _get_text_for_tts(self, llm_result: dict) -> str:
if "summary" in llm_result:
return llm_result["summary"]
if "full_analysis" in llm_result:
return llm_result["full_analysis"]
if "action_items" in llm_result:
return llm_result["action_items"]
if "qa" in llm_result:
return "\n".join(
f"Pregunta: {qa['question']}. Respuesta: {qa['answer']}"
for qa in llm_result["qa"]
)
return None
def _error_result(self, error: str, metrics: PipelineMetrics) -> dict:
return {
"status": "error",
"error": error,
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}Paso 7: Demo
Resumen de reunión
config = PipelineConfig(
task="summarize",
language="es",
generate_audio=True,
voice="nova",
vocabulary=["sprint", "deployment", "QA", "staging"]
)
pipeline = AudioPipeline(config)
result = pipeline.run("reunion_equipo.mp3")
if result["status"] == "success":
print("=== RESUMEN ===")
print(result["result"]["summary"])
print(f"\nDuración del audio: {result['metadata']['audio_duration_min']} min")
print(f"Palabras transcritas: {result['metadata']['transcript_words']}")
print(f"Costo total: ${result['costs']['total']:.4f}")
print(f"Tiempo total: {result['timing']['total']}s")
if result["audio_output_path"]:
print(f"Audio del resumen: {result['audio_output_path']}")Extracción de action items
config = PipelineConfig(task="action_items", language="es")
pipeline = AudioPipeline(config)
result = pipeline.run("standup.mp3")
if result["status"] == "success":
print(result["result"]["action_items"])Q&A sobre el audio
config = PipelineConfig(
task="qa",
language="es",
questions=[
"¿Cuál fue la decisión sobre el presupuesto?",
"¿Quién es responsable del deploy?",
"¿Cuándo es la fecha de entrega?"
]
)
pipeline = AudioPipeline(config)
result = pipeline.run("planning.mp3")
if result["status"] == "success":
for qa in result["result"]["qa"]:
print(f"P: {qa['question']}")
print(f"R: {qa['answer']}\n")Análisis completo
config = PipelineConfig(
task="full",
language="es",
generate_audio=True,
tts_model="tts-1-hd",
voice="echo"
)
pipeline = AudioPipeline(config)
result = pipeline.run("board_meeting.mp3")
if result["status"] == "success":
print(result["result"]["full_analysis"])Extensión 1: Batch Processing
Procesar múltiples archivos de audio en lote:
from concurrent.futures import ThreadPoolExecutor
class BatchAudioPipeline:
def __init__(self, config: PipelineConfig = None, max_workers: int = 3):
self.config = config or PipelineConfig()
self.max_workers = max_workers
def run_batch(self, audio_paths: list[str]) -> dict:
pipeline = AudioPipeline(self.config)
results = []
def process_one(path: str) -> dict:
result = pipeline.run(path)
result["source_file"] = Path(path).name
return result
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
results = list(executor.map(process_one, audio_paths))
total_cost = sum(
r.get("costs", {}).get("total", 0)
for r in results
)
total_time = sum(
r.get("timing", {}).get("total", 0)
for r in results
)
successful = sum(1 for r in results if r["status"] == "success")
return {
"results": results,
"summary": {
"total_files": len(audio_paths),
"successful": successful,
"failed": len(audio_paths) - successful,
"total_cost_usd": round(total_cost, 4),
"total_time_s": round(total_time, 2)
}
}Uso
batch = BatchAudioPipeline(
config=PipelineConfig(task="summarize", language="es"),
max_workers=3
)
audio_files = list(Path("recordings").glob("*.mp3"))
results = batch.run_batch([str(f) for f in audio_files])
print(f"Procesados: {results['summary']['successful']}/{results['summary']['total_files']}")
print(f"Costo total: ${results['summary']['total_cost_usd']}")Extensión 2: Diarización con AssemblyAI
Extender el pipeline para identificar speakers:
import assemblyai as aai
class DiarizedAudioPipeline(AudioPipeline):
def __init__(self, config: PipelineConfig = None):
super().__init__(config)
aai.settings.api_key = os.getenv("ASSEMBLYAI_API_KEY")
def run_with_diarization(self, audio_path: str) -> dict:
base_result = self.run(audio_path)
if base_result["status"] != "success":
return base_result
try:
diarization = self._diarize(audio_path)
base_result["diarization"] = diarization
if diarization["utterances"]:
dialogue = self._format_dialogue(diarization["utterances"])
base_result["result"]["dialogue"] = dialogue
except Exception as e:
base_result["diarization_error"] = str(e)
return base_result
def _diarize(self, audio_path: str) -> dict:
config = aai.TranscriptionConfig(
speaker_labels=True,
language_code=self.config.language
)
transcript = aai.Transcriber().transcribe(audio_path, config=config)
if transcript.status == aai.TranscriptStatus.error:
raise RuntimeError(f"Error de diarización: {transcript.error}")
utterances = [
{
"speaker": u.speaker,
"text": u.text,
"start_ms": u.start,
"end_ms": u.end
}
for u in (transcript.utterances or [])
]
speakers = list(set(u["speaker"] for u in utterances))
return {
"utterances": utterances,
"speakers": speakers,
"speaker_count": len(speakers)
}
def _format_dialogue(self, utterances: list[dict]) -> str:
lines = []
for u in utterances:
start_min = u["start_ms"] // 60000
start_sec = (u["start_ms"] % 60000) // 1000
timestamp = f"[{start_min:02d}:{start_sec:02d}]"
lines.append(f"{timestamp} Speaker {u['speaker']}: {u['text']}")
return "\n\n".join(lines)Extensión 3: Exportar Resultados
def export_results(result: dict, output_dir: str, formats: list[str] = None):
if formats is None:
formats = ["json", "md", "txt"]
Path(output_dir).mkdir(parents=True, exist_ok=True)
if "json" in formats:
json_path = Path(output_dir) / "pipeline_result.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
if "md" in formats:
md_path = Path(output_dir) / "pipeline_result.md"
md_content = _build_markdown_report(result)
with open(md_path, "w", encoding="utf-8") as f:
f.write(md_content)
if "txt" in formats:
txt_path = Path(output_dir) / "transcript.txt"
with open(txt_path, "w", encoding="utf-8") as f:
f.write(result.get("transcript", ""))
def _build_markdown_report(result: dict) -> str:
lines = ["# Audio Pipeline Report\n"]
meta = result.get("metadata", {})
lines.append(f"**Duración:** {meta.get('audio_duration_min', 'N/A')} min")
lines.append(f"**Palabras:** {meta.get('transcript_words', 'N/A')}")
lines.append(f"**Idioma:** {meta.get('language_detected', 'N/A')}")
lines.append(f"**Tarea:** {meta.get('task', 'N/A')}\n")
lines.append("## Resultado\n")
llm_result = result.get("result", {})
for key, value in llm_result.items():
lines.append(f"### {key.replace('_', ' ').title()}\n")
if isinstance(value, list):
for item in value:
if isinstance(item, dict):
lines.append(f"**{item.get('question', '')}**")
lines.append(f"{item.get('answer', '')}\n")
else:
lines.append(f"- {item}")
else:
lines.append(str(value))
lines.append("")
costs = result.get("costs", {})
if costs:
lines.append("## Costos\n")
lines.append("| Servicio | Costo USD |")
lines.append("|----------|-----------|")
for service, cost in costs.items():
lines.append(f"| {service} | ${cost:.6f} |")
lines.append("")
timing = result.get("timing", {})
if timing:
lines.append("## Timing\n")
lines.append("| Paso | Segundos |")
lines.append("|------|----------|")
for step, secs in timing.items():
lines.append(f"| {step} | {secs}s |")
return "\n".join(lines)Troubleshooting del Proyecto
Problema: Pipeline falla silenciosamente
Síntoma: status: "error" sin información útil.
Solución: Agregar logging detallado:
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("audio_pipeline")
class AudioPipelineDebug(AudioPipeline):
def run(self, audio_path: str) -> dict:
logger.info(f"Iniciando pipeline para: {audio_path}")
logger.info(f"Config: task={self.config.task}, lang={self.config.language}")
result = super().run(audio_path)
if result["status"] == "error":
logger.error(f"Pipeline falló: {result['error']}")
else:
logger.info(f"Pipeline exitoso. Costo: ${result['costs'].get('total', 0):.4f}")
return resultProblema: Transcripción parcial en audio largo
Síntoma: Solo se transcribe una parte del audio.
Solución: Verificar que el chunking funciona correctamente. El pipeline maneja automáticamente archivos >25MB, pero si el archivo ya está bajo el límite y es largo (ej: 50 min de audio comprimido en 20MB), la transcripción puede perder contexto en pausas largas.
config = PipelineConfig(
language="es",
vocabulary=["sprint", "deployment"],
auto_preprocess=True
)Problema: Costo inesperadamente alto en batch
Síntoma: Batch de 50 archivos cuesta mucho más de lo esperado.
Solución: Estimar costos antes de ejecutar:
def estimate_batch_cost(audio_paths: list[str]) -> dict:
total_duration = 0
for path in audio_paths:
audio = AudioSegment.from_file(path)
total_duration += len(audio) / 1000 / 60
return {
"files": len(audio_paths),
"total_duration_min": round(total_duration, 1),
"estimated_whisper_cost": round(total_duration * 0.006, 2),
"estimated_llm_cost": round(len(audio_paths) * 0.002, 2),
"estimated_total": round(total_duration * 0.006 + len(audio_paths) * 0.002, 2)
}Checklist de Completitud
Funcionalidad Core
- Validación de audio (formato, tamaño, calidad)
- Preprocesamiento automático (conversión, normalización)
- Transcripción con Whisper (single y chunked)
- Procesamiento con LLM (summarize, action_items, qa, full)
- Generación de audio con TTS (opcional)
- Manejo de errores en cada paso
Calidad
- Tracking de costos por servicio
- Tracking de timing por paso
- Soporte para vocabulario técnico
- Texto preprocesado para TTS (sin markdown, sin bullets)
Extensiones
- Batch processing con ThreadPoolExecutor
- Diarización con AssemblyAI
- Exportar resultados (JSON, Markdown, TXT)
- Estimación de costos pre-ejecución
Testing
- Funciona con audio corto (<1 min)
- Funciona con audio largo (>25MB, chunking)
- Funciona con formato no estándar (conversión)
- Maneja errores gracefully (archivo no encontrado, API error)
- Genera audio de salida correctamente
Ejercicios
Ejercicio 1: Pipeline con caché
Extiende AudioPipeline para cachear transcripciones en disco usando el hash MD5 del archivo de audio como key. Si el archivo ya fue transcrito, no vuelvas a llamar a Whisper.
Ver solución
import hashlib
class CachedAudioPipeline(AudioPipeline):
def __init__(self, config: PipelineConfig = None, cache_dir: str = "transcript_cache"):
super().__init__(config)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _get_cache_key(self, audio_path: str) -> str:
h = hashlib.md5()
with open(audio_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return h.hexdigest()
def run(self, audio_path: str) -> dict:
metrics = PipelineMetrics()
metrics.start_timer("total")
validation = validate_audio(audio_path)
working_path = audio_path
if validation.get("needs_preprocessing") and self.config.auto_preprocess:
prep = preprocess_audio(audio_path, self.config.output_dir)
working_path = prep["output_path"]
cache_key = self._get_cache_key(working_path)
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
cached = json.loads(cache_file.read_text(encoding="utf-8"))
transcript_data = cached
transcript_data["from_cache"] = True
else:
transcript_data = transcribe_audio(working_path, self.config, metrics)
cache_file.write_text(
json.dumps(transcript_data, ensure_ascii=False),
encoding="utf-8"
)
transcript_data["from_cache"] = False
llm_result = process_with_llm(transcript_data["text"], self.config, metrics)
audio_output = None
if self.config.generate_audio:
result_text = self._get_text_for_tts(llm_result)
if result_text:
audio_output = generate_audio_output(result_text, self.config, metrics)
metrics.stop_timer("total")
return {
"status": "success",
"transcript": transcript_data["text"],
"result": llm_result,
"audio_output_path": audio_output["audio_path"] if audio_output else None,
"from_cache": transcript_data.get("from_cache", False),
"metadata": {
"audio_duration_min": transcript_data.get("duration_min"),
"transcript_words": transcript_data.get("word_count"),
"language_detected": transcript_data.get("language"),
"chunks_processed": transcript_data.get("chunks", 1),
"task": self.config.task
},
"costs": metrics.costs,
"timing": {k: v for k, v in metrics.timing.items() if not k.startswith("_")}
}Ejercicio 2: Pipeline con reporte comparativo
Crea una función que ejecute el mismo audio con diferentes configuraciones (summarize, action_items, full) y genere un reporte comparativo de costos y resultados.
Ver solución
def compare_pipeline_tasks(audio_path: str) -> dict:
tasks = ["summarize", "action_items", "full"]
results = {}
for task in tasks:
config = PipelineConfig(task=task, language="es")
pipeline = AudioPipeline(config)
result = pipeline.run(audio_path)
results[task] = {
"status": result["status"],
"result": result.get("result"),
"cost": result.get("costs", {}).get("total", 0),
"time": result.get("timing", {}).get("total", 0),
"transcript_words": result.get("metadata", {}).get("transcript_words", 0)
}
comparison = {
"results": results,
"cheapest_task": min(results.items(), key=lambda x: x[1]["cost"])[0],
"fastest_task": min(results.items(), key=lambda x: x[1]["time"])[0],
"total_cost": round(sum(r["cost"] for r in results.values()), 4),
"total_time": round(sum(r["time"] for r in results.values()), 2)
}
return comparisonResumen
- El Audio Pipeline integra validación → preprocesamiento → transcripción → LLM → TTS en un sistema configurable.
- Soporta 4 tareas: summarize, action_items, qa, full analysis.
- Incluye tracking de costos y timing por cada paso del pipeline.
- Preprocesamiento automático: conversión de formato, normalización de audio, manejo de archivos grandes.
- Extensiones: batch processing, diarización con AssemblyAI, exportación multi-formato.
- El pipeline es el componente de audio del Document Analyzer del Módulo 8.
- Costo típico: ~$0.10-0.20 para 15 minutos de audio con resumen y TTS.
Próximo módulo: Módulo 6 — RAG Multimodal. Ya procesas texto, imágenes y audio; ahora construirás un sistema de búsqueda inteligente que combina todas las modalidades para responder preguntas sobre documentos con figuras.