내 PDF 문서를 AI가 읽는다 — 사내 지식 RAG 챗봇 구축

📌 0. 시리즈

📌 1. 들어가며

이 포스트에서 만들 것

이 포스트를 끝까지 따라하면 사내 PDF / Word 문서를 넣기만 하면 아래 결과물이 완성됩니다.

입력: 사내 문서 폴더 (PDF, DOCX, TXT)
  예: 제품 매뉴얼, 사규, 영업 자료, 기술 문서
   ↓
[1단계] 문서 인덱싱
  └─ 청크 분할 → 임베딩 생성 → ChromaDB 저장

[2단계] 질의응답
  └─ 질문 → 유사 청크 검색 → LLM 답변 생성

출력:
  ├─ 웹 챗봇 UI (Gradio / Streamlit)
  ├─ 출처 문서명 + 페이지 번호 표시
  └─ REST API 서버 (사내 시스템 연동용)

사용 예시:
  질문: "연차 신청 절차가 어떻게 되나요?"
  답변: "사규 3조 2항에 따르면 연차 신청은 3일 전에...
         [출처: 인사규정.pdf, 12페이지]"

RAG란 — 모델을 바꾸지 않고 지식을 추가하는 방법

일반 LLM의 한계:
  ├─ 학습 데이터 마감일 이후 정보 모름
  ├─ 사내 비공개 문서 내용 모름
  └─ 출처 없이 그럴듯한 거짓말 생성 (Hallucination)

RAG (Retrieval-Augmented Generation):
  파인튜닝 없이 외부 지식을 실시간으로 LLM에 주입하는 기법

  [사용자 질문]
       ↓
  [벡터 DB에서 관련 문서 검색]
       ↓
  [검색된 문서 + 질문을 LLM에 전달]
       ↓
  [LLM이 문서를 근거로 답변 생성]

  핵심 장점:
  ✅ 모델 재학습 불필요 (비용 0)
  ✅ 문서 업데이트 즉시 반영
  ✅ 출처 추적 가능 → Hallucination 대폭 감소
  ✅ 사내 기밀 문서를 외부로 보내지 않아도 됨 (로컬 실행)

📌 2. 환경 준비

2-1. CPU만 있어도 동작 가능

GPU 있는 경우:
  → Ollama + Qwen2.5:7b / Llama3.2 로컬 실행
  → 임베딩 모델 GPU 가속
  → 빠른 응답 (1~3초)

CPU만 있는 경우:
  → Ollama + Llama3.2:3b (경량 모델)
  → 임베딩은 CPU로도 충분 (bge-m3 small)
  → 응답 시간 5~15초 (실용적 수준)

API 방식 (외부 서버 활용):
  → OpenAI GPT-4o-mini API (저비용)
  → Ollama 없이 바로 시작 가능
  → 비용: 약 $0.0001 / 질문 1회

2-2. LlamaIndex, ChromaDB, sentence-transformers 설치

bash# Python 환경 생성
python -m venv venv
source venv/bin/activate       # Linux/Mac
venv\Scripts\activate          # Windows

# 핵심 RAG 프레임워크
pip install llama-index
pip install llama-index-vector-stores-chroma
pip install llama-index-embeddings-huggingface
pip install llama-index-llms-ollama

# 벡터 DB
pip install chromadb            # 로컬 경량 벡터 DB
pip install qdrant-client        # 대용량 벡터 DB (선택)

# 임베딩 모델
pip install sentence-transformers FlagEmbedding   # bge-m3

# 문서 로더
pip install pymupdf              # PDF (PyMuPDF)
pip install pdfplumber           # PDF 표/레이아웃 특화
pip install python-docx          # DOCX
pip install openpyxl             # XLSX

# UI
pip install gradio
pip install streamlit

# Ollama 설치 (로컬 LLM)
# https://ollama.com/download
ollama pull qwen2.5:7b           # 권장 (한국어 우수)
ollama pull llama3.2:3b          # 경량 옵션

# 설치 확인
python -c "import llama_index; import chromadb; print('✅ 설치 완료')"

📌 3. RAG 핵심 개념

3-1. 임베딩이란 — 텍스트를 숫자 벡터로 바꾸는 원리

임베딩(Embedding):
  텍스트를 의미가 담긴 고차원 숫자 벡터로 변환하는 과정

예시:
  "연차 신청 방법" → [0.23, -0.41, 0.87, ..., 0.15]  (768차원)
  "휴가 사용 절차" → [0.21, -0.39, 0.85, ..., 0.13]  (768차원)
  "파이썬 설치법" → [-0.55, 0.72, -0.31, ..., 0.88]  (768차원)

의미가 비슷할수록 벡터가 가깝다:
  "연차 신청 방법" ↔ "휴가 사용 절차"  → 코사인 유사도 0.95 (매우 유사)
  "연차 신청 방법" ↔ "파이썬 설치법"  → 코사인 유사도 0.12 (관련 없음)

코사인 유사도 공식:
  similarity = (A · B) / (||A|| × ||B||)
  → 1에 가까울수록 의미가 같음
  → 0에 가까울수록 의미가 다름

