LangChain과 Qdrant를 활용한 S3 데이터 인제스트
LangChain과 Qdrant를 활용한 S3 데이터 인제스트 (tutorials-build-essentials-data-ingestion-beginners)
벡터 저장소로의 데이터 인제스트(ingestion) 는 효과적인 검색·검색 알고리즘을 구축하는 데 필수적이에요. 특히 데이터의 거의 80%가 비정형이라 정해진 형식이 없기 때문이에요.
이 튜토리얼에서는 AWS S3에서 직접 데이터를 가져와 Qdrant로 넣는 간결한 데이터 인제스트 파이프라인을 만들어 볼 거예요. 벡터 임베딩을 살펴보면서 비정형 데이터를 문서를 의미적으로 검색할 수 있는 형식으로 변환해 볼 거예요. 비정형 데이터 속에 숨은 통찰을 발견할 새로운 방법을 찾아볼 준비를 해 봐요!
| 시간: 30분 | 난이도: 초급 |
|---|
인제스트 워크플로우 아키텍처
이 워크플로우에서는 클라우드 스토리지, 자연어 처리(NLP) 도구, 임베딩 기술을 사용해 강력한 문서 인제스트·분석 파이프라인을 구축할 거예요. S3 버킷의 원시 데이터로 시작해 LangChain으로 전처리하고, 텍스트와 이미지 모두에 임베딩 API를 적용한 뒤 결과를 유사성 검색에 최적화된 벡터 데이터베이스인 Qdrant에 저장할 거예요.
그림 1: 데이터 인제스트 워크플로우 아키텍처
이 워크플로우의 각 구성 요소를 살펴볼게요.
- S3 버킷: 우리의 출발점이에요. PDF, 이미지, 텍스트 등 다양한 파일 형식을 위한 중앙 집중식 확장 가능 스토리지 솔루션이에요.
- LangChain: 파이프라인의 오케스트레이터 역할을 해요. 추출, 전처리를 처리하고 임베딩 생성을 위한 데이터 흐름을 관리해요. PDF 처리를 단순화해서 여기서 OCR(광학 문자 인식)을 적용할 걱정을 할 필요가 없어요.
- Qdrant: 벡터 데이터베이스로서 Qdrant는 임베딩과 그 payload를 저장해 모든 콘텐츠 유형에서 효율적인 유사성 검색과 검색을 가능하게 해요.
사전 준비 (Prerequisites)
이 섹션에서는 S3 버킷에서 데이터를 인제스트하는 단계별 안내를 볼 거예요. 하지만 그 전에 모든 사전 준비 사항을 갖췄는지 확인해 볼게요.
| 샘플 데이터 | 텍스트 형식의 제품 리뷰와 해당 이미지를 담은 폴더로 구성된 샘플 데이터셋을 사용해요. |
| AWS 계정 | S3 서비스에 접근할 수 있는 활성 AWS 계정 |
| Qdrant Cloud | 컬렉션 관리와 쿼리 실행을 위해 WebUI에 접근할 수 있는 Qdrant Cloud 계정 |
| LangChain | 모든 것을 연결하는 인기 프레임워크 |
지원되는 문서 유형
인제스트에 사용되는 문서는 PDF, 텍스트 파일, 이미지 등 다양한 유형이 될 수 있어요. 테스트와 실험을 위해 지원되는 문서 유형의 폴더로 구성된 구조화된 S3 버킷을 만들 거예요.
Python 환경
Python 환경(Python 3.9 이상)에 다음 라이브러리가 설치되어 있어야 해요.
boto3
langchain-community
langchain
python-dotenv
unstructured
unstructured[pdf]
qdrant_client
fastembed
액세스 키: AWS 액세스 키, S3 시크릿 키, Qdrant API 키를 .env 파일에 저장해 쉽게 접근할 수 있게 해요. 샘플 .env 파일은 다음과 같아요.
ACCESS_KEY = ""
SECRET_ACCESS_KEY = ""
QDRANT_KEY = ""
코드가 PDF 처리를 지원하지만, 샘플 데이터에는 현재 PDF 파일이 없어요.
Step 1: S3에서 데이터 인제스트
LangChain 프레임워크는 PDF, 이미지, 텍스트 파일 같은 형식의 문서를 로드하는 내장 지원 덕분에 AWS S3 같은 스토리지 서비스에서 데이터를 쉽게 인제스트하게 해줘요.
LangChain을 S3에 연결하려면 S3DirectoryLoader를 사용해요. 이 로더는 S3 버킷에서 파일을 LangChain 파이프라인으로 직접 로드할 수 있게 해줘요.
예시: LangChain이 S3에서 파일을 로드하도록 구성
S3 버킷에서 데이터를 인제스트하도록 LangChain을 설정하는 방법은 다음과 같아요.
from langchain_community.document_loaders import S3DirectoryLoader
# S3 문서 로더 초기화
loader = S3DirectoryLoader(
"product-dataset", # S3 버킷 이름
"p_1", #S3 폴더 이름 (첫 번째 제품 데이터 포함)
aws_access_key_id=aws_access_key_id, # AWS 액세스 키
aws_secret_access_key=aws_secret_access_key # AWS 시크릿 액세스 키
)
# 지정된 S3 버킷에서 문서 로드
docs = loader.load()
Step 2: 문서를 임베딩으로 변환
임베딩은 여기서 핵심 비결이에요. 쉽게 비교할 수 있는 형태로 "의미"를 포착하는 데이터(텍스트, 이미지, 오디오)의 수치 표현이에요. 텍스트와 이미지를 임베딩으로 변환하면 빠르고 효율적으로 유사성 검색을 수행할 수 있어요. 임베딩을 Qdrant에서 데이터의 의미 있는 통찰을 저장하고 검색하는 다리로 생각해 보세요.
임베딩 생성에 사용할 모델
시작하기 위해 두 가지 강력한 모델을 사용할 거예요.
- 텍스트 데이터 변환을 위한
sentence-transformers/all-MiniLM-L6-v2임베딩 - 이미지 데이터를 위한
CLIP(Contrastive Language-Image Pretraining)
문서 처리 함수
다음으로 문서 파이프라인의 각기 다른 파일 유형을 처리할 process_text와 process_image 두 함수를 정의할 거예요. process_text 함수는 텍스트 문서에서 원시 콘텐츠를 추출해 반환하고, process_image는 S3 소스에서 이미지를 검색해 메모리로 로드해요.
from PIL import Image
def process_text(doc):
source = doc.metadata['source'] # 문서 소스 추출 (예: S3 URL)
text = doc.page_content # 텍스트 파일에서 콘텐츠 추출
print(f"Processing text from {source}")
return source, text
def process_image(doc):
source = doc.metadata['source'] # 문서 소스 추출 (예: S3 URL)
print(f"Processing image from {source}")
bucket_name, object_key = parse_s3_url(source) # S3 URL 파싱
response = s3.get_object(Bucket=bucket_name, Key=object_key) # S3에서 이미지 가져오기
img_bytes = response['Body'].read()
img = Image.open(io.BytesIO(img_bytes))
return source, img
문서 처리를 위한 헬퍼 함수
S3에서 이미지를 검색하려면, 헬퍼 함수 parse_s3_url이 S3 URL을 버킷과 주요 구성 요소로 분해해요. 이는 S3 스토리지에서 이미지를 가져오는 데 필수적이에요.
def parse_s3_url(s3_url):
parts = s3_url.replace("s3://", "").split("/", 1)
bucket_name = parts[0]
object_key = parts[1]
return bucket_name, object_key
Step 3: 임베딩을 Qdrant에 로드
이제 문서가 처리되고 임베딩으로 변환됐으니, 다음 단계는 이 임베딩을 Qdrant에 로드하는 거예요.
Qdrant에서 컬렉션 생성
Qdrant에서 데이터는 컬렉션으로 구성돼요. 각 컬렉션은 임베딩(또는 포인트) 집합과 연관된 메타데이터(payload)를 나타내요. 앞서 생성한 임베딩을 저장하려면 먼저 컬렉션을 만들어야 해요.
텍스트와 이미지 임베딩을 모두 저장하는 Qdrant 컬렉션을 만드는 방법은 다음과 같아요.
def create_collection(collection_name):
qdrant_client.create_collection(
collection_name,
vectors_config={
"text_embedding": models.VectorParams(
size=384, # 텍스트 임베딩의 차원
distance=models.Distance.COSINE, # 비교에 cosine 유사도 사용
),
"image_embedding": models.VectorParams(
size=512, # 이미지 임베딩의 차원
distance=models.Distance.COSINE, # 비교에 cosine 유사도 사용
),
},
)
create_collection("products-data")
이 함수는 텍스트(384차원)와 이미지(512차원) 임베딩을 저장하는 컬렉션을 만들고, 컬렉션 내에서 임베딩을 비교하는 데 cosine 유사도를 사용해요.
컬렉션이 준비되면 임베딩을 Qdrant에 로드할 수 있어요. 여기에는 임베딩과 연관 메타데이터(payload)를 지정된 컬렉션에 삽입(또는 업데이트)하는 작업이 포함돼요.
임베딩을 Qdrant에 로드하는 코드는 다음과 같아요.
def ingest_data(points):
operation_info = qdrant_client.upsert(
collection_name="products-data", # 데이터가 삽입되는 컬렉션
points=points
)
return operation_info
인제스트 설명
- 데이터 포인트 Upsert:
qdrant_client의 upsert 메서드는 각 PointStruct를 지정된 컬렉션에 삽입해요. 같은 ID의 포인트가 이미 있다면 새 값으로 업데이트돼요. - 작업 정보: 함수는 성공 상태나 잠재적 오류 같은 upsert 작업의 세부 정보를 담은
operation_info를 반환해요.
인제스트 코드 실행
함수를 호출하고 데이터를 인제스트하는 방법은 다음과 같아요.
from qdrant_client import models
if __name__ == "__main__":
collection_name = "products-data"
create_collection(collection_name)
for i in range(1,6): # 다섯 개의 문서
folder = f"p_{i}"
loader = S3DirectoryLoader(
"product-dataset",
folder,
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key
)
docs = loader.load()
points, text_review, product_image = [], "", ""
for idx, doc in enumerate(docs):
source = doc.metadata['source']
if source.endswith(".txt") or source.endswith(".pdf"):
_text_review_source, text_review = process_text(doc)
elif source.endswith(".png"):
product_image_source, product_image = process_image(doc)
if text_review:
point = models.PointStruct(
id=idx, # 각 포인트의 고유 식별자
vector={
"text_embedding": models.Document(
text=text_review, model="sentence-transformers/all-MiniLM-L6-v2"
),
"image_embedding": models.Image(
image=product_image, model="Qdrant/clip-ViT-B-32-vision"
),
},
payload={"review": text_review, "product_image": product_image_source},
)
points.append(point)
operation_info = ingest_data(points)
print(operation_info)
PointStruct는 다음 핵심 매개변수로 초기화돼요.
- id: 각 임베딩의 고유 식별자로, 일반적으로 증가하는 인덱스예요.
- vector: 임베딩할 텍스트·이미지 입력을 담는 딕셔너리예요.
qdrant-client는 내부적으로 FastEmbed를 사용해 이 입력들로부터 벡터 표현을 로컬에서 자동 생성해요. - payload: 제품 리뷰와 이미지 참조 같은 추가 메타데이터를 저장하는 딕셔너리로, 검색 중 검색과 맥락에 매우 유용해요.
이 코드는 S3 버킷에서 폴더를 동적으로 로드하고, 텍스트와 이미지 파일을 따로 처리한 뒤 전용 리스트에 임베딩과 연관 데이터를 저장해요. 그런 다음 각 데이터 항목에 대해 PointStruct를 만들고 인제스트 함수를 호출해 Qdrant에 로드해요.
Qdrant WebUI 대시보드 살펴보기
임베딩이 Qdrant에 로드되면 WebUI 대시보드로 컬렉션을 시각화하고 관리할 수 있어요. 대시보드는 컬렉션과 그 데이터를 보는 명확하고 구조화된 인터페이스를 제공해요. 다음 섹션에서 자세히 살펴볼게요.
Step 4: Qdrant WebUI에서 데이터 시각화
Qdrant WebUI에서 데이터 시각화를 시작하려면 Overview 섹션으로 이동해 Access the database를 선택해요.
그림 2: Qdrant UI에서 데이터베이스 접근
프롬프트가 나타나면 API 키를 입력해요. 접속하면 컬렉션과 해당 데이터 포인트를 볼 수 있어요. 컬렉션이 다음과 같이 표시될 거예요.
그림 3: Qdrant의 product-data 컬렉션
Qdrant에 인제스트된 가장 최근 포인트를 살펴볼게요.
그림 4: product-data 컬렉션에 추가된 최신 포인트
Qdrant WebUI의 검색 기능은 컬렉션 전반에 걸쳐 벡터 검색을 수행할 수 있게 해줘요. 필터와 매개변수를 적용할 수 있는 옵션 덕분에 관련 임베딩을 검색하고 데이터 내 관계를 탐색하는 것이 쉬워져요. 시작하려면 왼쪽 패널의 Console로 이동해 쿼리를 만들 수 있어요.
그림 5: Qdrant Console 개요
첫 번째 쿼리는 모든 컬렉션을 검색하고, 두 번째는 product-data 컬렉션에서 포인트를 가져오며, 세 번째는 샘플 쿼리를 수행해요. 이는 Qdrant UI에서 데이터와 상호작용하는 것이 얼마나 간단한지 보여줘요.
이제 쿼리로 데이터베이스에서 일부 문서를 검색해 볼게요.
그림 6: 관련 문서를 검색하기 위한 Qdrant 클라이언트 쿼리
이 예시에서는 디자인이 개선된 Phones(폰) 를 쿼리했어요. 그런 다음 OpenAI로 텍스트를 벡터로 변환하고 디자인 개선을 강조하는 관련 폰 리뷰를 검색했어요.
결론 (Conclusion)
이 가이드에서 S3 버킷을 설정하고, 다양한 데이터 유형을 인제스트하고, 임베딩을 Qdrant에 저장했어요. LangChain을 사용해 텍스트·이미지 파일을 동적으로 처리해 각 파일 유형을 작업하기 쉽게 만들었어요.
이제 여러분 차례예요. 비디오 같은 다양한 데이터 유형으로 실험해 보고, 애플리케이션을 강화하기 위해 Qdrant의 고급 기능을 탐색해 보세요. 시작하려면 가입해 주세요.
