LangChain + RAG로 지능형 Telegram FAQ 봇 만들기: 문서 기반 질의응답 시스템

사용자 질문
    ↓
Telegram Bot API
    ↓
LangChain RAG Pipeline
    ├── 1. Query Embedding (OpenAI)
    ├── 2. Vector Search (ChromaDB)
    ├── 3. Context Retrieval (Top-K)
    └── 4. LLM Generation (GPT-4o)
    ↓
답변 + 출처 표시

pip install langchain langchain-openai langchain-community \
  chromadb python-telegram-bot tiktoken \
  pypdf docx2txt unstructured

# config.py
import os

TELEGRAM_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

# RAG 설정
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
TOP_K = 4
MODEL_NAME = "gpt-4o"
EMBEDDING_MODEL = "text-embedding-3-small"

# indexer.py
from langchain_community.document_loaders import (
    DirectoryLoader,
    PyPDFLoader,
    TextLoader,
    Docx2txtLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

def load_documents(docs_dir: str):
    """다양한 형식의 문서 로딩"""
    loaders = {
        "**/*.pdf": PyPDFLoader,
        "**/*.txt": TextLoader,
        "**/*.md": TextLoader,
        "**/*.docx": Docx2txtLoader,
    }

    all_docs = []
    for glob_pattern, loader_cls in loaders.items():
        loader = DirectoryLoader(
            docs_dir,
            glob=glob_pattern,
            loader_cls=loader_cls,
            show_progress=True,
        )
        docs = loader.load()
        all_docs.extend(docs)
        print(f"Loaded {len(docs)} docs from {glob_pattern}")

    return all_docs

def create_vector_store(docs_dir: str, persist_dir: str = "./chroma_db"):
    """문서를 청크로 분할하고 벡터 스토어에 저장"""
    # 문서 로딩
    documents = load_documents(docs_dir)
    print(f"Total documents: {len(documents)}")

    # 텍스트 분할
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["\n\n", "\n", ".", "!", "?", ",", " "],
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Total chunks: {len(chunks)}")

    # 임베딩 생성 & 벡터 스토어 저장
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_metadata={"hnsw:space": "cosine"},
    )

    print(f"Vector store created at {persist_dir}")
    return vectorstore

if __name__ == "__main__":
    create_vector_store("./docs")

# rag_chain.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

SYSTEM_PROMPT = """당신은 회사 FAQ 도우미입니다. 제공된 컨텍스트를 기반으로 질문에 답변하세요.

규칙:
1. 컨텍스트에 있는 정보만 사용하세요.
2. 확실하지 않으면 "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 답하세요.
3. 답변 끝에 참고한 문서 출처를 표시하세요.
4. 간결하고 명확하게 답변하세요.

컨텍스트:
{context}"""

def create_rag_chain(persist_dir: str = "./chroma_db"):
    """RAG 체인 생성"""
    # 벡터 스토어 로드
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        persist_directory=persist_dir,
        embedding_function=embeddings,
    )

    # 리트리버 설정
    retriever = vectorstore.as_retriever(
        search_type="mmr",  # Maximal Marginal Relevance
        search_kwargs={
            "k": 4,
            "fetch_k": 10,
            "lambda_mult": 0.7,
        },
    )

    # LLM
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0.1,
        max_tokens=1024,
    )

    # 대화 메모리 (최근 5턴)
    memory = ConversationBufferWindowMemory(
        k=5,
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
    )

    # 프롬프트
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(SYSTEM_PROMPT),
        HumanMessagePromptTemplate.from_template("{question}"),
    ])

    # 체인 생성
    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True,
        combine_docs_chain_kwargs={"prompt": prompt},
        verbose=False,
    )

    return chain

