ما نبنيه
بوتان صغيران، وعقل مشترك واحد. ملف يُسمى llm.py يملك كل شيء يتصل بالـ API: الشخصية، مقيد المعدل، ومنطق إعادة المحاولة. يقوم بوت Discord وبوت Telegram بعد ذلك بترجمة أحداث المنصة إلى قائمة رسائل والعكس. أضف منصة ثالثة لاحقاً وستكتب فقط المحوّل.
ثبّت التبعيات باستخدام pip install httpx discord.py python-telegram-bot. صدّر API_KEY من صفحة حسابك، بالإضافة إلى DISCORD_TOKEN و/أو TELEGRAM_TOKEN من إعدادات البوت الخاصة بكل منصة. يفترض كلا البوتين أدناه استخدام Python 3.10 أو أحدث.
صيغة الطلب هي شكل OpenAI chat-completions العادي، مغطى في الدليل السريع متعدد اللغات. لا يحتاج أي شيء هنا إلى SDK.
الوحدة الأساسية المشتركة: الشخصية، المقيد وإعادة المحاولة
ضع الجزء الخطير في مكان واحد. يحتفظ الوحدة أدناه بنافذة منزلقة لأوقات الطلبات ويتوقّف عندما يقترب العدد من 240 طلبًا في الدقيقة، وهو هامش متعمد تحت حد 300 طلب في الدقيقة المسموح لكل مفتاح. يحدّ المسمّك (semaphore) من التوازي إلى أربعة حتى لا يُطلق قناة مزدحمة عشرين طلبًا في آن واحد.
# llm.py - shared by both bots
import asyncio
import os
import time
from collections import deque
import httpx
URL = "https://api.unrestrictedaiapi.com/v1/chat/completions"
PERSONA = "You are Pip, a sardonic bar-room storyteller. Keep replies under 150 words."
_stamps = deque() # timestamps of recent calls
_gate = asyncio.Semaphore(4) # at most 4 requests in flight
_client = httpx.AsyncClient(timeout=60.0)
async def _wait_for_slot(limit=240, window=60.0):
# Stay below the 300 requests/minute per-key limit with headroom.
while True:
now = time.monotonic()
while _stamps and now - _stamps[0] > window:
_stamps.popleft()
if len(_stamps) < limit:
_stamps.append(now)
return
await asyncio.sleep(window - (now - _stamps[0]) + 0.05)
async def chat(history):
"""history: list of {"role","content"} dicts, oldest first."""
messages = [{"role": "system", "content": PERSONA}] + list(history)
async with _gate:
for attempt in range(3):
await _wait_for_slot()
r = await _client.post(
URL,
headers={"Authorization": f"Bearer {os.environ['API_KEY']}"},
json={"model": "uncensored", "messages": messages, "max_tokens": 400},
)
if r.status_code in (429, 503):
await asyncio.sleep(2 ** attempt)
continue
if r.status_code == 403:
return "I can't continue with that."
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
return "The service is busy. Try again in a moment."انظر إلى معالجة الحالة. 429 أو 503 ينام لمدة 1، ثم 2، ثم 4 ثوانٍ ويحاول مجددًا. 403 يعني أن فلتر المحتوى قد عمل، لذا يجيب البوت بسطر محايد بدل إعادة المحاولة. أي خطأ آخر يرمي استثناءً، والذي ستسجله مكتبة منصتك. إذا رأيت 402، رصيدك فارغ أو انتهت التجربة؛ قم بشحن الرصيد المسبق الدفع.
محوّل Discord
تحتاج بوتات Discord إلى تفعيل قصد محتوى الرسالة في بوابة المطورين وكذلك في الكود، أو يصل msg.content فارغًا. البوت يجيب فقط عند الإشارة إليه، مما يحافظ عليه من قراءة كل رسالة في خادم مزدحم.
# discord_bot.py
import os
import time
from collections import defaultdict, deque
import discord
from llm import chat
intents = discord.Intents.default()
intents.message_content = True # also enable it in the developer portal
bot = discord.Client(intents=intents)
HISTORY_TURNS = 10 # user+assistant messages kept per channel
history = defaultdict(lambda: deque(maxlen=HISTORY_TURNS))
last_use = {} # user id -> last request time
COOLDOWN = 5.0 # seconds between requests per user
def chunks(text, size=1900):
return [text[i:i + size] for i in range(0, len(text), size)] or [""]
@bot.event
async def on_message(msg: discord.Message):
if msg.author.bot or bot.user not in msg.mentions:
return
# Age gate: guild text channels must be flagged NSFW. No DMs.
if not isinstance(msg.channel, discord.TextChannel) or not msg.channel.is_nsfw():
await msg.reply("I only chat in channels marked age-restricted (NSFW).")
return
now = time.monotonic()
if now - last_use.get(msg.author.id, 0) < COOLDOWN:
await msg.add_reaction("\u23f3")
return
last_use[msg.author.id] = now
text = msg.clean_content.replace(f"@{bot.user.display_name}", "").strip()
if not text:
return
h = history[msg.channel.id]
h.append({"role": "user", "content": f"{msg.author.display_name}: {text}"})
async with msg.channel.typing():
answer = await chat(h)
h.append({"role": "assistant", "content": answer})
for part in chunks(answer):
await msg.channel.send(part)
bot.run(os.environ["DISCORD_TOKEN"])السجل هو deque بـ maxlen=10 مرتبط بمعرف القناة، لذا كل قناة هي محادثة منفصلة وتنزح الأدوار القديمة تلقائيًا. إضافة اسم عرض المتحدث كبادئة لكل سطر مستخدم تتيح للنموذج التمييز بين الأشخاص في غرفة مشتركة. عشرة رسائل هي افتراض محافظ؛ نافذة السياق هي 100,000 رمز، لذا يمكنك رفعها كثيرًا قبل أن يهم الأمر.
محوّل Telegram
python-telegram-bot v20 وما بعده غير متزامن، مما يتوافق بشكل طبيعي مع الوحدة المشتركة. chat_data هو قاموس لكل محادثة تحتفظ به المكتبة لك، لذا لا تحتاج السجل وحالة التبريد إلى كود تخزين إضافي.
# telegram_bot.py (python-telegram-bot v20+)
import os
import time
from collections import deque
from telegram import Update
from telegram.ext import (ApplicationBuilder, CommandHandler, ContextTypes,
MessageHandler, filters)
from llm import chat
async def adult(update: Update, context: ContextTypes.DEFAULT_TYPE):
context.chat_data["adult"] = True
await update.message.reply_text("Noted. Say something.")
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"This bot produces adult fiction. Send /adult only if you are 18 or older."
)
async def talk(update: Update, context: ContextTypes.DEFAULT_TYPE):
cd = context.chat_data
if not cd.get("adult"):
await update.message.reply_text("Send /adult to confirm you are 18+.")
return
if time.monotonic() - cd.get("last", 0) < 4:
return # per-chat cooldown
cd["last"] = time.monotonic()
h = cd.setdefault("history", deque(maxlen=10))
h.append({"role": "user", "content": update.message.text})
await context.bot.send_chat_action(update.effective_chat.id, "typing")
answer = await chat(h)
h.append({"role": "assistant", "content": answer})
for i in range(0, len(answer), 4000): # Telegram limit is 4096 chars
await update.message.reply_text(answer[i:i + 4000])
app = ApplicationBuilder().token(os.environ["TELEGRAM_TOKEN"]).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("adult", adult))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, talk))
app.run_polling()يُعدّ run_polling() أبسط طريقة للبدء ولا يحتاج إلى عنوان URL عام. ضع في اعتبارك أن chat_data يعيش في الذاكرة، لذا فإن إعادة التشغيل تمسحه. إذا كنت تحتاج إلى حفظ السجل بعد إعادة التشغيل، احفظ محتويات deque بنفسك في ملف أو قاعدة بيانات.
فلتر العمر وقاعدة قناة NSFW
هذا الـ API مخصص للبالغين فقط، وأي شيء تبنيه عليه يرث هذه المسؤولية. تعامل مع فلتر العمر كميزة، وليس كفكرة لاحقة.
- Discord: اجب فقط في قنوات السيرفر التي تحمل علامة مقيد العمر. الفحص
channel.is_nsfw()لا يكلف شيئاً ويرفض البوت في أماكن أخرى، بما في ذلك الرسائل المباشرة. - Telegram: اطلب تأكيداً صريحاً عبر الأمر
/adultلكل محادثة قبل أي توليد، وحدد القاعدة في نص الترحيب. - كلاهما: لا يقدمان البوت في المساحات الموجهة للشباب، ويجب إزالة الوصول فورًا إذا علمت أن المستخدم يقل عمره عن 18 عامًا.
أمر التأكيد هو حاجز، وليس إثباتاً للعمر. محتوى البالغين الذي يتضمن قاصرين يتم حظره دائماً بواسطة الـ API برمز 403، بما في ذلك الخيال الأدبي ولعب الأدوار، ويجب ألا يحاول البوت تجاوز هذا الرد.
تشغيله واختباره بدون حشد
ابدأ كل بوت في طرفية منفصلة باستخدام python discord_bot.py أو python telegram_bot.py. لأول تشغيل، أنشئ خادم اختبار خاص أو محادثة مع نفسك فقط. امشِ عبر قائمة تحقق قصيرة وستلتقط تقريباً كل خطأ للمبتدئين.
- أشر إلى البوت في قناة غير NSFW وتأكد من رفضه. هذا يثبت أن الحاجز يعمل.
- اذكره في قناة محددة وبعث رسالتين بسرعة. يجب أن تحصل الرسالة الثانية على تفاعل الساعة الرملية، مما يثبت أن فترة الانتظار تعمل.
- أجرِ محادثة مكونة من أربع رسائل، ثم ارجع إلى الرسالة الأولى. إذا تذكر البوت، فإن السجل متصل بشكل صحيح.
- اضبط مفتاح الـ API مؤقتاً على قيمة خاطئة وتأكد من ظهور الفشل في سجلاتك بدلاً من اختفائه بصمت.
- الصق كتلة نصية طويلة وشاهد الرد ينقسم إلى رسائل متعددة تحت حد طول المنصة.
إذا بدت الردات عامة، فإن الشخصية هي عادة المشكلة بدلاً من الكود. غيّر جملة واحدة من PERSONA، أعد التشغيل، وقارن. لأن الشخصية تعيش في الوحدة الأساسية، فإن تعديل واحد يغير كلا البوتين في آن واحد، وهو بالضبط سبب جدوى الهيكلية مقابل الملف الإضافي.
عندما تكون جاهزاً للنشر، أي مدير عمليات دائم التشغيل سيعمل. البوتات تحتفظ باتصال طويل الأمد مع منصتها وتقوم بطلبات HTTPS قصيرة الصادرة إلى الـ API، لذا لا تحتاج إلى منافذ داخلة وتستهلك ذاكرة قليلة جداً. أضف سياسة إعادة التشغيل حتى لا يترك التعطل مجتمعك يتحدث إلى بوت صامت.
حدود المعدل، حجم السجل والتكلفة
ثلاثة مقابض تتحكم في سلوك البوت تحت الحمل. يوقف تبريد كل مستخدم شخصًا واحدًا من احتكار المفتاح. تحمي النافذة العالمية سقف 300 طلب في الدقيقة. يُمهّد المُعدِّل الانبعاثات. إذا شغلت عدة عمليات بوت على مفتاح واحد، فإنها تشترك في ذلك السقف، لذا خفّض limit لكل عملية وفقًا لذلك.
قدّر الإنفاق قبل دعوة الحشود. الافتراضات: كل استدعاء يرسل حوالي 1,200 رمز إدخال (الشخصية وعشرة أدوار) ويعيد 250 رمزًا. تكلفة الاستدعاء الواحد هي 1,200 × $0.25 / 1,000,000 + 250 × $1.00 / 1,000,000 = $0.0003 + $0.00025 = $0.00055. خادم يُنتج 2,000 ردًا يوميًا سينفق حوالي $1.10 يوميًا تحت هذه الافتراضات. أرقامك ستختلف، لذا سجّل حقل usage لمدة أسبوع وأعد الحساب.
لضبط الشخصية نفسها، اقرأ دليل الموجّه. الأسعار مدرجة على صفحة التسعير.
تقوية البوت قبل مشاركته
أبقِ الرموز خارج التحكم في المصدر؛ اقرأها من البيئة كما هو موضح. حد طول الرسالة قبل إرسالها إلى الـ API حتى لا تحرق الرصيد بلصقة ضخمة واحدة. سجل الأخطاء ولكن ليس نص الرسالة، لأن مستخدميك يتوقعون معقولاً أن تبقى محادثاتهم خاصة. أضف أمر /reset الذي يمسح deque القناة، وهو أسرع حل عندما تنحرف المحادثة. أخيرًا، ضع ملاحظة مرئية في ملف تعريف البوت تقول إنه يكتب خيالًا للكبار وأن ردوده مولدة.
يحتاج خيار تصميم آخر إلى جملة: البوتات تنتظر الرد الكامل بدلاً من بثه. تحرير رسالة محادثة رمزاً تلو الآخر يؤدي مباشرة إلى حدود تحرير المنصة، وبالنسبة للردود المقربة من 150 كلمة، فإن الانتظار قصير. إذا أردت لاحقاً الكتابة الحية في واجهة الويب الخاصة بك، فعّل stream: true هناك، حيث تتحكم في النقل. حتى ذلك الحين، مؤشر الكتابة الذي ترسله بالفعل هو أرخص طريقة لإظهار أن البوت يعمل، ويحافظ على الكود قصيراً بما يكفي للقراءة في جلسة واحدة ومراجعته قبل دعوة أي شخص.
