import pandas as pd
from openai import OpenAI
import numpy as np
from typing import List # 타입 힌트를 위한 라이브러리
from scipy import spatial # 코사인 유사도 계산을 위한 라이브러리
client = OpenAI()
############################################################################
# 질문과 학습 데이터를 비교해 컨텍스트를 만드는 함수
# - 사용자의 질문에 대한 적절한 답변을 생성하기 위해 학습 데이터와 질문을
# 비교하여 가장 유사한 문장을 찾아 해당 문장을 기반으로 컨텍스트를 생성
############################################################################
def create_context(question, df, max_len=1800):
# 질문을 벡터화 하기 model = 'text-embedding-3-small' / 'text-embedding-ada-002' 사용
question_embedding = (
client.embeddings.create(input=[question], model="text-embedding-ada-002")
.data[0]
.embedding
)
# 질문과 학습 데이터와 비교하여 코사인 유사도를 계산하고 'distances' 열에 유사도를 저장
df["distances"] = distances_from_embeddings(
question_embedding,
df["embeddings"].apply(eval).apply(np.array).values,
distance_metric="cosine",
)
returns = [] # 결과를 저장할 리스트
cur_len = 0 # 컨텍스트의 현재 길이
# 학습 데이터를 유사도 순으로 정렬하고 토큰 개수 한도까지 컨텍스트에 추가
for _, row in df.sort_values("distances", ascending=True).iterrows():
cur_len += row["n_tokens"] + 4 # 텍스트 길이를 현재 길이에 더하기
if cur_len > max_len: # 텍스트가 너무 길면 루프 종료
break
returns.append(row["text"]) # 컨텍스트 목록에 텍스트 추가하기
return "\n\n###\n\n".join(returns) # 컨텍스트 목록을 반환
########################################################
# 문맥에 따라 질문에 답하는 기능
########################################################
def answer_question(question, conversation_history):
df = pd.read_csv("./02_hotel_chatbot/embeddings2.csv") # 임베딩된 데이터를 불러옴
context = create_context(
question, df, max_len=200
) # 질문과 학습 데이터를 비교해 컨텍스트 생성
# OpenAI에 전달할 프롬프트 생성
prompt = f"당신은 어느 호텔 직원입니다. 문맥에 따라 고객의 질문에 정중하게 대답해 주십시오. 컨텍스트가 질문에 대답할 수 없는 경우 '모르겠습니다'라고 대답하세요.\n\n컨텍스트: {context}\n\n---\n\n질문: {question}\n답변:"
conversation_history.append({"role": "user", "content": prompt})
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo", # GPT-3.5 모델 사용
messages=conversation_history, # 대화 기록을 전달
temperature=1, # 낮은 온도는 더 정확한 답변을 생성
)
return response.choices[0].message.content.strip() # 답변 반환
except Exception as e:
print(e)
return "" # 예외가 발생하면 빈 문자열 반환
def distances_from_embeddings(
query_embedding: List[float],
embeddings: List[List[float]],
distance_metric="cosine",
) -> List[List]:
# 쿼리 임베딩과 임베딩 목록 사이의 거리를 반환합니다.
distance_metrics = {
"cosine": spatial.distance.cosine, # 코사인 유사도
"L1": spatial.distance.cityblock, # L1 거리
"L2": spatial.distance.euclidean, # L2 거리
"Linf": spatial.distance.chebyshev, # L무한 거리
}
distances = [
distance_metrics[distance_metric](query_embedding, embedding) # 쿼리 임베딩과 임베딩 목록 사이의 거리 계산
for embedding in embeddings # 모든 임베딩에 대해 거리 계산
]
return distances # 거리 목록 반환