한국어 최적 임베딩 모델 (2026 기준):
  bge-m3              → 다국어 최강, 한국어 우수, 무료
  multilingual-e5-large → 다국어 균형, 경량
  ko-sroberta-multitask → 한국어 특화, 빠름

3-2. 벡터 DB란 — 유사한 벡터를 빠르게 찾는 구조

일반 DB (MySQL, PostgreSQL):
  → 정확한 키워드 일치 검색
  → "연차" 검색 → "연차" 포함된 문서만 반환
  → "휴가 사용" 검색하면 "연차" 포함 문서 못 찾음

벡터 DB (ChromaDB, Qdrant, Pinecone):
  → 의미 유사도 기반 검색 (ANN: Approximate Nearest Neighbor)
  → "연차 신청 방법" 검색 → "휴가 사용 절차" 문서도 반환
  → 키워드가 달라도 의미가 같으면 검색됨

내부 인덱싱 구조:
  HNSW (Hierarchical Navigable Small World)
    → 벡터들을 그래프로 연결
    → 검색 시 가까운 노드를 따라가며 최근접 벡터 탐색
    → 100만 개 벡터도 수십 ms 내 검색 가능

3-3. 청크 분할 전략

왜 청크로 나누나:
  LLM의 컨텍스트 창(Context Window)에는 한계가 있음
  → 100페이지 문서를 통째로 넣을 수 없음
  → 적절한 크기로 잘라서 관련 부분만 전달

[전략 1] 고정 크기 분할 (Fixed-size Chunking)
  장점: 구현 단순, 예측 가능한 토큰 수
  단점: 문장 중간에서 잘릴 수 있음

  문서 전체
  ├─ 청크 1: 0~500 토큰
  ├─ 청크 2: 500~1000 토큰  (50토큰 오버랩)
  ├─ 청크 3: 950~1450 토큰  (오버랩으로 문맥 유지)
  └─ ...

  chunk_size    = 512   # 토큰 수
  chunk_overlap = 50    # 앞뒤 청크와 겹치는 토큰 수

[전략 2] 의미 단위 분할 (Semantic Chunking)
  장점: 의미 완결 단위로 분할 → 검색 정확도↑
  단점: 처리 시간 다소 느림

  문서 전체
  ├─ 청크 1: "제 1장 총칙..."   ← 의미 단락 기준
  ├─ 청크 2: "제 2조 연차..."   ← 제목/단락 기준
  └─ 청크 3: "별표 1 서식..."   ← 섹션 기준

  → 제목, 줄바꿈 2회, 구분선 등을 기준으로 분할
  → 실무에서 권장

📌 4. [1단계] 문서 인덱싱 파이프라인

4-1. PDF 로드 (PyMuPDF / pdfplumber)

python# indexing/document_loader.py
import fitz          # PyMuPDF
import pdfplumber
import os
from pathlib import Path
from docx import Document as DocxDocument

def load_pdf_pymupdf(pdf_path: str) -> list[dict]:
    """
    PyMuPDF로 PDF 로드
    → 일반 텍스트 PDF에 최적
    반환: [{"text": "...", "page": 1, "source": "파일명"}, ...]
    """
    doc    = fitz.open(pdf_path)
    pages  = []
    source = os.path.basename(pdf_path)

    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text("text").strip()

        if len(text) < 20:    # 너무 짧은 페이지 스킵 (빈 페이지 등)
            continue

        pages.append({
            "text":   text,
            "page":   page_num + 1,
            "source": source,
            "path":   pdf_path,
        })

    doc.close()
    print(f"📄 {source}: {len(pages)}페이지 로드 완료")
    return pages

def load_pdf_pdfplumber(pdf_path: str) -> list[dict]:
    """
    pdfplumber로 PDF 로드
    → 표(Table)가 많은 PDF에 최적 (재무제표, 스펙시트 등)
    """
    pages  = []
    source = os.path.basename(pdf_path)

    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            # 일반 텍스트 추출
            text = page.extract_text() or ""

            # 표 추출 후 텍스트로 변환
            tables = page.extract_tables()
            for table in tables:
                for row in table:
                    row_text = " | ".join(
                        [str(cell) if cell else "" for cell in row]
                    )
                    text += f"\n{row_text}"

            text = text.strip()
            if len(text) < 20:
                continue

            pages.append({
                "text":   text,
                "page":   page_num + 1,
                "source": source,
                "path":   pdf_path,
            })

    print(f"📊 {source}: {len(pages)}페이지 (표 포함) 로드 완료")
    return pages

