Mở đầu: Khi Chatbot Không Còn "Nhìn" Được Sản Phẩm

Tôi đã làm việc với rất nhiều đội ngũ phát triển chatbot và hệ thống hỏi đáp tự động trong suốt 5 năm qua. Và một trong những vấn đề phổ biến nhất mà tôi gặp phải là: các hệ thống RAG (Retrieval-Augmented Generation) truyền thống chỉ hoạt động tốt với văn bản. Khi người dùng hỏi về một chiếc áo màu xanh có hoạ tiết hoa, chatbot không thể "nhìn" thấy sản phẩm đó trong cơ sở dữ liệu hình ảnh của doanh nghiệp. Bài viết này sẽ hướng dẫn bạn xây dựng một kiến trúc Multi-Modal RAG hoàn chỉnh, kết hợp khả năng truy xuất và hiểu cả hình ảnh lẫn văn bản — và tất cả đều có thể triển khai dễ dàng với HolySheep AI.

Case Study: Startup E-Commerce Ở TP.HCM Giải 70% Chi Phí AI

Bối cảnh: Một nền tảng thương mại điện tử tại TP.HCM với hơn 50,000 sản phẩm, trong đó 40% là sản phẩm thời trang với hình ảnh đặc thù (màu sắc, kiểu dáng, hoạ tiết). Đội ngũ muốn xây dựng chatbot tư vấn sản phẩm thông minh có thể "nhìn" ảnh và trả lời câu hỏi như "Cho tôi xem những áo sơ mi trắng có viền xanh".

Điểm đau với nhà cung cấp cũ: Họ đã sử dụng một nhà cung cấp API AI quốc tế với độ trễ trung bình 1.2 giây cho mỗi truy vấn multi-modal, chi phí $4,200/tháng cho 8 triệu token hình ảnh và văn bản. Đội ngũ kỹ thuật phải duy trì 3 service riêng biệt: embedding service, vector database, và LLM inference — mỗi service lại có config và monitoring riêng.

Lý do chọn HolySheep: Sau khi benchmark, họ nhận thấy HolySheep cung cấp endpoint multi-modal duy nhất xử lý cả embedding lẫn inference, với độ trễ thực tế dưới 180ms (so với 1.2 giây trước đó). Tỷ giá tính theo ¥1 = $1 giúp họ tiết kiệm 85% chi phí vận hành.

Các bước di chuyển cụ thể:

Kết quả sau 30 ngày go-live:

Kiến Trúc Multi-Modal RAG: Tổng Quan Hệ Thống

Sơ đồ luồng dữ liệu

┌─────────────────────────────────────────────────────────────────────┐
│                        MULTI-MODAL RAG ARCHITECTURE                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────┐    ┌──────────────────┐    ┌──────────────────────┐  │
│  │  INPUT   │───▶│  IMAGE PROCESS  │───▶│  TEXT EXTRACTION     │  │
│  │ (Mixed)  │    │  & OCR (nếu cần)│    │  from Images         │  │
│  └──────────┘    └──────────────────┘    └──────────────────────┘  │
│        │                    │                     │               │
│        ▼                    ▼                     ▼               │
│  ┌──────────────────────────────────────────────────────────────┐ │
│  │              HOLYSHEEP API (base_url + key)                  │ │
│  │                                                              │ │
│  │  • Vision Embedding ( CLIP-like )                            │ │
│  │  • Text Embedding ( multilingual )                            │ │
│  │  • LLM Inference ( context-aware )                            │ │
│  │                                                              │ │
│  └──────────────────────────────────────────────────────────────┘ │
│                              │                                     │
│                              ▼                                     │
│  ┌──────────────────────────────────────────────────────────────┐ │
│  │              VECTOR DATABASE (Hybrid Search)                │ │
│  │                                                              │ │
│  │  • Image vectors + Text vectors                              │ │
│  │  • Cross-modal similarity                                    │ │
│  │  • Metadata filtering                                        │ │
│  └──────────────────────────────────────────────────────────────┘ │
│                              │                                     │
│                              ▼                                     │
│  ┌──────────────────────────────────────────────────────────────┐ │
│  │              RESPONSE GENERATION                            │ │
│  │                                                              │ │
│  │  • Retrieved context (images + text)                          │ │
│  │  • RAG prompt with vision capabilities                       │ │
│  │  • Final answer with referenced images                       │ │
│  └──────────────────────────────────────────────────────────────┘ │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Cài Đặt Môi Trường Và Dependencies

