배경
- faiss 모듈 내 코드를 분석해서, 여러 용도로 응용하고자함.
코드
from __future__ import annotations # 미래 버전의 타입 힌트를 사용할 수 있게 함
import logging
import operator
import os
import pickle
import uuid
import warnings
from pathlib import Path
from typing import (
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Sized,
Tuple,
Union,
)
import numpy as np
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.runnables.config import run_in_executor
from langchain_core.vectorstores import VectorStore
# 문서 저장소와 관련된 클래스 및 유틸리티 함수들 불러오기
from langchain_community.docstore.base import AddableMixin, Docstore
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_community.vectorstores.utils import (
DistanceStrategy, # 거리 계산 방식 클래스
maximal_marginal_relevance, # 최대 한계 관련성 함수
)
# 로깅 객체 설정
logger = logging.getLogger(__name__) # 현재 모듈의 로깅 이름을 기준으로 로거 생성
def dependable_faiss_import(no_avx2: Optional[bool] = None) -> Any:
"""
faiss 라이브러리를 가져옵니다. faiss를 사용할 수 없는 경우 오류를 발생시킵니다.
FAISS_NO_AVX2 환경 변수가 설정된 경우, AVX2 최적화 없이 faiss를 로드합니다.
Args:
no_avx2 (Optional[bool]): AVX2 최적화 없이 faiss를 로드하려면 True로 설정.
이는 벡터 저장소의 호환성을 높여 다른 장치에서도 사용할 수 있게 합니다.
Returns:
faiss: 가져온 faiss 모듈 객체.
Raises:
ImportError: faiss 모듈을 가져올 수 없는 경우 예외가 발생합니다.
"""
# no_avx2가 명시되지 않았고, 환경 변수 'FAISS_NO_AVX2'가 설정되어 있다면,
# 환경 변수를 사용하여 no_avx2를 설정
if no_avx2 is None and "FAISS_NO_AVX2" in os.environ:
no_avx2 = bool(os.getenv("FAISS_NO_AVX2"))
try:
# no_avx2가 True일 경우 AVX2 최적화가 없는 faiss 모듈을 가져옴
if no_avx2:
from faiss import swigfaiss as faiss
else:
import faiss
except ImportError:
# faiss를 설치하지 않았을 경우 오류 메시지와 함께 ImportError를 발생
raise ImportError(
"faiss 파이썬 패키지를 가져올 수 없습니다. "
"CUDA가 지원되는 GPU를 사용하는 경우 `pip install faiss-gpu`를 설치하거나, "
"파이썬 버전에 맞게 `pip install faiss-cpu`를 설치하세요."
)
return faiss
def _len_check_if_sized(x: Any, y: Any, x_name: str, y_name: str) -> None:
"""
두 개의 입력 객체 x와 y가 Sized일 경우, 길이가 동일한지 검사합니다.
길이가 다르면 오류를 발생시킵니다.
Args:
x (Any): 비교할 첫 번째 객체.
y (Any): 비교할 두 번째 객체.
x_name (str): x 객체의 이름 (오류 메시지에 표시될 이름).
y_name (str): y 객체의 이름 (오류 메시지에 표시될 이름).
Raises:
ValueError: x와 y가 Sized 타입이면서 길이가 다를 경우 예외 발생.
"""
# x와 y가 Sized 타입인 경우 (길이를 측정할 수 있는 객체)
if isinstance(x, Sized) and isinstance(y, Sized) and len(x) != len(y):
# 길이가 다를 경우 ValueError 발생
raise ValueError(
f"{x_name}과 {y_name}의 길이가 같아야 하지만, "
f"len({x_name})={len(x)} 및 len({y_name})={len(y)} 입니다."
)
return
class FAISS(VectorStore):
"""
FAISS 벡터 스토어 클래스입니다.
`Meta Faiss` 라이브러리를 기반으로 하는 벡터 스토어로, 문서의 임베딩을 저장하고
검색할 수 있습니다. FAISS 인덱스는 효율적으로 벡터 데이터를 검색하는 데 사용됩니다.
사용하려면 `faiss` 파이썬 패키지가 설치되어 있어야 합니다.
예제:
.. code-block:: python
from langchain_community.embeddings.openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
embeddings = OpenAIEmbeddings()
texts = ["FAISS is an important library", "LangChain supports FAISS"]
faiss = FAISS.from_texts(texts, embeddings)
"""
def __init__(
self,
# Union을 사용해 두 타입을 허용 : 매개변수가 여러 타입의 값을 받을 수 있게 하여 더 유연하고 재사용 가능한 코드를 만들기 위해서
# Callable[[str], List[float]]: 텍스트를 입력으로 받아 임베딩 벡터(리스트 형태의 실수)로 반환
# Embeddings: 임베딩을 처리하는 객체
embedding_function: Union[
Callable[[str], List[float]],
Embeddings,
],
index: Any,
docstore: Docstore,
index_to_docstore_id: Dict[int, str],
# Optional: relevance_score_fn이 지정되지 않거나 None 값이 전달될 수 있도록 허용
# Callable은 함수 타입을 나타내며, 여기서는 하나의 float 값을 입력으로 받고 float 값을 반환하는 함수를 기대한다는 것을 나타냅
relevance_score_fn: Optional[Callable[[float], float]] = None,
normalize_L2: bool = False,
distance_strategy: DistanceStrategy = DistanceStrategy.EUCLIDEAN_DISTANCE,
):
"""
FAISS 인스턴스를 초기화합니다.
Args:
embedding_function: 텍스트를 벡터(임베딩)로 변환하는 함수 또는 Embeddings 객체.
index: FAISS 인덱스 객체. 벡터들을 저장하고 검색할 수 있는 데이터 구조입니다.
docstore: 문서를 저장하는 Docstore 객체. 메타데이터 및 원본 문서를 관리합니다.
index_to_docstore_id: 인덱스에서 문서 저장소로의 매핑을 나타내는 딕셔너리입니다.
relevance_score_fn: 선택적 함수. 검색 결과의 적합성을 평가하기 위해 유사도 점수를 변환합니다.
normalize_L2: L2 정규화 여부를 나타내는 플래그입니다.
distance_strategy: 거리 측정 전략 (유클리드 거리, 코사인 거리 등).
Raises:
경고: embedding_function이 Embeddings 객체가 아니면 경고 메시지를 출력합니다.
"""
if not isinstance(embedding_function, Embeddings):
logger.warning(
"`embedding_function`은 Embeddings 객체여야 합니다. 함수 전달 지원은 곧 제거될 예정입니다."
)
self.embedding_function = embedding_function # 임베딩을 수행하는 함수 또는 Embeddings 객체.
self.index = index # FAISS 인덱스 객체, 벡터 데이터의 저장과 검색을 수행.
self.docstore = docstore # 문서 저장소, 문서의 메타데이터 및 내용을 저장.
self.index_to_docstore_id = index_to_docstore_id # 인덱스의 ID와 문서 저장소의 ID 간의 매핑 정보.
self.distance_strategy = distance_strategy # 벡터 간 거리 계산 전략, 기본값은 유클리드 거리.
self.override_relevance_score_fn = relevance_score_fn # 검색 결과의 적합성을 평가하는 사용자 정의 함수.
# 벡터 추가 시 L2 정규화를 수행할지 여부. 벡터의 크기가 검색 정확도에 영향을 미치지 않도록 하는 것이 중요합니다.
# 특히 코사인 유사도를 사용할 경우, L2 정규화를 적용하면 벡터 간의 각도 차이만 비교할 수 있어 검색 품질이 향상됨
self._normalize_L2 = normalize_L2
# 거리 측정 전략이 유클리드가 아닐 때, L2 정규화가 설정되었을 경우 경고를 발생.
if (
self.distance_strategy != DistanceStrategy.EUCLIDEAN_DISTANCE
and self._normalize_L2
):
warnings.warn(
"L2 정규화는 유클리드 거리 외의 거리 측정 방법에서는 사용할 수 없습니다: "
f"metric type: {self.distance_strategy}"
)
@property
def embeddings(self) -> Optional[Embeddings]:
"""
Embeddings 객체를 반환하는 속성입니다.
Returns:
embedding_function이 Embeddings 객체일 경우 반환, 아니면 None 반환.
"""
return (
self.embedding_function
if isinstance(self.embedding_function, Embeddings)
else None
)
# 문자열 리스트(List[str])를 받아들이며, 각 문자열은 임베딩할 텍스트를 나타냅니다.
# -> List[List[float]]: 함수가 반환하는 타입을 지정함
# 외부 리스트는 여러 개의 벡터를 담고 있습니다. 내부 리스트는 하나의 벡터를 나타내며, 벡터의 각 요소는 float 타입의 숫자
# 예> "Hello world"와 "This is a test" 텍스트를 벡터화한 결과라고 가정
# [
# [0.1, 0.2, 0.3], # 첫 번째 텍스트의 벡터
# [0.4, 0.5, 0.6] # 두 번째 텍스트의 벡터
# ]
def _embed_documents(self, texts: List[str]) -> List[List[float]]:
"""
주어진 텍스트들을 벡터로 변환합니다.
Args:
texts: 임베딩할 문자열의 리스트.
Returns:
주어진 텍스트들의 벡터 리스트.
"""
if isinstance(self.embedding_function, Embeddings):
return self.embedding_function.embed_documents(texts) # Embeddings 객체 사용 시 벡터화
else:
return [self.embedding_function(text) for text in texts] # 함수로 전달 시 각 텍스트를 벡터화
async def _aembed_documents(self, texts: List[str]) -> List[List[float]]:
"""
주어진 텍스트들을 비동기적으로 벡터로 변환합니다.
Args:
texts: 임베딩할 문자열의 리스트.
Returns:
주어진 텍스트들의 벡터 리스트.
Raises:
Exception: embedding_function이 Embeddings 객체가 아니면 예외 발생.
"""
if isinstance(self.embedding_function, Embeddings):
return await self.embedding_function.aembed_documents(texts) # 비동기 임베딩 수행
else:
raise Exception(
"`embedding_function`은 Embeddings 객체여야 합니다. 함수 전달 지원은 곧 제거될 예정입니다."
)
def _embed_query(self, text: str) -> List[float]:
"""
주어진 쿼리를 벡터로 변환합니다.
Args:
text: 임베딩할 쿼리 문자열.
Returns:
벡터로 변환된 쿼리.
"""
if isinstance(self.embedding_function, Embeddings):
return self.embedding_function.embed_query(text)
else:
return self.embedding_function(text)
async def _aembed_query(self, text: str) -> List[float]:
"""
주어진 쿼리를 비동기적으로 벡터로 변환합니다.
Args:
text: 임베딩할 쿼리 문자열.
Returns:
벡터로 변환된 쿼리.
Raises:
Exception: embedding_function이 Embeddings 객체가 아니면 예외 발생.
"""
if isinstance(self.embedding_function, Embeddings):
return await self.embedding_function.aembed_query(text)
else:
raise Exception(
"`embedding_function`은 Embeddings 객체여야 합니다. 함수 전달 지원은 곧 제거될 예정입니다."
)
# 실시간으로 데이터를 추가하거나, 새로운 데이터를 반영하기 위해 데이터베이스를 업데이트해야 할 경우 임베딩 추가 기능이 필수적입니다.
def __add(
self,
# Iterable 타입을 사용하면 리스트, 튜플, 제너레이터 등의 모든 반복 가능한 형태를 인자로 받을 수 있슴
texts: Iterable[str],
# Iterable[List[float]]는 List[List[float]], Tuple[List[float], ...], 또는 제너레이터 (List[float] for _ in range(n))와 같은 다양한 형태의 입력을 지원
# List[List[float]] 대신 Iterable[List[float]]로 지정함으로써 메모리를 덜 사용하는 방식으로 데이터를 넘길 수 있슴
embeddings: Iterable[List[float]],
metadatas: Optional[Iterable[dict]] = None,
ids: Optional[List[str]] = None,
) -> List[str]:
"""
주어진 텍스트와 임베딩들을 벡터 인덱스 및 문서 저장소에 추가합니다.
Args:
texts: 추가할 문자열의 반복 가능한 객체.
embeddings: 각 텍스트에 해당하는 벡터의 리스트.
metadatas: 각 텍스트에 연결된 메타데이터 리스트 (옵션).
ids: 각 텍스트에 대한 고유 ID 리스트 (옵션).
Returns:
추가된 텍스트의 ID 리스트.
Raises:
ValueError: docstore가 AddableMixin을 지원하지 않거나 ID가 중복될 경우 예외 발생.
"""
faiss = dependable_faiss_import()
# isinstance로 self.docstore가 AddableMixin을 상속받는지 확인
# AddableMixin을 상속받으면 해당 객체가 add 메서드를 통해 항목 추가 기능을 제공할 수 있음을 보장.
# 여러 클래스에서 동일한 기능(항목 추가)을 구현해야 하는 경우, AddableMixin을 상속받아 간단히 해당 기능을 추가가능
if not isinstance(self.docstore, AddableMixin):
raise ValueError(
"텍스트를 추가하려면 기본 docstore가 항목 추가를 지원해야 합니다. "
f"{self.docstore}는 이 기능을 지원하지 않습니다."
)
# texts와 metadatas의 길이 일치 여부 확인, metadatas가 없을 경우 빈 딕셔너리 사용
_len_check_if_sized(texts, metadatas, "texts", "metadatas")
# metadatas가 None일 경우 빈 딕셔너리 {}를 텍스트마다 생성하도록 합
_metadatas = metadatas or ({} for _ in texts)
documents = [
Document(page_content=t, metadata=m) for t, m in zip(texts, _metadatas)
]
# documents, embeddings, ids의 길이 일치 여부 확인
_len_check_if_sized(documents, embeddings, "documents", "embeddings")
_len_check_if_sized(documents, ids, "documents", "ids")
# ID에 중복이 있는지 확인
if ids and len(ids) != len(set(ids)):
raise ValueError("ID 목록에 중복된 ID가 있습니다.")
# 인덱스에 벡터 추가
vector = np.array(embeddings, dtype=np.float32)
if self._normalize_L2:
faiss.normalize_L2(vector) # L2 정규화 옵션이 설정된 경우 벡터 정규화
self.index.add(vector)
# docstore와 인덱스에 문서 추가
# ids 파라미터가 주어지지 않은 경우, 각 텍스트마다 고유한 ID가 필요하므로, 파이썬의 uuid 모듈을 사용하여 UUID를 생성합니다.
# uuid.uuid4()는 임의의 고유 ID를 생성하는 함수로, UUID는 중복 가능성이 매우 낮은 무작위한 ID를 제공
ids = ids or [str(uuid.uuid4()) for _ in texts]
# self.docstore는 텍스트 문서를 저장하는 문서 저장소 객체입니다. 이 라인은 새로 생성한 ID와 Document 객체들을 docstore에 추가하는 역할
self.docstore.add({id_: doc for id_, doc in zip(ids, documents)})
# starting_len은 현재 인덱스의 길이를 저장하는 변수입니다. 이는 새로 추가할 문서의 위치를 결정하기 위해 필요,
# 새로 추가하는 문서들은 starting_len 이후의 인덱스 번호를 할당받게 됨
starting_len = len(self.index_to_docstore_id)
# index_to_id는 새로 추가할 문서들의 인덱스 번호와 ID를 매핑하는 딕셔너리
index_to_id = {starting_len + j: id_ for j, id_ in enumerate(ids)}
# self.index_to_docstore_id가 새로 추가된 문서까지 포함하여 업데이트
self.index_to_docstore_id.update(index_to_id)
return ids
def similarity_search_with_score_by_vector(
self,
embedding: List[float],
k: int = 4,
# 메타데이터 필터링 조건으로, 메타데이터에 기반한 필터링을 수행할 수 있습니다. 함수나 조건 딕셔너리 형태로 제공됩니다. 필터링 조건을 만족하는 문서만 최종 결과에 포함
# Callable(함수) 또는 Dict[str, Any](딕셔너리) 중 하나의 타입을 허용
# Dict[str, Any]는 필터로 사용될 키-값 쌍을 정의합니다. 딕셔너리의 각 키는 특정 필드를 나타내고, 그 값은 해당 필드에 적용할 조건을 나타냄
# None: 이 필터 파라미터가 기본적으로 None이라는 값으로 설정된다는 의미입니다. 따라서 사용자가 필터를 제공하지 않으면 필터가 적용되지 않습니다.
filter: Optional[Union[Callable, Dict[str, Any]]] = None,
# 필터 적용 전 검색할 최대 문서 수로, 필터링을 위해 더 많은 문서를 검색한 뒤 필터를 적용해 상위 k개의 문서를 선택합니다. 기본값은 20
fetch_k: int = 20,
# 선택적인 추가 매개변수로, score_threshold 등 추가 필터링 조건을 설정할 수 있습니다.
**kwargs: Any,
) -> List[Tuple[Document, float]]:
"""
주어진 임베딩 벡터와 가장 유사한 문서들을 반환합니다.
Args:
embedding: 유사한 문서를 찾을 임베딩 벡터.
k: 반환할 문서 수. 기본값은 4.
filter: 메타데이터에 따라 필터링하는 함수 또는 조건(딕셔너리). 기본값은 None.
fetch_k: 필터링 전 검색할 문서 수. 기본값은 20.
Returns:
쿼리와 가장 유사한 문서와 각 문서의 L2 거리 (유사도 점수)의 튜플 리스트.
"""
# FAISS 라이브러리가 설치되어 있는지 확인하고, 없는 경우 ImportError를 발생
faiss = dependable_faiss_import() # FAISS 모듈 불러오기
# 검색할 임베딩 벡터를 numpy 배열로 변환합니다. np.array([embedding], dtype=np.float32)로 벡터를 32비트 부동소수점 형식의 배열로 생성
vector = np.array([embedding], dtype=np.float32) # 쿼리 벡터를 numpy 배열로 변환
# _normalize_L2 옵션이 True일 경우 faiss.normalize_L2(vector)를 호출하여 벡터를 L2 정규화합니다. L2 정규화는 벡터의 길이를 1로 만들어, 방향만 비교하도록 함
if self._normalize_L2:
faiss.normalize_L2(vector)
# self.index.search 메서드를 사용해 벡터와 가장 유사한 상위 k개의 문서를 검색합니다. 필터가 없는 경우 상위 k개를 검색하고,
# 필터가 있으면 fetch_k 개수만큼 검색하여 이후 필터링 단계에서 더 많은 문서를 필터링
# scores는 각 검색 결과의 유사도 점수이며, 점수가 낮을수록 유사도가 높습니다 (유클리드 거리 기준)
scores, indices = self.index.search(vector, k if filter is None else fetch_k) # 벡터와 가장 가까운 문서 검색
docs = []
# 필터가 있을 경우 필터 함수 생성
if filter is not None:
filter_func = self._create_filter_func(filter)
# indices[0]는 검색된 문서의 인덱스 번호를 포함한 리스트입니다. 이를 순회하여 각 검색 결과를 처리
for j, i in enumerate(indices[0]):
# i == -1인 경우, 이는 FAISS가 충분한 결과를 반환하지 않았다는 것을 의미합니다. 이 경우 해당 인덱스를 건너뜁
if i == -1:
continue # 검색 결과가 부족할 때 해당 인덱스 건너뜀
_id = self.index_to_docstore_id[i] # 인덱스 번호로 해당 문서의 ID 찾기
doc = self.docstore.search(_id) # ID로 문서 검색
if not isinstance(doc, Document):
raise ValueError(f"ID {_id}에 대한 문서를 찾을 수 없습니다. 현재 값: {doc}")
# 필터가 설정되어 있으면 필터 조건을 만족하는지 확인
if filter is not None:
if filter_func(doc.metadata): # 필터 조건을 만족하는 경우 문서 추가
# 필터 조건을 만족할 경우 (doc, scores[0][j]) 튜플을 docs 리스트에 추가합니다. 여기서 scores[0][j]는 해당 문서의 유사도 점수
docs.append((doc, scores[0][j]))
else:
# 필터가 설정되지 않았다면, 조건 검사 없이 문서와 점수를 바로 추가
docs.append((doc, scores[0][j])) # 필터가 없으면 문서와 점수를 바로 추가
# 선택적 점수 임계값 필터링
score_threshold = kwargs.get("score_threshold")
if score_threshold is not None:
cmp = (
# MAX_INNER_PRODUCT 또는 JACCARD일 경우 operator.ge (크거나 같음) 연산자를 사용하고,
# 그외(유클리드 거리, 코사인 거리 )는 operator.le (작거나 같음) 연산자를 사용
# MAX_INNER_PRODUCT, JACCARD는 점수가 높을수록 유사도가 높은 방식이고, 다른 방식은 점수가 낮을수록 유사도가 높은 방식이기 때문
operator.ge
if self.distance_strategy in (DistanceStrategy.MAX_INNER_PRODUCT, DistanceStrategy.JACCARD)
else operator.le
)
docs = [
(doc, similarity) for doc, similarity in docs if cmp(similarity, score_threshold)
]
return docs[:k] # 상위 k개의 문서와 유사도 점수 반환