def load_docx(docx_path: str) -> list[dict]:
    """Word 문서 로드"""
    doc    = DocxDocument(docx_path)
    source = os.path.basename(docx_path)
    chunks = []
    buffer = []
    page_num = 1

    for para in doc.paragraphs:
        text = para.text.strip()
        if not text:
            continue

        # 제목 스타일이면 새 청크 시작
        if para.style.name.startswith("Heading"):
            if buffer:
                chunks.append({
                    "text":   "\n".join(buffer),
                    "page":   page_num,
                    "source": source,
                    "path":   docx_path,
                })
                page_num += 1
                buffer = []
            buffer.append(f"## {text}")
        else:
            buffer.append(text)

    if buffer:
        chunks.append({
            "text":   "\n".join(buffer),
            "page":   page_num,
            "source": source,
            "path":   docx_path,
        })

    print(f"📝 {source}: {len(chunks)}섹션 로드 완료")
    return chunks

def load_documents_from_folder(folder_path: str) -> list[dict]:
    """폴더 내 모든 문서 자동 로드"""
    all_pages = []
    folder    = Path(folder_path)
    files     = list(folder.rglob("*"))

    for file in files:
        ext = file.suffix.lower()
        try:
            if ext == ".pdf":
                # 표 포함 여부에 따라 로더 선택
                pages = load_pdf_pdfplumber(str(file))
            elif ext in [".docx", ".doc"]:
                pages = load_docx(str(file))
            elif ext == ".txt":
                text = file.read_text(encoding="utf-8")
                pages = [{"text": text, "page": 1,
                           "source": file.name, "path": str(file)}]
            else:
                continue

            all_pages.extend(pages)
        except Exception as e:
            print(f"⚠️  {file.name} 로드 실패: {e}")

    print(f"\n✅ 총 {len(all_pages)}페이지 로드 완료")
    return all_pages

4-2. 청크 분할 & 메타데이터 추가

python# indexing/chunker.py
from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser,
)
from llama_index.core import Document
from llama_index.embeddings.huggingface import HuggingFaceEmbedding

def pages_to_documents(pages: list[dict]) -> list[Document]:
    """페이지 딕셔너리 → LlamaIndex Document 객체 변환"""
    documents = []
    for page in pages:
        doc = Document(
            text     = page["text"],
            metadata = {
                "source":   page["source"],
                "page":     page["page"],
                "path":     page["path"],
            },
        )
        documents.append(doc)
    return documents

def chunk_fixed_size(documents: list[Document],
                     chunk_size=512, chunk_overlap=50) -> list:
    """
    고정 크기 청크 분할
    chunk_size:    청크 당 최대 토큰 수
    chunk_overlap: 청크 간 오버랩 토큰 수
    """
    splitter = SentenceSplitter(
        chunk_size    = chunk_size,
        chunk_overlap = chunk_overlap,
    )
    nodes = splitter.get_nodes_from_documents(documents)
    print(f"고정 크기 분할: {len(documents)}페이지 → {len(nodes)}청크")
    return nodes

def chunk_semantic(documents: list[Document],
                   embed_model, buffer_size=1,
                   breakpoint_threshold=95) -> list:
    """
    의미 단위 청크 분할 (Semantic Chunking)
    buffer_size:            문장을 묶는 단위
    breakpoint_threshold:   분할 기준 유사도 임계값 (높을수록 적게 분할)
    """
    splitter = SemanticSplitterNodeParser(
        buffer_size               = buffer_size,
        breakpoint_percentile_threshold = breakpoint_threshold,
        embed_model               = embed_model,
    )
    nodes = splitter.get_nodes_from_documents(documents)
    print(f"의미 단위 분할: {len(documents)}페이지 → {len(nodes)}청크")
    return nodes

def add_metadata_to_nodes(nodes: list, extra_meta: dict = {}) -> list:
    """청크에 추가 메타데이터 삽입"""
    for i, node in enumerate(nodes):
        node.metadata["chunk_id"]    = i
        node.metadata["chunk_total"] = len(nodes)
        node.metadata.update(extra_meta)

        # 청크 미리보기 (디버깅용)
        preview = node.text[:80].replace("\n", " ")
        node.metadata["preview"] = preview

    return nodes

4-3. bge-m3 임베딩 생성

python# indexing/embedder.py
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from FlagEmbedding import BGEM3FlagModel
import numpy as np

def load_bge_m3_llamaindex(device="cuda"):
    """
    LlamaIndex 통합 bge-m3 임베딩 모델
    → LlamaIndex 파이프라인과 바로 연동
    """
    embed_model = HuggingFaceEmbedding(
        model_name  = "BAAI/bge-m3",
        device      = device,
        embed_batch_size = 32,
    )
    print(f"✅ bge-m3 임베딩 모델 로드 완료 (device: {device})")
    return embed_model

def load_bge_m3_flagembedding():
    """
    FlagEmbedding 직접 사용 (더 많은 옵션)
    → Dense + Sparse + ColBERT 혼합 검색 지원
    """
    model = BGEM3FlagModel(
        "BAAI/bge-m3",
        use_fp16=True    # GPU 메모리 절약
    )
    return model

