Cosa stiamo costruendo
Due piccoli bot, un cervello condiviso. Un file chiamato llm.py gestisce tutto ciò che tocca l'API: la persona, il rate limiter, la logica di retry. Un bot Discord e un bot Telegram poi non fanno altro che tradurre gli eventi della piattaforma in una lista di messaggi e viceversa. Aggiungi una terza piattaforma in futuro e scriverai solo l'adattatore.
Installa le dipendenze con pip install httpx discord.py python-telegram-bot. Esporta API_KEY dalla pagina del tuo account, più DISCORD_TOKEN e/o TELEGRAM_TOKEN dalla configurazione bot di ciascuna piattaforma. Entrambi i bot qui sotto assumono Python 3.10 o superiore.
Il formato della richiesta è la forma standard di chat-completions OpenAI, coperta nella guida rapida multi-lingua. Nulla qui richiede un SDK.
Il core condiviso: persona, limiter e retry
Metti la parte pericolosa in un unico punto. Il modulo sottostante mantiene una finestra scorrevole dei timestamp delle richieste e attende quando si avvicina a 240 chiamate al minuto, un margine deliberato sotto le 300 per minuto consentite per chiave. Un semaforo limita la concorrenza a quattro così un canale affollato non può inviare venti richieste in una volta.
# 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."Guarda la gestione degli stati. Un 429 o 503 attende 1, poi 2 e infine 4 secondi e riprova. Un 403 indica che il filtro sui contenuti è intervenuto, quindi il bot risponde con una frase neutrale invece di riprovare. Qualsiasi altro errore genera un'eccezione, che la libreria della tua piattaforma registrerà. Se vedi un 402, il tuo saldo è vuoto o la prova è scaduta; ricarica il saldo prepagato.
L'adattatore Discord
I bot Discord richiedono l'attivazione dell'intent message-content nel portale developer e nel codice, altrimenti msg.content arriva vuoto. Il bot risponde solo quando viene menzionato, il che evita che legga ogni messaggio in un server affollato.
# 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"])La cronologia è un deque con maxlen=10 indicizzata per id canale, così ogni canale è la sua conversazione e i turni vecchi scadono automaticamente. Anteporre il nome visualizzato di ciascun utente alle sue righe permette al modello di distinguere le persone in una stanza condivisa. Dieci messaggi sono un default conservativo; la finestra di contesto è di 100.000 token, quindi puoi aumentarla molto prima che abbia importanza.
L'adattatore Telegram
python-telegram-bot v20 e successivi sono asincroni, il che si abbina naturalmente al core condiviso. chat_data è un dizionario per chat che la libreria mantiene per te, quindi cronologia e stato del cooldown non richiedono codice di storage aggiuntivo.
# 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() è il modo più semplice per avviare e non richiede un URL pubblico. Tieni presente che chat_data risiede in memoria, quindi un riavvio lo cancella. Se hai bisogno che la cronologia sopravviva ai riavvii, salva tu stesso i contenuti della deque in un file o database.
Controllo età e regola canale NSFW
Questa API è solo per adulti, e qualsiasi cosa tu costruisca su di essa eredita quella responsabilità. Tratta il controllo età come una funzionalità, non come un pensiero secondario.
- Discord: rispondi solo nei canali del server contrassegnati come limitati all'età. Il controllo
channel.is_nsfw()non costa nulla e il bot rifiuta altrove, incluse le chat private. - Telegram: richiedi una conferma esplicita
/adultper chat prima di qualsiasi generazione e indica la regola nel testo di benvenuto. - Entrambi: non offrire il bot in spazi rivolti ai giovani e rimuovi l'accesso prontamente se scopri che un utente ha meno di 18 anni.
Un comando di conferma è un cancello, non una prova di età. I contenuti sessuali che coinvolgono minori sono sempre bloccati dall'API con un 403, inclusa la narrativa e il roleplay, e il tuo bot non dovrebbe mai cercare di aggirare quella risposta.
Esecuzione e test senza folla
Avvia ogni bot nel suo terminale con python discord_bot.py o python telegram_bot.py. Per la prima esecuzione, crea un server di test privato o una chat con te stesso. Segui una breve checklist e catturerai quasi tutti i bug da principianti.
- Menziona il bot in un canale non NSFW e conferma che rifiuta. Questo prova che il cancello funziona.
- Menzionalo in un canale contrassegnato e invia due messaggi rapidamente. Il secondo dovrebbe ottenere la reazione clessidra, il che prova che il cooldown funziona.
- Tieni una conversazione di quattro messaggi, poi fai riferimento al primo messaggio. Se il bot ricorda, la cronologia è collegata correttamente.
- Imposta temporaneamente la chiave API su un valore errato e conferma che l'errore appaia nei tuoi log invece di scomparire silenziosamente.
- Incolla un lungo blocco di testo e guarda la risposta divisa in più messaggi sotto il limite di lunghezza della piattaforma.
Se le risposte sembrano generiche, il problema è di solito la personalità piuttosto che il codice. Modifica una frase di PERSONA, riavvia e confronta. Poiché la personalità risiede nel modulo condiviso, una modifica cambia entrambi i bot contemporaneamente, il che è esattamente il motivo per cui la struttura vale il file aggiuntivo.
Quando sei pronto per la produzione, qualsiasi gestore di processi sempre attivo è sufficiente. I bot mantengono una connessione a lungo termine con la loro piattaforma ed effettuano brevi chiamate HTTPS in uscita all'API, quindi non necessitano di porte in ingresso e quasi nessuna memoria. Aggiungi una politica di riavvio così un crash non lascia la tua comunità a parlare con un bot silenzioso.
Limiti di richiesta, dimensione cronologia e costi
Tre parametri controllano il comportamento del bot sotto carico. Il cooldown per utente impedisce a una persona di monopolizzare la chiave. La finestra globale protegge il limite di 300 richieste al minuto. Il semaforo smorza i picchi. Se esegui più processi di bot con una singola chiave, condividono quel limite, quindi riduci di conseguenza il limit di ogni processo.
Stima la spesa prima di invitare un pubblico. Ipotesi: ogni chiamata invia circa 1 200 token di input (persona più dieci turni) e restituisce 250 token. Una chiamata costa 1 200 x $0,25 / 1 000 000 + 250 x $1,00 / 1 000 000 = $0,0003 + $0,00025 = $0,00055. Un server che produce 2 000 risposte al giorno spenderebbe circa $1,10 al giorno secondo queste ipotesi. I tuoi numeri saranno diversi, quindi registra il campo usage per una settimana e ricalcola.
Per ottimizzare la persona stessa, leggi la guida al prompting. I prezzi sono elencati nella pagina dei prezzi.
Rafforzamento prima di condividere il bot
Tieni i token fuori dal controllo versione; leggi dall'ambiente come mostrato. Limita la lunghezza del messaggio prima di inviarlo all'API in modo che un incolla enorme non bruci il tuo saldo. Registra gli errori ma non il testo del messaggio, poiché gli utenti si aspetterebbero ragionevolmente che le loro chat rimangano private. Aggiungi un comando /reset che cancella la deque di un canale, che è la soluzione più rapida quando una conversazione si disallinea. Infine, inserisci una nota visibile nel profilo del bot che indica che scrive narrativa per adulti e che le sue risposte sono generate.
Un'altra scelta di progettazione merita una frase: i bot attendono la risposta completa invece di fare streaming. Modificare un messaggio chat token per token si scontra direttamente con i limiti di modifica della piattaforma e, per risposte limitate a circa 150 parole, l'attesa è breve. Se in seguito vuoi la digitazione live nel tuo frontend web, attiva stream: true lì, dove controlli il trasporto. Fino ad allora, l'indicatore di digitazione che invii già è il modo più economico per mostrare che il bot sta funzionando e mantiene il codice abbastanza breve da leggere in una sola seduta e verificare prima di invitare chiunque.
