ساخت ربات تلگرام فارسی با GPT در ۳۰ دقیقه — پایتون، استریم پاسخ، کنترل هزینه

ربات تلگرامی که با مدل زبانی جواب می‌دهد، پرتکرارترین پروژهٔ اول برنامه‌نویس‌های ایرانی است و به همان اندازه هم پرتکرارترین جایی است که هزینه از کنترل خارج می‌شود. در این مقاله یک ربات کامل می‌سازیم که پاسخ را همان‌طور که تولید می‌شود در پیام تلگرام نشان می‌دهد، چند پیام آخر هر کاربر را به‌عنوان حافظه نگه می‌دارد، و سه سازوکار برای کنترل هزینه دارد. کل کد زیر صد خط است و با python-telegram-bot نسخهٔ ۲۲ و SDK رسمی openai نوشته شده.

قبل از شروع یک نکتهٔ مهم که خیلی از آموزش‌ها نمی‌گویند: سرور ایران به api.telegram.org دسترسی ندارد. یعنی رباتت باید یا روی یک VPS خارج از ایران اجرا شود، یا از داخل ایران از طریق یک relay (مثلاً یک nginx روی VPS خارجی که فقط تلگرام را عبور می‌دهد) به تلگرام برسد. خبر خوب اینکه API مدل‌ها روی api.uttapen.ir از هر دو طرف قابل دسترس است، پس محل اجرای ربات را فقط تلگرام تعیین می‌کند، نه مدل. در کد، آدرس API تلگرام از متغیر محیطی خوانده می‌شود تا هر دو حالت بدون تغییر کد کار کند.

آنچه لازم داری

• توکن ربات از BotFather (دستور /newbot).
• یک کلید sk-up-… از داشبورد uttapen با مقداری شارژ؛ برای تست چند هزار تومان کافی است.
• پایتون ۳٫۱۱ به بالا و دو بسته: pip install python-telegram-bot openai
• جایی برای اجرا که به تلگرام برسد (بند بالا).

انتخاب مدل و برآورد هزینه

برای ربات عمومی، مدل باید سریع، ارزان و در فارسی روان باشد. پیش‌فرض کد openai/gpt-5-mini است. اگر رباتت پرترافیک است یا پاسخ‌های ساده می‌خواهی، openai/gpt-5-nano چند برابر ارزان‌تر است و برای پرسش‌وپاسخ کوتاه فارسی معمولاً کافی است. deepseek/deepseek-v4-flash هم گزینهٔ ارزان دیگری است.

یک پاسخ معمولی ربات حدود ۴۰۰ توکن ورودی (system prompt به‌علاوهٔ چند پیام قبلی) و ۲۰۰ توکن خروجی مصرف می‌کند. با قیمت‌های امروز کاتالوگ، هر پاسخ با gpt-5-mini چیزی حدود ۶۵ تومان و با gpt-5-nano حدود ۱۳ تومان درمی‌آید؛ یعنی هزار پاسخ در روز به‌ترتیب حدود ۶۵ هزار و ۱۳ هزار تومان. این‌ها برآورد است — عدد قطعی هر پاسخ را خودِ ربات از رویداد meta می‌خواند و لاگ می‌کند، که پایین‌تر می‌بینی.

کد ربات

# bot.py — ربات تلگرام فارسی با استریم پاسخ و کنترل هزینه
# pip install python-telegram-bot openai
import asyncio, os, time, logging
from collections import defaultdict, deque
from openai import AsyncOpenAI
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes, filters

logging.basicConfig(level=logging.INFO)
llm = AsyncOpenAI(
    base_url=os.getenv("UTTAPEN_BASE_URL", "https://api.uttapen.ir/v1"),
    api_key=os.environ["UTTAPEN_API_KEY"],
)
MODEL = os.getenv("BOT_MODEL", "openai/gpt-5-mini")
MAX_OUT = 500            # سقف توکن خروجی هر پاسخ = سقف هزینه
HISTORY = 6              # فقط ۶ پیام آخر هر کاربر به مدل می‌رود
EDIT_EVERY = 1.0         # ویرایش پیام تلگرام حداکثر هر ۱ ثانیه (محدودیت تلگرام)
SYSTEM = "تو دستیار فارسی‌زبان و مختصر هستی. جواب‌های کوتاه و دقیق بده؛ اگر مطمئن نیستی بگو نمی‌دانم."

