Image for post Transcripción de Audio con OpenAI Whisper y FastAPI: Construyendo una API Inteligente y Escalable

Transcripción de Audio con OpenAI Whisper y FastAPI: Construyendo una API Inteligente y Escalable

Escrito por Sergio Márquez

La capacidad de convertir voz a texto es una funcionalidad fundamental en muchas aplicaciones modernas. Desde asistentes virtuales y transcripción de reuniones hasta análisis de contenido multimedia y herramientas de accesibilidad, la demanda de soluciones precisas y eficientes es constante. Tradicionalmente, desarrollar un sistema de transcripción de voz ha sido un desafío complejo, pero modelos avanzados como OpenAI Whisper han democratizado esta capacidad, poniéndola al alcance de los desarrolladores.

Como desarrolladores, necesitamos integrar estas potentes capacidades en nuestras aplicaciones de manera eficiente y robusta. Este artículo te guiará paso a paso para construir una API inteligente con FastAPI que utilice OpenAI Whisper para transcribir archivos de audio, enfocándonos en la escalabilidad y la buena gestión de recursos. Aprenderás a manejar la subida de archivos de forma asíncrona, a interactuar con la API de Whisper y a utilizar las tareas en segundo plano de FastAPI para mantener tu API responsiva.

Contexto del Problema

Imagina que necesitas procesar grabaciones de reuniones, podcasts o notas de voz para extraer texto. Un enfoque síncrono, donde el cliente espera a que la transcripción se complete, puede llevar a tiempos de espera inaceptables, especialmente con archivos de audio largos. Esto no solo degrada la experiencia del usuario, sino que también puede bloquear los recursos del servidor. La solución ideal es una API que acepte el archivo de audio, confirme su recepción y luego procese la transcripción en segundo plano, permitiendo al cliente consultar el resultado más tarde.

Conceptos Clave

  • OpenAI Whisper: Es un modelo de IA de código abierto (y disponible a través de la API de OpenAI) entrenado en una vasta cantidad de datos de audio multilingües. Es capaz de transcribir voz a texto con alta precisión y detectar el idioma automáticamente. La API de Whisper soporta varios formatos de audio como mp3, mp4, mpeg, mpga, m4a, wav y webm, con un límite de tamaño de archivo de 25 MB por defecto. [1, 2, 13, 14]
  • FastAPI: Un framework web moderno y rápido para construir APIs con Python. Es conocido por su rendimiento, facilidad de uso y soporte nativo para programación asíncrona, lo que lo hace ideal para aplicaciones de IA.
  • Programación Asíncrona (AsyncIO): Permite que tu aplicación maneje múltiples operaciones de E/S (entrada/salida), como subir archivos o hacer llamadas a APIs externas, sin bloquear el hilo principal. Esto mejora significativamente la capacidad de respuesta y la concurrencia de tu API.
  • Background Tasks en FastAPI: Una característica crucial que permite ejecutar funciones en segundo plano después de que la API ha enviado una respuesta al cliente. Esto es perfecto para tareas que consumen tiempo, como la transcripción de audio, ya que evita que el cliente tenga que esperar a que la operación finalice. [4, 5, 10]
  • Manejo de Archivos con UploadFile y aiofiles: FastAPI utiliza la clase UploadFile para manejar archivos subidos, la cual es un wrapper alrededor de SpooledTemporaryFile. Esto significa que los archivos se almacenan en memoria hasta un cierto umbral y luego se escriben en disco, optimizando el uso de recursos. Para operaciones de escritura y lectura de archivos de forma asíncrona, utilizaremos la librería aiofiles. [17, 20, 23]

Implementación Paso a Paso

1. Configuración del Entorno

Primero, crea un entorno virtual y activa. Luego, instala las dependencias necesarias:


pip install fastapi uvicorn openai python-dotenv aiofiles

Crea un archivo .env en la raíz de tu proyecto para almacenar tu API Key de OpenAI de forma segura:


OPENAI_API_KEY="tu_api_key_de_openai_aqui"

