Co budujemy
Dwa małe boty, jedno wspólne mózg. Plik o nazwie llm.py zarządza wszystkim, co dotyka API: persona, limitator zapytań, logika ponawiania prób. Bot Discord i bot Telegram następnie nic innego nie robią, tylko tłumaczą zdarzenia platformy na listę wiadomości i z powrotem. Dodaj trzecią platformę później i napiszesz tylko adapter.
Zainstaluj zależności za pomocą pip install httpx discord.py python-telegram-bot. Wyeksportuj API_KEY ze twojej strony konta, oraz DISCORD_TOKEN i/lub TELEGRAM_TOKEN z konfiguracji botów każdej platformy. Poniższe boty zakładają Python 3.10 lub nowszy.
Format żądania to zwykły kształt chat-completions OpenAI, opisany w poradniku szybkiego startu w wielu językach. Nic tutaj nie wymaga SDK.
Wspólne jądro: persona, limitator i ponawianie prób
Umieść niebezpieczną część w jednym miejscu. Poniższy moduł utrzymuje przesuwne okno znaczników czasu zapytań i czeka, gdy zbliża się do 240 wywołań na minutę, świadomy margines poniżej 300 na minutę dozwolonych na klucz. Semafory ograniczają współbieżność do czterech, więc zajęty kanał nie może wysłać dwudziestu zapytań naraz.
# 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."Zwróć uwagę na obsługę statusów. Kod 429 lub 503 powoduje pauzę 1, 2, a następnie 4 sekundy i ponawia próbę. Kod 403 oznacza, że zadziałał filtr treści, więc bot odpowiada neutralnym zdaniem zamiast ponawiać próbę. Każde inne wyjątki zgłasza błąd, który Twoja biblioteka platformy zaloguje. Jeśli widzisz kod 402, Twoje saldo jest puste lub próba się skończyła; doładuj przedpłacone saldo.
Adapter Discord
Boty Discord potrzebują włączonego zamiaru message-content w portalu deweloperskim oraz w kodzie, w przeciwnym razie msg.content przychodzi pusty. Bot odpowiada tylko wtedy, gdy jest wspomniany, co utrzymuje go od czytania każdej wiadomości w zatłoczonym serwerze.
# 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"])Historia to deque z maxlen=10 kluczowany przez identyfikator kanału, więc każdy kanał jest swoją własną rozmową, a stare tury odpadają automatycznie. Poprzedzając każdą linię użytkownika nazwą wyświetlaną mówcy, pozwala to modelowi rozróżniać ludzi w wspólnym pokoju. Dziesięć wiadomości to konserwatywny domyślny; okno kontekstu ma 100 000 tokenów, więc możesz to podnieść znacznie, zanim to ma znaczenie.
Adapter Telegram
python-telegram-bot v20 i nowszy jest asynchroniczny, co naturalnie pasuje do wspólnego jądra. chat_data to słownik per-czat, który biblioteka utrzymuje dla Ciebie, więc historia i stan cooldown nie potrzebują dodatkowego kodu przechowywania.
# 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() to najprostszy sposób na uruchomienie i nie wymaga publicznego URL. Pamiętaj, że chat_data znajduje się w pamięci, więc restart go czyści. Jeśli potrzebujesz historii, która przetrwa restarty, zapisz zawartość deque samodzielnie w pliku lub bazie danych.
Filtr wiekowy i reguła kanału NSFW
To API jest tylko dla dorosłych, a wszystko, co na nim zbudujesz, dziedziczy tę odpowiedzialność. Traktuj filtr wiekowy jako funkcję, nie pośpiech.
- Discord: odpowiadaj tylko w kanałach gildii oznaczonych jako ograniczone wiekowo. Sprawdzenie
channel.is_nsfw()nic nie kosztuje, a bot odmawia w innych miejscach, w tym w wiadomościach bezpośrednich. - Telegram: wymagaj jawnego potwierdzenia
/adultdla każdego czatu przed jakąkolwiek generacją i wskaż regułę w tekście powitalnym. - Oba: nie oferuj bota w miejscach przeznaczonych dla młodych ludzi i usuń dostęp szybko, jeśli dowiesz się, że użytkownik ma poniżej 18 lat.
Polecenie potwierdzenia to bariera, nie dowód wieku. Treść seksualna involving małoletnich jest zawsze blokowana przez API z 403, w tym fikcja i roleplay, a Twój bot nigdy nie powinien próbować obejść tę odpowiedź.
Uruchomienie i testowanie bez tłumu
Uruchom każdy bot w osobnym terminalu przez python discord_bot.py lub python telegram_bot.py. Przy pierwszym uruchomieniu stwórz prywatny serwer testowy lub czat tylko z Tobą. Przejdź przez krótką listę kontrolną, a wyłapiesz prawie każdy początkujący błąd.
- Wspomnij bota w kanale nie-NSFW i potwierdź, że odmawia. To dowodzi, że brama działa.
- Wspomnij go w oznaczonym kanale i wyślij dwie wiadomości szybko. Druga powinna uzyskać reakcję klepsydry, co dowodzi, że cooldown działa.
- Prowadź rozmowę czterema wiadomościami, a następnie odwołaj się do pierwszej wiadomości. Jeśli bot pamięta, historia jest poprawnie podłączona.
- Tymczasowo ustaw klucz API na niewłaściwą wartość i potwierdź, że błąd pojawia się w Twoich logach zamiast znikać cicho.
- Wklej długi blok tekstu i obserwuj, jak odpowiedź dzieli się na kilka wiadomości poniżej limitu długości platformy.
Jeśli odpowiedzi wydają się ogólne, persona jest zwykle problemem, a nie kodem. Zmień jedno zdanie PERSONA, zrestartuj i porównaj. Ponieważ persona żyje w wspólnym module, jedna edycja zmienia oba boty naraz, co jest dokładnie powodem, dla którego struktura jest warta dodatkowego pliku.
Gdy będziesz gotowy do wdrożenia, wystarczy dowolny menedżer procesów działający ciągle. Boty utrzymują długotrwałe połączenie z platformą i wykonują krótkie połączenia HTTPS do API, więc nie potrzebują portów przychodzących i prawie nie zużywają pamięci. Dodaj politykę restartu, aby awaria nie sprawiła, że Twoja społeczność będzie rozmawiać z milczącym botem.
Limity zapytań, rozmiar historii i koszt
Trzy suwaki kontrolują zachowanie bota pod obciążeniem. Opóźnienie na użytkownika zapobiega monopolizacji klucza przez jedną osobę. Globalne okno chroni próg 300 zapytań na minutę. Semafor wygładza bursty. Jeśli uruchamiasz wiele procesów bota na jednym kluczu, dzielą ten próg, więc obniż limit każdego procesu odpowiednio.
Oszacuj wydatki przed zaproszeniem tłumu. Założenia: każde wywołanie wysyła około 1200 tokenów wejściowych (persona plus dziesięć tur) i zwraca 250 tokenów. Jedno wywołanie kosztuje 1200 x $0,25 / 1 000 000 + 250 x $1,00 / 1 000 000 = $0,0003 + $0,00025 = $0,00055. Serwer generujący 2000 odpowiedzi dziennie wyda około $1,10 dziennie przy tych założeniach. Twoje liczby będą inne, więc loguj pole usage przez tydzień i przelicz.
Aby dostroić samą personę, przeczytaj poradnik promptów. Ceny są wymienione na stronie z cenami.
Wzmocnienie przed udostępnieniem bota
Trzymaj tokeny poza kontrolą wersji; odczytuj je ze środowiska, jak pokazano. Ogranic długość wiadomości przed wysłaniem do API, aby jeden ogromny wklejony tekst nie spalił Twojego salda. Loguj błędy, ale nie tekst wiadomości, ponieważ użytkownicy słusznie oczekują prywatności czatów. Dodaj komendę /reset, która czyści deque kanału, co jest najszybszym rozwiązaniem, gdy rozmowa odbiegnie od tematu. Na końcu dodaj widoczną notatkę w profilu bota informującą, że pisze literaturę dla dorosłych i że jego odpowiedzi są generowane.
Jeszcze jedna decyzja projektowa zasługuje na zdanie: boty czekają na pełną odpowiedź zamiast ją strumieniować. Edytowanie wiadomości czatu token po tokenie natychmiast wchodzi w granice edycji platformy, a przy odpowiedziach ograniczonych do około 150 słów oczekiwanie jest krótkie. Jeśli później chcesz żywe wpisywanie w swoim własnym interfejsie webowym, włącz stream: true tam, gdzie kontrolujesz transport. Do tego czasu wskaźnik wpisywania, który już wysyłasz, jest najtańszym sposobem na pokazanie, że bot pracuje, i utrzymuje kod krótkim na tyle, aby dało się go przeczytać w jednej sesji i sprawdzić przed zaproszeniem kogokolwiek.
