필사 모드: 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()

# 실행
python main.py

# 우리가 직접 찍은 로그가 이 순서로 나오면 정상입니다.
# (Bolt가 자체적으로 출력하는 문구는 버전마다 다르니 기준으로 삼지 마세요.)
[INFO] socket mode handler started
[INFO] event=app_mention channel=C123ABC456 user=U061F7AUR ts=1515449522.000016
[INFO] question='배포 절차 알려줘' thread_ts=None
[INFO] retrieved 5 chunks in 0.42s
[INFO] llm answered in 6.1s len=1842
[INFO] chat_update ok

ack       Slack 서버에 수신 확인을 돌려준다
say       연결된 채널 ID로 chat.postMessage 를 호출한다
respond   연결된 response_url 을 이용한다
body      파싱된 요청 본문 전체
payload   요청 본문에서 핵심 데이터만 벗겨낸 것
event / message / command / action / shortcut / view / options
          각 리스너에서 payload 의 별칭
client    유효한 토큰이 들어간 WebClient 인스턴스
logger    로거
context   BoltContext 인스턴스
next      미들웨어 체인의 다음 단계로

# 인자는 이름으로 주입됩니다. 순서도, 전부 다 받을 필요도 없습니다.
@app.event("app_mention")
def handle_mention(event, say, client, logger):
    logger.info(
        "event=app_mention channel=%s user=%s ts=%s thread_ts=%s",
        event["channel"],
        event["user"],
        event["ts"],
        event.get("thread_ts"),
    )

14:02:10.114  event=app_mention ts=1515449522.000016   답변 생성 시작
14:02:10.140  event=app_mention ts=1515449522.000016   <- 재시도 1
14:03:10.203  event=app_mention ts=1515449522.000016   <- 재시도 2
14:08:10.377  event=app_mention ts=1515449522.000016   <- 재시도 3

# ts가 전부 같습니다. 사용자는 한 번 물었고, 우리는 네 번 답했습니다.

import threading

@app.event("app_mention")
def handle_mention(event, client, logger):
    # 리스너는 즉시 반환합니다. RAG는 여기서 돌리지 않습니다.
    threading.Thread(
        target=answer_in_background,
        args=(event, client, logger),
        daemon=True,
    ).start()


@app.command("/docbot-reindex")
def handle_reindex(ack, say):
    ack()  # 슬래시 커맨드는 문서가 ack()를 명시적으로 요구합니다.
    threading.Thread(target=reindex_in_background, args=(say,), daemon=True).start()

from collections import OrderedDict
import threading

_seen = OrderedDict()
_seen_lock = threading.Lock()


def already_handled(body) -> bool:
    """같은 event_id가 다시 오면 True. Slack의 재전송을 걸러냅니다."""
    event_id = body.get("event_id")
    if not event_id:
        return False
    with _seen_lock:
        if event_id in _seen:
            return True
        _seen[event_id] = True
        while len(_seen) > 5000:
            _seen.popitem(last=False)
    return False


@app.event("app_mention")
def handle_mention(body, event, client, logger):
    if already_handled(body):
        logger.info("duplicate event_id=%s skipped", body.get("event_id"))
        return
    threading.Thread(
        target=answer_in_background, args=(event, client, logger), daemon=True
    ).start()

{
  "type": "app_mention",
  "user": "U061F7AUR",
  "text": "<@U0LAN0Z89> is it everything a river should be?",
  "ts": "1515449522.000016",
  "channel": "C123ABC456",
  "event_ts": "1515449522000016"
}

# 최상위 멘션이면 이 메시지 자신이 스레드의 시작점이 됩니다.
# 스레드 안 멘션이면 부모의 ts가 이미 thread_ts에 들어 있습니다.
thread_ts = event.get("thread_ts") or event["ts"]

say(text=answer, thread_ts=thread_ts)

SECTION_LIMIT = 3000   # Block Kit section 블록의 text 상한
TEXT_LIMIT = 4000      # chat.postMessage / chat.update의 권장 상한