Trước tiên, hãy thiết lập môi trường Python với các thư viện cần thiết cho multi-modal RAG:

# Tạo virtual environment và cài đặt dependencies
python -m venv venv_multimodal_rag
source venv_multimodal_rag/bin/activate  # Linux/Mac

venv_multimodal_rag\Scripts\activate  # Windows



Cài đặt các thư viện cần thiết

pip install requests pillow python-dotenv qdrant-client
pip install numpy scikit-learn  # cho vector operations
pip install fastapi uvicorn     # cho API server (optional)


Kiểm tra cài đặt

python -c "import requests, PIL, qdrant_client; print('Setup thanh cong!')"
# config.py - Cấu hình HolySheep API
import os
from dotenv import load_dotenv

load_dotenv()


Cấu hình HolySheep - LUÔN LUÔN sử dụng base_url này

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")


Cấu hình Vector Database (ví dụ Qdrant)

QDRANT_HOST = os.getenv("QDRANT_HOST", "localhost")
QDRANT_PORT = int(os.getenv("QDRANT_PORT", "6333"))
COLLECTION_NAME = "multimodal_products"


Cấu hình RAG

EMBEDDING_MODEL = "clip-vit-base-patch32"
LLM_MODEL = "gpt-4.1"  # Hoặc deepseek-v3.2, claude-sonnet-4.5
CHUNK_SIZE = 256
RETRIEVAL_TOP_K = 5

print(f"Configuration loaded:")
print(f"  Base URL: {BASE_URL}")
print(f"  LLM Model: {LLM_MODEL}")
print(f"  Embedding Model: {EMBEDDING_MODEL}")

Module 1: Image Embedding Với HolySheep Vision API

HolySheep cung cấp endpoint vision mạnh mẽ cho phép embedding hình ảnh trực tiếp. Dưới đây là implementation chi tiết:

import base64
import requests
from PIL import Image
from io import BytesIO
from typing import List, Dict, Optional
import time

