Vector Embeddings

Vector embeddings enable semantic search by representing text, images, or other data as numerical vectors. VSKI provides native support for storing and searching embeddings.

The examples below use the Python SDK (pip install vski-sdk) and assume an authenticated client is in scope (see Collections).

Understanding Embeddings

Embeddings convert content (text, images, etc.) into high-dimensional vectors where similar items are close together in vector space. This enables:

• Semantic search (finding meaningfully similar content)
• Recommendation systems
• Clustering and classification
• Duplicate detection

Storing Embeddings

Upsert an Embedding

# Generate embedding (using OpenAI, Cohere, or local models)
embedding = await generate_embedding("Machine learning is amazing")

# Store embedding for a record
await client.embeddings.upsert(
    "documents",  # collection name
    "record-id",  # record ID
    embedding,    # vector
)

The upsert method creates or updates an embedding:

result = await client.embeddings.upsert(
    "documents",           # collection name
    "record-id-123",       # record ID
    [0.1, 0.2, 0.3, ...],  # list of floats (typically 128-1536 dimensions)
)

print(result["recordId"])    # "record-id-123"
print(result["dimensions"])  # 128 (or your vector dimension)
print(result["status"])      # "upserted"

Generating Embeddings

Using OpenAI

import os

import httpx

async def generate_embedding(text: str) -> list[float]:
    async with httpx.AsyncClient() as http:
        response = await http.post(
            "https://api.openai.com/v1/embeddings",
            headers={
                "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
            },
            json={
                "model": "text-embedding-3-small",
                "input": text,
            },
        )
        response.raise_for_status()
        return response.json()["data"][0]["embedding"]

Using Cohere

import os

import httpx

async def generate_embedding(text: str) -> list[float]:
    async with httpx.AsyncClient() as http:
        response = await http.post(
            "https://api.cohere.com/v2/embed",
            headers={
                "Authorization": f"Bearer {os.environ['COHERE_API_KEY']}",
            },
            json={
                "model": "embed-english-v3.0",
                "texts": [text],
                "input_type": "search_document",
                "embedding_types": ["float"],
            },
        )
        response.raise_for_status()
        return response.json()["embeddings"]["float"][0]

Using Local Models (sentence-transformers)

# pip install sentence-transformers
from sentence_transformers import SentenceTransformer

_model: SentenceTransformer | None = None

def generate_embedding(text: str) -> list[float]:
    global _model
    if _model is None:
        _model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

    vector = _model.encode(text, normalize_embeddings=True)
    return list(map(float, vector))

# sentence-transformers is synchronous; from async code run it in a thread:
#   embedding = await asyncio.to_thread(generate_embedding, text)

Searching Embeddings

Basic Similarity Search

query_embedding = await generate_embedding("AI and machine learning")

results = await client.embeddings.search("documents", query_embedding, limit=10)

print(results["results"])
# [
#   {
#     "record": {"id": "1", "title": "Machine Learning Tutorial", ...},
#     "distance": 0.23,
#   },
#   {
#     "record": {"id": "2", "title": "AI Research", ...},
#     "distance": 0.45,
#   },
#   ...
# ]
print(results["count"])  # 10 (number of results returned)
print(results["model"])  # echoed when the collection has a pinned model tag

Search with Threshold

Filter results by similarity threshold (lower distance = more similar):

results = await client.embeddings.search(
    "documents",
    query_embedding,
    limit=10,
    threshold=0.5,  # Only return results with distance <= 0.5
)

Custom Limit

Control how many results to return:

results = await client.embeddings.search(
    "documents",
    query_embedding,
    limit=5,  # Return top 5 most similar
)

Practical Examples

Semantic Document Search

# Store documents with embeddings
documents = await client.collection("documents").get_list(1, 1000)

for doc in documents["items"]:
    embedding = await generate_embedding(f"{doc['title']} {doc['content']}")
    # the model tag is pinned on the first tagged write
    await client.embeddings.upsert(
        "documents", doc["id"], embedding, model="text-embedding-3-small"
    )

