STS를 활용한 간단한 QNA봇 만들기

P stage sts 대회

P stage sts대회가 끝난 후, 우리가 열정적으로 돌리던 모델들은 서버에 외롭게 남겨졌다. 세상에 존재하는 AI중 8~90퍼센트는 실제로 사용이 되고 있지 않다는 article을 medium에서 봤던 것 같다. 대부분 더 좋은 성능을 내는, 더 빠른 모델을 찾기 위해 너무나 많은 자원을 소비하지만 실제로 우리 삶에 접목이 됬던 모델들은 극히 일부였던 것 같다. 사실 나 또한 이 모델을 활용할 생각은 해보지 못했는데, AI Tech 이번 주 강의가 AI 서비스 개발 기초다(정말 짜임새 있는 교육 과정이다). 이 과정에서 MLOps의 각 components를 알았고, 자원이 많이 요구되는 풀스택 개발을 통한 웹 개발이 아닌 voila, streamlit을 사용하는 방법을 알려주셨다. Java를 사용한 간단한 웹 개발도 너무나 어려웠던 나로써는 streamlit은 신세계였다. voila 또한 노트북 자체를 웹에서 볼 수 있다는 점에서 프로토타입으로써의 활용도가 높은 것 같다.

Business model

이번 주 강의 중에서 business model 부분이 정말 흥미로웠다. AI를 공부하면서도 AI에 치우치지 않고, business를 생각하려고 노력하는 사람으로써 P stage에서 만들었던 모델을 어떤 식으로 활용해야 할까에 대한 고민은 즐거웠다. 일단 당장 생각나는 것이 없어서 구글링을 해보았다. Semantic Textual similarity business model이라고 치니 정말 많은 자료가 나왔다. Plagiarism detection, keyword search, Customer service 등등 수 많은 주제 중에서 Customer service에 눈이 갔다. 그래서 sentence1을 입력으로 받고, 내가 준비한 수많은 sentence2에서 가장 상관계수가 높은 문장을 고른 후, 그 문장에 대한 정답을 출력하는 식으로 동작하려고 했다. 향후 모델을 발전시켜 어느 산업이든 질의응답 봇으로 사용할 수도 있겠다(물론 성능은 그리 높지는 않을 수도 있다)

Baseline 모델은 단순하게 ,하지만 파이프라인은 견고하게 해야했다. 먼저 어떤 주제에 대한 customer service인지는 정하지 않았고, 나에 대한 몇 가지 질문과 답을 준비해보았다.

10개의 질문만 만들었고, 내가 만든 페이지가 정상적으로 동작 시에 더 많은 요소들을 추가할 생각이었다.

Error

이번 주 강의를 들으면서, 실습을 하면서, 미션을 풀면서 왜 이렇게 많은 라이브러리 충돌이 일어나는지 알수 없을 정도로 너무나 많은 에러를 직면했다. 신기했던 점은 모든 에러가 다 구글에 있다는 점이었다. 시간은 굉장히 많이 들었지만 전부 다 해결할만한 문제들이었다. 너무 답답한 마음에 slack에 공개적으로 질문을 했고, 그에 대한 대답으로 마스터님이 의도하신 바라는 대답을 받았다. 그 때부터 에러를 만나도 조금더 담담해진 것 같다.

Anaconda 환경을 열 몇번을 밀어가며 환경을 맞춘 결과, 결국에는 알맞는 환경을 찾았고, 이제서야 streamlit 개발을 할 수 있었다. 

과정

가장 먼저 마스터님이 주신 코드를 살펴봐야 했다. 

streamlit run app.py로 돌릴 app.py와, 질문과 정답을 저장하고 model_name등을 저장할 config파일 config.yaml과, st.cache를 위한 confirm_button_hack.py와, 우리 모델 매개변수가 담긴 ckpt파일과, Model Class를 담아줄 model.py, model을 가져오고 predict를 해줄 predict.py와, 자질구레한 함수가 저장되어 있는 utils.py가 있다.

사실 한 문장에 대해서 결과를 반환함으로 따로 dataset과 dataloader는 구현하지는 않았다. 일단 가장 간단하게 동작만 하는 baseline을 만들고 싶었다.

