FAISS VectorDB와 gradio를 이용한 rag GPT 구현

FAISS VectorDB와 gradio를 이용한 rag GPT 구현

 

 

이전에 구현했던 txt 파일을 csv 파일로 변환 후 embedding해서 rag 기반 gpt를 만들었던 것을 응용해서 FAISS VectorDB와 Gradio를 이용해 RAG GPT 챗봇을 구현한다.

 

본격적으로 RAG GPT 챗봇을 구현하기 전에 vector database부터 알아보자.

1. Vector Database란?

Vector Database는 데이터베이스의 한 종류로, 주로 벡터 형식의 데이터를 저장하고 빠르게 검색할 수 있도록 설계된 데이터베이스다. 특히 유사도 검색을 위한 최적화된 구조를 제공하며, 벡터화된 데이터를 기반으로 가까운 항목을 찾는 데 매우 효율적이다.

 

2. 대표적인 Vector Database

1) FAISS (Facebook AI Similarity Search)
   - **Facebook AI**에서 개발한 오픈 소스 라이브러리.
   - 매우 빠른 유사도 검색과 군집화 기능을 제공.
   - GPU 가속을 지원해 대량의 데이터를 효율적으로 처리할 수 있음.

2) Milvus
   - **Zilliz**에서 개발한 오픈 소스 vectordb로, 높은 성능과 확장성을 자랑함.
   - 벡터 검색과 혼합 검색(hybrid search)을 지원해 다양한 유형의 데이터를 처리할 수 있음.
   - 분산 시스템을 통해 대규모 데이터셋에서도 효율적인 검색이 가능.

3) Weaviate
   - 의미론적 검색을 위해 설계된 오픈 소스 vectordb.
   - 데이터베이스 내에서 벡터를 자동으로 생성하거나 기존의 임베딩을 사용할 수 있음.
   - 다양한 NLP 모델과의 통합을 지원.

4) Pinecone
   - 클라우드 기반 벡터 데이터베이스 서비스로, 사용자가 인프라를 직접 관리할 필요 없이 벡터 검색 기능을 제공.
   - 쉽게 확장 가능하고 API 기반으로 간편하게 벡터 데이터를 관리할 수 있음.
   - 지리적으로 분산된 클러스터를 제공해 글로벌 검색 성능을 최적화함.

 

3. Vector Database의 특징

- 의미 기반 검색: 단순한 키워드 매칭이 아닌, 벡터 간의 거리(예: 코사인 유사도)를 이용해 데이터의 의미적 유사성을 파악.
- 빠른 검색 속도: 수백만에서 수십억 개의 벡터에 대한 검색을 실시간으로 처리할 수 있음.
- 확장성: 많은 vectordb는 분산 처리와 클러스터링을 통해 대규모 데이터를 효율적으로 관리할 수 있음.

 

4. 주요 사용 사례

- 추천 시스템: 사용자 선호도를 분석하고 유사한 제품이나 콘텐츠를 추천.
- 챗봇 및 QA 시스템: 의미적으로 유사한 질문과 답변을 매칭하여 더 자연스러운 대화와 검색을 가능하게 함.
- 이미지 검색: 이미지의 특징 벡터를 사용해 시각적 유사도를 기반으로 검색.
- 문서 검색: 임베딩 벡터를 사용하여 의미적으로 관련된 문서나 내용을 빠르게 찾음.

Vector Database는 AI와 NLP 애플리케이션의 핵심 기술로 자리잡고 있으며, 데이터의 의미를 기반으로 한 고속 검색과 처리가 필요할 때 필수적이다.

 

5. FAISS VectorDB와 gradio를 이용한 rag GPT 구현

facebook 에서 개발한 FAISS (Facebook AI Similarity Search)를 vectorDB로 사용해 RAG 기반 GPT챗봇을 만든다.

RAG 데이터로 사용하는 파일은 이전 글 실습에서 만들었던 호텔 고객응대 매뉴얼 파일이다.

 

1) faiss 설치

nvidia GPU가 있다면 gpu버전으로 nvidia GPU가 없다면 cpu버전으로 설치한다.

!pip install faiss-cpu

 

2) 라이브러리 임포트 및 openai 클라이언트, 임베딩 매개변수를 설정한다.

import os
import pandas as pd
import gradio as gr
import tiktoken
from openai import OpenAI
import numpy as np
import faiss
from typing import List

# OpenAI 클라이언트 설정
client = OpenAI()
client.api_key = os.environ['OPENAI_API_KEY']

# 임베딩 매개변수 설정
embedding_model = "text-embedding-3-small"
embedding_encoding = "cl100k_base"

 

3) txt 파일을 불러와서 dataframe으로 변환하는 함수를 작성한다.

def process_file(file_path):
    """
    txt 파일을 불러와서 dataframe으로 변환하는 함수
    """
    tokenizer = tiktoken.get_encoding(embedding_encoding)  # 함수 내부로 이동
    with open(file_path, "r", encoding="utf-8") as f:
        data = f.read()
    
    text2df = {}
    for content in data.split("\n\n"):  # 문단 단위로 분리
        temp = content.split("\n")
        text2df.setdefault("title", []).append(temp[0])
        text2df.setdefault("content", []).append(" ".join(temp[1:]))

    df = pd.DataFrame(text2df)
    df['n_tokens'] = df['content'].apply(lambda x: len(tokenizer.encode(x)))
    return df

 

