Créer un chatbot Telegram intelligent en 30 minutes (code inclus)

You: /newbot
BotFather: Alright, a new bot. How are we going to call it?
You: Mon Assistant IA
BotFather: Good. Now choose a username for your bot. It must end in 'bot'.
You: mon_assistant_ia_bot
BotFather: Done! Your token is: 1234567890:ABCdefGHIjklMNOpqrsTUVwxyz

/setdescription — Une description courte de votre bot
/setabouttext — Texte affiché dans le profil
/setuserpic — Photo de profil (optionnel)
/setcommands — Liste des commandes (on y reviendra)

start - Démarrer une conversation
reset - Réinitialiser la mémoire
stats - Voir les statistiques d'utilisation
help - Afficher l'aide

User → Telegram API → Bot Python → [Récupère historique] 
                                  ↓
                             LLM (OpenAI/Claude)
                                  ↓
                         [Sauvegarde réponse]
                                  ↓
                            Telegram API → User

mkdir telegram-chatbot-ia
cd telegram-chatbot-ia
python3 -m venv venv
source venv/bin/activate  # Sur Windows: venv\Scripts\activate

pip install python-telegram-bot openai anthropic python-dotenv

TELEGRAM_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxx
# Ou si vous utilisez Claude :
# ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxx

#!/usr/bin/env python3
"""
Chatbot Telegram intelligent avec mémoire et LLM
Auteur : AI-master.dev
Licence : MIT
"""

import os
import json
import logging
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv

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

import openai

# Configuration du logging (pour débugger)
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

# Chargement des variables d'environnement
load_dotenv()
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

# Configuration OpenAI
openai.api_key = OPENAI_API_KEY

# Dossier pour stocker les conversations
MEMORY_DIR = Path("conversations")
MEMORY_DIR.mkdir(exist_ok=True)


class ConversationMemory:
    """Gère la mémoire des conversations par utilisateur"""

    def __init__(self, user_id: int):
        self.user_id = user_id
        self.file_path = MEMORY_DIR / f"{user_id}.json"
        self.messages = self._load()

    def _load(self) -> list:
        """Charge l'historique depuis le fichier JSON"""
        if self.file_path.exists():
            with open(self.file_path, 'r', encoding='utf-8') as f:
                return json.load(f)
        return []

    def _save(self):
        """Sauvegarde l'historique dans le fichier JSON"""
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump(self.messages, f, ensure_ascii=False, indent=2)

    def add_message(self, role: str, content: str):
        """Ajoute un message à l'historique"""
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        # Limite à 50 messages pour éviter des contextes trop longs
        if len(self.messages) > 50:
            # Garde le premier message (system prompt) + les 49 derniers
            self.messages = [self.messages[0]] + self.messages[-49:]
        self._save()

    def reset(self):
        """Réinitialise la mémoire"""
        self.messages = []
        if self.file_path.exists():
            self.file_path.unlink()

    def get_context(self) -> list:
        """Retourne l'historique au format OpenAI"""
        # Filtre uniquement role et content (enlève timestamp)
        return [
            {"role": msg["role"], "content": msg["content"]}
            for msg in self.messages
        ]


async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Commande /start"""
    user = update.effective_user
    welcome_message = (
        f"👋 Bonjour {user.first_name} !\n\n"
        "Je suis votre assistant IA personnel. Je peux :\n"
        "• Répondre à vos questions\n"
        "• Me souvenir de nos conversations\n"
        "• Vous aider dans vos tâches quotidiennes\n\n"
        "Posez-moi n'importe quelle question !\n\n"
        "Commandes disponibles :\n"
        "/reset - Réinitialiser notre conversation\n"
        "/stats - Voir vos statistiques\n"
        "/help - Afficher l'aide"
    )
    await update.message.reply_text(welcome_message)


async def reset_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Commande /reset - Efface la mémoire"""
    user_id = update.effective_user.id
    memory = ConversationMemory(user_id)
    memory.reset()
    await update.message.reply_text(
        "🔄 Mémoire réinitialisée ! Notre conversation repart de zéro."
    )


