우리가 구축할 것
두 개의 작은 봇, 하나의 공유 코어. llm.py 파일은 API에 접근하는 모든 것을 담당합니다: 페르소나, 속도 제한기, 재시도 로직입니다. Discord 봇과 Telegram 봇은 플랫폼 이벤트를 메시지 목록으로 변환하고 다시 되돌리는 역할만 합니다. 나중에 다른 플랫폼을 추가하면 어댑터만 작성하면 됩니다.
pip install httpx discord.py python-telegram-bot로 종속성을 설치하세요. 계정 페이지에서 API_KEY를 내보내고, 각 플랫폼의 봇 설정에서 DISCORD_TOKEN 및/또는 TELEGRAM_TOKEN을 가져옵니다. 아래 봇들은 Python 3.10 이상을 가정합니다.
요청 형식은 평범한 OpenAI 채팅 완성 형태이며, 다국어 빠른 시작 가이드에서 다룹니다. 여기에는 SDK가 필요하지 않습니다.
공유 코어: 페르소나, 제한기 및 재시도
위험한 부분을 한 곳에 넣으세요. 아래 모듈은 요청 타임스탬프의 슬라이딩 윈도우를 유지하며 분당 240회 호출에 가까워지면 대기합니다. 이는 키당 분당 허용되는 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 어댑터
디스코드 봇은 개발자 포털과 코드 모두에서 메시지 콘텐츠 의도를 활성화해야 합니다. 그렇지 않으면 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()확인은 비용이 들지 않으며 봇은 다른 곳(다이렉트 메시지 포함)에서는 응답을 거부합니다. - 텔레그램: 모든 생성 전에 각 채팅에서 명시적인
/adult확인을 요구하고 환영 텍스트에 규칙을 명시하세요. - 공통: 청소년을 대상으로 하는 공간에 봇을 제공하지 마세요. 사용자가 18세 미만임을 알면 즉시 접근 권한을 제거하세요.
확인 명령어는 연령의 증명이라기보다는 차단 장치입니다. 미성년자가 관련된 성적 콘텐츠는 픽션과 역할극을 포함하여 항상 403 오류로 차단되므로 봇은 해당 응답을 우회하려고 시도해서는 안 됩니다.
실행 및 소규모 테스트
python discord_bot.py 또는 python telegram_bot.py로 각 봇을 별도의 터미널에서 시작하세요. 첫 실행 시에는 개인 테스트 서버나 자신만 있는 채팅을 만드세요. 짧은 체크리스트를 따라가면 거의 모든 초보자 버그를 잡을 수 있습니다.
- NSFW가 아닌 채널에서 봇을 언급하고 거부하는지 확인하세요. 이는 차단이 작동함을 증명합니다.
- 표시된 채널에서 언급하고 두 개의 메시지를 빠르게 보내세요. 두 번째 메시지는 모래시계 반응을 얻어야 하며, 이는 대기 시간이 작동함을 증명합니다.
- 4개 메시지의 대화를 나누고 첫 번째 메시지를 다시 참조하세요. 봇이 기억한다면 기록이 올바르게 연결된 것입니다.
- 임시로 API 키를 잘못된 값으로 설정하고 실패가 로깅되지만 조용히 사라지지 않는지 확인하세요.
- 긴 텍스트 블록을 붙여넣고 플랫폼의 길이 제한 아래에서 여러 메시지로 나누어 응답하는지 확인하세요.
응답이 일반적이라고 느껴진다면, 문제는 보통 코드보다 페르소나에 있습니다. PERSONA의 문장 하나를 변경하고, 다시 실행한 후 비교하세요. 페르소나가 공유 모듈에 있으므로 한 번의 수정으로 두 봇 모두에 동시에 변경이 적용됩니다. 이것이 바로 구조가 추가 파일을 사용하는 가치가 있는 이유입니다.
배포할 준비가 되면 항상 실행되는 프로세스 관리자면 충분합니다. 봇은 플랫폼에 장수명 연결을 유지하고 API로 짧은 아웃바운드 HTTPS 호출을 수행하므로, 인바운드 포트가 필요하지 않으며 메모리도 거의 사용하지 않습니다. 재시작 정책을 추가하여 충돌 시 커뮤니티가 침묵하는 봇과 대화하는 상황이 발생하지 않도록 하세요.
속도 제한, 기록 크기 및 비용
세 가지 조절 장치는 부하 하에서 봇의 동작을 제어합니다. 사용자별 대기 시간은 한 사람이 키를 독점하지 못하게 합니다. 전역 윈도우는 분당 300회 요청 상한을 보호합니다. 세마포로는 버스트를 완화합니다. 하나의 키로 여러 봇 프로세스를 실행하는 경우 해당 상한을 공유하므로 각 프로세스의 limit을 그에 따라 낮추세요.
대중을 초대하기 전에 지출을 추정하세요. 가정: 각 호출은 약 1,200개의 입력 토큰(페르소나 및 10개의 턴)을 보내고 250개의 토큰을 반환합니다. 한 호출의 비용은 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 필드를 일주일 동안 로깅하고 다시 계산하세요.
페르소나 자체를 조정하려면 프롬프트 가이드를 읽으세요. 가격은 가격 페이지에 나열되어 있습니다.
배포 전 견고성 강화
토큰을 버전 관리 시스템에 넣지 마세요. 표시된 대로 환경에서 읽어오세요. API에 보내기 전에 메시지 길이를 제한해 거대한 붙여넣기가 잔액을 소모하지 않도록 하세요. 오류는 로깅하지만 메시지 텍스트는 로깅하지 마세요. 사용자는 채팅이 비공개로 유지될 것으로 합리적으로 기대합니다. 채널의 deque를 지우는 /reset 명령어를 추가해 대화가 어긋났을 때 가장 빠르게 수정하세요. 마지막으로 봇 프로필에 성인 소설을 작성하며 응답이 생성됨을 명시하는 시각적 안내를 넣으세요.
한 문장 정도는 설명할 만한 또 다른 디자인 선택지가 있습니다: 봇이 스트리밍 대신 전체 응답을 기다린다는 것입니다. 채팅 메시지를 토큰 단위로 편집하면 플랫폼의 편집 제한에 직면하므로, 150단어 근처로 제한된 응답의 경우 대기 시간이 짧습니다. 나중에 자체 웹 프론트엔드에서 실시간 입력을 원한다면 stream: true를 활성화하세요. 그전까지 이미 전송하는 타이핑 표시는 봇이 작업 중임을 보여주는 가장 저렴한 방법이며, 코드를 짧게 유지하여 누군가를 초대하기 전에 한 번에 읽고 감사할 수 있게 합니다.
