실전 RAG Bot 프로젝트 — Q&A 봇 · 사내 지식베이스 · FastAPI/Slack/Discord 통합

3·4부에서 만든 로딩/전처리/리트리버·Generation 파트를 실전 서비스로 엮습니다. 이 글은 간단한 문서 Q&A 봇 → 사내 지식베이스 챗봇 → 웹/앱 통합 → FastAPI 백엔드 → Slack/Discord 봇 배포 순서로 진행하며, 즉시 실행 가능한 코드를 제공합니다. 

2025.08.16 - [RAG] - RAG Bot 만들기: Generation 파트 — Retrieval·LLM 결합, RetrievalQA, 프롬프트 전략, 대화 메모리

RAG Bot 만들기: Generation 파트 — Retrieval·LLM 결합, RetrievalQA, 프롬프트 전략, 대화 메모리

 

---

0. 준비물 & 설치

# Core
pip install -U langchain langchain-openai langchain-community langchain-text-splitters
# Vector store (Chroma 권장: 로컬/서버 둘 다 간단)
pip install -U chromadb langchain-chroma
# Loaders (필요한 것만 선택 설치)
pip install pypdf beautifulsoup4 lxml atlassian-python-api \
            langchain-google-community google-auth google-auth-oauthlib google-api-python-client
# API 서버
pip install fastapi uvicorn[standard]
# Bot (선택)
pip install slack_bolt discord.py

환경 변수

• OPENAI_API_KEY (필수)
• (선택) Google/Confluence 자격증명

---

1. 간단한 문서 기반 Q&A 봇 만들기

1-1. 데이터 적재·전처리·청킹

# 1) 샘플: 로컬 폴더에서 문서 로딩
from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader, CSVLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
import re

raw_docs = []
raw_docs += PyPDFLoader("docs/policy.pdf").load()
raw_docs += WebBaseLoader(["https://example.com/faq"]).load()
raw_docs += CSVLoader("data/faq.csv").load()

# 2) 가벼운 클리닝
ZWS = "\u200b"

def clean_text(t: str) -> str:
    t = t.replace(ZWS, "")
    t = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F]", "", t)
    t = re.sub(r"\s+", " ", t).strip()
    return t

base_docs = [Document(page_content=clean_text(d.page_content), metadata=d.metadata) for d in raw_docs]

# 3) 청킹
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
docs = splitter.split_documents(base_docs)
print(f"원문 {len(raw_docs)} → 청크 {len(docs)}")

1-2. 벡터 스토어(Chroma) 구축

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

emb = OpenAIEmbeddings(model="text-embedding-3-small")
vs = Chroma(collection_name="kb", persist_directory="indexes/chroma_kb", embedding_function=emb)
vs.add_documents(docs)  # 초기 적재
vs.persist()

retriever = vs.as_retriever(search_kwargs={"k": 4})

1-3. RAG 체인으로 답변 생성

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

PROMPT = ChatPromptTemplate.from_messages([
    ("system", "너는 정확하고 간결한 한국어 도우미다. 컨텍스트에 없으면 모른다고 답해라."),
    ("human", "컨텍스트:\n{context}\n\n질문: {question}")
])

def fmt(docs):
    return "\n\n".join(f"[source: {d.metadata.get('source','?')}]\n{d.page_content}" for d in docs)

qa = (
    {"context": retriever | fmt, "question": RunnablePassthrough()}
    | PROMPT | llm | StrOutputParser()
)

print(qa.invoke("노쇼 수수료는?"))

---

2. 사내 지식베이스 챗봇 구현하기

2-1. 사내 소스 로더(Confluence/Google Drive)

from langchain_community.document_loaders import ConfluenceLoader, UnstructuredFileIOLoader
from langchain_google_community import GoogleDriveLoader

conf_docs = ConfluenceLoader(
    url="https://<org>.atlassian.net/wiki",
    username="you@company.com",
    api_key="ATLASSIAN_API_TOKEN",
    space_key="SPACE",
    limit=100,
).load()

# Google Drive 폴더 전체 수집 (하위 포함)
gdocs = GoogleDriveLoader(
    folder_id="<folder-id>", recursive=True,
    file_loader_cls=UnstructuredFileIOLoader,
    file_loader_kwargs={"mode": "elements"},
).load()

2-2. 메타데이터·필터·권한

# 문서에 접근 레벨/팀/언어 등 태그 추가
for d in conf_docs + gdocs:
    d.metadata.setdefault("access", "employee")
    d.metadata.setdefault("team", d.metadata.get("space", "general"))
    d.metadata.setdefault("locale", "ko")

# 컬렉션에 병합 업서트
vs.add_documents(conf_docs + gdocs)
vs.persist()