4) 텍스트를 임베딩하는 함수를 작성한다.

def get_embedding(text, model):
    """
    텍스트를 임베딩하는 함수
    """
    text = text.replace("\n", " ")
    return client.embeddings.create(input=[text], model=model).data[0].embedding

 

5) 데이터 프레임에서 임베딩한 데이터를 FAISS VectorDB에 저장하는 함수를 작성한다.

def save_embeddings_to_faiss(df):
    """
    데이터 프레임에서 텍스트를 임베딩한 데이터를 faiss 벡터 DB에 저장
    """
    embeddings = np.array(df['embeddings'].tolist(), dtype='float32')
    index = faiss.IndexFlatL2(embeddings.shape[1])  # Euclidean distance 사용
    index.add(embeddings)
    faiss.write_index(index, "vector_index.faiss")

 

6) 파일을 불러와서 데이터프레임으로 만들고 임베딩 후 embeddings.csv로 저장 그리고 faiss DB에도 저장한다.

def process_and_embed(file):
    """
    텍스트 파일을 데이터프레임으로 변환하고 임베딩하는 함수
    """
    df = process_file(file.name)
    df['embeddings'] = df['content'].apply(lambda x: get_embedding(x, model=embedding_model))
    df.to_csv('embeddings.csv', index=False)
    save_embeddings_to_faiss(df)
    return "임베딩 완료 및 저장 완료"

 

7) 질문과 RAG를 비교해 유사도가 높은 항목을 퓨샷용 컨텍스트로 만드는 함수

def create_context(question, df, max_len=128000):
    """
    질문과 데이터프레임을 비교하여 컨텍스트를 생성하는 함수
    """
    q_embeddings = client.embeddings.create(input=[question], model=embedding_model).data[0].embedding
    
    index = faiss.read_index("vector_index.faiss")
    D, I = index.search(np.array([q_embeddings], dtype='float32'), len(df))
    
    returns = []
    cur_len = 0
    for i in I[0]:
        if i == -1:  # 유사도가 낮으면 제외
            continue
        row = df.iloc[i]
        cur_len += row['n_tokens'] + 4
        if cur_len > max_len:
            break
        returns.append(row['content'])
    
    return "\n\n###\n\n".join(returns)

 

8) 컨텍스트와 질문을 OpenAI API에게 전달해 답변을 받는 함수

def answer_question(question, conversation_history):
    """
    문맥에 따라 질문에 답변하는 함수
    """
    df = pd.read_csv('embeddings.csv')
    context = create_context(question, df, max_len=128000)
    
    if not context.strip():  # 컨텍스트가 없을 경우
        return "죄송합니다, 해당 질문에 대한 답변을 찾을 수 없습니다."
    
    prompt = f"다음 컨텍스트에 기반하여 질문에 답하세요. 컨텍스트 외의 정보는 사용하지 마세요.\n\n컨텍스트: {context}\n\n---\n\n질문: {question}\n답변:" 
    conversation_history.append({"role": "user", "content": question})

    # 진행 상황 표시
    print("답변을 생성 중입니다...")
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": prompt}],
        temperature=1,
        max_tokens=500
    )
    
    answer = response.choices[0].message.content.strip()
    conversation_history.append({"role": "assistant", "content": answer})
    return answer

 

9) gradio에서 질문과 답변을 출력하도록 하는 함수

def chat_interface(question, history):
    """
    Gradio 인터페이스에서 질문에 답변을 생성하는 함수
    """
    answer = answer_question(question, history)
    display_history = "<br>".join([f"<div style='text-align: right;'>{entry['content']}</div>" if entry['role'] == 'user' else f"<div style='text-align: left;'>{entry['content']}</div>" for entry in history])
    return display_history, answer

# Gradio 인터페이스 설정
demo = gr.Blocks()

 

10) gradio로 화면 구성

탭을 이용해서 txt 파일을 업로드 후 임베딩하는 기능과 질문과 답변을 하는 기능을 분리해 구현한다.

전체 코드를 실행하면 웹브라우저에서 localhost:7860으로 접속할 수 있다.

with demo:
    gr.Markdown("# 문서 임베딩 및 질문-응답 시스템")
    with gr.Tab("파일 업로드 및 임베딩"):
        file_input = gr.File()
        embed_button = gr.Button("임베딩 및 저장")
        embed_output = gr.Textbox()
        embed_button.click(process_and_embed, inputs=file_input, outputs=embed_output)
    
    with gr.Tab("질문 및 응답"):
        question_input = gr.Textbox(placeholder="질문을 입력하세요")
        conversation_display = gr.HTML()
        answer_output = gr.Textbox(label="답변")
        ask_button = gr.Button("질문하기")
        
        # 버튼 클릭 시 chat_interface 함수와 연결
        ask_button.click(chat_interface, inputs=[question_input, gr.State([])], outputs=[conversation_display, answer_output])

demo.launch(inline=False, server_name="0.0.0.0", server_port=7860)

 

11)  실행 화면

(1) txt 파일 업로드 및 임베딩

 

(2) 질문 입력 및 응답 화면

질문을 입력하고 질문하기 버튼을 누르면 질문과 답변이 누적되어 출력된다.