Bot de Telegram com IA usando Python

O que vamos construir

Um bot de Telegram completo que:

• Responde mensagens usando a API da OpenAI (GPT-4o-mini)
• Mantém histórico de conversa por usuário
• Tem comandos customizados (/resumir, /traduzir, /codigo)
• Suporta modo de streaming para respostas longas
• Inclui rate limiting para controle de custos

Pré-requisitos

1. Criar bot no Telegram via @BotFather
2. Obter token da API da OpenAI
3. Python 3.10+

pip install python-telegram-bot openai

Código Completo

1. Configuração e Imports

"""
bot_telegram_ia.py - Bot de Telegram com IA
"""

import os
import logging
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Optional

from telegram import Update, BotCommand
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    ContextTypes,
    filters,
)
from openai import OpenAI

# Configuração
TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# Logging
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO,
)
logger = logging.getLogger(__name__)

# Cliente OpenAI
openai_client = OpenAI(api_key=OPENAI_API_KEY)

2. Gerenciamento de Histórico

class ConversationManager:
    """Gerencia histórico de conversas por usuário."""

    def __init__(self, max_messages: int = 20, max_age_hours: int = 2):
        self.histories: dict[int, list[dict]] = defaultdict(list)
        self.max_messages = max_messages
        self.max_age_hours = max_age_hours
        self.last_activity: dict[int, datetime] = {}

    def get_history(self, user_id: int) -> list[dict]:
        """Retorna histórico do usuário, limpando mensagens antigas."""
        # Limpar se inativo por muito tempo
        if user_id in self.last_activity:
            if datetime.now() - self.last_activity[user_id] > timedelta(hours=self.max_age_hours):
                self.histories[user_id] = []

        self.last_activity[user_id] = datetime.now()
        return self.histories[user_id]

    def add_message(self, user_id: int, role: str, content: str):
        """Adiciona mensagem ao histórico."""
        self.histories[user_id].append({"role": role, "content": content})

        # Manter apenas as últimas N mensagens
        if len(self.histories[user_id]) > self.max_messages:
            # Manter system prompt + últimas mensagens
            self.histories[user_id] = self.histories[user_id][-self.max_messages:]

    def clear(self, user_id: int):
        """Limpa histórico do usuário."""
        self.histories[user_id] = []
        logger.info(f"Histórico limpo para usuário {user_id}")


class RateLimiter:
    """Controla taxa de uso por usuário."""

    def __init__(self, max_requests: int = 20, window_minutes: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(minutes=window_minutes)
        self.requests: dict[int, list[datetime]] = defaultdict(list)

    def check(self, user_id: int) -> bool:
        """Verifica se o usuário pode fazer mais requisições."""
        now = datetime.now()
        # Limpar requisições antigas
        self.requests[user_id] = [
            t for t in self.requests[user_id] if now - t < self.window
        ]
        return len(self.requests[user_id]) < self.max_requests

    def record(self, user_id: int):
        """Registra uma requisição."""
        self.requests[user_id].append(datetime.now())

    def remaining(self, user_id: int) -> int:
        """Retorna quantas requisições restam."""
        now = datetime.now()
        active = [t for t in self.requests[user_id] if now - t < self.window]
        return max(0, self.max_requests - len(active))


# Instâncias globais
conversations = ConversationManager()
rate_limiter = RateLimiter(max_requests=30, window_minutes=60)

SYSTEM_PROMPT = """Você é um assistente inteligente no Telegram, especializado em 
programação e inteligência artificial. 

REGRAS:
- Responda sempre em português brasileiro
- Seja conciso (Telegram tem tela pequena)
- Use formatação Markdown quando útil
- Para código, use blocos de código com a linguagem
- Se não souber algo, diga honestamente
- Seja amigável e técnico ao mesmo tempo"""

3. Função de Chat com IA

async def chat_with_ai(
    user_id: int,
    message: str,
    system_override: Optional[str] = None,
) -> str:
    """Envia mensagem para a IA e retorna resposta."""

    history = conversations.get_history(user_id)

    messages = [
        {"role": "system", "content": system_override or SYSTEM_PROMPT},
    ]
    messages.extend(history)
    messages.append({"role": "user", "content": message})

    try:
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0.7,
            max_tokens=2000,
        )

        assistant_message = response.choices[0].message.content

        # Salvar no histórico
        conversations.add_message(user_id, "user", message)
        conversations.add_message(user_id, "assistant", assistant_message)

        return assistant_message

    except Exception as e:
        logger.error(f"Erro na API OpenAI: {e}")
        return f"❌ Erro ao processar sua mensagem: {str(e)}"

