들어가며
RAG(Retrieval-Augmented Generation) 기반 서치 챗봇은 기업이 사용자의 질문에 대한 대답을 생성할 때 검색 결과에서 정보를 가져와 생성하는 방식입니다. 정확도 향상, 다양성 및 품질, 실시간 업데이트, 자동화 및 효율성, 고객 만족도 향상과 같은 이점으로 인해 기업들은 RAG 기반 서치 챗봇을 만들어 사용자 경험을 향상시키고 비즈니스 성과를 극대화하려고 합니다.
이전 글에서는 Pinecone이라는 벡터 DB에 데이터 셋을 임베딩하여 적재하고 챗봇에 DB 정보를 기반으로 한 질의응답을 할 수 있는 모델을 만들어 보았습니다. 답변에 대한 퀄리티는 증가했지만, 아쉬운 점은 문의를 남기고 답변을 받기까지 시간이 상당히 오래 걸렸다는 것인데요. 이런 레이턴시를 줄이기 위한 방법으로 캐싱 기술을 사용해 보려고 합니다.
이번 글에서는 캐싱 기술이란 무엇인지 알아보고, 캐시 스토리지 SaaS를 이용해 챗봇의 응답 속도를 개선한 방법에 대해 소개하겠습니다.
캐싱이란
캐싱은 이전에 검색하거나 계산한 데이터를 저장하는 기술입니다. LLM에서는 입력 요청의 임베딩에 대한 LLM 응답을 캐시하고, 다음 요청에서 의미상 유사한 요청이 들어오면 캐시된 응답을 제공할 수 있습니다. 캐싱 기술을 활용하면 응답시간을 줄임과 동시에, LLM을 요청하는 횟수를 줄여서 API 사용 비용까지 절감할 수 있습니다.
이번 챗봇에서 캐싱을 위해 제가 선택한 SaaS는 오픈소스 Redis를 클라우드 형태로 제공하는 Redis Insight입니다.
1. 캐시 스토리지 생성하기
먼저 Redis 사이트에 접속하여 캐시 스토리지를 생성해 줍니다.
2. 질문 내용을 저장할 수 있도록 코드 추가하기
기존 코드에서 질문 내용을 저장할 수 있도록 코드를 추가해 보았는데요. 이전 단계에서 redis 스토리지 생성 당시 만들어진 AWS host 정보와 계정 비밀번호를 불러옵니다.
그리고 사용자의 질문에 챗봇이 답변하는 get_response 함수에 매번 질문과 답변이 발생할 때마다 문장을 임베딩하여 redis 스토리지에 저장할 수 있도록 코드를 추가해 줍니다.
import redis
redis_client = redis.Redis(host='host', port=12519, password='pw')
def get_response(prompt, input_time):
response = openai.chat.completions.create(
model="gpt-3.5-turbo",
messages = [
{"role": "system", "content": "You are a Korean shopping assistant. Provide the answer in Korean"},
{"role": "user", "content": f"Customer Question: {prompt}"},
{"role": "assistant", "content": f"Information from the database:{query_top3_vector(prompt)}"},
{"role": "user", "content": "Please provide your insights and recommendations about the item based on the information from the database. If you don't have the relevant information."}
])
ai_response = response.choices[0].message.content
duration = datetime.now() - input_time
answer = (f"Duration: {duration} \nBot Answer: {ai_response}")
prompt_embedding = embed_query(prompt)
redis_client.set(prompt, ai_response)
redis_client.set(f"{prompt}_embedding", str(prompt_embedding))
return answer
이후 Redis Insight를 노트북에 설치하여 위에 생성해 둔 스토리지 정보를 불러올 수 있으며, 손쉽게 만들어 둔 인스턴스와 캐시된 정보들을 GUI를 통해 확인하실 수 있습니다. 아래와 같이 Part1에서 질문했던 내용이 저장되어 있는 걸 확인할 수 있습니다.
3. 질문 유사도 계산하기
질문과 AI의 답변은 저장이 잘 되었는데, 그렇다면 이후에 질문을 했을 때 해당 질문이 스토리지에 저장된 내용과 비슷하다는 것은 어떻게 인지시킬 수 있을까요?
그 방법으로 먼저 cosine 유사도를 계산할 수 있는 cosine_similarity 함수를 만들어 새로운 질문을 받았을 때 캐시된 기존의 질문과 유사도를 수치화하였습니다.
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
그리고 semantic_cache 함수를 만들어 기본적으로 모든 질문에 대한 답변을 하기 전 semantic_cache 함수를 거치면서 cosine 유사도를 확인하게 됩니다. 새로운 질문을 받았을 때 이전의 질문 리스트를 돌면서 유사도가 가장 높은 하나의 질문을 찾아낸 뒤, 해당 질문의 유사도가 특정 점수 이상을 만족하면 캐시된 답변을, 그렇지 못할 경우 AI에 질문을 하는 방식으로 구성하였습니다.
예시의 경우에는 유사도 0.7 이상이라는 기준값을 사용해 보았습니다.
def semantic_cache(prompt):
input_time = datetime.now()
prompt_embedding = embed_query(prompt)
max_similarity = -1
most_similar_question = None
for key in redis_client.keys("*_embedding"):
cached_embedding = eval(redis_client.get(key).decode('utf-8'))
similarity = cosine_similarity(prompt_embedding, cached_embedding)
if similarity > max_similarity:
max_similarity = similarity
most_similar_question = key.decode('utf-8').replace("_embedding", "")
if most_similar_question and max_similarity >= 0.7:
cached_response = redis_client.get(most_similar_question).decode('utf-8')
duration = datetime.now() - input_time
return (f"Duration: {duration} <br>Cached Answer: {cached_response}")
else:
# If no similar question is found, generate a new response
return get_response(prompt, input_time)
4. 챗봇과 대화해보기
다시 웹으로 돌아와 기존에 문의했던 질문과 비슷한 질문들을 던져보겠습니다.
최초로 질문이 캐시된 이후 문맥상 비슷한 질문들을 했을 때 캐시된 질문에 대한 답변을 가져오고 있는데요. 기존에 AI가 생성한 답변의 경우 약 10초의 시간이 걸렸다면, 캐시 후에는 약 3초로 리스폰스 시간이 줄어든 것을 볼 수 있습니다.
그런데 몇 가지 유사한 질문을 계속하던 중 문제점을 발견했습니다. 다른 제품을 검색했을 때도 이를 문맥상 비슷한 질문으로 인식하여 캐시된 답변을 가져온다는 것이었습니다. 아래 보시는 것처럼 카시오라는 제품 대신 태그호이어라는 시계 브랜드를 검색했으나 유사한 것으로 판단한 것인지 기존처럼 카시오 제품에 대한 캐시된 답변을 주고 있습니다.
답변 속도를 개선하기 위해 캐싱 기법을 사용했더니 오히려 부정확한 답변을 받는 결과를 초래한 것 같습니다. 그렇다면 다른 제품을 추천해달라고 했을 때 이를 유사한 답변으로 인식하지 않도록 방법을 바꿔봐야 할 것 같은데요. 이에 대한 방안을 Part 3에 이어서 소개하도록 하겠습니다.
여기까지 궁금한 사항이나 커피챗을 원하시면 언제든지 아래 메일로 연락 부탁드립니다 : )