class HolySheepVisionClient:
    """Client cho HolySheep Vision API - Multi-modal embedding"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def encode_image_to_base64(self, image_source: str) -> str:
        """
        Chuyển đổi image từ URL hoặc local path thành base64
        
        Args:
            image_source: URL hoặc đường dẫn local tới hình ảnh
        
        Returns:
            Chuỗi base64 của hình ảnh
        """
        if image_source.startswith(('http://', 'https://')):
            response = requests.get(image_source)
            response.raise_for_status()
            image_data = response.content
        else:
            with open(image_source, 'rb') as f:
                image_data = f.read()
        
        return base64.b64encode(image_data).decode('utf-8')
    
    def get_image_embedding(self, image_source: str) -> List[float]:
        """
        Lấy embedding vector cho một hình ảnh
        
        Args:
            image_source: URL hoặc đường dẫn local
        
        Returns:
            List[float] - embedding vector (1536 dimensions với CLIP)
        """
        # Encode image
        image_base64 = self.encode_image_to_base64(image_source)
        
        # Gọi HolySheep Vision API
        # Lưu ý: Sử dụng endpoint /embeddings với model vision
        payload = {
            "model": "clip-vit-base-patch32",
            "input": {
                "type": "image",
                "content": image_base64
            }
        }
        
        start_time = time.time()
        response = self.session.post(
            f"{self.base_url}/embeddings",
            json=payload,
            timeout=30
        )
        latency_ms = (time.time() - start_time) * 1000
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        embedding = result['data'][0]['embedding']
        
        print(f"✓ Image embedding retrieved in {latency_ms:.1f}ms")
        return embedding
    
    def batch_get_embeddings(self, image_sources: List[str]) -> List[List[float]]:
        """
        Lấy embeddings cho nhiều hình ảnh cùng lúc
        
        Args:
            image_sources: Danh sách URL hoặc đường dẫn
        
        Returns:
            List[List[float]] - danh sách embedding vectors
        """
        embeddings = []
        
        # Xử lý batch với HolySheep để tối ưu chi phí
        for source in image_sources:
            try:
                embedding = self.get_image_embedding(source)
                embeddings.append(embedding)
            except Exception as e:
                print(f"⚠ Lỗi khi xử lý {source}: {e}")
                embeddings.append(None)
        
        return embeddings


Ví dụ sử dụng

if __name__ == "__main__":
    client = HolySheepVisionClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    # Test với một hình ảnh mẫu
    test_image = "https://example.com/product-sample.jpg"
    
    try:
        embedding = client.get_image_embedding(test_image)
        print(f"Embedding dimension: {len(embedding)}")
    except Exception as e:
        print(f"Kiểm tra thất bại: {e}")

Module 2: Text Embedding Đa Ngôn Ngữ

HolySheep hỗ trợ text embedding với nhiều ngôn ngữ, bao gồm tiếng Việt, tiếng Trung, tiếng Anh — hoàn hảo cho các ứng dụng thương mại điện tử đa quốc gia:

import requests
from typing import List, Dict, Union
import time

class HolySheepTextClient:
    """Client cho HolySheep Text Embedding API - Hỗ trợ đa ngôn ngữ"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def get_text_embedding(
        self, 
        text: str, 
        model: str = "text-embedding-3-large"
    ) -> List[float]:
        """
        Lấy embedding vector cho văn bản
        
        Args:
            text: Văn bản cần embedding
            model: Model embedding (mặc định: text-embedding-3-large)
        
        Returns:
            List[float] - embedding vector
        """
        payload = {
            "model": model,
            "input": text
        }
        
        start_time = time.time()
        response = self.session.post(
            f"{self.base_url}/embeddings",
            json=payload,
            timeout=30
        )
        latency_ms = (time.time() - start_time) * 1000
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        embedding = result['data'][0]['embedding']
        
        print(f"✓ Text embedding retrieved in {latency_ms:.1f}ms")
        return embedding
    
    def get_texts_embeddings(
        self, 
        texts: List[str], 
        model: str = "text-embedding-3-large"
    ) -> List[List[float]]:
        """
        Batch embedding cho nhiều văn bản
        
        Args:
            texts: Danh sách văn bản
            model: Model embedding
        
        Returns:
            List[List[float]] - danh sách embedding vectors
        """
        payload = {
            "model": model,
            "input": texts
        }
        
        start_time = time.time()
        response = self.session.post(
            f"{self.base_url}/embeddings",
            json=payload,
            timeout=60
        )
        latency_ms = (time.time() - start_time) * 1000
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        embeddings = [item['embedding'] for item in result['data']]
        
        print(f"✓ Batch embedding ({len(texts)} texts) retrieved in {latency_ms:.1f}ms")
        return embeddings
    
    def cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
        """Tính cosine similarity giữa 2 vectors"""
        import numpy as np
        v1 = np.array(vec1)
        v2 = np.array(vec2)
        return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))


Ví dụ sử dụng

if __name__ == "__main__":
    client = HolySheepTextClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    # Test với tiếng Việt
    test_texts = [
        "áo sơ mi nam trắng cổ điển",
        "quần jeans nam rách gối",
        "giày thể thao nữ màu hồng",
        "áo phông unisex form rộng"
    ]
    
    embeddings = client.get_texts_embeddings(test_texts)
    print(f"Embedding dimensions: {len(embeddings[0])}")
    
    # Test similarity
    sim = client.cosine_similarity(embeddings[0], embeddings[3])
    print(f"Similarity giữa 'áo sơ mi' và 'áo phông': {sim:.4f}")

Module 3: Hybrid Search Với Qdrant Vector Database

Để kết hợp tìm kiếm hình ảnh và văn bản, tôi khuyên sử dụng Qdrant — một vector database mã nguồn mở với khả năng hybrid search tuyệt vời:

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter
from qdrant_client.models import MatchAny, MatchValue
from typing import List, Dict, Tuple, Optional
import numpy as np

