디스코드 봇 만들기#17RAG로 문서 기반 답변봇 만들기! 우리 서버 전용 FAQ 봇 (파이썬 예제)

pip install discord.py openai anthropic numpy

my-rag-bot/
├── docs/              ← 여기에 문서(.txt / .md)를 넣어요
│   ├── 서버규칙.txt
│   └── 모드설치가이드.md
├── build_index.py     ← 색인 만드는 스크립트 (한 번만 실행)
└── rag_bot.py         ← 실제 봇

CHUNK_SIZE = 500      # 조각 하나의 글자 수
CHUNK_OVERLAP = 100   # 조각끼리 겹칠 글자 수

def split_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    chunks, start = [], 0
    while start < len(text):
        piece = text[start:start + size].strip()
        if piece:
            chunks.append(piece)
        start += size - overlap
    return chunks

from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
EMBED_MODEL = "text-embedding-3-small"

def embed_all(texts):
    vectors = []
    for i in range(0, len(texts), 100):     # 100개씩 끊어서 요청
        batch = texts[i:i + 100]
        res = client.embeddings.create(model=EMBED_MODEL, input=batch)
        vectors.extend([d.embedding for d in res.data])
        print(f"  임베딩 {i + len(batch)}/{len(texts)} 완료")
    return vectors

import numpy as np

# 미리 전부 길이 1로 맞춰두면(정규화) 나중에 곱셈만 하면 끝!
MATRIX = np.array([it["embedding"] for it in ITEMS], dtype="float32")
MATRIX = MATRIX / np.linalg.norm(MATRIX, axis=1, keepdims=True)

def search(question, top_k=3):
    res = openai_client.embeddings.create(model=EMBED_MODEL, input=[question])
    q = np.array(res.data[0].embedding, dtype="float32")
    q = q / np.linalg.norm(q)

    scores = MATRIX @ q                      # 한 방에 전체 유사도 계산
    top = np.argsort(scores)[::-1][:top_k]   # 점수 높은 순 top_k개
    return [(ITEMS[i], float(scores[i])) for i in top]

from anthropic import Anthropic

claude = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
CHAT_MODEL = "claude-sonnet-5"

SYSTEM_PROMPT = (
    "당신은 디스코드 서버의 문서 도우미입니다. "
    "반드시 아래 '문서' 안의 내용만 근거로 한국어로 답하세요. "
    "문서에 없는 내용이면 지어내지 말고 '문서에서 찾지 못했어요'라고 답하세요. "
    "답변은 3~5문장으로 짧게 정리하세요."
)

def make_answer(question, hits):
    context = "\n\n".join(
        f"[{i + 1}] ({it['source']})\n{it['text']}"
        for i, (it, _) in enumerate(hits)
    )
    msg = claude.messages.create(
        model=CHAT_MODEL,
        max_tokens=600,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": f"문서:\n{context}\n\n질문: {question}"}],
    )
    return msg.content[0].text

@bot.tree.command(name="질문", description="서버 문서에서 찾아 답변해드려요")
@discord.app_commands.describe(내용="궁금한 내용을 적어주세요")
async def ask(interaction: discord.Interaction, 내용: str):
    await interaction.response.defer(thinking=True)   # ① 3초 제한 회피
    hits = await asyncio.to_thread(search, 내용)       # ② 블로킹 → 스레드로
    answer = await asyncio.to_thread(make_answer, 내용, hits)

    embed = discord.Embed(title="📚 문서 기반 답변", description=answer[:4000],
                          color=0x5865F2)
    await interaction.followup.send(embed=embed)

# 파이썬 3.9+ / openai 2.x / numpy 2.x
# 문서를 조각내서 임베딩으로 바꾼 뒤 index.json 으로 저장합니다.
import os
import json
import glob
from openai import OpenAI

DOCS_DIR = "docs"
INDEX_FILE = "index.json"
EMBED_MODEL = "text-embedding-3-small"
CHUNK_SIZE = 500
CHUNK_OVERLAP = 100

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def split_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """긴 글을 size 글자씩, overlap 만큼 겹치게 잘라줍니다."""
    chunks, start = [], 0
    while start < len(text):
        piece = text[start:start + size].strip()
        if piece:
            chunks.append(piece)
        start += size - overlap
    return chunks

def load_chunks():
    items = []
    paths = sorted(glob.glob(os.path.join(DOCS_DIR, "*.txt")) +
                   glob.glob(os.path.join(DOCS_DIR, "*.md")))
    for path in paths:
        with open(path, encoding="utf-8") as f:
            text = f.read()
        for i, piece in enumerate(split_text(text)):
            items.append({
                "source": os.path.basename(path),
                "chunk_id": i,
                "text": piece,
            })
    return items