def embed_texts_batch(model, texts: list[str],
                      batch_size=32) -> np.ndarray:
    """
    텍스트 리스트 → 임베딩 배치 생성
    반환: shape (N, 1024) numpy 배열
    """
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch   = texts[i:i + batch_size]
        encoded = model.encode(
            batch,
            batch_size          = batch_size,
            max_length          = 512,
            return_dense        = True,
            return_sparse       = False,
            return_colbert_vecs = False,
        )
        all_embeddings.append(encoded["dense_vecs"])
        print(f"임베딩 진행: {min(i+batch_size, len(texts))}/{len(texts)}")

    return np.vstack(all_embeddings)

4-4. ChromaDB에 벡터 저장

python# indexing/vector_store.py
import chromadb
from chromadb.config import Settings
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore

def build_chroma_index(nodes, embed_model,
                        collection_name="company_docs",
                        persist_dir="./chroma_db"):
    """
    청크 노드 + 임베딩 → ChromaDB 저장 & 인덱스 생성
    persist_dir: 로컬 저장 경로 (재시작 후에도 유지)
    """
    # ChromaDB 클라이언트 초기화
    chroma_client = chromadb.PersistentClient(
        path     = persist_dir,
        settings = Settings(anonymized_telemetry=False),
    )

    # 컬렉션 생성 (이미 있으면 불러옴)
    chroma_collection = chroma_client.get_or_create_collection(
        name     = collection_name,
        metadata = {"hnsw:space": "cosine"},   # 코사인 유사도
    )

    # LlamaIndex VectorStore 연결
    vector_store    = ChromaVectorStore(
        chroma_collection=chroma_collection
    )
    storage_context = StorageContext.from_defaults(
        vector_store=vector_store
    )

    # 인덱스 빌드 (청크 임베딩 생성 + ChromaDB 저장)
    print(f"🔨 인덱스 빌드 중... ({len(nodes)}개 청크)")
    index = VectorStoreIndex(
        nodes,
        storage_context = storage_context,
        embed_model     = embed_model,
        show_progress   = True,
    )

    print(f"✅ ChromaDB 저장 완료: {persist_dir}/{collection_name}")
    print(f"   총 벡터 수: {chroma_collection.count()}")
    return index

def load_chroma_index(embed_model,
                       collection_name="company_docs",
                       persist_dir="./chroma_db"):
    """
    기존 ChromaDB 인덱스 불러오기
    → 문서 재처리 없이 바로 검색 가능
    """
    chroma_client     = chromadb.PersistentClient(path=persist_dir)
    chroma_collection = chroma_client.get_collection(collection_name)

    vector_store    = ChromaVectorStore(
        chroma_collection=chroma_collection
    )
    storage_context = StorageContext.from_defaults(
        vector_store=vector_store
    )

    index = VectorStoreIndex.from_vector_store(
        vector_store,
        embed_model = embed_model,
    )

    print(f"✅ ChromaDB 인덱스 로드: {chroma_collection.count()}개 벡터")
    return index

# ─── 전체 인덱싱 파이프라인 한 번에 실행 ─────────────────

def run_indexing_pipeline(docs_folder: str,
                           collection_name: str = "company_docs",
                           persist_dir: str     = "./chroma_db",
                           chunk_size: int      = 512):
    """
    폴더 경로 → ChromaDB 인덱스 자동 생성
    """
    from indexing.document_loader import load_documents_from_folder
    from indexing.chunker         import (pages_to_documents,
                                          chunk_fixed_size,
                                          add_metadata_to_nodes)
    from indexing.embedder        import load_bge_m3_llamaindex

    print("=" * 50)
    print("📚 문서 인덱싱 파이프라인 시작")
    print("=" * 50)

    # 1. 문서 로드
    print("\n[1/4] 문서 로드 중...")
    pages = load_documents_from_folder(docs_folder)

    # 2. LlamaIndex Document 변환
    print("\n[2/4] Document 변환 중...")
    documents = pages_to_documents(pages)

    # 3. 임베딩 모델 로드
    print("\n[3/4] 임베딩 모델 로드 중...")
    embed_model = load_bge_m3_llamaindex()

    # 4. 청크 분할
    nodes = chunk_fixed_size(documents, chunk_size=chunk_size)
    nodes = add_metadata_to_nodes(nodes)

    # 5. ChromaDB 저장
    print("\n[4/4] ChromaDB 저장 중...")
    index = build_chroma_index(nodes, embed_model,
                                collection_name, persist_dir)

    print("\n✅ 인덱싱 완료!")
    return index

# 실행 예시
if __name__ == "__main__":
    index = run_indexing_pipeline(
        docs_folder     = "./company_docs",
        collection_name = "company_docs",
        persist_dir     = "./chroma_db",
        chunk_size      = 512,
    )

📌 5. [2단계] 질의응답 파이프라인

5-1. 질문 임베딩 생성

python# retrieval/query_engine.py
from llama_index.core             import VectorStoreIndex
from llama_index.llms.ollama      import Ollama
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers   import VectorIndexRetriever
from llama_index.core.postprocessor import SimilarityPostprocessor

def setup_llm(model_name="qwen2.5:7b", temperature=0.1):
    """
    Ollama 로컬 LLM 설정
    temperature=0.1 → 사실 기반 답변에 최적 (낮을수록 보수적)
    """
    llm = Ollama(
        model       = model_name,
        base_url    = "http://localhost:11434",
        temperature = temperature,
        request_timeout = 120.0,
        context_window  = 8192,
    )
    return llm