history: dict[int, deque] = defaultdict(lambda: deque(maxlen=HISTORY))

async def stream_answer(chat_id: int, user_text: str):
    """پاسخ مدل را تکه‌تکه yield می‌کند و در پایان هزینهٔ تومانی همین درخواست را."""
    history[chat_id].append({"role": "user", "content": user_text})
    stream = await llm.chat.completions.create(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM}, *history[chat_id]],
        max_tokens=MAX_OUT,
        stream=True,
        extra_headers={"X-Uttapen-Include-Meta": "1"},   # رویداد uttapen.meta قبل از [DONE]
    )
    full, cost = "", None
    async for chunk in stream:
        if getattr(chunk, "object", "") == "uttapen.meta":
            cost = chunk.model_extra.get("cost_toman")
            continue
        delta = chunk.choices[0].delta.content if chunk.choices else None
        if delta:
            full += delta
            yield full, None
    history[chat_id].append({"role": "assistant", "content": full})
    yield full, cost

async def on_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    sent = await update.message.reply_text("…")
    last_edit, last_text = 0.0, ""
    try:
        async for text, cost in stream_answer(chat_id, update.message.text):
            now = time.monotonic()
            if cost is None and (now - last_edit < EDIT_EVERY or text == last_text):
                continue
            last_edit, last_text = now, text
            await sent.edit_text(text[:4000] or "…")
        if cost is not None:
            logging.info("chat=%s cost=%s toman", chat_id, cost)
    except Exception as e:
        logging.exception("llm failed")
        await sent.edit_text("الان نتوانستم جواب بدهم؛ چند لحظه بعد دوباره بپرس.")

async def on_reset(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    history.pop(update.effective_chat.id, None)
    await update.message.reply_text("حافظهٔ گفتگو پاک شد.")

def main():
    app = (Application.builder()
           .token(os.environ["TELEGRAM_BOT_TOKEN"])
           .base_url(os.getenv("TELEGRAM_API_BASE", "https://api.telegram.org/bot"))  # relay خارج از ایران
           .build())
    app.add_handler(CommandHandler("reset", on_reset))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_message))
    app.run_polling()

if __name__ == "__main__":
    main()

اجرا:

export TELEGRAM_BOT_TOKEN=123456:ABC...
export UTTAPEN_API_KEY=sk-up-...
# اگر از ایران و از طریق relay به تلگرام می‌رسی:
# export TELEGRAM_API_BASE=https://your-relay.example.com/bot
python bot.py

کد چطور کار می‌کند

استریم و ویرایش پیام. به‌جای اینکه کاربر ده ثانیه منتظر بماند، اول یک پیام «…» می‌فرستیم و بعد همان پیام را با متنِ تا این لحظه ویرایش می‌کنیم. تلگرام برای editMessageText محدودیت نرخ دارد و اگر برای هر توکن ویرایش کنی، خطای flood می‌گیری. برای همین EDIT_EVERY ویرایش را به یکی در ثانیه محدود می‌کند و اگر متن تغییری نکرده باشد، اصلاً ویرایش نمی‌فرستد. ویرایش آخر بعد از پایان استریم همیشه انجام می‌شود، چون در آن مرحله cost مقدار دارد و شرط throttle نادیده گرفته می‌شود.

پیام‌های خام، نه Markdown. عمداً parse_mode نمی‌فرستیم. وقتی متن وسط استریم ویرایش می‌شود، ممکن است یک بلوک کد یا * باز مانده باشد و تلگرام کل ویرایش را رد کند. اگر قالب‌بندی می‌خواهی، فقط در ویرایش آخر parse_mode بده و متن را قبلش escape کن.