def chunk_for_slack(answer: str, limit: int = TEXT_LIMIT) -> list[str]:
    """문단 경계에서 자릅니다. 아무 데서나 끊으면 코드 블록이 깨집니다."""
    parts, buf = [], ""
    for para in answer.split("\n\n"):
        if len(buf) + len(para) + 2 > limit:
            if buf:
                parts.append(buf)
                buf = ""
            while len(para) > limit:
                parts.append(para[:limit])
                para = para[limit:]
            buf = para
        else:
            buf = f"{buf}\n\n{para}" if buf else para
    if buf:
        parts.append(buf)
    return parts

증상    답변이 문장 중간에서 뚝 끊긴다
확인    len(answer) 를 로그로 찍는다  ->  3214
진단    Block Kit section 블록(3,000)에 넣었다. text 필드(4,000)였다면 통과했을 길이다.

증상    답변은 나오는데 몇 초 뒤 갱신이 멈춘다
확인    응답 코드와 헤더를 찍는다  ->  429 / Retry-After: 12
진단    chat.update 를 초당 여러 번 불렀다. Tier 3(분당 50회 이상)를 넘겼다.

[채널 #dev-help]
지우     @DocBot 스테이징 배포 절차 알려줘                14:02:10

DocBot   :mag: 문서를 검색하고 있습니다...                14:02:10
         (같은 메시지가 6초 뒤 답변으로 바뀝니다)

DocBot   @지우                                            14:02:16
         스테이징 배포는 다음 순서입니다.
         1. main 브랜치에 머지
         2. CI 통과 확인
         3. deploy-staging 워크플로 수동 실행

         :page_facing_up: 참고 문서:
         - documents/deployment-guide.md
         - documents/ci-cd.md

# main.py — 확인 응답 / 중복 제거 / 스레드 / 길이까지 반영한 형태
import os
import threading
import logging

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from rag_chain import RAGChain

logging.basicConfig(level=logging.INFO)
app = App(token=os.environ["SLACK_BOT_TOKEN"])
rag = RAGChain()


def answer_in_background(event, client, logger):
    channel = event["channel"]
    thread_ts = event.get("thread_ts") or event["ts"]
    question = event["text"].split(">", 1)[-1].strip()
    logger.info("question=%r thread_ts=%s", question, event.get("thread_ts"))

    placeholder = client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text=":mag: 문서를 검색하고 있습니다...",
    )
    try:
        result = rag.ask(question)
        body = f"<@{event['user']}>\n\n{result['answer']}"
        if result["sources"]:
            lines = "\n".join(f"- `{s}`" for s in result["sources"])
            body += f"\n\n:page_facing_up: *참고 문서:*\n{lines}"

        parts = chunk_for_slack(body)
        client.chat_update(channel=channel, ts=placeholder["ts"], text=parts[0])
        for extra in parts[1:]:
            client.chat_postMessage(channel=channel, thread_ts=thread_ts, text=extra)
        logger.info("answered len=%d parts=%d", len(body), len(parts))
    except Exception:
        # 예외 문자열을 채널에 그대로 올리지 않습니다. 자세한 건 로그로.
        logger.exception("rag failed")
        client.chat_update(
            channel=channel,
            ts=placeholder["ts"],
            text="답변을 만들지 못했습니다. 잠시 후 다시 시도해주세요.",
        )


@app.event("app_mention")
def handle_mention(body, event, client, logger):
    if already_handled(body):
        return
    threading.Thread(
        target=answer_in_background, args=(event, client, logger), daemon=True
    ).start()


if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

[INFO] event=app_mention channel=C123ABC456 user=U061F7AUR ts=1515449522.000016 thread_ts=None
[INFO] question='스테이징 배포 절차 알려줘' thread_ts=None
[INFO] retrieved 5 chunks in 0.42s
[INFO] llm answered in 6.1s
[INFO] answered len=1842 parts=1

# 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]

