Slack Bot + LangChain RAG 챗봇 구축 실전 가이드 — 사내 문서 검색 봇 만들기

# 인덱싱 파이프라인 (오프라인)
# 문서 → 청킹 → 임베딩 → 벡터 DB(ChromaDB)

# 질의 파이프라인 (온라인)
# Slack 메시지 → 임베딩 → 벡터 검색 → LLM 생성 → Slack 응답

mkdir slack-rag-bot && cd slack-rag-bot

# 가상환경
python -m venv .venv
source .venv/bin/activate

# 의존성
pip install \
  langchain==0.2.16 \
  langchain-openai==0.1.25 \
  langchain-community==0.2.16 \
  chromadb==0.5.3 \
  slack-bolt==1.20.0 \
  python-dotenv==1.0.1 \
  unstructured==0.15.0 \
  tiktoken==0.7.0

# .env
OPENAI_API_KEY=sk-xxx
SLACK_BOT_TOKEN=xoxb-xxx
SLACK_APP_TOKEN=xapp-xxx
SLACK_SIGNING_SECRET=xxx
CHROMA_PERSIST_DIR=./chroma_db
DOCS_DIR=./documents

slack-rag-bot/
├── .env
├── main.py              # Slack Bot 엔트리포인트
├── indexer.py           # 문서 인덱싱
├── rag_chain.py         # RAG 체인
├── config.py            # 설정
├── documents/           # 사내 문서 (Markdown, PDF 등)
│   ├── deployment-guide.md
│   ├── k8s-access.md
│   └── onboarding.pdf
└── chroma_db/           # 벡터 DB 저장소

# indexer.py
import os
from pathlib import Path
from langchain_community.document_loaders import (
    DirectoryLoader,
    UnstructuredMarkdownLoader,
    PyPDFLoader,
    TextLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from dotenv import load_dotenv

load_dotenv()


def load_documents(docs_dir: str):
    """다양한 형식의 문서 로드"""
    documents = []

    # Markdown 파일
    md_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.md",
        loader_cls=UnstructuredMarkdownLoader,
        show_progress=True
    )
    documents.extend(md_loader.load())

    # PDF 파일
    pdf_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.pdf",
        loader_cls=PyPDFLoader,
        show_progress=True
    )
    documents.extend(pdf_loader.load())

    # 텍스트 파일
    txt_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.txt",
        loader_cls=TextLoader,
        show_progress=True
    )
    documents.extend(txt_loader.load())

    print(f"총 {len(documents)}개 문서 로드됨")
    return documents


def split_documents(documents):
    """문서를 청크로 분할"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
        separators=["\n## ", "\n### ", "\n\n", "\n", " ", ""]
    )

    chunks = text_splitter.split_documents(documents)
    print(f"총 {len(chunks)}개 청크 생성됨")
    return chunks


def create_vectorstore(chunks, persist_dir: str):
    """벡터 DB 생성"""
    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",
        chunk_size=500
    )

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_metadata={"hnsw:space": "cosine"}
    )

    print(f"벡터 DB 생성 완료: {persist_dir}")
    return vectorstore


def index_documents():
    """전체 인덱싱 파이프라인"""
    docs_dir = os.getenv("DOCS_DIR", "./documents")
    persist_dir = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")

    # 로드 → 청킹 → 임베딩 → 저장
    documents = load_documents(docs_dir)
    chunks = split_documents(documents)
    vectorstore = create_vectorstore(chunks, persist_dir)

    return vectorstore


if __name__ == "__main__":
    index_documents()

# 인덱싱 실행
python indexer.py

# rag_chain.py
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()


class RAGChain:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Chroma(
            persist_directory=os.getenv("CHROMA_PERSIST_DIR", "./chroma_db"),
            embedding_function=self.embeddings
        )
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",  # Maximum Marginal Relevance
            search_kwargs={
                "k": 5,
                "fetch_k": 20,
                "lambda_mult": 0.7
            }
        )
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0.1,
            max_tokens=2000
        )
        self.chain = self._build_chain()

    def _build_chain(self):
        """RAG 체인 구성"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """당신은 사내 문서 기반 Q&A 어시스턴트입니다.
아래 컨텍스트를 기반으로 질문에 답변하세요.

규칙:
1. 컨텍스트에 있는 정보만 사용하세요.
2. 확실하지 않으면 "관련 문서를 찾지 못했습니다"라고 답하세요.
3. 답변에 출처 문서를 포함하세요.
4. 코드나 명령어가 있으면 코드 블록으로 포맷하세요.