5-2. 유사도 검색 (Top-K 청크 추출)

pythondef build_query_engine(index: VectorStoreIndex, llm,
                        top_k: int     = 5,
                        min_score: float = 0.4):
    """
    질의응답 엔진 구성
    top_k:     검색할 청크 수 (많을수록 더 많은 문맥 제공)
    min_score: 최소 유사도 점수 (낮은 청크 필터링)
    """
    # 검색기 설정
    retriever = VectorIndexRetriever(
        index           = index,
        similarity_top_k = top_k,
    )

    # 유사도 임계값 이하 청크 제거
    postprocessor = SimilarityPostprocessor(
        similarity_cutoff = min_score
    )

    # 시스템 프롬프트 (한국어 + 출처 강제)
    from llama_index.core import PromptTemplate

    QA_PROMPT = PromptTemplate(
        """당신은 사내 문서를 기반으로 정확하게 답변하는 AI 어시스턴트입니다.

규칙:
1. 반드시 아래 문서 내용만 참고하여 답변하세요.
2. 문서에 없는 내용은 "해당 내용은 문서에서 찾을 수 없습니다."라고 답하세요.
3. 답변 마지막에 반드시 출처를 표시하세요: [출처: 파일명, N페이지]

참고 문서:
---------------------
{context_str}
---------------------

질문: {query_str}

답변:"""
    )

    query_engine = RetrieverQueryEngine.from_args(
        retriever         = retriever,
        llm               = llm,
        node_postprocessors = [postprocessor],
        text_qa_template  = QA_PROMPT,
        streaming         = True,   # 스트리밍 출력 활성화
    )

    return query_engine

5-3. LLM에 컨텍스트 + 질문 전달

pythondef format_context(source_nodes: list) -> str:
    """
    검색된 청크 → LLM에 전달할 컨텍스트 문자열 구성
    """
    context_parts = []
    for i, node in enumerate(source_nodes):
        meta     = node.metadata
        source   = meta.get("source", "알 수 없음")
        page     = meta.get("page", "?")
        score    = node.score if hasattr(node, "score") else 0

        context_parts.append(
            f"[문서 {i+1}] {source} / {page}페이지 (유사도: {score:.2f})\n"
            f"{node.text}"
        )

    return "\n\n".join(context_parts)

5-4. 답변 생성 & 출처 표시

pythondef ask(query_engine, question: str,
        streaming: bool = True) -> dict:
    """
    질문 입력 → 답변 + 출처 반환
    """
    print(f"\n❓ 질문: {question}")
    print("─" * 50)

    if streaming:
        # 스트리밍 모드: 답변이 실시간으로 출력됨
        response      = query_engine.query(question)
        response_text = ""

        print("🤖 답변: ", end="", flush=True)
        for token in response.response_gen:
            print(token, end="", flush=True)
            response_text += token
        print()

    else:
        response      = query_engine.query(question)
        response_text = str(response)
        print(f"🤖 답변: {response_text}")

    # 출처 정리
    sources = []
    if hasattr(response, "source_nodes"):
        for node in response.source_nodes:
            meta = node.metadata
            sources.append({
                "source":  meta.get("source", "알 수 없음"),
                "page":    meta.get("page", "?"),
                "score":   round(node.score, 3) if hasattr(node, "score") else 0,
                "preview": meta.get("preview", ""),
            })

    # 출처 출력
    if sources:
        print("\n📚 참고 문서:")
        for s in sources:
            print(f"  └─ {s['source']} / {s['page']}페이지"
                  f" (유사도: {s['score']})")

    return {
        "answer":  response_text,
        "sources": sources,
    }

# ─── 전체 RAG 파이프라인 실행 ─────────────────────────────

def run_rag_pipeline(docs_folder: str, question: str):
    from indexing.embedder import load_bge_m3_llamaindex
    from indexing.vector_store import load_chroma_index

    embed_model  = load_bge_m3_llamaindex()
    index        = load_chroma_index(embed_model)
    llm          = setup_llm("qwen2.5:7b")
    query_engine = build_query_engine(index, llm, top_k=5)
    result       = ask(query_engine, question)
    return result

# 테스트
if __name__ == "__main__":
    result = run_rag_pipeline(
        docs_folder = "./company_docs",
        question    = "연차 신청은 며칠 전에 해야 하나요?",
    )

📌 6. 대용량 문서 처리 — Qdrant 활용

6-1. ChromaDB vs Qdrant 비교

6-2. Qdrant 로컬 셀프호스팅 설정

bash# Docker로 Qdrant 실행 (권장)
docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v $(pwd)/qdrant_data:/qdrant/storage \
  qdrant/qdrant:latest

# 실행 확인
curl http://localhost:6333/health
# → {"title":"qdrant - vector search engine","version":"..."}