رویداد meta برای هزینه. هدر X-Uttapen-Include-Meta: 1 باعث می‌شود قبل از [DONE] یک رویداد اضافه با object برابر uttapen.meta بیاید که cost_toman، balance_toman و hold_toman دارد. SDK این رویداد را مثل یک chunk معمولی تحویل می‌دهد؛ ما آن را با object تشخیص می‌دهیم و از حلقهٔ متن جدا می‌کنیم. عدد cost_toman همان چیزی است که از کیف پول کم شده، پس لاگ آن برابر با صورتحساب واقعی ربات است. بدون این هدر، استریم دقیقاً فرمت OpenAI است؛ مستندات استریم جزئیات را دارد.

حافظهٔ گفتگو. برای هر چت یک deque با طول ثابت داریم. شش پیام آخر (سه رفت‌وبرگشت) کافی است تا ربات یادش بماند دربارهٔ چه حرف می‌زدید، و در عین حال هزینهٔ ورودی هر درخواست را محدود نگه می‌دارد. دستور /reset این حافظه را پاک می‌کند. این حافظه در RAM است و با restart می‌پرد؛ برای ربات واقعی آن را در Redis یا SQLite بگذار.

خطا بدون ترکیدن. هر مشکلی در تماس با مدل (موجودی ناکافی، محدودیت نرخ، خطای upstream) گرفته می‌شود، در لاگ می‌آید و کاربر یک پیام فارسی معقول می‌بیند. فهرست کدهای خطا و معنی هر کدام در مستندات خطاها هست؛ حداقل 402 را جدا هندل کن تا وقتی شارژ تمام شد، بدانی چرا ربات ساکت شده.

سه قفل برای کنترل هزینه

1. max_tokens — سقف خروجی هر پاسخ. با ۵۰۰ توکن، هیچ پاسخی نمی‌تواند بیشتر از حدود ۱۳۰ تومان با gpt-5-mini هزینه داشته باشد، هرچقدر هم کاربر مدل را به پرحرفی وادار کند. این عدد را بر اساس کاربرد تنظیم کن؛ برای پرسش‌وپاسخ کوتاه ۳۰۰ هم زیاد است.
2. پنجرهٔ حافظه — بزرگ‌ترین منبع هزینهٔ پنهان در ربات‌های چت، فرستادن کل تاریخچه در هر درخواست است. با پنجرهٔ ثابت، هزینهٔ ورودی هم ثابت می‌ماند.
3. سقف ماهانهٔ کلید — در داشبورد برای کلید ربات یک سقف تومانی بگذار. اگر رباتت ویروسی شد یا کسی حلقه‌ای برایش ساخت، بعد از سقف 429 می‌گیرد و کیف پول اصلی دست‌نخورده می‌ماند. برای هر ربات یک کلید جدا بساز تا مصرفش در گزارش‌ها تفکیک شود.

یک قفل چهارم که در کد نیست ولی برای ربات عمومی لازم است: محدودیت نرخ به‌ازای هر کاربر تلگرام (مثلاً ۲۰ پیام در ساعت). بدون آن، یک نفر می‌تواند سهم روزانهٔ همه را مصرف کند.

نکته‌های فارسی

• system prompt را فارسی بنویس؛ مدل‌ها وقتی دستور به زبان پاسخ باشد، کمتر به انگلیسی می‌پرند.
• اگر می‌خواهی اعداد در پاسخ فارسی باشند، صریح در system prompt بگو؛ پیش‌فرض بیشتر مدل‌ها ارقام لاتین است.
• تلگرام متن راست‌به‌چپ را خودش درست نشان می‌دهد، ولی وقتی پاسخ ترکیب فارسی و کد است، بلوک کد را در ویرایش آخر با parse_mode="MarkdownV2" بفرست تا چپ‌چین شود.
• طول پیام تلگرام حداکثر ۴٬۰۹۶ کاراکتر است؛ کد پاسخ را در ۴٬۰۰۰ می‌بُرد. اگر پاسخ‌های بلند می‌خواهی، ادامه را در پیام دوم بفرست.

چند سناریوی رایج که ربات‌ها را خراب می‌کند

