[챗봇] GPT API를 활용하여 서울시 문화행사 공공프로그램 챗봇 만들기

최근에 진행하고 있는 토요일 스터디에 챗봇을 개발하자고 의견을 내었다.
감사하게도 다들 좋아해주셨고 다들 눈에 흥미가 가득 생겼다. 10월 쯤에 나가려고 하신 분도 재밌을거 같다며 더 오래할 수도 있다고 말해주시기 까지 했다.

 

나는 빠르게 구현 후 비용이 안들게 대체하는 방법을 택했다. 그래서 GPT API를 먼저 사용했다.

이후에 계속 업로드 하겠지만 GPT까진 아니더라도 GPT의 80%정도만 되도 좋을 것 같다. 

 

그래서,  어떤 챗봇을 만들껀데?

우선 내가 데이터가 있어야한다. 바로 서울 열린데이터 광장에 들어갔다.

다양한 데이터가 있었는데 "서울시 문화행사 공공서비스예약 정보" 데이터가 눈에 들어왔다.

 

공공서비스를 알려주는 챗봇을 만들어보자!

 

링크: https://data.seoul.go.kr/dataList/OA-2269/S/1/datasetView.do

열린데이터광장 메인

---

1. 데이터를 다운 받아준다.

- 처음에 csv로 다운받았는데 인코딩 에러가 나서 좀더 익숙한 데이터 형태인 json으로 다운 받았다.

데이터 형태는 이러하고 "svcstatnm"을 보면 "접수종료"라고 되어있다.

내가 사용할 데이터만 따로 추출해야한다. 그외에도 전처리 할 사항이 많았다.

import pandas as pd
import json
import os
import openai
from config import *

def get_embedding(text):
    response = client.embeddings.create(
        input=text,
        model='text-embedding-ada-002'
    )
    print(response)
    return response.data[0].embedding
def convert_to_text_format(row):
    policy_name = row['서비스명']
    department = "서울시청"
    program_desc = f"{row['장소명']}에서는 {row['서비스명']} 프로그램을 제공합니다. "
    target_audience = f"이 프로그램은 {row['이용대상']}을 대상으로 하며, 매일 {row['최소시간']}부터 {row['최대시간']}까지 진행됩니다. "
    additional_info = f"이 프로그램은 {row['지역명']}에서 운영됩니다. 참가 신청은 {row['장소명']}의 공식 웹사이트에서 가능합니다. "
    url_info = f"자세한 내용은 {row['서비스 URL']} 에서 확인할 수 있으며, 추가 문의는 해당 웹사이트를 참조하시기 바랍니다."

    return f"프로그램 이름: {policy_name}\n{department}에서는 {program_desc}{target_audience}{additional_info}{url_info}"

def Main_preprocess(input_path, output_csv, output_txt, save_df, save_txt):
    with open(input_path, 'r', encoding='utf-8') as file:
        data = json.load(file)

    need_data = []
    for dt in data["DATA"]:
        if dt['svcstatnm'] == "접수중":
            row = {
                'usetgtinfo': dt.get('usetgtinfo'),
                'v_max': dt.get('v_max'),
                'v_min': dt.get('v_min'),
                'minclassnm': dt.get('minclassnm'),
                'placenm': dt.get('placenm'),
                'svcurl': dt.get('svcurl', ''),  # 존재하지 않으면 빈 문자열
                'svcnm': dt.get('svcnm', ''),  # 존재하지 않으면 빈 문자열
                'areanm': dt.get('areanm', '')  # 존재하지 않으면 빈 문자열
            }
            need_data.append(row)

    df = pd.DataFrame(need_data)

    df = df.rename(columns={
        'v_max': '최대시간',
        'v_min': '최소시간',
        'usetgtinfo': '이용대상',
        'minclassnm': '분류명',
        'placenm': '장소명',
        'svcurl': '서비스 URL',
        'svcnm': '서비스명',
        'areanm': '지역명'
    })

    if save_df:
        df.to_csv(output_csv, index=False, encoding='utf-8-sig')

    text_data = df.apply(convert_to_text_format, axis=1)

    if save_txt:
        for index, text in text_data.items():
            file_name = f"{index}.txt"
            file_path = os.path.join(output_txt, file_name)
            with open(file_path, "w", encoding="utf-8") as file:
                file.write(text)

 

필요한 컬럼만 가져와 컬럼명 바꾸고 csv로 저장한다.

그리고 아래와 같은 형태로 데이터를 만들어 개별 txt파일로 저장했다.

    policy_name = row['서비스명']
    department = "서울시청"
    program_desc = f"{row['장소명']}에서는 {row['서비스명']} 프로그램을 제공합니다. "
    target_audience = f"이 프로그램은 {row['이용대상']}을 대상으로 하며, 매일 {row['최소시간']}부터 {row['최대시간']}까지 진행됩니다. "
    additional_info = f"이 프로그램은 {row['지역명']}에서 운영됩니다. 참가 신청은 {row['장소명']}의 공식 웹사이트에서 가능합니다. "
    url_info = f"자세한 내용은 {row['서비스 URL']} 에서 확인할 수 있으며, 추가 문의는 해당 웹사이트를 참조하시기 바랍니다."