python# indexing/vector_store_qdrant.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams,
    PointStruct, Filter, FieldCondition, MatchValue
)
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.core import VectorStoreIndex, StorageContext
import uuid

def build_qdrant_index(nodes, embed_model,
                        collection_name="company_docs",
                        qdrant_url="http://localhost:6333"):
    """Qdrant 인덱스 구성"""
    client = QdrantClient(url=qdrant_url)

    # 컬렉션 생성 (bge-m3: 1024차원)
    try:
        client.create_collection(
            collection_name = collection_name,
            vectors_config  = VectorParams(
                size     = 1024,           # bge-m3 차원
                distance = Distance.COSINE
            ),
        )
        print(f"✅ Qdrant 컬렉션 생성: {collection_name}")
    except Exception:
        print(f"⚠️  컬렉션 이미 존재: {collection_name}")

    # LlamaIndex 연동
    vector_store    = QdrantVectorStore(
        client          = client,
        collection_name = collection_name,
    )
    storage_context = StorageContext.from_defaults(
        vector_store=vector_store
    )

    index = VectorStoreIndex(
        nodes,
        storage_context = storage_context,
        embed_model     = embed_model,
        show_progress   = True,
    )

    info = client.get_collection(collection_name)
    print(f"✅ Qdrant 저장 완료: {info.points_count}개 벡터")
    return index

def qdrant_filtered_search(client, collection_name: str,
                            query_vector: list,
                            source_filter: str = None,
                            top_k: int = 5) -> list:
    """
    메타데이터 필터링 기반 검색
    source_filter: 특정 파일만 검색 (예: "인사규정.pdf")
    """
    search_filter = None
    if source_filter:
        search_filter = Filter(
            must=[FieldCondition(
                key   = "source",
                match = MatchValue(value=source_filter)
            )]
        )

    results = client.search(
        collection_name = collection_name,
        query_vector    = query_vector,
        query_filter    = search_filter,
        limit           = top_k,
        with_payload    = True,
    )
    return results

📌 7. 챗봇 UI 구성

7-1. Gradio로 5분 만에 웹 UI 만들기

python# ui/gradio_app.py
import gradio as gr
from indexing.embedder     import load_bge_m3_llamaindex
from indexing.vector_store import load_chroma_index
from retrieval.query_engine import setup_llm, build_query_engine, ask

# 전역 초기화 (앱 시작 시 1회)
print("🚀 모델 로드 중...")
embed_model  = load_bge_m3_llamaindex()
index        = load_chroma_index(embed_model)
llm          = setup_llm("qwen2.5:7b")
query_engine = build_query_engine(index, llm, top_k=5)
print("✅ RAG 챗봇 준비 완료")

def chat(message: str, history: list):
    """
    Gradio 채팅 함수
    message: 현재 사용자 입력
    history: [(user, bot), ...] 이전 대화 기록
    """
    if not message.strip():
        return "", history

    # RAG 답변 생성
    result = ask(query_engine, message, streaming=False)

    # 출처 포맷팅
    sources_text = ""
    if result["sources"]:
        sources_text = "\n\n📚 **참고 문서:**\n"
        for s in result["sources"]:
            sources_text += (
                f"- {s['source']} / {s['page']}페이지"
                f" (유사도: {s['score']})\n"
            )

    bot_response = result["answer"] + sources_text
    history.append((message, bot_response))
    return "", history

def add_documents(files):
    """문서 추가 업로드 기능"""
    if not files:
        return "파일을 선택해주세요."

    import shutil, os
    os.makedirs("./company_docs", exist_ok=True)

    for file in files:
        dest = os.path.join("./company_docs", os.path.basename(file.name))
        shutil.copy(file.name, dest)

    return f"✅ {len(files)}개 파일 업로드 완료. 재인덱싱 후 반영됩니다."

# ─── Gradio UI 구성 ───────────────────────────────────────

with gr.Blocks(title="사내 지식 RAG 챗봇",
               theme=gr.themes.Soft()) as demo:

    gr.Markdown("# 🤖 사내 지식 RAG 챗봇")
    gr.Markdown("사내 문서를 기반으로 정확한 답변을 제공합니다.")

    with gr.Row():
        with gr.Column(scale=3):
            chatbot = gr.Chatbot(
                label  = "대화",
                height = 500,
                bubble_full_width = False,
            )
            with gr.Row():
                msg = gr.Textbox(
                    label       = "질문 입력",
                    placeholder = "예: 연차 신청 절차가 어떻게 되나요?",
                    scale       = 9,
                )
                send_btn = gr.Button("전송", scale=1, variant="primary")

            gr.Examples(
                examples=[
                    "연차 신청은 며칠 전에 해야 하나요?",
                    "재택근무 신청 방법을 알려주세요.",
                    "복리후생 항목에는 무엇이 있나요?",
                ],
                inputs=msg,
            )

        with gr.Column(scale=1):
            gr.Markdown("### 📄 문서 추가")
            file_upload = gr.File(
                label        = "PDF / DOCX / TXT 업로드",
                file_count   = "multiple",
                file_types   = [".pdf", ".docx", ".txt"],
            )
            upload_btn    = gr.Button("업로드", variant="secondary")
            upload_status = gr.Textbox(label="업로드 상태", interactive=False)

    # 이벤트 연결
    send_btn.click(chat, [msg, chatbot], [msg, chatbot])
    msg.submit(chat, [msg, chatbot], [msg, chatbot])
    upload_btn.click(add_documents, [file_upload], [upload_status])

