構築するもの
2つの小さなボット、1つの共有ロジック。llm.pyというファイルが、APIにアクセスするすべての処理を管理します。ペルソナ、レートリミッター、リトライロジックです。DiscordボットとTelegramボットは、プラットフォームのイベントをメッセージのリストに変換し、その逆の変換を行うことだけをします。後で第3のプラットフォームを追加する場合、アダプターを1つ書くだけで済みます。
依存関係をpip install httpx discord.py python-telegram-botでインストールします。アカウントページからAPI_KEYを、各プラットフォームのボット設定からそれぞれDISCORD_TOKENおよび/またはTELEGRAM_TOKENを環境変数として設定します。以下のボットは、Python 3.10以降を前提としています。
リクエスト形式は、多言語クイックスタートでカバーされている標準的なOpenAIチャット完了の形式です。ここにはSDKは必要ありません。
共有コア：ペルソナ、リミッター、リトライ
危険な部分を1か所にまとめます。以下のモジュールは、リクエストのタイムスタンプのスライディングウィンドウを保持し、1分あたり240回の呼び出しに近づくと待機します。これは1分あたりAPIキーごとに許可される300回という制限の意図的なマージンです。セマフォは同時実行数を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 をキーとする deque で、maxlen=10 です。これにより、各チャンネルが独自の会話となり、古いターンは自動的に削除されます。各ユーザーの行に話者の表示名をプレフィックスとして付けることで、モデルは共有ルームで人々を区別できます。10メッセージは保守的なデフォルトです。コンテキストウィンドウは 100,000 トークンなので、実用上の問題が生じるまで大幅に引き上げることができます。
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 ではないチャンネルでボットをメンションし、拒否することを確認します。これでゲートが機能していることが証明されます。
- フラグ付きチャンネルでそれをメンションし、2つのメッセージを素早く送信します。2番目のメッセージは砂時計のリアクションを取得するはずです。これでクールダウンが機能していることが証明されます。
- 4メッセージの会話を保持し、最初のメッセージを参照してください。ボットがそれを覚えている場合、履歴は正しく実装されています。
- APIキーを一時的に間違った値に設定し、失敗がログに表示され、静かに消えないことを確認してください。
- 長いテキストブロックを貼り付け、プラットフォームの長さ制限の下で返信が複数のメッセージに分割される様子を確認します。
返信が汎用的に感じられる場合、通常はコードではなくペルソナが問題です。PERSONA の1文を変更し、再起動して比較してください。ペルソナは共有モジュールにあるため、1回の編集で両方のボットが変更されます。これが、余分なファイルの価値がある理由です。
デプロイの準備ができたら、常時稼働プロセスマネージャーであればどれでも使用できます。ボットはプラットフォームに対して長寿命な接続を維持し、APIに対して短い外部HTTPS呼び出しを行うため、受信ポートは必要なく、メモリもほとんど消費しません。クラッシュが発生した際にボットが沈黙したままにならないよう、再起動ポリシーを追加してください。
レート制限、履歴サイズ、コスト
ボットが負荷下で動作する方法を制御する3つのノブがあります。ユーザーごとのクールダウンは、1人のユーザーがキーを独占するのを防ぎます。グローバルウィンドウは1分あたり300リクエストの上限を保護します。セマフォはバーストを平滑化します。1つのキーで複数のボットプロセスを実行する場合、それらはその上限を共有するため、各プロセスの limit をそれに応じて下げてください。
大勢を招待する前に支出を見積もってください。前提：各呼び出しは約 1,200 入力トークン（ペルソナと10ターン分）を送信し、250 トークンを返します。1呼び出しのコストは 1,200 x $0.25 / 1,000,000 + 250 x $1.00 / 1,000,000 = $0.0003 + $0.00025 = $0.00055 です。1日あたり 2,000 件の返信を生成するサーバーは、これらの前提の下で1日あたり約 $1.10 を消費します。あなたの数字は異なるため、usage フィールドを1週間ログに記録し、再計算してください。
ペルソナ自体を調整するには、プロンプティングガイド をお読みください。料金は 料金ページ に記載されています。
ボット公開前の強化
トークンをソースコード管理から除外し、環境変数から読み取ってください（上記参照）。APIに送信する前にメッセージの長さを制限し、巨大な貼り付けで残高が枯渇しないようにしてください。エラーはログに記録しますが、メッセージテキストは記録しません。ユーザーはチャットが非公開であることを合理的に期待するからです。チャネルのdequeをクリアする/resetコマンドを追加します。これは会話が脱線した場合の最も迅速な修正方法です。最後に、ボットのプロフィールに目に見える注記を置き、それが成人向けフィクションを生成しており、その返信は生成されたものであることを明記してください。
もう一つの設計判断にも一言触れる価値があります。ボットはストリーミングではなく、完全な返信を待ってから返答します。チャットメッセージをトークン単位で編集する場合、プラットフォームの編集制限に直ちに突き当たります。また、150語程度に制限された返信では、待機時間は短くなります。後で独自のWebフロントエンドでリアルタイムのタイピングを実現したい場合は、トランスポートを制御する箇所にあるstream: trueを有効にしてください。それまでの間は、すでに送信しているタイピングインジケーターがボットが動作中であることを示す最も安価な方法であり、コードを短く保って、誰でも招く前に一息で読み込み監査できるようにします。
