RAG 기반 FAQ 챗봇 프로덕션 구축 가이드: 벡터 DB 선정부터 운영 최적화까지

FAQ 문서 수집 -> 전처리/정규화 -> 청킹 -> 임베딩 생성 -> 벡터 DB 저장 -> 메타데이터 인덱싱

사용자 질문 -> 쿼리 전처리 -> 임베딩 변환 -> 벡터 검색 + BM25 -> 리랭킹 -> 프롬프트 구성 -> LLM 답변 생성 -> 후처리/가드레일

"""
FAQ 전용 청킹 전략.
질문-답변 쌍을 하나의 단위로 유지하면서,
Parent-Child 구조로 검색 정밀도와 컨텍스트를 동시에 확보한다.
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List, Tuple
import re
import hashlib


def parse_faq_pairs(raw_text: str) -> List[Tuple[str, str, dict]]:
    """FAQ 원문에서 질문-답변 쌍을 추출한다."""
    faq_pattern = re.compile(
        r"(?:Q|질문)\s*[.:]\s*(.+?)\n+"
        r"(?:A|답변)\s*[.:]\s*(.+?)(?=\n(?:Q|질문)\s*[.:]|\Z)",
        re.DOTALL
    )
    pairs = []
    for i, match in enumerate(faq_pattern.finditer(raw_text)):
        question = match.group(1).strip()
        answer = match.group(2).strip()
        metadata = {
            "faq_id": hashlib.md5(question.encode()).hexdigest()[:8],
            "source_type": "faq",
            "question": question,
            "pair_index": i,
        }
        pairs.append((question, answer, metadata))
    return pairs


def create_faq_chunks(
    faq_pairs: List[Tuple[str, str, dict]],
    child_chunk_size: int = 200,
    child_chunk_overlap: int = 50,
) -> Tuple[List[Document], List[Document]]:
    """
    Parent-Child 청킹 전략으로 FAQ 문서를 분할한다.
    - Parent: 질문 + 전체 답변 (LLM 컨텍스트용)
    - Child: 답변을 작은 청크로 분할 (검색 정밀도용)
    """
    parent_docs = []
    child_docs = []

    child_splitter = RecursiveCharacterTextSplitter(
        chunk_size=child_chunk_size,
        chunk_overlap=child_chunk_overlap,
        separators=["\n\n", "\n", ". ", " "],
    )

    for question, answer, metadata in faq_pairs:
        # Parent 문서: 질문 + 전체 답변
        parent_content = f"질문: {question}\n답변: {answer}"
        parent_id = metadata["faq_id"]
        parent_doc = Document(
            page_content=parent_content,
            metadata={**metadata, "doc_type": "parent", "parent_id": parent_id},
        )
        parent_docs.append(parent_doc)

        # Child 문서: 답변을 세분화하여 검색 정밀도 향상
        answer_chunks = child_splitter.split_text(answer)
        for j, chunk in enumerate(answer_chunks):
            child_content = f"질문: {question}\n답변 일부: {chunk}"
            child_doc = Document(
                page_content=child_content,
                metadata={
                    **metadata,
                    "doc_type": "child",
                    "parent_id": parent_id,
                    "chunk_index": j,
                },
            )
            child_docs.append(child_doc)

    return parent_docs, child_docs


# 사용 예시
raw_faq = """
Q: 환불 처리는 며칠이 걸리나요?
A: 환불은 요청일로부터 영업일 기준 3-5일 이내에 처리됩니다.
카드 결제의 경우 카드사 처리 기간이 추가로 2-3일 소요될 수 있습니다.
무통장 입금의 경우 등록된 계좌로 직접 환불됩니다.