폴더 구성

텍스트 파일 내부

위와 같은 형식으로 데이터가 저장이 되면 이제 임베딩을 진행하면 된다.

 

임베딩은 openai의 'text-embedding-ada-002' 모델을 사용하여 진행하였다. 시간은 10분 내로 소요되었다.
임베딩 값은 하나의 csv 파일에 txt파일 내부의 글과 임베딩 값이 하나의 로우에 담기도록하고 저장했다.

이제 데이터는 준비되었다. 이걸 가지고 쿼리에 따라 매칭시켜서 UI에 표시해주면 된다. 

---

import ast
from numpy import dot
from numpy.linalg import norm
from data_preprocess import *
import numpy as np


import streamlit as st
from streamlit_chat import message


file_path = './embedding_data.csv'

if os.path.isfile(file_path):
    print(f"{file_path.split('/')[-1]} 파일이 존재합니다.")
    df = pd.read_csv(file_path)
    df['embedding'] = df['embedding'].apply(ast.literal_eval) # ast.literal_eval 원래의 형변환으로 바꿔줌 str -> float

def cos_sim(A, B):
    return dot(A, B)/(norm(A)*norm(B))

def return_answer_candidate(df, query):
    query_embedding = get_embedding(query)

    df["similarity"] = df.embedding.apply(lambda x: cos_sim(np.array(x), np.array(query_embedding)))
    top_three_doc = df.sort_values("similarity", ascending=False).head(3)
    return top_three_doc

def create_prompt(df, query):
    result = return_answer_candidate(df, query)
    system_role = f"""You are an artificial intelligence language model named "공공 프로그램 도우미" that specializes in summarizing \
    and answering documents about Seoul's public programs.
    You need to take a given document and return a very detailed summary of the document in the query language.
    Here are the document: 
            doc 1 :{str(result.iloc[0]['text'])}
            doc 2 :{str(result.iloc[1]['text'])}
            doc 3 :{str(result.iloc[2]['text'])}
    You must return in Korean. Return a accurate answer based on the document.
    """
    user_content = f"""User question: "{str(query)}". """

    messages = [
        {"role": "system", "content": system_role},
        {"role": "user", "content": user_content}
    ]
    return messages

def generate_response(messages):
    result = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=0.4,
        max_tokens=500)
    print(result.choices[0].message.content)
    return result.choices[0].message.content


# user_input = '청소년이 참가할 수 있는 서울시 공공 프로그램을 알려줘'
# prompt = create_prompt(df, user_input)
# chatbot_response = generate_response(prompt)

# 화면에 보여주기 위해 챗봇의 답변을 저장할 공간 할당
if 'generated' not in st.session_state:
    st.session_state['generated'] = []

# 화면에 보여주기 위해 사용자의 답변을 저장할 공간 할당
if 'past' not in st.session_state:
    st.session_state['past'] = []

# 사용자의 입력이 들어오면 user_input에 저장하고 Send 버튼을 클릭하면
# submitted의 값이 True로 변환.
with st.form('form', clear_on_submit=True):
    user_input = st.text_input('서울시에 어떤 공공 프로그램이 궁금한가요?', '', key='input')
    submitted = st.form_submit_button('Send')

# submitted의 값이 True면 챗봇이 답변을 하기 시작
if submitted and user_input:
    # 프롬프트 생성
    prompt = create_prompt(df, user_input)
    # 생성한 프롬프트를 기반으로 챗봇 답변을 생성
    chatbot_response = generate_response(prompt)
    # 화면에 보여주기 위해 사용자의 질문과 챗봇의 답변을 각각 저장
    st.session_state['past'].append(user_input)
    st.session_state['generated'].append(chatbot_response)

# 사용자의 질문과 챗봇의 답변을 순차적으로 화면에 출력
if st.session_state['generated']:
    for i in reversed(range(len(st.session_state['generated']))):
        message(st.session_state['past'][i], is_user=True, key=str(i) + '_user')
        message(st.session_state['generated'][i], key=str(i))

 

빠르게 작업한 거라 아직 부족한게 많다. 기능별로 차근차근 수정해나가면 되니까 걱정은 없다.

 

먼저 임베딩한 데이터를 읽어서 float형태로 바꿔준다.

그리고 코사인 유사도로 유사한 벡터를 가져와서 gpt 3.5-turbo에 넣은 후 응답이 자연스럽게 만들었다. 

 

streamlit을 사용하여 화면에 보여지는 페이지는 단순하게 진행했다. 답변은 누적해서 쌓이게 했다. 

 

해당 소스는 streamlit run main.py로 하면 돌아간다. 

스터디원들에게 공유할땐 ngrok을 사용해서 8080포트로 연결 후 링크를 공유하고 테스트 해봤다. 

다들 좋아해주셨다.

 

다만, 아쉬운 점은 데이터를 더 다양하게 가져와서 벡터 디비에 저장할것이고,

한국어에 성능이 우수하고 상업적으로 사용가능한 임베딩 모델로 교체한 다음,

생성 모델을 학습 시켜서 gpt로 부터 독립적으로 챗봇을 확장하는 것이 목표기 때문에 지금은 1사이클 테스트 정도로 봐주면 좋을 것 같다.