class Model(pl.LightningModule):
    def __init__(self, config):
        super().__init__()
        self.save_hyperparameters()

        self.model_name = config.model.model_name
        self.lr = config.train.learning_rate
        self.plm = transformers.AutoModelForSequenceClassification.from_pretrained(
            pretrained_model_name_or_path=self.model_name, num_labels=1)

    def forward(self, x):
        x = self.plm(x)['logits']
        return x

먼저 model.py다. 전에 사용했던 모델 클라스를 그대로 들고 왔고, 단순히 query 문장에 대해서 모델의 예측 결과만 필요함으로 forward까지만 구현했다.

import torch
import streamlit as st
from model import Model
import transformers
import yaml
from typing import Tuple
from omegaconf import OmegaConf


@st.cache(allow_output_mutation=True)
def load_model():
    config = OmegaConf.load(f'config.yaml')  # 질문과 정답, model_name등의 인자가 담긴 config 파일  
    model = Model.load_from_checkpoint(model_name='kykim/electra-kor-base', lr=1e-05, checkpoint_path=config['model_path']) # .ckpt파일을 불러오므로 load_from_checkpoint사용
    return model
def tokenizing(s1, s2):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu") 
    config = OmegaConf.load(f'config.yaml')
    data = []
    tokenizer = transformers.AutoTokenizer.from_pretrained(config['model_name'], max_length=50)
    for s in s2: # qeury문과 질문들을 [SEP]으로 합친 후 tokenizer에 넘겨줌
        text = s1+'[SEP]'+s
        outputs = tokenizer(text, add_special_tokens=True, padding='max_length', truncation=True)
        data.append(outputs['input_ids'])
    return data

def get_prediction(model:Model, text) -> Tuple[torch.Tensor, torch.Tensor]:
    config = OmegaConf.load(f'config.yaml')
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    qnas = [config['qna'][i][0] for i in range(len(config['qna']))] # 준비한 질문들 리스트
    data =tokenizing(text, qnas)
    outputs = model.forward(torch.tensor(data)) # forward해서 각 질문과의 유사도 예측
    return torch.max(outputs), torch.argmax(outputs)

predict.py다. 먼저 streamlit은 파일을 수정하거나 입력하면 자동으로 rendering된다고 한다. 모델을 캐시에 저장해서 속도를 향상시키려면 load_model함수 위에 @st.cache()가 필요하다. 그 안에 있는 인자는 output을 바꿀 수 있게 할 것이냐는 조건인데, 만약 없을 시에는 페이지에 다음과 같은 에러가 발생한다

모델을 저장하고 불러오는 여러 가지 방법이 있는데, 다음과 같다

• torch.save(model, path) 
  • 가중치 뿐만 아니라 optimizer, lr, batch_size 등 모델 전부 저장됨
  • 지난 번에 학습한 모델을 이어서 학습할 경우라면 사용
  • 모델의 모든 정보가 있는만큼 용량이 크다
• torch.save(model.state_dict(), path)
  • 모델의 가중치만 딕셔너리 형태로 저장
  • 이어서 학습할 것이 아니라 예측에 사용될거면 사용
  • model 전체를 저장한 것보다는 용량이 적다
• .ckpt
  • 학습 중간 가장 성능이 좋았던 모델을 저장한 경우 사용
  • Model.load_from_checkpoint(*args, checkpoint_path={path})로 불러올 수 있다

저장이 되었던 확장자에 대해서도 궁금했는데, .pt, .pth, ..pwf 다 큰 차이점이 없다고 한다. 그러나 Python Path(.pth)랑 확장자가 겹치므로 .pth를 사용하지 않는 것을 권장한다고 한다.

다음 tokenizing 함수는 P stage에서 사용했던 코드를 재사용하였다. s1은 쿼리로 하나의 문자열이고, s2는 준비된 질문들의 리스트이다. s2에서 각 요소를 가져와서 s1과 [SEP]으로 이어준 후 tokenizer에 넣어준 후 반환해준다.