Q: 해외 배송이 가능한가요?
A: 현재 미국, 일본, 중국, 동남아시아 지역으로 해외 배송이 가능합니다.
해외 배송비는 지역과 무게에 따라 달라지며, 관세는 수령인 부담입니다.
배송 소요 기간은 지역에 따라 7-14 영업일입니다.
"""

pairs = parse_faq_pairs(raw_faq)
parents, children = create_faq_chunks(pairs)
print(f"Parent 문서: {len(parents)}개, Child 문서: {len(children)}개")

"""
Pinecone 벡터 DB 설정과 FAQ 문서 인덱싱.
네임스페이스로 문서 유형을 분리하고 메타데이터 필터링을 활용한다.
"""
from pinecone import Pinecone, ServerlessSpec
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from typing import List
import os
import time


def setup_pinecone_index(
    index_name: str = "faq-chatbot",
    dimension: int = 1536,
    metric: str = "cosine",
) -> None:
    """Pinecone 인덱스를 생성한다. 이미 존재하면 건너뛴다."""
    pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

    existing_indexes = [idx.name for idx in pc.list_indexes()]
    if index_name not in existing_indexes:
        pc.create_index(
            name=index_name,
            dimension=dimension,
            metric=metric,
            spec=ServerlessSpec(cloud="aws", region="us-east-1"),
        )
        # 인덱스가 준비될 때까지 대기
        while not pc.describe_index(index_name).status["ready"]:
            time.sleep(1)
        print(f"인덱스 '{index_name}' 생성 완료")
    else:
        print(f"인덱스 '{index_name}' 이미 존재")


def index_faq_documents(
    parent_docs: List[Document],
    child_docs: List[Document],
    index_name: str = "faq-chatbot",
) -> PineconeVectorStore:
    """
    Parent-Child 구조의 FAQ 문서를 Pinecone에 인덱싱한다.
    - Child 문서: 'search' 네임스페이스 (검색용)
    - Parent 문서: 'context' 네임스페이스 (LLM 컨텍스트용)
    """
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # Child 문서 인덱싱 (검색 대상)
    child_store = PineconeVectorStore.from_documents(
        documents=child_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="search",
    )
    print(f"Child 문서 {len(child_docs)}개 인덱싱 완료 (namespace: search)")

    # Parent 문서 인덱싱 (컨텍스트 제공용)
    parent_store = PineconeVectorStore.from_documents(
        documents=parent_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="context",
    )
    print(f"Parent 문서 {len(parent_docs)}개 인덱싱 완료 (namespace: context)")

    return child_store


# 실행
setup_pinecone_index()
child_vectorstore = index_faq_documents(parents, children)

"""
LangChain 기반 FAQ 챗봇 구현.
Parent-Child 검색 + 커스텀 프롬프트 + 대화 이력 관리를 통합한다.
"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document
from typing import List, Dict
import os


# 1. 컴포넌트 초기화
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

child_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="search",
)
parent_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="context",
)


# 2. Parent-Child 검색기 구현
def retrieve_with_parent_lookup(query: str, k: int = 3) -> List[Document]:
    """
    Child 청크로 검색한 뒤, 매칭된 Parent 문서를 반환한다.
    이렇게 하면 검색 정밀도는 Child 수준, 컨텍스트는 Parent 수준으로 확보된다.
    """
    # Step 1: Child 청크에서 유사도 검색
    child_results = child_store.similarity_search(query, k=k * 2)

    # Step 2: 중복 제거하여 고유 parent_id 추출
    seen_parent_ids = set()
    unique_parent_ids = []
    for doc in child_results:
        pid = doc.metadata.get("parent_id")
        if pid and pid not in seen_parent_ids:
            seen_parent_ids.add(pid)
            unique_parent_ids.append(pid)
        if len(unique_parent_ids) >= k:
            break

    # Step 3: Parent 문서 검색
    parent_results = parent_store.similarity_search(
        query,
        k=k,
        filter={"parent_id": {"$in": unique_parent_ids}},
    )

    return parent_results


