Summary: The TwelveLabs handler for MindsDB allows you to search and summarize video content directly within MindsDB, streamlining the integration of these features into your applications.

Description: This guide outlines how you can use the handler and how the handler interfaces with the TwelveLabs Video Understanding Platform to combine TwelveLabs’ state-of-the-art foundation models for video understanding with MindsDB’s platform for building customized AI solutions.

Code explanation: Our blog post, Build a Powerful Video Summarization Tool with Twelve Lans, MindsDB, and Slack, walks you through the steps required to configure the TwelveLabs integration in MindsDB, deploy the TwelveLabs model for summarization within MindsDB, and automate the whole flow through a Slack bot that will periodically post the video summarizations as announcements.

GitHub: TwelveLabs Handler.

Use the handler

This section assumes the following:

To use the platform, you need an API key:

1
If you don’t have an account, sign up for a free account.
2
Go to the API Keys page.
3
If you need to create a new key, select the Create API Key button. Enter a name and set the expiration period. The default is 12 months.
4
Select the Copy icon next to your key to copy it to your clipboard.
You have a running MindsDB database inside of a Docker container. If not, see the Docker for MindsDB section of the MindsDB documentation

Typically, the steps for using the handler are as follows:

Install the required dependencies inside the Docker container:
```
pip install mindsdb[twelve_labs] 
```
Open the MindsDB SQL Editor.
Create an ML engine. Use the CREATE ML_ENGINE statement, replacing the placeholders surrounded by <> with your values:

CREATE ML_ENGINE <YOUR_ENGINE_NAME> 
from twelve_labs
USING
    twelve_labs_api_key = '<YOUR_API_KEY>'

The example below creates an ML engine named twelve_labs_engine:

CREATE ML_ENGINE twelve_labs_engine 
from twelve_labs
USING
    twelve_labs_api_key = 'tlk_111'

Create a model. Use the CREATE_MODEL statement to create a model. The PREDICT clause specifies the name of the column that will contain the results of the task. The USING clause specifies the parameters for the model. The parameters depend on the task you want to perform. The available tasks are search and summarization, and the parameters for each task are described in the Creating Models section of the handler’s GitHub Readme file.
The example below creates a model for the search task:

CREATE MODEL mindsdb.twelve_labs_search
PREDICT search_results
USING
  engine = 'twelve_labs_engine',
  task = 'search',
  engine_id = 'marengo2.7',
  index_name = 'index_1',
  index_options = ['visual', 'conversation', 'text_in_video', 'logo'],
  video_urls = ['https://.../video_1.mp4', 'https://.../video_2.mp4'],
  search_options = ['visual', 'conversation', 'text_in_video', 'logo'],
  search_query_column = 'query';

The example below creates a model for the summarization task:

CREATE MODEL mindsdb.twelve_labs_summarization
PREDICT summarization_results
USING
  engine = 'twelve_labs_engine',
  task = 'summarization',
  engine_id = 'pegasus1',
  index_name = 'index_1',
  index_options = ['visual', 'conversation'],
  video_urls = ['https://.../video_1.mp4', 'https://.../video_2.mp4'],
  summarization_type = 'summary';

(Optional) Check the status of the video indexing process. The TwelveLabs Video Understanding Platform requires some time to index videos. You can search or summarize your videos only after the indexing process is complete. Use the DESCRIBE statement to check the status of the indexing process, replacing the placeholder surrounded by <> with your the name your model:
```
DESCRIBE mindsdb.<YOUR_MODEL_NAME>;
```
The example below checks the status of a model named twelve_labs_summarization:
```
DESCRIBE mindsdb.twelve_labs_summarization;
```
You should see the status as complete in the STATUS column. In case of an error, check the ERROR column, which contains detailed information about the error.
Retrieve the identifiers of the indexed videos. Perform this step if you want to summarize a video. To retrieve the identifiers, use the DESCRIBE statement on the indexed_videos table of your model, replacing the placeholder surrounded by <> with the name of your model:
```
DESCRIBE mindsdb.<YOUR_MODEL_NAME>.indexed_videos;
```
The example below retrieves the identifiers of the videos uploaded to a model named twelve_labs_summarization:
```
DESCRIBE mindsdb.twelve_labs_summarization.indexed_videos;
```
Make predictions. Use the SELECT statement to make predictions using the model created in the previous step. The WHERE clause specifies the condition for the prediction. The condition depends on the task you want to perform. The available tasks are search and summarization, and the conditions for each task are described in the Making Predictions section of the handler’s GitHub Readme file:
The example below performs a search request. Ensure you replace the placeholder surrounded by <> with your query

Search:
In the SQL query below, ensure you replace the placeholders surrounded by <> with your values:
```
SELECT *
FROM mindsdb.<YOUR_MODEL_NAME>
WHERE query = '<YOUR_QUERY>';
```
The example below makes predictions for the search task using a model named twelve_labs_search:
```
SELECT *
FROM mindsdb.twelve_labs_search
WHERE query = 'Soccer player scoring a goal';
```
Summarize:

