FastEmbed Integration

pip install qdrant-client[fastembed]
# or for GPU support
pip install qdrant-client[fastembed-gpu]

def get_embedding_size(self, model_name: str) -> int:
    """
    Get embedding size for a model.

    Parameters:
    - model_name: Name of the embedding model

    Returns:
        int: Vector dimension for the model
    """

# Get embedding size for a model
size = client.get_embedding_size("sentence-transformers/all-MiniLM-L6-v2")
print(f"Embedding size: {size}")  # 384

# Create collection with correct size
client.create_collection(
    collection_name="semantic_collection",
    vectors_config=models.VectorParams(size=size, distance=models.Distance.COSINE)
)

def upload_collection(
    self,
    collection_name: str,
    vectors: Union[
        Iterable[VectorStruct],
        Iterable[PointStruct], 
        Iterable[Record],
        Iterable[Document],
        Iterable[ImageDocument]
    ],
    ids: Optional[Iterable[PointId]] = None,
    batch_size: int = 100,
    parallel: int = 1,
    max_retries: int = 3,
    payload: Optional[Iterable[Payload]] = None,
    wait: bool = True,
    shard_key_selector: Optional[ShardKeySelector] = None,
) -> None:
    """
    Upload collection with automatic embedding generation.

    Parameters:
    - collection_name: Name of the collection
    - vectors: Documents, images, or vectors to upload
    - ids: Point IDs (generated if not provided)
    - batch_size: Number of items per batch
    - parallel: Number of parallel processing threads
    - max_retries: Maximum retry attempts
    - payload: Additional metadata for points
    - wait: Wait for operation to complete
    - shard_key_selector: Shard key for routing
    """

from qdrant_client import models

# Upload text documents with auto-embedding
model_name = "sentence-transformers/all-MiniLM-L6-v2"
documents = [
    models.Document(text="Qdrant has Langchain integrations", model=model_name),
    models.Document(text="Qdrant also has Llama Index integrations", model=model_name),
    models.Document(text="Vector databases are useful for AI applications", model=model_name)
]

client.upload_collection(
    collection_name="semantic_search",
    vectors=documents,
    ids=[1, 2, 3],
    payload=[
        {"source": "langchain-docs"},
        {"source": "llamaindex-docs"},
        {"source": "general"}
    ]
)

# Upload images with auto-embedding
image_documents = [
    models.ImageDocument(image="path/to/image1.jpg", model="clip-ViT-B-32"),
    models.ImageDocument(image="path/to/image2.jpg", model="clip-ViT-B-32")
]

client.upload_collection(
    collection_name="image_search",
    vectors=image_documents,
    payload=[{"type": "photo"}, {"type": "illustration"}]
)

def query_qdrant(
    self,
    collection_name: str,
    query: Union[str, Document, ImageDocument],
    query_filter: Optional[Filter] = None,
    limit: int = 10,
    search_params: Optional[SearchParams] = None,
    **kwargs
) -> List[QueryResponse]:
    """
    Semantic search with automatic query embedding.

    Parameters:
    - collection_name: Name of the collection
    - query: Text query, Document, or ImageDocument
    - query_filter: Filter conditions for results
    - limit: Maximum number of results
    - search_params: Search algorithm parameters

    Returns:
        List[QueryResponse]: Semantic search results
    """

# Text semantic search
results = client.query_qdrant(
    collection_name="semantic_search",
    query=models.Document(text="AI and machine learning", model=model_name),
    limit=5
)

# Simple text query (uses default model)
results = client.query_qdrant(
    collection_name="semantic_search",
    query="vector search applications",
    limit=5
)

# Image semantic search
results = client.query_qdrant(
    collection_name="image_search",
    query=models.ImageDocument(image="query_image.jpg", model="clip-ViT-B-32"),
    limit=5
)

