외부 API 없이 문서 기반 AI 챗봇 만들기 (Hugging Face 활용)

외부 API 없이 문서 기반 AI 챗봇 만들기 (Hugging Face 활용)

안녕하세요. 파파대니입니다.

최근 인공지능 기술이 급격히 발전하면서, 챗봇은 다양한 분야에서 중요한 도구로 자리 잡고 있습니다.

그러나 많은 챗봇 개발 과정에서 외부 API 사용이 필수적으로 요구되며, 이는 비용 문제나 데이터 보안 우려를 불러일으킬 수 있습니다.

이에 따라, 외부 API에 의존하지 않고도 문서 기반 RAG(Retrieval-Augmented Generation) 챗봇을 구현하는 방법에 대한 관심이 높아지고 있습니다.

이 글에서는 비용 절감과 데이터 보호를 고려해 로컬 환경에서 완전히 작동하는 RAG 챗봇을 구현하는 방법을 단계별로 자세히 안내합니다.

특히, Hugging Face 모델과 FAISS를 활용하여 효율적이고 실용적인 챗봇을 구축할 수 있는 솔루션을 제공합니다.

이 글은 챗봇 개발 초보자부터 전문가까지 누구나 따라 할 수 있는 실용적인 내용을 담고 있으며, 외부 API 없이도 고성능의 문서 기반 AI 챗봇을 만들 수 있는 자신감을 드릴 것입니다.

 

1. 개발 환경 설정: 필요 도구 설치

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

# 필수 라이브러리 설치
pip install -U \
    langchain \
    pypdf \
    faiss-cpu \
    sentence-transformers \
    python-dotenv \
    transformers \
    torch

---

2. 문서 전처리 스크립트

import os
import logging
from typing import List
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document

class DocumentProcessor:
    def __init__(self, chunk_size=1000, chunk_overlap=200):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def load_documents(self, folder_path: str) -> List[Document]:
        """
        지정된 폴더의 모든 PDF 문서 로드
        """
        documents = []
        for filename in os.listdir(folder_path):
            if filename.endswith('.pdf'):
                filepath = os.path.join(folder_path, filename)
                loader = PyPDFLoader(filepath)
                documents.extend(loader.load())
                self.logger.info(f"문서 로드: {filename}")
        return documents

    def split_documents(self, documents: List[Document]) -> List[Document]:
        """
        문서를 지정된 크기로 분할
        """
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap
        )
        
        split_docs = text_splitter.split_documents(documents)
        self.logger.info(f"총 {len(split_docs)}개 문서 청크 생성")
        return split_docs

    def clean_text(self, text: str) -> str:
        """
        텍스트 정제 메서드
        """
        # 특수문자, 불필요한 공백 제거 등
        text = text.strip()
        text = ' '.join(text.split())
        return text

    def process_documents(self, folder_path: str) -> List[Document]:
        """
        문서 로드, 분할, 정제 통합 메서드
        """
        documents = self.load_documents(folder_path)
        split_docs = self.split_documents(documents)
        
        # 텍스트 정제
        cleaned_docs = [
            Document(
                page_content=self.clean_text(doc.page_content),
                metadata=doc.metadata
            ) for doc in split_docs
        ]
        
        return cleaned_docs

# 사용 예시
if __name__ == "__main__":
    processor = DocumentProcessor()
    processed_docs = processor.process_documents('./company_docs')

---

3. 임베딩 및 벡터 스토어 구현

import os
from dotenv import load_dotenv
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from document_processor import DocumentProcessor

class VectorStoreManager:
    def __init__(self, embedding_model='huggingface'):
        load_dotenv()
        self.embedding_model = self._select_embedding_model(embedding_model)
    
    def _select_embedding_model(self, model_type):
        """
        임베딩 모델 선택
        """
        if model_type == 'huggingface':
            return HuggingFaceEmbeddings(
                model_name='all-MiniLM-L6-v2'
            )
        else:
            raise ValueError("지원되지 않는 임베딩 모델")

    def create_vector_store(self, documents):
        """
        문서 벡터화 및 벡터 저장소 생성
        """
        vector_store = FAISS.from_documents(
            documents, 
            self.embedding_model
        )
        return vector_store

    def save_vector_store(self, vector_store, path='./vector_db'):
        """
        벡터 저장소 로컬에 저장
        """
        vector_store.save_local(path)
        print(f"벡터 DB 저장 완료: {path}")

    def load_vector_store(self, path='./vector_db'):
        """
        저장된 벡터 저장소 로드
        """
        return FAISS.load_local(
            path, 
            self.embedding_model
        )

# 사용 예시
if __name__ == "__main__":
    processor = DocumentProcessor()
    docs = processor.process_documents('./company_docs')
    
    vm = VectorStoreManager()
    vector_store = vm.create_vector_store(docs)
    vm.save_vector_store(vector_store)

---

4. RAG 기반 챗봇 구현

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.llms import HuggingFacePipeline
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from vector_database import VectorStoreManager

class RAGChatbot:
    def __init__(self, vector_store, model_name='gpt2', temperature=0.7):
        # Hugging Face 모델 로드
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForCausalLM.from_pretrained(model_name)

        # 파이프라인 생성
        text_gen_pipeline = pipeline(
            "text-generation", 
            model=model, 
            tokenizer=tokenizer,
            device=0  # GPU 사용 시
        )

        # LangChain과 통합
        self.llm = HuggingFacePipeline(pipeline=text_gen_pipeline)
        
        # 메모리 및 검색 체인 설정
        self.memory = ConversationBufferMemory(
            memory_key="chat_history", 
            return_messages=True
        )
        self.retrieval_chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=vector_store.as_retriever(search_kwargs={"k": 3}),
            memory=self.memory,
            verbose=True
        )

    def get_response(self, query):
        result = self.retrieval_chain({"question": query})
        return result['answer']

    def reset_memory(self):
        self.memory.clear()

# 사용 예시
if __name__ == "__main__":
    vm = VectorStoreManager()
    vector_store = vm.load_vector_store()
    
    chatbot = RAGChatbot(vector_store)
    
    while True:
        query = input("질문: ")
        if query.lower() in ['exit', 'quit']:
            break
        
        response = chatbot.get_response(query)
        print("답변:", response)

---

 

RAG 챗봇은 단순한 대화형 AI를 넘어, 사용자가 제공한 문서를 기반으로 유용한 정보를 제공할 수 있는 강력한 도구입니다.

이제 이 가이드를 기반으로 자신만의 맞춤형 AI 챗봇을 설계하고 구현해 보세요.

앞으로 AI 기술을 활용해 더 많은 문제를 해결할 수 있는 가능성을 발견하시기를 바랍니다.

여러분의 성공적인 챗봇 개발 여정을 응원합니다!

 

 

⬇️ 참고하면 좋은 글 ⬇️

 

생성형 AI와 대화형 AI 차이: 무엇이 다를까?

생성형 AI와 대화형 AI 차이: 무엇이 다를까?

 

RAG와 Fine-tuning: 생성형 AI와 대화형 AI 모두 활용 가능한 기술

RAG와 Fine-tuning: 생성형 AI와 대화형 AI 모두 활용 가능한 기술

 

회사의 문서로 챗봇 만들기: RAG와 Fine-Tuning 기법 완벽 가이드

회사의 문서로 챗봇 만들기: RAG와 Fine-Tuning 기법 완벽 가이드