🚀 免費嘗試 Zilliz Cloud,完全托管的 Milvus,體驗速度提升 10 倍!立即嘗試

milvus-logo
LFAI
主頁
  • 整合
  • Home
  • Docs
  • 整合

  • 嵌入模型

  • 十二實驗室

進階視訊搜尋:利用 Twelve Labs 和 Milvus 進行語意檢索

簡介

歡迎來到這個使用Twelve Labs Embed API和 Milvus 實現語義視頻搜索的綜合教程。在本指南中,我們將探討如何利用Twelve Labs 先進的多模態嵌入(multimodal embeddings)和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 key 設定為環境變數,以確保安全性:

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_twelvelabs_demo.db 的檔案中。這種以檔案為基礎的方式非常適合開發和測試用途。

建立視訊嵌入的 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 Embed API 生成嵌入式内容

要使用 Twelve Labs Embed 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 的 embeddings:

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 產生 embeddings。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. 定義一個函式 perform_similarity_search,接受查詢向量並在 Milvus 套件中搜尋相似的嵌入式元件。
  2. 使用 Milvus 客戶端的搜尋方法找出最相似的向量。
  3. 指定我們要擷取的輸出欄位,包括匹配視訊片段的元資料。
  4. 提供一個範例,說明如何將此函式用於查詢視訊,首先產生其嵌入,然後用於搜尋。
  5. 列印搜尋結果,包括相關的元資料和相似度得分。

透過實作這些函式,您已建立了完整的工作流程,可在 Milvus 中儲存影片內嵌並執行相似性搜尋。此設定可根據 Twelve Labs 的 Embed API 所產生的多模態嵌入內容,有效率地檢索相似的視訊內容。

優化效能

好了,讓我們將此應用程式提升到更高層次!在處理大規模的視訊集合時,效能是關鍵。為了最佳化,我們應該執行批次處理,以產生嵌入並插入 Milvus。這樣,我們就能同時處理多個視訊,大幅縮短整體處理時間。此外,我們還可以利用Milvus 的分割功能來更有效率地組織資料,例如依視訊類別或時間段來組織資料。這可讓我們只搜尋相關的分區,從而加快查詢速度。

另一個優化技巧是對經常存取的嵌入或搜尋結果使用快取機制。這可大幅改善常用查詢的回應時間。別忘了根據您的特定資料集和查詢模式來微調 Milvus 的索引參數- 稍作調整就能大大提升搜尋效能。

進階功能

現在,讓我們加入一些很酷的功能,讓我們的應用程式脫穎而出!我們可以實作混合搜尋,結合文字與視訊查詢。事實上,Twelve Labs Embed API 也可以為您的文字查詢產生文字嵌入。試想一下,讓使用者同時輸入文字描述和範例視訊片段 - 我們會為兩者產生嵌入式內容,並在 Milvus 中執行加權搜尋。這將帶給我們超精確的結果。

另一個很棒的新增功能是在視訊中進行時間搜尋我們可以將長影片分割成較小的片段,每個片段都有自己的 embedding。如此一來,使用者就可以找到影片中的特定時刻,而不只是整個片段。而且,為什麼不加入一些基本的視訊分析呢?我們可以使用 embeddings 來聚類相似的視訊片段、偵測趨勢,甚至找出大型視訊集合中的異常值。

錯誤處理和日誌

讓我們面對現實吧,事情可能會出錯,當出錯時,我們需要做好準備。實施強大的錯誤處理是非常重要的。我們應該在 try-except 區塊中包覆我們的 API 呼叫和資料庫作業,並在發生故障時向使用者提供資訊豐富的錯誤訊息。對於與網路相關的問題,使用指數遞減 (exponential backoff) 實作重試 (retries) 有助於從容處理暫時的故障。

至於日誌,它是我們調試和監控的好朋友。我們應該使用Python 的日誌模組來追蹤整個應用程式中的重要事件、錯誤和效能指標。讓我們設定不同的日誌層級 - DEBUG 用於開發,INFO 用於一般操作,ERROR 用於重要問題。別忘了實施日誌輪換以管理檔案大小。有了適當的日誌,我們就能快速找出並解決問題,確保我們的視訊搜尋應用程式即使在擴充時也能順暢運作。

總結

恭喜您!您現在已經使用 Twelve Labs 的 Embed API 和 Milvus 建立了一個功能強大的語意視訊搜尋應用程式。此整合功能可讓您以前所未有的精確度和效率處理、儲存和擷取視訊內容。透過利用多模態嵌入,您建立了一個能夠理解視訊資料細微差異的系統,為內容發現、推薦系統和進階視訊分析開啟了令人振奮的可能性。

當您繼續開發和完善您的應用程式時,請記住 Twelve Labs 的進階嵌入生成與 Milvus 的可擴展向量儲存的結合,為應付更複雜的視訊理解挑戰提供了堅實的基礎。我們鼓勵您嘗試使用所討論的進階功能,並推動視訊搜尋與分析的可能性。