컨텍스트:
{context}"""),
            ("human", "{question}")
        ])

        def format_docs(docs):
            formatted = []
            for i, doc in enumerate(docs):
                source = doc.metadata.get("source", "unknown")
                formatted.append(f"[문서 {i+1}] ({source})\n{doc.page_content}")
            return "\n\n---\n\n".join(formatted)

        chain = (
            {"context": self.retriever | format_docs, "question": RunnablePassthrough()}
            | prompt
            | self.llm
            | StrOutputParser()
        )

        return chain

    def ask(self, question: str) -> dict:
        """질문에 답변"""
        # 관련 문서 검색
        docs = self.retriever.invoke(question)

        # LLM 생성
        answer = self.chain.invoke(question)

        # 출처 문서 정보
        sources = list(set(
            doc.metadata.get("source", "unknown") for doc in docs
        ))

        return {
            "answer": answer,
            "sources": sources,
            "num_docs": len(docs)
        }

    def refresh_index(self):
        """인덱스 새로고침"""
        from indexer import index_documents
        self.vectorstore = index_documents()
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",
            search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7}
        )
        self.chain = self._build_chain()

1. https://api.slack.com/apps 에서 새 앱 생성
2. Socket Mode 활성화
3. Bot Token Scopes 추가:
   - app_mentions:read
   - chat:write
   - im:history
   - im:read
   - im:write
4. Event Subscriptions 활성화:
   - app_mention
   - message.im
5. 워크스페이스에 설치

# main.py
import os
import logging
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from rag_chain import RAGChain
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO)

# Slack App 초기화
app = App(token=os.environ["SLACK_BOT_TOKEN"])

# RAG Chain 초기화
rag = RAGChain()


@app.event("app_mention")
def handle_mention(event, say, client):
    """@멘션으로 질문받기"""
    user = event["user"]
    text = event["text"]
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])

    # 봇 멘션 제거
    question = text.split(">", 1)[-1].strip()

    if not question:
        say(
            text="질문을 입력해주세요! 예: `@DocBot 배포 절차 알려줘`",
            thread_ts=thread_ts
        )
        return

    # 로딩 메시지
    loading_msg = client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text=":mag: 문서를 검색하고 있습니다..."
    )

    try:
        # RAG 질의
        result = rag.ask(question)

        # 응답 포맷
        response = f"<@{user}>\n\n{result['answer']}"

        if result["sources"]:
            sources_text = "\n".join(f"• `{s}`" for s in result["sources"])
            response += f"\n\n:page_facing_up: *참고 문서:*\n{sources_text}"

        # 로딩 메시지 업데이트
        client.chat_update(
            channel=channel,
            ts=loading_msg["ts"],
            text=response
        )

    except Exception as e:
        logging.error(f"RAG error: {e}")
        client.chat_update(
            channel=channel,
            ts=loading_msg["ts"],
            text=f"죄송합니다, 오류가 발생했습니다: {str(e)}"
        )


@app.event("message")
def handle_dm(event, say):
    """DM으로 질문받기"""
    if event.get("channel_type") != "im":
        return
    if event.get("bot_id"):
        return

    question = event["text"]

    try:
        result = rag.ask(question)

        response = result["answer"]
        if result["sources"]:
            sources_text = "\n".join(f"• `{s}`" for s in result["sources"])
            response += f"\n\n:page_facing_up: *참고 문서:*\n{sources_text}"

        say(text=response)

    except Exception as e:
        say(text=f"오류가 발생했습니다: {str(e)}")


@app.command("/docbot-reindex")
def handle_reindex(ack, say):
    """슬래시 커맨드로 인덱스 새로고침"""
    ack()
    say("인덱스를 새로고침합니다... :hourglass_flowing_sand:")

    try:
        rag.refresh_index()
        say("인덱스 새로고침 완료! :white_check_mark:")
    except Exception as e:
        say(f"인덱스 새로고침 실패: {str(e)}")


if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    print("Slack RAG Bot started!")
    handler.start()

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 인덱싱 후 봇 시작
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  slack-rag-bot:
    build: .
    env_file: .env
    volumes:
      - ./documents:/app/documents
      - ./chroma_db:/app/chroma_db
    restart: unless-stopped

# 빌드 및 실행
docker compose up -d

# 로그 확인
docker compose logs -f

from langchain.storage import LocalFileStore
from langchain.embeddings import CacheBackedEmbeddings

store = LocalFileStore("./embedding_cache")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    document_embedding_cache=store,
    namespace="text-embedding-3-small"
)

from langchain.memory import ConversationBufferWindowMemory

# 스레드별 메모리 관리
thread_memories = {}

def get_memory(thread_ts: str) -> ConversationBufferWindowMemory:
    if thread_ts not in thread_memories:
        thread_memories[thread_ts] = ConversationBufferWindowMemory(
            k=5,
            memory_key="chat_history",
            return_messages=True
        )
    return thread_memories[thread_ts]