Project HowAbout RAG API - Outro: TPM Vector Store 개선

이전 포스트에서 TPM 문제에 대한 해결책을 실제로 구현해 보고 비교해 봤다.

 

먼저 Vector Store를 개선해보려고 한다.

1. Vector Store 생성 파트
   • 사용 토큰 수 조정
   • Batch 적용
   • Embedding Model 변경
2. Query Translation & Generation 파트
   • Tenacity 혹은 Backoff 적용

 

문제 개선에 대한 동기와 그 선정과정이 궁금하다면 아래의 링크로 가면 된다.

Project HowAbout RAG API - Outro: Optimization - TPM 문제 정의

 

 

사용 토큰 수 조정

먼저 토큰이 정확히 얼마나 어디서 쓰이는지 확인하기 위해 진행했다. 그래서 먼저 tiktoken을 활용해서 사용되는 토큰의 규모를 파악했다. 또한 기본적으로 시도할 수 있는 부분이 적절한 토큰수를 찾는 거라고 판단했다. (토크가 무조건 많다고 해서 좋은 게 아니니깐)

 

특히, LangChain의 Document의 Metadata 부분을 활용한다면 토큰수도 줄이면서 정확성과 효율성을 더 높일 수 있다고 판단됐다.

 

그래서 아래와 같은 기존의 코드를 변환하였다.

Meta Data 활용 전

test_docuemtns_no_meta = []
import uuid
from uuid import uuid4
from langchain.schema import Document

for row in data:
  text = f"{row['ad_gu']} [SEP] {row['ad_dong']} [SEP] {row['address']} [SEP] {row['location']} [SEP] " \
               f"{row['description']} [SEP] {row['rating']} [SEP] {row['share_link']} [SEP] " \
               f"{' '.join(row['reviews'])} [SEP] {row['info']}"
  clean_text = text.replace("\n", " ")
  test_docuemtns_no_meta.append(Document(page_content=clean_text))

# Only for Faiss -> comment it out for Pinecone
uuids = [str(uuid4()) for _ in range(len(test_docuemtns_no_meta))]

 

Meta Data 활용 후

test_documents_meta = []
import uuid
from uuid import uuid4
from langchain.schema import Document

for row in data:
  # Text
  text = f"{row['address']} [SEP] {row['location']} [SEP] " \
  f"{row['description']} [SEP] {row['rating']} [SEP] {row['share_link']} [SEP] " \
  f"{' '.join(row['reviews'])} [SEP] {row['info']}"
  # Metadata
  metadata = {
      "ad_gu": row['ad_gu'],
      "ad_dong": row['ad_dong'],
  }
  clean_text = text.replace("\n", " ")
  test_documents_meta.append(Document(
      page_content=clean_text,
      metadata=metadata
  ))

# Only for Faiss -> comment it out for Pinecone
uuids = [str(uuid4()) for _ in range(len(test_documents_meta))]

 

그리고 이를 바탕으로 나온 결과는 아래와 같다. 

 

물론 예상과는 다르게 Meta Data로 지정한 부분이 생각보다 양이 적어 토큰의 수가 크게 줄지는 않았다 하지만 그래도 Meta Data를 사용하는 게 정확성과 효율성 면에서 높다고 판단된다. 

 

물론 프로젝트에서 Meta Data를 활용하는 구조가 명확해야 하니 무조건 좋다고는 할 수 없다.

 

그리고 이미 Concatenation과 데이터 Preprocessing이 이미 전화하여 토큰 수 추가적으로 줄이는 것은 멈추기로 했다. 

 

다음으로 Batch 적용이다.

 

Batch API를 먼저 사용해보려고 했으나 Batch API의 경우 Request를 나눠서 하는 것이기 때문에 적용하기 Vector Store를 적용하기에는 무리가 있었다. 물론 추후에 다른 방식에서 시도해 볼 예정이다. 그래서 Batch API 대신에 Batch의 개념을 적용하기로 결정했다. 또한 이 부분에서 TPM뿐만이 아니라 비용을 낮추기 위해 append와 같이 매번 새로 만드는 게 아닌 기존의 Vector Store에 추가하는 방식을 구현했다.

 

먼저 위의 토큰 수를 바탕으로 최적의 Batch 사이즈를 계산해 봤다. 그래서 아래와 같은 결과가 나왔다. 

 

'text-embedding-3-mini'(비용 문제로 large에서 전환하지만 사용제약은 동일하다.)의 경우 분당 100만개 이기 때문에 위와 같이 안정적인 선에서 분당 72만 개로 숫자를 낮춘 이후에 총 31개의 Batch로 나누어 진행했다.

 

그래서 Batch를 아래의 상황에 맞춰 추가를 진행하였다.

1. 새로운 데이터셋에 대한 중복검사가 필요 없는 경우: Merge
2. 새로운 데이터셋에 중복검사가 필요한 경우: append 및 중복 검사

먼저 Merge의 경우 아래처럼 간단하게 시도해 봤다.

Vector Store: Merging

 

그리고 중복검사를 동반한 Append의 경우 아래처럼 진행했다.

Vector Store: Append & Duplication Check

 

기본적으로 기존 Vector Store의 Document의 page_content를 Set으로 담았다. 이는 FAISS의 경우 page_content와 id를 직접 입력해야 하닌 다른 Vector Store의 경우 이와 같은 부분이 다르기 때문에 page_content를 직접 비교했다.

 

이런 과정이 수고스러울 수 있다. 왜냐하면 단순히 sleep만 추가해도 TPM 문제의 경우 쉽게 해결되기 때문이다.(실시간이 아닐 경우와 Vector Store 생성과정에서) 하지만 반대로 Open AI를 사용하는 비용 Vector Store가 생성되는 과정의 오류시 처음부터 다시 시작해야 하는 점을 생각했을 때 생각해 보면 위와 같은 안정성은 중요하다고 생각한다. 여기서 추가로 아래와 같이 try-catch를 활용하여 문제 발생 시 바로 저장을 하는 로직까지 추가하였다.(당연한 건데 까먹었었다.)

 

from time import sleep

# Batch 순회
existing_texts = {doc.page_content for doc in vector_store.docstore._dict.values()}
for i in range(len(batches)):
  print(f"Batch {i+1}/{len(batches)}")
  checked_batch = []
  checked_batch_id = []
  dup_count = 0
  # Batch duplication check with original dataset or saved data
  for j in range(len(batches[i])):
    if batches[i][j].page_content not in existing_texts:
      checked_batch.append(batches[i][j])
      checked_batch_id.append(batches_id[i][j])
    else: 
      dup_count+=1
  print("Duplicate Found: {}".format(dup_count))
  print("New Embeddings: {}".format(len(checked_batch)))
  
  if (len(checked_batch) == 0):
    continue

  try:
    vector_store.add_documents(documents=checked_batch, ids=checked_batch_id)
    print("Current Vector Store Size: ", vector_store.index.ntotal)
    sleep(5)
  except Exception as e:
    vector_store.save_local(faiss_index_path)
    print(e)

 

위의 내용들이 적용된 전체코드는 아래의 깃헙 저장소에서 확인할 수 있다. 

rag-optimization/vector_store_optimization

RAG/rag-optimization/vector_store_optimization.ipynb at main · jwywoo/RAG

 

 

다음으로 Tenacity와 Python의 Backoff library를 적용해 보겠다.