# 예: 팀 기반 필터링 리트리버
team_ret = vs.as_retriever(search_kwargs={"k": 4, "filter": {"team": "support"}})
print(qa.invoke({"question": "영수증 발급 정책?", "context": team_ret}))  # (개념 예시)

권한 모델: API단에서 사용자 토큰→팀/권한을 해석해 필터 조건으로 강제하세요(Zero-trust). Pinecone/Weaviate/pgvector 등도 메타 필터를 제공합니다.

2-3. 평가/관측(LangSmith)

운영 전후로 run_on_dataset 평가와 태깅을 넣어 리콜/정확도·지연·비용을 추적하세요.

---

3. 웹/앱 서비스와의 통합 (최소 예제)

3-1. 간단 웹 위젯 (HTML)

<!doctype html>
<html>
  <body>
    <input id="q" placeholder="질문을 입력하세요" />
    <button onclick="ask()">Ask</button>
    <pre id="out"></pre>
    <script>
      async function ask(){
        const r = await fetch("/ask", {method:"POST", headers:{"Content-Type":"application/json"}, body: JSON.stringify({question: document.getElementById('q').value})});
        const j = await r.json();
        document.getElementById('out').textContent = j.answer;
      }
    </script>
  </body>
</html>

3-2. CORS/보안

• 도메인 화이트리스트로 CORS 제한, Auth 헤더(JWT/쿠키)로 사용자 식별.
• API 게이트웨이/역프록시에서 속도 제한·IP 허용 목록 적용.

---

4. FastAPI 백엔드 연동

4-1. 단일턴 /ask + 멀티턴 /chat

from fastapi import FastAPI
from pydantic import BaseModel
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

app = FastAPI()
_session = {}

def get_hist(session_id: str):
    if session_id not in _session:
        _session[session_id] = ChatMessageHistory()
    return _session[session_id]

class AskIn(BaseModel):
    question: str

@app.post("/ask")
def ask(in_: AskIn):
    return {"answer": qa.invoke(in_.question)}

# 멀티턴 (세션 유지)
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "컨텍스트 기반으로만 답해라. 모르면 모른다고 답해라."),
    MessagesPlaceholder("chat_history"),
    ("human", "컨텍스트:\n{context}\n\n질문: {question}")
])
rag_chat = ({"context": retriever | fmt, "question": RunnablePassthrough()} | chat_prompt | llm | StrOutputParser())
rag_with_hist = RunnableWithMessageHistory(rag_chat, get_hist, input_messages_key="question", history_messages_key="chat_history")

class ChatIn(BaseModel):
    session_id: str
    question: str

@app.post("/chat")
def chat(in_: ChatIn):
    cfg = {"configurable": {"session_id": in_.session_id}}
    ans = rag_with_hist.invoke({"question": in_.question}, config=cfg)
    return {"answer": ans}

4-2. 실행

uvicorn app:app --reload --port 8000

---

5. Slack/Discord Bot 배포

5-1. Slack (Bolt for Python, Events API)

# slack_app.py
import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_APP_TOKEN = os.getenv("SLACK_APP_TOKEN")  # xapp- with Socket Mode

app = App(token=SLACK_BOT_TOKEN)

@app.event("app_mention")
@app.message(re.compile(r"^/ask\s+(.+)", re.I))
def handle_events(message, say, context):
    text = message.get("text", "")
    q = text.split(" ", 1)[-1] if "/ask" in text else text
    answer = qa.invoke(q)
    say(answer[:3000])

if __name__ == "__main__":
    SocketModeHandler(app, SLACK_APP_TOKEN).start()

• 슬래시 커맨드(/ask)를 만들어도 좋습니다(응답 URL로 답변).
• 사내 권한·채널 제한, 민감어 필터링을 추가하세요.

5-2. Discord (discord.py)

# discord_bot.py
import os, asyncio
import discord

TOKEN = os.getenv("DISCORD_TOKEN")
intents = discord.Intents.default(); intents.message_content = True
client = discord.Client(intents=intents)

@client.event
async def on_message(message: discord.Message):
    if message.author.bot: return
    if message.content.startswith("!ask "):
        q = message.content[5:]
        ans = qa.invoke(q)
        await message.channel.send(ans[:1900])

asyncio.run(client.start(TOKEN))

---

6. 마무리

• Q&A 봇: 로컬·웹·CSV/PDF를 통합해 빠르게 구축
• 사내 챗봇: Confluence/Google Drive 등과 연동, 메타 필터로 권한 제어
• 서비스 통합: FastAPI로 엔드포인트 제공, Slack/Discord 봇으로 사내 유통
• 확장: Pinecone/Weaviate/pgvector/Redis Vector 교체, 하이브리드/리랭킹/메모리/평가 추가