김태영님의 "문서로더부터 벡터스토어까지 (랭체인없이) 쌩 파이썬으로 만들어보기" 코드 공유

본 포스팅에 사용된 코드는 (주)인공지능팩토리 대표이사이자 마이크로소프트 지역 디렉터(Microsoft Regional Director)이신 김태영님의 “[랭체인러닝데이] 문서로더부터 벡터스토어까지 (랭체인없이) 쌩 파이썬으로 만들어보기” 유튜브 강의 영상을 바탕으로 작성되었습니다. 해당 코드에 대한 모든 권리는 김태영님께 있음을 밝힙니다.

강의 영상에서는 코드를 직접 공유하지 않으셨기에, 본 포스팅에서는 강의 내용을 토대로 직접 코드를 작성하여 정리해 보았습니다. 코드 작성 과정에서 강의 내용을 최대한 반영하고자 노력하였으나, 일부 변형 또는 오류가 있을 수 있습니다.

이 강의에서 다루는 내용은 랭체인(LangChain)에서 RAG(Retrieval-Augmented Generation)을 위해 사용하는 각 컴포넌트들이 어떻게 구현되었는지를 보여줍니다. 랭체인은 언어 모델과 외부 데이터를 연결하여 더 강력한 자연어 처리 애플리케이션을 만들 수 있게 해주는 프레임워크이며, RAG는 질의에 답할 때 외부 데이터를 활용하여 언어 모델의 성능을 향상시키는 기술입니다. 이 강의는 이러한 최신 기술들의 내부 동작을 이해하는 데 큰 도움이 될 것입니다.

원본 코드와 더 자세한 설명은 김태영님의 유튜브 강의를 참고해 주시기 바랍니다.

필요 라이브러리 설치 (OpenAI, TQDM, ChromaDB, Tiktoken, Sentence Transformers)

pip install openai tqdm chromadb tiktoken sentence-transformers

RAG를 위한 연설문 텍스트 데이터 다운로드 (State of the Union)

import urllib.request

urllib.request.urlretrieve(
    "https://raw.githubusercontent.com/hwchase17/chat-your-data/master/state_of_the_union.txt",
    filename="state_of_the_union.txt"
)

SimpleTextLoader 구현해보기

class SimpleTextLoader:

  def __init__(self, file_path):
    self.file_path = file_path

  def load(self):
    text = ''
    with open(self.file_path, 'r', encoding='utf-8') as file:
      text = file.read()
    return text

SimpleCharacterTextSplitter 구현해보기

from tqdm import tqdm
class SimpleCharacterTextSplitter:

  def __init__(self, chunk_size, chunk_overlap, separator_pattern='\n\n'):
    self.chunk_size = chunk_size
    self.chunk_overlap = chunk_overlap
    self.separator_pattern = separator_pattern

  def split_documents(self, documents):

    # 개행 두 개(\n\n)만큼으로 짜르는데, 이걸 청크라고 부르지 않고, 이걸 다시 청크 사이즈만큼 합친다.
    splits = documents.split(self.separator_pattern)

    chunks = []
    current_chunk = splits[0]

    for split in tqdm(splits[1:], desc="splitting..."):

      if len(current_chunk) + len(split) + len(self.separator_pattern) > self.chunk_size:
        chunks.append(current_chunk.strip())
        current_chunk = split
      else:
        current_chunk += self.separator_pattern
        current_chunk += split

    if current_chunk:
      chunks.append(current_chunk.strip())

    return chunks

SimpleOpenAIEmbeddings 구현해보기

from openai import OpenAI

class SimpleOpenAIEmbeddings:

  def embed_query(self, text):
    client = OpenAI()
    response = client.embeddings.create(
        input=text,
        model="text-embedding-ada-002"
    )
    return response.data[0].embedding

SimpleVectorStore 구현해보기

# 청크로 자른 docs와 임베딩벡터로 바꾼 것들을 차곡차곡 쌓아준다
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SimpleVectorStore:
  def __init__(self, docs, embedding):
    self.embedding = embedding
    self.documents = []
    self.vectors = []

    for doc in tqdm(docs, desc="embedding..."):
      self.documents.append(doc)
      vector = self.embedding.embed_query(doc)
      self.vectors.append(vector)

  def similarity_search(self, query, k=4):
    query_vector = self.embedding.embed_query(query)

    if not self.vectors:
      return []

    similarities = cosine_similarity([query_vector], self.vectors)[0]
    sorted_doc_similarities = sorted(zip(self.documents, similarities), key=lambda x: x[1], reverse=True)

    return sorted_doc_similarities[:k]

  def as_retriever(self, k=4):
    return SimpleRetriever(self, k)

SimpleRetriever 구현해보기

class SimpleRetriever:
  def __init__(self, vector_store, k=4):
    self.vector_store = vector_store
    self.k = k

  def get_relevant_documents(self, query):
    docs = self.vector_store.similarity_search(query, self.k)
    return docs

유사도 검색

raw_documents = SimpleTextLoader('state_of_the_union.txt').load()
text_splitter = SimpleCharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
documents = text_splitter.split_documents(raw_documents)
db = SimpleVectorStore(documents, SimpleOpenAIEmbeddings())

query = "what did the president say about Ketanji Brown Jackson"
docs = db.similarity_search(query)

유사도 검색 기반 검색기(Similarity-search-based retriever)

retriever = db.as_retriever()
unique_docs = retriever.get_relevant_documents(query="what did the president say about Ketanji Brown Jackson")

print(unique_docs)

SimpleRetrievalQA 구현해보기

import openai
system_prompt_template = ("You are a helpful assistant. "
"Based on the following content, "
"kindly and comprehensively respond to user questions."
"[Content]"
"{content}"
"")

class SimpleRetrievalQA():

  def __init__(self, retriever):
    self.retriever = retriever

  def invoke(self, query):

    docs = retriever.get_relevant_documents(query)

    for i, doc in enumerate(docs):
      print("[#" + str(i) + "]", doc[1])
      print(doc[0])
      print("")

    completion = openai.chat.completions.create(
        model="gpt-3.5-turbo-1106",
        messages=[
            {"role": "system", "content": system_prompt_template.format(content=docs)}, 
            {"role": "user", "content": query}
        ]
    )

    return completion.choices[0].message.content

chain = SimpleRetrievalQA(retriever)

answer = chain.invoke("what did the president say about Ketanji Brown Jackson")
print("")
print(">>>>", answer)

본 포스팅에서는 RAG 구현에 있어 핵심이라고 생각되는 컴포넌트들의 코드만 정리하였습니다. 더 자세한 내용은 영상을 참고해 주시기 바랍니다.