고급 비디오 검색: 시맨틱 검색을 위한 Twelve Labs와 Milvus 활용하기

소개

Twelve Labs Embed API와 Milvus를 사용해 시맨틱 비디오 검색을 구현하는 방법에 대한 종합적인 튜토리얼에 오신 것을 환영합니다. 이 가이드에서는 Twelve Labs의 고급 멀티모달 임베딩과 Milvus의 효율적인 벡터 데이터베이스를 활용하여 강력한 동영상 검색 솔루션을 구축하는 방법을 살펴봅니다. 이러한 기술을 통합함으로써 개발자는 동영상 콘텐츠 분석의 새로운 가능성을 열어 콘텐츠 기반 동영상 검색, 추천 시스템, 동영상 데이터의 뉘앙스를 이해하는 정교한 검색 엔진과 같은 애플리케이션을 구현할 수 있습니다.

이 튜토리얼에서는 개발 환경 설정부터 기능적인 시맨틱 비디오 검색 애플리케이션 구현까지 전체 과정을 안내합니다. 동영상에서 멀티모달 임베딩 생성하기, Milvus에 효율적으로 저장하기, 유사도 검색을 수행하여 관련 콘텐츠를 검색하기 등의 핵심 개념을 다룹니다. 동영상 분석 플랫폼, 콘텐츠 검색 도구를 구축하든, 동영상 검색 기능으로 기존 애플리케이션을 개선하든, 이 가이드는 프로젝트에서 Twelve Labs와 Milvus의 결합된 강점을 활용할 수 있는 지식과 실용적인 단계를 제공합니다.

전제 조건

시작하기 전에 다음이 준비되어 있는지 확인하세요:

Twelve Labs API 키(키가 없는 경우 https://api.twelvelabs.io 에서 등록) 시스템에 설치된 Python 3.7 이상 버전

개발 환경 설정

프로젝트를 위한 새 디렉토리를 만들고 해당 디렉토리로 이동합니다:

mkdir video-search-tutorial
cd video-search-tutorial

가상 환경을 설정합니다(선택 사항이지만 권장 사항):

python -m venv venv
source venv/bin/activate  # On Windows, use `venv\Scripts\activate`

필요한 Python 라이브러리를 설치합니다:

pip install twelvelabs pymilvus

프로젝트를 위한 새 Python 파일을 만듭니다:

touch video_search.py

이 video_search.py 파일이 튜토리얼에 사용하는 기본 스크립트가 됩니다. 다음으로 보안을 위해 Twelve Labs API 키를 환경 변수로 설정합니다:

export TWELVE_LABS_API_KEY='your_api_key_here'

Milvus에 연결하기

Milvus와의 연결을 설정하기 위해 MilvusClient 클래스를 사용합니다. 이 접근 방식은 연결 프로세스를 간소화하고 로컬 파일 기반 Milvus 인스턴스로 작업할 수 있어 튜토리얼에 적합합니다.

from pymilvus import MilvusClient

# Initialize the Milvus client
milvus_client = MilvusClient("milvus_twelvelabs_demo.db")

print("Successfully connected to Milvus")

이 코드는 밀버스_트와엘랩스_데모라는 파일에 모든 데이터를 저장하는 새로운 밀버스 클라이언트 인스턴스를 생성합니다. 이 파일 기반 접근 방식은 개발 및 테스트 목적에 이상적입니다.

동영상 임베딩을 위한 Milvus 컬렉션 만들기

이제 Milvus에 연결되었으므로 동영상 임베딩과 관련 메타데이터를 저장할 컬렉션을 만들어 보겠습니다. 컬렉션 스키마를 정의하고 컬렉션이 아직 존재하지 않는 경우 컬렉션을 생성합니다.

# Initialize the collection name
collection_name = "twelvelabs_demo_collection"

# Check if the collection already exists and drop it if it does
if milvus_client.has_collection(collection_name=collection_name):
    milvus_client.drop_collection(collection_name=collection_name)

# Create the collection
milvus_client.create_collection(
    collection_name=collection_name,
    dimension=1024  # The dimension of the Twelve Labs embeddings
)

print(f"Collection '{collection_name}' created successfully")

이 코드에서는 먼저 컬렉션이 이미 존재하는지 확인하고 존재하는 경우 삭제합니다. 이렇게 하면 깨끗한 상태에서 시작할 수 있습니다. 컬렉션의 크기는 1024이며, 이는 Twelve Labs 임베딩의 출력 크기와 일치합니다.

Twelve Labs 임베드 API로 임베딩 생성하기

Twelve Labs 임베드 API를 사용하여 동영상에 대한 임베딩을 생성하기 위해 Twelve Labs Python SDK를 사용합니다. 이 프로세스에는 임베딩 작업을 생성하고, 완료될 때까지 기다린 다음, 결과를 검색하는 과정이 포함됩니다. 이를 구현하는 방법은 다음과 같습니다:

먼저 Twelve Labs SDK가 설치되어 있는지 확인하고 필요한 모듈을 가져옵니다:

from twelvelabs import TwelveLabs
from twelvelabs.models.embed import EmbeddingsTask
import os

# Retrieve the API key from environment variables
TWELVE_LABS_API_KEY = os.getenv('TWELVE_LABS_API_KEY')

Twelve Labs 클라이언트를 초기화합니다:

twelvelabs_client = TwelveLabs(api_key=TWELVE_LABS_API_KEY)

지정된 동영상 URL에 대한 임베딩을 생성하는 함수를 만듭니다:

def generate_embedding(video_url):
    """
    Generate embeddings for a given video URL using the Twelve Labs API.

    This function creates an embedding task for the specified video URL using
    the Marengo-retrieval-2.6 engine. It monitors the task progress and waits
    for completion. Once done, it retrieves the task result and extracts the
    embeddings along with their associated metadata.

    Args:
        video_url (str): The URL of the video to generate embeddings for.

    Returns:
        tuple: A tuple containing two elements:
            1. list: A list of dictionaries, where each dictionary contains:
                - 'embedding': The embedding vector as a list of floats.
                - 'start_offset_sec': The start time of the segment in seconds.
                - 'end_offset_sec': The end time of the segment in seconds.
                - 'embedding_scope': The scope of the embedding (e.g., 'shot', 'scene').
            2. EmbeddingsTaskResult: The complete task result object from Twelve Labs API.

    Raises:
        Any exceptions raised by the Twelve Labs API during task creation,
        execution, or retrieval.
    """

    # Create an embedding task
    task = twelvelabs_client.embed.task.create(
        engine_name="Marengo-retrieval-2.6",
        video_url=video_url
    )
    print(f"Created task: id={task.id} engine_name={task.engine_name} status={task.status}")

    # Define a callback function to monitor task progress
    def on_task_update(task: EmbeddingsTask):
        print(f"  Status={task.status}")

    # Wait for the task to complete
    status = task.wait_for_done(
        sleep_interval=2,
        callback=on_task_update
    )
    print(f"Embedding done: {status}")

    # Retrieve the task result
    task_result = twelvelabs_client.embed.task.retrieve(task.id)

    # Extract and return the embeddings
    embeddings = []
    for v in task_result.video_embeddings:
        embeddings.append({
            'embedding': v.embedding.float,
            'start_offset_sec': v.start_offset_sec,
            'end_offset_sec': v.end_offset_sec,
            'embedding_scope': v.embedding_scope
        })
    
    return embeddings, task_result

이 함수를 사용하여 동영상에 대한 임베딩을 생성합니다:

# Example usage
video_url = "https://example.com/your-video.mp4"

# Generate embeddings for the video
embeddings, task_result = generate_embedding(video_url)

print(f"Generated {len(embeddings)} embeddings for the video")
for i, emb in enumerate(embeddings):
    print(f"Embedding {i+1}:")
    print(f"  Scope: {emb['embedding_scope']}")
    print(f"  Time range: {emb['start_offset_sec']} - {emb['end_offset_sec']} seconds")
    print(f"  Embedding vector (first 5 values): {emb['embedding'][:5]}")
    print()

이 구현을 통해 Twelve Labs Embed API를 사용하여 모든 동영상 URL에 대한 임베딩을 생성할 수 있습니다. generate_embedding 함수는 작업 생성부터 결과 검색까지 전체 프로세스를 처리합니다. 이 함수는 메타데이터(시간 범위 및 범위)와 함께 임베딩 벡터가 포함된 사전 목록을 반환하며, 프로덕션 환경에서는 네트워크 문제나 API 제한과 같은 잠재적인 오류를 처리해야 한다는 점을 잊지 마세요. 또한 특정 사용 사례에 따라 재시도 또는 보다 강력한 오류 처리를 구현할 수도 있습니다.

Milvus에 임베딩 삽입하기

Twelve Labs Embed API를 사용하여 임베딩을 생성한 다음 단계는 이러한 임베딩을 메타데이터와 함께 Milvus 컬렉션에 삽입하는 것입니다. 이 과정을 통해 나중에 효율적인 유사도 검색을 위해 동영상 임베딩을 저장하고 색인을 생성할 수 있습니다.

Milvus에 임베딩을 삽입하는 방법은 다음과 같습니다:

def insert_embeddings(milvus_client, collection_name, task_result, video_url):
    """
    Insert embeddings into the Milvus collection.

    Args:
        milvus_client: The Milvus client instance.
        collection_name (str): The name of the Milvus collection to insert into.
        task_result (EmbeddingsTaskResult): The task result containing video embeddings.
        video_url (str): The URL of the video associated with the embeddings.

    Returns:
        MutationResult: The result of the insert operation.

    This function takes the video embeddings from the task result and inserts them
    into the specified Milvus collection. Each embedding is stored with additional
    metadata including its scope, start and end times, and the associated video URL.
    """
    data = []

    for i, v in enumerate(task_result.video_embeddings):
        data.append({
            "id": i,
            "vector": v.embedding.float,
            "embedding_scope": v.embedding_scope,
            "start_offset_sec": v.start_offset_sec,
            "end_offset_sec": v.end_offset_sec,
            "video_url": video_url
        })

    insert_result = milvus_client.insert(collection_name=collection_name, data=data)
    print(f"Inserted {len(data)} embeddings into Milvus")
    return insert_result

# Usage example
video_url = "https://example.com/your-video.mp4"

# Assuming this function exists from previous step
embeddings, task_result = generate_embedding(video_url)

# Insert embeddings into the Milvus collection
insert_result = insert_embeddings(milvus_client, collection_name, task_result, video_url)
print(insert_result)

이 기능은 임베딩 벡터, 시간 범위, 소스 동영상 URL 등 모든 관련 메타데이터를 포함하여 삽입할 데이터를 준비합니다. 그런 다음 Milvus 클라이언트를 사용하여 이 데이터를 지정된 컬렉션에 삽입합니다.

유사도 검색 수행

임베딩이 Milvus에 저장되면 유사도 검색을 수행하여 쿼리 벡터를 기반으로 가장 관련성이 높은 동영상 세그먼트를 찾을 수 있습니다. 이 기능을 구현하는 방법은 다음과 같습니다:

def perform_similarity_search(milvus_client, collection_name, query_vector, limit=5):
    """
    Perform a similarity search on the Milvus collection.

    Args:
        milvus_client: The Milvus client instance.
        collection_name (str): The name of the Milvus collection to search in.
        query_vector (list): The query vector to search for similar embeddings.
        limit (int, optional): The maximum number of results to return. Defaults to 5.

    Returns:
        list: A list of search results, where each result is a dictionary containing
              the matched entity's metadata and similarity score.

    This function searches the specified Milvus collection for embeddings similar to
    the given query vector. It returns the top matching results, including metadata
    such as the embedding scope, time range, and associated video URL for each match.
    """
    search_results = milvus_client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=limit,
        output_fields=["embedding_scope", "start_offset_sec", "end_offset_sec", "video_url"]
    )

    return search_results
    
