Was wir bauen
Zwei kleine Bots, ein gemeinsamer Kern. Eine Datei namens llm.py übernimmt alles, was die API betrifft: die Persona, den Ratenbegrenzer und die Wiederholungslogik. Ein Discord-Bot und ein Telegram-Bot machen dann nichts anderes, als Plattform-Ereignisse in eine Liste von Nachrichten und zurück zu übersetzen. Wenn du später eine dritte Plattform hinzufügst, schreibst du nur noch den Adapter.
Installiere die Abhängigkeiten mit pip install httpx discord.py python-telegram-bot. Exportiere API_KEY von deiner Kontoseite, sowie DISCORD_TOKEN und/oder TELEGRAM_TOKEN aus der jeweiligen Bot-Einrichtung. Beide Bots setzen Python 3.10 oder neuer voraus.
Das Anfrageformat entspricht der einfachen OpenAI-Chat-Completion-Struktur, die in der mehrsprachigen Schnellstartanleitung beschrieben ist. Hierfür wird kein SDK benötigt.
Der gemeinsame Kern: Persona, Limiter und Wiederholungen
Platziere den kritischen Teil an einer zentralen Stelle. Das untenstehende Modul hält ein Gleitfenster von Anfrage-Zeitstempeln und wartet, wenn es 240 Aufrufe pro Minute erreicht, was eine bewusste Pufferung unter den pro Schlüssel erlaubten 300 Anfragen pro Minute ist. Ein Semaphore begrenzt die Parallelität auf vier, damit ein aktiver Kanal nicht zwanzig Anfragen gleichzeitig auslöst.
# 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."Schau dir die Fehlerbehandlung an. Bei einem 429 oder 503 wartest du 1, dann 2 und schließlich 4 Sekunden und versuchst es erneut. Ein 403 bedeutet, dass der Inhaltsfilter ausgelöst hat; das Bot antwortet dann mit einer neutralen Zeile statt zu wiederholen. Jeder andere Fehler wird ausgelöst, was deine Plattformbibliothek protokolliert. Wenn du ein 402 siehst, ist dein Guthaben leer oder das Testguthaben ist abgelaufen; lade das Prepaid-Guthaben auf.
Der Discord-Adapter
Discord-Bots benötigen die Aktivierung der message-content-Intention im Entwicklerportal sowie im Code, sonst kommt msg.content leer an. Der Bot antwortet nur bei Erwähnung, was verhindert, dass er jede Nachricht in einem aktiven Server liest.
# 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"])Der Verlauf ist ein deque mit maxlen=10, key-basiert nach Kanal-ID, sodass jeder Kanal eine eigene Unterhaltung ist und alte Nachrichten automatisch abfallen. Das Voranstellen des Anzeigenamens jedes Nutzers ermöglicht es dem Modell, Personen in einem gemeinsamen Raum zu unterscheiden. Zehn Nachrichten sind ein konservativer Standardwert; das Kontextfenster umfasst 100.000 Token, sodass du diesen Wert deutlich erhöhen kannst, bevor dies relevant wird.
Der Telegram-Adapter
python-telegram-bot v20 und höher ist asynchron, was sich natürlich mit dem gemeinsamen Kern kombinieren lässt. chat_data ist ein pro Chat gespeichertes Wörterbuch der Bibliothek, sodass Verlauf und Abklingzeit-Zustand keinen zusätzlichen Speichercode benötigen.
# 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() ist die einfachste Startmethode und benötigt keine öffentliche URL. Beachte, dass chat_data im Speicher lebt, also wird es bei einem Neustart gelöscht. Wenn der Verlauf auch Neustarts überleben soll, speichere die Deque-Inhalte selbst in einer Datei oder Datenbank.
Altersprüfung und die NSFW-Kanalregel
Diese API ist nur für Erwachsene, und alles, was du darauf aufbaust, übernimmt diese Verantwortung. Betrachte die Altersprüfung als Feature, nicht als Nachgedanke.
- Discord: antworte nur in Guild-Kanälen, die als altersbeschränkt markiert sind. Die Prüfung
channel.is_nsfw()kostet nichts und der Bot verweigert andernorts, auch in Direktnachrichten. - Beide: biete den Bot nicht in Bereichen an, die auf Jugendliche abzielen, und entziehe den Zugang schnell, wenn du erfährst, dass ein Nutzer unter 18 ist.
Ein Bestätigungsbefehl ist eine Schranke, kein Altersnachweis. Sexuelle Inhalte mit Minderjährigen werden von der API immer mit einem 403 blockiert, einschließlich Fiktion und Rollenspiel, und dein Bot sollte niemals versuchen, diese Antwort zu umgehen.
Ausführen und Testen ohne große Nutzerzahl
- Erwähne den Bot in einem Nicht-NSFW-Kanal und bestätige, dass er sich weigert. Dies beweist, dass das Tor funktioniert.
- Erwähne ihn in einem markierten Kanal und sende schnell zwei Nachrichten. Die zweite sollte die Sanduhr-Reaktion auslösen, was beweist, dass die Abklingzeit funktioniert.
- Führe eine Vier-Nachrichten-Konversation, und beziehe dich dann auf die erste Nachricht. Wenn sich der Bot erinnert, ist der Verlauf korrekt verdrahtet.
- Setze den API-Schlüssel vorübergehend auf einen falschen Wert und bestätige, dass der Fehler in deinen Logs erscheint, anstatt stillschweigend zu verschwinden.
- Füge einen langen Textblock ein und beobachte, wie die Antwort in mehrere Nachrichten unterhalb der plattformspezifischen Längengrenze aufgeteilt wird.
Wenn du bereit zum Deployen bist, reicht jeder dauerhaft laufende Prozessmanager. Die Bots halten eine langfristige Verbindung zu ihrer Plattform und stellen kurze ausgehende HTTPS-Anfragen an die API, daher benötigen sie keine eingehenden Ports und fast keinen Speicher. Füge eine Neustart-Richtlinie hinzu, damit ein Absturz nicht dazu führt, dass deine Community mit einem stillen Bot spricht.