¡Importante! Nunca expongas tu API Key directamente en el código fuente ni la subas a repositorios públicos. Usa siempre variables de entorno. [14, 21]

2. Estructura Básica de la API con FastAPI

Crearemos un archivo main.py. Necesitaremos importar FastAPI, UploadFile, File, BackgroundTasks y otras utilidades.


import os
import uuid
import shutil
from typing import Dict

import aiofiles
from fastapi import FastAPI, UploadFile, File, BackgroundTasks, HTTPException, status
from openai import OpenAI
from dotenv import load_dotenv

# Cargar variables de entorno
load_dotenv()

app = FastAPI()

# Configuración de OpenAI
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("La variable de entorno OPENAI_API_KEY no está configurada.")

client = OpenAI(api_key=OPENAI_API_KEY)

# Directorio para guardar archivos temporales
UPLOAD_DIR = "./temp_audio_uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

# Diccionario para almacenar el estado de las tareas de transcripción
transcription_tasks: Dict[str, Dict] = {}

# Formatos de audio soportados por OpenAI Whisper
SUPPORTED_AUDIO_TYPES = [
    "audio/mpeg", # mp3
    "audio/mp4",  # mp4, m4a
    "audio/wav",
    "audio/webm",
    "audio/ogg",
    "video/mp4",  # para archivos mp4 que contienen audio
    "video/webm"  # para archivos webm que contienen audio
]

MAX_FILE_SIZE_MB = 25 # Límite de 25MB para la API de Whisper [1, 2]

async def transcribe_audio_task(task_id: str, file_path: str):
    """Tarea de fondo para transcribir audio con OpenAI Whisper."""
    try:
        with open(file_path, "rb") as audio_file:
            transcription = client.audio.transcriptions.create(
                model="whisper-1",
                file=audio_file
            )
        transcription_tasks[task_id]["status"] = "completed"
        transcription_tasks[task_id]["result"] = transcription.text
    except Exception as e:
        transcription_tasks[task_id]["status"] = "failed"
        transcription_tasks[task_id]["error"] = str(e)
    finally:
        # Limpiar el archivo temporal después de la transcripción
        if os.path.exists(file_path):
            os.remove(file_path)

@app.post("/transcribe/")
async def upload_audio_for_transcription(
    file: UploadFile = File(..., description=f"Archivo de audio para transcribir (máx. {MAX_FILE_SIZE_MB}MB)"),
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    """Sube un archivo de audio y programa su transcripción con OpenAI Whisper.

    Retorna un ID de tarea para consultar el estado de la transcripción.
    """
    # 1. Validación del tipo de archivo
    if file.content_type not in SUPPORTED_AUDIO_TYPES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Tipo de archivo no soportado. Tipos permitidos: {', '.join(SUPPORTED_AUDIO_TYPES)}"
        )

    # 2. Generar un nombre de archivo único y ruta temporal
    file_extension = file.filename.split(".")[-1] if "." in file.filename else "tmp"
    unique_filename = f"{uuid.uuid4()}.{file_extension}"
    temp_file_path = os.path.join(UPLOAD_DIR, unique_filename)

    # 3. Guardar el archivo de forma asíncrona y validar tamaño
    file_size = 0
    try:
        async with aiofiles.open(temp_file_path, "wb") as out_file:
            while content := await file.read(1024 * 1024): # Leer en chunks de 1MB
                file_size += len(content)
                if file_size > MAX_FILE_SIZE_MB * 1024 * 1024:
                    raise HTTPException(
                        status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                        detail=f"El archivo excede el límite de {MAX_FILE_SIZE_MB}MB para la API de Whisper."
                    )
                await out_file.write(content)
    except HTTPException as e:
        # Si hay un error de tamaño, eliminar el archivo parcial
        if os.path.exists(temp_file_path):
            os.remove(temp_file_path)
        raise e
    except Exception as e:
        if os.path.exists(temp_file_path):
            os.remove(temp_file_path)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Error al guardar el archivo: {str(e)}"
        )

    # 4. Crear una tarea de transcripción y añadirla a las tareas de fondo
    task_id = str(uuid.uuid4())
    transcription_tasks[task_id] = {
        "status": "pending",
        "filename": file.filename,
        "file_path": temp_file_path,
        "result": None,
        "error": None
    }
    background_tasks.add_task(transcribe_audio_task, task_id, temp_file_path)

    return {"message": "Transcripción iniciada", "task_id": task_id}