def embed_documents(
    self,
    documents: List[Union[str, Document]],
    model_name: Optional[str] = None,
    **kwargs
) -> List[List[float]]:
    """
    Generate embeddings for text documents.

    Parameters:
    - documents: List of text strings or Document objects
    - model_name: Embedding model name

    Returns:
        List[List[float]]: Generated embeddings
    """

def embed_sparse_documents(
    self,
    documents: List[Union[str, Document]],
    model_name: Optional[str] = None,
    **kwargs
) -> List[SparseVector]:
    """
    Generate sparse embeddings for text documents.

    Parameters:
    - documents: List of text strings or Document objects
    - model_name: Sparse embedding model name

    Returns:
        List[SparseVector]: Generated sparse embeddings
    """

class Document(BaseModel):
    text: str  # Text content to embed
    model: str  # Embedding model name
    options: Optional[Dict[str, Any]] = None  # Model-specific options

# Document options examples:
# {"cuda": True}  # Use GPU acceleration
# {"normalize": True}  # Normalize embeddings
# {"batch_size": 32}  # Custom batch size

class ImageDocument(BaseModel):
    image: Union[str, bytes, ImageInput]  # Image path, bytes, or PIL Image
    model: str  # Image embedding model name
    options: Optional[Dict[str, Any]] = None  # Model-specific options

class QueryResponse(BaseModel):
    id: Union[str, int]  # Point ID
    embedding: Optional[List[float]]  # Generated embedding
    sparse_embedding: Optional[SparseVector] = None  # Sparse embedding
    metadata: Dict[str, Any]  # Point payload
    document: str  # Original document text
    score: float  # Similarity score

class FastEmbedMisc:
    @classmethod
    def is_installed(cls) -> bool:
        """Check if FastEmbed is installed."""
    
    @classmethod
    def get_text_models(cls) -> Set[str]:
        """Get available text embedding models."""
    
    @classmethod
    def get_image_models(cls) -> Set[str]:
        """Get available image embedding models."""
    
    @classmethod  
    def get_sparse_models(cls) -> Set[str]:
        """Get available sparse embedding models."""

# Install GPU version
# pip install qdrant-client[fastembed-gpu]

# Use GPU in document options
document = models.Document(
    text="Accelerated embedding generation",
    model="sentence-transformers/all-MiniLM-L6-v2",
    options={"cuda": True}
)

# Large batch upload with optimized settings
documents = [models.Document(text=f"Document {i}", model=model_name) for i in range(10000)]

client.upload_collection(
    collection_name="large_collection",
    vectors=documents,
    batch_size=200,  # Larger batches for efficiency
    parallel=4       # Multiple parallel workers
)

from qdrant_client.hybrid import FusionQuery

# Upload documents with both dense and sparse embeddings
dense_docs = [models.Document(text=text, model="all-MiniLM-L6-v2") for text in texts]
sparse_docs = [models.Document(text=text, model="Qdrant/bm25") for text in texts]

# Create collection with multiple vectors
client.create_collection(
    collection_name="hybrid_collection",
    vectors_config={
        "dense": models.VectorParams(size=384, distance=models.Distance.COSINE),
        "sparse": models.SparseVectorParams()
    }
)

# Perform hybrid search
fusion_query = FusionQuery(
    dense_query=models.Document(text="search query", model="all-MiniLM-L6-v2"),
    sparse_query=models.Document(text="search query", model="Qdrant/bm25")
)

results = client.query_points(
    collection_name="hybrid_collection",
    query=fusion_query,
    limit=10
)

class FastEmbedNotInstalled(Exception):
    """Raised when FastEmbed is not installed but required."""

class ModelNotFound(Exception):
    """Raised when specified model is not available."""

class EmbeddingGenerationError(Exception):
    """Raised when embedding generation fails."""

from qdrant_client.fastembed_common import FastEmbedMisc

if not FastEmbedMisc.is_installed():
    print("FastEmbed not installed. Install with: pip install qdrant-client[fastembed]")
else:
    print("Available text models:", FastEmbedMisc.get_text_models())