我們正在建立什麼
兩個小型機器人，共用一個大腦。名為 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 次的餘量。訊號量將並行請求數限制為四個，因此繁忙的通道不會同時觸發二十次請求。
# 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 機器人需要在開發者入口網站與程式碼中開啟 message-content 權限，否則 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"])歷史紀錄是一個以頻道 ID 為鍵的 deque，設定 maxlen=10，因此每個頻道都是獨立的對話，舊的對話輪次會自動移除。在每個使用者輸入前加上顯示名稱，讓模型能在共用房間中區分不同使用者。10 則訊息是一個保守的預設值；上下文視窗為 100,000 token，因此您可以大幅增加此數值而不影響效能。
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 中的一個句子，重新啟動並進行比較。由於角色設定位於共用模組中，一次編輯即可同時變更兩個機器人，這正是這種結構值得額外增加一個檔案的原因。
當你準備部署時，任何持續運行的程序管理器皆可勝任。機器人與平台保持長連線，並向 API 發出短暫的出站 HTTPS 請求，因此不需要開放入站連接埠，也幾乎不佔用記憶體。加入重啟策略，以免當機時讓你的社群面對一個無回應的機器人。
速率限制、歷史大小與成本
有三個參數控制機器人在負載下的行為。每使用者冷卻可防止單一人員壟佔金鑰。全域視窗保護每分鐘 300 次請求的上限。訊號量可平滑突發流量。若在同一金鑰上執行多個機器人程序，它們共用該上限，因此請相應降低每個程序的 limit。
在邀請大量使用者前估算支出。假設：每次呼叫發送約 1,200 個輸入 token（人格加上十輪對話）並返回 250 個 token。一次呼叫的成本為 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 欄位一週後重新計算。
分享機器人前的強化措施
請將 token 排除在版本控制之外；如範例所示從環境變數讀取。在將訊息傳送至 API 之前限制其長度，以免單次過長的貼文耗盡你的額度。記錄錯誤但不記錄訊息內容，因為使用者合理預期他們的對話將保持私密。新增一個 /reset 指令以清除頻道的佇列，這是對話偏離軌道時最快的修復方式。最後，在機器人的個人資料中放上明顯的註記，說明它撰寫成人小說，且其回覆是由機器生成的。
另一個設計選擇值得說明：機器人會等待完整回覆，而非串流輸出。逐 token 編輯聊天訊息會觸及平台編輯限制，而回覆若限制在 150 字以內，等待時間很短。若你稍後想在自訂網頁前端實現即時打字效果，可在控制傳輸層的地方開啟 stream: true。在此之前，你已發送的打字指示器是顯示機器人運作的最便宜方式，且程式碼足夠短，可一次讀完並在邀請他人前進行稽核。
