Traducción inteligente de textos largos: Estrategia con Python y Gemini para lidiar con los límites de tokens

Infografía conceptual sobre traducción automatizada de documentos usando scripts de Python y la IA de Google Gemini


Guía de producción para traductores y desarrolladores: optimización por bloques (batching) y manejo inteligente del delay dinámico de la API de Google.

Quienes trabajamos en el desarrollo de software o en la traducción de documentos masivos (como bases de datos de conocimiento o colecciones de tarjetas de estudio), nos enfrentamos recurrentemente a las restricciones severas de las arquitecturas en la nube. Recientemente, Google unificó sus herramientas: el paquete de Python clásico fue depreciado en favor del nuevo ecosistema integrado google-genai.

Junto a este salto de versión, la capa gratuita para desarrolladores incorporó severas cuotas de seguridad por modelo. Si bien el modelo clásico gemini-2.5-flash impone un límite estricto de alrededor de 24 peticiones totales por día (RPD) en ciertos entornos, existen variantes optimizadas como Gemini 3.1 Flash Lite que extienden este cupo a unas generosas 500 solicitudes diarias (RPD) con 15 peticiones por minuto (RPM). Saber elegir el modelo adecuado dentro de la consola de cuotas diferencia a un script funcional de un bloqueo permanente por el error 429: RESOURCE_EXHAUSTED.

En este artículo de ingeniería aplicada, resolveremos este problema atacando tres frentes esenciales: la modernización del entorno de ejecución, la selección inteligente del modelo con mayor cuota disponible y el rediseño del algoritmo para implementar **procesamiento semántico por bloques (Batching)** acompañado de un **sistema de retardo dinámico**.

⚠️ Alerta de obsolescencia: Si tu terminal arroja errores de tipo AttributeError: module 'google.generativeai' has no attribute 'GenerativeModel' o advertencias críticas sobre el ciclo de vida de Python, estás utilizando un entorno inferior a Python 3.10. El nuevo SDK exige imperativamente entornos de ejecución modernos.

La Estrategia Definitiva: Batching y Smart Retry Delay

Para ganarle a los límites restrictivos de una API gratuita sin recurrir a costosas pasarelas comerciales, el código debe estructurarse bajo tres pilares de optimización:

  1. Mapeo de estructuras complejas: No debemos pasar los archivos estructurales (como arrays de objetos JavaScript o JSON) como si fueran texto plano, ya que la IA corre el riesgo de romper las claves sintácticas. Debemos deserializar los datos en memoria utilizando Python.
  2. Selección de modelos de Alta Cuota (Lite): Utilizar variantes como gemini-3.1-flash-lite nos permite trabajar con bloques más pequeños y precisos (por ejemplo, de a 10 elementos) sin temor a agotar las peticiones diarias disponibles.
  3. Cálculo dinámico de espera: Cuando la API se sature, en lugar de utilizar pausas fijas (que agotan los reintentos inútilmente), analizaremos el cuerpo del error mediante expresiones regulares para extraer la instrucción exacta de recuperación (Please retry in X seconds), pausando el hilo del script el tiempo matemáticamente necesario.

Implementación: Script completo

Para comenzar, debemos asegurar la instalación limpia del SDK unificado de Google en nuestro entorno virtual:

pip install google-genai

A continuación, presentamos la solución de código corregida y adaptada. Este script lee un archivo de tarjetas con estructura de datos, procesa los campos de forma segura agrupados por bloques utilizando el modelo Lite e imprime simultáneamente un registro histórico detallado de la terminal en un archivo físico .txt:

import json
import re
import time
import sys
from google import genai
from google.genai import types
from google.genai.errors import APIError

# Configuración del sistema de Logs automático requerido para auditorías
class Logger(object):
    def __init__(self, archivo_log="registro_traduccion.txt"):
        self.terminal = sys.stdout
        self.log = open(archivo_log, "w", encoding="utf-8")

    def write(self, message):
        self.terminal.write(message)
        self.log.write(message)
        self.log.flush()

    def flush(self):
        self.terminal.flush()
        self.log.flush()

# Redireccionamos la salida estándar al archivo y a la pantalla
sys.stdout = Logger()

# Inicializa el cliente moderno del estándar unificado google-genai
client = genai.Client()

SYSTEM_PROMPT = (
    "Actuás como un traductor profesional nativo. Traducí el siguiente bloque de datos en formato JSON "
    "del portugués al español. Mantené un tono formal, teológico y social adecuado para un catecismo. "
    "Es CRÍTICO que devuelvas ÚNICAMENTE la estructura JSON válida traducida, respetando los mismos IDs. "
    "No agregues texto de introducción, ni resúmenes, ni bloques de código markdown."
)

def extraer_json_de_js(ruta_archivo):
    """Extrae y parsea el array JSON interno oculto en un archivo JavaScript."""
    with open(ruta_archivo, 'r', encoding='utf-8') as f:
        contenido = f.read()
    
    match = re.search(r'=\s*(\s*\[.*\])\s*;?\s*$', contenido, re.DOTALL)
    if not match:
        match = re.search(r'\[.*\]', contenido, re.DOTALL)
        
    if match:
        json_texto = match.group(0) if match.group(0).startswith('[') else match.group(1)
        return json.loads(json_texto)
    else:
        raise ValueError("No se pudo mapear una estructura de datos válida.")