In the SQL query below, ensure you replace the placeholders surrounded by <> with your values:
```
SELECT *
FROM mindsdb.<YOUR_MODEL_NAME>
WHERE video_id = '<YOUR_VIDEO_ID>';
```
The example below makes predictions for the summarization task using a model named twelve_labs_summarization:
```
SELECT *
FROM mindsdb.twelve_labs_summarization
WHERE video_id = '660bfa6766995fbd9fd662ee';
```

Integration with TwelveLabs

For brevity, the sections below outline the key components for integrating MindsDB with the TwelveLabs Video Understanding Platform:

Initialize a client
Create indexes
Upload videos
Perform downstream tasks such as search or classification.

For all the components, refer to the TwelveLabs Handler page on GitHub.

Initialize a client

The constructor sets up a new TwelveLabsAPIClient object that establishes a connection to the TwelveLabs Video Understanding Platform:

def __init__(self, api_key: str, base_url: str = None):
    """
    The initializer for the TwelveLabsAPIClient.
    Parameters
    ----------
    api_key : str
        The TwelveLabs API key.
    base_url : str, Optional
        The base URL for the TwelveLabs API. Defaults to the base URL in the TwelveLabs handler settings.
    """
    self.api_key = api_key
    self.headers = {
        'Content-Type': 'application/json',
        'x-api-key': self.api_key
    }
    self.base_url = base_url if base_url else twelve_labs_handler_config.BASE_URL

Create indexes

To create indexes, the create_index method invokes the POST method of the /indexes endpoint:

def create_index(self, index_name: str, index_options: List[str], engine_id: Optional[str] = None, addons: Optional[List[str]] = None) -> str:
    """
    Create an index.
    Parameters
    ----------
    index_name : str
        Name of the index to be created.
    index_options : List[str]
        List of that specifies how the platform will process the videos uploaded to this index.
    engine_id : str, Optional
        ID of the engine. If not provided, the default engine is used.
    addons : List[str], Optional
        List of addons that should be enabled for the index.
    Returns
    -------
    str
        ID of the created index.
    """
    # TODO: change index_options to engine_options?
    # TODO: support multiple engines per index?
    body = {
        "index_name": index_name,
        "engines": [{
            "engine_name": engine_id if engine_id else twelve_labs_handler_config.DEFAULT_ENGINE_ID,
            "engine_options": index_options
        }],
        "addons": addons,
    }
    result = self._submit_request(
        method="POST",
        endpoint="/indexes",
        data=body,
    )
    logger.info(f"Index {index_name} successfully created.")
    return result['_id']

Upload videos

To upload videos to the TwelveLabs Video Understanding Platform and index them, the handler invokes the POST method of the /tasks endpoint:

def create_video_indexing_tasks(self, index_id: str, video_urls: List[str] = None, video_files: List[str] = None) -> List[str]:
    """
    Create video indexing tasks.
    Parameters
    ----------
    index_id : str
        ID of the index.
    video_urls : List[str], Optional
        List of video urls to be indexed. Either video_urls or video_files should be provided. This validation is handled by TwelveLabsHandlerModel.
    video_files : List[str], Optional
        List of videos to be indexed. Either video_urls or video_files should be provided. This validation is handled by TwelveLabsHandlerModel.
    Returns
    -------
    List[str]
        List of task IDs created.
    """
    task_ids = []
    if video_urls:
        logger.info("video_urls has been set, therefore, it will be given precedence.")
        logger.info("Creating video indexing tasks for video urls.")
        for video_url in video_urls:
            task_ids.append(
                self._create_video_indexing_task(
                    index_id=index_id,
                    video_url=video_url
                )
            )
    elif video_files:
        logger.info("video_urls has not been set, therefore, video_files will be used.")
        logger.info("Creating video indexing tasks for videos.")
        for video_file in video_files:
            task_ids.append(
                self._create_video_indexing_task(
                    index_id=index_id,
                    video_file=video_file
                )
            )
    return task_ids
