Cómo Crear un Bot de Telegram con IA y Webhooks en Python: Streaming de Respuestas y Despliegue Serverless
Telegram se ha consolidado como la plataforma predilecta para desarrolladores y usuarios avanzados que buscan interfaces de comunicación ultrarrápidas, ligeras y con una API de bots infinitamente más abierta y versátil que la de WhatsApp o Slack. Sin embargo, la mayoría de los tutoriales de bots con inteligencia artificial en internet incurren en dos errores de novato garrafales: utilizar sondeos ineficientes (Long Polling) que colapsan la CPU del servidor, y esperar a que el LLM termine de redactar toda la respuesta (tardando 10 segundos en blanco) antes de enviar el primer mensaje.
En este tutorial aprenderás a construir una solución de nivel profesional: una arquitectura serverless asíncrona basada en Webhooks con FastAPI y Python, y un sistema de streaming de tokens en tiempo real con estrangulamiento (throttling) que va escribiendo la respuesta en pantalla progresivamente sin superar los límites de rate limit de Telegram. Siguiendo las especificaciones oficiales de Telegram Bot API Reference y la documentación de FastAPI Modern Python Web Framework, desplegarás un bot listo para soportar miles de usuarios concurrentes.
Índice de Contenidos
- 👉 1. Requisitos Iniciales: Creación del Bot y Obtención del Token en BotFather
- 👉 2. Arquitectura: Por Qué Abandonar Long Polling y Adoptar Webhooks Asíncronos
- 👉 3. Configuración del Servidor: FastAPI, Python-Telegram-Bot y Variables de Entorno
- 👉 4. Streaming de Tokens en Tiempo Real: El Algoritmo de Throttling para Evitar el Error 429
- 👉 5. Código Fuente Completo: Servidor Asíncrono con Manejo de Mensajes y LLM
- 👉 6. Tabla Comparativa: Long Polling vs. Webhooks en VPS vs. Serverless
- 👉 7. Despliegue en Producción: Certificados SSL Gratuitos y Hosting en la Nube
- 👉 8. Preguntas Frecuentes sobre la Creación de Bots de Telegram con IA
1. Requisitos Iniciales: Creación del Bot y Obtención del Token en BotFather
El primer paso es registrar la identidad del bot en los servidores centrales de Telegram:
- Abre la aplicación de Telegram y busca al usuario oficial @BotFather.
- Envía el comando
/newboty sigue las instrucciones para asignar un nombre público y un identificador único que termine enbot(por ejemplo:CiberbotiaAsistenteBot). - BotFather te entregará un token criptográfico de acceso HTTP API de la forma
123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ. Guárdalo celosamente en tu archivo.env. - Opcional: Deshabilita el acceso a grupos con
/setprivacysi deseas que el bot solo responda en conversaciones privadas directas.
Desarrollo asíncrono en Python conectando la API de Telegram con modelos de lenguaje.
2. Arquitectura: Por Qué Abandonar Long Polling y Adoptar Webhooks Asíncronos
El método clásico Long Polling mantiene una conexión HTTP abierta continua preguntando a Telegram cada segundo: ‘¿Hay nuevos mensajes?’. Esto consume ancho de banda, mantiene la CPU despierta y se rompe ante reinicios de red.
En una arquitectura de Webhooks, tú expones un endpoint HTTPS público en tu servidor. Cuando un usuario envía un mensaje, los propios servidores de Telegram realizan una petición POST instantánea hacia tu URL. Tu servidor despierta, procesa el mensaje de forma asíncrona y vuelve a reposo, permitiendo escalar a miles de usuarios con un coste computacional ínfimo.
3. Configuración del Servidor: FastAPI, Python-Telegram-Bot y Variables de Entorno
Instalamos las librerías necesarias en un entorno virtual limpio:
pip install fastapi uvicorn python-telegram-bot openai python-dotenv
Creamos nuestro archivo .env con las credenciales maestras:
TELEGRAM_BOT_TOKEN="tu_token_de_botfather"
OPENAI_API_KEY="tu_clave_de_openai"
WEBHOOK_URL="https://tu-dominio.com/webhook"
4. Streaming de Tokens en Tiempo Real: El Algoritmo de Throttling para Evitar el Error 429
Para emular la experiencia interactiva de ChatGPT, queremos que Telegram vaya mostrando el texto a medida que el LLM lo genera palabra a palabra. Sin embargo, Telegram impone un límite estricto: no permite editar el mismo mensaje más de una vez por segundo (Error 429 Too Many Requests).
La solución técnica es implementar un búfer con estrangulamiento temporal (throttling): acumulamos los tokens entrantes del generador asíncrono y actualizamos el mensaje en Telegram mediante edit_message_text únicamente cuando hayan transcurrido al menos 1.2 segundos desde la última actualización o cuando se complete la respuesta final. Si quieres aprender a conectar bases de datos con RAG a este bot, consulta nuestro tutorial de RAG con LangChain y Python.
Infraestructura serverless con microservicios en FastAPI para despliegue de webhooks escalables.
5. Código Fuente Completo: Servidor Asíncrono con Manejo de Mensajes y LLM
A continuación se presenta la implementación completa y lista para producción en un único archivo main.py:
import os
import time
import asyncio
from fastapi import FastAPI, Request
from telegram import Update, Bot
from openai import AsyncOpenAI
from dotenv import load_dotenv
load_dotenv()
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
bot = Bot(token=TOKEN)
client = AsyncOpenAI()
app = FastAPI()
@app.post("/webhook")
async def telegram_webhook(request: Request):
req_json = await request.json()
update = Update.de_json(req_json, bot)
if update.message and update.message.text:
chat_id = update.message.chat_id
user_prompt = update.message.text
# Enviar mensaje inicial de espera
sent_msg = await bot.send_message(chat_id=chat_id, text="Pensando...")
msg_id = sent_msg.message_id
# Iniciar streaming desde OpenAI
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Eres un asistente técnico conciso y útil de Ciberbotia."},
{"role": "user", "content": user_prompt}
],
stream=True
)
full_text = ""
last_edit_time = time.time()
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
full_text += delta
# Throttling de 1.3 segundos para respetar los límites de Telegram
if time.time() - last_edit_time > 1.3 and full_text.strip():
try:
await bot.edit_message_text(chat_id=chat_id, message_id=msg_id, text=full_text)
last_edit_time = time.time()
except Exception:
pass
# Actualización final garantizada
try:
await bot.edit_message_text(chat_id=chat_id, message_id=msg_id, text=full_text)
except Exception:
pass
return {"status": "ok"}
6. Tabla Comparativa: Long Polling vs. Webhooks en VPS vs. Serverless
Evaluación de arquitecturas de despliegue para bots de mensajería:
| Arquitectura | Consumo de Recursos | Latencia de Entrada | Facilidad de Configuración | Coste Operativo |
|---|---|---|---|---|
| Long Polling Local | Alto en reposo (Bucle continuo) | ~500 ms – 1000 ms | ⭐⭐⭐⭐⭐ Extremadamente Fácil | Gratis (Tu PC) |
| Webhooks en VPS (FastAPI) | Muy Bajo (Orientado a eventos) | < 50 ms (Instantánea) | Media (Requiere Nginx y SSL) | Bajo (VPS 4-5 $/mes) |
| Serverless (AWS Lambda / Vercel) | Cero en reposo (Escala a cero) | Variable (Cold starts de ~1s) | Media / Alta | Gratis en tiers básicos |
7. Despliegue en Producción: Certificados SSL Gratuitos y Hosting en la Nube
Telegram exige de forma innegociable que la URL del Webhook esté protegida bajo HTTPS con un certificado SSL válido. Para desplegar gratis o a muy bajo coste sin lidiar con configuración manual de certificados:
- Render / Koyeb / Fly.io: Plataformas modernas de PaaS donde subes tu repositorio de GitHub y te asignan automáticamente un subdominio HTTPS seguro (ej.
https://mi-bot.onrender.com). - Registrar el Webhook: Una vez desplegado, abres en tu navegador una petición única:
https://api.telegram.org/bot<TU_TOKEN>/setWebhook?url=https://tu-dominio.com/webhook. Telegram responderá{"ok":true,"result":true,"description":"Webhook was set"}y tu bot estará operativo al 100%.
Monitorización de peticiones HTTP en tiempo real para optimizar la latencia del webhook.
8. Preguntas Frecuentes sobre la Creación de Bots de Telegram con IA
¿Puedo enviar mensajes con formato Markdown enriquecido o código?
Sí. Basta con añadir el parámetro parse_mode="MarkdownV2" o parse_mode="HTML" en la llamada a send_message o edit_message_text. Si usas MarkdownV2, asegúrate de escapar caracteres especiales como puntos y guiones para evitar errores de sintaxis.
¿Cómo gestiono la memoria de conversaciones pasadas para cada usuario?
En una base de datos ligera como SQLite o Redis, almacenas una lista con los últimos 6 o 10 mensajes intercambiados indexados por el chat_id del usuario. Cada vez que recibes un nuevo mensaje, recuperas ese historial y lo concatenas en el array messages que envías al modelo.
¿Puedo conectar un modelo open source local (como LLaMA 3 con Ollama) a mi bot de Telegram?
Totalmente. Sustituyes la llamada de OpenAI por la URL local de Ollama (http://localhost:11434/v1). Si tu servidor FastAPI corre en la misma máquina o en tu red local con túneles seguros como Cloudflare Tunnels o Ngrok, el bot responderá utilizando tu hardware local con coste cero por mensaje.
