Customer Support Chatbot 구축 - 완벽 핸즈온 가이드 #2

🎯 학습 목표

• RAG (Retrieval-Augmented Generation) 아키텍처 이해
• Vector Database를 활용한 의미 검색
• Fine-tuning vs Prompting 비교
• LoRA/PEFT를 활용한 효율적인 모델 커스터마이징
• 실전 챗봇 배포

📋 사전 준비

1. 개발 환경 설정

# 새 프로젝트 디렉토리
mkdir customer-support-chatbot
cd customer-support-chatbot

# 가상환경 생성
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 필수 패키지 설치
pip install openai anthropic streamlit python-dotenv
pip install chromadb sentence-transformers
pip install langchain langchain-openai langchain-community
pip install pandas numpy tiktoken
pip install datasets huggingface_hub  # Fine-tuning용

# Fine-tuning 고급 (선택)
pip install torch transformers peft accelerate bitsandbytes

2. 프로젝트 구조

mkdir -p src data/knowledge_base data/conversations utils
touch .env app.py requirements.txt

customer-support-chatbot/
├── src/
│   ├── rag_engine.py          # RAG 시스템 핵심
│   ├── vector_store.py        # Vector DB 관리
│   ├── embeddings.py          # 임베딩 생성
│   ├── chatbot.py             # 챗봇 로직
│   ├── finetuning.py          # Fine-tuning 코드
│   └── evaluator.py           # 성능 평가
├── data/
│   ├── knowledge_base/        # FAQ, 문서들
│   └── conversations/         # 대화 로그
├── utils/
│   └── data_loader.py         # 데이터 로딩 유틸
├── app.py                     # Streamlit UI
└── .env

---

🚀 Step 1: 지식 베이스 구축 (20분)

data/knowledge_base/faq.json

{
  "faqs": [
    {
      "id": 1,
      "category": "배송",
      "question": "배송은 얼마나 걸리나요?",
      "answer": "일반 배송은 주문 후 2-3일 소요되며, 제주도 및 도서산간 지역은 1-2일 추가됩니다. 새벽 배송은 당일 밤 12시 이전 주문시 다음날 오전 7시 전 도착합니다."
    },
    {
      "id": 2,
      "category": "배송",
      "question": "배송비는 얼마인가요?",
      "answer": "3만원 이상 구매시 무료배송이며, 3만원 미만 구매시 배송비 3,000원이 부과됩니다. 제주도 및 도서산간 지역은 추가 배송비 3,000원이 발생합니다."
    },
    {
      "id": 3,
      "category": "반품",
      "question": "반품은 어떻게 하나요?",
      "answer": "상품 수령 후 7일 이내 마이페이지에서 반품 신청 가능합니다. 단순 변심의 경우 왕복 배송비(6,000원)가 차감됩니다. 상품 하자의 경우 전액 환불됩니다."
    },
    {
      "id": 4,
      "category": "반품",
      "question": "반품 불가 상품이 있나요?",
      "answer": "신선식품, 개봉한 화장품, 속옷 등은 반품이 불가능합니다. 또한 주문 제작 상품의 경우에도 반품이 제한됩니다."
    },
    {
      "id": 5,
      "category": "결제",
      "question": "어떤 결제 방법을 사용할 수 있나요?",
      "answer": "신용카드, 체크카드, 계좌이체, 무통장입금, 네이버페이, 카카오페이를 지원합니다. 할부는 5만원 이상 구매시 가능합니다."
    },
    {
      "id": 6,
      "category": "결제",
      "question": "무통장입금 입금 기한은?",
      "answer": "주문 후 3일 이내 입금하지 않으면 자동으로 주문이 취소됩니다. 입금자명과 주문자명이 다를 경우 고객센터로 연락 부탁드립니다."
    },
    {
      "id": 7,
      "category": "회원",
      "question": "회원 등급은 어떻게 되나요?",
      "answer": "일반(0-30만원), 실버(30-100만원), 골드(100-300만원), VIP(300만원 이상) 등급이 있으며, 등급별로 최대 5% 할인 혜택이 제공됩니다."
    },
    {
      "id": 8,
      "category": "회원",
      "question": "적립금은 어떻게 사용하나요?",
      "answer": "구매 금액의 1-5%가 적립되며, 5,000원 이상부터 1원 단위로 사용 가능합니다. 적립금은 적립일로부터 1년간 유효합니다."
    },
    {
      "id": 9,
      "category": "교환",
      "question": "사이즈 교환은 어떻게 하나요?",
      "answer": "상품 수령 후 7일 이내 교환 신청 가능합니다. 왕복 배송비는 고객 부담이며, 동일 상품의 다른 사이즈로만 교환 가능합니다."
    },
    {
      "id": 10,
      "category": "쿠폰",
      "question": "쿠폰은 어디서 받나요?",
      "answer": "마이페이지 쿠폰함에서 다운로드 가능하며, 회원가입 시 10% 할인 쿠폰이 자동 지급됩니다. 중복 사용은 불가능합니다."
    }
  ]
}