def embed_all(texts):
    vectors = []
    for i in range(0, len(texts), 100):
        batch = texts[i:i + 100]
        res = client.embeddings.create(model=EMBED_MODEL, input=batch)
        vectors.extend([d.embedding for d in res.data])
        print(f"  임베딩 {i + len(batch)}/{len(texts)} 완료")
    return vectors

def main():
    items = load_chunks()
    if not items:
        print("❌ docs 폴더에 .txt 또는 .md 파일을 먼저 넣어주세요!")
        return
    print(f"조각 {len(items)}개를 임베딩할게요...")
    vectors = embed_all([it["text"] for it in items])
    for it, vec in zip(items, vectors):
        it["embedding"] = vec
    with open(INDEX_FILE, "w", encoding="utf-8") as f:
        json.dump({"model": EMBED_MODEL, "items": items}, f, ensure_ascii=False)
    print(f"✅ {INDEX_FILE} 저장 완료! (조각 {len(items)}개)")

if __name__ == "__main__":
    main()

# discord.py 2.x / 파이썬 3.9+ / openai 2.x / anthropic 0.x / numpy 2.x
import os
import json
import asyncio

import numpy as np
import discord
from discord.ext import commands
from openai import OpenAI
from anthropic import Anthropic

DISCORD_TOKEN = os.environ["DISCORD_TOKEN"]
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
claude = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "claude-sonnet-5"
TOP_K = 3          # 몇 조각을 참고할지
MIN_SCORE = 0.25   # 이보다 유사도가 낮으면 "못 찾았어요"

# --- 색인 불러오기 ---
with open("index.json", encoding="utf-8") as f:
    index = json.load(f)

ITEMS = index["items"]
MATRIX = np.array([it["embedding"] for it in ITEMS], dtype="float32")
MATRIX = MATRIX / np.linalg.norm(MATRIX, axis=1, keepdims=True)

def search(question, top_k=TOP_K):
    res = openai_client.embeddings.create(model=EMBED_MODEL, input=[question])
    q = np.array(res.data[0].embedding, dtype="float32")
    q = q / np.linalg.norm(q)
    scores = MATRIX @ q
    top = np.argsort(scores)[::-1][:top_k]
    return [(ITEMS[i], float(scores[i])) for i in top]

SYSTEM_PROMPT = (
    "당신은 디스코드 서버의 문서 도우미입니다. "
    "반드시 아래 '문서' 안의 내용만 근거로 한국어로 답하세요. "
    "문서에 없는 내용이면 지어내지 말고 '문서에서 찾지 못했어요'라고 답하세요. "
    "답변은 3~5문장으로 짧게 정리하세요."
)

def make_answer(question, hits):
    context = "\n\n".join(
        f"[{i + 1}] ({it['source']})\n{it['text']}"
        for i, (it, _) in enumerate(hits)
    )
    msg = claude.messages.create(
        model=CHAT_MODEL,
        max_tokens=600,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": f"문서:\n{context}\n\n질문: {question}"}],
    )
    return msg.content[0].text

intents = discord.Intents.default()

class RagBot(commands.Bot):
    def __init__(self):
        super().__init__(command_prefix="!", intents=intents)

    async def setup_hook(self):
        await self.tree.sync()

bot = RagBot()

@bot.event
async def on_ready():
    print(f"✅ 로그인 완료: {bot.user} / 문서 조각 {len(ITEMS)}개")

@bot.tree.command(name="질문", description="서버 문서에서 찾아 답변해드려요")
@discord.app_commands.describe(내용="궁금한 내용을 적어주세요")
async def ask(interaction: discord.Interaction, 내용: str):
    await interaction.response.defer(thinking=True)
    try:
        hits = await asyncio.to_thread(search, 내용)
        if not hits or hits[0][1] < MIN_SCORE:
            await interaction.followup.send("🤔 문서에서 관련 내용을 찾지 못했어요.")
            return
        answer = await asyncio.to_thread(make_answer, 내용, hits)
    except Exception as e:
        await interaction.followup.send(f"⚠️ 답변을 만들지 못했어요: {e}")
        return

    embed = discord.Embed(
        title="📚 문서 기반 답변",
        description=answer[:4000],
        color=0x5865F2,
    )
    embed.add_field(
        name="🔎 참고한 문서",
        value="\n".join(
            f"[{i + 1}] {it['source']} (유사도 {score:.2f})"
            for i, (it, score) in enumerate(hits)
        ),
        inline=False,
    )
    embed.set_footer(text=f"질문: {내용[:80]}")
    await interaction.followup.send(embed=embed)

bot.run(DISCORD_TOKEN)

python build_index.py    # ① 색인 만들기 (문서 바뀔 때만)
python rag_bot.py        # ② 봇 실행

조각 12개를 임베딩할게요...
  임베딩 12/12 완료
✅ index.json 저장 완료! (조각 12개)