Overview

What You'll Build

What You'll Learn

What You'll Need

Stage Your Videos

CREATE DATABASE IF NOT EXISTS TWELVE_LABS_DEMO;
CREATE SCHEMA IF NOT EXISTS TWELVE_LABS_DEMO.VIDEO_INTELLIGENCE;
USE SCHEMA TWELVE_LABS_DEMO.VIDEO_INTELLIGENCE;

CREATE STAGE IF NOT EXISTS VIDEO_STAGE
  DIRECTORY  = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

snow stage copy ./videos/*.mp4 @TWELVE_LABS_DEMO.VIDEO_INTELLIGENCE.VIDEO_STAGE

ALTER STAGE VIDEO_STAGE REFRESH;
SELECT RELATIVE_PATH, SIZE FROM DIRECTORY(@VIDEO_STAGE);

aside positive Already have video in S3, GCS, or Azure? Create an external stage on a storage integration instead. The rest of this guide works the same with either kind of stage.

CREATE STAGE IF NOT EXISTS SEARCH_UPLOADS
  DIRECTORY  = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

Embed Videos with Marengo

CREATE OR REPLACE TABLE VIDEO_EMBEDDINGS AS
WITH raw AS (
    SELECT
        REPLACE(RELATIVE_PATH, '.mp4', '') AS episode,
        AI_MULTI_EMBED(
            'twelvelabs-marengo-embed-3-0',
            TO_FILE('@VIDEO_STAGE', RELATIVE_PATH)
        ) AS emb
    FROM DIRECTORY(@VIDEO_STAGE)
)
SELECT
    episode,
    f.value['embedding']::VECTOR(FLOAT, 512) AS embedding_vec,
    f.value['embedding_option']::STRING      AS modality,
    f.value['start_sec']::FLOAT              AS start_sec,
    f.value['end_sec']::FLOAT                AS end_sec
FROM raw, LATERAL FLATTEN(input => raw.emb['value']) f;

SELECT modality, COUNT(*) AS segments, COUNT(DISTINCT episode) AS videos
FROM VIDEO_EMBEDDINGS
GROUP BY modality;

aside positive Marengo puts text, images, and video in the same vector space. That's why a text query or an uploaded photo can be matched directly against video segments. You'll use this in the app.

Index with Cortex Search

CREATE OR REPLACE CORTEX SEARCH SERVICE VIDEO_SEARCH_SERVICE
  TEXT INDEXES   EPISODE
  VECTOR INDEXES EMBEDDING_VEC
  ATTRIBUTES     EPISODE, MODALITY, START_SEC, END_SEC
  WAREHOUSE      = <your_warehouse>
  TARGET_LAG     = '1 day'
AS (
    SELECT EPISODE, EMBEDDING_VEC, MODALITY, START_SEC, END_SEC
    FROM VIDEO_EMBEDDINGS
    WHERE MODALITY = 'visual'
);

(Optional) Analyze Video with Pegasus

aside negative Private Preview: twelvelabs-pegasus-1-2 is not yet in the public AI_COMPLETE model list. If your account isn't enrolled, skip this step and leave ENABLE_PEGASUS = False in the app. Search works without it. Contact your Snowflake account team to request access.

SELECT AI_COMPLETE(
    'twelvelabs-pegasus-1-2',
    'Summarize this video in 3 bullets, then list the main characters as JSON.',
    TO_FILE('@VIDEO_STAGE', '<your_video>.mp4')
) AS analysis;

Build the Streamlit App

import json
import streamlit as st
from snowflake.core import Root

st.set_page_config(page_title="Twelve Labs x Snowflake", layout="wide")

DB, SCHEMA = "TWELVE_LABS_DEMO", "VIDEO_INTELLIGENCE"
VIDEO_STAGE = f"@{DB}.{SCHEMA}.VIDEO_STAGE"
UPLOAD_STAGE = f"@{DB}.{SCHEMA}.SEARCH_UPLOADS"
ENABLE_PEGASUS = False  # Pegasus is Private Preview; set True if your account is enrolled

conn = st.connection("snowflake")
session = conn.session()
search_svc = (Root(session).databases[DB].schemas[SCHEMA]
              .cortex_search_services["VIDEO_SEARCH_SERVICE"])


def embed_query(text=None, image=None) -> list:
    """Embed a text string or an uploaded image with Marengo."""
    if image is not None:
        name = image.name.replace(" ", "_")
        session.file.put_stream(image, f"{UPLOAD_STAGE}/{name}",
                                auto_compress=False, overwrite=True)
        sql = f"""SELECT AI_MULTI_EMBED('twelvelabs-marengo-embed-3-0',
                  TO_FILE('{UPLOAD_STAGE}', ?)):value[0]['embedding']::VARCHAR AS v"""
        row = session.sql(sql, params=[name]).collect()[0]
    else:
        sql = """SELECT AI_MULTI_EMBED('twelvelabs-marengo-embed-3-0', ?)
                 :value[0]['embedding']::VARCHAR AS v"""
        row = session.sql(sql, params=[text]).collect()[0]
    return json.loads(row["V"])


def presigned_url(stage: str, path: str) -> str:
    sql = f"SELECT GET_PRESIGNED_URL({stage}, ?, 3600) AS url"
    return session.sql(sql, params=[path]).collect()[0]["URL"]


def fmt(sec: float) -> str:
    return f"{int(sec // 60)}:{int(sec % 60):02d}"


tab_names = ["Embed + Search"] + (["Analyze"] if ENABLE_PEGASUS else [])
tabs = st.tabs(tab_names)
tab_search = tabs[0]

# ── Search ──────────────────────────────────────────
with tab_search:
    query = st.text_input("Describe a scene")
    image = st.file_uploader("...or search with an image", type=["jpg", "jpeg", "png"])

    if st.button("Search", type="primary") and (query or image):
        with st.spinner("Searching..."):
            vec = embed_query(text=query, image=image)
            resp = search_svc.search(
                multi_index_query={"EMBEDDING_VEC": [{"vector": vec}]},
                columns=["EPISODE", "START_SEC", "END_SEC"],
                limit=50,
            )
            # Adjacent segments are usually the same scene; keep hits spread out.
            results, kept = [], {}
            for r in resp.results:
                ep, start = r["EPISODE"], float(r["START_SEC"])
                prior = kept.setdefault(ep, [])
                if len(prior) < 2 and all(abs(start - p) >= 45 for p in prior):
                    prior.append(start)
                    results.append(r)
                if len(results) == 6:
                    break
            urls = {ep: presigned_url(VIDEO_STAGE, f"{ep}.mp4")
                    for ep in {r["EPISODE"] for r in results}}
            st.session_state.hits = (results, urls)

    if "hits" in st.session_state:
        results, urls = st.session_state.hits
        cols = st.columns(2)
        for i, r in enumerate(results):
            with cols[i % 2].container(border=True):
                start, end = float(r["START_SEC"]), float(r["END_SEC"])
                st.markdown(f"**{r['EPISODE']}** · `{fmt(start)} – {fmt(end)}`")
                st.video(urls[r["EPISODE"]], start_time=int(start))

# ── Analyze (optional, Pegasus) ─────────────────────
if ENABLE_PEGASUS:
    with tabs[1]:
        files = [r["RELATIVE_PATH"] for r in
                 session.sql(f"SELECT RELATIVE_PATH FROM DIRECTORY({VIDEO_STAGE}) ORDER BY 1").collect()]
        left, right = st.columns([2, 3])
        with left:
            video = st.selectbox("Video", files)
            if video:
                st.video(presigned_url(VIDEO_STAGE, video))
            prompt = st.text_area("Prompt", placeholder="Ask anything about this video...")
            run = st.button("Analyze with Pegasus", type="primary")
        with right:
            if run and video and prompt:
                with st.spinner("Pegasus is watching..."):
                    sql = f"""SELECT AI_COMPLETE('twelvelabs-pegasus-1-2', ?,
                              TO_FILE('{VIDEO_STAGE}', ?)) AS a"""
                    answer = session.sql(sql, params=[prompt, video]).collect()[0]["A"]
                st.markdown(answer)

Run it

[connections.snowflake]
account   = "<account_identifier>"
user      = "<user>"
authenticator = "externalbrowser"
role      = "<role>"
warehouse = "<warehouse>"

pip install streamlit snowflake-snowpark-python snowflake
streamlit run streamlit_app.py

Try it

Conclusion and Resources

What You Learned

Next Steps

Related Resources