class HybridSearchEngine:
    """
    Hybrid Search Engine kết hợp image và text embeddings
    """
    
    def __init__(
        self, 
        host: str = "localhost", 
        port: int = 6333,
        collection_name: str = "multimodal_products"
    ):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = collection_name
        self.vector_dim = 1536  # CLIP + Text-embedding-3-large dimension
        
    def create_collection(self, recreate: bool = False):
        """
        Tạo collection cho multi-modal vectors
        
        Args:
            recreate: Nếu True, xóa collection cũ và tạo mới
        """
        if recreate:
            self.client.delete_collection(collection_name=self.collection_name)
            print(f"✓ Deleted existing collection '{self.collection_name}'")
        
        # Kiểm tra xem collection đã tồn tại chưa
        collections = self.client.get_collections().collections
        collection_names = [c.name for c in collections]
        
        if self.collection_name not in collection_names:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.vector_dim,
                    distance=Distance.COSINE
                )
            )
            print(f"✓ Created collection '{self.collection_name}'")
        else:
            print(f"Collection '{self.collection_name}' already exists")
    
    def insert_product(
        self,
        product_id: str,
        image_url: str,
        product_name: str,
        description: str,
        category: str,
        image_embedding: List[float],
        text_embedding: List[float]
    ) -> str:
        """
        Chèn một sản phẩm vào database với cả image và text embeddings
        
        Args:
            product_id: ID duy nhất của sản phẩm
            image_url: URL hình ảnh sản phẩm
            product_name: Tên sản phẩm
            description: Mô tả sản phẩm
            category: Danh mục sản phẩm
            image_embedding: Vector embedding từ hình ảnh
            text_embedding: Vector embedding từ text
        
        Returns:
            str: ID của point được chèn
        """
        # Combine image và text embeddings (có thể weighted)
        combined_embedding = np.mean([image_embedding, text_embedding], axis=0)
        combined_embedding = combined_embedding.tolist()
        
        point = PointStruct(
            id=product_id,
            vector=combined_embedding,
            payload={
                "image_url": image_url,
                "product_name": product_name,
                "description": description,
                "category": category,
                "image_embedding_raw": image_embedding,  # Lưu riêng để search cross-modal
                "text_embedding_raw": text_embedding
            }
        )
        
        operation_info = self.client.upsert(
            collection_name=self.collection_name,
            points=[point]
        )
        
        return operation_info
    
    def search_by_text(
        self, 
        query_embedding: List[float], 
        top_k: int = 5,
        category_filter: Optional[str] = None
    ) -> List[Dict]:
        """
        Tìm kiếm sản phẩm bằng text embedding
        
        Args:
            query_embedding: Query vector
            top_k: Số lượng kết quả
            category_filter: Lọc theo danh mục
        
        Returns:
            List[Dict]: Danh sách sản phẩm phù hợp
        """
        search_filter = None
        if category_filter:
            search_filter = Filter(
                must=[MatchValue(key="category", value=category_filter)]
            )
        
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k,
            query_filter=search_filter,
            with_payload=True
        )
        
        return [
            {
                "id": hit.id,
                "score": hit.score,
                "product_name": hit.payload["product_name"],
                "description": hit.payload["description"],
                "category": hit.payload["category"],
                "image_url": hit.payload["image_url"]
            }
            for hit in results
        ]
    
    def search_by_image(
        self, 
        query_embedding: List[float], 
        top_k: int = 5
    ) -> List[Dict]:
        """
        Tìm kiếm sản phẩm bằng image embedding (cross-modal retrieval)
        """
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k,
            with_payload=True
        )
        
        return [
            {
                "id": hit.id,
                "score": hit.score,
                "product_name": hit.payload["product_name"],
                "description": hit.payload["description"],
                "category": hit.payload["category"],
                "image_url": hit.payload["image_url"]
            }
            for hit in results
        ]
    
    def hybrid_search(
        self,
        text_query: str,
        image_url: Optional[str],
        text_embedding: List[float],
        image_embedding: Optional[List[float]],
        top_k: int = 5,
        text_weight: float = 0.6,
        image_weight: float = 0.4
    ) -> List[Dict]:
        """
        Hybrid search kết hợp cả text và image queries
        
        Args:
            text_query: Query text
            image_url: URL hình ảnh (optional)
            text_embedding: Text embedding vector
            image_embedding: Image embedding vector (nếu có)
            top_k: Số kết quả
            text_weight: Trọng số cho text (0.0 - 1.0)
            image_weight: Trọng số cho image (0.0 - 1.0)
        
        Returns:
            List[Dict]: Kết quả hybrid search
        """
        if image_embedding is None:
            # Chỉ search bằng text
            return self.search_by_text(text_embedding, top_k)
        
        # Normalize weights
        total_weight = text_weight + image_weight
        text_weight /= total_weight
        image_weight /= total_weight
        
        # Combine vectors
        combined_vector = np.add(
            np.multiply(text_embedding, text_weight),
            np.multiply(image_embedding, image_weight)
        ).tolist()
        
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=combined_vector,
            limit=top_k,
            with_payload=True
        )
        
        return [
            {
                "id": hit.id,
                "score": hit.score,
                "product_name": hit.payload["product_name"],
                "description": hit.payload["description"],
                "category": hit.payload["category"],
                "image_url": hit.payload["image_url"]
            }
            for hit in results
        ]


