高度なビデオ理解アプリケーションの構築：マルチモーダルAIに向けたTwelve Labs Embed APIとLanceDBの統合

!pip install twelvelabs lancedb

from twelvelabs import TwelveLabs
import os

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

# Initialize the Twelve Labs client
twelvelabs_client = TwelveLabs(api_key=TWELVE_LABS_API_KEY)

export TWELVE_LABS_API_KEY='your_api_key_here'

import lancedb
import pyarrow as pa

# Create or connect to a LanceDB database
db = lancedb.connect("my_video_embeddings_db")

# Define the schema for your video embeddings
schema = pa.schema(
    [
        pa.field("video_id", pa.string()),
        pa.field("embedding", pa.list_(pa.float32(), 1024)),
        pa.field("start_time", pa.float32()),
        pa.field("end_time", pa.float32()),
        pa.field("video_url", pa.string())
    ]
)

# Create a table for storing video embeddings
table = db.create_table("video_embeddings", schema=schema, mode="overwrite")

from twelvelabs.models.embed import EmbeddingsTask

def generate_embedding(video_url: str) -> tuple[List[Dict[str, Any]], Any]:
    """Generate embeddings for a given video URL."""
    task = twelvelabs_client.embed.task.create(
        engine_name="Marengo-retrieval-2.6",
        video_url=video_url
    )
    
    def on_task_update(task: EmbeddingsTask):
        print(f"  Status={task.status}")

    task.wait_for_done(sleep_interval=2, callback=on_task_update)
    task_result = twelvelabs_client.embed.task.retrieve(task.id)

    embeddings = [{
        'embedding': v.embedding.float,
        'start_offset_sec': v.start_offset_sec,
        'end_offset_sec': v.end_offset_sec,
        'embedding_scope': v.embedding_scope
    } for v in task_result.video_embeddings]
    
    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()

import uuid

def insert_embeddings(embeddings: List[Dict[str, Any]], video_url: str) -> None:
    """Insert embeddings into the database."""
    data_to_insert = [{
        "id": str(uuid.uuid4()),
        "embedding": emb['embedding'],
        "start_time": emb['start_offset_sec'],
        "end_time": emb['end_offset_sec'],
        "video_url": video_url
    } for emb in embeddings]
    
    table.add(data_to_insert)
    print(f"Inserted {len(data_to_insert)} embeddings for video: {video_url}")

# Use the function to insert the embeddings we generated earlier
insert_embeddings(embeddings, video_url)

# Verify the insertion
print(f"Total embeddings in the table: {len(table)}")

def similarity_search(query_embedding: List[float], k: int = 5) -> List[Dict[str, Any]]:
    """Perform a similarity search on the embeddings table."""
    return table.search(query_embedding).limit(k).to_list()

def get_text_embedding(text_query: str) -> List[float]:
    """Generate a text embedding for a given text query."""
    return twelvelabs_client.embed.text(text_query).embedding.float

# Example usage: Text-based search
text_query = "A person explaining machine learning concepts"
query_embedding = get_text_embedding(text_query)
search_results = similarity_search(query_embedding)

print(f"Top 5 results for query: '{text_query}'")
for i, result in enumerate(search_results):
    print(f"Result {i+1}:")
    print(f"  Video URL: {result['video_url']}")
    print(f"  Segment: {result['start_time']} - {result['end_time']} seconds")
    print(f"  Similarity Score: {result['_distance']}")
    print()

def get_video_segment(video_url: str, start_time: float, end_time: float) -> str:
    """Retrieve a video segment URL from a video URL using start and end times."""
    return f"{video_url}?start={start_time}&end={end_time}"

# Retrieve specific video segment for the top result
top_result = search_results[0]
video_segment = get_video_segment(top_result['video_url'], 
                                  top_result['start_time'], 
                                  top_result['end_time'])
print(f"Link to top result video segment: {video_segment}")

# Example of filtering results by metadata
filtered_results = table.search(query_embedding).where("start_time > 10 AND end_time < 60").limit(5).to_list()

from transformers import AutoTokenizer, AutoModelForSeq2SeqLMv

# Constants
MODEL_NAME = "google/flan-t5-base"
MAX_QUERY_TOKENS = 77

# Initialize model, tokenizer, and database
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME)

def generate_response(prompt: str, max_length: int = 100) -> str:
    """Generate a response using FLAN-T5 model."""
    inputs = tokenizer(prompt, return_tensors="pt", max_length=512, truncation=True)
    outputs = model.generate(**inputs, max_length=max_length, num_return_sequences=1)
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

def retrieve_relevant_segments(query: str, k: int = 3) -> list:
    """Retrieve relevant video segments based on the query."""
    truncated_query = ' '.join(query.split()[:MAX_QUERY_TOKENS])
    query_embedding = get_text_embedding(truncated_query)
    return similarity_search(query_embedding, k)

def create_context(results: list) -> str:
    """Create context from retrieved video segments."""
    return "\n".join(
        f"Video segment from {result['start_time']} to {result['end_time']} seconds: {result['video_url']}"
        for result in results
    )

def rag_workflow(user_query: str) -> str:
    """Execute the RAG (Retrieval-Augmented Generation) workflow."""
    relevant_segments = retrieve_relevant_segments(user_query)
    context = create_context(relevant_segments)
    
    prompt = f"""Context:
{context}

User Query: {user_query}

Based on the context provided about the video segments, please answer the user's query:"""
    
    return generate_response(prompt)

# Example usage
user_query = "What is the purpose of the video?"
rag_response = rag_workflow(user_query)
print("User Query:", user_query)
print("\nRAG Response:", rag_response)