utils/data_loader.py

import json
import os
from typing import List, Dict

class DataLoader:
    """지식 베이스 데이터 로딩"""
    
    @staticmethod
    def load_faqs(file_path: str = "data/knowledge_base/faq.json") -> List[Dict]:
        """FAQ 데이터 로드"""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"FAQ 파일을 찾을 수 없습니다: {file_path}")
        
        with open(file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        return data['faqs']
    
    @staticmethod
    def create_documents(faqs: List[Dict]) -> List[Dict]:
        """FAQ를 문서 형식으로 변환"""
        documents = []
        
        for faq in faqs:
            # 질문과 답변을 하나의 문서로
            doc = {
                'id': faq['id'],
                'text': f"질문: {faq['question']}\n답변: {faq['answer']}",
                'category': faq['category'],
                'metadata': {
                    'question': faq['question'],
                    'answer': faq['answer'],
                    'category': faq['category']
                }
            }
            documents.append(doc)
        
        return documents

# 테스트
if __name__ == "__main__":
    loader = DataLoader()
    faqs = loader.load_faqs()
    print(f"✅ {len(faqs)}개의 FAQ 로드 완료")
    
    docs = loader.create_documents(faqs)
    print(f"✅ {len(docs)}개의 문서 생성 완료")
    print("\n첫 번째 문서:")
    print(docs[0]['text'])

---

🔍 Step 2: Vector Store 구축 (25분)

src/embeddings.py

from sentence_transformers import SentenceTransformer
from typing import List
import numpy as np

class EmbeddingGenerator:
    """임베딩 생성기"""
    
    def __init__(self, model_name: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"):
        """
        한국어 지원 임베딩 모델 초기화
        - paraphrase-multilingual-MiniLM-L12-v2: 다국어 지원, 빠름
        - jhgan/ko-sroberta-multitask: 한국어 특화, 정확
        """
        print(f"임베딩 모델 로딩 중: {model_name}")
        self.model = SentenceTransformer(model_name)
        self.dimension = self.model.get_sentence_embedding_dimension()
        print(f"✅ 임베딩 차원: {self.dimension}")
    
    def embed_text(self, text: str) -> List[float]:
        """단일 텍스트 임베딩"""
        embedding = self.model.encode(text, convert_to_tensor=False)
        return embedding.tolist()
    
    def embed_texts(self, texts: List[str]) -> List[List[float]]:
        """여러 텍스트 임베딩 (배치 처리)"""
        embeddings = self.model.encode(texts, convert_to_tensor=False, show_progress_bar=True)
        return embeddings.tolist()
    
    def similarity(self, text1: str, text2: str) -> float:
        """두 텍스트 간 유사도 계산"""
        emb1 = self.model.encode(text1)
        emb2 = self.model.encode(text2)
        
        # 코사인 유사도
        similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
        return float(similarity)

# 테스트
if __name__ == "__main__":
    embedder = EmbeddingGenerator()
    
    # 유사한 문장
    text1 = "배송은 얼마나 걸리나요?"
    text2 = "배송 기간이 어떻게 되나요?"
    text3 = "반품 방법을 알려주세요"
    
    print(f"\n'{text1}' vs '{text2}' 유사도: {embedder.similarity(text1, text2):.4f}")
    print(f"'{text1}' vs '{text3}' 유사도: {embedder.similarity(text1, text3):.4f}")

src/vector_store.py

import chromadb
from chromadb.config import Settings
from typing import List, Dict, Optional
import uuid

class VectorStore:
    """ChromaDB를 사용한 벡터 저장소"""
    
    def __init__(self, collection_name: str = "customer_support", persist_directory: str = "./chroma_db"):
        """벡터 DB 초기화"""
        self.client = chromadb.Client(Settings(
            persist_directory=persist_directory,
            anonymized_telemetry=False
        ))
        
        # 컬렉션 생성 또는 가져오기
        try:
            self.collection = self.client.get_collection(name=collection_name)
            print(f"✅ 기존 컬렉션 로드: {collection_name}")
        except:
            self.collection = self.client.create_collection(
                name=collection_name,
                metadata={"description": "Customer support knowledge base"}
            )
            print(f"✅ 새 컬렉션 생성: {collection_name}")
    
    def add_documents(self, documents: List[Dict], embeddings: List[List[float]]):
        """문서와 임베딩을 벡터 DB에 저장"""
        ids = [str(doc['id']) for doc in documents]
        texts = [doc['text'] for doc in documents]
        metadatas = [doc['metadata'] for doc in documents]
        
        self.collection.add(
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas,
            ids=ids
        )
        
        print(f"✅ {len(documents)}개 문서 저장 완료")
    
    def search(self, query_embedding: List[float], top_k: int = 3) -> List[Dict]:
        """유사한 문서 검색"""
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        # 결과 포맷팅
        retrieved_docs = []
        for i in range(len(results['ids'][0])):
            doc = {
                'id': results['ids'][0][i],
                'text': results['documents'][0][i],
                'metadata': results['metadatas'][0][i],
                'distance': results['distances'][0][i] if 'distances' in results else None
            }
            retrieved_docs.append(doc)
        
        return retrieved_docs
    
    def search_by_text(self, query: str, embedder, top_k: int = 3) -> List[Dict]:
        """텍스트로 직접 검색"""
        query_embedding = embedder.embed_text(query)
        return self.search(query_embedding, top_k)
    
    def count(self) -> int:
        """저장된 문서 개수"""
        return self.collection.count()
    
    def delete_collection(self):
        """컬렉션 삭제"""
        self.client.delete_collection(self.collection.name)
        print(f"🗑️ 컬렉션 삭제: {self.collection.name}")

# 테스트
if __name__ == "__main__":
    import sys
    sys.path.append('..')
    from utils.data_loader import DataLoader
    from embeddings import EmbeddingGenerator
    
    # 1. 데이터 로드
    loader = DataLoader()
    faqs = loader.load_faqs()
    documents = loader.create_documents(faqs)
    
    # 2. 임베딩 생성
    embedder = EmbeddingGenerator()
    texts = [doc['text'] for doc in documents]
    embeddings = embedder.embed_texts(texts)
    
    # 3. 벡터 DB에 저장
    vector_store = VectorStore()
    vector_store.add_documents(documents, embeddings)
    
    print(f"\n총 {vector_store.count()}개 문서 저장됨")
    
    # 4. 검색 테스트
    query = "배송비가 궁금해요"
    print(f"\n🔍 검색어: {query}")
    results = vector_store.search_by_text(query, embedder, top_k=3)
    
    for i, result in enumerate(results, 1):
        print(f"\n{i}. [{result['metadata']['category']}]")
        print(f"   {result['metadata']['question']}")
        print(f"   거리: {result['distance']:.4f}")

---

🤖 Step 3: RAG 챗봇 엔진 구축 (30분)

src/rag_engine.py

import os
from typing import List, Dict, Optional
from openai import OpenAI
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

class RAGEngine:
    """RAG 기반 고객 지원 챗봇 엔진"""
    
    def __init__(self, vector_store, embedder, model_type: str = "gpt"):
        """
        Args:
            vector_store: VectorStore 인스턴스
            embedder: EmbeddingGenerator 인스턴스
            model_type: 'gpt' 또는 'claude'
        """
        self.vector_store = vector_store
        self.embedder = embedder
        self.model_type = model_type
        
        if model_type == "gpt":
            self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
            self.model = "gpt-4o-mini"
        else:
            self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
            self.model = "claude-3-5-sonnet-20241022"
    
    def retrieve_context(self, query: str, top_k: int = 3) -> List[Dict]:
        """쿼리와 관련된 문서 검색"""
        return self.vector_store.search_by_text(query, self.embedder, top_k)
    
    def build_prompt(self, query: str, context_docs: List[Dict]) -> str:
        """프롬프트 구성"""
        # 컨텍스트 조합
        context_text = "\n\n".join([
            f"[{i+1}] {doc['metadata']['question']}\n{doc['metadata']['answer']}"
            for i, doc in enumerate(context_docs)
        ])
        
        prompt = f"""당신은 친절한 고객 지원 상담원입니다. 아래 지식 베이스를 참고하여 고객의 질문에 답변해주세요.

## 지식 베이스
{context_text}

## 답변 규칙
1. 지식 베이스에 정보가 있으면 그것을 바탕으로 정확하게 답변하세요
2. 지식 베이스에 없는 내용은 "죄송하지만 해당 내용은 확인이 어렵습니다. 고객센터(1588-XXXX)로 문의해주세요"라고 안내하세요
3. 친절하고 공손한 말투를 사용하세요
4. 답변은 간결하고 명확하게 작성하세요

## 고객 질문
{query}

## 답변"""
        
        return prompt
    
    def generate_response(self, prompt: str) -> str:
        """LLM을 사용하여 답변 생성"""
        if self.model_type == "gpt":
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "당신은 친절한 고객 지원 상담원입니다."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,  # 일관된 답변을 위해 낮게 설정
                max_tokens=500
            )
            return response.choices[0].message.content
        
        else:  # Claude
            response = self.client.messages.create(
                model=self.model,
                max_tokens=500,
                temperature=0.3,
                messages=[{"role": "user", "content": prompt}]
            )
            return response.content[0].text
    
    def query(self, user_query: str, return_context: bool = False) -> Dict:
        """전체 RAG 파이프라인 실행"""
        # 1. 관련 문서 검색
        context_docs = self.retrieve_context(user_query, top_k=3)
        
        # 2. 프롬프트 구성
        prompt = self.build_prompt(user_query, context_docs)
        
        # 3. 답변 생성
        answer = self.generate_response(prompt)
        
        result = {
            "query": user_query,
            "answer": answer,
            "model": self.model
        }
        
        if return_context:
            result["context_docs"] = context_docs
            result["prompt"] = prompt
        
        return result

# 테스트
if __name__ == "__main__":
    import sys
    sys.path.append('..')
    from utils.data_loader import DataLoader
    from embeddings import EmbeddingGenerator
    from vector_store import VectorStore
    
    # 초기화
    loader = DataLoader()
    embedder = EmbeddingGenerator()
    vector_store = VectorStore()
    
    # 벡터 DB가 비어있으면 데이터 로드
    if vector_store.count() == 0:
        print("벡터 DB 초기화 중...")
        faqs = loader.load_faqs()
        documents = loader.create_documents(faqs)
        texts = [doc['text'] for doc in documents]
        embeddings = embedder.embed_texts(texts)
        vector_store.add_documents(documents, embeddings)
    
    # RAG 엔진 생성
    rag = RAGEngine(vector_store, embedder, model_type="gpt")
    
    # 테스트 쿼리
    test_queries = [
        "배송비가 얼마예요?",
        "반품하고 싶은데 어떻게 해야 하나요?",
        "적립금은 언제 소멸되나요?",
        "iPhone 15 Pro는 언제 재입고되나요?"  # 지식 베이스에 없는 질문
    ]
    
    for query in test_queries:
        print(f"\n{'='*60}")
        print(f"❓ 질문: {query}")
        print(f"{'='*60}")
        
        result = rag.query(query, return_context=True)
        
        print(f"\n📚 검색된 문서:")
        for i, doc in enumerate(result['context_docs'], 1):
            print(f"  {i}. [{doc['metadata']['category']}] {doc['metadata']['question']}")
        
        print(f"\n💬 답변:\n{result['answer']}")

---

🎨 Step 4: Streamlit 챗봇 UI (25분)

app.py

import streamlit as st
import sys
sys.path.append('src')
sys.path.append('utils')

from rag_engine import RAGEngine
from vector_store import VectorStore
from embeddings import EmbeddingGenerator
from data_loader import DataLoader

# 페이지 설정
st.set_page_config(
    page_title="고객 지원 챗봇",
    page_icon="🤖",
    layout="wide"
)

# CSS 스타일링
st.markdown("""
<style>
    .user-message {
        background-color: #E3F2FD;
        padding: 15px;
        border-radius: 10px;
        margin: 10px 0;
    }
    .bot-message {
        background-color: #F5F5F5;
        padding: 15px;
        border-radius: 10px;
        margin: 10px 0;
    }
    .context-doc {
        background-color: #FFF9C4;
        padding: 10px;
        border-radius: 5px;
        margin: 5px 0;
        font-size: 0.9em;
    }
</style>
""", unsafe_allow_html=True)

# 초기화
@st.cache_resource
def initialize_system():
    """시스템 초기화 (캐싱)"""
    loader = DataLoader()
    embedder = EmbeddingGenerator()
    vector_store = VectorStore()
    
    # 벡터 DB가 비어있으면 데이터 로드
    if vector_store.count() == 0:
        with st.spinner("지식 베이스 초기화 중..."):
            faqs = loader.load_faqs()
            documents = loader.create_documents(faqs)
            texts = [doc['text'] for doc in documents]
            embeddings = embedder.embed_texts(texts)
            vector_store.add_documents(documents, embeddings)
    
    return embedder, vector_store

embedder, vector_store = initialize_system()

# 세션 상태 초기화
if 'messages' not in st.session_state:
    st.session_state.messages = []
if 'rag_engine' not in st.session_state:
    st.session_state.rag_engine = None

# 헤더
st.title("🤖 AI 고객 지원 챗봇")
st.markdown("무엇을 도와드릴까요? 배송, 반품, 결제 등 궁금하신 점을 물어보세요!")

# 사이드바 - 설정
with st.sidebar:
    st.header("⚙️ 설정")
    
    model_type = st.selectbox(
        "AI 모델 선택",
        ["GPT-4o-mini", "Claude Sonnet"],
        help="답변 생성에 사용할 AI 모델"
    )
    
    show_context = st.checkbox(
        "검색된 문서 표시",
        value=False,
        help="답변 생성에 사용된 관련 문서 표시"
    )
    
    top_k = st.slider(
        "검색할 문서 수",
        min_value=1,
        max_value=5,
        value=3,
        help="유사도 기준 상위 K개 문서 검색"
    )
    
    # RAG 엔진 생성/업데이트
    model_map = {
        "GPT-4o-mini": "gpt",
        "Claude Sonnet": "claude"
    }
    
    if st.session_state.rag_engine is None or \
       st.session_state.rag_engine.model_type != model_map[model_type]:
        st.session_state.rag_engine = RAGEngine(
            vector_store,
            embedder,
            model_type=model_map[model_type]
        )
    
    st.markdown("---")
    
    # 통계
    st.subheader("📊 시스템 정보")
    st.metric("지식 베이스 문서", vector_store.count())
    st.metric("대화 수", len(st.session_state.messages) // 2)
    
    st.markdown("---")
    
    # 초기화 버튼
    if st.button("🔄 대화 초기화"):
        st.session_state.messages = []
        st.rerun()
    
    st.markdown("---")
    
    # 예시 질문
    st.subheader("💡 예시 질문")
    example_questions = [
        "배송비가 얼마예요?",
        "반품은 어떻게 하나요?",
        "적립금 사용 방법은?",
        "회원 등급 혜택은?",
        "무통장입금 기한은?"
    ]
    
    for question in example_questions:
        if st.button(question, key=f"example_{question}", use_container_width=True):
            st.session_state.messages.append({
                "role": "user",
                "content": question
            })
            st.rerun()

# 메인 채팅 영역
chat_container = st.container()

with chat_container:
    # 이전 대화 표시
    for message in st.session_state.messages:
        if message["role"] == "user":
            st.markdown(f"""
            <div class="user-message">
                <strong>👤 고객:</strong><br>
                {message["content"]}
            </div>
            """, unsafe_allow_html=True)
        else:
            st.markdown(f"""
            <div class="bot-message">
                <strong>🤖 상담원:</strong><br>
                {message["content"]}
            </div>
            """, unsafe_allow_html=True)
            
            # 컨텍스트 문서 표시
            if show_context and "context_docs" in message:
                with st.expander("📚 참고한 문서"):
                    for i, doc in enumerate(message["context_docs"], 1):
                        st.markdown(f"""
                        <div class="context-doc">
                            <strong>{i}. [{doc['metadata']['category']}]</strong><br>
                            Q: {doc['metadata']['question']}<br>
                            <small>유사도: {1 - doc['distance']:.2%}</small>
                        </div>
                        """, unsafe_allow_html=True)

# 입력 영역
user_input = st.chat_input("궁금하신 점을 입력해주세요...")

if user_input:
    # 사용자 메시지 추가
    st.session_state.messages.append({
        "role": "user",
        "content": user_input
    })
    
    # RAG 쿼리 실행
    with st.spinner("답변 생성 중..."):
        result = st.session_state.rag_engine.query(
            user_input,
            return_context=show_context
        )
        
        # 봇 응답 추가
        bot_message = {
            "role": "assistant",
            "content": result["answer"]
        }
        
        if show_context:
            bot_message["context_docs"] = result["context_docs"]
        
        st.session_state.messages.append(bot_message)
    
    st.rerun()

# Footer
st.markdown("---")
st.markdown("""
<div style='text-align: center; color: #666; font-size: 0.9em;'>
    💡 이 챗봇은 RAG (Retrieval-Augmented Generation) 기술을 사용합니다.<br>
    실시간 지식 베이스 검색으로 정확한 답변을 제공합니다.
</div>
""", unsafe_allow_html=True)

실행

streamlit run app.py

---

🎯 Step 5: Few-shot vs Zero-shot 비교 (20분)

src/chatbot.py

from typing import List, Dict
from openai import OpenAI
import os

class ChatbotComparison:
    """다양한 프롬프팅 기법 비교"""
    
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def zero_shot(self, query: str) -> str:
        """Zero-shot: 예시 없이 바로 답변"""
        prompt = f"""당신은 고객 지원 상담원입니다.

고객 질문: {query}

답변:"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content
    
    def few_shot(self, query: str, examples: List[Dict]) -> str:
        """Few-shot: 예시를 제공하여 답변"""
        examples_text = "\n\n".join([
            f"고객: {ex['question']}\n상담원: {ex['answer']}"
            for ex in examples
        ])
        
        prompt = f"""당신은 고객 지원 상담원입니다. 다음 예시를 참고하여 답변해주세요.

# 예시 대화
{examples_text}

# 실제 상담
고객: {query}
상담원:"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content
    
    def rag_based(self, query: str, context_docs: List[Dict]) -> str:
        """RAG: 검색된 문서 기반 답변"""
        context = "\n\n".join([
            f"[문서 {i+1}]\n{doc['metadata']['question']}\n{doc['metadata']['answer']}"
            for i, doc in enumerate(context_docs)
        ])
        
        prompt = f"""당신은 고객 지원 상담원입니다. 아래 지식 베이스를 참고하여 답변해주세요.

# 지식 베이스
{context}

# 고객 질문
{query}

# 답변 (지식 베이스에 근거하여 정확하게 답변)"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content

# 비교 실험
if __name__ == "__main__":
    import sys
    sys.path.append('..')
    from embeddings import EmbeddingGenerator
    from vector_store import VectorStore
    
    chatbot = ChatbotComparison()
    embedder = EmbeddingGenerator()
    vector_store = VectorStore()
    
    query = "배송비가 궁금합니다"
    
    print("="*60)
    print(f"질문: {query}\n")
    
    # 1. Zero-shot
    print("1️⃣ Zero-shot (예시 없음)")
    print("-"*60)
    answer = chatbot.zero_shot(query)
    print(answer)
    print()
    
    # 2. Few-shot
    print("2️⃣ Few-shot (예시 제공)")
    print("-"*60)
    examples = [
        {
            "question": "반품은 어떻게 하나요?",
            "answer": "상품 수령 후 7일 이내 마이페이지에서 반품 신청 가능합니다."
        },
        {
            "question": "적립금은 어떻게 쓰나요?",
            "answer": "5,000원 이상부터 사용 가능하며, 적립일로부터 1년간 유효합니다."
        }
    ]
    answer = chatbot.few_shot(query, examples)
    print(answer)
    print()
    
    # 3. RAG
    print("3️⃣ RAG (지식 베이스 검색)")
    print("-"*60)
    context_docs = vector_store.search_by_text(query, embedder, top_k=2)
    answer = chatbot.rag_based(query, context_docs)
    print(answer)
    print()
    
    print("="*60)
    print("💡 결론:")
    print("- Zero-shot: 일반적이고 때론 부정확할 수 있음")
    print("- Few-shot: 더 일관된 형식, 하지만 여전히 일반적")
    print("- RAG: 가장 정확하고 구체적인 답변 (실제 지식 베이스 활용)")

---

🎓 Step 6: Fine-tuning (고급, 30분)

src/finetuning.py

from openai import OpenAI
import os
import json
from typing import List, Dict

class FineTuner:
    """OpenAI Fine-tuning"""
    
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def prepare_training_data(self, faqs: List[Dict], output_file: str = "data/training_data.jsonl"):
        """Fine-tuning용 데이터 준비"""
        training_data = []
        
        for faq in faqs:
            example = {
                "messages": [
                    {"role": "system", "content": "당신은 친절한 고객 지원 상담원입니다."},
                    {"role": "user", "content": faq['question']},
                    {"role": "assistant", "content": faq['answer']}
                ]
            }
            training_data.append(example)
        
        # JSONL 형식으로 저장
        with open(output_file, 'w', encoding='utf-8') as f:
            for example in training_data:
                f.write(json.dumps(example, ensure_ascii=False) + '\n')
        
        print(f"✅ {len(training_data)}개의 훈련 데이터 생성: {output_file}")
        return output_file
    
    def upload_file(self, file_path: str):
        """훈련 파일 업로드"""
        with open(file_path, 'rb') as f:
            response = self.client.files.create(
                file=f,
                purpose='fine-tune'
            )
        
        print(f"✅ 파일 업로드 완료: {response.id}")
        return response.id
    
    def create_fine_tune_job(self, file_id: str, model: str = "gpt-4o-mini-2024-07-18"):
        """Fine-tuning 작업 생성"""
        job = self.client.fine_tuning.jobs.create(
            training_file=file_id,
            model=model
        )
        
        print(f"✅ Fine-tuning 작업 생성: {job.id}")
        print(f"   상태: {job.status}")
        return job.id
    
    def check_status(self, job_id: str):
        """Fine-tuning 상태 확인"""
        job = self.client.fine_tuning.jobs.retrieve(job_id)
        print(f"작업 ID: {job.id}")
        print(f"상태: {job.status}")
        print(f"모델: {job.model}")
        if job.fine_tuned_model:
            print(f"Fine-tuned 모델: {job.fine_tuned_model}")
        return job
    
    def test_model(self, model_id: str, query: str):
        """Fine-tuned 모델 테스트"""
        response = self.client.chat.completions.create(
            model=model_id,
            messages=[
                {"role": "system", "content": "당신은 친절한 고객 지원 상담원입니다."},
                {"role": "user", "content": query}
            ]
        )
        return response.choices[0].message.content

# 사용 예시
if __name__ == "__main__":
    import sys
    sys.path.append('..')
    from utils.data_loader import DataLoader
    
    loader = DataLoader()
    faqs = loader.load_faqs()
    
    finetuner = FineTuner()
    
    # 1. 훈련 데이터 준비
    print("📝 Step 1: 훈련 데이터 준비")
    training_file = finetuner.prepare_training_data(faqs)
    
    print("\n⚠️ Fine-tuning은 비용이 발생합니다!")
    print("계속하려면 아래 주석을 해제하고 실행하세요.\n")
    
    # # 2. 파일 업로드
    # print("📤 Step 2: 파일 업로드")
    # file_id = finetuner.upload_file(training_file)
    
    # # 3. Fine-tuning 작업 생성
    # print("\n🔧 Step 3: Fine-tuning 작업 생성")
    # job_id = finetuner.create_fine_tune_job(file_id)
    
    # print("\n⏳ Fine-tuning이 완료될 때까지 기다립니다 (보통 10-30분 소요)")
    # print(f"   상태 확인: finetuner.check_status('{job_id}')")

---

📊 Step 7: 성능 평가 (20분)

src/evaluator.py

from typing import List, Dict
import time

class ChatbotEvaluator:
    """챗봇 성능 평가"""
    
    def __init__(self, rag_engine):
        self.rag_engine = rag_engine
    
    def evaluate_retrieval(self, test_cases: List[Dict]) -> Dict:
        """검색 성능 평가"""
        results = {
            "total": len(test_cases),
            "correct": 0,
            "mrr": 0.0,  # Mean Reciprocal Rank
            "details": []
        }
        
        reciprocal_ranks = []
        
        for case in test_cases:
            query = case['query']
            expected_category = case['expected_category']
            
            # 문서 검색
            context_docs = self.rag_engine.retrieve_context(query, top_k=5)
            
            # 정답 문서 순위 찾기
            rank = None
            for i, doc in enumerate(context_docs, 1):
                if doc['metadata']['category'] == expected_category:
                    rank = i
                    break
            
            if rank:
                results["correct"] += 1
                reciprocal_ranks.append(1.0 / rank)
            else:
                reciprocal_ranks.append(0.0)
            
            results["details"].append({
                "query": query,
                "expected": expected_category,
                "found_at_rank": rank,
                "retrieved_categories": [doc['metadata']['category'] for doc in context_docs]
            })
        
        results["accuracy"] = results["correct"] / results["total"]
        results["mrr"] = sum(reciprocal_ranks) / len(reciprocal_ranks)
        
        return results
    
    def evaluate_response_quality(self, test_cases: List[Dict]) -> Dict:
        """답변 품질 평가"""
        results = {
            "total": len(test_cases),
            "avg_response_time": 0.0,
            "responses": []
        }
        
        total_time = 0.0
        
        for case in test_cases:
            query = case['query']
            
            start_time = time.time()
            result = self.rag_engine.query(query, return_context=True)
            response_time = time.time() - start_time
            
            total_time += response_time
            
            results["responses"].append({
                "query": query,
                "answer": result["answer"],
                "response_time": response_time,
                "context_count": len(result["context_docs"])
            })
        
        results["avg_response_time"] = total_time / results["total"]
        
        return results

# 평가 실행
if __name__ == "__main__":
    import sys
    sys.path.append('..')
    from rag_engine import RAGEngine
    from vector_store import VectorStore
    from embeddings import EmbeddingGenerator
    from utils.data_loader import DataLoader
    
    # 초기화
    loader = DataLoader()
    embedder = EmbeddingGenerator()
    vector_store = VectorStore()
    
    if vector_store.count() == 0:
        faqs = loader.load_faqs()
        documents = loader.create_documents(faqs)
        texts = [doc['text'] for doc in documents]
        embeddings = embedder.embed_texts(texts)
        vector_store.add_documents(documents, embeddings)
    
    rag = RAGEngine(vector_store, embedder, model_type="gpt")
    evaluator = ChatbotEvaluator(rag)
    
    # 테스트 케이스
    test_cases = [
        {"query": "배송비가 얼마예요?", "expected_category": "배송"},
        {"query": "반품하고 싶어요", "expected_category": "반품"},
        {"query": "적립금 사용법", "expected_category": "회원"},
        {"query": "결제 방법", "expected_category": "결제"},
        {"query": "무통장입금 기한", "expected_category": "결제"}
    ]
    
    print("🔍 검색 성능 평가")
    print("="*60)
    retrieval_results = evaluator.evaluate_retrieval(test_cases)
    print(f"정확도: {retrieval_results['accuracy']:.2%}")
    print(f"MRR: {retrieval_results['mrr']:.4f}")
    
    print("\n💬 답변 품질 평가")
    print("="*60)
    quality_results = evaluator.evaluate_response_quality(test_cases[:3])
    print(f"평균 응답 시간: {quality_results['avg_response_time']:.2f}초")
    
    for resp in quality_results['responses']:
        print(f"\nQ: {resp['query']}")
        print(f"A: {resp['answer'][:100]}...")
        print(f"⏱️ {resp['response_time']:.2f}초 | 📚 {resp['context_count']}개 문서 참조")

---

🚀 Step 8: 프로덕션 배포 (15분)

requirements.txt

openai==1.59.5
anthropic==0.42.0
streamlit==1.41.1
python-dotenv==1.0.1
chromadb==0.6.3
sentence-transformers==3.3.1
langchain==0.3.14
langchain-openai==0.2.14
langchain-community==0.3.13
pandas==2.2.3
numpy==2.2.1
tiktoken==0.8.0
torch==2.5.1
transformers==4.47.1
peft==0.14.0
accelerate==1.2.1

배포 체크리스트

# 1. 환경 변수 설정 확인
cat .env

# 2. 의존성 설치
pip install -r requirements.txt

# 3. 지식 베이스 초기화
python -c "from src.vector_store import VectorStore; from src.embeddings import EmbeddingGenerator; from utils.data_loader import DataLoader; loader = DataLoader(); embedder = EmbeddingGenerator(); vs = VectorStore(); faqs = loader.load_faqs(); docs = loader.create_documents(faqs); texts = [d['text'] for d in docs]; embs = embedder.embed_texts(texts); vs.add_documents(docs, embs)"

# 4. 로컬 테스트
streamlit run app.py

# 5. Streamlit Cloud 배포 (선택)
# - GitHub 리포지토리에 푸시
# - https://streamlit.io/cloud 에서 연결
# - Secrets에 API 키 추가

---

🎓 학습 과제 및 개선 방향

필수 과제

1. ✅ 기본 RAG 챗봇 완성
2. ✅ Zero-shot vs Few-shot 비교
3. ✅ Vector DB 검색 성능 측정
4. ✅ Streamlit UI 구현

심화 과제

1. 다국어 지원: 영어, 일본어 FAQ 추가
2. 대화 히스토리: 이전 대화 컨텍스트 활용
3. 감정 분석: 고객 불만 자동 감지
4. A/B 테스팅: 다른 프롬프트 전략 비교
5. Fine-tuning: 실제 대화 로그로 모델 개선

추가 기능 아이디어

• 🔊 음성 입력/출력 (STT/TTS)
• 📸 이미지 업로드 (상품 사진으로 문의)
• 📊 상담 통계 대시보드
• 🔔 실시간 알림 (고객센터 연결 필요 시)
• 🌐 다중 언어 자동 감지