4. Handlers de Comandos

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /start."""
    user = update.effective_user
    await update.message.reply_text(
        f"👋 Olá, {user.first_name}!\n\n"
        f"Sou um bot com IA integrada. Pode me perguntar qualquer coisa!\n\n"
        f"📋 *Comandos disponíveis:*\n"
        f"/resumir — Resuma um texto\n"
        f"/traduzir — Traduza um texto\n"
        f"/codigo — Gere código Python\n"
        f"/limpar — Limpe o histórico\n"
        f"/status — Veja seu uso\n"
        f"/ajuda — Mais informações\n\n"
        f"Ou simplesmente me envie uma mensagem! 🤖",
        parse_mode="Markdown",
    )


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /ajuda."""
    await update.message.reply_text(
        "🤖 *Como me usar:*\n\n"
        "1️⃣ Envie qualquer mensagem e eu respondo com IA\n"
        "2️⃣ Mantenho contexto da conversa (lembro do que falamos)\n"
        "3️⃣ Use comandos especiais para tarefas específicas\n\n"
        "📌 *Dicas:*\n"
        "• Seja específico nas perguntas\n"
        "• Para código, diga a linguagem desejada\n"
        "• Use /limpar se quiser começar conversa nova\n"
        f"• Limite: {rate_limiter.max_requests} mensagens por hora",
        parse_mode="Markdown",
    )


async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /limpar."""
    user_id = update.effective_user.id
    conversations.clear(user_id)
    await update.message.reply_text(
        "🧹 Histórico limpo! Podemos começar uma conversa nova."
    )


async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /status."""
    user_id = update.effective_user.id
    remaining = rate_limiter.remaining(user_id)
    history_size = len(conversations.get_history(user_id))

    await update.message.reply_text(
        f"📊 *Seu Status:*\n\n"
        f"💬 Mensagens no histórico: {history_size}\n"
        f"⏳ Requisições restantes: {remaining}/{rate_limiter.max_requests}\n"
        f"🔄 Reset em: 60 minutos",
        parse_mode="Markdown",
    )


async def summarize_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /resumir <texto>."""
    user_id = update.effective_user.id

    if not context.args:
        await update.message.reply_text(
            "📝 Use: `/resumir <texto para resumir>`",
            parse_mode="Markdown",
        )
        return

    texto = " ".join(context.args)
    await update.message.reply_text("⏳ Resumindo...")

    system = "Você é um assistente que cria resumos concisos e claros em português."
    response = await chat_with_ai(
        user_id, f"Resuma o seguinte texto:\n\n{texto}", system_override=system
    )
    await update.message.reply_text(f"📋 *Resumo:*\n\n{response}", parse_mode="Markdown")


async def translate_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /traduzir <texto>."""
    user_id = update.effective_user.id

    if not context.args:
        await update.message.reply_text(
            "🌐 Use: `/traduzir <texto em qualquer idioma>`",
            parse_mode="Markdown",
        )
        return

    texto = " ".join(context.args)
    await update.message.reply_text("⏳ Traduzindo...")

    system = (
        "Você é um tradutor profissional. Traduza o texto para português brasileiro "
        "se estiver em outro idioma, ou para inglês se estiver em português."
    )
    response = await chat_with_ai(
        user_id, f"Traduza: {texto}", system_override=system
    )
    await update.message.reply_text(f"🌐 *Tradução:*\n\n{response}", parse_mode="Markdown")


