Milvus와 EmbedAnything으로 RAG 구축하기

임베드애니씽은 텍스트, PDF, 이미지, 오디오 등을 지원하는 초고속 경량 임베딩 파이프라인으로 Rust에 내장되어 있습니다.

이 튜토리얼에서는 Milvus와 함께 EmbedAnything을 사용하여 검색 증강 생성(RAG) 파이프라인을 구축하는 방법을 보여드리겠습니다. 특정 데이터베이스와 긴밀하게 결합하는 대신, 임베드애니씽은 플러그형 어댑터 시스템을 사용합니다. 어댑터는 임베딩의 포맷, 인덱싱 및 대상 벡터 저장소에 저장되는 방식을 정의하는 래퍼 역할을 합니다.

EmbedAnything과 Milvus 어댑터를 페어링하면 단 몇 줄의 코드만으로 다양한 파일 유형에서 임베딩을 생성하고 이를 Milvus에 효율적으로 저장할 수 있습니다.

⚠️ 참고: EmbedAnything의 어댑터는 Milvus에 삽입을 처리하지만, 즉시 검색을 지원하지는 않습니다. 전체 RAG 파이프라인을 구축하려면 MilvusClient를 별도로 인스턴스화하고 애플리케이션의 일부로 검색 로직(예: 벡터를 통한 유사성 검색)을 구현해야 합니다.

준비

종속성 및 환경

$ pip install -qU pymilvus milvus-lite openai embed_anything

리포지토리 및 로드 어댑터 복제하기

다음으로, 임베드애니싱 리포지토리를 복제하고 examples/adapters 디렉토리를 파이썬 경로에 추가합니다. 이 디렉토리에 커스텀 Milvus 어댑터 구현을 저장하여 EmbedAnything이 벡터 삽입을 위해 Milvus와 통신할 수 있도록 합니다.

import sys

# Clone the EmbedAnything repository if not already cloned
![ -d "EmbedAnything" ] || git clone https://github.com/StarlightSearch/EmbedAnything.git

# Add the `examples/adapters` directory to the Python path
sys.path.append("EmbedAnything/examples/adapters")
print("✅ EmbedAnything cloned and adapter path added.")

✅ EmbedAnything cloned and adapter path added.

이 RAG 파이프라인에서는 OpenAI를 LLM으로 사용합니다. 환경 변수로 OPENAI_API_KEY API 키를 준비해야 합니다.

import os
from openai import OpenAI

os.environ["OPENAI_API_KEY"] = "sk-***********"

openai_client = OpenAI()

RAG 빌드

Milvus 초기화

파일을 임베드하기 전에 Milvus와 상호 작용하는 두 가지 컴포넌트를 준비해야 합니다:

1. MilvusVectorAdapter - 이것은 EmbedAnything용 Milvus 어댑터이며, 벡터 수집 (즉, 임베딩 삽입 및 인덱스 생성) 에만 사용됩니다. 현재 검색 작업은 지원하지 않습니다.
2. MilvusClient - 공식 클라이언트는 pymilvus 에서 벡터 검색, 필터링, 수집 관리 등 Milvus의 모든 기능에 액세스할 수 있습니다.

혼동을 피하기 위해:

• MilvusVectorAdapter 은 벡터 저장을 위한 "쓰기 전용" 도구라고 생각하세요.
• MilvusClient 은 실제로 쿼리를 수행하고 RAG용 문서를 검색하는 "읽기 및 검색" 엔진이라고 생각하시면 됩니다.

import embed_anything
from embed_anything import (
    WhichModel,
    EmbeddingModel,
)
from milvus_db import MilvusVectorAdapter
from pymilvus import MilvusClient

# Official Milvus client for full operations
milvus_client = MilvusClient(uri="./milvus.db", token="")

# EmbedAnything adapter for pushing embeddings into Milvus
index_name = "embed_anything_milvus_collection"
milvus_adapter = MilvusVectorAdapter(
    uri="./milvus.db", token="", collection_name=index_name
)