def extraer_tiempo_espera(mensaje_error):
    """Analiza de manera dinámica el error de cuota para extraer los segundos solicitados por Google."""
    match = re.search(r'Please retry in ([0-9.]+)\s*s', mensaje_error, re.IGNORECASE)
    if match:
        return float(match.group(1))
    
    match_delay = re.search(r"'retryDelay':\s*'([0-9.]+)\s*s'", mensaje_error, re.IGNORECASE)
    if match_delay:
        return float(match_delay.group(1))
    return None

def traducir_bloque_tarjetas(bloque_tarjetas):
    """Envía un lote de elementos aplicando el modelo Lite y control dinámico de cuotas."""
    texto_input = json.dumps(bloque_tarjetas, ensure_ascii=False, indent=2)
    intento = 0
    max_intentos = 3
    
    while intento < max_intentos:
        try:
            # Aprovechamos el modelo 3.1 Flash Lite por su cuota extendida de 500 RPD
            response = client.models.generate_content(
                model='gemini-3.1-flash-lite',
                contents=f"Traducí este JSON al español (campos 'q' y 'a'):\n\n{texto_input}",
                config=types.GenerateContentConfig(
                    system_instruction=SYSTEM_PROMPT,
                    response_mime_type="application/json" 
                ),
            )
            time.sleep(4) # Respetamos el límite de 15 RPM (1 petición cada 4 segundos)
            return json.loads(response.text.strip())
            
        except APIError as e:
            intento += 1
            error_msg = str(e)
            print(f"⚠️ Alerta de API (Intento {intento}/{max_intentos})")
            
            segundos_espera = extraer_tiempo_espera(error_msg)
            if segundos_espera:
                # Margen de seguridad para absorber la latencia del canal de red
                tiempo_final = segundos_espera + 2.0
                print(f"📢 Cuota superada. Esperando de forma inteligente {tiempo_final} segundos...")
                time.sleep(tiempo_final)
            else:
                tiempo_defecto = 30 * intento
                print(f"No se detectó tiempo explícito. Pausa por defecto: {tiempo_defecto}s.")
                time.sleep(tiempo_defecto)
        except Exception as ge:
            intento += 1
            print(f"⚠️ Error general del sistema: {ge}")
            time.sleep(15)
    return None

def iniciar_proceso_traduccion(ruta_entrada, ruta_salida, tamaño_bloque=10):
    print("=========================================================")
    print("Iniciando motor de traducción inteligente por bloques")
    print("=========================================================\n")
    
    try:
        datos_originales = extraer_json_de_js(ruta_entrada)
    except Exception as e:
        print(f"Error crítico de lectura: {e}")
        return

    total_items = len(datos_originales)
    print(f"Total de objetos a procesar: {total_items}")
    
    datos_traducidos = []
    
    for i in range(0, total_items, tamaño_bloque):
        bloque = datos_originales[i:i + tamaño_bloque]
        print(f"📦 Despachando lote: Índices {i} a {i + len(bloque)}...")
        
        bloque_traducido = traducir_bloque_tarjetas(bloque)
        
        if bloque_traducido:
            if isinstance(bloque_traducido, list):
                datos_traducidos.extend(bloque_traducido)
                print("✅ Lote procesado e integrado.")
            else:
                datos_traducidos.extend(bloque)
        else:
            print("❌ Bloque fallido de forma permanente. Abortando para proteger los índices.")
            break
            
    # Escritura final en formato compatible
    json_formateado = json.dumps(datos_traducidos, ensure_ascii=False, indent=4)
    contenido_final = f"// Archivo compilado de forma automática\nconst dehonFlashcards_es = {json_formateado};"
    
    with open(ruta_salida, 'w', encoding='utf-8') as f:
        f.write(contenido_final)
    print(f"\nProceso concluido. Archivo de salida listo en: {ruta_salida}")

if __name__ == "__main__":
    iniciar_proceso_traduccion("catecismo_dehon_pt.js", "catecismo_dehon_es.js", tamaño_bloque=10)

Ventajas de la arquitectura avanzada de código

Al implementar este paradigma de desarrollo, no solo solucionamos las restricciones físicas impuestas por la infraestructura básica, sino que elevamos el estándar operativo del software:

  • Mitigación eficiente de solicitudes diarias: Al saltar al modelo 3.1 Flash Lite, las 11 solicitudes requeridas para procesar un documento entero representan apenas el 2% de tu cuota de 500 RPD, dejando el entorno libre para pruebas complejas en paralelo.
  • Tolerancia a fallos quirúrgica: El script intercepta de manera controlada el error genérico del cliente y extrae el retardo exacto mediante expresiones regulares, automatizando por completo las esperas por saturación horaria.
  • Auditoría completa: Al redirigir el flujo mediante la clase Logger, el programador puede desentenderse de la ejecución sabiendo que cada bit de información quedará registrado en un archivo histórico.

Conclusión

Los entornos de Inteligencia Artificial modernos exigen que los desarrolladores abandonen los enfoques de fuerza bruta. El diseño de arquitecturas eficientes basados en el procesamiento por lotes y la auditoría interna de cuotas representa la única frontera válida para extraer el máximo potencial a los modelos fundacionales a gran escala.

Bibliografía de referencia

Comentarios