[VectorDB] Milvus 연결 + 컬렉션 및 스키마 생성하기

안녕하세요? 이번에는 Python으로 Milvus 연결 + 컬렉션 및 스키마 생성하는 법을 알려드리겠습니다.

 

1. 서버에서 Milvus를 구동합니다.

저는 간단하게 블로그 작성용 + 테스트용으로 사용을 할려고 해서 docker-compose를 사용해서 진행하겠습니다.

다른분들은 따라 하실때 docker-container(standalone)로 사용하셔도 무관합니다.

 

2. 정상적으로 구동이 되었는지 확인

 

3. vscode를 킵니다.

필요한 모듈 설치

pip install pymilvus
# pymilvus 모듈을 설치합니다.

 

설치가 완료되면 아래와 같이 연결 테스트를 진행합니다.

from pymilvus import connections

connection_args = {
    "host": "192.168.XX.XX",
    "port": "19530",
    "user": "root",
    "password": "Milvus"
}

connections.connect(
    alias="default",
    **connection_args
)

is_connected = connections.has_connection("default")

if is_connected:
    print("정상적으로 연결되었습니다.")
    
else:
    print("연결에 실패했습니다.")

 

아래 사진과 같이 연결되었다는 문구가 나오면 정상적으로 연결이 되었다고 보시면 됩니다.

정상적으로 연결된 모습

※ has_connection 함수는 특정 alias 이름으로 설정된 연결이 현재 활성화되어 있는지 확인하는 함수입니다.

지정한 alias로 연결된 Milvus 서버가 존재하고, 그 연결이 유효하다면 True를 반환하고,

그렇지 않으면 False를 반환합니다.

 

다들 연결하시는데는 문제가 없으실 겁니다! 더군다나 root 계정과 passwd 비밀번호가 없어도 접속이 가능하거든요^^

ㄴ 해당 문제는 나중에 보안 설정할때 알려드리도록 하겠습니다.

 

만약 url 주소(http 혹은 https)로 연결하고 싶다고 하시면 아래와 같이 사용하실 수 있습니다.

connection_args = {
    "uri": http://192.168.XX.XX",
    "user": "root",
    "password": "Milvus"
}

이렇게 변경만 해도 정상적으로 연결이 가능합니다.

 

---

추가적으로 알아두면 좋은 함수들

Milvus 연결과 관련된 몇 가지 유용한 함수들이 있습니다.

 

1. 현재 활성화된 연결 목록 확인하기

이 함수는 현재 활성화된 모든 연결(`alias`)의 목록을 반환합니다.

Milvus에 연결할 때 `alias`라는 이름을 지정하게 되는데, 이 함수는 이러한 이름들을 리스트 형태로 반환합니다.

aliases = connections.list_connections()
print(aliases)

 

2. 특정 연결에 대한 정보 확인하기

이 함수는 지정된 alias에 대한 연결 정보(주소, 포트, 사용자 이름 등)를 반환합니다.

예를 들어, alias로 "default"를 사용했다면, 이 alias가 참조하는 Milvus 서버의 연결 정보를 가져올 수 있습니다.

connection_info = connections.get_connection_addr("default")
print(connection_info)

---

자! 이제 연결이 되었으니 이번에는 컬렉션 검증 및 생성하는 방법을 해보겠습니다.

 

필요한 모듈

from pymilvus import (
    CollectionSchema, DataType, FieldSchema, Collection, connections, utility
)

Milvus와의 연결

Milvus와 연결하기 위해 MilvusManager 클래스를 정의하겠습니다.

이 클래스는 데이터베이스와 연결을 관리하는 기능을 포함하고 있습니다.

 

class MilvusManager:
    def __init__(self):
        self.dbname = "default"
        self.connection_args = {
            "uri": "주소",
            "db_name": self.dbname,
        }

    def connect(self):
        try:
            print("Milvus와 연결 중...")
            connections.connect(
                alias=self.dbname,
                **self.connection_args
            )

            if connections.has_connection(self.dbname):
                print("Milvus와 연결되었습니다.")

            else:
                print("Milvus 연결 실패: 연결이 확인되지 않았습니다.")
                return False
            
        except Exception as e:
            print(f"Milvus 연결 실패: {e}")
            return False
            
        return True

위 코드에서 MilvusManager 클래스는 데이터베이스와의 연결을 관리하는 기능을 제공합니다.

connect 메서드는 Milvus에 연결을 시도하고, 연결 상태를 출력합니다.

 

 

컬렉션 생성 및 검증

이제 MilvusManager 클래스에 컬렉션을 생성하고 검증하는 기능을 추가하겠습니다.

컬렉션을 생성하기 전에 같은 이름의 컬렉션이 이미 존재하는지 확인하고, 존재하지 않으면 새로운 컬렉션을 생성합니다.

또한, 벡터 필드에 대한 인덱스를 생성하여 검색 성능을 최적화합니다.