# 3. 프롬프트 설계
FAQ_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """당신은 고객 FAQ를 전문으로 답변하는 AI 어시스턴트입니다.
아래 규칙을 반드시 준수하세요:

1. 제공된 FAQ 문서에 기반하여만 답변하세요.
2. FAQ 문서에 답변이 없으면 "해당 질문에 대한 답변을 찾지 못했습니다. 고객센터(1234-5678)로 문의해 주세요."라고 안내하세요.
3. 추측하거나 FAQ에 없는 정보를 생성하지 마세요.
4. 답변 시 관련 FAQ 문서의 출처를 함께 안내하세요.
5. 친절하고 간결하게 답변하세요.

참고할 FAQ 문서:
{context}"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])


# 4. 체인 구성
def format_docs(docs: List[Document]) -> str:
    """검색된 문서를 프롬프트에 포함할 형식으로 변환한다."""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source_info = doc.metadata.get("faq_id", "unknown")
        formatted.append(
            f"[FAQ-{source_info}]\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)


faq_chain = (
    {
        "context": RunnableLambda(
            lambda x: format_docs(retrieve_with_parent_lookup(x["question"]))
        ),
        "question": RunnablePassthrough() | RunnableLambda(lambda x: x["question"]),
        "chat_history": RunnableLambda(lambda x: x.get("chat_history", [])),
    }
    | FAQ_PROMPT
    | llm
    | StrOutputParser()
)


# 5. 실행
response = faq_chain.invoke({
    "question": "환불 처리 기간이 얼마나 걸리나요?",
    "chat_history": [],
})
print(response)

"""
BM25 + Dense 하이브리드 검색 구현.
Reciprocal Rank Fusion(RRF)으로 두 검색 결과를 결합한다.
"""
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from typing import List, Dict, Tuple
import numpy as np


class HybridRetriever:
    """BM25와 벡터 검색을 결합하는 하이브리드 검색기."""

    def __init__(
        self,
        vector_store,
        documents: List[Document],
        bm25_k: int = 10,
        vector_k: int = 10,
        rrf_k: int = 60,
        alpha: float = 0.5,
    ):
        self.vector_store = vector_store
        self.bm25_retriever = BM25Retriever.from_documents(
            documents, k=bm25_k
        )
        self.vector_k = vector_k
        self.rrf_k = rrf_k
        self.alpha = alpha  # 0=BM25 only, 1=Dense only

    def _reciprocal_rank_fusion(
        self,
        bm25_results: List[Document],
        vector_results: List[Document],
    ) -> List[Tuple[Document, float]]:
        """RRF 알고리즘으로 두 검색 결과를 결합한다."""
        doc_scores: Dict[str, Tuple[Document, float]] = {}

        # BM25 결과에 RRF 점수 부여
        for rank, doc in enumerate(bm25_results):
            doc_key = doc.page_content[:100]
            score = (1 - self.alpha) / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # Dense 결과에 RRF 점수 부여
        for rank, doc in enumerate(vector_results):
            doc_key = doc.page_content[:100]
            score = self.alpha / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # RRF 점수 기준 내림차순 정렬
        sorted_results = sorted(
            doc_scores.values(), key=lambda x: x[1], reverse=True
        )
        return sorted_results

    def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
        """하이브리드 검색을 수행한다."""
        # 병렬 검색 (프로덕션에서는 asyncio 사용 권장)
        bm25_results = self.bm25_retriever.invoke(query)
        vector_results = self.vector_store.similarity_search(
            query, k=self.vector_k
        )

        # RRF로 결합
        fused = self._reciprocal_rank_fusion(bm25_results, vector_results)

        return [doc for doc, score in fused[:top_k]]


# 사용 예시
hybrid_retriever = HybridRetriever(
    vector_store=child_store,
    documents=children,  # BM25용 원본 문서
    alpha=0.6,  # Dense 가중치 60%
)
results = hybrid_retriever.retrieve("오류 코드 P4021 해결 방법")

                    +------------------+
                    |   Load Balancer  |
                    +--------+---------+
                             |
              +--------------+--------------+
              |                             |
    +---------v---------+    +---------v---------+
    |  API Server (1)   |    |  API Server (2)   |
    |  FastAPI + Uvicorn|    |  FastAPI + Uvicorn|
    +---------+---------+    +---------+---------+
              |                        |
    +---------v------------------------v---------+
    |              Redis Cache                    |
    |   (쿼리 임베딩 캐싱, 답변 캐싱)            |
    +-----+-------------+-------------+----------+
          |             |             |
+---------v---+ +-------v-----+ +----v----------+
| Pinecone    | | BM25 Index  | | LLM API       |
| (Dense)     | | (Sparse)    | | (OpenAI/Azure)|
+-------------+ +-------------+ +---------------+

"""
RAGAS 기반 FAQ 챗봇 품질 평가.
Golden Dataset에 대해 자동으로 평가를 수행하고 지표를 산출한다.
"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from datasets import Dataset
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from typing import List, Dict
import json
from datetime import datetime


def prepare_evaluation_dataset(
    test_cases: List[Dict],
    retriever,
    chain,
) -> Dataset:
    """
    테스트 케이스를 RAGAS 평가 형식으로 변환한다.
    각 질문에 대해 실제 검색과 답변 생성을 수행한다.
    """
    eval_data = {
        "question": [],
        "answer": [],
        "contexts": [],
        "ground_truth": [],
    }

    for case in test_cases:
        question = case["question"]

        # 실제 검색 수행
        retrieved_docs = retriever.retrieve(question, top_k=5)
        contexts = [doc.page_content for doc in retrieved_docs]

        # 실제 답변 생성
        answer = chain.invoke({
            "question": question,
            "chat_history": [],
        })

        eval_data["question"].append(question)
        eval_data["answer"].append(answer)
        eval_data["contexts"].append(contexts)
        eval_data["ground_truth"].append(case["expected_answer"])

    return Dataset.from_dict(eval_data)


