Módulo 1: Introducción a IA Multimodal
7. Troubleshooting Multimodal
Descripción
Trabajar con APIs multimodales implica errores que no existen en APIs de solo texto: imágenes con formato incorrecto, audios demasiado grandes, codificación Base64 corrupta, límites de tokens invisibles y costos que se disparan sin aviso. Esta cápsula te da un framework para diagnosticar, prevenir y resolver los errores más comunes.
Por qué importa: En producción, un error no manejado tumba toda la cadena. Si tu sistema recibe una imagen TIFF (no soportada), falla silenciosamente. Si un audio excede 25MB, Whisper lo rechaza. Si no controlas reintentos, un rate limit genera cientos de requests fallidos. Saber troubleshootear separa un prototipo frágil de un sistema robusto.
Conexión con el módulo: Las cápsulas 02-06 te enseñaron a usar vision, audio, combinaciones, el landscape y los formatos. Esta cápsula cierra el ciclo: qué hacer cuando algo falla. Los patrones que aprendas aquí los usarás directamente en el Clasificador multimodal (cápsula 08) y en cada módulo posterior.
Categoría 1: Errores de Formato y Encoding
"Invalid image format"
El proveedor rechaza la imagen porque el formato no está soportado (TIFF, BMP, HEIC) o el archivo está corrupto.
from PIL import Image
from pathlib import Path
import io
SUPPORTED_FORMATS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
def convert_image_format(image_path: str, target_format: str = "PNG") -> bytes:
"""Convierte cualquier imagen a un formato soportado por las APIs."""
with Image.open(image_path) as img:
if img.mode in ("RGBA", "LA", "P") and target_format == "JPEG":
img = img.convert("RGB")
buffer = io.BytesIO()
img.save(buffer, format=target_format)
return buffer.getvalue()
# Uso: convertir un .tiff no soportado a PNG
png_bytes = convert_image_format("documento.tiff", "PNG")
Errores de Base64
Un Base64 mal codificado produce errores crípticos como Invalid base64 o respuestas vacías.
import base64
def safe_base64_encode(file_path: str) -> str:
"""Codifica archivo a Base64 con validación de integridad."""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"Archivo no encontrado: {file_path}")
with open(path, "rb") as f:
raw_bytes = f.read()
if len(raw_bytes) == 0:
raise ValueError("El archivo está vacío")
encoded = base64.b64encode(raw_bytes).decode("utf-8")
decoded = base64.b64decode(encoded)
assert len(decoded) == len(raw_bytes), "Longitud no coincide post-decode"
return encoded
MIME type incorrecto
Enviar image/png cuando el archivo es JPEG causa rechazos silenciosos en Anthropic.
import mimetypes
MIME_MAP = {
".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".gif": "image/gif", ".webp": "image/webp",
".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4",
}
def detect_mime_type(file_path: str) -> str:
"""Detecta el MIME type real del archivo."""
ext = Path(file_path).suffix.lower()
if ext in MIME_MAP:
return MIME_MAP[ext]
mime, _ = mimetypes.guess_type(file_path)
if mime:
return mime
raise ValueError(f"No se pudo determinar MIME type para: {file_path}")
Audio con formato no soportado
Whisper solo acepta: mp3, mp4, mpeg, mpga, m4a, wav, webm.
from pydub import AudioSegment
WHISPER_FORMATS = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm"}
def convert_audio_for_whisper(audio_path: str, output_format: str = "mp3") -> str:
"""Convierte audio a formato compatible con Whisper."""
path = Path(audio_path)
if path.suffix.lower() in WHISPER_FORMATS:
return audio_path
audio = AudioSegment.from_file(str(path))
output_path = str(path.with_suffix(f".{output_format}"))
audio.export(output_path, format=output_format)
return output_path
Categoría 2: Errores de Tamaño y Límites
Imagen demasiado grande (>20MB)
OpenAI/Gemini rechazan imágenes >20MB. Anthropic: >5MB.
from PIL import Image
import io
def resize_image_to_limit(image_path: str, max_size_mb: float = 4.0) -> bytes:
"""Reduce calidad y dimensiones progresivamente hasta cumplir el límite."""
with Image.open(image_path) as img:
if img.mode != "RGB":
img = img.convert("RGB")
quality = 95
current = img.copy()
while True:
buffer = io.BytesIO()
current.save(buffer, format="JPEG", quality=quality, optimize=True)
if len(buffer.getvalue()) / (1024 * 1024) <= max_size_mb:
return buffer.getvalue()
if quality > 40:
quality -= 10
else:
w, h = current.size
current = current.resize((w // 2, h // 2), Image.LANCZOS)
quality = 85
if current.size[0] < 100:
raise ValueError("No se pudo reducir sin perder calidad útil")
Audio demasiado largo (>25MB)
Para archivos largos, divide en segmentos antes de transcribir.
from pydub import AudioSegment
import math
from openai import OpenAI
def split_audio(audio_path: str, max_size_mb: float = 24.0) -> list[str]:
"""Divide audio en chunks que no excedan el límite de Whisper."""
audio = AudioSegment.from_file(audio_path)
file_size_mb = Path(audio_path).stat().st_size / (1024 * 1024)
if file_size_mb <= max_size_mb:
return [audio_path]
num_chunks = math.ceil(file_size_mb / max_size_mb)
chunk_ms = len(audio) // num_chunks
paths = []
for i in range(num_chunks):
chunk = audio[i * chunk_ms : min((i + 1) * chunk_ms, len(audio))]
path = f"./chunks/chunk_{i:03d}.mp3"
Path("./chunks").mkdir(exist_ok=True)
chunk.export(path, format="mp3")
paths.append(path)
return paths
def transcribe_long_audio(audio_path: str) -> str:
"""Transcribe audio largo dividiéndolo en chunks."""
client = OpenAI()
chunks = split_audio(audio_path)
texts = []
for chunk_path in chunks:
with open(chunk_path, "rb") as f:
result = client.audio.transcriptions.create(
model="whisper-1", file=f, language="es"
)
texts.append(result.text)
return " ".join(texts)
Context window excedido
Cuando envías múltiples imágenes, puedes exceder el context window sin darte cuenta.
import math
def estimate_image_tokens(width: int, height: int, detail: str = "auto") -> int:
"""Estima tokens de una imagen en OpenAI (low=85 fijo, high=170 base + 85/tile)."""
if detail == "low":
return 85
scale = min(2048 / max(width, height), 1.0)
sw, sh = int(width * scale), int(height * scale)
short = min(sw, sh)
if short > 768:
ratio = 768 / short
sw, sh = int(sw * ratio), int(sh * ratio)
return 170 + 85 * math.ceil(sw / 512) * math.ceil(sh / 512)
def check_context_budget(images: list[dict], text_tokens: int, max_ctx: int = 128_000) -> dict:
"""Verifica si un request cabe en el context window."""
img_tokens = sum(estimate_image_tokens(i["w"], i["h"], i.get("detail", "auto")) for i in images)
total = text_tokens + img_tokens
return {"fits": total <= max_ctx, "total": total, "remaining": max_ctx - total}
Categoría 3: Errores de API y Red
Rate limit → exponential backoff
import time
import random
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError
def safe_api_call(func, *args, max_retries: int = 5, base_delay: float = 1.0, **kwargs):
"""Wrapper con retry y backoff exponencial para cualquier llamada API."""
last_error = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except RateLimitError as e:
last_error = e
delay = min(base_delay * (2 ** attempt), 60) + random.uniform(0, 1)
print(f" [Rate limit] Intento {attempt+1}/{max_retries}, esperando {delay:.1f}s")
time.sleep(delay)
except APITimeoutError as e:
last_error = e
time.sleep(base_delay * (2 ** attempt))
except APIConnectionError as e:
last_error = e
time.sleep(base_delay * 2)
except Exception:
raise
raise last_error
# Uso
client = OpenAI()
response = safe_api_call(
client.chat.completions.create,
model="gpt-4o",
messages=[{"role": "user", "content": "Hola"}],
max_tokens=100,
)
Autenticación (401)
from openai import OpenAI, AuthenticationError
def validate_api_key(api_key: str | None = None) -> dict:
"""Valida que la API key de OpenAI sea funcional."""
try:
client = OpenAI(api_key=api_key) if api_key else OpenAI()
client.models.list()
return {"valid": True, "error": None}
except AuthenticationError as e:
return {"valid": False, "error": f"API key inválida: {e}"}
except Exception as e:
return {"valid": False, "error": f"Error inesperado: {e}"}
Errores 500 → fallback
from openai import InternalServerError
def call_with_fallback(primary_func, fallback_func, *args, **kwargs):
"""Intenta con primario; si falla con 500, usa fallback."""
try:
return safe_api_call(primary_func, *args, max_retries=3, **kwargs)
except InternalServerError:
print(" [Fallback] Servidor primario con errores. Usando fallback...")
return safe_api_call(fallback_func, *args, max_retries=3, **kwargs)
Categoría 4: Errores de Calidad
Los errores de calidad no lanzan excepciones — el request funciona, pero el resultado es malo.
OCR impreciso → preprocesamiento
from PIL import Image, ImageEnhance, ImageFilter
import io
def preprocess_for_ocr(image_path: str) -> bytes:
"""Preprocesa imagen para maximizar precisión del OCR."""
with Image.open(image_path) as img:
img = img.convert("L")
if img.width < 1000 or img.height < 1000:
scale = max(1000 / img.width, 1000 / img.height)
img = img.resize((int(img.width * scale), int(img.height * scale)), Image.LANCZOS)
img = ImageEnhance.Contrast(img).enhance(2.0)
img = img.filter(ImageFilter.SHARPEN)
img = img.point(lambda x: 255 if x > 128 else 0)
buffer = io.BytesIO()
img.save(buffer, format="PNG")
return buffer.getvalue()
Transcripción imprecisa → language hints + prompt
from openai import OpenAI
def transcribe_with_hints(audio_path: str, language: str = "es", context: str = "") -> str:
"""Transcribe con hints de idioma y contexto para mayor precisión.
El prompt le da vocabulario esperado: nombres, acrónimos, términos técnicos.
"""
client = OpenAI()
with open(audio_path, "rb") as f:
result = client.audio.transcriptions.create(
model="whisper-1", file=f, language=language,
prompt=context, temperature=0.0,
)
return result.text
# Uso
transcript = transcribe_with_hints(
"reunion.mp3", language="es",
context="Reunión sobre deployment de FastAPI con Docker. Participantes: María, Carlos.",
)
Clasificación incorrecta → prompt + temperature=0
from openai import OpenAI
import base64
def classify_with_precision(image_path: str, categories: list[str]) -> dict:
"""Clasifica imagen con prompt optimizado para consistencia."""
client = OpenAI()
with open(image_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
cats = ", ".join(categories)
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": [
{"type": "text", "text": f"Clasifica esta imagen en EXACTAMENTE una categoría: {cats}. Responde SOLO el nombre."},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
]}],
max_tokens=50, temperature=0,
)
result = response.choices[0].message.content.strip()
return {"category": result, "valid": result in categories}
Alucinaciones → pedir nivel de confianza
from openai import OpenAI
import base64, json
def extract_with_verification(image_path: str, fields: list[str]) -> dict:
"""Extrae datos pidiendo confianza por campo para detectar alucinaciones."""
client = OpenAI()
with open(image_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
fields_str = ", ".join(fields)
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": [
{"type": "text", "text": (
f"Extrae estos campos de la imagen: {fields_str}.\n"
f"Responde JSON: {{\"campo\": {{\"value\": \"...\", \"confidence\": \"high|medium|low\", "
f"\"source\": \"visible|inferido|no encontrado\"}}}}.\n"
f"Si un campo NO es visible, value=null y confidence=\"low\". NO inventes."
)},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
]}],
max_tokens=1000, temperature=0,
)
raw = response.choices[0].message.content
try:
clean = raw.split("```json")[-1].split("```")[0] if "```" in raw else raw
return json.loads(clean.strip())
except json.JSONDecodeError:
return {"raw_response": raw, "parse_error": True}
Categoría 5: Errores de Costos
Cost tracker con alertas
import time
import functools
class CostTracker:
"""Rastrea costos acumulados de llamadas API multimodales."""
PRICING = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"whisper-1": {"per_minute": 0.006},
"tts-1": {"per_1k_chars": 0.015},
"dall-e-3": {"per_image": 0.040},
}
def __init__(self, budget_limit: float = 1.0):
self.total_cost = 0.0
self.budget_limit = budget_limit
self.calls = []
def log_call(self, model: str, **kwargs) -> float:
cost = self._estimate(model, **kwargs)
self.total_cost += cost
self.calls.append({"model": model, "cost": cost, "time": time.time(), **kwargs})
if self.total_cost > self.budget_limit:
print(f" ALERTA: ${self.total_cost:.4f} excede presupuesto ${self.budget_limit:.2f}")
return cost
def _estimate(self, model: str, **kw) -> float:
p = self.PRICING.get(model, {})
if "input_tokens" in kw:
return (kw["input_tokens"]/1e6)*p.get("input",0) + (kw.get("output_tokens",0)/1e6)*p.get("output",0)
if "audio_minutes" in kw:
return kw["audio_minutes"] * p.get("per_minute", 0)
if "characters" in kw:
return (kw["characters"]/1000) * p.get("per_1k_chars", 0)
return kw.get("images_generated", 0) * p.get("per_image", 0)
def summary(self) -> dict:
return {"total": round(self.total_cost, 6), "calls": len(self.calls),
"remaining": round(self.budget_limit - self.total_cost, 6)}
Decorador de monitoreo
def track_cost(model: str, tracker: CostTracker):
"""Decorador que registra costo de cada llamada automáticamente."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
cost_kw = {}
if hasattr(result, "usage"):
cost_kw["input_tokens"] = result.usage.prompt_tokens
cost_kw["output_tokens"] = result.usage.completion_tokens
tracker.log_call(model, **cost_kw)
return result
return wrapper
return decorator
# Uso
tracker = CostTracker(budget_limit=0.50)
@track_cost("gpt-4o", tracker)
def analyze_image(image_path: str):
client = OpenAI()
with open(image_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
return client.chat.completions.create(
model="gpt-4o", max_tokens=300,
messages=[{"role": "user", "content": [
{"type": "text", "text": "Describe esta imagen."},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
]}],
)
Función de Diagnóstico Integral
from pathlib import Path
from PIL import Image
import os
def diagnose_multimodal_input(file_path: str) -> dict:
"""Diagnostica problemas comunes antes de enviar a una API."""
path = Path(file_path)
issues, suggestions = [], []
if not path.exists():
return {"valid": False, "issues": ["Archivo no encontrado"],
"suggestions": [f"Verifica la ruta: {file_path}"]}
size_mb = path.stat().st_size / (1024 * 1024)
if size_mb == 0:
issues.append("Archivo vacío (0 bytes)")
elif size_mb > 20:
issues.append(f"Archivo muy grande: {size_mb:.1f}MB")
suggestions.append("Usa resize_image_to_limit() o split_audio()")
ext = path.suffix.lower()
img_exts = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
audio_exts = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm"}
if ext not in (img_exts | audio_exts):
issues.append(f"Formato no soportado: {ext}")
suggestions.append("Convierte a PNG/JPEG (imagen) o MP3/WAV (audio)")
if ext in img_exts:
try:
with Image.open(path) as img:
img.verify()
with Image.open(path) as img:
w, h = img.size
if w < 50 or h < 50:
issues.append(f"Imagen muy pequeña: {w}x{h}")
except Exception as e:
issues.append(f"Imagen corrupta: {e}")
if size_mb > 5:
suggestions.append("Para Anthropic, reduce a <5MB")
if ext in audio_exts and size_mb > 25:
issues.append(f"Audio excede 25MB de Whisper: {size_mb:.1f}MB")
suggestions.append("Usa split_audio() para dividir en chunks")
if not os.environ.get("OPENAI_API_KEY"):
issues.append("OPENAI_API_KEY no configurada")
return {"valid": len(issues) == 0, "size_mb": round(size_mb, 2),
"format": ext, "issues": issues, "suggestions": suggestions}
Pipeline: Preprocesamiento de Imagen por Proveedor
from PIL import Image
import base64, io
from pathlib import Path
def prepare_image_pipeline(image_path: str, provider: str = "openai") -> dict:
"""Pipeline completa: valida → convierte → redimensiona → codifica → payload."""
path = Path(image_path)
if not path.exists():
raise FileNotFoundError(image_path)
limits = {
"openai": {"max_mb": 20, "fmts": {".png",".jpg",".jpeg",".gif",".webp"}},
"anthropic": {"max_mb": 5, "fmts": {".png",".jpg",".jpeg",".gif",".webp"}},
"google": {"max_mb": 20, "fmts": {".png",".jpg",".jpeg",".gif",".webp"}},
}
cfg = limits.get(provider, limits["openai"])
ext = path.suffix.lower()
if ext not in cfg["fmts"]:
img_bytes = convert_image_format(image_path, "PNG")
mime = "image/png"
else:
with open(path, "rb") as f:
img_bytes = f.read()
mime = detect_mime_type(image_path)
if len(img_bytes) / (1024*1024) > cfg["max_mb"]:
img_bytes = resize_image_to_limit(image_path, cfg["max_mb"] * 0.9)
mime = "image/jpeg"
b64 = base64.b64encode(img_bytes).decode()
payloads = {
"openai": {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
"anthropic": {"type": "image", "source": {"type": "base64", "media_type": mime, "data": b64}},
"google": {"mime_type": mime, "data": b64},
}
return {"payload": payloads[provider], "provider": provider,
"final_size_mb": round(len(img_bytes)/(1024*1024), 2), "mime": mime}
Referencia Rápida
| Error | Causa | Solución |
|---|---|---|
Invalid image format | TIFF, BMP, HEIC | convert_image_format() → PNG |
Invalid base64 | Encoding corrupto | safe_base64_encode() |
File too large | >20MB img, >25MB audio | resize_image_to_limit(), split_audio() |
Rate limit exceeded | Muchas requests/min | safe_api_call() con backoff |
401 Unauthorized | API key inválida | validate_api_key() |
500 Server Error | Problema del proveedor | Retry + fallback |
Context length exceeded | Demasiados tokens | check_context_budget() |
| OCR impreciso | Baja resolución/contraste | preprocess_for_ocr() |
| Transcripción mala | Sin language hint | transcribe_with_hints() |
| Clasificación errónea | Prompt vago | Prompt específico + temperature=0 |
| Alucinaciones | Datos inventados | extract_with_verification() |
| Costos altos | Sin monitoreo | CostTracker + @track_cost |
Ejercicios
Ejercicio 1: Validador de imágenes (Fácil)
Crea validate_image(path) -> dict que verifique: archivo existe, extensión soportada (PNG, JPEG, GIF, WebP), tamaño < 20MB, PIL puede abrirlo, dimensiones >= 50x50. Retorna valid, issues y metadata.
Ver solución
from PIL import Image
from pathlib import Path
def validate_image(path: str) -> dict:
p = Path(path)
issues, metadata = [], {}
if not p.exists():
return {"valid": False, "issues": ["Archivo no encontrado"], "metadata": {}}
ext = p.suffix.lower()
if ext not in {".png", ".jpg", ".jpeg", ".gif", ".webp"}:
issues.append(f"Formato no soportado: {ext}")
size_mb = p.stat().st_size / (1024 * 1024)
metadata["size_mb"] = round(size_mb, 2)
if size_mb > 20:
issues.append(f"Excede 20MB: {size_mb:.1f}MB")
try:
with Image.open(p) as img:
img.verify()
with Image.open(p) as img:
w, h = img.size
metadata.update({"width": w, "height": h, "format": img.format})
if w < 50 or h < 50:
issues.append(f"Muy pequeña: {w}x{h}")
except Exception as e:
issues.append(f"No se pudo abrir: {e}")
return {"valid": len(issues) == 0, "issues": issues, "metadata": metadata}
Ejercicio 2: Validador de audio (Fácil)
Crea validate_audio(path) -> dict que verifique: archivo existe, extensión compatible con Whisper, tamaño < 25MB. Si excede, incluye cuántos chunks serían necesarios.
Ver solución
from pathlib import Path
import math
def validate_audio(path: str) -> dict:
p = Path(path)
issues, metadata = [], {}
if not p.exists():
return {"valid": False, "issues": ["Archivo no encontrado"], "metadata": {}}
ext = p.suffix.lower()
if ext not in {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm"}:
issues.append(f"Formato no compatible con Whisper: {ext}")
size_mb = p.stat().st_size / (1024 * 1024)
metadata["size_mb"] = round(size_mb, 2)
metadata["format"] = ext
if size_mb > 25:
chunks = math.ceil(size_mb / 24)
issues.append(f"Excede 25MB: {size_mb:.1f}MB")
metadata["chunks_needed"] = chunks
return {"valid": len(issues) == 0, "issues": issues, "metadata": metadata}
Ejercicio 3: Diagnóstico de conectividad API (Medio)
Crea test_api_connectivity() -> dict que pruebe la conectividad con OpenAI para chat, vision y whisper. Para cada servicio reporta: available, latency_ms, error.
Ver solución
from openai import OpenAI
import time
def test_api_connectivity() -> dict:
client = OpenAI()
results = {}
# Test Chat
try:
start = time.time()
client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "test"}],
max_tokens=1,
)
results["chat"] = {"available": True, "latency_ms": round((time.time()-start)*1000)}
except Exception as e:
results["chat"] = {"available": False, "error": str(e)}
# Test Models (verifica modelos disponibles sin consumir tokens)
try:
start = time.time()
models = client.models.list()
model_ids = [m.id for m in models.data]
results["models"] = {
"available": True,
"latency_ms": round((time.time()-start)*1000),
"vision": any("gpt-4o" in m for m in model_ids),
"whisper": any("whisper" in m for m in model_ids),
}
except Exception as e:
results["models"] = {"available": False, "error": str(e)}
return results
Ejercicio 4: Pipeline de preprocesamiento universal (Medio)
Crea preprocess_image(path, provider) -> bytes que abra cualquier imagen (incluyendo TIFF, BMP), convierta a JPEG, redimensione según el límite del proveedor (openai: 20MB, anthropic: 5MB) y retorne bytes listos para Base64.
Ver solución
from PIL import Image
import io
def preprocess_image(path: str, provider: str = "openai") -> bytes:
limits = {"openai": 20, "anthropic": 5, "google": 20}
target_mb = limits.get(provider, 20) * 0.85
with Image.open(path) as img:
if img.mode != "RGB":
img = img.convert("RGB")
quality = 95
current = img.copy()
while True:
buffer = io.BytesIO()
current.save(buffer, format="JPEG", quality=quality, optimize=True)
if len(buffer.getvalue()) / (1024 * 1024) <= target_mb:
return buffer.getvalue()
if quality > 50:
quality -= 10
else:
w, h = current.size
current = current.resize((w*3//4, h*3//4), Image.LANCZOS)
quality = 85
if current.size[0] < 100:
raise ValueError("No se pudo reducir suficiente")
Ejercicio 5: Wrapper completo con logging, retry y fallback (Difícil)
Construye una clase RobustAPIClient que: envuelva OpenAI con retry automático (exponential backoff), registre cada llamada (timestamp, modelo, tokens, costo, error), tenga presupuesto máximo configurable, soporte fallback (gpt-4o → gpt-4o-mini), y exponga analyze_image(path, prompt).
Pistas:
- Usa
@dataclassparaAPILogcon campos: timestamp, model, input_tokens, output_tokens, cost, error - En
_retry_call: capturaRateLimitError,APITimeoutError,InternalServerErrory haz retry con2**attempt + random.uniform(0, 1) - En
analyze_image: si el primary falla después de todos los retries, intenta con el fallback - Antes de cada llamada, verifica que
self.spent < self.budget summary()retorna dict con calls, spent, remaining, errors
Resumen
- Errores de formato son la causa #1 de fallos. Valida formato, MIME type y encoding antes de enviar.
- Errores de tamaño tienen soluciones mecánicas:
resize_image_to_limit()para imágenes,split_audio()para audio. - Errores de API requieren retry con exponential backoff. Nunca hagas retry inmediato.
- Errores de calidad no lanzan excepciones. Combate OCR impreciso con preprocesamiento, transcripciones malas con language hints, y alucinaciones pidiendo nivel de confianza.
- Errores de costos se previenen con monitoreo activo. Usa
CostTrackery establece presupuestos. diagnose_multimodal_input()es tu primera línea de defensa: ejecútala antes de cualquier llamada.safe_api_call()es tu segunda línea: envuelve toda llamada con retry, timeout y logging.
Recursos Adicionales
- OpenAI Error Codes Reference — Catálogo completo de errores de la API con soluciones.
- OpenAI Rate Limits — Documentación sobre rate limits, tiers y estrategias.
- Pillow Documentation — Referencia para manipulación de imágenes en Python.
- pydub — Manipulación de audio: conversión, split, compresión.
- OpenAI Vision Guide — Formatos y límites para modelos de visión.
- Anthropic Vision Docs — Límites específicos de Claude para imágenes.
- Exponential Backoff (Google Cloud) — Patrón de backoff exponencial con jitter.
- Counting Tokens (OpenAI Cookbook) — Estimar tokens y costos.