def create_collection(self, collection_name):
    try:
        # Milvus 연결
        if not self.connect():
            return

        # 컬렉션이 이미 존재하는지 확인
        if utility.has_collection(collection_name, using=self.dbname):
            print(f"Collection '{collection_name}'이 이미 존재합니다.")
            return

        # 필드 구성
        fields = [
            FieldSchema(name="pk", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="text_vector", dtype=DataType.FLOAT_VECTOR, dim=1024),
        ]

        # 컬렉션 스키마 생성
        collection_schema = CollectionSchema(
            fields=fields,
            enable_dynamic_fields=True,
            description=f"{collection_name}의 컬렉션"
        )

        # 컬렉션 생성
        collection = Collection(name=collection_name, schema=collection_schema, using=self.dbname)
        print(f"Collection: '{collection_name}' 생성되었습니다.")
        
        # 인덱스 파라미터 설정
        index_params = {
                "metric_type": "COSINE",
                "index_type": "HNSW",
                "params": {"M": 64, "efConstruction": 200}
        }
         
		# 벡터 필드에 대한 인덱스 생성
        collection.create_index(field_name="text_vector", index_params=index_params)
        print(f"'text_vector' 인덱스 생성 완료")

    except Exception as e:
        print(f"생성 실패: {e}")

코드 설명

1. 컬렉션 중복 확인: utility.has_collection 메서드를 사용해 동일한 이름의 컬렉션이 이미 존재하는지 확인합니다.
   존재할 경우, 컬렉션을 다시 생성하지 않고 종료합니다.
   
   
2. 필드 구성: 컬렉션의 필드는 FieldSchema 클래스를 사용해 정의합니다.
   예제에서는 기본 키 필드(pk), 텍스트 필드(text), 그리고 1024차원의 벡터 필드(text_vector)를 정의했습니다.
   저는 임베딩 모델을 BGE-M3 모델을 사용해서 dim을 1024차원으로 설정하였습니다.
   
   
3. 컬렉션 스키마 생성: 정의된 필드를 사용해 컬렉션의 스키마를 생성합니다.
   이 과정에서 스키마에 대한 설명도 추가할 수 있습니다.
   
   
4. 컬렉션 생성: Collection 클래스를 사용해 컬렉션을 생성합니다.
   생성된 컬렉션의 이름과 함께 성공 메시지를 출력합니다.
   
   
5. 벡터 필드 인덱스 생성: 생성된 컬렉션의 벡터 필드(text_vector)에 대해 인덱스를 생성합니다.
   이를 통해 데이터베이스에서 벡터 검색을 더 빠르고 효율적으로 수행할 수 있습니다.

---

Milvus에서 지원하는 메트릭 유형은 Euclidean distance (L2), Inner product (IP), Cosine similarity (COSINE)이며,
이들 메트릭에 따라 다양한 인덱스 유형(FLAT, IVF_FLAT, IVF_SQ8, IVF_PQ, GPU_IVF_FLAT, GPU_IVF_PQ, HNSW, DISKANN)을 사용해 벡터 검색 성능을 최적화할 수 있습니다.

https://milvus.io/docs/index-vector-fields.md?tab=floating

Index Vector Fields | Milvus Documentation

 

각 유형별 간단한 설명입니다.

메트릭 유형

1. Euclidean distance (L2): 벡터 간의 유클리드 거리(직선 거리)를 계산하여 유사성을 평가합니다. 일반적인 거리 측정 방법입니다.
2. Inner product (IP): 두 벡터의 내적을 계산하여 유사성을 평가하며, 주로 유사도가 높을수록 값이 커지는 경우에 사용됩니다.
3. Cosine similarity (COSINE): 두 벡터 간의 코사인 각도를 사용하여 유사성을 평가합니다. 방향에 초점을 맞추며 크기보다는 패턴 유사성을 측정합니다.

인덱스 유형

1. FLAT: 모든 벡터를 순차적으로 비교하는 가장 기본적인 인덱스 유형으로, 정확도가 높지만 속도가 느립니다.
2. IVF_FLAT: 벡터를 여러 클러스터로 나누고, 각 클러스터 내에서 FLAT 검색을 수행하여 속도를 향상시킵니다.
3. IVF_SQ8: IVF_FLAT 방식에 양자화를 적용해 메모리 사용량을 줄이면서 검색 성능을 유지하는 인덱스 유형입니다.
4. IVF_PQ: IVF 방식을 활용하며, 벡터를 압축해 메모리 효율성을 높인 인덱스 유형으로, 대규모 데이터에 적합합니다.
5. GPU_IVF_FLAT: GPU를 사용해 IVF_FLAT 알고리즘을 가속화하여 대규모 데이터에서도 빠른 검색을 제공합니다.
6. GPU_IVF_PQ: GPU를 활용해 IVF_PQ 알고리즘을 가속화하여 메모리 효율성을 유지하면서도 높은 검색 성능을 제공합니다.
7. HNSW: 그래프 기반의 인덱스로, 검색 속도와 정확도 모두에서 높은 성능을 제공하며 특히 고차원 데이터에 유리합니다.
8. DISKANN: 디스크 기반의 인덱스로, 대규모 데이터셋을 메모리에 적재하지 않고도 효율적으로 검색할 수 있는 방법입니다.

저는 검색 속도와 정확성이 높은 HNSW와 Cosine유사도를 사용했습니다.

---

실행 예시

이제 MilvusManager 클래스를 사용해 실제로 컬렉션을 생성해볼 수 있습니다.

manager = MilvusManager()
manager.create_collection("example_collection")

 

실행결과

이 포스팅에서는 Python을 사용해 Milvus에 컬렉션을 생성하고 관리하는 방법을 알아보았습니다.

pymilvus 모듈을 활용해 데이터베이스 연결, 컬렉션 중복 확인, 필드 정의 및 컬렉션 생성을 단계별로 구현해 보았습니다.

Milvus를 처음 접하는 개발자들에게 도움이 되길 바랍니다.