@app.get("/transcription-status/{task_id}")
async def get_transcription_status(task_id: str):
    """Consulta el estado y el resultado de una tarea de transcripción."""
    task = transcription_tasks.get(task_id)
    if not task:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="ID de tarea no encontrado.")

    if task["status"] == "completed":
        # Una vez completada, se puede eliminar la tarea del diccionario si no se necesita persistencia
        # del historial de tareas en memoria.
        # del transcription_tasks[task_id]
        return {"status": task["status"], "result": task["result"]}
    elif task["status"] == "failed":
        return {"status": task["status"], "error": task["error"]}
    else:
        return {"status": task["status"], "message": "Transcripción en progreso o pendiente."}

@app.get("/health")
async def health_check():
    return {"status": "ok", "message": "API is running"}

3. Ejecutar la Aplicación

Guarda el código anterior como main.py y ejecuta tu aplicación FastAPI con Uvicorn:


uvicorn main:app --reload

Ahora puedes acceder a la documentación interactiva de tu API en http://127.0.0.1:8000/docs.

Mini Proyecto / Aplicación Sencilla

El código proporcionado ya implementa una aplicación sencilla con dos endpoints clave:

  • POST /transcribe/: Permite subir un archivo de audio. La API lo valida, lo guarda temporalmente y luego inicia la transcripción con OpenAI Whisper en segundo plano. Inmediatamente devuelve un task_id al cliente, sin hacerlo esperar por la transcripción completa.
  • GET /transcription-status/{task_id}: Con este endpoint, el cliente puede usar el task_id recibido para consultar el estado de su transcripción. Una vez que la tarea se completa, recibirá el texto transcrito.

Probando la API