class RAGBot:
    """사용자별 대화 컨텍스트를 관리하는 RAG 봇"""

    def __init__(self, persist_dir: str = "./chroma_db"):
        self.persist_dir = persist_dir
        self.user_chains: dict[int, ConversationalRetrievalChain] = {}

    def get_chain(self, user_id: int):
        """사용자별 체인 (대화 메모리 분리)"""
        if user_id not in self.user_chains:
            self.user_chains[user_id] = create_rag_chain(self.persist_dir)
        return self.user_chains[user_id]

    async def ask(self, user_id: int, question: str) -> tuple[str, list[str]]:
        """질문에 답변하고 출처를 반환"""
        chain = self.get_chain(user_id)
        result = chain.invoke({"question": question})

        answer = result["answer"]
        sources = []
        for doc in result.get("source_documents", []):
            source = doc.metadata.get("source", "Unknown")
            page = doc.metadata.get("page", "")
            if page:
                sources.append(f"{source} (p.{page})")
            else:
                sources.append(source)

        # 중복 제거
        sources = list(dict.fromkeys(sources))
        return answer, sources

    def reset_memory(self, user_id: int):
        """사용자의 대화 메모리 초기화"""
        if user_id in self.user_chains:
            del self.user_chains[user_id]

# bot.py
import logging
from telegram import Update, BotCommand
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes,
)
from rag_chain import RAGBot
from config import TELEGRAM_TOKEN

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

rag_bot = RAGBot()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """시작 명령어"""
    welcome = (
        "안녕하세요! FAQ 도우미입니다.\n\n"
        "궁금한 것을 자유롭게 물어보세요.\n"
        "회사 문서를 기반으로 답변해 드립니다.\n\n"
        "명령어:\n"
        "/reset - 대화 초기화\n"
        "/sources - 검색 가능한 문서 목록"
    )
    await update.message.reply_text(welcome)

async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """대화 메모리 초기화"""
    user_id = update.effective_user.id
    rag_bot.reset_memory(user_id)
    await update.message.reply_text("대화가 초기화되었습니다.")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """일반 메시지 처리"""
    user_id = update.effective_user.id
    question = update.message.text

    # 타이핑 표시
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

    try:
        answer, sources = await rag_bot.ask(user_id, question)

        # 답변 포맷팅
        response = answer
        if sources:
            response += "\n\n📚 참고 문서:\n"
            for src in sources[:3]:
                response += f"  • {src}\n"

        await update.message.reply_text(response)

    except Exception as e:
        logger.error(f"Error: {e}")
        await update.message.reply_text(
            "죄송합니다. 답변을 생성하는 중 오류가 발생했습니다."
        )

async def post_init(application: Application):
    """봇 시작 시 명령어 등록"""
    commands = [
        BotCommand("start", "봇 시작"),
        BotCommand("reset", "대화 초기화"),
        BotCommand("sources", "검색 가능한 문서 목록"),
    ]
    await application.bot.set_my_commands(commands)

def main():
    app = Application.builder().token(TELEGRAM_TOKEN).post_init(post_init).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("reset", reset))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    logger.info("Bot started")
    app.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

# 문서 인덱싱
RUN python indexer.py

CMD ["python", "bot.py"]

# docker-compose.yml
services:
  faq-bot:
    build: .
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./docs:/app/docs
      - chroma-data:/app/chroma_db
    restart: unless-stopped

volumes:
  chroma-data:

docker-compose up -d

# watcher.py - 문서 변경 감지 및 자동 재인덱싱
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class DocChangeHandler(FileSystemEventHandler):
    def __init__(self, indexer_fn):
        self.indexer_fn = indexer_fn
        self.last_indexed = 0

    def on_modified(self, event):
        if event.is_directory:
            return
        # 디바운스 (5초 이내 중복 방지)
        now = time.time()
        if now - self.last_indexed < 5:
            return
        self.last_indexed = now

        print(f"Document changed: {event.src_path}")
        self.indexer_fn()

def watch_docs(docs_dir, indexer_fn):
    handler = DocChangeHandler(indexer_fn)
    observer = Observer()
    observer.schedule(handler, docs_dir, recursive=True)
    observer.start()
    return observer

from functools import lru_cache
import hashlib

class CachedRAGBot(RAGBot):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache: dict[str, tuple[str, list[str]]] = {}

    async def ask(self, user_id: int, question: str):
        cache_key = hashlib.md5(question.lower().strip().encode()).hexdigest()

        if cache_key in self.cache:
            return self.cache[cache_key]

        answer, sources = await super().ask(user_id, question)
        self.cache[cache_key] = (answer, sources)
        return answer, sources