if __name__ == "__main__":
    demo.launch(
        server_name = "0.0.0.0",
        server_port = 7860,
        share       = False,     # True: 외부 공개 URL 생성
    )

7-2. Streamlit으로 대화 히스토리 구현

python# ui/streamlit_app.py
import streamlit as st

st.set_page_config(
    page_title = "사내 RAG 챗봇",
    page_icon  = "🤖",
    layout     = "wide",
)

# ─── 모델 초기화 (캐시로 재로드 방지) ────────────────────

@st.cache_resource
def init_rag():
    from indexing.embedder     import load_bge_m3_llamaindex
    from indexing.vector_store import load_chroma_index
    from retrieval.query_engine import setup_llm, build_query_engine

    embed_model  = load_bge_m3_llamaindex()
    index        = load_chroma_index(embed_model)
    llm          = setup_llm("qwen2.5:7b")
    query_engine = build_query_engine(index, llm, top_k=5)
    return query_engine

query_engine = init_rag()

# ─── 대화 히스토리 상태 관리 ─────────────────────────────

if "messages" not in st.session_state:
    st.session_state.messages = []

if "sources_history" not in st.session_state:
    st.session_state.sources_history = []

# ─── UI ──────────────────────────────────────────────────

st.title("🤖 사내 지식 RAG 챗봇")
st.caption("사내 문서를 기반으로 정확한 답변을 제공합니다.")

with st.sidebar:
    st.header("⚙️ 설정")
    top_k = st.slider("검색 청크 수 (Top-K)", 1, 10, 5)
    model = st.selectbox(
        "LLM 모델",
        ["qwen2.5:7b", "qwen2.5:14b", "llama3.2:3b"]
    )

    st.header("📄 문서 업로드")
    uploaded = st.file_uploader(
        "PDF / DOCX / TXT",
        accept_multiple_files = True,
        type                  = ["pdf", "docx", "txt"],
    )
    if uploaded and st.button("인덱싱 시작"):
        with st.spinner("문서 인덱싱 중..."):
            import os, shutil
            os.makedirs("./company_docs", exist_ok=True)
            for f in uploaded:
                with open(f"./company_docs/{f.name}", "wb") as fp:
                    fp.write(f.read())
            st.success(f"✅ {len(uploaded)}개 문서 저장 완료")
            st.info("앱 재시작 후 새 문서가 반영됩니다.")

    if st.button("대화 초기화"):
        st.session_state.messages = []
        st.session_state.sources_history = []
        st.rerun()

# ─── 대화 기록 출력 ──────────────────────────────────────

for i, msg in enumerate(st.session_state.messages):
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

        # 출처 접기 표시
        if msg["role"] == "assistant" and i < len(st.session_state.sources_history):
            sources = st.session_state.sources_history[i]
            if sources:
                with st.expander("📚 참고 문서 보기"):
                    for s in sources:
                        st.markdown(
                            f"- **{s['source']}** / {s['page']}페이지 "
                            f"`유사도: {s['score']}`\n"
                            f"  > {s['preview']}"
                        )

# ─── 질문 입력 ────────────────────────────────────────────

if question := st.chat_input("질문을 입력하세요..."):
    # 사용자 메시지 추가
    st.session_state.messages.append(
        {"role": "user", "content": question}
    )
    with st.chat_message("user"):
        st.markdown(question)

    # RAG 답변 생성
    with st.chat_message("assistant"):
        with st.spinner("문서 검색 중..."):
            from retrieval.query_engine import ask
            result = ask(query_engine, question, streaming=False)

        st.markdown(result["answer"])

        # 출처 표시
        if result["sources"]:
            with st.expander("📚 참고 문서 보기"):
                for s in result["sources"]:
                    st.markdown(
                        f"- **{s['source']}** / {s['page']}페이지 "
                        f"`유사도: {s['score']}`\n"
                        f"  > {s['preview']}"
                    )

    # 히스토리 저장
    st.session_state.messages.append(
        {"role": "assistant", "content": result["answer"]}
    )
    st.session_state.sources_history.append(result["sources"])

# 실행: streamlit run ui/streamlit_app.py

📌 8. 결과 확인 & 트러블슈팅

관련 없는 청크가 검색될 때

python# 원인 1: min_score 임계값이 너무 낮음
# → similarity_cutoff 높이기

postprocessor = SimilarityPostprocessor(
    similarity_cutoff = 0.6   # 0.4 → 0.6으로 올리기
)

# 원인 2: 청크 크기가 너무 크거나 작음
# 청크 너무 클 때 (1000토큰):
#   → 여러 주제가 섞여 검색 정확도 하락
#   → chunk_size=512 이하로 줄이기