Puedes usar la interfaz de Swagger UI (http://127.0.0.1:8000/docs) para probar los endpoints. Para el endpoint /transcribe/, selecciona un archivo de audio (.mp3, .wav, etc.) de menos de 25MB. Después de enviarlo, recibirás un task_id. Luego, usa ese task_id en el endpoint /transcription-status/{task_id} para ver el progreso y el resultado.

Alternativamente, puedes usar curl o una librería HTTP como httpx en Python:


import httpx
import asyncio

async def test_transcription_api():
    audio_file_path = "./sample_audio.mp3" # Asegúrate de tener un archivo de audio aquí

    # Crear un archivo de audio de ejemplo si no existe (solo para demostración)
    # En un entorno real, usarías un archivo existente.
    if not os.path.exists(audio_file_path):
        print(f"Creando archivo de audio de ejemplo: {audio_file_path}")
        # Esto es solo un placeholder, en la realidad necesitarías un archivo de audio real.
        # Puedes grabar tu voz o descargar un pequeño mp3.
        with open(audio_file_path, "wb") as f:
            f.write(b"Simulated audio content") # Esto NO es un archivo de audio real
        print("¡ADVERTENCIA! El archivo de audio de ejemplo no es un MP3 válido. Por favor, reemplázalo con un archivo MP3 real para probar la transcripción.")

    async with httpx.AsyncClient() as client:
        print(f"Subiendo archivo {audio_file_path}...")
        with open(audio_file_path, "rb") as f:
            files = {'file': (os.path.basename(audio_file_path), f, 'audio/mpeg')}
            response = await client.post("http://127.0.0.1:8000/transcribe/", files=files, timeout=60.0)

        print(f"Respuesta de subida: {response.status_code} - {response.json()}")

        if response.status_code == 200:
            task_id = response.json().get("task_id")
            if task_id:
                print(f"Transcripción iniciada. ID de tarea: {task_id}")
                while True:
                    await asyncio.sleep(5) # Esperar 5 segundos antes de consultar de nuevo
                    status_response = await client.get(f"http://127.0.0.1:8000/transcription-status/{task_id}")
                    status_data = status_response.json()
                    print(f"Estado de la tarea {task_id}: {status_data}")

                    if status_data.get("status") in ["completed", "failed"]:
                        break
            else:
                print("No se recibió un task_id.")
        else:
            print("Error al iniciar la transcripción.")

# Para ejecutar la función de prueba:
# asyncio.run(test_transcription_api())

Nota: Para que el script de prueba funcione correctamente, debes reemplazar "./sample_audio.mp3" con la ruta a un archivo de audio MP3 real y válido en tu sistema. El contenido simulado b"Simulated audio content" no es un MP3 funcional y solo sirve para evitar errores de archivo no encontrado.

Errores Comunes y Depuración

  • ValueError: OPENAI_API_KEY no está configurada.: Asegúrate de que tu archivo .env esté en la raíz del proyecto y contenga OPENAI_API_KEY="tu_clave", y que load_dotenv() se ejecute correctamente.
  • HTTPException 400: Tipo de archivo no soportado.: Verifica que el archivo que estás subiendo sea uno de los formatos soportados por OpenAI Whisper (mp3, wav, mp4, etc.) y que el Content-Type de la petición sea correcto. [2, 16]
  • HTTPException 413: El archivo excede el límite de 25MB.: La API de Whisper tiene un límite de 25MB por archivo. Si necesitas transcribir archivos más grandes, deberás dividirlos en segmentos más pequeños antes de subirlos. [1, 6]
  • openai.AuthenticationError: Tu API Key de OpenAI es inválida o ha caducado. Revisa tu clave en el panel de OpenAI.
  • Archivos temporales no eliminados: Aunque la tarea de fondo incluye una limpieza, en caso de fallos inesperados, los archivos podrían persistir. Considera implementar una tarea de limpieza periódica para el directorio UPLOAD_DIR.
  • Errores de red/conexión: Las llamadas a APIs externas pueden fallar. Implementa reintentos (retries) o un manejo de errores más sofisticado para la llamada a client.audio.transcriptions.create.

Aprendizaje Futuro

Esta implementación es un excelente punto de partida, pero hay muchas formas de expandirla y mejorarla para un entorno de producción:

  • Streaming de Audio: Explorar cómo transcribir audio en tiempo real a medida que se recibe, aunque la API de Whisper actualmente procesa archivos completos. Esto podría implicar el uso de otras librerías o servicios de streaming de voz. [2]
  • Persistencia de Tareas: Actualmente, el estado de las tareas se guarda en un diccionario en memoria (transcription_tasks). Para una aplicación robusta, deberías persistir esta información en una base de datos (SQL, NoSQL) o un caché (Redis) para que las tareas sobrevivan a reinicios del servidor.
  • Colas de Mensajes: Para un procesamiento de tareas en segundo plano más robusto y escalable, integra una cola de mensajes como Celery con RabbitMQ o Redis. Esto desacoplará la recepción de la petición del procesamiento de la transcripción, permitiendo manejar picos de carga.
  • Procesamiento de Audio Avanzado: Implementar funcionalidades como detección de oradores (diarización), segmentación de audio o eliminación de ruido antes de la transcripción.
  • Despliegue en Producción: Dockerizar la aplicación y desplegarla en plataformas en la nube como Railway, Render, Google Cloud Run, AWS ECS o Kubernetes. Asegúrate de configurar correctamente las variables de entorno y los volúmenes para los archivos temporales. [23]
  • Manejo de Archivos Grandes: Para archivos de audio que exceden los 25MB, implementa una lógica para dividirlos en segmentos más pequeños, transcribir cada segmento y luego unir las transcripciones. Librerías como pydub pueden ser útiles para esto. [6]

Con esta guía, tienes las herramientas y el conocimiento para construir tu propia API de transcripción de audio con FastAPI y OpenAI Whisper, abriendo un mundo de posibilidades para tus aplicaciones de IA.

Volver al blog