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

# 이 글의 import → 2026-08 기준 위치
langchain.chains.ConversationalRetrievalChain
  → langchain_classic.chains.conversational_retrieval.base.ConversationalRetrievalChain
    (0.1.17부터 deprecated)
langchain.chains.RetrievalQA
  → langchain_classic.chains.retrieval_qa.base.RetrievalQA
langchain.memory.ConversationBufferMemory
  → langchain_classic.memory.buffer.ConversationBufferMemory
langchain.text_splitter.RecursiveCharacterTextSplitter
  → langchain_text_splitters.RecursiveCharacterTextSplitter
langchain_community.vectorstores.Chroma
  → langchain_chroma.Chroma   (community 0.2.9부터 deprecated)
langchain_community.vectorstores.FAISS
  → 이동 경로 없음. langchain-faiss 패키지는 존재하지 않는다.

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

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

thread_config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    thread_config,
)

pip install "python-telegram-bot[rate-limiter]==22.8" \
  langchain langchain-classic langchain-openai \
  langchain-chroma langchain-text-splitters \
  chromadb tiktoken pypdf docx2txt

사용자 질문
    ↓
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")

$ python indexer.py
100%|█████████████████████████| 12/12 [00:04<00:00,  2.71it/s]
Loaded 47 docs from **/*.pdf
100%|█████████████████████████| 31/31 [00:00<00:00, 240.11it/s]
Loaded 31 docs from **/*.txt
Loaded 9 docs from **/*.md
Loaded 3 docs from **/*.docx
Total documents: 90
Total chunks: 412
Vector store created at ./chroma_db

# smoke_test.py — 봇 없이 검색만 확인한다
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = Chroma(
    embedding_function=embeddings,
    persist_directory="./chroma_db",
)

QUERIES = [
    "연차는 며칠부터 쓸 수 있나요",   # 문서에 있는 질문
    "재택근무 신청 절차",             # 문서에 있는 질문
    "회사 강아지 이름이 뭐야",        # 문서에 없는 질문 = 음성 대조군
]

for q in QUERIES:
    docs = store.similarity_search(q, k=3)
    print(f"\nQ: {q}  -> {len(docs)} hits")
    for d in docs:
        src = d.metadata.get("source", "?")
        print(f"   {src}: {d.page_content[:60]}...")

Q: 연차는 며칠부터 쓸 수 있나요  -> 3 hits
   docs/hr-policy.pdf: 연차 유급휴가는 입사일로부터 1년간 80퍼센트 이상 출근한 근로자에게...
   docs/hr-policy.pdf: 연차 사용 촉진 제도에 따라 미사용 연차는 매년 12월 31일 기준으로...
   docs/onboarding.md: 입사 첫 해에는 1개월 개근 시 1일의 유급휴가가 발생하며...

Q: 회사 강아지 이름이 뭐야  -> 3 hits
   docs/office-guide.md: 사무실 출입 카드는 1층 안내데스크에서 수령하며 분실 시...
   docs/hr-policy.pdf: 경조사 휴가는 다음 기준에 따라 부여합니다. 본인 결혼 5일...
   docs/onboarding.md: 슬랙 채널 안내는 다음과 같습니다. 전사 공지는 general 채널...

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

retriever = store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 4, "score_threshold": 0.5},
)

docs = retriever.invoke(question)
if not docs:
    # LLM을 아예 호출하지 않는다. 여기서 끝내는 것이 정답이다.
    return "제공된 문서에서 해당 정보를 찾을 수 없습니다.", []

logger.info(
    "retrieval chat_id=%s q=%r hits=%d sources=%s",
    chat_id,
    question[:80],
    len(docs),
    [d.metadata.get("source") for d in docs],
)

async def debug(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = " ".join(context.args)
    if not query:
        await update.message.reply_text("사용법: /debug 검색어")
        return

    docs = retriever.invoke(query)
    if not docs:
        await update.message.reply_text("검색 결과 없음 (임계값 미달)")
        return

    lines = []
    for i, d in enumerate(docs, 1):
        src = d.metadata.get("source", "?")
        lines.append(f"{i}. {src}\n   {d.page_content[:120]}")
    await update.message.reply_text("\n".join(lines))

application.add_handler(CommandHandler("debug", debug))

# 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()

from telegram.constants import MessageLimit

def split_for_telegram(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]:
    """문단 경계 -> 줄 경계 -> 강제 절단 순으로 자른다."""
    if len(text) <= limit:
        return [text]

    parts: list[str] = []
    buf = ""
    for block in text.split("\n\n"):
        if len(block) > limit:
            for line in block.split("\n"):
                while len(line) > limit:
                    parts.append(line[:limit])
                    line = line[limit:]
                if len(buf) + len(line) + 1 > limit:
                    parts.append(buf)
                    buf = line
                else:
                    buf = f"{buf}\n{line}" if buf else line
            continue
        if len(buf) + len(block) + 2 > limit:
            parts.append(buf)
            buf = block
        else:
            buf = f"{buf}\n\n{block}" if buf else block
    if buf:
        parts.append(buf)
    return parts

async def reply_long(update: Update, text: str) -> None:
    for part in split_for_telegram(text):
        await update.message.reply_text(part)

from telegram.constants import ChatAction

await context.bot.send_chat_action(
    chat_id=update.effective_chat.id,
    action=ChatAction.TYPING,
)

# https://core.telegram.org/bots/faq
"In a single chat, avoid sending more than one message per second."
"In a group, bots are not be able to send more than 20 messages per minute."
"For bulk notifications, bots are not able to broadcast more than about
 30 messages per second, unless they enable paid broadcasts."

pip install "python-telegram-bot[rate-limiter]"

from telegram.ext import AIORateLimiter, ApplicationBuilder

application = (
    ApplicationBuilder()
    .token(TELEGRAM_TOKEN)
    .rate_limiter(AIORateLimiter())
    .concurrent_updates(True)
    .build()
)

from telegram.ext import ApplicationBuilder, PicklePersistence

persistence = PicklePersistence(filepath="bot_state.pickle")

application = (
    ApplicationBuilder()
    .token(TELEGRAM_TOKEN)
    .persistence(persistence)
    .build()
)

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

# 프로덕션에서는 InMemorySaver 대신 PostgresSaver를 쓰라고 문서가 안내한다.
# 정확한 import 경로는 사용 중인 버전의 문서에서 확인할 것.
agent = create_agent(model="openai:gpt-5.5", tools=[], checkpointer=InMemorySaver())

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    config = {"configurable": {"thread_id": str(update.effective_chat.id)}}
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": update.message.text}]},
        config,
    )
    # 반환 구조는 버전에 따라 다르다. 정확한 API는 사용 중인 버전의 문서에서 확인할 것.
    await reply_long(update, extract_answer(result))

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)

run_polling(poll_interval=0.0, timeout=datetime.timedelta(seconds=10),
            bootstrap_retries=0, allowed_updates=None, drop_pending_updates=None,
            close_loop=True, stop_signals=None)

run_webhook(listen='127.0.0.1', port=80, url_path='', cert=None, key=None,
            bootstrap_retries=0, webhook_url=None, allowed_updates=None,
            drop_pending_updates=None, ip_address=None, max_connections=40,
            close_loop=True, stop_signals=None, secret_token=None, unix=None)

# 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