🤖 آموزش ساخت ربات تلگرامی پولساز با مدل‌های زبانی LLM | راهنمای جامع از صفر تا صد

فصل اول: مدل‌های زبانی بزرگ (LLM) چیست و چرا برای ربات تلگرام انقلابی هستند؟

۱.۱ مدل زبانی بزرگ (LLM) چیست؟

مدل‌های زبانی بزرگ یا Large Language Models (LLM)، سیستم‌های هوش مصنوعی پیشرفته‌ای هستند که بر روی حجم عظیمی از داده‌های متنی (کتاب‌ها، مقالات، کدها، وبسایت‌ها) آموزش دیده‌اند. این مدل‌ها می‌توانند:

• متن‌های طبیعی و روان مانند یک انسان تولید کنند
• سوالات پیچیده را درک کرده و پاسخ‌های دقیق بدهند
• خلاصه‌سازی، ترجمه، تحلیل احساسات و بسیاری کارهای دیگر انجام دهند
• کدنویسی، ریاضیات و منطق را به خوبی درک کنند
• مکالمات طولانی را به خاطر بسپارند و ادامه دهند

معروف‌ترین LLMها عبارتند از: GPT-4 (ChatGPT)، Claude (Anthropic)، Gemini (Google)، Llama (Meta) و Mistral. هر کدام مزایا و معایب خاص خود را دارند.

۱.۲ مقایسه بهترین LLMها برای ربات تلگرام

۱.۳ چرا ربات تلگرام با LLM موفق‌تر است؟

ربات‌های سنتی تلگرام معمولاً بر اساس کلمات کلیدی و پاسخ‌های از پیش تعیین شده کار می‌کنند. اما ربات‌های مبتنی بر LLM:

• مکالمه طبیعی: مثل یک انسان واقعی صحبت می‌کنند و کاربر حس می‌کند با یک شخص حرف می‌زند
• درک بهتر: حتی اگر کاربر سوال را اشتباه یا نامفهوم بپرسد، ربات متوجه منظور می‌شود
• یادگیری مداوم: با هر مکالمه، پاسخ‌های بهتری ارائه می‌دهد (در صورت استفاده از حافظه)
• قابلیت‌های بی‌نهایت: از مشاوره روانشناسی تا حل مسئله ریاضی، هر کاری می‌تواند انجام دهد

فصل دوم: گام‌های عملی ساخت ربات تلگرام با LLM

pip install python-telegram-bot==20.3
pip install openai
pip install anthropic
pip install google-generativeai
pip install requests
pip install python-dotenv
                

import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters

# راه‌اندازی لاگر
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)

# توکن ربات خود را اینجا قرار دهید
TOKEN = 'YOUR_BOT_TOKEN_HERE'

async def start(update: Update, context):
    """دستور /start"""
    await update.message.reply_text('سلام! من یک ربات هوشمند هستم. چطور می‌توانم کمک کنم؟')

async def echo(update: Update, context):
    """پاسخ به پیام‌های متنی"""
    user_message = update.message.text
    await update.message.reply_text(f'شما گفتید: {user_message}')

def main():
    # ساخت اپلیکیشن
    app = Application.builder().token(TOKEN).build()
    
    # اضافه کردن هندلرها
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    
    # شروع ربات
    print("ربات در حال اجراست...")
    app.run_polling()

if __name__ == '__main__':
    main()
                

import openai
import os
from dotenv import load_dotenv

load_dotenv()  # بارگذاری متغیرهای محیطی

# دریافت API Key از متغیر محیطی
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
openai.api_key = OPENAI_API_KEY

async def handle_message(update: Update, context):
    user_message = update.message.text
    
    # نمایش پیام کاربر در ترمینال
    print(f"User: {user_message}")
    
    try:
        # ارسال درخواست به GPT
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",  # یا "gpt-4" برای کیفیت بالاتر
            messages=[
                {"role": "system", "content": "تو یک دستیار هوشمند و مفید هستی. به زبان فارسی پاسخ بده."},
                {"role": "user", "content": user_message}
            ],
            max_tokens=500,
            temperature=0.7
        )
        
        bot_response = response.choices[0].message.content
        
        # ارسال پاسخ به کاربر
        await update.message.reply_text(bot_response)
        
    except Exception as e:
        print(f"خطا: {e}")
        await update.message.reply_text("متاسفانه مشکلی پیش آمده. لطفاً دوباره تلاش کنید.")
                

فصل سوم: روش‌های درآمدزایی از ربات تلگرام هوشمند

فصل چهارم: جدول مقایسه هزینه و درآمد مدل‌های مختلف

فصل پنجم: چالش ۷ روزه راه‌اندازی ربات پولساز