# 2026-08 기준으로 다시 깐다면
pip install \
  "langchain==1.3.15" \
  "langchain-core==1.5.5" \
  "langchain-classic==1.0.8" \
  "langchain-openai" \
  "langchain-text-splitters" \
  "langchain-chroma>=0.1.2" \
  "slack-bolt==1.30.0" \
  "python-dotenv"

# 이 글에 나온 import -> 2026-08 기준 위치
# langchain.memory.ConversationBufferMemory
#   -> langchain_classic.memory.buffer.ConversationBufferMemory
# langchain.embeddings.CacheBackedEmbeddings
#   -> langchain_classic.embeddings.cache.CacheBackedEmbeddings
# langchain.storage.LocalFileStore
#   -> langchain_classic.storage.file_system.LocalFileStore
# langchain.chains.RetrievalQA
#   -> langchain_classic.chains.retrieval_qa.base.RetrievalQA

# 텍스트 분할기는 전용 패키지로 나왔습니다.
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    add_start_index=True,
)

# Chroma도 전용 패키지가 정식입니다.
# (langchain_community.vectorstores.Chroma 는 community 0.2.9부터 deprecated)
from langchain_chroma import Chroma

# LCEL 조각들은 langchain-core에 그대로 남아 있습니다.
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document

# 검색기 옵션 정리
#   search_type   : 'similarity'(기본) / 'mmr' / 'similarity_score_threshold'
#   search_kwargs : k(기본 4), score_threshold, fetch_k(기본 20),
#                   lambda_mult(기본 0.5), filter
#   이 글이 쓴 lambda_mult=0.7 은 다양성보다 관련도에 무게를 둔 값입니다.

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[search_internal_docs],
    checkpointer=InMemorySaver(),
)

# 여기가 Slack 봇과 정확히 맞아떨어지는 지점입니다.
# 스레드 하나 = 대화 하나. thread_ts를 그대로 thread_id로 씁니다.
thread_ts = event.get("thread_ts") or event["ts"]
thread_config = {"configurable": {"thread_id": thread_ts}}

# langchain-core의 VectorStore 구현 일부
async def asimilarity_search(self, query: str, k: int = 4, **kwargs: Any) -> list[Document]:
    return await run_in_executor(None, self.similarity_search, query, k=k, **kwargs)


# Bolt 비동기 쪽 임포트 경로도 함께 적어둡니다.
# from slack_bolt.app.async_app import AsyncApp          (짧은 별칭: slack_bolt.async_app)
# from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
#   (slack_bolt.adapter.socket_mode.aiohttp 에도 있습니다)
# await handler.start_async()

증상                       먼저 볼 것              그다음
같은 답이 2~4회            event_id 중복 여부      3초 초과 -> 재시도
멘션 하나에 답 2회         로그의 이벤트 이름      app_mention + message 이중 구독
검색 결과가 빈다           임베딩 모델 일치        persist_directory / 컬렉션 이름
답이 문장 중간에서 끊긴다  답변 길이               3,000(Block Kit) 대 4,000(text)
스레드 밖에 답이 뜬다      thread_ts 로그값        답글의 ts 를 그대로 쓰고 있는지

# ask()가 프롬프트 길이를 함께 돌려주도록 한 줄만 늘려두면
# 비용 문제는 청구서가 아니라 로그에서 먼저 보입니다.
try:
    result = rag.ask(question)
    logger.info("prompt_chars=%d chunks=%d", result["prompt_chars"], result["num_docs"])
except Exception:
    logger.exception("rag failed question=%r", question[:200])
    client.chat_update(
        channel=channel,
        ts=placeholder["ts"],
        text="답변을 만들지 못했습니다. 잠시 후 다시 시도해주세요.",
    )

# 사람이 문서를 열 때
사용자 -> 문서 저장소 -> 권한 확인 -> 열람 허용 또는 거부

# 봇이 답할 때
사용자 -> 봇 -> 봇의 권한으로 이미 인덱싱된 벡터 DB -> 답변
              (질문한 사람이 누구인지는 여기서 아무 영향이 없습니다)