# Search semantically
async def search_documents(query: str):
    embedding = await generate_embedding(query)
    results = await client.embeddings.search(
        "documents", embedding, limit=10, threshold=0.5
    )

    return [r["record"] for r in results["results"]]

# Usage
results = await search_documents("artificial intelligence tutorials")

Product Recommendations

# Store product embeddings
products = await client.collection("products").get_list(1, 1000)

for product in products["items"]:
    text = f"{product['name']} {product['description']} {product['category']}"
    embedding = await generate_embedding(text)
    await client.embeddings.upsert("products", product["id"], embedding)

# Find similar products
async def find_similar_products(product_id: str, limit: int = 5):
    product = await client.collection("products").get_one(product_id)
    text = f"{product['name']} {product['description']} {product['category']}"
    embedding = await generate_embedding(text)

    results = await client.embeddings.search(
        "products",
        embedding,
        limit=limit + 1,  # +1 to exclude the product itself
    )

    return [
        r["record"] for r in results["results"] if r["record"]["id"] != product_id
    ][:limit]

# Usage
similar = await find_similar_products("product-123", 5)

FAQ Search

# Store FAQ embeddings
faqs = await client.collection("faqs").get_list(1, 1000)

for faq in faqs["items"]:
    text = f"{faq['question']} {faq['answer']}"
    embedding = await generate_embedding(text)
    await client.embeddings.upsert("faqs", faq["id"], embedding)

# Semantic FAQ search
async def search_faqs(query: str):
    embedding = await generate_embedding(query)
    results = await client.embeddings.search(
        "faqs", embedding, limit=3, threshold=0.6
    )

    return [r["record"] for r in results["results"]]

# Usage
answers = await search_faqs("How do I reset my password?")

Content Moderation

# Store flagged content embeddings
flagged = await client.collection("flagged_content").get_list(1, 1000)

for item in flagged["items"]:
    embedding = await generate_embedding(item["content"])
    await client.embeddings.upsert("flagged_content", item["id"], embedding)

# Check if new content is similar to flagged content
async def check_similarity(content: str):
    embedding = await generate_embedding(content)
    results = await client.embeddings.search(
        "flagged_content", embedding, limit=3, threshold=0.3
    )

    if results["results"]:
        print("Content similar to flagged items found!")
        return [r["record"] for r in results["results"]]

    return []

Plagiarism Detection

# Store document embeddings for comparison
async def index_for_plagiarism_check(documents: list[dict]):
    for doc in documents:
        embedding = await generate_embedding(doc["content"])
        await client.embeddings.upsert("documents", doc["id"], embedding)

# Check for potential plagiarism
async def check_plagiarism(content: str):
    embedding = await generate_embedding(content)
    results = await client.embeddings.search(
        "documents",
        embedding,
        limit=5,
        threshold=0.2,  # Low threshold for high similarity
    )

    return [
        {
            "document": r["record"],
            "similarity": 1 - r["distance"],  # Convert distance to similarity
        }
        for r in results["results"]
    ]

Hybrid Search (Text + Vector)

Combine keyword search with semantic search:

import asyncio

async def hybrid_search(query: str, collection: str):
    embedding = await generate_embedding(query)

    # Vector search and text search in parallel
    vector_results, text_results = await asyncio.gather(
        client.embeddings.search(collection, embedding, limit=10),
        client.collection(collection).search(query, per_page=10),
    )

    # Combine and deduplicate
    combined: dict = {}

    for r in vector_results["results"]:
        combined[r["record"]["id"]] = {
            "record": r["record"],
            "vector_score": 1 - r["distance"],
            "text_score": 0.0,
        }

    for r in text_results["items"]:
        if r["id"] in combined:
            combined[r["id"]]["text_score"] = 1.0
        else:
            combined[r["id"]] = {"record": r, "vector_score": 0.0, "text_score": 1.0}

    # Calculate combined score (70% vector, 30% text)
    ranked = sorted(
        (
            {
                **entry["record"],
                "score": entry["vector_score"] * 0.7 + entry["text_score"] * 0.3,
            }
            for entry in combined.values()
        ),
        key=lambda item: item["score"],
        reverse=True,
    )

    return ranked[:10]

