5.4. 4단계. 지식 저장 (Store)

벡터화된 데이터를 효율적으로 관리하고 검색할 수 있도록 저장하는 과정

본 페이지는 45:42 지점부터 59:33 지점까지 보시면 됩니다.

5.4.1. 필요성

단순히 문서를 벡터로 변환했다고 해서 즉시 활용할 수 있는 것이 아니라, 빠르고 정확한 검색이 가능하도록 저장해야 하며, 적절한 인덱싱 기법을 적용하여 검색 속도를 최적화해야만 합니다. “지식 저장 (Store)” 단계는 효율적인 검색을 위한 최적화된 저장소 구축 과정입니다.

본 튜토리얼에서는 파이썬 코드로 저장만 구현하고 인덱싱 과정은 다루지 않습니다. Vector Store를 사용하실 때 인덱싱 기능을 활용하실 수 있습니다.

5.4.2. 파이썬 구현

본 튜토리얼에서는 비효율적인 방식이지만 벡터 데이터를 리스트에 저장하고 이를 pickle 포맷으로 파일로 저장하는 방식으로 간략히 구현하여, Vector Store의 동작을 이해해보겠습니다.

list 클래스를 상속받은 VectorStore 클래스를 구현하여 VectorStore 내에서 벡터 데이터를 저장하고 관리할 수 있도록 구현합니다. 다음 4개의 메서드를 지원합니다.

  1. make 메서드 : 문서 리스트를 받아서 벡터 데이터 리스트를 생성 (Embed 단계)

  2. save 메서드 : 현재의 벡터 데이터 리스트를 디스크에 파일로 저장

  3. load 메서드 : 디스크에 저장된 벡터 데이터를 로딩하여 리스트 화

  4. search 메서드 : 질문 문자열을 받아서 유사 문서 목록을 검색

5.4.2.1. embed 함수를 make 메서드로 리팩토링

먼저 3단계. 지식 표현 (Embed) 단계에서 구현한 embed 함수를 VectorStore.make 클래스 함수로 리팩토링합니다.

 1class VectorStore(list):
 2    # 지식에 사용한 임베딩 모델과 질문에 사용할 임베딩 모델은 동일해야만 합니다.
 3    # 각각 임베딩 모델명을 지정하지 않고, 임베딩 모델명을 클래스 변수로 선언하여
 4    # 모델명 변경의 용이성을 확보합니다.
 5    embedding_model = "text-embedding-3-small"
 6
 7    @classmethod
 8    def make(cls, doc_list: List[Document]) -> "VectorStore":
 9        vector_store = cls()
10
11        for doc in doc_list:
12            response = client.embeddings.create(
13                model=cls.embedding_model,
14                input=doc.page_content,
15            )
16            vector_store.append(
17                {
18                    "document": doc.model_copy(),
19                    "embedding": response.data[0].embedding,
20                }
21            )
22
23        return vector_store
24
25     # TODO: 이어서 구현할 예정입니다.
26
27     # 벡터 스토어 문서/임베딩 데이터를 지정 경로에 파일로 저장
28     # def save(self, vector_store_path: Path) -> None: ...
29
30     # 지정 경로의 파일을 읽어서 벡터 스토어 문서/임베딩 데이터 복원
31     # @classmethod
32     # def load(cls, vector_store_path: Path) -> "VectorStore": ...
33
34     # 질의 문자열을 받아서, 벡터 스토어에서 유사 문서를 최대 k개 반환
35     # def search(self, question: str, k: int = 4) -> List[Document]:
36
37
38# vector_store = embed(doc_list)
39vector_store = VectorStore.make(doc_list)

5.4.2.2. save 메서드와 load 메서드 구현

벡터 데이터가 메모리에만 저장되어있는 상황이기에 프로세스가 종료되면 데이터가 사라집니다. 따라서 데이터를 디스크에 저장하고 관리할 필요가 있습니다.

파이썬 객체를 파일로 저장하는 가장 간단한 방법으로서 pickle이 파이썬 기본에서 지원 됩니다. 물론 JSON이나 CSV 등 다양한 포맷으로 저장할 수 있습니다. 물론 벡터 데이터를 저장하는 효율적인 방법은 아니지만, 구현이 간단합니다.

 1import pickle
 2from pathlib import Path
 3
 4class VectorStore(list):
 5    # ...
 6
 7    def save(self, vector_store_path: Path) -> None:
 8        """
 9        벡터 스토어 문서/임베딩 데이터를 지정 경로에 파일로 저장
10        """
11        with vector_store_path.open("wb") as f:
12            # 리스트(self)를 pickle 포맷으로 파일(f)에 저장
13            pickle.dump(self, f)
14
15    @classmethod
16    def load(cls, vector_store_path: Path) -> "VectorStore":
17        """
18        지정 경로의 파일을 읽어서 벡터 스토어 문서/임베딩 데이터 복원
19        """
20        with vector_store_path.open("rb") as f:
21            # pickle 포맷으로 파일(f)에서 리스트(VectorStore)를 로딩
22            return pickle.load(f)
23
24     # TODO: 이어서 구현할 예정입니다.
25
26     # 질의 문자열을 받아서, 벡터 스토어에서 유사 문서를 최대 k개 반환
27     # def search(self, question: str, k: int = 4) -> List[Document]:

5.4.2.4. VectorStore 클래스 현재 상황

 1import pickle
 2from pathlib import Path
 3from typing import List
 4
 5import numpy as np
 6import openai
 7from langchain_community.utils.math import cosine_similarity
 8from langchain_core.documents import Document
 9
10
11client = openai.Client()
12
13
14def load() -> List[Document]:
15    file_path = "빽다방.txt"
16    지식: str = open(file_path, "rt", encoding="utf-8").read()
17    docs = [
18        Document(
19            # 의미있는 메타데이터가 있다면, 맘껏 더 담으시면 됩니다.
20            metadata={"source": file_path},
21            page_content=지식,
22        )
23    ]
24    return docs
25
26
27def split(src_doc_list: List[Document]) -> List[Document]:
28    new_doc_list = []
29    for doc in src_doc_list:
30        for new_page_content in doc.page_content.split("\n\n"):
31            new_doc_list.append(
32                Document(
33                    metadata=doc.metadata.copy(),
34                    page_content=new_page_content,
35                )
36            )
37    return new_doc_list
38
39
40class VectorStore(list):
41    embedding_model = "text-embedding-3-small"
42
43    @classmethod
44    def make(cls, doc_list: List[Document]) -> "VectorStore":
45        vector_store = cls()
46
47        for doc in doc_list:
48            response = client.embeddings.create(
49                model=cls.embedding_model,
50                input=doc.page_content,
51            )
52            vector_store.append(
53                {
54                    "document": doc.model_copy(),
55                    "embedding": response.data[0].embedding,
56                }
57            )
58
59        return vector_store
60
61    def save(self, vector_store_path: Path) -> None:
62        """
63        현재의 벡터 데이터 리스트를 지정 경로에 파일로 저장
64        """
65        with vector_store_path.open("wb") as f:
66            # 리스트(self)를 pickle 포맷으로 파일(f)에 저장
67            pickle.dump(self, f)
68
69    @classmethod
70    def load(cls, vector_store_path: Path) -> "VectorStore":
71        """
72        지정 경로에 저장된 파일을 읽어서 벡터 데이터 리스트를 반환
73        """
74        with vector_store_path.open("rb") as f:
75            # pickle 포맷으로 파일(f)에서 리스트(VectorStore)를 로딩
76            return pickle.load(f)
77
78     # TODO: 이어서 구현할 예정입니다.
79
80     # 질의 문자열을 받아서, 벡터 스토어에서 유사 문서를 최대 k개 반환
81     # def search(self, question: str, k: int = 4) -> List[Document]:

위 VectorStore 클래스는 다음과 같이 활용할 수 있습니다.

 1def main():
 2    vector_store_path = Path("vector_store.pickle")
 3
 4    # 지정 경로에 파일이 없으면
 5    # 문서를 로딩하고 분할하여 벡터 데이터를 생성하고 해당 경로에 저장합니다.
 6    if not vector_store_path.is_file():
 7        doc_list = load()
 8        print(f"loaded {len(doc_list)} documents")
 9        doc_list = split(doc_list)
10        print(f"split into {len(doc_list)} documents")
11        vector_store = VectorStore.make(doc_list)
12        vector_store.save(vector_store_path)
13        print(f"created {len(vector_store)} items in vector store")
14     # 지정 경로에 파일이 있으면, 로딩하여 VectorStore 객체를 복원합니다.
15    else:
16        vector_store = VectorStore.load(vector_store_path)
17        print(f"loaded {len(vector_store)} items in vector store")
18
19    # TODO: RAG를 통해 지식에 기반한 AI 답변을 구해보겠습니다.
20    question = "빽다방 카페인이 높은 음료와 가격은?"
21    print(f"RAG를 통해 '{question}' 질문에 대해서 지식에 기반한 AI 답변을 구해보겠습니다.")
22
23
24if __name__ == "__main__":
25    main()

  1. 첫번째 실행에서는 vector_store.pickle 파일이 없으므로 load, split, make, save 순서로 호출되어, VectoreStore 객체를 생성하고 파일로 백업합니다.

  2. 이후 실행에서는 vector_store.pickle 파일이 있으므로 load 함수를 호출하여, VectorStore 객체를 복원합니다.

  3. 재생성하실려면 vector_store.pickle 파일을 삭제하고 다시 실행해주세요.

참고: batch API를 활용해서 임베딩 비용을 50% 절감하실 수 있습니다.

임베딩 API를 활용하면 즉시 임베딩 처리가 가능합니다. 그러나 대부분의 문서 임베딩 작업은 실시간 처리가 아닌 대량 데이터를 천천히 처리해도 문제가 없는 경우가 많습니다. 이런 경우 Batch API를 활용하면 비용을 크게 절감할 수 있습니다.

OpenAI 가격 페이지에 따르면 Batch 방식을 사용할 경우 비용이 실시간 처리 방식 대비 50% 저렴 합니다.