# define the query vector
# We use the embedding inserted previously as an example. In practice, you can replace it with any video embedding you want to query.
query_vector = task_result.video_embeddings[0].embedding.float

# Perform a similarity search on the Milvus collection
search_results = perform_similarity_search(milvus_client, collection_name, query_vector)

print("Search Results:")
for i, result in enumerate(search_results[0]):
    print(f"Result {i+1}:")
    print(f"  Video URL: {result['entity']['video_url']}")
    print(f"  Time Range: {result['entity']['start_offset_sec']} - {result['entity']['end_offset_sec']} seconds")
    print(f"  Similarity Score: {result['distance']}")
    print()

이 구현은 다음을 수행합니다:

1. 쿼리 벡터를 받아 Milvus 컬렉션에서 유사한 임베딩을 검색하는 함수 perform_similarity_search를 정의합니다.
2. Milvus 클라이언트의 검색 방법을 사용하여 가장 유사한 벡터를 찾습니다.
3. 일치하는 동영상 세그먼트에 대한 메타데이터를 포함하여 검색할 출력 필드를 지정합니다.
4. 쿼리 동영상에 이 함수를 사용하여 먼저 임베딩을 생성한 다음 검색에 사용하는 예시를 제공합니다.
5. 관련 메타데이터 및 유사도 점수를 포함한 검색 결과를 인쇄합니다.

이러한 기능을 구현하면 Milvus에 동영상 임베딩을 저장하고 유사도 검색을 수행하는 완벽한 워크플로우를 만들 수 있습니다. 이 설정을 통해 Twelve Labs의 임베드 API에서 생성된 멀티모달 임베딩을 기반으로 유사한 동영상 콘텐츠를 효율적으로 검색할 수 있습니다.

성능 최적화

