我们要构建什么
两个小型机器人，一个共享核心。名为 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 次调用限制下留有 deliberate 余量。信号量将并发限制为 4，防止繁忙频道同时触发 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 错误表示触发了内容过滤器，机器人将返回中性行而非重试。其他任何错误将抛出异常，由你的平台库记录。如果看到 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 为键、maxlen=10 的 deque，因此每个频道都是独立的对话，旧消息会自动移除。在每个用户行前加上发言者的显示名称，使模型能在共享房间中区分用户。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 字限制的回复，等待时间很短。如果稍后希望在自己的 Web 前端实现实时打字，请在那里开启 stream: true，由你控制传输。在此之前，你已发送的打字指示器是显示机器人正在工作的最便宜方式，且代码足够短，可以在一次阅读中阅读并在邀请任何人之前进行审计。
