지난, 게시글에서는 벡터 디비 초기 세팅 및 사용방법에 대해서 알아봤다.
사실 조금 아쉬웠던 점은 openai ada-2 임베딩 모델을 사용했던 것이였다.
오늘은, 임베딩 과정에서 돈이 들지 않고 정확도가 심하게 떨어지지 않는 허깅페이스 올라간 오픈 모델을 사용하여 임베딩 후 벡터 디비에 저장 및 서치 하는 방법에 대해서 설명하겠다.
초기, 클러스터 생성 부분은 아래의 링크를 참고하세요.
https://lena-train-cfg.tistory.com/88
[챗봇/벡터디비] Weaviate란? 벡터 디비 설정, 저장, 쿼리 테스트하기
벡터 데이터베이스란?대량의 고차원 데이터 저장 및 조회를 위해 벡터 임베딩 및 쿼리 벡터 유사도 비교 기반 신속하게 인덱싱하는 데이터 베이스 어떻게 쓰일까?벡터 데이터베이스는 다차원
lena-train-cfg.tistory.com
데이터 전처리
지난번에 아쉬웠던 부분은 데이터의 전처리이다.
내가 다운받은 공공데이터는 심각할 정도로 정리가 안되어있는 데이터이다.
(괄호 부터 월, 시간등 순서가 없이 프로그램 명에 자유롭게 나열 되어있다.)
전처리 소스가 조금 긴데 정리하자면
1. 타임스태프 값을 ms 단위에서 날짜 및 시간으로 변환하는데 UTC가 아닌 현지 시간으로 변경
2. 가격 정보는 새로운 컬럼으로 빼고 서비스 대상은 불필요한 정보 제거
3. 서비스명에 정규표현식을 활용하여 규칙성 확보(transform_text, extract_and_remove함수)
import pandas as pd
from datetime import datetime
import json
import re
def convert_timestamps(df, columns):
for col in columns:
df[col] = pd.to_datetime(df[col], unit='ms', utc=False, errors='coerce')
return df
def extract_pricing_info(row):
if row['결제방법'] == '유료':
target = row['서비스대상']
# 괄호 안의 내용을 추출하여 키-값 쌍 생성
pricing_info = {}
for item in target.split(', '):
if '(' in item and ')' in item:
key = item.split('(')[0].strip()
value = item.split('(')[1].split(')')[0].strip()
pricing_info[key] = value
# 키-값 쌍을 문자열로 변환
return ', '.join([f"{k}:{v}" for k, v in pricing_info.items()])
else:
return None
# 정규 표현식을 사용한 전처리 함수
def extract_conditions_and_clean_target(target):
# 1. 괄호 안의 내용 추출
conditions = re.findall(r'\((.*?)\)', target)
clean_target = re.sub(r'\(.*?\)', '', target).strip()
clean_target = clean_target.replace(')', ' ').strip()
# 2. / 뒤의 내용 추출
if '/' in clean_target:
parts = clean_target.split('/')
clean_target = parts[0].strip() # / 앞의 내용
condition_from_slash = parts[1].strip() # / 뒤의 내용
if condition_from_slash:
conditions.append(condition_from_slash)
# 조건이 있는 경우만 반환
if conditions:
return clean_target, ', '.join(conditions)
else:
return clean_target, None
def extract_conditions(service_name):
# [] 안의 내용 추출
condition1 = re.findall(r'\[(.*?)\]', service_name)
# () 안의 내용 추출
condition2 = re.findall(r'\((.*?)\)', service_name)
# 서비스명에서 []와 () 안의 내용 모두 제거
clean_service_name = re.sub(r'[\[\]\(\)]', ' ', service_name).strip()
# 조건1과 조건2를 반환
return clean_service_name, ', '.join(condition1) if condition1 else None, ', '.join(
condition2) if condition2 else None
def expand_month_range(match):
start_month = int(match.group(1))
end_month = int(match.group(2))
return ' '.join(f'{month}월' for month in range(start_month, end_month + 1))
def expand_grade_range(match):
start_grade = int(match.group(1))
end_grade = int(match.group(2))
return ' '.join(f'{grade}학년' for grade in range(start_grade, end_grade + 1))
def convert_time_format(match):
hour = match.group(1)
minute = match.group(2)
return f'{hour}시 {minute}분'
def expand_weekdays_range(match):
days = ["월", "화", "수", "목", "금", "토", "일"]
start_day = match.group(1)
end_day = match.group(2)
start_index = days.index(start_day)
end_index = days.index(end_day)
if start_index > end_index:
return ' '.join(days[start_index:end_index + 1])
return ' '.join(days[start_index:end_index + 1])
def convert_to_24_hour_format(match):
period = match.group(1)
start_hour = int(match.group(2))
end_hour = int(match.group(3))
# 오후인 경우, 12시간을 더해 24시간 형식으로 변환
if period == "오후":
start_hour += 12
end_hour += 12
# 변환된 시간을 "오후 14시~16시" 형식으로 반환
return f"{period} {start_hour}시~{end_hour}시"
def expand_time_range(match):
period = match.group(1) # "오전" 또는 "오후"
start_hour = int(match.group(2)) # 시작 시간
end_hour = int(match.group(3)) # 종료 시간
# 12시간제 범위로 변환하지 않고 그대로 사용
expanded_times = [f"{period} {hour}시" for hour in range(start_hour, end_hour + 1)]
return ' '.join(expanded_times)
def expand_weekend_dates(match):
# 그룹에서 날짜 부분을 추출
days = match.group(1)
# 날짜를 '/' 기준으로 분리하고 각 날짜 뒤에 '일'을 추가
expanded_days = ' '.join([f"{day}일" for day in days.split('/')])
return f"주말 {expanded_days}"
def transform_text(text):
# 주말 변환 (주말 -> 토 일)
text = re.sub(r'\b주말\b', '토 일', text)
# 연도 변환 (2024 -> 2024년)
text = re.sub(r'(\b\d{4}\b)', r'\1년', text)
# 날짜 변환 (8/23 -> 8월 23일)
text = re.sub(r'(\b\d{1,2})/(\d{1,2}\b)', r'\1월 \2일', text)
# 날짜 변환 (6.13 -> 6월 13일)
text = re.sub(r'(\b\d{1,2})\.(\d{1,2}\b)', r'\1월 \2일', text)
# 월 범위를 개별 월로 확장 (7~8월 -> 7월 8월)
text = re.sub(r'(\d{1,2})~(\d{1,2})월', expand_month_range, text)
# 주말 날짜 범위를 확장 (예: 주말 17/18 -> 주말 17일 18일)
text = re.sub(r'주말\s*(\d{1,2}/\d{1,2})', expand_weekend_dates, text)
# 학년 범위를 개별 학년으로 확장 (1~3학년 -> 1학년 2학년 3학년)
text = re.sub(r'(\d)~(\d)학년', expand_grade_range, text)
# 시간 형식 변환 (13:30 -> 13시 30분)
text = re.sub(r'(\d{1,2}):(\d{2})', convert_time_format, text)
# 정규 표현식과 re.sub를 사용해 변환
text = re.sub(r'(오전|오후)\s*(\d{1,2})~(\d{1,2})시', convert_to_24_hour_format, text)
# 24시간제에서 12시간제로 변환 (예: "오후 13시~16시" -> "오후 1시~4시")
text = re.sub(r'(오전|오후)\s*(\d{1,2})시~(\d{1,2})시', expand_time_range, text)
# 요일 범위를 개별 요일로 확장 (월~금 -> 월 화 수 목 금)
text = re.sub(r'(월|화|수|목|금|토|일)~(월|화|수|목|금|토|일)', expand_weekdays_range, text)
# 앞에 0이 붙은 월을 수정 (08월 -> 8월)
text = re.sub(r'\b0(\d)월', r'\1월', text)
# " - "와 같은 형식의 특수 문자 제거 (앞뒤 공백 포함)
text = re.sub(r'\s*-\s*', ' ', text)
# 특정 숫자와 점(.) 형식 제거 (예: 6.)
text = re.sub(r'\b\d+\.\s*', ' ', text)
# 기타 특수 문자 제거 (Ⅱ 추가)
text = re.sub(r'\b(in)\b|[\"_<>,:「」\'~ㆍ\*\)\(\|–·〔〕Ⅱ★★①②!?I/&《》“” + ‘ ’\.]', ' ', text)
# 여분의 공백 제거
text = re.sub(r'\s+', ' ', text).strip()
return text
def extract_and_remove(text):
# 월일 추출 (숫자+월 숫자+일 형태만)
month_day = re.findall(r'\b(?:\d{4}년)?\s*(?:\d{1,2}월)?\s*(?:\d{1,2}일)?\b', text)
non_empty_values = [item for item in month_day if item.strip() != '']
if non_empty_values:
df1.at[index, '년월일'] = non_empty_values[0]
text = re.sub(r'\b(?:\d{4}년)?\s*(?:\d{1,2}월)?\s*(?:\d{1,2}일)?\b', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
# 학년 추출
grade = re.findall(r'\d학년', text)
if grade:
df1.at[index, '학년'] = grade[0]
text = re.sub(r'\d학년', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
# 요일 추출 (한글 문자로 구성된 요일만)
days = re.findall(r'\b(?:월|화|수|목|금|토|일)(?:요일)?\b', text)
if days:
df1.at[index, '요일'] = ' '.join(days)
text = re.sub(r'\b(?:월|화|수|목|금|토|일)(?:요일)?\b', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
# 시간 추출 (오전/오후 + 숫자시 숫자분 형태)
time = re.findall(r'((?:오전|오후)?\s*\d{1,2}시(?:\s*\d{1,2}분)?)', text)
if time:
df1.at[index, '시간'] = time[0]
text = re.sub(r'((?:오전|오후)?\s*\d{1,2}시(?:\s*\d{1,2}분)?)', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
# "정원" 뒤에 오는 숫자를 추출
capacity = re.search(r'정원\s*(\d+)명', text)
if capacity:
df1.at[index, '정원'] = capacity.group(1)
# "대기" 또는 "대기자" 뒤에 오는 숫자를 추출
waiting = re.search(r'(대기|대기자)\s*(\d+)명', text)
if waiting:
df1.at[index, '대기'] = waiting.group(2)
text = re.sub(r'정원\s*\d+명', '', text)
text = re.sub(r'(대기|대기자)\s*\d+명', '', text)
text = re.sub(r'\s+', ' ', text).strip()
return text.strip()
input_path = './data/서울시 문화행사 공공서비스예약 정보.json'
with open(input_path, 'r', encoding='utf-8') as file:
data = json.load(file)
df = pd.DataFrame(data['DATA'])
description_mapping = {key.lower(): value for key, value in data['DESCRIPTION'].items()}
df.rename(columns=description_mapping, inplace=True)
df1 = df.copy()
columns_to_convert = ['접수종료일시', '서비스개시종료일시', '접수시작일시', '서비스개시시작일시']
df1 = convert_timestamps(df1, columns_to_convert)
# 전처리1 서비스 대상 ,로 표시 중복은 하나만, 괄호는 ,로 연결 유로 일경우 가격 새로운 컬럼에 기입
df1['서비스대상'], df1['서비스조건'] = zip(*df1['서비스대상'].apply(extract_conditions_and_clean_target))
df1['서비스명'], df1['조건1'], df1['조건2'] = zip(*df1['서비스명'].apply(extract_conditions))
df1['서비스명'] = df1['서비스명'].apply(transform_text)
# 새로운 컬럼들 추가
df1['년월일'] = None
df1['학년'] = None
df1['요일'] = None
df1['시간'] = None
df1['정원'] = None
df1['대기'] = None
# 데이터프레임의 각 행에 대해 처리
for index, row in df1.iterrows():
df1.at[index, '서비스명'] = extract_and_remove(row['서비스명'])
서비스명이 훨씬 간결하고 깔끔해진 것을 볼 수 있다.
Query, Answer 딕셔너리 생성
임베딩 모델을 거치기 전 나의 챗봇은 질문과 답을 하는 데이터로 구성이 되어있다.
벡터 디비에 인설트할때도 아래의 데이터 형식으로 구성이 되어있기때문에 해당 형식으로 질문-답형식으로 만들어줘야한다.
가장 질문이 많이 나올 것 같은 리스트를 추려서 질문-답 데이터를 만들었다.
아래 소스는 데이터 어떻게 만들었는지 참고용이다.
대략 11,168개의 질문-답 데이터가 만들어진다.
import json
import pandas as pd
data = pd.read_csv('./data/preprocess_data.csv')
print(data.columns)
questions_and_answers = []
for ind in range(len(data)):
item = data.iloc[ind]
item = item.fillna("없음")
questions_and_answers.append({
"question": f"{item['서비스명']}은 무료인가요?",
"answer": item['결제방법'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 유로인가요?",
"answer": item['결제방법'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 얼마인가요?",
"answer": item['결제방법'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 지금 예약할 수 있나요?",
"answer": item['서비스상태'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 종료됬나요?",
"answer": item['서비스상태'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 진행하나요?",
"answer": item['서비스상태'],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}은 언제 열리나요?",
"answer": item["서비스개시시작일시"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}의 장소는 어디인가요?",
"answer": item["장소명"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']} 예약 링크는 무엇인가요?",
"answer": item["바로가기URL"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']} 예약은 어디서 하나요?",
"answer": item["바로가기URL"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']} 취소 기간은 언제까지인가요?",
"answer": item["취소기간 기준정보"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}의 참가 대상은 누구인가요?",
"answer": item['서비스대상'],
"category": item["소분류명"]
})
if item['서비스대상'] in ["성인", "제한없음"]:
questions_and_answers.append({
"question": "성인만 참여 가능한 프로그램은 무엇인가요?",
"answer": item["서비스명"],
"category": item["소분류명"]
})
if item['서비스대상'] in ["유아", "제한없음"]:
questions_and_answers.append({
"question": "유아만 참여 가능한 프로그램은 무엇인가요?",
"answer": item["서비스명"],
"category": item["소분류명"]
})
if item['서비스대상'] in ["초등학생", "제한없음"]:
questions_and_answers.append({
"question": "초등학생만 참여 가능한 프로그램은 무엇인가요?",
"answer": item["서비스명"],
"category": item["소분류명"]
})
if item['서비스대상'] in ["가족", "제한없음"]:
questions_and_answers.append({
"question": "가족이 참여 가능한 프로그램은 무엇인가요?",
"answer": item["서비스명"],
"category": item["소분류명"]
})
if item['서비스대상'] in ["어린이", "제한없음"]:
questions_and_answers.append({
"question": "어린이가 참여 가능한 프로그램은 무엇인가요?",
"answer": item["서비스명"],
"category": item["소분류명"]
})
questions_and_answers.append({
"question": f"{item['서비스명']}의 장소가 어디인가요?",
"answer": item["장소명"],
"category": item["소분류명"]
})
file_path = './data/question_data.json'
with open(file_path, 'w', encoding='utf-8') as json_file:
json.dump(questions_and_answers, json_file, ensure_ascii=False, indent=4)임베딩 모델 선정 및 적용
내가 사용한 오픈 모델은 jhgan/ko-sroberta-nli 이다.
sentence transformer 모델이고 768 벡터 차원으로 맵핑한다.
11,168개의 질문 데이터 임베딩 모델 인코딩하는데 대략 3.45분 소요된다.
https://huggingface.co/jhgan/ko-sroberta-nli
jhgan/ko-sroberta-nli · Hugging Face
ko-sroberta-nli This is a sentence-transformers model: It maps sentences & paragraphs to a 768 dimensional dense vector space and can be used for tasks like clustering or semantic search. Usage (Sentence-Transformers) Using this model becomes easy when you
huggingface.co
from sentence_transformers import SentenceTransformer
sentences = ["안녕하세요?", "한국어 문장 임베딩을 위한 버트 모델입니다."]
model = SentenceTransformer('jhgan/ko-sroberta-nli')
embeddings = model.encode(sentences)
print(embeddings)
사용방법은 모델 선언해서 내문장 리스트로 담은다음 modle.encode("내 문장 리스트")를 통과시키면 된다.
아래 소스는 벡터 디비에 임베딩 모델 대체해서 데이터 인설트하는 것이다.
작업 순서
1. 내 질문 데이터 읽어오기
2. 임베딩 모델 선언
3. weaviate 클라이언트 연결, 컬렉션 생성, 벡터라이저는 None
* text2vec-huggingface 모듈이 있는데 이걸 사용해서 api 통신하는거 유료인지 확인 필요, 무료면 해당 모듈 사용
4. 벡터 변환은 question만 한정했고, 임베딩 모델은 numpy array로 반환되기 때문에 to_list()로 리스트 형식으로 바꿔줘야함
5. 데이터 인설트
import json
import time
import weaviate
import weaviate.classes as wvc
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from sentence_transformers import SentenceTransformer
with open('./data/question_data.json', 'r', encoding='utf-8') as json_file:
data = json.load(json_file)
model = SentenceTransformer('jhgan/ko-sroberta-nli')
# 클라이언트 연결, v4
client = weaviate.connect_to_weaviate_cloud(
cluster_url = "", # Replace with your Weaviate endpoint
auth_credentials=wvc.init.Auth.api_key(api_key="") # Replace with your Weaviate instance API key
)
# 데이터, 컬렉션 객체 구성, 내 데이터로 변경
questions = client.collections.create(
name="seoul_public_programs_myvector",
vectorizer_config=wvc.config.Configure.Vectorizer.none())
questions = client.collections.get("seoul_public_programs_myvector")
question_objs = list()
for i, d in enumerate(data):
question_objs.append(wvc.data.DataObject(
properties={
"answer": d["answer"],
"question": d["question"],
"category": d["category"],
},
vector = model.encode(d['question']).tolist()
))
questions.data.insert_many(question_objs)
데이터가 잘 들어간것으로 확인 된다.
테스트
questions.query.near_vector를 사용해서 벡터 검색을 진행했다.
여기서도 주의할 것이 np.array에서 tolist()로 리스트 형식으로 변경해야하는 점이다.
user_input = model.encode("나의 답십리 프로그램은 얼마?").tolist()
response = questions.query.near_vector(
near_vector=user_input,
limit=1,
return_metadata=wvc.query.MetadataQuery(certainty=True)
)
user_input = model.encode("느낌 있는 박물관 안내 장소는 어디?").tolist()
response = questions.query.near_vector(
near_vector=user_input,
limit=1,
return_metadata=wvc.query.MetadataQuery(certainty=True)
)
개선할 점
1. 자료 조사하다가 multi-hop 태스크가 있다는 걸 확인했는데, 실제 사람들이 질문할때는 이러한 복잡한 질문을 할 수 있다는 것이다. 그래서 그래프 형식으로 데이터를 구축했던거 같은데 더 찾아보고 적용하면 좋을 것 같다.
* Multi-hop 질문이란 여러 문서에 대한 복합적인 추론을 통해 응답을 찾을 수 있는 복잡한 질문을 뜻한다.
2. 쓸만한 LLM 모델을 찾고 학습 시키는 것이다.
embedding 모델은 우선 여기서 픽스하고, 추가적으로 문제 생기면 바꿀 거나 추가 학습 시킬건데, 이제는 답변을 생성해주는 LLM 모델을 찾아서 추가 학습 하거나 사용해야한다.
참고링크
https://weaviate.io/developers/weaviate/starter-guides/custom-vectors
Bring your own vectors | Weaviate
Weaviate is a vector database. Vector databases store data objects and vectors that represent those objects. The vector representation is also called an "embedding."
weaviate.io
'Dev 개발할레나' 카테고리의 다른 글
| [챗봇/벡터디비] Weaviate란? 벡터 디비 설정, 저장, 쿼리 테스트하기 (1) | 2024.08.28 |
|---|---|
| [챗봇] GPT API를 활용하여 서울시 문화행사 공공프로그램 챗봇 만들기 (0) | 2024.08.20 |
| [Docker] 사용법, 명령어 총 정리, CheatSheet (2) | 2024.08.07 |