async def stats_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Commande /stats - Affiche les statistiques"""
    user_id = update.effective_user.id
    memory = ConversationMemory(user_id)

    total_messages = len(memory.messages)
    user_messages = sum(1 for m in memory.messages if m["role"] == "user")
    assistant_messages = sum(1 for m in memory.messages if m["role"] == "assistant")

    stats_text = (
        f"📊 **Vos statistiques**\n\n"
        f"Total de messages : {total_messages}\n"
        f"Vos messages : {user_messages}\n"
        f"Mes réponses : {assistant_messages}\n"
    )

    if memory.messages:
        first_msg = memory.messages[0].get("timestamp", "")
        stats_text += f"\nPremière conversation : {first_msg[:10]}"

    await update.message.reply_text(stats_text, parse_mode='Markdown')


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Commande /help"""
    help_text = (
        "🤖 **Guide d'utilisation**\n\n"
        "Envoyez-moi simplement un message et je vous répondrai !\n\n"
        "**Commandes disponibles :**\n"
        "/start - Démarrer le bot\n"
        "/reset - Effacer la mémoire de conversation\n"
        "/stats - Voir vos statistiques d'utilisation\n"
        "/help - Afficher ce message\n\n"
        "**Exemples de questions :**\n"
        "• Explique-moi la blockchain en termes simples\n"
        "• Donne-moi une recette de pâtes carbonara\n"
        "• Aide-moi à rédiger un email professionnel\n"
    )
    await update.message.reply_text(help_text, parse_mode='Markdown')


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Gère les messages texte et génère une réponse via LLM"""
    user_id = update.effective_user.id
    user_message = update.message.text

    # Affiche un indicateur "typing..."
    await update.message.chat.send_action("typing")

    try:
        # Charge la mémoire de l'utilisateur
        memory = ConversationMemory(user_id)

        # Ajoute le message de l'utilisateur
        memory.add_message("user", user_message)

        # Prépare le contexte pour l'API
        messages = [
            {
                "role": "system",
                "content": (
                    "Tu es un assistant IA utile, sympathique et concis. "
                    "Tu réponds en français de manière claire et naturelle. "
                    "Tu es là pour aider l'utilisateur du mieux possible."
                )
            }
        ] + memory.get_context()

        # Appel à l'API OpenAI
        response = openai.chat.completions.create(
            model="gpt-3.5-turbo",  # ou "gpt-4", "gpt-4-turbo"
            messages=messages,
            temperature=0.7,
            max_tokens=1000
        )

        # Extraction de la réponse
        assistant_reply = response.choices[0].message.content

        # Sauvegarde la réponse dans la mémoire
        memory.add_message("assistant", assistant_reply)

        # Envoie la réponse à l'utilisateur
        await update.message.reply_text(assistant_reply)

        logger.info(f"User {user_id}: {user_message[:50]}... | Response: {assistant_reply[:50]}...")

    except Exception as e:
        logger.error(f"Erreur lors du traitement du message : {e}")
        await update.message.reply_text(
            "❌ Désolé, une erreur est survenue. Réessayez dans quelques instants."
        )


async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Gère les erreurs globales"""
    logger.error(f"Update {update} caused error {context.error}")


def main():
    """Point d'entrée principal"""
    # Vérification des variables d'environnement
    if not TELEGRAM_TOKEN:
        raise ValueError("TELEGRAM_TOKEN manquant dans .env")
    if not OPENAI_API_KEY:
        raise ValueError("OPENAI_API_KEY manquant dans .env")

    # Création de l'application
    app = Application.builder().token(TELEGRAM_TOKEN).build()

    # Enregistrement des handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("reset", reset_command))
    app.add_handler(CommandHandler("stats", stats_command))
    app.add_handler(CommandHandler("help", help_command))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    # Gestion des erreurs
    app.add_error_handler(error_handler)

    # Démarrage du bot
    logger.info("🚀 Bot démarré ! Appuyez sur Ctrl+C pour arrêter.")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == '__main__':
    main()

python bot.py

2025-02-26 14:30:15,123 - __main__ - INFO - 🚀 Bot démarré ! Appuyez sur Ctrl+C pour arrêter.

class ConversationMemory:
    def __init__(self, user_id: int):
        self.user_id = user_id
        self.file_path = MEMORY_DIR / f"{user_id}.json"

messages = [
    {
        "role": "system",
        "content": "Tu es un assistant IA utile..."
    }
] + memory.get_context()

# Bot expert en développement
"Tu es un développeur senior expert en Python et IA. Tu donnes des réponses techniques précises avec du code."

# Bot coach sportif
"Tu es un coach sportif bienveillant. Tu motives l'utilisateur et donnes des conseils personnalisés."

# Bot de cuisine
"Tu es un chef cuisinier. Tu donnes des recettes détaillées et des astuces de pro."

if len(self.messages) > 50:
    self.messages = [self.messages[0]] + self.messages[-49:]

import anthropic

client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))