async def code_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para /codigo <descrição>."""
    user_id = update.effective_user.id

    if not context.args:
        await update.message.reply_text(
            "💻 Use: `/codigo <descreva o que quer programar>`",
            parse_mode="Markdown",
        )
        return

    descricao = " ".join(context.args)
    await update.message.reply_text("⏳ Gerando código...")

    system = (
        "Você é um programador senior. Gere código Python limpo, "
        "comentado e funcional. Inclua docstrings e type hints."
    )
    response = await chat_with_ai(
        user_id, f"Gere código Python para: {descricao}", system_override=system
    )
    await update.message.reply_text(response, parse_mode="Markdown")

5. Handler de Mensagens

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler para mensagens de texto normais."""
    user_id = update.effective_user.id
    user_message = update.message.text

    # Rate limiting
    if not rate_limiter.check(user_id):
        remaining = rate_limiter.remaining(user_id)
        await update.message.reply_text(
            f"⚠️ Limite de mensagens atingido. "
            f"Aguarde alguns minutos e tente novamente.\n"
            f"Restantes: {remaining}/{rate_limiter.max_requests}"
        )
        return

    rate_limiter.record(user_id)

    # Indicar que está "digitando"
    await update.message.chat.send_action("typing")

    # Processar com IA
    response = await chat_with_ai(user_id, user_message)

    # Telegram tem limite de 4096 caracteres por mensagem
    if len(response) > 4000:
        # Dividir em partes
        parts = [response[i:i+4000] for i in range(0, len(response), 4000)]
        for part in parts:
            await update.message.reply_text(part, parse_mode="Markdown")
    else:
        try:
            await update.message.reply_text(response, parse_mode="Markdown")
        except Exception:
            # Se Markdown falhar, enviar sem formatação
            await update.message.reply_text(response)


async def handle_error(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handler global de erros."""
    logger.error(f"Erro: {context.error}", exc_info=context.error)
    if update and update.message:
        await update.message.reply_text(
            "❌ Ocorreu um erro inesperado. Tente novamente."
        )

6. Inicialização do Bot

async def post_init(application: Application):
    """Configuração pós-inicialização."""
    # Registrar comandos no menu do Telegram
    commands = [
        BotCommand("start", "Iniciar o bot"),
        BotCommand("ajuda", "Ver instruções de uso"),
        BotCommand("resumir", "Resumir um texto"),
        BotCommand("traduzir", "Traduzir um texto"),
        BotCommand("codigo", "Gerar código Python"),
        BotCommand("limpar", "Limpar histórico"),
        BotCommand("status", "Ver status de uso"),
    ]
    await application.bot.set_my_commands(commands)
    logger.info("✅ Bot inicializado com sucesso!")


def main():
    """Ponto de entrada do bot."""
    print("🤖 Iniciando Bot de Telegram com IA...")

    # Criar aplicação
    app = Application.builder().token(TELEGRAM_TOKEN).post_init(post_init).build()

    # Registrar handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("ajuda", help_command))
    app.add_handler(CommandHandler("limpar", clear_command))
    app.add_handler(CommandHandler("status", status_command))
    app.add_handler(CommandHandler("resumir", summarize_command))
    app.add_handler(CommandHandler("traduzir", translate_command))
    app.add_handler(CommandHandler("codigo", code_command))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    # Handler de erros
    app.add_error_handler(handle_error)

    # Rodar bot
    print("✅ Bot rodando! Pressione Ctrl+C para parar.")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

Como Rodar

# Instalar dependências
pip install python-telegram-bot openai

# Configurar variáveis de ambiente
export TELEGRAM_BOT_TOKEN="seu-token-do-botfather"
export OPENAI_API_KEY="sk-..."

# Executar
python bot_telegram_ia.py

Testando o Bot

1. Abra o Telegram e busque seu bot pelo username
2. Envie /start para iniciar
3. Faça perguntas naturalmente: "Como fazer web scraping em Python?"
4. Use comandos: /codigo calculadora simples
5. Teste tradução: /traduzir Hello, how are you?

Expansões Possíveis

• Inline mode: responder em qualquer chat mencionando o bot
• Imagens: integrar DALL-E para geração de imagens
• Voz: transcrever áudios com Whisper e responder
• Grupos: funcionar em grupos com menção
• Persistência: salvar histórico em banco de dados (Redis/PostgreSQL)
• Admin panel: comandos especiais para administradores

Pronto! Você tem um bot de Telegram com IA funcional em menos de 300 linhas de Python. 🚀