get_prediction() 함수는 예측을 하는 함수다. Model.predict를 사용할 수도 있겠지만, dataloader를 같이 넣어줘야 한다는 복잡성이 있어서 단순하게 매개변수로 받아온 model을 forward해주어서 예측값을 얻었다. 반환해줄 때는 모든 질문들 중에서 가장 상관계수가 높은 값(max)과 가장 상관계수가 높은 질문의 인덱스(argmax)를 반환해주었다.

import streamlit as st

import io
import os
import yaml

from PIL import Image
from omegaconf import OmegaConf
from predict import load_model, get_prediction
from model import Model
from confirm_button_hack import cache_on_button_press
os.environ['KMP_DUPLICATE_LIB_OK']='True'
# SETTING PAGE CONFIG TO WIDE MODE
st.set_page_config(layout="wide")

def main():
    st.title("Yongchan's QNA") # 제목

    config = OmegaConf.load(f'config.yaml') #config 파일
    

    model = load_model()  # 모델 불러오기
    model.eval() 

    text_input = st.text_input("용찬이에 대해 물어보세요!") # text란 만듦, 이 값을 받아와서 비교

    if text_input:
        st.write("용찬이가 대답을 생각하고 있습니다...") # loading되는 중에 띄울 문구
        p, idx = get_prediction(model, text_input) # get_prediction으로 가져온 최대 상관계수와 가장 상관계수가 높은 질문의 인덱스
        if float(p) > 2.0: # 가장 높은 값이 2.0이하라면 질문이 config에 존재하지 않는 것으로 판단
            st.write(config['qna'][int(idx)][1]) # 질문에 대한 답 출력
        else:
            st.write('생각해본 적 없는 질문이네요.. 조금만 시간을 주시겠어요?') # 준비되어있지 않은 질문할 시에 출력
main()

먼저 st.title()로 제목을 지어준다. 그 다음 config 파일 로드 한 후 model을 load_model로 가져오고, model.eval()로 평가 상태로 바꿔준다. 다음 query 질문을 입력 받아야 하는데, 이를 st.text_input으로 받아온다. 만약 받아온 값이 존재한다면 대답을 해야 하므로 먼저 로딩되는 시간동안 띄울 문구를 st.write으로 정해준다. get_prediction으로 받아온 최고 상관계수와 인덱스를 통해 만약 상관계수가 2.0보다 작다면 query로 들어온 질문이 준비된 질문들과 관계가 없을 가능성이 크므로 생각해본 적 없는 질문이네요.. 조금만 시간을 주시겠어요? 라고 출력하고, 만약 상관계수가 더 크다면 config파일에서 해당하는 인덱스의 1번 째 인덱스(0은 질문, 1은 답)을 가져와서 출력한다.

 

결과

Streamlit은 녹화하는 기능이 있어서 동영상을 업로드하려고 하였는데 로딩 에러가 계속 나서 사진으로 첨부하겠다. 먼저 메인 화면으로 query 질문을 입력으로 넣으면 답변을 출력하는 페이지다.

만약 준비되있는 질문들 중에서 2.0 이하의 점수들만 존재한다면 관련 있는 질문이 존재하지 않는 것으로 판단하고 따로 설정해둔 문구가 출력된다

아쉬웠던 점

1. 프로토타입으로 만들었던 웹이었기에 기능이 단순하게 디자인이 허접한 점이 아쉬웠다. 그러나 프로토타입용으로 streamlit을 많이 사용하므로 디자인은 크게 중요하지 않을 수도 있겠다는 생각이 들었다. 나중에 Flask로 웹 개발을 할 때는 디자인에 신경 써보고 싶다

2. 모델이 예측을 수요하는데 10초가 넘게 걸린다. 이러한 latency 문제를 최종 프로젝트에서 해결해야할텐데 학습을 진행하면서 단순히 모델의 성능을 올리는데 집중하는 것이 아니라 경량화하는 방법도 찾아봐야하겠다.

3. 모델의 예측력이 생각보다 높지 않아 올바른 정답을 출력하는 경우가 많이 없었다. 입력으로 받는 데이터와 학습 데이터와의 차이에서 오는 결과일 수도 있겠다는 생각을 했고, 향후 학습 데이터와 입력 데이터의 분포 차이를 확인하고 전처리를 해주는 MLOps Component를 배우면서 해결해야 하겠다는 생각이 들었다.