دو پیام پشت سر هم. کاربر هنوز جواب اول کامل نشده، دومی را می‌فرستد. با کد بالا هر دو هم‌زمان پردازش می‌شوند و تاریخچه به هم می‌ریزد، چون پیام دوم قبل از ثبت پاسخ اول به حافظه اضافه می‌شود. ساده‌ترین راه، یک asyncio.Lock به‌ازای هر چت است: تا پاسخ قبلی تمام نشده، پیام بعدی منتظر می‌ماند. راه بهتر برای ربات پرترافیک، صف کردن پیام‌های هر کاربر و ادغام آن‌ها در یک درخواست.

پیام خیلی بلند. کسی یک مقالهٔ ده‌هزارکلمه‌ای paste می‌کند. هزینهٔ ورودی این پیام به‌تنهایی از ده‌ها پاسخ معمولی بیشتر است و تا وقتی در پنجرهٔ حافظه بماند، در هر درخواست بعدی هم تکرار می‌شود. قبل از افزودن به تاریخچه، طول پیام را محدود کن (مثلاً ۲٬۰۰۰ کاراکتر) و اگر بلندتر بود، به کاربر بگو خلاصه‌اش کند.

ربات در گروه. در گروه، ربات به‌طور پیش‌فرض همهٔ پیام‌ها را نمی‌بیند، ولی اگر privacy mode را خاموش کنی، هر پیام گروه یک درخواست به مدل می‌شود و هزینه چند برابر می‌شود. در گروه فقط به پیام‌هایی جواب بده که ربات را mention کرده‌اند یا reply به پیام خودش هستند.

پیام غیرمتنی. عکس، استیکر و voice با فیلتر filters.TEXT نادیده گرفته می‌شوند و کاربر بی‌جواب می‌ماند. حداقل یک handler اضافه کن که بگوید فعلاً فقط متن می‌فهمد. اگر بعداً خواستی عکس را هم بفهمد، همان مدل‌های vision کاتالوگ با یک پیام image_url کار را انجام می‌دهند.

قطع شدن استریم. اگر اتصال به API وسط پاسخ قطع شود، متنِ تا آن لحظه در پیام تلگرام مانده و ناقص است. در بلوک except به‌جای پیام خطای عمومی، می‌توانی متن ناقص را نگه داری و یک خط «پاسخ ناتمام ماند» به انتهایش اضافه کنی تا کاربر بداند چه شد. هزینهٔ بخش تولیدشده تا لحظهٔ قطع محاسبه می‌شود، نه کل max_tokens.

از تست تا production

این کد با polling کار می‌کند که برای شروع و برای VPS بدون دامنه ساده‌ترین راه است. برای ترافیک بالا webhook بهتر است؛ python-telegram-bot هر دو را با یک تغییر در main پشتیبانی می‌کند. حافظه را به Redis ببر، برای هر کاربر rate limit بگذار، و لاگ هزینه را جایی جمع کن که بتوانی روزانه نگاهش کنی. گزارش مصرف تفکیک‌شده به روز و مدل هم در داشبورد هست، پس اگر لاگ خودت را گم کردی، مرجع آنجاست.

دو محدودیت را هم صادقانه بگوییم. اول، مسیر مدل از ایران به relay خارج از کشور و بعد به upstream می‌رود و حدود صد میلی‌ثانیه تأخیر اضافه دارد؛ در ربات تلگرام که خودش چند صد میلی‌ثانیه تأخیر شبکه دارد، عملاً حس نمی‌شود. دوم، متن پیام کاربران به ارائه‌دهندهٔ مدل می‌رسد؛ gateway چیزی ذخیره نمی‌کند، ولی اگر رباتت دادهٔ حساس می‌گیرد، این را به کاربرانت بگو.

بخش مدل این کد (تابع stream_answer) روی gateway اجرا و تست شده و رویداد meta همان‌طور که توضیح دادیم می‌رسد. بخش تلگرام الگوی استاندارد python-telegram-bot نسخهٔ ۲۲ است. اگر همین حالا شروع می‌کنی: توکن از BotFather، کلید از داشبورد، و سی دقیقه بعد رباتت جواب می‌دهد.