def run_ragas_evaluation(dataset: Dataset) -> Dict:
    """RAGAS 평가를 실행하고 결과를 반환한다."""
    eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o", temperature=0))
    eval_embeddings = LangchainEmbeddingsWrapper(
        OpenAIEmbeddings(model="text-embedding-3-small")
    )

    result = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
        ],
        llm=eval_llm,
        embeddings=eval_embeddings,
    )

    # 결과 저장
    report = {
        "timestamp": datetime.now().isoformat(),
        "dataset_size": len(dataset),
        "metrics": {
            "faithfulness": float(result["faithfulness"]),
            "answer_relevancy": float(result["answer_relevancy"]),
            "context_precision": float(result["context_precision"]),
            "context_recall": float(result["context_recall"]),
        },
    }

    # 배포 게이트: 모든 지표가 임계값 이상이어야 통과
    thresholds = {
        "faithfulness": 0.9,
        "answer_relevancy": 0.85,
        "context_precision": 0.8,
        "context_recall": 0.9,
    }
    report["deployment_gate"] = all(
        report["metrics"][k] >= v for k, v in thresholds.items()
    )

    with open(f"eval_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    return report


# 실행 예시
test_cases = [
    {
        "question": "환불 처리는 며칠이 걸리나요?",
        "expected_answer": "환불은 요청일로부터 영업일 기준 3-5일 이내에 처리됩니다.",
    },
    {
        "question": "해외 배송비는 얼마인가요?",
        "expected_answer": "해외 배송비는 지역과 무게에 따라 달라지며, 관세는 수령인 부담입니다.",
    },
]

# eval_dataset = prepare_evaluation_dataset(test_cases, hybrid_retriever, faq_chain)
# report = run_ragas_evaluation(eval_dataset)
# print(f"배포 게이트 통과: {report['deployment_gate']}")

"""
FAQ 챗봇 운영 모니터링.
요청별 지표 수집, 이상 탐지, 알림 발송을 담당한다.
"""
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from prometheus_client import (
    Counter,
    Histogram,
    Gauge,
    start_http_server,
)

logger = logging.getLogger(__name__)

# Prometheus 메트릭 정의
REQUEST_COUNT = Counter(
    "faq_chatbot_requests_total",
    "Total FAQ chatbot requests",
    ["status", "category"],
)
RESPONSE_LATENCY = Histogram(
    "faq_chatbot_response_seconds",
    "Response latency in seconds",
    buckets=[0.5, 1.0, 2.0, 3.0, 5.0, 10.0],
)
RETRIEVAL_LATENCY = Histogram(
    "faq_chatbot_retrieval_seconds",
    "Retrieval latency in seconds",
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0],
)
LLM_TOKENS_USED = Counter(
    "faq_chatbot_llm_tokens_total",
    "Total LLM tokens consumed",
    ["type"],  # prompt, completion
)
FALLBACK_RATE = Gauge(
    "faq_chatbot_fallback_rate",
    "Current fallback (no answer) rate",
)
ACTIVE_REQUESTS = Gauge(
    "faq_chatbot_active_requests",
    "Currently processing requests",
)


@dataclass
class RequestMetrics:
    """단일 요청의 메트릭을 수집하는 컨텍스트 매니저."""
    question: str
    start_time: float = field(default_factory=time.time)
    retrieval_time: Optional[float] = None
    llm_time: Optional[float] = None
    total_time: Optional[float] = None
    status: str = "success"
    is_fallback: bool = False
    prompt_tokens: int = 0
    completion_tokens: int = 0

    def record_retrieval(self):
        self.retrieval_time = time.time() - self.start_time

    def record_llm_start(self):
        self._llm_start = time.time()

    def record_llm_end(self, prompt_tokens: int, completion_tokens: int):
        self.llm_time = time.time() - self._llm_start
        self.prompt_tokens = prompt_tokens
        self.completion_tokens = completion_tokens

    def finalize(self):
        self.total_time = time.time() - self.start_time

        # Prometheus 메트릭 기록
        REQUEST_COUNT.labels(
            status=self.status, category="faq"
        ).inc()
        RESPONSE_LATENCY.observe(self.total_time)

        if self.retrieval_time:
            RETRIEVAL_LATENCY.observe(self.retrieval_time)

        LLM_TOKENS_USED.labels(type="prompt").inc(self.prompt_tokens)
        LLM_TOKENS_USED.labels(type="completion").inc(
            self.completion_tokens
        )

        # 로그 기록 (구조화된 로깅)
        logger.info(
            "faq_request_completed",
            extra={
                "question_preview": self.question[:50],
                "total_time_ms": round(self.total_time * 1000),
                "retrieval_time_ms": round(
                    (self.retrieval_time or 0) * 1000
                ),
                "status": self.status,
                "is_fallback": self.is_fallback,
                "tokens": self.prompt_tokens + self.completion_tokens,
            },
        )


# Prometheus 메트릭 서버 시작
# start_http_server(8001)  # /metrics 엔드포인트 노출