Twelve Labs と Milvus を使用したマルチモーダル検索拡張生成（RAG）アプリケーションの構築

TWELVELABS_API_KEY="your_twelvelabs_key"
COLLECTION_NAME="your_collection_name"
URL="your_milvus_url"
TOKEN="your_milvus_token"
OPENAI_API_KEY="your_openai_key"

├── pages/
│   ├── add_product_page.py
│   └── visual_search.py
├── .gitignore
├── app.py
└── utils.py  
└── requirements.txt

pip install -r requirements.txt

# Load environment variables
COLLECTION_NAME = os.getenv('COLLECTION_NAME')
URL = os.getenv('URL')
TOKEN = os.getenv('TOKEN')
TWELVELABS_API_KEY = os.getenv('TWELVELABS_API_KEY')

# Initialize connections
openai_client = OpenAI()
connections.connect(uri=URL, token=TOKEN)
collection = Collection(COLLECTION_NAME)
collection.load()

twelvelabs_client = TwelveLabs(api_key=TWELVELABS_API_KEY)

# Generate text and segmented video embeddings for a product
def generate_embedding(product_info):
    try:
        st.write("Starting embedding generation process...")
        st.write(f"Processing product: {product_info['title']}")
        
        # Initialize the Twelve Labs client
 twelvelabs_client = TwelveLabs(api_key=TWELVELABS_API_KEY)
        st.write("TwelveLabs client initialized successfully")
        
        st.write("Attempting to generate text embedding...")

        # Formatting the Text Data of Product Catalogue
        text = f"product type: {product_info['title']}. " \
               f"product description: {product_info['desc']}. " \
               f"product category: fashion apparel."
               
        st.write(f"Generating embedding for text: {text}")
        
        # Generating the Text Embeddings
        text_embedding = twelvelabs_client.embed.create(
            model_name="Marengo-retrieval-2.7",
            text=text
        ).text_embedding.segments[0].embeddings_float
        st.write("Text embedding generated successfully")
        
        # Create and wait for video embedding task
        st.write("Creating video embedding task...")
        video_task = twelvelabs_client.embed.task.create(
            model_name="Marengo-retrieval-2.7",
            video_url=product_info['video_url'],
            video_clip_length=6
        )
        
        def on_task_update(task):
            st.write(f"Video processing status: {task.status}")
        
        st.write("Waiting for video processing to complete...")
        video_task.wait_for_done(sleep_interval=2, callback=on_task_update)
        
        # Retrieve segmented video embeddings
        video_task = video_task.retrieve()
        if not video_task.video_embedding or not video_task.video_embedding.segments:
            raise Exception("Failed to retrieve video embeddings")
        
        video_segments = video_task.video_embedding.segments
        st.write(f"Retrieved {len(video_segments)} video segments")
        
        video_embeddings = []
        for segment in video_segments:
            video_embeddings.append({
                'embedding': segment.embeddings_float,
                'metadata': {
                    'scope': 'clip',
                    'start_time': segment.start_offset_sec,
                    'end_time': segment.end_offset_sec,
                    'video_url': product_info['video_url']
                }
            })
        
        return {
            'text_embedding': text_embedding,
            'video_embeddings': video_embeddings
        }, None
        
    except Exception as e:
        st.error("Error in embedding generation")
        st.error(f"Error message: {str(e)}")
        return None, str(e)

# Insert text and all video segment embeddings into Milvus Collection
def insert_embeddings(embeddings_data, product_info):
    try:
        metadata = {
            "product_id": product_info['product_id'],
            "title": product_info['title'],
            "description": product_info['desc'],
            "video_url": product_info['video_url'],
            "link": product_info['link']
        }
        
        # Insert text embedding
        text_entry = {
            "id": int(uuid.uuid4().int & (1<<63)-1),
            "vector": embeddings_data['text_embedding'],
            "metadata": metadata,
            "embedding_type": "text"
        }
        collection.insert([text_entry])
        st.write("Text embedding inserted successfully")
        
        # Insert each video segment embedding
        for video_segment in embeddings_data['video_embeddings']:
            video_entry = {
                "id": int(uuid.uuid4().int & (1<<63)-1),
                "vector": video_segment['embedding'],
                "metadata": {**metadata, **video_segment['metadata']},
                "embedding_type": "video"
            }
            collection.insert([video_entry])
        
        st.write(f"Inserted {len(embeddings_data['video_embeddings'])} video segment embeddings")
        return True
        
    except Exception as e:
        st.error(f"Error inserting embeddings: {str(e)}")
        return False

# Load and Read Query Image
image_path = "path/to/your/image.jpg" 
with open(image_path, 'rb') as f: 
    image_file = f.read()

# Intializing of client and Generation of Image Emebedding
twelvelabs_client = TwelveLabs(api_key=TWELVELABS_API_KEY)
image_embedding = twelvelabs_client.embed.create(
    model_name="Marengo-retrieval-2.7",
    image_file=image_file
).image_embedding.segments[0].embeddings_float

