azure openAI (GPT4.0, ada-002)

azure openAI에 있는 GPT 모델을 사용해서 나만의 비서를 만들어보자!
(최신 자료는 gpt가 모르고, 정확도가 떨어지는 경우가 있으니)

구조

모델을 그대로 가져다 쓸거라 구조는 단순합니다.

1. 필요한 문서를 크롤링으로 가져와서 text화 합니다
2. 모든 text를 embed model에 집어넣고 indexing 된 data를 csv로 저장합니다
3. 메세지를 받으면 gpt model이 indexing된 data를 참고해서 유사도 검색을 할 수 있도록 설정합니다

 

Crawling

문서 페이지에서 어떤 데이터들을 가져올 건지 확인해야 합니다

페이지에서 공통적으로 사용하는 html class를 찾고, 해당 class에 원하는 데이터들이 담겨있는지 만 확인하면 됩니다.

 

예시로 해본 jenkins페이지에서는 section이라는 class를 공통으로 쓰는 거 같으니  크롤링 포인트를 section class로 잡고 text를 추출합니다.
(디테일하게 text를 정제할 순 있지만, 노력 대비 더 좋은 퍼포먼스가 나오진 않는 거 같았습니다.)

 

필요한 문서 페이지를 urls 배열에 전부 넣고 txt파일로 전부 저장합니다.

 

from bs4 import BeautifulSoup
import cloudscraper
import re


def getDocuments(index, url):

    # cloudflare CAPTCHA를 우회하기 위해 cloudscraper사용
    scraper = cloudscraper.create_scraper()
    response = scraper.get(url)
    response.raise_for_status()

    # HTML 파싱
    soup = BeautifulSoup(response.text, 'html.parser')
    large_spans = soup.find_all('div', {'class': 'section'})
    for large_span in large_spans:
        # "<" 로 시작해서 ">"로 끝나는건 모두 지우는 정규식
        cleaned_text = re.sub(r'<.*?>', '', str(large_span))

        with open(f"./jenkins/{index}.txt", 'w', encoding='utf-8') as file:
            file.write(cleaned_text)

urls = [
    "https://www.jenkins.io/doc/book/pipeline/",
    "https://www.jenkins.io/doc/book/pipeline/getting-started/",
]
for index, url in enumerate(urls):
    documents =[]
    getDocuments(index, url)

 

Embeding

1. 모든 문서를 embeding 한 뒤 생성된 indexing data를 csv로 저장합니다.
   (data 형태는 vector이기에 db에 넣고 쓰려면, es나 redis를 사용하는 게 보편적)
2. system명령어를 주입해서 약간의 custom이 가능합니다.
3. 사용자 message를 받아서 embeding 한 후 message data와 indexing data 간의 유사도를 측정해서 가장 비슷한 문서를 return 합니다.
4. gpt 4.0 모델에서는 2번과 3번을 조합하여 최종 message를 return 합니다.

 

생성되는 csv sample

 

import os
import openai
import faiss
import numpy as np
import pandas as pd
openai.api_type = "azure"
openai.api_base = "https://xxx.openai.azure.com/"
openai.api_version = "2023-03-15-preview"
openai.api_key = 'xxxx'

 

 

deployment_name='chat'
embed_name='ada'
documents = []
index= None
embedding_matrix = None
embeddings = None
file = None
df = None

def get_embedding(text):
    text = text.replace("\n", " ")
    while True:
        try:
            response = openai.Embedding.create(input=[text], engine=embed_name)
            return response["data"][0]["embedding"]
        except openai.error.RateLimitError:
            print("limit rate")
            time.sleep(30)

def runBot(user_input):
    global index, embedding_matrix, embeddings, file, documents
    file_path = 'embeddings.csv'

    # 1. article파일을 크롤링해서 받아옵니다.
    if not os.path.exists(file_path):
        documents = get_article()
        with open(file_path, 'w'):
            pass
        df = None
    else:
        # 1.1 파일이 있을땐 csv파일을 가져와서 data로 사용합니다.
        df = pd.read_csv(file_path)

    directory_path = '.\\docs'
    files = list_files_in_directory(directory_path)
    for file in files:
        read_text_file_to_list(f'{directory_path}\\{file}')

    # 2. 크롤링으로 정제된 article 전체를 embedding 합니다  
    if index is None and df is None :
        embeddings = [get_embedding(doc) for doc in documents]
        embedding_matrix = np.array(embeddings)
        # 2.1 L2메서드를 사용해서 유클리드 거리(Euclidean distance)를 계산하는 방식으로 인덱스를 생성합니다.
        index = faiss.IndexFlatL2(embedding_matrix.shape[1])
        # 2.2 임베딩 행렬의 각 행(각각의 문서 임베딩)에 대한 참조를 할 수 있게 인덱스에 임베딩 행렬을 추가
        index.add(embedding_matrix)
        df = pd.DataFrame(embeddings)
        df.to_csv(file_path, index=False)
    else:
        embeddings = df
        embedding_matrix = np.array(df)
        index = faiss.IndexFlatL2(embedding_matrix.shape[1])
        index.add(df)
               
    system_message = "Please tell me in the kindest way possible"

    # 챗봇 객체 생성
    chatbot = Chatbot(index, documents, system_message)
    response = chatbot.chat(f'{user_input}')

    return file, response

 

 

class Chatbot:
    def __init__(self, index, documents, system_message):
        self.index = index
        self.documents = documents
        self.system_message = system_message
        self.chat_history = []



    def find_similar_document(self, user_embedding):
        _, top_indices = self.index.search(np.array([user_embedding]), 1)
        top_index = top_indices[0][0]
        return self.documents[top_index]

    def chat(self, user_input):
        # 3. user_input기반으로 유사한 article을 찾고, gpt가 해당 article 기반으로 말을 만들어줍니다.
        user_embedding = get_embedding(user_input)
            # 3.1 user_intput과 article index를 비교해서 가장 비슷한 article 파일을 return합니다.
        similar_document = self.find_similar_document(user_embedding)

        system_message = self.system_message + " " + similar_document
            # 3.2 system message와 similar_document를 조합해서 gpt 모델에서 사용할 최종 message를 만듭니다.
        messages = [{"role": "system", "content": system_message}]
        for message in self.chat_history:
            messages.append(message)
        messages.append({"role": "user", "content": user_input})
            # 3.3 최종 message 생성
        response = openai.ChatCompletion.create(
            engine=deployment_name,
            temperature = 0,
            messages=messages
        )  
            # 3.4 가장 점수가 높은 message를 리턴 합니다.
        assistant_message = response.choices[0].message.content
        self.chat_history.append({"role": "user", "content": user_input})
        self.chat_history.append({"role": "assistant", "content": assistant_message})

        return assistant_message