자, 이제 이 앱을 한 단계 업그레이드해 봅시다! 대규모 동영상 컬렉션을 다룰 때는 성능이 핵심입니다. 최적화를 위해서는 임베딩 생성 및 삽입을 위한 일괄 처리를 Milvus에 구현해야 합니다. 이렇게 하면 여러 동영상을 동시에 처리하여 전체 처리 시간을 크게 단축할 수 있습니다. 또한 Milvus의 파티셔닝 기능을 활용하여 동영상 카테고리 또는 기간별로 데이터를 보다 효율적으로 구성할 수 있습니다. 이렇게 하면 관련성이 높은 파티션만 검색할 수 있어 쿼리 속도가 빨라집니다.

또 다른 최적화 요령은 자주 액세스하는 임베딩이나 검색 결과에 캐싱 메커니즘을 사용하는 것입니다. 이렇게 하면 인기 있는 쿼리의 응답 시간을 크게 개선할 수 있습니다. 특정 데이터 세트와 쿼리 패턴에 따라 Milvus의 인덱스 매개변수를 미세 조정하는 것을 잊지 마세요. 여기서 약간의 조정만으로도 검색 성능을 향상시키는 데 큰 도움이 될 수 있습니다.

고급 기능

이제 앱을 돋보이게 하는 몇 가지 멋진 기능을 추가해 보겠습니다! 텍스트와 동영상 쿼리를 결합하는 하이브리드 검색을 구현할 수 있습니다. 실제로 Twelve Labs 임베드 API는 텍스트 쿼리에 대한 텍스트 임베딩도 생성할 수 있습니다. 사용자가 텍스트 설명과 샘플 동영상 클립을 모두 입력할 수 있다고 상상해 보세요. 이 두 가지 모두에 대한 임베딩을 생성하고 Milvus에서 가중치 검색을 수행할 수 있습니다. 이렇게 하면 매우 정확한 결과를 얻을 수 있습니다.

또 다른 멋진 추가 기능은 동영상 내 일시적 검색입니다. 긴 동영상을 각각 고유한 임베딩이 있는 작은 세그먼트로 나눌 수 있습니다. 이렇게 하면 사용자는 전체 클립뿐만 아니라 동영상 내에서 특정 순간을 찾을 수 있습니다. 그리고 기본적인 동영상 분석 기능도 추가하면 어떨까요? 임베딩을 사용하여 유사한 동영상 세그먼트를 클러스터링하고, 추세를 감지하거나, 대규모 동영상 컬렉션에서 이상값을 식별할 수도 있습니다.

오류 처리 및 로깅

현실을 직시하자, 문제가 발생할 수 있으며, 문제가 발생하면 이에 대비해야 합니다. 강력한 오류 처리 기능을 구현하는 것이 중요합니다. API 호출과 데이터베이스 작업을 시도 예외 블록으로 래핑하여 실패 시 사용자에게 유익한 오류 메시지를 제공해야 합니다. 네트워크 관련 문제의 경우 기하급수적 백오프를 사용하여 재시도를 구현하면 일시적인 결함을 원활하게 처리하는 데 도움이 될 수 있습니다.

로깅은 디버깅과 모니터링을 위한 가장 좋은 친구입니다. 애플리케이션 전체에서 중요한 이벤트, 오류 및 성능 메트릭을 추적하려면 Python의 로깅 모듈을 사용해야 합니다. 개발을 위한 DEBUG, 일반 운영을 위한 INFO, 중요한 문제를 위한 ERROR 등 다양한 로그 수준을 설정해 보겠습니다. 그리고 파일 크기를 관리하기 위해 로그 로테이션을 구현하는 것도 잊지 마세요. 적절한 로깅을 설정하면 문제를 신속하게 파악하고 해결할 수 있으므로 확장 시에도 동영상 검색 앱이 원활하게 실행될 수 있습니다.

결론

축하합니다! 이제 Twelve Labs의 Embed API와 Milvus를 사용하여 강력한 시맨틱 동영상 검색 애플리케이션을 구축하셨습니다. 이 통합을 통해 전례 없는 정확도와 효율성으로 동영상 콘텐츠를 처리, 저장 및 검색할 수 있습니다. 멀티모달 임베딩을 활용하면 동영상 데이터의 뉘앙스를 이해하는 시스템을 구축하여 콘텐츠 검색, 추천 시스템 및 고급 동영상 분석에 대한 흥미로운 가능성을 열 수 있습니다.

애플리케이션을 계속 개발하고 개선해 나가면서 Twelve Labs의 고급 임베딩 생성 기능과 Milvus의 확장 가능한 벡터 스토리지의 조합은 훨씬 더 복잡한 동영상 이해 문제를 해결할 수 있는 강력한 기반을 제공한다는 점을 기억하세요. 앞서 설명한 고급 기능을 실험해보고 동영상 검색 및 분석의 한계를 뛰어넘어보시기 바랍니다.