search_params = {
    "metric_type": "COSINE",
    "params": {
        "nprobe": 1024,
        "ef": 64
    }
}

# Search relevant video segments
results = collection.search(
    data=[image_embedding],
    anns_field="vector",
    param=search_params,
    limit=2,
    expr="embedding_type == 'video'",
    output_fields=["metadata"]
)

search_results = []
for hits in results:
    for hit in hits:
        metadata = hit.metadata
        # Convert score from [-1,1] to [0,100] range
        similarity = round((hit.score + 1) * 50, 2)
        similarity = max(0, min(100, similarity))
        
        search_results.append({
            'Title': metadata.get('title', ''),
            'Description': metadata.get('description', ''),
            'Link': metadata.get('link', ''),
            'Start Time': f"{metadata.get('start_time', 0):.1f}s",
            'End Time': f"{metadata.get('end_time', 0):.1f}s",
            'Video URL': metadata.get('video_url', ''),
            'Similarity': f"{similarity}%",
            'Raw Score': hit.score
        })

# Sort by similarity score in descending order
search_results.sort(key=lambda x: float(x['Similarity'].rstrip('%')), reverse=True)

# Sample question
question_with_context = f"fashion product: Suggest me black dresses for a party"

# Generate embedding for the question
question_embedding = twelvelabs_client.embed.create(
    model_name="Marengo-retrieval-2.7",
    text=question_with_context
).text_embedding.segments[0].embeddings_float

# Configure search parameters
search_params = {
    "metric_type": "COSINE",
    "params": {
        "nprobe": 1024,
        "ef": 64
    }
}

# Search for relevant text embeddings
text_results = collection.search(
    data=[question_embedding],
    anns_field="vector",
    param=search_params,
    limit=2,  # Get top 2 text matches
    expr="embedding_type == 'text'",
    output_fields=["metadata"]
)

# Search for relevant video segments
video_results = collection.search(
    data=[question_embedding],
    anns_field="vector",
    param=search_params,
    limit=3,  # Get top 3 video segments
    expr="embedding_type == 'video'",
    output_fields=["metadata"]
)

# Process text results
text_docs = []
for hits in text_results:
    for hit in hits:
        metadata = hit.metadata
        similarity = round((hit.score + 1) * 50, 2)
        similarity = max(0, min(100, similarity))
        
        text_docs.append({
            "title": metadata.get('title', 'Untitled'),
            "description": metadata.get('description', 'No description available'),
            "product_id": metadata.get('product_id', ''),
            "video_url": metadata.get('video_url', ''),
            "link": metadata.get('link', ''),
            "similarity": similarity,
            "raw_score": hit.score,
            "type": "text"
        })

# Process video results
video_docs = []
for hits in video_results:
    for hit in hits:
        metadata = hit.metadata
        similarity = round((hit.score + 1) * 50, 2)
        similarity = max(0, min(100, similarity))
        
        video_docs.append({
            "title": metadata.get('title', 'Untitled'),
            "description": metadata.get('description', 'No description available'),
            "product_id": metadata.get('product_id', ''),
            "video_url": metadata.get('video_url', ''),
            "link": metadata.get('link', ''),
            "similarity": similarity,
            "raw_score": hit.score,
            "start_time": metadata.get('start_time', 0),
            "end_time": metadata.get('end_time', 0),
            "type": "video"
        })

# Handle case when no results found
if not text_docs and not video_docs:
    response_data = {
        "response": "I couldn't find any matching products. Try describing what you're looking for differently.",
        "metadata": None
    }
else:
    # Create context from text results for LLM
    text_context = "\n\n".join([
        f"Product: {doc['title']}\nDescription: {doc['description']}\nLink: {doc['link']}"
        for doc in text_docs
    ])

    # Prepare messages for OpenAI
    messages = [
        {
            "role": "system",
            "content": """You are a professional fashion advisor and AI shopping assistant.
            Organize your response in the following format:

            First, provide a brief, direct answer to the user's query
            Then, describe any relevant products found that match their request, including:
               - Product name and key features
               - Why this product matches their needs
               - Style suggestions for how to wear or use the item
            Finally, provide any additional style advice or recommendations
            
            Keep your response engaging and natural while maintaining this clear structure.
            Focus on being helpful and specific rather than promotional."""
        },
        {
            "role": "user",
            "content": f"""Query: {question}

Available Products:
{text_context}

Please provide fashion advice and product recommendations based on these options."""
        }
    ]

    # Get response from OpenAI
    chat_response = openai_client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=0.5,
        max_tokens=500
    )

    # Format final response
    response_data = {
        "response": chat_response.choices[0].message.content,
        "metadata": {
            "sources": text_docs + video_docs,
            "total_sources": len(text_docs) + len(video_docs),
            "text_sources": len(text_docs),
            "video_sources": len(video_docs)
        }
    }