Integración de Gemini API en bot de Telegram: guía para desarrolladores

Integración asíncrona de la API de Gemini en un bot de Telegram: Guía práctica con aiogram 3.x

Cuando tu bot de Telegram deja de repente de responder por un error 429 de la API de Gemini, no es solo un fallo: es deuda técnica. En este artículo, lo desglosamos paso a paso: cómo integrar correctamente Gemini en un bot asíncrono con aiogram 3.x para evitar bloquear el event loop, sortear los límites de RPM/TPM e implementar llamadas a funciones sin sacrificar el rendimiento.

Arquitectura: De un manejador simple a una solución de producción

El flujo básico de interacción usuario-bot vía Gemini parece directo: solicitud → procesamiento → respuesta. Pero los escenarios del mundo real demandan capas adicionales. Llamar a una API síncrona de forma asíncrona sin una envoltura adecuada congelará el event loop de aiogram, sobre todo bajo carga pesada. Por eso, los componentes clave son:

• Cola de Asyncio — para amortiguar las solicitudes entrantes
• Limitador de tasa — para cumplir con los límites de Gemini (5–60 RPM)
• Caché de respuestas — para ahorrar tokens en solicitudes repetidas
• Manejador de streaming — para evitar los timeouts de Telegram (10 segundos)

La API de Gemini impone límites estrictos incluso en planes de pago: la latencia máxima de respuesta puede llegar a 15 segundos, y el nivel gratuito se limita a solo 5 solicitudes por minuto. Sin colas y caché, la experiencia de usuario será un desastre.

Configuración del cliente: SDK síncrono en un entorno asíncrono

El SDK de Google GenAI sigue siendo síncrono a pesar de las esperanzas de la comunidad. Eso significa que llamar a generate_content() directamente en un manejador de aiogram bloqueará todo el event loop. La solución: envolver las llamadas en asyncio.to_thread o usar run_in_executor.

import asyncio
from google import genai

class AsyncGeminiClient:
    def __init__(self, model: str = "gemini-3-flash-preview"):
        self.client = genai.Client()
        self.model = model
    
    async def generate(self, prompt: str) -> str:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None,
            self._sync_generate,
            prompt
        )
        return response
    
    def _sync_generate(self, prompt: str) -> str:
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt
        )
        return response.text

Este es un envoltorio mínimo y funcional. Para producción, añade manejo de excepciones y métricas de ejecución.

Integración con aiogram: De /start a respuestas en streaming

Un manejador básico necesita hacer más que solo aceptar mensajes: gestionar el estado mostrando "typing", dividir respuestas largas y almacenar resultados en caché. Telegram limita la longitud de los mensajes a 4096 caracteres: ignóralo y tendrás respuestas cortadas.

@router.message()
async def handle_message(message: types.Message):
    await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    try:
        response = await gemini.generate(message.text)
        if len(response) > 4000:
            for i in range(0, len(response), 4000):
                await message.answer(response[i:i+4000])
        else:
            await message.answer(response)
    except Exception as e:
        logging.error(f"Gemini error: {e}")
        await message.answer("Chto-then poshlo not so. Poprobuyte pozzhe.")

Para solicitudes de larga duración, usa streaming: actualiza un solo mensaje a medida que se genera el texto. Esto evita timeouts y mejora la UX.

Llamadas a funciones: Cuando Gemini se convierte en un agente

Function Calling permite que el modelo invoque APIs externas según el contexto de la solicitud. Por ejemplo, si un usuario dice "reserva una reunión para mañana a las 3 PM", Gemini puede analizar los parámetros y llamar a tu función schedule_meeting.

Pasos clave:

• Describe la función en formato JSON Schema — para que el modelo entienda su firma.
• Pasa la descripción en GenerateContentConfig como tools.
• Maneja function_call en la respuesta y ejecuta la llamada real.
• Alimenta el resultado de vuelta al chat como parte de la conversación.

Ejemplo de declaración de función:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Withzdayot meetingsu with ukazannymi uchastnikami",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

Esto convierte un bot estático en un asistente dinámico capaz de interactuar con sistemas externos.

Caché y limitación de tasa: Protección contra quiebras y baneos

Gemini cobra por tokens. Sin caché, pagarás por cada pregunta repetida. Un caché simple en memoria con hash MD5 del prompt y TTL lo soluciona:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # check TTL and vozvrat values
    
    def set(self, key: str, value: str):
        self._cache[key] = (value, datetime.now())
    
    @staticmethod
    def hash_prompt(prompt: str) -> str:
        return hashlib.md5(prompt.lower().strip().encode()).hexdigest()

La limitación de tasa es esencial incluso en niveles de pago. Impleméntala como un context manager asíncrono o middleware:

class AsyncRateLimiter:
    def __init__(self, max_requests: int, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
        self._lock = asyncio.Lock()
    
    async def acquire(self):
        async with self._lock:
            now = time.time()
            self.requests = [t for t in self.requests if now - t < self.time_window]
            if len(self.requests) >= self.max_requests:
                sleep_time = self.time_window - (now - self.requests[0])
                await asyncio.sleep(sleep_time + 0.1)
                return await self.acquire()
            self.requests.append(now)

Errores comunes y cómo evitarlos

Tres trampas clásicas que todos cometen:

• Error 429 (RESOURCE_EXHAUSTED) — se activa al exceder RPM, TPM o RPD. Solución: limitador de tasa local + backoff exponencial.
• Timeouts de Telegram — el servidor espera 10 segundos por una respuesta. Si Gemini es lento, usa streaming o envía mensajes intermedios.
• Modelos obsoletos — Google retira regularmente versiones antiguas. En marzo de 2026, los modelos Pro pasaron a solo por suscripción. Siempre verifica la disponibilidad del modelo antes de desplegar.

Lecciones clave:

• El SDK síncrono de Gemini necesita envoltura con run_in_executor — o bloquearás el event loop.
• Implementa limitación de tasa en el lado del bot, no confíes en la API.
• El streaming es la única forma de evitar el timeout de 10 segundos de Telegram en respuestas largas.
• El caché ahorra dinero y reduce la latencia en repeticiones.
• Function Calling evoluciona tu bot de generador de texto a agente completo.

Adoptar estas prácticas reduce los incidentes en producción y mantiene la estabilidad a medida que crece la carga. No tomes atajos en la arquitectura: ahorra en tokens en su lugar.

— Editorial Team