Embedding Update Strategies

Real-time Updates

Update embeddings when content changes:

# Create record with embedding
record = await client.collection("documents").create({
    "title": "New Document",
    "content": "Content here...",
})

embedding = await generate_embedding(f"{record['title']} {record['content']}")
await client.embeddings.upsert("documents", record["id"], embedding)

# Update embedding when record is updated
async def update_document(record_id: str, data: dict):
    updated = await client.collection("documents").update(record_id, data)

    embedding = await generate_embedding(f"{updated['title']} {updated['content']}")
    await client.embeddings.upsert("documents", record_id, embedding)

    return updated

Batch Updates

Reindex all records:

async def reindex_collection(collection_name: str):
    records = await client.collection(collection_name).get_list(1, 1000)

    for record in records["items"]:
        text = f"{record.get('title', '')} {record.get('content', '')}"
        embedding = await generate_embedding(text)
        await client.embeddings.upsert(collection_name, record["id"], embedding)

        # Rate limiting
        await asyncio.sleep(0.1)

    print(f"Reindexed {len(records['items'])} records")

Performance Optimization

Batch Processing

Process multiple embeddings in parallel:

async def batch_embed(
    texts: list[str], chunk_size: int = 32
) -> list[list[float]]:
    # Process in parallel, in chunks, to bound concurrency
    embeddings: list[list[float]] = []
    for start in range(0, len(texts), chunk_size):
        chunk = texts[start : start + chunk_size]
        embeddings.extend(
            await asyncio.gather(*(generate_embedding(text) for text in chunk))
        )

    return embeddings

Caching

Cache embeddings for frequently accessed content:

import time

embedding_cache: dict[str, tuple[float, list[float]]] = {}  # key -> (expires_at, vector)
cache_ttl = 3600.0  # seconds

async def get_cached_embedding(text: str) -> list[float]:
    cache_key = text.lower().strip()

    hit = embedding_cache.get(cache_key)
    if hit and hit[0] > time.monotonic():
        return hit[1]

    embedding = await generate_embedding(text)
    embedding_cache[cache_key] = (time.monotonic() + cache_ttl, embedding)

    return embedding

Dimension Selection

Choose appropriate embedding dimensions:

# Small models (faster, less memory): 128-384 dimensions
#   all-MiniLM-L6-v2 -> 384 dimensions
# Medium models (good balance): 768-1024 dimensions
# Large models (more accurate, slower): 1536+ dimensions
#   text-embedding-3-small -> 1536 dimensions

# Dimensions are fixed per collection by the first stored vector; the server
# rejects mismatched vectors (400). Tag the model on the first upsert and
# check what a collection holds:
info = await client.embeddings.info("documents")
print(info["dimensions"])  # e.g. 384
print(info["model"])       # the pinned model tag (None when untagged)
print(info["metric"])      # "l2" (euclidean distance)
print(info["count"])       # number of stored vectors

API Endpoints

Method	Endpoint	Description
POST	/api/collections/:name/embeddings	Upsert embedding for a record
POST	/api/collections/:name/embeddings/search	Search similar embeddings

Best Practices

1. Consistent text - Use the same text format for all embeddings
2. Normalize text - Clean and normalize text before embedding
3. Choose right model - Select embedding model based on use case
4. Batch when possible - Process multiple embeddings together
5. Cache results - Cache embeddings for repeated queries
6. Monitor performance - Track embedding generation and search times
7. Use appropriate dimensions - Balance accuracy and performance
8. Update embeddings - Re-embed when content changes
9. Set proper thresholds - Tune similarity thresholds for your use case
10. Test thoroughly - Validate search quality with real queries