Marengo-retrieval-2.6

pip install twelvelabs pinecone

from pinecone import Pinecone, ServerlessSpec

# Initialize a Pinecone client
# You can create an API key in the Pinecone console at app.pinecone.io
pc = Pinecone(api_key="API_KEY")

index_name = "marengo-retrieval-2-6"

if not pc.has_index(index_name):
  pc.create_index(
      name=index_name,
      dimension=1024,
      metric="cosine",
      spec=ServerlessSpec(
          cloud='aws',
          region='us-east-1'
      )
  )

index = pc.Index(index_name)

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

# Initialize the Twelve Labs client
# You can find the Twelve Labs key in your account upon signing up at api.twelvelabs.io
twelvelabs_client = TwelveLabs(api_key="TL_API_KEY")

def on_task_update(task: EmbeddingsTask):
   print(f"  Status={task.status}")

def generate_embedding(video_file):
   """
   Create and manage the embedding tasks
   """

   # Create an embedding task
   task = twelvelabs_client.embed.task.create(
       engine_name="Marengo-retrieval-2.6",
       video_file=video_file
   )
   print(f"Created task: id={task.id} engine_name={task.engine_name} 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

def ingest_data(video_file, index_name):
   """
   Generate embeddings for video and store in Pinecone
   """

   # Generate embeddings using Twelve Labs Embed API
   embeddings, task_result = generate_embedding(video_file)

   # Prepare vectors for upsert
   vectors_to_upsert = []
   for i, emb in enumerate(embeddings):
       vector_id = f"{video_name}_{i}"
       vectors_to_upsert.append((vector_id, emb['embedding'], {
           'video_file': video_file,
           'video_segment': i,
           'start_time': emb['start_offset_sec'],
           'end_time': emb['end_offset_sec'],
           'scope': emb['embedding_scope']
       }))

   # Upsert embeddings to Pinecone
   index.upsert(vectors=vectors_to_upsert)

   return f"Ingested {len(embeddings)} embeddings for {video_file}"

video_file = "Rare Beauty By Selena Gomez - Makeup Made To Feel Good In.mp4"

result = ingest_data(video_file)

print(result)

def get_text_embedding(text_query):
   # Twelve Labs Embed API supports text-to-embedding
   text_embedding = twelvelabs_client.embed.create(
     engine_name="Marengo-retrieval-2.6",
     text=text_query,
     text_truncate="start"
   )

   return text_embedding.text_embedding.float

def retrieve_similar_content(query, index_name, top_k=5):
   """
   Retrieve similar content based on query embedding
   """
   # Generate query embedding
   query_embedding = get_text_embedding(query)

   # Connect to Pinecone index
   index = pc.Index(index_name)

   # Query Pinecone for similar vectors
   results = index.query(vector=query_embedding, top_k=top_k, include_metadata=True)

   return results

text_query = "Lipstick"
similar_content = retrieve_similar_content(text_query)

print(f"Query: '{text_query}'")
print(f"Top {len(similar_content['matches'])} similar content:")
for i, match in enumerate(similar_content['matches']):
   print(f"{i+1}. Score: {match['score']:.4f}")
   print(f"   Video File: {match['metadata']['video_file']}")
   print(f"   Video ID: {match['metadata']['video_segment']}")
   print(f"   Time range: {match['metadata']['start_time']} - {match['metadata']['end_time']} seconds")
   print(f"   Scope: {match['metadata']['scope']}")
   print()