فصل ششم: کد کامل یک ربات تلگرام حرفه‌ای با سیستمی‌پرداخت

import os
import json
import sqlite3
import logging
from datetime import datetime, timedelta
from dotenv import load_dotenv
import openai
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters, ContextTypes

# بارگذاری متغیرهای محیطی
load_dotenv()

# تنظیمات
TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
ADMIN_ID = int(os.getenv('ADMIN_ID', '0'))

openai.api_key = OPENAI_API_KEY

# راه‌اندازی دیتابیس
def init_db():
    conn = sqlite3.connect('bot_database.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS users
                 (user_id INTEGER PRIMARY KEY,
                  username TEXT,
                  first_name TEXT,
                  subscription_end TEXT,
                  message_count INTEGER DEFAULT 0,
                  join_date TEXT)''')
    c.execute('''CREATE TABLE IF NOT EXISTS messages
                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
                  user_id INTEGER,
                  role TEXT,
                  content TEXT,
                  timestamp TEXT)''')
    conn.commit()
    conn.close()

init_db()

# توابع کمکی
def get_user(user_id):
    conn = sqlite3.connect('bot_database.db')
    c = conn.cursor()
    c.execute("SELECT * FROM users WHERE user_id = ?", (user_id,))
    user = c.fetchone()
    conn.close()
    return user

def add_user(user_id, username, first_name):
    conn = sqlite3.connect('bot_database.db')
    c = conn.cursor()
    join_date = datetime.now().isoformat()
    c.execute("INSERT OR IGNORE INTO users (user_id, username, first_name, join_date) VALUES (?, ?, ?, ?)",
              (user_id, username, first_name, join_date))
    conn.commit()
    conn.close()

def is_premium(user_id):
    user = get_user(user_id)
    if user and user[3]:  # subscription_end
        end_date = datetime.fromisoformat(user[3])
        return end_date > datetime.now()
    return False

def add_message_history(user_id, role, content):
    conn = sqlite3.connect('bot_database.db')
    c = conn.cursor()
    timestamp = datetime.now().isoformat()
    c.execute("INSERT INTO messages (user_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
              (user_id, role, content, timestamp))
    conn.commit()
    conn.close()

def get_recent_history(user_id, limit=10):
    conn = sqlite3.connect('bot_database.db')
    c = conn.cursor()
    c.execute("SELECT role, content FROM messages WHERE user_id = ? ORDER BY timestamp DESC LIMIT ?",
              (user_id, limit))
    messages = c.fetchall()
    conn.close()
    return list(reversed(messages))

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    add_user(user.id, user.username, user.first_name)
    
    welcome_text = f"""
🤖 سلام {user.first_name} عزیز!

به ربات هوشمند من خوش آمدی. من با استفاده از پیشرفته‌ترین مدل‌های هوش مصنوعی ساخته شده‌ام.

✨ قابلیت‌های من:
• پاسخ به هر سوالی در هر زمینه‌ای
• تولید محتوا، ترجمه، خلاصه‌سازی
• کدنویسی و حل مسائل ریاضی
• مکالمه طبیعی و دوستانه

💰 نحوه استفاده:
• نسخه رایگان: ۱۰ پیام در روز
• نسخه پریمیوم: نامحدود + پاسخ‌های دقیق‌تر

برای مشاهده تعرفه‌ها، دستور /pricing را ارسال کنید.
    """
    
    keyboard = [
        [InlineKeyboardButton("📊 تعرفه و اشتراک", callback_data='pricing')],
        [InlineKeyboardButton("❓ راهنما", callback_data='help')],
        [InlineKeyboardButton("👤 وضعیت من", callback_data='status')]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    
    await update.message.reply_text(welcome_text, parse_mode='HTML', reply_markup=reply_markup)

async def pricing(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    if query:
        await query.answer()
        message = query.message
    else:
        message = update.message
    
    pricing_text = """
💎 تعرفه اشتراک هوش ناب

🎁 نسخه رایگان
• ۱۰ پیام در روز
• پاسخ‌های پایه
• پشتیبانی معمولی
• هزینه: ۰ تومان

⭐ نسخه پریمیوم ماهانه
• پیام نامحدود
• پاسخ‌های دقیق و پیشرفته
• اولویت در پاسخگویی
• حفظ تاریخچه مکالمات
• هزینه: ۵۰,۰۰۰ تومان / ماه

👑 نسخه ویژه سالانه
• تمام امکانات پریمیوم
• ۴ ماه رایگان
• مشاوره اختصاصی
• هزینه: ۵۰۰,۰۰۰ تومان / سال

برای خرید اشتراک، روی دکمه زیر کلیک کنید:
    """
    
    keyboard = [[InlineKeyboardButton("💰 خرید اشتراک ماهانه", callback_data='buy_monthly')],
                [InlineKeyboardButton("🌟 خرید اشتراک سالانه", callback_data='buy_yearly')]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    
    if query:
        await message.edit_text(pricing_text, parse_mode='HTML', reply_markup=reply_markup)
    else:
        await message.reply_text(pricing_text, parse_mode='HTML', reply_markup=reply_markup)

async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    if query:
        await query.answer()
        user_id = query.from_user.id
        message = query.message
    else:
        user_id = update.effective_user.id
        message = update.message
    
    premium = is_premium(user_id)
    user_data = get_user(user_id)
    
    if premium and user_data and user_data[3]:
        end_date = datetime.fromisoformat(user_data[3])
        days_left = (end_date - datetime.now()).days
        status_text = f"✅ وضعیت: پریمیوم فعال\n📅 زمان باقی‌مانده: {days_left} روز"
    else:
        status_text = "⚠️ وضعیت: رایگان\nبرای استفاده از امکانات کامل، اشتراک تهیه کنید."
    
    await message.reply_text(status_text, parse_mode='HTML')

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    user_message = update.message.text
    
    # بررسی محدودیت برای کاربران رایگان
    if not is_premium(user_id):
        conn = sqlite3.connect('bot_database.db')
        c = conn.cursor()
        today = datetime.now().date().isoformat()
        c.execute("SELECT COUNT(*) FROM messages WHERE user_id = ? AND date(timestamp) = ?",
                  (user_id, today))
        count = c.fetchone()[0]
        conn.close()
        
        if count >= 10:
            await update.message.reply_text("⚠️ شما به سقف پیام‌های روزانه خود رسیده‌اید. برای استفاده نامحدود، اشتراک پریمیوم تهیه کنید.\n/price")
            return
    
    # نمایش وضعیت تایپ
    await update.message.chat.send_action(action="typing")
    
    # ذخیره پیام کاربر
    add_message_history(user_id, "user", user_message)
    
    # دریافت تاریخچه مکالمه
    history = get_recent_history(user_id, 10)
    messages = [{"role": "system", "content": "تو یک دستیار هوشمند و حرفه‌ای هستی. به زبان فارسی پاسخ بده."}]
    
    for role, content in history:
        messages.append({"role": role, "content": content})
    
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=800 if is_premium(user_id) else 300,
            temperature=0.7
        )
        
        bot_response = response.choices[0].message.content
        
        # ذخیره پاسخ
        add_message_history(user_id, "assistant", bot_response)
        
        # اضافه کردن تبلیغات برای کاربران رایگان
        if not is_premium(user_id):
            bot_response += "\n\n---\n✨ برای حذف تبلیغات و پیام نامحدود، اشتراک پریمیوم تهیه کنید: /pricing"
        
        await update.message.reply_text(bot_response)
        
    except Exception as e:
        logging.error(f"Error: {e}")
        await update.message.reply_text("متاسفانه خطایی رخ داده است. لطفاً چند لحظه دیگر تلاش کنید.")

async def buy_subscription(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()
    
    plan = query.data  # buy_monthly یا buy_yearly
    
    if plan == 'buy_monthly':
        amount = 50000
        description = "اشتراک ماهانه ربات هوشمند"
    else:
        amount = 500000
        description = "اشتراک سالانه ربات هوشمند"
    
    # در اینجا کد اتصال به درگاه پرداخت (مثل Zarinpal) قرار می‌گیرد
    # به دلیل محدودیت فضا، کد پرداخت کامل قرار داده نمی‌شود
    
    await query.edit_message_text(
        f"💰 درخواست خرید {description} ثبت شد.\n"
        f"مبلغ: {amount:,} تومان\n\n"
        "به زودی لینک پرداخت برای شما ارسال خواهد شد.",
        parse_mode='HTML'
    )

def main():
    app = Application.builder().token(TOKEN).build()
    
    # هندلرها
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("pricing", pricing))
    app.add_handler(CommandHandler("status", status))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    app.add_handler(CallbackQueryHandler(pricing, pattern='pricing'))
    app.add_handler(CallbackQueryHandler(status, pattern='status'))
    app.add_handler(CallbackQueryHandler(buy_subscription, pattern='buy_'))
    
    print("ربات در حال اجراست...")
    app.run_polling()

if __name__ == '__main__':
    main()
            

کد کامل ربات حرفه‌ای با سیستم اشتراک، دیتابیس و تاریخچه مکالمه

فصل هفتم: سوالات متداول (FAQ)

📚 مقالات پیشنهادی هوش ناب