# 청크 너무 작을 때 (50토큰):
#   → 문맥이 잘려 답변 품질 하락
#   → chunk_size=256 이상으로 늘리기

# 원인 3: 임베딩 모델이 한국어에 최적화 안 됨
# → multilingual-e5-large → bge-m3으로 교체

# 검색 결과 디버깅
def debug_retrieval(index, embed_model, question, top_k=10):
    """검색된 청크 상세 출력"""
    retriever = VectorIndexRetriever(
        index=index, similarity_top_k=top_k
    )
    nodes = retriever.retrieve(question)

    print(f"\n🔍 '{question}' 검색 결과 (Top {top_k}):")
    for i, node in enumerate(nodes):
        print(f"\n[{i+1}] 유사도: {node.score:.4f}")
        print(f"     출처: {node.metadata.get('source')} / "
              f"{node.metadata.get('page')}페이지")
        print(f"     내용: {node.text[:150]}...")

답변이 문서와 다를 때 (Hallucination)

python# 원인 1: top_k가 너무 낮아 관련 청크 누락
# → top_k=3 → top_k=7로 늘리기

# 원인 2: LLM temperature가 높음
# → temperature=0.7 → 0.1로 낮추기
llm = setup_llm("qwen2.5:7b", temperature=0.1)

# 원인 3: 시스템 프롬프트가 약함 → 더 강하게 명시
STRICT_QA_PROMPT = PromptTemplate("""
당신은 오직 제공된 문서 내용만을 기반으로 답변하는 AI입니다.

⚠️ 엄격한 규칙:
- 문서에 없는 내용은 절대 추측하거나 생성하지 마세요.
- "~인 것 같습니다", "~일 수도 있습니다" 표현을 사용하지 마세요.
- 문서에 정보가 없으면: "해당 내용은 제공된 문서에서 찾을 수 없습니다."
- 모든 답변 끝에 [출처: 파일명, N페이지] 형식으로 출처를 표시하세요.

참고 문서:
---------------------
{context_str}
---------------------

질문: {query_str}
답변:""")

검색 정확도 높이는 법 (Re-ranking)

python# Re-ranking: 1차 검색 결과를 더 정밀한 모델로 재정렬
# → Top-K=20 으로 넓게 검색 → Re-ranker로 Top-3으로 압축

from llama_index.core.postprocessor import SentenceTransformerRerank

def build_query_engine_with_reranker(index, llm,
                                      top_k=20, rerank_top=5):
    """
    Re-ranking 적용 쿼리 엔진
    top_k:      1차 검색 수 (넓게)
    rerank_top: 최종 사용할 청크 수 (좁게)
    """
    retriever = VectorIndexRetriever(
        index            = index,
        similarity_top_k = top_k,    # 넓게 검색
    )

    # Cross-Encoder 기반 Re-ranker
    reranker = SentenceTransformerRerank(
        model   = "cross-encoder/ms-marco-MiniLM-L-2-v2",
        top_n   = rerank_top,         # 최종 선택
    )

    query_engine = RetrieverQueryEngine.from_args(
        retriever           = retriever,
        llm                 = llm,
        node_postprocessors = [reranker],
        text_qa_template    = STRICT_QA_PROMPT,
    )

    print(f"✅ Re-ranking 엔진: Top-{top_k} → Re-rank → Top-{rerank_top}")
    return query_engine

# 하이브리드 검색 (벡터 + BM25 키워드 검색 혼합)
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.core.retrievers import QueryFusionRetriever

def build_hybrid_query_engine(index, nodes, llm, top_k=5):
    """
    벡터 검색 + BM25 키워드 검색 혼합
    → 키워드 일치 + 의미 유사도 동시 고려
    """
    vector_retriever = VectorIndexRetriever(
        index=index, similarity_top_k=top_k
    )
    bm25_retriever   = BM25Retriever.from_defaults(
        nodes=nodes, similarity_top_k=top_k
    )

    # 두 검색 결과 Reciprocal Rank Fusion으로 합산
    hybrid_retriever = QueryFusionRetriever(
        retrievers      = [vector_retriever, bm25_retriever],
        similarity_top_k = top_k,
        num_queries      = 1,     # 쿼리 재생성 없음
        mode             = "reciprocal_rerank",
    )

    query_engine = RetrieverQueryEngine.from_args(
        retriever        = hybrid_retriever,
        llm              = llm,
        text_qa_template = STRICT_QA_PROMPT,
    )

    print("✅ 하이브리드 검색 엔진 (벡터 + BM25) 준비 완료")
    return query_engine

✅ 완성 체크리스트

• PDF / DOCX 로드 및 텍스트 추출 확인
• 청크 분할 결과 확인 (청크 수, 평균 길이)
• ChromaDB 인덱스 생성 및 벡터 수 확인
• 기본 질의응답 동작 확인 (출처 표시 포함)
• Gradio 또는 Streamlit UI 실행 확인
• 대화 히스토리 유지 확인
• Re-ranking 적용 후 검색 정확도 향상 확인