def _create_video_indexing_task(self, index_id: str, video_url: str = None, video_file: str = None) -> str:
    """
    Create a video indexing task.
    Parameters
    ----------
    index_id : str
        ID of the index.
    video_url : str, Optional
        URL of the video to be indexed. Either video_url or video_file should be provided. This validation is handled by TwelveLabsHandlerModel.
    video_file : str, Optional
        Path to the video file to be indexed. Either video_url or video_file should be provided. This validation is handled by TwelveLabsHandlerModel.
    Returns
    -------
    str
        ID of the created task.
    """
    body = {
        "index_id": index_id,
    }
    file_to_close = None
    if video_url:
        body['video_url'] = video_url
    elif video_file:
        import mimetypes
        # WE need the file open for the duration of the request. Maybe simplify it with context manager later, but needs _create_video_indexing_task re-written
        file_to_close = open(video_file, 'rb')
        mime_type, _ = mimetypes.guess_type(video_file)
        body['video_file'] = (file_to_close.name, file_to_close, mime_type)
    result = self._submit_multi_part_request(
        method="POST",
        endpoint="/tasks",
        data=body,
    )
    if file_to_close:
        file_to_close.close()
    task_id = result['_id']
    logger.info(f"Created video indexing task {task_id} for {video_url if video_url else video_file} successfully.")
    # update the video title
    video_reference = video_url if video_url else video_file
    task = self._get_video_indexing_task(task_id=task_id)
    self._update_video_metadata(
        index_id=index_id,
        video_id=task['video_id'],
        metadata={
            "video_reference": video_reference
        }
    )
    return task_id

Once the video has been uploaded to the platform, the handler monitors the indexing process using the GET method of the /tasks/{task_id} endpoint:

def poll_for_video_indexing_tasks(self, task_ids: List[str]) -> None:
    """
    Poll for video indexing tasks to complete.
    Parameters
    ----------
    task_ids : List[str]
        List of task IDs to be polled.
    Returns
    -------
    None
    """
    for task_id in task_ids:
        logger.info(f"Polling status of video indexing task {task_id}.")
        is_task_running = True
        while is_task_running:
            task = self._get_video_indexing_task(task_id=task_id)
            status = task['status']
            logger.info(f"Task {task_id} is in the {status} state.")
            wait_durtion = task['process']['remain_seconds'] if 'process' in task else twelve_labs_handler_config.DEFAULT_WAIT_DURATION
            if status in ('pending', 'indexing', 'validating'):
                logger.info(f"Task {task_id} will be polled again in {wait_durtion} seconds.")
                time.sleep(wait_durtion)
            elif status == 'ready':
                logger.info(f"Task {task_id} completed successfully.")
                is_task_running = False
            else:
                logger.error(f"Task {task_id} failed with status {task['status']}.")
                # TODO: update Exception to be more specific
                raise Exception(f"Task {task_id} failed with status {task['status']}.")
    logger.info("All videos indexed successffully.")

Perform downstream tasks

The handler supports the following downstream tasks - search and summarize videos. See the sections below for details.

Search videos

To perform search requests, the handler invokes the POST method of the /search endpoint:

def search_index(self, index_id: str, query: str, search_options: List[str]) -> Dict:
    """
    Search an index.
    Parameters
    ----------
    index_id : str
        ID of the index.
    query : str
        Query to be searched.
    search_options : List[str]
        List of search options to be used.
    Returns
    -------
    Dict
        Search results.
    """
    body = {
        "index_id": index_id,
        "query": query,
        "search_options": search_options
    }
    data = []
    result = self._submit_request(
        method="POST",
        endpoint="/search",
        data=body,
    )
    data.extend(result['data'])
    while 'next_page_token' in result['page_info']:
        result = self._submit_request(
            method="GET",
            endpoint=f"/search/{result['page_info']['next_page_token']}"
        )
        data.extend(result['data'])
    logger.info(f"Search for index {index_id} completed successfully.")
    return data

Summarize videos

To summarize videos, the handler invokes the POST method of the summarize endpoint:

    def summarize_videos(self, video_ids: List[str], summarization_type: str, prompt: str) -> Dict:
        """
        Summarize videos.
        Parameters
        ----------
        video_ids : List[str]
            List of video IDs.
        summarization_type : str
            Type of the summary to be generated. Supported types are 'summary', 'chapter' and 'highlight'.
        prompt: str
            Prompt to be used for the Summarize task
        Returns
        -------
        Dict
            Summary of the videos.
        """
        results = []
        results = [self.summarize_video(video_id, summarization_type, prompt) for video_id in video_ids]
        logger.info(f"Summarized videos {video_ids} successfully.")
        return results
    def summarize_video(self, video_id: str, summarization_type: str, prompt: str) -> Dict:
        """
        Summarize a video.
        Parameters
        ----------
        video_id : str
            ID of the video.
        summarization_type : str
            Type of the summary to be generated. Supported types are 'summary', 'chapter' and 'highlight'.
        prompt: str
            Prompt to be used for the Summarize task
        Returns
        -------
        Dict
            Summary of the video.
        """
        body = {
            "video_id": video_id,
            "type": summarization_type,
            "prompt": prompt
        }
        result = self._submit_request(
            method="POST",
            endpoint="/summarize",
            data=body,
        )
        logger.info(f"Video {video_id} summarized successfully.")
        return result

Next steps

After reading this page, you have several options:

Use the handler: Inspect the TwelveLabs Handler page on GitHub to better understand its features and start using it in your applications.
Explore further: Try the applications built by the community or our sample applications to get more insights into the TwelveLabs Video Understanding Platform’s diverse capabilities and learn more about integrating the platform into your applications.