# Delete existing collection if it exists
if milvus_client.has_collection(index_name):
    milvus_client.drop_collection(index_name)

# Create a new collection with dimension matching the embedding model later used
milvus_adapter.create_index(dimension=384)

Ok - Milvus DB connection established.
Collection 'embed_anything_milvus_collection' created with index.

임베딩 모델 초기화 및 PDF 문서 임베딩하기

이제 임베딩 모델을 초기화하겠습니다. 텍스트 임베딩을 생성하기 위한 가볍지만 강력한 모델인 문장 변환기 라이브러리의 all-MiniLM-L12-v2 model 을 사용하겠습니다. 이 모델은 384차원 임베딩을 생성하므로 Milvus 컬렉션 차원이 384로 설정되어 있는 것과 일치합니다. 이 정렬은 매우 중요하며 Milvus에 저장된 벡터 차원과 모델에서 생성된 차원 간의 호환성을 보장합니다.

임베드애니씽은 훨씬 더 많은 임베딩 모델을 지원합니다. 자세한 내용은 공식 문서를 참조하세요.

# Initialize the embedding model
model = EmbeddingModel.from_pretrained_hf(
    WhichModel.Bert, model_id="sentence-transformers/all-MiniLM-L12-v2"
)

이제 PDF 파일을 임베드해 보겠습니다. 임베드애니씽을 사용하면 PDF(및 기타 여러 문서)를 쉽게 처리하고 임베딩을 Milvus에 직접 저장할 수 있습니다.

# Embed a PDF file
data = embed_anything.embed_file(
    "./pdf_files/WhatisMilvus.pdf",
    embedder=model,
    adapter=milvus_adapter,
)

Converted 12 embeddings for insertion.
Successfully inserted 12 embeddings.

응답 검색 및 생성

다시 말씀드리지만, 현재 EmbedAnything의 MilvusVectorAdapter 은 벡터 수집 및 인덱싱만을 위한 경량 추상화입니다. 검색 또는 검색 쿼리는 지원하지 않습니다. 따라서 RAG 파이프라인을 구축하기 위해 관련 문서를 검색하려면 MilvusClient 인스턴스(milvus_client)를 직접 사용하여 Milvus 벡터 저장소를 쿼리해야 합니다.

Milvus에서 관련 문서를 검색하는 함수를 정의합니다.

def retrieve_documents(question, top_k=3):
    query_vector = list(
        embed_anything.embed_query([question], embedder=model)[0].embedding
    )
    search_res = milvus_client.search(
        collection_name=index_name,
        data=[query_vector],
        limit=top_k,
        output_fields=["text"],
    )
    docs = [(res["entity"]["text"], res["distance"]) for res in search_res[0]]
    return docs

RAG 파이프라인에서 검색된 문서를 사용하여 응답을 생성하는 함수를 정의합니다.

def generate_rag_response(question):
    retrieved_docs = retrieve_documents(question)
    context = "\n".join([f"Text: {doc[0]}\n" for doc in retrieved_docs])
    system_prompt = (
        "You are an AI assistant. Provide answers based on the given context."
    )
    user_prompt = f"""
    Use the following pieces of information to answer the question. If the information is not in the context, say you don't know.
    
    Context:
    {context}
    
    Question: {question}
    """
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
    )
    return response.choices[0].message.content

샘플 질문으로 RAG 파이프라인을 테스트해 보겠습니다.

question = "How does Milvus search for similar documents?"
answer = generate_rag_response(question)
print(f"Question: {question}")
print(f"Answer: {answer}")

Question: How does Milvus search for similar documents?
Answer: Milvus searches for similar documents primarily through Approximate Nearest Neighbor (ANN) search, which finds the top K vectors closest to a given query vector. It also supports various other types of searches, such as filtering search under specified conditions, range search within a specified radius, hybrid search based on multiple vector fields, and keyword search based on BM25. Additionally, it can perform reranking to adjust the order of search results based on additional criteria, refining the initial ANN search results.