# Dans handle_message :
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1000,
    messages=memory.get_context()
)
assistant_reply = response.content[0].text

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [InlineKeyboardButton("💡 Conseil du jour", callback_data='tip')],
        [InlineKeyboardButton("📊 Mes stats", callback_data='stats')],
        [InlineKeyboardButton("❓ Aide", callback_data='help')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await update.message.reply_text(
        "Que puis-je faire pour vous ?",
        reply_markup=reply_markup
    )

from telegram.ext import CallbackQueryHandler

async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    if query.data == 'tip':
        await query.message.reply_text("💡 Astuce : Posez des questions précises pour de meilleures réponses !")
    elif query.data == 'stats':
        await stats_command(update, context)

app.add_handler(CallbackQueryHandler(button_handler))

import requests

def search_web(query: str) -> str:
    """Recherche sur le web via DuckDuckGo Instant Answer API"""
    url = f"https://api.duckduckgo.com/?q={query}&format=json"
    response = requests.get(url)
    data = response.json()
    return data.get('AbstractText', 'Aucun résultat trouvé.')

# Dans handle_message, avant l'appel LLM :
if "cherche sur internet" in user_message.lower():
    search_query = user_message.replace("cherche sur internet", "").strip()
    search_result = search_web(search_query)
    await update.message.reply_text(f"🔍 Résultat : {search_result}")
    return

from collections import defaultdict
from datetime import datetime, timedelta

user_last_message = defaultdict(lambda: datetime.min)

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    now = datetime.now()

    # Maximum 1 message toutes les 3 secondes
    if now - user_last_message[user_id] < timedelta(seconds=3):
        await update.message.reply_text("⏳ Doucement ! Attendez quelques secondes.")
        return

    user_last_message[user_id] = now
    # ... reste du code

# /etc/systemd/system/telegram-bot.service
[Unit]
Description=Telegram Chatbot IA
After=network.target

[Service]
Type=simple
User=botuser
WorkingDirectory=/home/botuser/telegram-chatbot-ia
ExecStart=/home/botuser/telegram-chatbot-ia/venv/bin/python bot.py
Restart=always

[Install]
WantedBy=multi-user.target

sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot
sudo systemctl status telegram-bot

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY bot.py .
COPY .env .

CMD ["python", "bot.py"]

# docker-compose.yml
version: '3.8'
services:
  bot:
    build: .
    restart: always
    volumes:
      - ./conversations:/app/conversations

docker-compose up -d

from telegram.ext import Application

app = Application.builder().token(TELEGRAM_TOKEN).build()
# ... ajoutez vos handlers

# Fonction Lambda
def lambda_handler(event, context):
    update = Update.de_json(json.loads(event['body']), app.bot)
    app.process_update(update)
    return {'statusCode': 200}

curl -X POST https://api.telegram.org/bot<TOKEN>/setWebhook \
  -d "url=https://your-function-url.amazonaws.com/webhook"

logger.info(f"User {user_id} | Model: gpt-3.5-turbo | Tokens: {response.usage.total_tokens} | Latency: {latency}ms")

ADMIN_CHAT_ID = 123456789  # Votre user_id Telegram

async def send_admin_alert(message: str):
    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=f"⚠️ ALERTE : {message}")

def update_stats():
    stats = {
        "total_users": len(list(MEMORY_DIR.glob("*.json"))),
        "total_messages": sum(len(json.load(open(f))) for f in MEMORY_DIR.glob("*.json")),
        "last_update": datetime.now().isoformat()
    }
    with open("stats.json", "w") as f:
        json.dump(stats, f)

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/stats')
def stats():
    return jsonify(json.load(open("stats.json")))

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Obtenir la météo d'une ville",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                }
            }
        }
    }
]

# Dans handle_message :
response = openai.chat.completions.create(
    model="gpt-4-turbo",
    messages=messages,
    tools=tools
)

if response.choices[0].message.tool_calls:
    # Exécutez la fonction et renvoyez le résultat
    pass

async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    photo = await update.message.photo[-1].get_file()
    await photo.download_to_drive('photo.jpg')

    # Analyse avec GPT-4 Vision
    response = openai.chat.completions.create(
        model="gpt-4-vision-preview",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Décris cette image"},
                    {"type": "image_url", "image_url": {"url": "photo.jpg"}}
                ]
            }
        ]
    )

    await update.message.reply_text(response.choices[0].message.content)

app.add_handler(MessageHandler(filters.PHOTO, handle_photo))

async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    voice = await update.message.voice.get_file()
    await voice.download_to_drive('voice.ogg')

    # Transcription avec Whisper
    audio_file = open('voice.ogg', 'rb')
    transcript = openai.audio.transcriptions.create(
        model="whisper-1",
        file=audio_file
    )

    # Traitez le texte transcrit comme un message normal
    await handle_message(update, context)

import asyncio

async def send_daily_tip():
    users = [f.stem for f in MEMORY_DIR.glob("*.json")]
    tip = "💡 Astuce du jour : Posez des questions ouvertes pour des réponses plus complètes !"

    for user_id in users:
        await context.bot.send_message(chat_id=int(user_id), text=tip)

# Programmez avec asyncio ou un cron externe

if len(memory.messages) > 30:
    # Résumez les 20 premiers messages
    old_context = memory.get_context()[:20]
    summary_prompt = "Résume cette conversation en 2-3 phrases."
    summary = openai.chat.completions.create(...)
    # Remplacez les 20 messages par le résumé

if len(user_message) < 50 and "?" not in user_message:
    model = "gpt-3.5-turbo"
else:
    model = "gpt-4-turbo"

import hashlib

def get_cache_key(message):
    return hashlib.md5(message.encode()).hexdigest()

# Avant l'appel LLM :
cache_key = get_cache_key(user_message)
if cache_key in response_cache:
    return response_cache[cache_key]

moderation = openai.moderations.create(input=user_message)
if moderation.results[0].flagged:
    await update.message.reply_text("❌ Message inapproprié détecté.")
    return