Ví dụ sử dụng

if __name__ == "__main__":
    search_engine = HybridSearchEngine(
        host="localhost",
        port=6333,
        collection_name="multimodal_products"
    )
    
    # Tạo collection
    search_engine.create_collection(recreate=False)
    
    print("✓ Hybrid Search Engine initialized")

Module 4: RAG Inference Với HolySheep LLM

Bây giờ, hãy xây dựng module RAG inference sử dụng HolySheep LLM để generate câu trả lời dựa trên context đã retrieve:

import requests
from typing import List, Dict, Optional
import json
import time

class MultiModalRAG:
    """
    Multi-Modal RAG System sử dụng HolySheep LLM
    Hỗ trợ cả text và image context
    """
    
    def __init__(
        self, 
        api_key: str, 
        base_url: str = "https://api.holysheep.ai/v1",
        model: str = "gpt-4.1"
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.model = model
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def build_rag_prompt(
        self, 
        query: str, 
        retrieved_products: List[Dict],
        include_images: bool = True
    ) -> List[Dict]:
        """
        Xây dựng prompt cho RAG với multi-modal context
        
        Args:
            query: Câu hỏi của user
            retrieved_products: Kết quả từ hybrid search
            include_images: Có bao gồm hình ảnh trong context không
        
        Returns:
            List[Dict]: Messages format cho LLM
        """
        # System prompt
        system_message = {
            "role": "system",
            "content": """Bạn là trợ lý tư vấn sản phẩm thông minh cho cửa hàng thương mại điện tử.
Nhiệm vụ của bạn:
1. Phân tích câu hỏi của khách hàng để hiểu nhu cầu
2. Đề xuất sản phẩm phù hợp từ danh sách được cung cấp
3. Giải thích tại sao sản phẩm đó phù hợp với yêu cầu
4. Trả lời bằng tiếng Việt, thân thiện và chuyên nghiệp

LUÔN bao gồm:
- Tên sản phẩm và hình ảnh (nếu có)
- Mô tả ngắn gọn đặc điểm nổi bật
- Lý do sản phẩm phù hợp với nhu cầu

Nếu không tìm thấy sản phẩm phù hợp, hãy gợi ý các sản phẩm tương tự hoặc hỏi thêm thông tin."""
        }
        
        # Build context từ retrieved products
        context_parts = []
        
        for idx, product in enumerate(retrieved_products, 1):
            context_parts.append(f"""
Sản phẩm #{idx}:
- Tên: {product.get('product_name', 'N/A')}
- Mô tả: {product.get('description', 'N/A')}
- Danh mục: {product.get('category', 'N/A')}
- Hình ảnh: {product.get('image_url', 'N/A')}
- Độ phù hợp: {product.get('score', 0):.2%}""")
        
        context_text = "\n".join(context_parts)
        
        # User message với context
        user_message = {
            "role": "user", 
            "content": f"""Khách hàng hỏi: "{query}"

Danh sách sản phẩm liên quan được tìm thấy:
{context_text}

Hãy tư vấn sản phẩm phù hợp nhất cho khách hàng."""
        }
        
        return [system_message, user_message]
    
    def generate_response(
        self, 
        query: str, 
        retrieved_products: List[Dict],
        temperature: float = 0.7,
        max_tokens: int = 1000
    ) -> Dict:
        """
        Generate câu trả lời từ RAG context
        
        Args:
            query: Câu hỏi của user
            retrieved_products: Kết quả hybrid search
            temperature: Creativity level (0.0 - 1.0)
            max_tokens: Số token tối đa cho response
        
        Returns:
            Dict với response và metadata
        """
        messages = self.build_rag_prompt(query, retrieved_products)
        
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        start_time = time.time()
        response = self.session.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            timeout=60
        )
        latency_ms = (time.time() - start_time) * 1000
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()

Tài nguyên liên quan
Bài viết liên quan