สิ่งที่เรากำลังสร้าง
บอทขนาดเล็กสองตัว ใช้สมองร่วมกัน ไฟล์ llm.py จัดการทุกสิ่งที่โต้ตอบกับ API: พรอมต์, ตัวจำกัดอัตรา, และตรรกะการเรียกซ้ำ บอท Discord และบอท Telegram ทำหน้าที่แปลงเหตุการณ์แพลตฟอร์มเป็นรายการข้อความและส่งกลับ เพิ่มแพลตฟอร์มที่สามในภายหลังและคุณเขียนเพียงตัวเชื่อมต่อ
ติดตั้ง dependencies ด้วย pip install httpx discord.py python-telegram-bot. ส่งออก API_KEY จาก หน้าบัญชีของคุณ รวมถึง DISCORD_TOKEN และ/หรือ TELEGRAM_TOKEN จากการตั้งค่าบอทของแต่ละแพลตฟอร์ม บอททั้งสองด้านล่างสมมติว่าใช้ Python 3.10 ขึ้นไป
รูปแบบคำขอคือรูปแบบ chat-completions แบบธรรมดา ซึ่งครอบคลุมใน คู่มือเริ่มต้นใช้งานหลายภาษา ไม่ต้องใช้ SDK
แกนกลางร่วมกัน: บุคลิก ตัวจำกัด และการลองใหม่
วางส่วนที่เสี่ยงในจุดเดียว โมดูลด้านล่างรักษาหน้าต่างเลื่อนของเวลาเกิดคำขอและรอเมื่อเข้าใกล้ 240 การเรียกต่อนาที ซึ่งเป็นขอบเขตที่จงใจไว้ต่ำกว่า 300 ต่อนาทีต่อคีย์ ตัวจำกัดความพร้อมใช้งานจำกัดความพร้อมใช้งานสูงสุดไว้ที่สี่ เพื่อให้ช่องที่วุ่นวายไม่สามารถยิงคำขอ 20 คำขอพร้อมกันได้
# 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 หมายถึงตัวกรองเนื้อหาทำงาน บอทจึงตอบด้วยข้อความที่เป็นกลางแทนที่จะลองใหม่ ข้อผิดพลาดอื่นจะ raise ซึ่งไลบรารีแพลตฟอร์มของคุณจะบันทึก หากเห็น 402 แสดงว่ายอดเงินหมดหรือเครดิตทดลองใช้หมด เติมยอดเงินแบบเติมเงินล่วงหน้า
ตัวปรับแต่ง Discord
บอท Discord ต้องเปิด intent เนื้อหาข้อความในพอร์ทัลผู้พัฒนาและในโค้ด มิฉะนั้น 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 แยกตาม id ช่อง ทำให้แต่ละช่องเป็นบทสนทนาของตัวเองและรอบการสนทนาเก่าจะหลุดออกโดยอัตโนมัติ การเติมชื่อแสดงของผู้ใช้ต่อหน้าข้อความแต่ละบรรทัดช่วยให้โมเดลแยกแยะผู้คนในห้องร่วมกันได้ ข้อความสิบข้อความเป็นค่าเริ่มต้นที่ปลอดภัย หน้าต่างบริบทรองรับ 100,000 โทเคน ดังนั้นคุณสามารถเพิ่มค่านี้ได้มากก่อนที่มันจะสำคัญ
ตัวปรับแต่ง Telegram
python-telegram-bot v20 ขึ้นไปเป็น async ซึ่งเข้าคู่กับโมดูลหลักได้อย่างเป็นธรรมชาติ 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: ตอบเฉพาะในช่อง guild ที่ทำเครื่องหมายว่าจำกัดอายุ การตรวจสอบ
channel.is_nsfw()ไม่มีค่าใช้จ่ายและบอทจะปฏิเสธที่อื่น รวมถึงข้อความส่วนตัว - Telegram: ต้องมีการยืนยัน
/adultที่ชัดเจนต่อแชทก่อนการสร้างข้อความใดๆ และระบุกฎในข้อความต้อนรับ - ทั้งสอง: อย่าเสนอบอทในพื้นที่ที่มุ่งเป้าไปที่เยาวชน และลบการเข้าถึงทันทีหากคุณทราบว่าผู้ใช้มีอายุต่ำกว่า 18 ปี
คำสั่งยืนยันเป็นประตู ไม่ใช่การพิสูจน์อายุ เนื้อหาทางเพศที่เกี่ยวข้องกับเด็กจะถูกบล็อกโดย API เสมอด้วย 403 รวมถึงนวนิยายและบทบาทสมมติ และบอทของคุณไม่ควรพยายามหลบเลี่ยงการตอบสนองนั้น
การรันและการทดสอบโดยไม่ใช้กลุ่มใหญ่
เริ่มบอทแต่ละตัวในเทอร์มินัลของตัวเองด้วย python discord_bot.py หรือ python telegram_bot.py สำหรับการรันครั้งแรก ให้สร้างเซิร์ฟเวอร์ทดสอบส่วนตัวหรือแชทที่มีเฉพาะคุณเท่านั้น เดินผ่านรายการตรวจสอบสั้นๆ และคุณจะจับข้อบกพร่องของผู้เริ่มต้นได้เกือบทั้งหมด
- กล่าวถึงบอทในช่องที่ไม่ใช่ NSFW และยืนยันว่ามันปฏิเสธ สิ่งนี้พิสูจน์ว่าประตูทำงาน
- กล่าวถึงมันในช่องที่ทำเครื่องหมายและส่งข้อความสองข้อความอย่างรวดเร็ว ข้อความที่สองควรได้รับปฏิกิริยานาฬิกาทราย ซึ่งพิสูจน์ว่าการหน่วงเวลาทำงาน
- ทำการสนทนาสี่ข้อความ จากนั้นอ้างอิงกลับไปที่ข้อความแรก หากบอทจำได้ ประวัติถูกต่อสายถูกต้อง
- ตั้งค่าคีย์ API เป็นค่าผิดชั่วคราวและยืนยันว่าความล้มเหลวปรากฏในบันทึกของคุณแทนที่จะหายไปอย่างเงียบๆ
- วางบล็อกข้อความยาวและดูการตอบกลับแยกออกเป็นข้อความหลายข้อความภายใต้ขีดจำกัดความยาวของแพลตฟอร์ม
หากการตอบกลับดูทั่วไป ปัญหาอาจอยู่ที่พรอมต์มากกว่าโค้ด แก้ไขประโยคเดียวใน PERSONA รีสตาร์ท แล้วเปรียบเทียบ เนื่องจากพรอมต์อยู่ในโมดูลร่วม การแก้ไขหนึ่งครั้งจะเปลี่ยนบอททั้งสองทันที ซึ่งเป็นเหตุผลที่โครงสร้างนี้คุ้มค่ากับไฟล์เพิ่มเติม
เมื่อพร้อมที่จะใช้งาน ตัวจัดการกระบวนการแบบ always-on ใดๆ ก็ใช้ได้ บอทมีการเชื่อมต่อระยะยาวกับแพลตฟอร์มและทำ HTTPS call แบบสั้นไปยัง API จึงไม่ต้องใช้พอร์ตขาเข้าและใช้หน่วยความจำน้อยมาก เพิ่มนโยบายรีสตาร์ทเพื่อให้การ crash ไม่ทำให้ชุมชนของคุณคุยกับบอทที่เงียบ
ขีดจำกัดอัตรา ขนาดประวัติ และค่าใช้จ่าย
ปุ่มควบคุมสามตัวควบคุมพฤติกรรมของบอทภายใต้ภาระ การเว้นระยะสำหรับผู้ใช้รายบุคคลหยุดไม่ให้บุคคลหนึ่งผูกขาดคีย์ หน้าต่างทั่วโลกปกป้องเพดาน 300 คำขอต่อนาที ตัวจำกัดความพร้อมใช้งานทำให้การระเบิดเรียบลง หากคุณรันกระบวนการบอทหลายตัวบนคีย์เดียว พวกมันจะแบ่งปันเพดานนั้น ดังนั้นให้ลด limit ของแต่ละกระบวนการตามลำดับ
ประมาณการค่าใช้จ่ายก่อนเชิญฝูงคน สมมติฐาน: แต่ละการเรียกส่งโทเคนอินพุตประมาณ 1,200 (พรอมต์บวกสิบตา) และส่งคืน 250 โทเคน การเรียกหนึ่งครั้งมีค่าใช้จ่าย 1,200 x $0.25 / 1,000,000 + 250 x $1.00 / 1,000,000 = $0.0003 + $0.00025 = $0.00055 เซิร์ฟเวอร์ที่ผลิตการตอบกลับ 2,000 ครั้งต่อวันจะใช้เงินประมาณ $1.10 ต่อวันภายใต้สมมติฐานเหล่านั้น ตัวเลขของคุณจะแตกต่างกัน ดังนั้นบันทึกฟิลด์ usage เป็นเวลาหนึ่งสัปดาห์แล้วคำนวณใหม่
เพื่อปรับแต่งบุคลิกภาพเอง อ่าน คู่มือพรอมต์ ราคาแสดงบน หน้าราคา
การทำให้แข็งแรงก่อนแชร์บอท
เก็บโทเคนออกจากแหล่งควบคุม อ่านจากสภาพแวดล้อมตามที่แสดง จำกัดความยาวข้อความก่อนส่งไปยัง API เพื่อให้การวางข้อความขนาดใหญ่หนึ่งครั้งไม่สามารถเผาผลาญยอดเงินของคุณได้ บันทึกข้อผิดพลาดแต่ไม่ใช่ข้อความ เนื่องจากผู้ใช้ของคุณคาดหวังว่าการสนทนาของพวกเขาจะคงความเป็นส่วนตัว เพิ่มคำสั่ง /reset ที่ล้าง deque ของช่อง ซึ่งเป็นวิธีแก้ไขที่เร็วที่สุดเมื่อการสนทนาเบี่ยงเบน สุดท้าย ใส่หมายเหตุที่มองเห็นได้ในโปรไฟล์บอทว่าเขียนนวนิยายสำหรับผู้ใหญ่และการตอบกลับของมันเป็นสิ่งที่สร้างขึ้น
การออกแบบอีกอย่างที่ควรกล่าวถึง: บอทรอการตอบกลับเต็มรูปแบบแทนที่จะสตรีม การแก้ไขข้อความโทเคนต่อโทเคนชนขีดจำกัดการแก้ไขของแพลตฟอร์ม และการรอสำหรับคำตอบที่จำกัดประมาณ 150 คำนั้นสั้น หากต้องการการพิมพ์สดในเว็บฟรอนต์เอนด์ของคุณ ให้เปิด stream: true ที่นั่นซึ่งคุณควบคุมการขนส่ง จนกว่าจะถึงตอนนั้น ตัวบ่งชี้การพิมพ์ที่คุณส่งอยู่แล้วเป็นวิธีที่ถูกที่สุดในการแสดงว่าบอทกำลังทำงาน และทำให้โค้ดสั้นพอที่จะอ่านในครั้งเดียวและตรวจสอบก่อนเชิญใคร
