# AI recipe recommendation agent
Source: https://docs.vectoraidb.actian.com/academy/articles/AI-recipe-recommendation-agent
Build an AI-powered recipe recommendation agent using Actian VectorAI DB that matches user cravings to recipes through semantic search, filters by dietary restrictions and available ingredients, and learns preferences over time.
Recipe recommendation is a deceptively hard search problem. A user who says "I want something warm and comforting with chicken" is expressing a feeling, not a set of keywords. Traditional keyword search fails for four reasons:
* "Warm and comforting" is a semantic concept — it maps to soups, stews, casseroles, and curries, but none of those words appear in the query.
* Dietary restrictions create hard constraints — a gluten-free user must never see recipes with wheat flour, regardless of semantic relevance.
* Available ingredients create soft preferences — "I have chicken, garlic, and tomatoes" should boost recipes using those ingredients without excluding others.
* User taste evolves — someone who keeps rating Thai dishes highly should see more Thai cuisine in future recommendations.
This article builds an AI recipe recommendation agent where Actian VectorAI DB handles all four dimensions: semantic understanding through embeddings, hard constraints through payload filters, soft preferences through `should`/`min_should` logic, and preference learning through payload updates.
## Prerequisites
Before starting this tutorial, make sure the following are in place:
* Python 3.10 or later is installed.
* An Actian VectorAI DB instance is running locally or accessible at a network address.
* Basic familiarity with Python async/await syntax is assumed.
## Architecture overview
The system takes a natural-language craving, converts it into an embedding, and searches Actian VectorAI DB with structured filters. Results are ranked by a combination of semantic similarity and stored preferences, and feedback loops back into the database to refine future recommendations.
The diagram below shows the end-to-end data flow from the user query through embedding, search, ranking, and preference learning.
```mermaid theme={null}
flowchart LR
User[User query - craving + constraints] --> Embed[Embedding model 384-dim]
Embed --> Search[Semantic search + filter DSL]
Recipes[(Actian VectorAI DB - recipes + metadata)] --> Search
Search --> Rank[Preference-weighted ranking]
Rank --> Results[Recommended recipes]
Results --> Feedback[User feedback]
Feedback --> Learn[Preference learning with set_payload]
Learn --> Recipes
```
## Environment setup
Before running any code, install the Actian VectorAI Python client and the `sentence-transformers` library. The following command installs both packages into the active Python environment.
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
## Implementation
The following steps walk through each layer of the recommendation agent — from initial setup and data ingestion to semantic search, constraint filtering, preference learning, and administration.
### Step 1: Import dependencies and configure
The block below imports all required modules, sets the server address and collection name, loads the embedding model once so every call reuses it, and defines two helper functions for single and batch embedding. Running this block prints the active configuration and confirms the environment is ready before any further steps.
```python theme={null}
import asyncio
from datetime import datetime, timezone
from sentence_transformers import SentenceTransformer
# Core Actian VectorAI client, distance metric, filter DSL, index param types, and point structures
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
Field,
FieldType,
FilterBuilder,
IntegerIndexParams,
TextIndexParams,
BoolIndexParams,
FloatIndexParams,
PointStruct,
PrefetchQuery,
SearchParams,
VectorParams,
)
# HNSW graph configuration for tuning the approximate nearest-neighbor index
from actian_vectorai.models.collections import HnswConfigDiff
# Fusion strategy (RRF) for multisignal ranking and word-tokenizer type for text indexes
from actian_vectorai.models.enums import Fusion, TokenizerType
# Server address and collection name used throughout this guide
SERVER = "localhost:6574"
COLLECTION = "Recipe-Recommendations"
EMBED_DIM = 384 # Output dimension of all-MiniLM-L6-v2
# Load the model once at module level so it is reused on every call
model = SentenceTransformer("all-MiniLM-L6-v2")
def embed_text(text: str) -> list[float]:
"""Embed a single string and return a flat list of floats."""
return model.encode(text).tolist()
def embed_texts(texts: list[str]) -> list[list[float]]:
"""Embed a list of strings in a single forward pass for efficiency."""
return model.encode(texts).tolist()
print(f"Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"Embedding: all-MiniLM-L6-v2 ({EMBED_DIM}-dim)")
```
#### Expected output
Running the block above prints the server address, collection name, and embedding model name to confirm the configuration loaded correctly.
```text theme={null}
Server: localhost:6574
Collection: Recipe-Recommendations
Embedding: all-MiniLM-L6-v2 (384-dim)
```
### Step 2: Create the recipe collection with payload indexes
Each recipe has structured metadata covering cuisine, dietary tags, ingredients, cook time, difficulty, and rating. The function below creates the vector collection with cosine similarity and a tuned HNSW graph, then registers eight payload indexes — one for each filter pattern used later in this guide. Running this function creates the collection if it does not already exist, attaches all eight indexes, and prints a confirmation message.
```python theme={null}
async def create_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
# Create the collection with cosine similarity and tuned HNSW graph parameters
await client.collections.get_or_create(
name=COLLECTION,
vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine),
hnsw_config=HnswConfigDiff(m=16, ef_construct=128),
)
# Keyword indexes support exact-match and multivalue filters on string fields
await client.points.create_field_index(
COLLECTION, field_name="cuisine",
field_type=FieldType.FieldTypeKeyword,
)
await client.points.create_field_index(
COLLECTION, field_name="diet_tags",
field_type=FieldType.FieldTypeKeyword,
)
await client.points.create_field_index(
COLLECTION, field_name="meal_type",
field_type=FieldType.FieldTypeKeyword,
)
# Word-tokenized text index enables ingredient keyword search (e.g. "garlic")
await client.points.create_field_index(
COLLECTION, field_name="ingredients_text",
field_type=FieldType.FieldTypeText,
field_index_params=TextIndexParams(
tokenizer=TokenizerType.Word,
lowercase=True,
min_token_len=2,
),
)
# Range-enabled integer index supports numeric comparisons such as lte(30)
await client.points.create_field_index(
COLLECTION, field_name="cook_time_min",
field_type=FieldType.FieldTypeInteger,
field_index_params=IntegerIndexParams(range=True, is_principal=True),
)
# Principal float index supports rating threshold filters such as gte(4.0)
await client.points.create_field_index(
COLLECTION, field_name="rating",
field_type=FieldType.FieldTypeFloat,
field_index_params=FloatIndexParams(is_principal=True),
)
# Boolean index enables strict vegetarian-only filtering
await client.points.create_field_index(
COLLECTION, field_name="is_vegetarian",
field_type=FieldType.FieldTypeBool,
field_index_params=BoolIndexParams(),
)
await client.points.create_field_index(
COLLECTION, field_name="difficulty",
field_type=FieldType.FieldTypeKeyword,
)
print(f"Collection '{COLLECTION}' ready with 8 payload indexes.")
asyncio.run(create_collection())
```
The table below shows each field, the index type chosen, and the filter operation it enables.
| `Field` | Index type | Filter pattern |
| ------------------ | --------------------- | ---------------------------------------------- |
| `cuisine` | Keyword | `eq("thai")`, `any_of(["thai", "indian"])` |
| `diet_tags` | Keyword | `any_of(["gluten-free", "dairy-free"])` |
| `meal_type` | Keyword | `eq("dinner")`, `any_of(["lunch", "dinner"])` |
| `ingredients_text` | Text (word tokenizer) | `text("chicken")` — full-text substring search |
| `cook_time_min` | Integer (range) | `lte(30)`, `between(15, 45)` |
| `rating` | Float (principal) | `gte(4.0)` — ordering by rating |
| `is_vegetarian` | Bool | `eq(True)` — boolean constraint |
| `difficulty` | Keyword | `eq("easy")`, `except_of(["hard"])` |
The `TextIndexParams` configuration with the `Word` tokenizer and `lowercase=True` enables ingredient keyword search. Calling `Field("ingredients_text").text("garlic")` finds any recipe whose ingredients list contains the word "garlic", regardless of case.
### Step 3: Prepare the recipe dataset
The list below defines twelve recipes spanning multiple cuisines, meal types, and dietary profiles. Each recipe includes the structured metadata that the indexes from step 2 will filter against. Running this block loads the dataset into memory and prints a count confirming all twelve recipes are ready for ingestion.
```python theme={null}
# Each dict represents one recipe with its description (used for embedding), structured metadata
# (used for payload filtering), and an ingredients list (joined into ingredients_text for text search)
recipes = [
{
"name": "Thai Green Curry with Chicken",
"description": "Aromatic coconut-based curry with tender chicken, bamboo shoots, and Thai basil in a spicy green paste.",
"cuisine": "thai",
"meal_type": "dinner",
"ingredients": ["chicken breast", "coconut milk", "green curry paste", "bamboo shoots", "thai basil", "fish sauce", "palm sugar", "kaffir lime leaves"],
"diet_tags": ["gluten-free", "dairy-free"],
"is_vegetarian": False,
"cook_time_min": 35,
"difficulty": "medium",
"rating": 4.7,
"servings": 4,
"calories_per_serving": 380,
},
{
"name": "Classic Italian Margherita Pizza",
"description": "Thin-crust pizza with San Marzano tomato sauce, fresh mozzarella, and basil leaves baked in a hot oven.",
"cuisine": "italian",
"meal_type": "dinner",
"ingredients": ["pizza dough", "san marzano tomatoes", "fresh mozzarella", "basil", "olive oil", "salt"],
"diet_tags": ["vegetarian"],
"is_vegetarian": True,
"cook_time_min": 20,
"difficulty": "medium",
"rating": 4.5,
"servings": 2,
"calories_per_serving": 520,
},
{
"name": "Japanese Miso Ramen",
"description": "Rich miso-based broth with ramen noodles, soft-boiled egg, chashu pork, corn, and green onions.",
"cuisine": "japanese",
"meal_type": "dinner",
"ingredients": ["ramen noodles", "miso paste", "pork belly", "soft-boiled egg", "corn", "green onions", "nori", "sesame oil"],
"diet_tags": ["dairy-free"],
"is_vegetarian": False,
"cook_time_min": 60,
"difficulty": "hard",
"rating": 4.8,
"servings": 2,
"calories_per_serving": 620,
},
{
"name": "Mexican Street Corn Salad",
"description": "Grilled corn kernels tossed with lime, chili powder, cotija cheese, cilantro, and creamy mayo.",
"cuisine": "mexican",
"meal_type": "lunch",
"ingredients": ["corn", "cotija cheese", "lime", "chili powder", "cilantro", "mayonnaise", "garlic"],
"diet_tags": ["gluten-free", "vegetarian"],
"is_vegetarian": True,
"cook_time_min": 15,
"difficulty": "easy",
"rating": 4.3,
"servings": 4,
"calories_per_serving": 210,
},
{
"name": "Indian Butter Chicken",
"description": "Tender chicken pieces in a creamy tomato-based sauce with butter, cream, and aromatic spices like garam masala.",
"cuisine": "indian",
"meal_type": "dinner",
"ingredients": ["chicken thighs", "tomato puree", "butter", "cream", "garam masala", "cumin", "garlic", "ginger", "fenugreek"],
"diet_tags": ["gluten-free"],
"is_vegetarian": False,
"cook_time_min": 45,
"difficulty": "medium",
"rating": 4.9,
"servings": 4,
"calories_per_serving": 450,
},
{
"name": "Mediterranean Quinoa Bowl",
"description": "Protein-packed quinoa bowl with roasted vegetables, chickpeas, feta cheese, olives, and lemon tahini dressing.",
"cuisine": "mediterranean",
"meal_type": "lunch",
"ingredients": ["quinoa", "chickpeas", "bell pepper", "cucumber", "feta cheese", "kalamata olives", "cherry tomatoes", "tahini", "lemon"],
"diet_tags": ["vegetarian", "gluten-free"],
"is_vegetarian": True,
"cook_time_min": 25,
"difficulty": "easy",
"rating": 4.4,
"servings": 2,
"calories_per_serving": 340,
},
{
"name": "Korean Bibimbap",
"description": "Mixed rice bowl topped with sautéed vegetables, seasoned beef, a fried egg, and spicy gochujang sauce.",
"cuisine": "korean",
"meal_type": "dinner",
"ingredients": ["rice", "beef", "spinach", "carrots", "zucchini", "bean sprouts", "egg", "gochujang", "sesame oil", "garlic"],
"diet_tags": ["dairy-free"],
"is_vegetarian": False,
"cook_time_min": 40,
"difficulty": "medium",
"rating": 4.6,
"servings": 2,
"calories_per_serving": 480,
},
{
"name": "French Onion Soup",
"description": "Deeply caramelized onions simmered in rich beef broth, topped with crusty bread and melted Gruyère cheese.",
"cuisine": "french",
"meal_type": "dinner",
"ingredients": ["onions", "beef broth", "butter", "gruyère cheese", "baguette", "thyme", "bay leaf", "white wine"],
"diet_tags": [],
"is_vegetarian": False,
"cook_time_min": 75,
"difficulty": "medium",
"rating": 4.5,
"servings": 4,
"calories_per_serving": 310,
},
{
"name": "Chickpea and Spinach Curry",
"description": "Hearty vegan curry with chickpeas and spinach in a spiced coconut tomato sauce, served over basmati rice.",
"cuisine": "indian",
"meal_type": "dinner",
"ingredients": ["chickpeas", "spinach", "coconut milk", "tomatoes", "onion", "garlic", "ginger", "cumin", "turmeric", "coriander"],
"diet_tags": ["vegan", "gluten-free", "dairy-free", "vegetarian"],
"is_vegetarian": True,
"cook_time_min": 30,
"difficulty": "easy",
"rating": 4.6,
"servings": 4,
"calories_per_serving": 280,
},
{
"name": "American BBQ Pulled Pork Sandwich",
"description": "Slow-smoked pork shoulder shredded and tossed in tangy BBQ sauce, served on a brioche bun with coleslaw.",
"cuisine": "american",
"meal_type": "lunch",
"ingredients": ["pork shoulder", "bbq sauce", "brioche bun", "cabbage", "apple cider vinegar", "paprika", "brown sugar", "garlic powder"],
"diet_tags": ["dairy-free"],
"is_vegetarian": False,
"cook_time_min": 240,
"difficulty": "hard",
"rating": 4.7,
"servings": 6,
"calories_per_serving": 550,
},
{
"name": "Greek Lemon Chicken Soup (Avgolemono)",
"description": "Silky egg-lemon soup with tender chicken, orzo pasta, and fresh dill, a classic Greek comfort dish.",
"cuisine": "greek",
"meal_type": "dinner",
"ingredients": ["chicken", "orzo", "eggs", "lemon", "chicken broth", "dill", "olive oil", "onion"],
"diet_tags": ["dairy-free"],
"is_vegetarian": False,
"cook_time_min": 40,
"difficulty": "medium",
"rating": 4.4,
"servings": 6,
"calories_per_serving": 290,
},
{
"name": "Vietnamese Pho Bo",
"description": "Fragrant beef broth infused with star anise, cinnamon, and cloves, served with rice noodles, rare beef, and fresh herbs.",
"cuisine": "vietnamese",
"meal_type": "dinner",
"ingredients": ["beef bones", "rice noodles", "rare beef", "star anise", "cinnamon", "cloves", "ginger", "fish sauce", "bean sprouts", "thai basil", "lime", "hoisin sauce"],
"diet_tags": ["gluten-free", "dairy-free"],
"is_vegetarian": False,
"cook_time_min": 180,
"difficulty": "hard",
"rating": 4.9,
"servings": 4,
"calories_per_serving": 420,
},
]
print(f"{len(recipes)} recipes loaded.")
```
### Step 4: Embed and ingest recipes
The function below batch-embeds all recipe descriptions in a single model forward pass, constructs one point per recipe by pairing its vector with its full metadata payload, and upserts all twelve points in one call. The ingredients list is joined into a space-separated string and stored as `ingredients_text` so the text index can match individual ingredient words. Running this function inserts all twelve recipes into the collection and prints the total stored count to confirm the write succeeded.
```python theme={null}
async def ingest_recipes():
# Embed all descriptions at once to avoid one model call per recipe
descriptions = [r["description"] for r in recipes]
vectors = embed_texts(descriptions)
points = []
for i, (recipe, vector) in enumerate(zip(recipes, vectors)):
payload = {**recipe}
# Convert the ingredients list to a space-separated string for the text index
payload["ingredients_text"] = " ".join(recipe["ingredients"])
points.append(PointStruct(id=i, vector=vector, payload=payload))
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.upsert(COLLECTION, points=points)
# Flush writes to disk before reading the count to ensure accuracy
await client.vde.flush(COLLECTION)
count = await client.vde.get_vector_count(COLLECTION)
print(f"Ingested {len(points)} recipes. Total in collection: {count}")
asyncio.run(ingest_recipes())
```
#### Expected output
The function batch-embeds all twelve recipe descriptions in a single model forward pass, converts each description into a 384-dimensional vector, and upserts all twelve points into the collection in one call. The `ingredients_text` field is constructed by joining each recipe's ingredient list into a space-separated string so the word-tokenized text index can match individual ingredient words. After writing, the collection is flushed to disk and the total point count is retrieved to confirm that all twelve records were stored successfully.
```text theme={null}
Ingested 12 recipes. Total in collection: 12
```
### Step 5: Basic semantic search — "what am I craving?"
The simplest recommendation matches a craving to recipe descriptions by meaning alone, with no structural filters applied. The function below embeds the query string and searches the collection by cosine similarity, returning the top results ordered by score. Running this block with the query "I want something warm and comforting with a rich broth" returns broth-based recipes ranked by how closely their descriptions match the expressed feeling.
```python theme={null}
async def search_by_craving(query: str, top_k: int = 5):
"""Return the top matching recipes for a natural-language craving, with no filters applied."""
vec = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
# Search by cosine similarity only — no payload filter is applied
results = await client.points.search(
COLLECTION, vector=vec, limit=top_k, with_payload=True,
) or []
return results
query = "I want something warm and comforting with a rich broth"
results = asyncio.run(search_by_craving(query))
print(f"Query: {query}\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} {p['name']} [{p['cuisine']}] {p['cook_time_min']}min ★{p['rating']}")
```
#### Expected output
The function embeds the natural-language craving "I want something warm and comforting with a rich broth" into a 384-dimensional vector and performs a cosine similarity search across all twelve recipes with no payload filters applied. Results are returned in descending score order, where each score reflects how closely a recipe's embedded description matches the semantic meaning of the query. Broth-based dishes such as soups, ramen, and pho rank highest because their descriptions carry similar semantic content to the expressed feeling.
```text theme={null}
Query: I want something warm and comforting with a rich broth
score=0.6812 Vietnamese Pho Bo [vietnamese] 180min ★4.9
score=0.6534 French Onion Soup [french] 75min ★4.5
score=0.6210 Greek Lemon Chicken Soup (Avgolemono) [greek] 40min ★4.4
score=0.5890 Japanese Miso Ramen [japanese] 60min ★4.8
score=0.4567 Chickpea and Spinach Curry [indian] 30min ★4.6
```
### Step 6: Dietary restrictions — hard constraints with `must`
Dietary restrictions are non-negotiable — a gluten-free user must never see a recipe containing gluten regardless of how high it scores semantically. The function below adds a `must` condition for each supplied diet tag, so only recipes that carry every tag are returned. Running this block with `diet_tags=["gluten-free", "dairy-free"]` returns only the recipes whose `diet_tags` array contains both labels.
```python theme={null}
async def search_with_diet(query: str, diet_tags: list[str], top_k: int = 5):
"""Return semantically matched recipes that satisfy all supplied dietary constraints."""
vec = embed_text(query)
# Each diet tag becomes a separate must condition, enforcing AND logic across all tags
fb = FilterBuilder()
for tag in diet_tags:
fb = fb.must(Field("diet_tags").any_of([tag]))
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION, vector=vec, limit=top_k,
filter=filter_obj, with_payload=True,
) or []
return results
query = "spicy curry with coconut"
results = asyncio.run(search_with_diet(query, diet_tags=["gluten-free", "dairy-free"]))
print(f"Query: {query}")
print(f"Diet: gluten-free AND dairy-free\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} {p['name']} tags={p['diet_tags']}")
```
#### Expected output
The function embeds the query "spicy curry with coconut" and applies two `must` conditions — one for `"gluten-free"` and one for `"dairy-free"` — so only recipes whose `diet_tags` array contains both labels are eligible. The filter `Field("diet_tags").any_of(["gluten-free"])` matches any recipe that carries the tag, and wrapping each tag in `must` enforces AND logic so every supplied restriction must be satisfied before a recipe can appear in the results. The scores reflect semantic closeness to the craving within the filtered candidate set.
```text theme={null}
Query: spicy curry with coconut
Diet: gluten-free AND dairy-free
score=0.7123 Thai Green Curry with Chicken tags=['gluten-free', 'dairy-free']
score=0.6234 Chickpea and Spinach Curry tags=['vegan', 'gluten-free', 'dairy-free', 'vegetarian']
score=0.4567 Vietnamese Pho Bo tags=['gluten-free', 'dairy-free']
```
### Step 7: Available ingredients — soft preferences with `should` and `min_should`
Unlike dietary restrictions, ingredient availability is a soft preference. Recipes that use available ingredients should rank higher, but recipes that do not use them should not be excluded entirely. The function below adds a `should` condition for each ingredient and requires at least `min_match` of them to appear in the result. Running this block with `available=["chicken", "garlic", "tomatoes", "onion", "cream"]` and `min_match=2` returns recipes that contain at least two of those five ingredients, ranked by semantic similarity to the query.
```python theme={null}
async def search_with_available_ingredients(
query: str,
available: list[str],
min_match: int = 1,
top_k: int = 5,
):
"""Return recipes that match the craving and contain at least min_match of the available ingredients."""
vec = embed_text(query)
# Each ingredient is a should condition — preferred but not required
fb = FilterBuilder()
for ingredient in available:
fb = fb.should(Field("ingredients_text").text(ingredient))
# Require at least min_match of the should conditions to be satisfied
fb = fb.min_should(min_match)
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION, vector=vec, limit=top_k,
filter=filter_obj, with_payload=True,
) or []
return results
query = "quick dinner tonight"
available = ["chicken", "garlic", "tomatoes", "onion", "cream"]
results = asyncio.run(search_with_available_ingredients(query, available, min_match=2))
print(f"Query: {query}")
print(f"Available: {available} (at least 2 must match)\n")
for r in results:
p = r.payload
matched = [ing for ing in available if ing in p.get("ingredients_text", "").lower()]
print(f" score={r.score:.4f} {p['name']} matched={matched}")
```
The snippet below shows the relationship between `should` and `min_should`. Each `should` call adds one OR candidate; `min_should` sets the minimum number of those candidates that must match for a point to qualify.
```python theme={null}
fb = FilterBuilder()
fb = fb.should(Field("ingredients_text").text("chicken")) # OR candidate 1
fb = fb.should(Field("ingredients_text").text("garlic")) # OR candidate 2
fb = fb.should(Field("ingredients_text").text("tomatoes")) # OR candidate 3
fb = fb.min_should(2) # The point must match at least 2 of the 3 candidates
```
The `min_should` value controls how strictly the result set matches the available pantry. The table below shows how each value changes the behavior.
| `min_should` | Behavior |
| ------------ | ----------------------------------------------- |
| 1 (default) | At least one ingredient matches — very lenient. |
| 2 | At least two ingredients match — moderate. |
| 3 | All three ingredients match — strict. |
#### Expected output
The function embeds the query "quick dinner tonight" and applies a `should` condition for each of the five available ingredients — chicken, garlic, tomatoes, onion, and cream — with `min_should(2)` requiring that at least two of them appear in a recipe's `ingredients_text` field. Recipes are ranked by cosine similarity to the craving vector within the filtered candidate set, and each result shows which available ingredients it matched.
```text theme={null}
Query: quick dinner tonight
Available: ['chicken', 'garlic', 'tomatoes', 'onion', 'cream'] (at least 2 must match)
score=0.5432 Indian Butter Chicken matched=['chicken', 'garlic', 'cream']
score=0.4987 Chickpea and Spinach Curry matched=['tomatoes', 'onion', 'garlic']
score=0.4321 Greek Lemon Chicken Soup (Avgolemono) matched=['chicken', 'onion']
```
### Step 8: Exclude allergens — `must_not` and `except_of`
Some ingredients must be strictly excluded because of allergies or strong dislikes. The function below adds a `must_not` condition for each ingredient to exclude, so no returned recipe contains any of them in its `ingredients_text` field. Running this block with `exclude=["pork", "fish sauce"]` returns only recipes whose ingredient lists contain neither ingredient, ranked by semantic similarity to the query.
```python theme={null}
async def search_excluding_ingredients(
query: str,
exclude: list[str],
top_k: int = 5,
):
"""Return semantically matched recipes that contain none of the excluded ingredients."""
vec = embed_text(query)
# Each must_not condition removes any recipe containing that ingredient
fb = FilterBuilder()
for ingredient in exclude:
fb = fb.must_not(Field("ingredients_text").text(ingredient))
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION, vector=vec, limit=top_k,
filter=filter_obj, with_payload=True,
) or []
return results
query = "creamy pasta or rice dish"
results = asyncio.run(search_excluding_ingredients(query, exclude=["pork", "fish sauce"]))
print(f"Query: {query}")
print(f"Excluding: pork, fish sauce\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} {p['name']} [{p['cuisine']}]")
```
#### Expected output
The function embeds the query "creamy pasta or rice dish" and applies a `must_not` condition for each excluded ingredient — pork and fish sauce — so any recipe whose `ingredients_text` field contains either word is removed from the candidate set before scoring. The remaining recipes are ranked by cosine similarity to the craving vector, and each result shows the cuisine it belongs to. Dishes like Japanese Miso Ramen (pork belly) and Vietnamese Pho Bo (fish sauce) are absent from the results because they were eliminated by the exclusion filters.
```text theme={null}
Query: creamy pasta or rice dish
Excluding: pork, fish sauce
score=0.6012 Indian Butter Chicken [indian]
score=0.5781 Mediterranean Quinoa Bowl [mediterranean]
score=0.5432 Chickpea and Spinach Curry [indian]
score=0.4987 Korean Bibimbap [korean]
score=0.4321 Greek Lemon Chicken Soup (Avgolemono) [greek]
```
### Step 9: Combined constraints — the full recommendation query
The function below combines all constraint types into a single search call: dietary filters, a cook-time ceiling, difficulty exclusions, a rating floor, cuisine preferences, and ingredient boosts. Running this block with `diet_tags=["gluten-free"]`, `max_cook_time=60`, `exclude_difficulty=["hard"]`, `min_rating=4.0`, and `preferred_ingredients=["chicken", "coconut milk"]` returns gluten-free dinner recipes that take no more than 60 minutes, are not hard difficulty, have a rating of at least 4.0, and preferably contain chicken or coconut milk.
```python theme={null}
async def full_recommendation(
craving: str,
diet_tags: list[str] = None,
max_cook_time: int = None,
exclude_difficulty: list[str] = None,
preferred_ingredients: list[str] = None,
vegetarian_only: bool = False,
min_rating: float = None,
preferred_cuisines: list[str] = None,
top_k: int = 5,
):
"""Return the top recommendations that satisfy all hard constraints and prefer the soft ones."""
vec = embed_text(craving)
fb = FilterBuilder()
# Hard constraints — every must condition must be satisfied for a recipe to qualify
if diet_tags:
for tag in diet_tags:
fb = fb.must(Field("diet_tags").any_of([tag]))
if max_cook_time:
fb = fb.must(Field("cook_time_min").lte(float(max_cook_time)))
if exclude_difficulty:
fb = fb.must_not(Field("difficulty").any_of(exclude_difficulty))
if vegetarian_only:
fb = fb.must(Field("is_vegetarian").eq(True))
if min_rating:
fb = fb.must(Field("rating").gte(min_rating))
if preferred_cuisines:
fb = fb.must(Field("cuisine").any_of(preferred_cuisines))
# Soft preferences — boost recipes that match, but do not exclude those that do not
if preferred_ingredients:
for ing in preferred_ingredients:
fb = fb.should(Field("ingredients_text").text(ing))
fb = fb.min_should(1)
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION, vector=vec, limit=top_k,
filter=filter_obj, with_payload=True,
# Higher hnsw_ef improves recall at the cost of slightly more compute
params=SearchParams(hnsw_ef=128),
) or []
return results
results = asyncio.run(full_recommendation(
craving="something spicy and satisfying for dinner",
diet_tags=["gluten-free"],
max_cook_time=60,
exclude_difficulty=["hard"],
min_rating=4.0,
preferred_ingredients=["chicken", "coconut milk"],
))
print("=== Full recommendation ===")
print("Craving: something spicy and satisfying")
print("Constraints: gluten-free, <=60min, not hard, rating>=4.0")
print("Preferences: chicken, coconut milk\n")
for r in results:
p = r.payload
print(
f" score={r.score:.4f} {p['name']} [{p['cuisine']}] "
f"{p['cook_time_min']}min {p['difficulty']} ★{p['rating']} "
f"tags={p['diet_tags']}"
)
```
#### Expected output
The function runs a single search combining all constraint types. The craving "something spicy and satisfying for dinner" is embedded into a vector and used for cosine similarity scoring. Hard `must` conditions enforce that results are gluten-free, take no more than 60 minutes, exclude hard-difficulty recipes, and have a rating of at least 4.0. Soft `should` conditions boost recipes that contain chicken or coconut milk without excluding those that do not. The `hnsw_ef=128` parameter is passed to increase recall at query time. Each result shows the cuisine, cook time, difficulty, rating, and dietary tags.
```text theme={null}
=== Full recommendation ===
Craving: something spicy and satisfying
Constraints: gluten-free, <=60min, not hard, rating>=4.0
Preferences: chicken, coconut milk
score=0.6234 Thai Green Curry with Chicken [thai] 35min medium ★4.7 tags=['gluten-free', 'dairy-free']
score=0.5890 Indian Butter Chicken [indian] 45min medium ★4.9 tags=['gluten-free']
score=0.5432 Chickpea and Spinach Curry [indian] 30min easy ★4.6 tags=['vegan', 'gluten-free', 'dairy-free', 'vegetarian']
```
### Step 10: Batch recommendations for meal planning
The `search_batch` method sends multiple queries to the server in a single network call instead of one call per meal. The function below builds one query per meal — each with its own craving vector, meal-type filter, cook-time limit, and optional vegetarian flag — then dispatches all queries at once. Running this block with the three meal requests defined below returns a ranked list for each meal and prints a summary of scores and cook times.
```python theme={null}
async def meal_plan(meals: list[dict], top_k: int = 3):
"""Generate recommendations for multiple meals in one batch call to avoid per-query round-trips."""
searches = []
for meal in meals:
vec = embed_text(meal["craving"])
# Each meal gets its own filter: meal type, optional cook time, optional vegetarian flag
fb = FilterBuilder()
fb = fb.must(Field("meal_type").any_of([meal.get("meal_type", "dinner")]))
if meal.get("max_cook_time"):
fb = fb.must(Field("cook_time_min").lte(float(meal["max_cook_time"])))
if meal.get("vegetarian"):
fb = fb.must(Field("is_vegetarian").eq(True))
searches.append({
"vector": vec,
"limit": top_k,
"filter": fb.build(),
"with_payload": True,
})
# All meal queries execute in a single gRPC round-trip
async with AsyncVectorAIClient(url=SERVER) as client:
batch_results = await client.points.search_batch(COLLECTION, searches=searches)
return batch_results
meals = [
{"craving": "light healthy lunch", "meal_type": "lunch", "max_cook_time": 30, "vegetarian": True},
{"craving": "hearty warm dinner", "meal_type": "dinner", "max_cook_time": 60},
{"craving": "quick Asian dinner", "meal_type": "dinner", "max_cook_time": 45},
]
all_results = asyncio.run(meal_plan(meals))
for i, (meal, results) in enumerate(zip(meals, all_results)):
print(f"\nMeal {i+1}: '{meal['craving']}' ({meal['meal_type']}, <={meal['max_cook_time']}min)")
for r in results:
p = r.payload
print(f" score={r.score:.4f} {p['name']} {p['cook_time_min']}min ★{p['rating']}")
```
#### Expected output
The function builds three independent search queries — one for a light vegetarian lunch under 30 minutes, one for a hearty dinner under 60 minutes, and one for a quick Asian dinner under 45 minutes — and dispatches all three in a single `search_batch` call. Each query uses its own craving vector and filter combination. Without batching, three meals require three separate network round-trips. With `search_batch`, all three queries execute in a single gRPC call, reducing latency from three round-trips to one. Each meal's results are printed with their similarity score, cook time, and rating.
```text theme={null}
Meal 1: 'light healthy lunch' (lunch, <=30min)
score=0.5678 Mediterranean Quinoa Bowl 25min ★4.4
score=0.4321 Mexican Street Corn Salad 15min ★4.3
Meal 2: 'hearty warm dinner' (dinner, <=60min)
score=0.6123 Indian Butter Chicken 45min ★4.9
score=0.5890 Thai Green Curry with Chicken 35min ★4.7
score=0.5432 Korean Bibimbap 40min ★4.6
Meal 3: 'quick Asian dinner' (dinner, <=45min)
score=0.5987 Thai Green Curry with Chicken 35min ★4.7
score=0.5654 Korean Bibimbap 40min ★4.6
score=0.4890 Chickpea and Spinach Curry 30min ★4.6
```
### Step 11: User preference learning
When a user rates a recipe, its metadata can be updated to influence future recommendations without re-ingesting the entire dataset. The function below fetches the current payload for a recipe, merges the new user rating into the existing `user_ratings` map, recomputes the aggregate average across all stored ratings, and writes only the changed fields back using `set_payload`. Running this block records a rating of 5.0 from user-alice and 4.5 from user-bob for recipe 0 (Thai Green Curry), and a rating of 4.8 from user-alice for recipe 4 (Indian Butter Chicken).
```python theme={null}
async def record_user_feedback(recipe_id: int, user_id: str, user_rating: float, liked: bool):
"""Merge a new user rating into the recipe's payload without overwriting unrelated fields."""
async with AsyncVectorAIClient(url=SERVER) as client:
# Fetch the current payload to read existing user_ratings before merging
existing = await client.points.get(COLLECTION, ids=[recipe_id], with_payload=True)
if not existing:
print(f"Recipe {recipe_id} not found.")
return
payload = existing[0].payload or {}
user_ratings = payload.get("user_ratings", {})
user_ratings[user_id] = {
"rating": user_rating,
"liked": liked,
"rated_at": datetime.now(timezone.utc).isoformat(),
}
# Recompute the aggregate average across all users after adding the new rating
total_user_ratings = [v["rating"] for v in user_ratings.values()]
avg_user_rating = sum(total_user_ratings) / len(total_user_ratings)
# Write only the updated fields; all other payload fields remain unchanged
await client.points.set_payload(
COLLECTION,
payload={
"user_ratings": user_ratings,
"avg_user_rating": avg_user_rating,
"total_ratings": len(total_user_ratings),
},
ids=[recipe_id],
)
print(f"Recorded feedback for recipe {recipe_id}: user={user_id}, rating={user_rating}, liked={liked}")
asyncio.run(record_user_feedback(0, "user-alice", 5.0, True))
asyncio.run(record_user_feedback(0, "user-bob", 4.5, True))
asyncio.run(record_user_feedback(4, "user-alice", 4.8, True))
```
### Step 12: Personalized recommendations with preference boosting
After recording feedback, recommendations can be personalized by blending the current query vector with vectors from previously liked recipes. The function below collects the stored vectors for all recipes a user has liked, averages them into a taste-profile vector, then issues a two-stage prefetch query: one stage retrieves candidates by the current craving vector and the other by the taste-profile vector. Reciprocal Rank Fusion (RRF) then merges the two ranked lists into a single result. Running this block for "user-alice" — who liked Thai Green Curry and Butter Chicken — returns results that reflect both the current query and her recorded preferences.
```python theme={null}
async def personalized_recommend(
query: str,
user_id: str,
top_k: int = 5,
):
"""Blend the current craving with the user's taste history using prefetch and RRF fusion."""
vec = embed_text(query)
# Retrieve all recipe points, including their vectors, to find which ones this user liked
async with AsyncVectorAIClient(url=SERVER) as client:
all_count = await client.vde.get_vector_count(COLLECTION)
all_points = await client.points.get(
COLLECTION, ids=list(range(all_count)), with_payload=True, with_vectors=True,
)
# Collect the embedding vectors of every recipe this user has marked as liked
liked_vectors = []
for p in all_points:
ur = (p.payload or {}).get("user_ratings", {})
if user_id in ur and ur[user_id].get("liked"):
if p.vectors:
liked_vectors.append(p.vectors)
# If no preference history exists, fall back to plain semantic search
if not liked_vectors:
async with AsyncVectorAIClient(url=SERVER) as client:
return await client.points.search(
COLLECTION, vector=vec, limit=top_k, with_payload=True,
) or []
# Average the liked recipe vectors into a single taste-profile vector
avg_liked = [sum(dim) / len(liked_vectors) for dim in zip(*liked_vectors)]
# Run two prefetch queries and merge their ranked results with RRF
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
COLLECTION,
query={"fusion": Fusion.RRF},
prefetch=[
PrefetchQuery(query=vec, limit=10), # Candidates matching the current craving
PrefetchQuery(query=avg_liked, limit=10), # Candidates matching the historical taste profile
],
limit=top_k,
with_payload=True,
)
return results
results = asyncio.run(personalized_recommend("dinner tonight", user_id="user-alice"))
print("=== Personalized for user-alice ===")
print("(Alice liked Thai Green Curry and Butter Chicken)\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} {p['name']} [{p['cuisine']}] ★{p['rating']}")
```
The diagram below shows how the two prefetch stages combine into a single ranked result. Recipes that appear in both candidate lists — matching both the current craving and the historical taste profile — rank highest after fusion.
```text theme={null}
Prefetch 1: Search by craving vector → 10 candidates (what the user wants now)
Prefetch 2: Search by avg-liked vector → 10 candidates (what the user has liked before)
RRF fusion: merge by rank → top 5 (balances current craving with historical taste)
```
### Step 13: Delete user data — GDPR compliance
To honor a right-to-erasure request, all stored ratings and preference data for a specific user must be removed from every recipe in the collection. The function below iterates over all recipe points, removes the target user's entry from each `user_ratings` map using `set_payload`, recomputes the aggregate statistics from the remaining ratings, and writes the updated payload back. Running this block for "user-bob" removes his ratings from every recipe that stored them and prints the count of updated records.
```python theme={null}
async def delete_user_data(user_id: str):
"""Remove all ratings and preference data for the given user from every recipe in the collection."""
async with AsyncVectorAIClient(url=SERVER) as client:
total = await client.vde.get_vector_count(COLLECTION)
all_points = await client.points.get(
COLLECTION, ids=list(range(total)), with_payload=True,
)
updated = 0
for p in all_points:
ur = (p.payload or {}).get("user_ratings", {})
if user_id in ur:
# Remove this user's entry, then recompute aggregates from the remaining ratings
del ur[user_id]
total_ratings = [v["rating"] for v in ur.values()]
avg = sum(total_ratings) / len(total_ratings) if total_ratings else 0.0
await client.points.set_payload(
COLLECTION,
payload={
"user_ratings": ur,
"avg_user_rating": avg,
"total_ratings": len(total_ratings),
},
ids=[p.id],
)
updated += 1
print(f"Removed {user_id}'s data from {updated} recipes.")
asyncio.run(delete_user_data("user-bob"))
```
### Step 14: Collection administration
The function below queries four collection endpoints in sequence to gather health metrics — state, recipe count, segment count, storage bytes, and index memory usage — then flushes any pending writes to disk. Running this block after all previous steps prints a full snapshot of the collection state and confirms that buffered writes have been persisted.
```python theme={null}
async def admin():
"""Print collection health metrics and flush pending writes to disk."""
async with AsyncVectorAIClient(url=SERVER) as client:
# Collect metadata from four separate endpoints in sequence
count = await client.vde.get_vector_count(COLLECTION)
state = await client.vde.get_state(COLLECTION)
stats = await client.vde.get_stats(COLLECTION)
info = await client.collections.get_info(COLLECTION)
print(f"Collection: {COLLECTION}")
print(f" State: {state}")
print(f" Recipes: {count}")
print(f" Segments: {info.segments_count}")
print(f" Storage: {stats.storage_bytes / 1024:.1f} KB")
print(f" Index: {stats.index_memory_bytes / 1024:.1f} KB")
# Persist any buffered writes to disk
await client.vde.flush(COLLECTION)
print(" Flushed to disk.")
# Uncomment the line below to permanently remove the collection:
# await client.collections.delete(COLLECTION)
asyncio.run(admin())
```
## Filter patterns used in this article
The table below summarizes every filter pattern used in the recommendation agent, the API call that implements it, and an example value.
| Pattern | API | Example |
| --------------------------- | --------------------------------------------- | ---------------------------------- |
| Exact match | `Field("cuisine").eq("thai")` | Match one cuisine. |
| Multivalue match | `Field("cuisine").any_of(["thai", "indian"])` | Match any of several cuisines. |
| Exclusion | `Field("difficulty").except_of(["hard"])` | Exclude hard recipes. |
| Full-text ingredient search | `Field("ingredients_text").text("garlic")` | Keyword search in ingredient list. |
| Numeric range | `Field("cook_time_min").lte(30)` | Maximum cook time. |
| Float threshold | `Field("rating").gte(4.0)` | Minimum rating. |
| Boolean | `Field("is_vegetarian").eq(True)` | Vegetarian only. |
| AND logic | `FilterBuilder().must(...)` | All constraints must match. |
| OR logic | `FilterBuilder().should(...)` | Preferred but not required. |
| Minimum match | `FilterBuilder().min_should(2)` | At least N preferences match. |
| Exclusion logic | `FilterBuilder().must_not(...)` | Allergen or ingredient exclusion. |
## Actian VectorAI features used
The table below maps each Actian VectorAI feature to the API method and its role in the recommendation pipeline.
| Feature | API | Purpose |
| ----------------------- | ----------------------------------------------------- | ------------------------------------------ |
| Collection creation | `collections.get_or_create(hnsw_config=...)` | Recipe vector space. |
| Point upsert | `points.upsert()` | Store recipe embeddings with metadata. |
| Semantic search | `points.search(filter=..., params=...)` | Craving-to-recipe matching. |
| Search batch | `points.search_batch(searches=[...])` | Multimeal planning in one call. |
| Server-side fusion | `query(query={"fusion": Fusion.RRF}, prefetch=[...])` | Personalized preference fusion. |
| Prefetch | `PrefetchQuery(query=..., limit=...)` | Multisignal candidate retrieval. |
| Point retrieval | `points.get(with_vectors=True)` | Load liked recipe vectors. |
| Payload merge | `points.set_payload(payload=...)` | Record user ratings. |
| Keyword index | `FieldType.FieldTypeKeyword` | Cuisine, diet tags, meal type, difficulty. |
| Text index | `TextIndexParams(tokenizer=Word, lowercase=True)` | Full-text ingredient search. |
| Bool index | `BoolIndexParams()` | Vegetarian flag. |
| Integer index (range) | `IntegerIndexParams(range=True, is_principal=True)` | Cook time range queries. |
| Float index (principal) | `FloatIndexParams(is_principal=True)` | Rating threshold and ordering. |
| `any_of` filter | `Field("diet_tags").any_of([...])` | Multivalue dietary matching. |
| `except_of` filter | `Field("difficulty").except_of([...])` | Difficulty exclusion. |
| `text` filter | `Field("ingredients_text").text("garlic")` | Ingredient keyword search. |
| `should` / `min_should` | `FilterBuilder().should(...).min_should(2)` | Soft ingredient preferences. |
| `must_not` | `FilterBuilder().must_not(...)` | Allergen exclusion. |
| Vector count | `vde.get_vector_count()` | Collection statistics. |
| Collection stats | `vde.get_stats()` | Storage monitoring. |
| Flush | `vde.flush()` | Persist to disk. |
## Conclusion
Recipe recommendation is a microcosm of every real-world search problem: semantic understanding for vague queries, hard constraints for safety (allergies), soft preferences for personalization (pantry ingredients), and evolving taste. This system illustrates how each Actian VectorAI feature maps to a concrete product need:
* `text()` filters with the word tokenizer turn ingredient lists into searchable keyword fields without a separate text search engine.
* `should()` and `min_should()` express "at least 2 of these ingredients should match" — exactly the pantry-matching behavior expected from a recommendation system.
* `any_of()` and `except_of()` handle multivalue fields like dietary tags and difficulty levels naturally.
* `search_batch()` makes meal planning practical by eliminating per-query network overhead.
* Server-side RRF fusion with `prefetch` blends current cravings with historical preferences without client-side ranking logic.
* `set_payload()` enables incremental preference learning without re-ingesting the entire recipe dataset.
The result is a recommendation system where the vector database handles not just similarity search, but also filtering, fusion, and state management.
## Next steps
Master the full Filter DSL with all field types.
Improve relevance with multistage pipelines.
Add recipe image search with named vectors.
Build persistent memory for AI agents.
# Multivector document intelligence with visual RAG
Source: https://docs.vectoraidb.actian.com/academy/articles/Multivector-Document-Intelligence-with-Visual-RAG
Build a multimodal document intelligence system that embeds PDF pages as images with CLIP, retrieves them via Actian VectorAI DB, and generates answers using GPT-4o vision.
This article walks through building a visual RAG pipeline that treats each PDF page as an image, embeds it with CLIP, stores it in Actian VectorAI DB, and uses GPT-4o vision to answer questions directly from retrieved page images — no text extraction required.
Traditional document RAG systems extract text from PDFs, chunk it, embed the chunks, and then retrieve them for LLM-based answer generation. This approach works well for text-heavy documents, but fails when critical information lives in:
* Charts and graphs—revenue trends, system architecture diagrams
* Tables—financial statements, comparison matrices
* Images—product photos, screenshots, annotated figures
* Complex layouts—multicolumn reports, slide decks, scanned documents
Text extraction loses this visual information entirely.
*Multivector Document Intelligence* takes a different approach inspired by the Contextualized Late Interaction over PaliGemma (ColPali) architecture. Instead of extracting text, it treats each PDF page as an image. Every page is embedded using a vision model (CLIP), stored as a vector, and retrieved based on visual and semantic similarity. A vision-language model (GPT-4o) then reads the retrieved page images to generate answers.
As a result, the system can answer questions about charts, tables, diagrams, and layouts, not just plain text.
You will build the full pipeline using:
* `clip-ViT-B-32` for page-level image embeddings (512-dimensional dense vectors).
* `actian-vectorai-client` for vector storage and semantic retrieval.
* `openai` GPT-4o vision API for answer generation from retrieved page images.
* `pdf2image` for converting PDF pages to images.
## Architecture overview
The diagram below shows the three phases of the pipeline. During ingestion, each PDF page is rendered to a high-resolution image, embedded with CLIP, saved to disk, and stored as a vector in Actian VectorAI DB. During semantic retrieval, a user's text query is encoded into the same CLIP vector space and compared against stored page vectors using cosine similarity, returning the top-K most relevant pages. Finally, during visual RAG answer generation, the retrieved page images are base64-encoded and sent to GPT-4o vision alongside the original query, producing a Markdown-formatted answer grounded in the actual page content.
```mermaid theme={null}
flowchart TB
subgraph ingestion [PDF Ingestion Pipeline]
PDF["PDF Document"]
Render["pdf2image - Render pages at 200 DPI"]
CLIP["CLIP ViT-B-32 - 512-dim Image Embedding"]
Store["Actian VectorAI DB - PointStruct: vector + payload"]
SaveImg["Save Page Images to Disk"]
PDF --> Render
Render --> CLIP
Render --> SaveImg
CLIP --> Store
end
subgraph retrieval [Semantic Retrieval]
Query["User Question - text"]
CLIPText["CLIP Text Encoder - same 512-dim space"]
Search["client.points.search - cosine similarity"]
TopK["Top-K Page Results - score, source_file, page_number"]
Query --> CLIPText
CLIPText --> Search
Store --> Search
Search --> TopK
end
subgraph rag [Visual RAG - Answer Generation]
Pages["Retrieved Page Images - base64 encoded"]
VLM["OpenAI GPT-4o Vision - reads page images + query"]
Answer["Generated Answer - Markdown formatted"]
TopK --> Pages
Pages --> VLM
Query --> VLM
VLM --> Answer
end
```
## Why visual document RAG
Standard text-based RAG pipelines lose critical information when documents contain visual content. This section explains where text extraction falls short and how the multivector approach addresses it.
### The problem with text extraction
Standard RAG pipelines use libraries like PyPDF2 or pdfplumber to extract text. But consider a financial report PDF:
* Page 3 has a revenue chart — text extraction produces nothing useful.
* Page 7 has a comparison table — extraction loses row/column alignment.
* Page 12 has an architecture diagram — extraction ignores it entirely.
### The multivector approach
Instead of extracting text, the pipeline:
1. Render each page as a high-resolution image (200 DPI).
2. Embed the image with CLIP — capturing visual layout, text, charts, and diagrams.
3. Store the embedding in Actian VectorAI DB with page metadata.
4. At query time, encode the text query with CLIP's text encoder (same vector space).
5. Retrieve the most visually relevant pages via cosine similarity.
6. Send the page images to GPT-4o vision to generate an answer.
With layout preserved as pixels, every element on the page — text, tables, charts, images — contributes to retrieval and answer generation.
## Environment setup
The pipeline depends on an Actian VectorAI DB instance, a Python environment, an OpenAI API key, and at least one PDF document. The sections below describe each requirement.
### Actian VectorAI DB instance
The pipeline stores and retrieves page-level CLIP vectors through a VectorAI DB server.
* A running Actian VectorAI DB server accessible over gRPC (default port `6574`)
* The server URL — for local development this is typically `http://localhost:6574`
* If you do not have an instance running yet, then follow the [installation guide](/docs/guides/index) to set one up before continuing
### Python environment
All dependencies are installed through `pip` into a standard Python environment.
* Python 3.10 or later
* `pip` for package installation
### OpenAI account
GPT-4o generates answers from the retrieved page images during the final RAG step.
* An OpenAI API key with access to `gpt-4o` (required for Step 6 answer generation)
* Set the key as an environment variable before running the tutorial (covered below)
### PDF documents
The tutorial references `annual_report.pdf` as a placeholder. Substitute any PDF you have available — the pipeline processes one or more files and renders every page as an image.
***
### Install Python packages
Install the required dependencies:
```bash theme={null}
pip install actian-vectorai-client sentence-transformers pillow pdf2image openai
```
System dependency for PDF rendering:
```bash theme={null}
# macOS
brew install poppler
# Ubuntu / Debian
sudo apt-get install -y poppler-utils
```
Set your OpenAI API key before running the answer-generation steps:
```bash theme={null}
export OPENAI_API_KEY="sk-..."
```
The key is read at runtime via `os.getenv("OPENAI_API_KEY")` in Step 6. Without it, the GPT-4o vision call will raise an authentication error.
These provide:
* `actian-vectorai-client` — Actian VectorAI Python SDK (async client, gRPC transport)
* `sentence-transformers` — CLIP ViT-B-32 for image and text embeddings
* `pillow` — Image processing
* `pdf2image` — Converts PDF pages to PIL images (requires poppler)
* `openai` — GPT-4o vision API for answer generation
***
## Implementation
The following steps build the pipeline end-to-end, from importing dependencies to running a full RAG query against ingested documents.
### Step 1: Import dependencies and configure
Load all libraries, set the VectorAI server address and collection name, and initialize the CLIP model so every subsequent step can reference them.
```python theme={null}
import base64
import hashlib
import io
import os
from openai import OpenAI
from pdf2image import convert_from_bytes
from PIL import Image
from sentence_transformers import SentenceTransformer
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
HnswConfigDiff,
PointStruct,
VectorParams,
)
from actian_vectorai.models.points import ScoredPoint
SERVER = "localhost:6574"
COLLECTION = "Multivector-DocIntel"
CLIP_DIM = 512
clip_model = SentenceTransformer("clip-ViT-B-32")
PAGE_IMAGES_DIR = "page_images"
os.makedirs(PAGE_IMAGES_DIR, exist_ok=True)
print(f"VectorAI Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"CLIP model loaded ({CLIP_DIM}-dim)")
```
#### Why this step matters
Every component is configured upfront. The key settings are listed below.
* `SERVER` — The Actian VectorAI gRPC endpoint (default `localhost:6574`).
* `COLLECTION` — The collection name for document page vectors.
* `clip_model` — The CLIP ViT-B-32 model that embeds both page images and text queries into the same 512-dimensional space.
* `PAGE_IMAGES_DIR` — The local directory where rendered page images are saved for later vision-language model (VLM) input.
#### Expected output
Running the configuration block prints the following to confirm each component loaded successfully:
```text theme={null}
VectorAI Server: localhost:6574
Collection: Multivector-DocIntel
CLIP model loaded (512-dim)
```
### Step 2: Define embedding helpers
CLIP encodes both images and text into the same 512-dim vector space, so cosine similarity can compare a text query against page-image vectors. In practice, related text and visuals tend to sit closer together than unrelated pairs, but ranking is not guaranteed: results depend on query wording, the document set, and how strongly each page matches the query in CLIP's representation.
```python theme={null}
def embed_image(image_bytes: bytes) -> list[float]:
"""Generate a 512-dim CLIP embedding for an image."""
image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
return clip_model.encode(image).tolist()
def embed_text_clip(text: str) -> list[float]:
"""Encode text into the same 512-dim CLIP space used for page images."""
return clip_model.encode(text).tolist()
def image_to_bytes(img: Image.Image, fmt: str = "PNG") -> bytes:
"""Convert a PIL image to bytes."""
buf = io.BytesIO()
img.save(buf, format=fmt)
return buf.getvalue()
def pil_image_to_base64(img: Image.Image) -> str:
"""Convert a PIL image to a base64-encoded string for the vision API."""
buf = io.BytesIO()
img.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode("utf-8")
```
#### Why this step matters
Each helper has a distinct role in the pipeline.
* `embed_image` — Converts a rendered page image into a CLIP vector for storage.
* `embed_text_clip` — Converts a user's text query into the same CLIP space for retrieval.
* `pil_image_to_base64` — Prepares page images for the GPT-4o vision API, which accepts base64-encoded images.
### Step 3: Initialize the VectorAI collection
Create the collection with 512-dim cosine distance and HNSW indexing.
`collections.get_or_create` takes the same `vectors_config` and optional `hnsw_config` arguments as `collections.create` in the [Create a collection](/docs/fundamentals/collections/create-collection-task) guide, including `hnsw_config=HnswConfigDiff(m=..., ef_construct=...)`. Match the `actian-vectorai-client` version you install to the docs or SDK release you are using.
```python theme={null}
import asyncio
async def ensure_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
await client.collections.get_or_create(
name=COLLECTION,
vectors_config=VectorParams(size=CLIP_DIM, distance=Distance.Cosine),
hnsw_config=HnswConfigDiff(m=32, ef_construct=256),
)
print(f"Collection '{COLLECTION}' ready.")
asyncio.run(ensure_collection())
```
#### Why this step matters
The collection stores one vector per document page, configured with the following settings.
* `Distance.Cosine` — appropriate for normalized CLIP embeddings.
* `HnswConfigDiff(m=32, ef_construct=256)` — high-recall HNSW settings for accurate retrieval.
#### Expected output
If the collection already exists, then `get_or_create` is a no-op and prints the same confirmation:
```text theme={null}
Collection 'Multivector-DocIntel' ready.
```
### Step 4: Ingest a PDF document
Ingestion runs end-to-end for each PDF: each page is rendered to an image, embedded with CLIP, saved to disk, and upserted into VectorAI.
```python theme={null}
import hashlib
def page_point_id(filename: str, page_number: int) -> int:
"""Generate a stable, deterministic integer ID from filename + page number.
Using a hash of filename + page_number means re-ingesting the same file
always produces the same IDs, so upsert is idempotent and there are no
collisions even if points are deleted or ingests run concurrently.
"""
key = f"{filename}:{page_number}"
return int(hashlib.md5(key.encode()).hexdigest()[:16], 16)
async def ingest_pdf(pdf_bytes: bytes, filename: str) -> int:
"""Convert PDF to page images, embed each with CLIP, store in VectorAI."""
await ensure_collection()
pages = convert_from_bytes(pdf_bytes, dpi=200)
if not pages:
return 0
async with AsyncVectorAIClient(url=SERVER) as client:
points = []
for page_idx, page_img in enumerate(pages):
page_number = page_idx + 1
page_img_rgb = page_img.convert("RGB")
img_bytes = image_to_bytes(page_img_rgb)
vector = embed_image(img_bytes)
image_filename = f"{filename}_page_{page_number}.png"
save_path = os.path.join(PAGE_IMAGES_DIR, image_filename)
page_img_rgb.save(save_path, "PNG")
payload = {
"source_file": filename,
"page_number": page_number,
"image_filename": image_filename,
}
points.append(
PointStruct(
id=page_point_id(filename, page_number),
vector=vector,
payload=payload,
)
)
await client.points.upsert(COLLECTION, points=points)
await client.vde.flush(COLLECTION)
total = await client.vde.get_vector_count(COLLECTION)
print(f"Ingested {len(points)} pages from '{filename}'. Total: {total}")
return len(points)
```
#### Why this step matters
The ingestion pipeline performs four operations per page.
1. **Render** — `convert_from_bytes(pdf_bytes, dpi=200)` produces high-resolution page images.
2. **Embed** — CLIP converts each page image into a 512-dim vector.
3. **Save** — The rendered image is saved to disk for later retrieval by the VLM.
4. **Store** — `client.points.upsert()` inserts the vector with metadata payload.
Each point's ID is derived from `page_point_id(filename, page_number)` — an MD5 hash of the filename and page number, truncated to a 64-bit integer. Using a deterministic hash means re-ingesting the same file always produces the same IDs, so the upsert is idempotent and there are no collisions from deletions, concurrent ingests, or re-runs. The `get_vector_count` call is no longer needed for ID generation.
The payload stores `source_file`, `page_number`, and `image_filename` so retrieved results can be traced back to their source.
#### Example usage
The following snippet reads a PDF from disk and passes it to `ingest_pdf`, which renders each page, embeds it, saves the image, and upserts the vector into the collection.
```python theme={null}
with open("annual_report.pdf", "rb") as f:
pdf_bytes = f.read()
pages_ingested = asyncio.run(ingest_pdf(pdf_bytes, "annual_report.pdf"))
```
#### Expected output
The output confirms how many pages were ingested and the running total in the collection:
```text theme={null}
Ingested 24 pages from 'annual_report.pdf'. Total: 24
```
### Step 5: Semantic search for document pages
Search for pages that are visually and semantically similar to a text query.
Vector similarity search uses `AsyncVectorAIClient.points.search` with the query embedding as `vector`, a result cap as `limit`, and `with_payload=True` to return metadata. That matches the pattern and parameter table in [Similarity search basics](/academy/tutorials/similarity-search) (Step 4).
```python theme={null}
async def search_pages(query_text: str, top_k: int = 5) -> list[dict]:
"""Semantic search for document pages similar to a text query."""
query_vector = embed_text_clip(query_text)
async with AsyncVectorAIClient(url=SERVER) as client:
raw: list[ScoredPoint] = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
) or []
return [
{
"id": r.id,
"score": float(r.score or 0.0),
"source_file": (r.payload or {}).get("source_file", ""),
"page_number": (r.payload or {}).get("page_number", 0),
"image_filename": (r.payload or {}).get("image_filename", ""),
}
for r in raw
]
```
#### Why this step matters
The text query is encoded using CLIP's text encoder into the same 512-dim space as the page images. Actian VectorAI's `points.search` ranks pages by cosine similarity to the query vector—the mathematically nearest neighbors in that space, which approximate relevance but are not a formal guarantee of correctness.
Because CLIP was trained on image-text pairs, a query like "quarterly revenue breakdown" will often place relevant pages—those with charts or tables that match the intent—higher in the similarity list than unrelated pages. Treat top-K results as a best-effort shortlist: you may still need to tune `top_k`, rephrase queries, or add filters for production accuracy.
#### Example usage
The following snippet runs a semantic search for pages related to quarterly revenue and prints each result's page number, source file, and similarity score.
```python theme={null}
results = asyncio.run(search_pages("What is the quarterly revenue?", top_k=3))
for r in results:
print(f" page {r['page_number']} of {r['source_file']} score={r['score']:.4f}")
```
#### Expected output
Each result shows the page number, source file, and cosine similarity score:
```text theme={null}
page 5 of annual_report.pdf score=0.2834
page 3 of annual_report.pdf score=0.2651
page 12 of annual_report.pdf score=0.2403
```
### Step 6: Generate answers with GPT-4o vision
The retrieved page images are sent to OpenAI's GPT-4o vision model along with the user's question.
```python theme={null}
def generate_answer(query_text: str, image_filenames: list[str],
model: str = "gpt-4o") -> str:
"""Send retrieved page images + query to OpenAI GPT-4o vision API."""
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": (
"You are a helpful assistant that answers questions based on "
"the provided document page images. Read the images carefully "
"and provide accurate answers based only on what is visible in "
"the images. If the information is not present in the images, "
"say so rather than guessing. Answer in Markdown and highlight "
"the most important parts."
),
},
]
user_content = [{"type": "text", "text": query_text}]
for img_filename in image_filenames[:10]:
img_path = os.path.join(PAGE_IMAGES_DIR, img_filename)
if not os.path.isfile(img_path):
continue
img = Image.open(img_path).convert("RGB")
b64 = pil_image_to_base64(img)
user_content.append({
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{b64}"},
})
messages.append({"role": "user", "content": user_content})
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=1000,
)
return response.choices[0].message.content
```
#### Why this step matters
At answer time, the workflow becomes a true visual RAG pipeline. GPT-4o can perform the following tasks.
* Read text from page images (no separate OCR pipeline required)
* Interpret charts and graphs
* Parse tables and extract specific values
* Interpret diagrams and architecture drawings
The function sends up to 10 page images as base64-encoded PNGs to the vision API. The system prompt instructs the model to answer only from what is visible in the images and to state when information is not present, rather than inferring beyond the provided content.
### Step 7: End-to-end RAG pipeline
Connect search and answer generation into a single function.
```python theme={null}
async def rag_query(query_text: str, top_k: int = 3) -> dict:
"""Full RAG pipeline: search similar pages, then generate answer via VLM."""
results = await search_pages(query_text, top_k=top_k)
if not results:
return {
"answer": "No relevant document pages found. Please upload documents first.",
"sources": results,
"query": query_text,
}
image_filenames = [r["image_filename"] for r in results if r.get("image_filename")]
answer = generate_answer(query_text, image_filenames)
return {
"answer": answer,
"sources": results,
"query": query_text,
}
```
#### Why this step matters
The RAG pipeline has three stages.
1. **Retrieve** — Find the top-K most relevant pages via CLIP similarity search on Actian VectorAI.
2. **Load** — Get the page image filenames from the search results payload.
3. **Generate** — Send images + query to GPT-4o vision for answer generation.
The response includes both the generated answer and the source pages, so users can verify the answer against the original document.
#### Example usage
The following snippet runs a complete RAG query, then prints a truncated preview of the generated answer alongside the source pages and their similarity scores.
```python theme={null}
result = asyncio.run(rag_query("What were the key financial highlights?"))
print(f"Answer: {result['answer'][:200]}...")
print(f"Sources: {len(result['sources'])} pages")
for s in result["sources"]:
print(f" - page {s['page_number']} of {s['source_file']} (score={s['score']:.4f})")
```
#### Expected output
The `answer` field contains raw Markdown text returned by GPT-4o — it is not rendered here, so `**Revenue**` and list hyphens appear as literal characters rather than formatted output.
```text theme={null}
Answer: Based on the financial report pages, the key highlights include:
- **Revenue** grew 15% year-over-year to $2.4B
- Operating margin improved to 22.3%
- Free cash flow increased by $180M...
Sources: 3 pages
- page 5 of annual_report.pdf (score=0.2834)
- page 3 of annual_report.pdf (score=0.2651)
- page 12 of annual_report.pdf (score=0.2403)
```
### Step 8: Collection administration
Use the following operations to inspect the collection, list ingested documents, flush data to disk, or delete the collection entirely.
```python theme={null}
async def admin_operations():
async with AsyncVectorAIClient(url=SERVER) as client:
count = await client.vde.get_vector_count(COLLECTION)
print(f"Total indexed pages: {count}")
# List all ingested documents by scrolling through all points
doc_names = set()
offset = None
while True:
result = await client.points.scroll(
COLLECTION,
limit=500,
offset=offset,
with_payload=True,
with_vectors=False,
)
batch, next_offset = result
for p in (batch or []):
source = (p.payload or {}).get("source_file", "")
if source:
doc_names.add(source)
if next_offset is None:
break
offset = next_offset
print(f"Ingested documents: {sorted(doc_names)}")
# Flush to disk
await client.vde.flush(COLLECTION)
print("Collection flushed.")
# To delete everything:
# await client.collections.delete(COLLECTION)
asyncio.run(admin_operations())
```
#### Expected output
Running the admin operations prints the total vector count, a sorted list of ingested document names, and a flush confirmation:
```text theme={null}
Total indexed pages: 24
Ingested documents: ['annual_report.pdf']
Collection flushed.
```
***
## How the visual RAG pipeline differs from text RAG
The table below compares the two approaches across each stage of the pipeline. It covers everything from input processing to answer generation, helping you decide which approach fits your documents.
| Aspect | Traditional text RAG | Visual document RAG |
| ---------------------- | ---------------------------------------------- | ----------------------------------------- |
| **Input processing** | Extract text, chunk into passages | Render pages as images |
| **Embedding model** | Text embedder (such as text-embedding-3-small) | Vision model (CLIP ViT-B-32) |
| **What gets embedded** | Text chunks (500-1000 tokens) | Full page images (200 DPI) |
| **Charts and tables** | Lost during extraction | Preserved as visual content |
| **Retrieval unit** | Text chunk | Document page |
| **Answer generation** | LLM reads retrieved text | VLM reads retrieved page images |
| **OCR dependency** | Requires text extraction | No OCR needed — VLM reads images directly |
***
## Actian VectorAI features used
The following table summarises the SDK methods used in this article, mapping each feature to its API call and its role in the pipeline.
| Feature | API | Purpose |
| ----------------------- | ------------------------------------ | ------------------------------------------------ |
| **Collection creation** | `client.collections.get_or_create()` | Create 512-dim cosine vector space with HNSW |
| **Batch point upsert** | `client.points.upsert()` | Store CLIP page vectors with metadata payload |
| **Semantic search** | `client.points.search()` | Find visually similar pages by cosine similarity |
| **Point scroll** | `client.points.scroll()` | Page through all points for document listing |
| **Vector count** | `client.vde.get_vector_count()` | Track total indexed pages |
| **Flush** | `client.vde.flush()` | Persist vectors to disk after ingestion |
| **Delete collection** | `client.collections.delete()` | Clean up all data |
***
## The ColPali inspiration
This system is inspired by the ColPali architecture, which demonstrated that:
1. Treating document pages as images avoids lossy text extraction
2. Vision encoders capture layout, typography, and visual elements
3. Late interaction between query tokens and page patch embeddings improves retrieval
The implementation simplifies ColPali by using:
* CLIP instead of PaliGemma for embedding — simpler to deploy and widely available
* Single vector per page instead of multivector patch embeddings — compatible with standard vector databases without specialised infrastructure
* GPT-4o vision instead of a specialised reader model for answer generation — no custom training required
This trade-off produces a practical system that works with Actian VectorAI DB while preserving the core insight: visual document understanding beats text extraction for rich documents.
***
## When to use visual document RAG
Visual document RAG is not the right fit for every use case. The sections below outline where it excels and where text-based RAG remains the better option.
### Best suited for
This approach works best for documents where critical information is encoded visually rather than as plain text.
* Financial reports with charts and tables
* Slide decks and presentations
* Technical manuals with diagrams
* Scanned documents and forms
* Multi-column layouts and complex formatting
### Consider text RAG instead when
Text RAG is the better choice when documents are text-dominant and token-level precision matters more than visual fidelity.
* Documents are purely text-based (novels, articles)
* Token-level precision matters more than page-level retrieval
* You need to process thousands of pages per query (VLM calls add cost at scale)
***
## Next steps
Now that you have built a full visual RAG pipeline, explore these topics to extend and improve your system:
Combine vector similarity with structured constraints
Learn the core retrieval workflow
Add `must`, `should`, and `must_not` conditions
Measure and improve search result accuracy
# Next-gen product discovery with multimodal AI
Source: https://docs.vectoraidb.actian.com/academy/articles/Next-Gen-Product-Discovery-with-Multimodal-AI
Build a multimodal hybrid search system combining CLIP dense embeddings and BM25 sparse scoring for semantic and keyword product retrieval using Actian VectorAI DB.
This tutorial builds a multimodal hybrid search system that combines dense semantic embeddings with sparse keyword scoring to retrieve product images by both meaning and exact terms. By the end, you have a working system that handles queries like "dark blue french connection jeans for men" — matching on brand, color, and semantic similarity at the same time.
Modern search systems struggle when queries combine semantic meaning and exact keywords. A user searching for "dark blue french connection jeans for men" expects results that satisfy all of the following:
* The exact terms: "jeans", "french connection", "blue".
* The semantic meaning of the query.
* Visual similarity with product images.
Traditional keyword search cannot understand meaning, while pure vector search may ignore exact tokens like brand names or product attributes. The solution is hybrid search — combining sparse keyword retrieval with dense semantic embeddings, fused using Reciprocal Rank Fusion (RRF) or Distribution-Based Score Fusion (DBSF).
This tutorial builds the system using the following components:
* CLIP ViT-B-32 embeddings (512-dim) for semantic understanding of images and text.
* BM25 sparse scoring for keyword relevance.
* Actian VectorAI DB for scalable vector storage and retrieval.
* Actian VectorAI SDK fusion algorithms (RRF and DBSF) for combining dense and sparse results.
By the end, the system retrieves product images using both semantic similarity and keyword relevance.
## Prerequisites
Before starting, make sure you have the following in place:
* A running Actian VectorAI instance reachable at `localhost:6574`.
* Python 3.10 or later.
* A set of product images and associated metadata (product name, category, color, gender, and so on). This tutorial uses a fashion product dataset as its example, but the same pipeline applies to any product catalog.
## Architecture overview
The system is structured around two pipelines. The product registration pipeline takes a product image and its metadata, generates a 512-dimensional CLIP embedding from the image, concatenates the metadata into a searchable text string for BM25, and stores both in Actian VectorAI DB as a single point. The hybrid search pipeline takes a user query — text or image — runs a dense CLIP search server-side and a sparse BM25 search client-side, then fuses the results using RRF or DBSF to produce a single ranked output.
The diagram below shows how these two pipelines connect, from product registration through to final ranked results:
```mermaid theme={null}
flowchart TB
subgraph registration [Product Registration]
Image["Product Image"]
Meta["Product Metadata — Name, category, color, etc."]
CLIP["CLIP ViT-B-32 — 512-dim dense embedding"]
BM25Text["Build product text — Concatenate metadata fields"]
Image --> CLIP
Meta --> BM25Text
CLIP --> Upsert["Actian VectorAI DB — PointStruct: vector + payload"]
BM25Text --> Upsert
end
subgraph search [Hybrid Search Query]
Query["User Query — Text or image"]
DenseSearch["Dense Search — CLIP embedding, server-side"]
SparseSearch["Sparse Search — BM25 scoring, client-side"]
Query --> DenseSearch
Query --> SparseSearch
end
subgraph fusion [Result Fusion]
DenseResults["Dense Results — ScoredPoint list"]
SparseResults["Sparse Results — ScoredPoint list"]
DenseSearch --> DenseResults
SparseSearch --> SparseResults
RRF["Reciprocal Rank Fusion / DBSF"]
DenseResults --> RRF
SparseResults --> RRF
RRF --> FinalResults["Ranked Results"]
end
Upsert --> DenseSearch
Upsert --> SparseSearch
```
## Why hybrid search matters
Real-world queries usually contain two types of signals — keyword signals and semantic signals. The sections below explain each type and describe how hybrid search combines them.
### Keyword signals
Sparse search targets exact tokens such as the brand name, color, and product type. For the query below, BM25 matches against three distinct tokens:
```text theme={null}
french connection blue jeans
```
The tokens BM25 matches against are:
* French connection (brand).
* Blue (color).
* Jeans (product type).
Sparse retrieval methods like BM25 work well for these exact-match cases.
### Semantic signals
Dense embeddings from CLIP capture semantic meaning, allowing results to surface even when no exact tokens match. For the query below, the product description may not contain these exact words, but the system still returns similar items:
```text theme={null}
casual dark denim for men
```
CLIP maps this query into the same vector space as product image embeddings, so semantically related products surface regardless of exact wording.
### How hybrid search combines both
Instead of choosing one method, this approach combines both using fusion. The Actian VectorAI SDK provides two built-in fusion algorithms:
* Reciprocal Rank Fusion (RRF) — Rank-based merging that ignores raw scores. Use this when dense and sparse scores are on different scales.
* Distribution-Based Score Fusion (DBSF) — Normalizes and averages scores using mean and standard deviation. Use this when you want score-aware blending.
The `alpha` parameter controls the weight balance in RRF. Higher values favor dense results; lower values favor sparse results:
```text theme={null}
alpha = 1.0 → 100% dense (visual similarity)
alpha = 0.5 → equal blend
alpha = 0.0 → 100% sparse (keyword matching)
```
## Environment setup
The following command installs the three packages required for image processing, CLIP embeddings, and the Actian VectorAI SDK. Run this before proceeding with the implementation:
```bash theme={null}
pip install actian-vectorai-client sentence-transformers pillow
```
***
## Implementation
The following steps build the complete multimodal hybrid search system, from loading the CLIP model and initializing the collection through to running dense, sparse, and fused queries.
### Step 1: Import dependencies and configure
The block below imports all required libraries, sets the server address and collection name, and loads the CLIP model. Running it prints the configured server address, collection name, and CLIP model dimensionality, confirming everything is ready before any collections or vectors are created.
```python theme={null}
# Standard library imports for I/O, math, unique IDs, and token counting
import io
import math
import uuid
from collections import Counter
# PIL for image decoding; SentenceTransformer loads the CLIP model
from PIL import Image
from sentence_transformers import SentenceTransformer
# Actian VectorAI SDK: async client, vector config, point types, and fusion functions
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
PointStruct,
VectorParams,
reciprocal_rank_fusion,
distribution_based_score_fusion,
)
from actian_vectorai.models.collections import HnswConfigDiff
from actian_vectorai.models.points import ScoredPoint
# Server address, collection name, and vector dimensionality used throughout this tutorial
SERVER = "localhost:6574"
COLLECTION = "NextGen-Purchase"
DENSE_DIM = 512 # CLIP ViT-B-32 outputs 512-dimensional vectors
# Load CLIP once at module level — all embedding calls reuse this instance
clip_model = SentenceTransformer("clip-ViT-B-32")
print(f"VectorAI Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"CLIP model loaded ({DENSE_DIM}-dim)")
```
The CLIP model is loaded once at module level so that every subsequent call to `embed_image` or `embed_text_clip` reuses the same instance without reloading weights. On the first run, `SentenceTransformer("clip-ViT-B-32")` downloads the model weights before returning. Running this block confirms the configured server address, collection name, and CLIP model dimensionality, verifying that everything is ready before any collections or vectors are created.
#### Expected output
Running this block prints the server address, collection name, and the CLIP model dimensionality, confirming the configuration is valid before any collections or vectors are created.
```text theme={null}
VectorAI Server: localhost:6574
Collection: NextGen-Purchase
CLIP model loaded (512-dim)
```
### Step 2: Define embedding helpers
CLIP maps images and text into the same 512-dimensional vector space. This shared space is what enables cross-modal search — a text query can retrieve products whose embeddings were generated from images, because both live in the same space. The two functions below handle each input type separately but produce vectors that are directly comparable.
```python theme={null}
def embed_image(image_bytes: bytes) -> list[float]:
"""Return a 512-dim CLIP vector for a raw image."""
# Decode bytes to a PIL image and convert to RGB before encoding
image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
return clip_model.encode(image).tolist()
def embed_text_clip(text: str) -> list[float]:
"""Return a 512-dim CLIP vector for a text string."""
# CLIP's shared embedding space allows direct comparison with image vectors
return clip_model.encode(text).tolist()
```
Because both functions use the same CLIP model and vector space, an image of blue jeans and the text "blue jeans" produce nearby vectors, making text-to-image retrieval possible without storing images in the database.
### Step 3: Build the BM25 text representation
BM25 scoring operates on plain text rather than vectors. The function below concatenates all product metadata fields into a single lowercase string, which is stored in the point payload at registration time and scored against the query at search time.
```python theme={null}
def build_product_text(metadata: dict) -> str:
"""Concatenate metadata fields into a single lowercase string for BM25 scoring."""
# Pull all descriptive fields a user might search by keyword
fields = [
metadata.get("product_name", ""),
metadata.get("gender", ""),
metadata.get("category", ""),
metadata.get("sub_category", ""),
metadata.get("article_type", ""),
metadata.get("color", ""),
metadata.get("season", ""),
metadata.get("usage", ""),
]
# Drop empty strings, join with spaces, and lowercase for consistent tokenization
return " ".join(f for f in fields if f).strip().lower()
```
The function pulls eight metadata fields — product name, gender, category, sub-category, article type, color, season, and usage — drops any empty values, joins them with spaces, and lowercases the result. This produces a single plain-text document per product that BM25 can tokenize and score at query time.
#### Expected output
For a denim product with complete metadata, the concatenated string looks like this. BM25 uses this string to match query tokens such as "french connection", "jeans", and "blue" at search time.
```text theme={null}
dark blue french connection jeans men apparel bottomwear jeans blue winter casual
```
### Step 4: Implement client-side BM25 scoring
The BM25 function below takes a list of query tokens, the text of a single document, and corpus-level statistics (average document length, per-token document frequency, and total document count). It returns a float relevance score for that document. Scores of zero indicate no token overlap between the query and the document.
```python theme={null}
def bm25_score(
query_tokens: list[str],
doc_text: str,
avg_dl: float,
doc_freq: dict[str, int],
total_docs: int,
k1: float = 1.5,
b: float = 0.75,
) -> float:
"""Return the BM25 relevance score for a single document against a set of query tokens."""
doc_tokens = doc_text.lower().split()
dl = len(doc_tokens)
tf_map = Counter(doc_tokens)
score = 0.0
for qt in query_tokens:
tf = tf_map.get(qt, 0)
if tf == 0:
continue # Token not present in this document — contributes nothing to score
df = doc_freq.get(qt, 0)
# IDF boosts tokens that are rare across the corpus
idf = math.log((total_docs - df + 0.5) / (df + 0.5) + 1.0)
numerator = tf * (k1 + 1)
# Length normalization controlled by b penalizes unusually long documents
denominator = tf + k1 * (1 - b + b * (dl / max(avg_dl, 1)))
score += idf * (numerator / denominator)
return score
```
The formula combines three signals to rank documents. Term frequency (TF) measures how often a query token appears in the document. Inverse document frequency (IDF) boosts tokens that are rare across the entire corpus, such as a brand name that appears in only a handful of products. Document length normalization prevents longer documents from receiving unfairly high scores simply because they contain more words. The defaults `k1=1.5` and `b=0.75` are standard BM25 values that work well across most text corpora.
### Step 5: Initialize the VectorAI collection
The function below creates the `NextGen-Purchase` collection if it does not already exist. `get_or_create` is idempotent, so calling it on every startup is safe — it returns immediately if the collection is present. Running the block prints a confirmation that the collection is ready to accept vectors.
```python theme={null}
import asyncio
async def ensure_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
# get_or_create returns immediately if the collection already exists
await client.collections.get_or_create(
name=COLLECTION,
vectors_config=VectorParams(size=DENSE_DIM, distance=Distance.Cosine),
# m=32 sets the number of HNSW graph connections per node
# ef_construct=256 controls index build quality — higher means better recall
hnsw_config=HnswConfigDiff(m=32, ef_construct=256),
)
print(f"Collection '{COLLECTION}' ready.")
asyncio.run(ensure_collection())
```
The block creates the `NextGen-Purchase` collection using 512-dimensional CLIP vectors with cosine distance. The HNSW parameters `m=32` and `ef_construct=256` balance recall quality against indexing speed. Because `get_or_create` is idempotent, this call is safe to repeat on every startup — it returns immediately when the collection already exists.
#### Expected output
Running this block prints a confirmation that the collection is ready. If the collection already exists, the message is identical — `get_or_create` does not raise an error on repeat calls.
```text theme={null}
Collection 'NextGen-Purchase' ready.
```
### Step 6: Register a product
The function below registers a single product. It generates a 512-dim CLIP embedding from the product image, builds the BM25 text string from the metadata, and upserts both as a single point in the collection. After registration, it flushes the collection to disk and prints the product name alongside the updated total vector count.
```python theme={null}
async def register_product(image_bytes: bytes, metadata: dict) -> str:
"""Register a product as a vector point with CLIP embedding and BM25 text payload."""
# Encode the product image into a 512-dim CLIP vector
dense_vector = embed_image(image_bytes)
# Build a keyword-searchable string from all metadata fields
product_text = build_product_text(metadata)
# Generate a UUID for the payload — not used as the point ID in this implementation
point_id = str(uuid.uuid4())
payload = {
"point_id": point_id,
"product_text": product_text, # Scored by BM25 at query time
**metadata, # All metadata fields returned with search results
}
async with AsyncVectorAIClient(url=SERVER) as client:
# Use current vector count as the integer point ID
existing = await client.vde.get_vector_count(COLLECTION)
point = PointStruct(
id=existing,
vector=dense_vector,
payload=payload,
)
await client.points.upsert(COLLECTION, points=[point])
# flush() persists buffered writes to disk before returning
await client.vde.flush(COLLECTION)
total = await client.vde.get_vector_count(COLLECTION)
print(f"Registered '{metadata.get('product_name')}'. Total: {total}")
return point_id
```
Each point stored in VectorAI DB contains three things:
* Vector — A 512-dim CLIP image embedding used for dense similarity search.
* payload.product\_text — The concatenated metadata string scored by BM25 at query time.
* Additional payload fields — All original metadata fields, returned alongside each search result.
The `vde.flush()` call persists any buffered writes to disk before the function returns, ensuring the point is available for search immediately after registration.
### Step 7: Dense search (server-side)
The two functions below perform dense similarity searches using CLIP embeddings. Both encode the query into a 512-dim vector and send it to the Actian VectorAI server, which runs HNSW approximate nearest-neighbor search and returns a ranked list of `ScoredPoint` objects. The only difference between the two is the query input type.
```python theme={null}
async def search_by_image(image_bytes: bytes, top_k: int = 10):
"""Search by visual similarity using a CLIP image embedding."""
# Encode the query image into a CLIP vector for server-side search
dense_vector = embed_image(image_bytes)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=dense_vector,
limit=top_k,
with_payload=True, # Include metadata in each returned result
)
return results or []
async def search_by_text_dense(query: str, top_k: int = 10):
"""Search by semantic meaning using a CLIP text embedding."""
# CLIP maps this text into the same vector space as product image embeddings
dense_vector = embed_text_clip(query)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=dense_vector,
limit=top_k,
with_payload=True,
)
return results or []
```
Searching by image finds products that are visually similar to the query image. Searching by text finds products whose image embeddings are close to the text query in CLIP's shared vector space. Both return a list of `ScoredPoint` objects sorted by descending cosine similarity.
### Step 8: Sparse BM25 search (client-side)
Unlike dense search, BM25 runs entirely on the client. The function below fetches all points from the collection in batches of 500, computes BM25 scores locally by comparing query tokens against the `product_text` payload field of each point, and returns the top-K results sorted by descending score.
```python theme={null}
async def bm25_search(query: str, top_k: int = 10) -> list[ScoredPoint]:
"""Score all collection points against the query using BM25, return top-K results."""
async with AsyncVectorAIClient(url=SERVER) as client:
total = await client.vde.get_vector_count(COLLECTION)
if total == 0:
return []
# Retrieve all points in batches to avoid large single requests
batch_size = 500
all_points = []
for start in range(0, total, batch_size):
end = min(start + batch_size, total)
batch = await client.points.get(
COLLECTION,
ids=list(range(start, end)),
with_payload=True, # product_text payload field is required for scoring
)
all_points.extend(batch or [])
if not all_points:
return []
query_tokens = query.lower().split()
texts = [p.payload.get("product_text", "") for p in all_points]
total_docs = len(texts)
avg_dl = sum(len(t.split()) for t in texts) / max(total_docs, 1)
# Compute how many documents each token appears in — used for IDF calculation
doc_freq: dict[str, int] = {}
for t in texts:
for tok in set(t.split()):
doc_freq[tok] = doc_freq.get(tok, 0) + 1
scored = []
for p, text in zip(all_points, texts):
s = bm25_score(query_tokens, text, avg_dl, doc_freq, total_docs)
if s > 0:
scored.append(ScoredPoint(
id=p.id,
version=getattr(p, "version", 0),
score=s,
payload=p.payload,
))
# Sort by score descending and return only the top-K results
scored.sort(key=lambda x: x.score or 0, reverse=True)
return scored[:top_k]
```
BM25 catches keyword matches that dense embeddings miss. For a query like "french connection jeans", BM25 strongly scores products that contain the exact brand name in their metadata, even when the CLIP embedding does not distinguish brand names from other descriptive terms.
### Step 9: Hybrid search with fusion
The function below runs both dense and sparse searches in sequence and merges the results using either RRF or DBSF. It fetches `top_k * 5` candidates (up to 50) from each search before fusing, giving the fusion algorithm a broad enough input to rerank effectively before returning the final `top_k` results.
```python theme={null}
async def hybrid_search(
query_text: str | None = None,
image_bytes: bytes | None = None,
alpha: float = 0.5,
top_k: int = 10,
fusion_method: str = "rrf",
):
"""
Run dense and sparse search, then fuse the results.
alpha controls RRF weight balance:
alpha=1.0 — Full weight on dense (visual similarity)
alpha=0.0 — Full weight on sparse (keyword/BM25)
Note: alpha applies to RRF only. DBSF uses score normalization
and ignores the weights parameter in this implementation.
"""
if not query_text and not image_bytes:
return []
# Fetch a wider candidate pool than top_k so fusion has enough items to rerank
fetch_k = min(top_k * 5, 50)
dense_results = []
sparse_results = []
# Stage 1 — Dense search: encode the query as a CLIP vector
if image_bytes:
dense_vector = embed_image(image_bytes)
elif query_text:
dense_vector = embed_text_clip(query_text)
else:
dense_vector = None
if dense_vector is not None:
async with AsyncVectorAIClient(url=SERVER) as client:
dense_results = await client.points.search(
COLLECTION,
vector=dense_vector,
limit=fetch_k,
with_payload=True,
) or []
# Stage 2 — Sparse BM25 search: score all points client-side
if query_text:
sparse_results = await bm25_search(query_text, top_k=fetch_k)
# If only one source produced results, return those directly without fusion
if not dense_results and not sparse_results:
return []
if not sparse_results:
return dense_results[:top_k]
if not dense_results:
return sparse_results[:top_k]
# Stage 3 — Fusion: merge both result lists into a single ranked output
weights = [alpha, 1.0 - alpha]
if fusion_method == "dbsf":
# DBSF normalizes scores by distribution — alpha/weights are not used
fused = distribution_based_score_fusion(
[dense_results, sparse_results], limit=top_k
)
else:
# RRF merges by rank position; weights shift influence toward dense or sparse
fused = reciprocal_rank_fusion(
[dense_results, sparse_results], limit=top_k, weights=weights
)
return fused
```
This function runs three stages in sequence. First, it encodes the query and runs a dense CLIP similarity search server-side against Actian VectorAI DB. Second, it runs a client-side BM25 search over the `product_text` payload field. Third, it passes both result lists to either `reciprocal_rank_fusion` or `distribution_based_score_fusion` from the Actian VectorAI SDK.
The `alpha` parameter shifts the weight between the two sources in RRF. The table below shows how different values change the balance:
| Alpha | Behavior |
| ----- | ----------------------------------------------------------- |
| `1.0` | 100% dense — Pure visual/semantic similarity. |
| `0.7` | 70% dense, 30% sparse — Mostly semantic with keyword boost. |
| `0.5` | Equal blend — Balanced hybrid. |
| `0.3` | 30% dense, 70% sparse — Mostly keyword with semantic boost. |
| `0.0` | 100% sparse — Pure BM25 keyword matching. |
### Step 10: Run the end-to-end hybrid search
The block below runs the same query through all four search modes — dense-only, sparse-only, RRF-fused, and DBSF-fused — and prints a ranked result list for each. Running it lets you compare how each approach ranks "Dark Blue French Connection Jeans" against other products in the collection.
```python theme={null}
async def run_hybrid_demo():
query = "dark blue french connection jeans for men"
# Dense-only: server-side CLIP similarity search
dense_results = await search_by_text_dense(query, top_k=5)
print("=== Dense Results (CLIP) ===")
for r in dense_results:
print(f" score={r.score:.4f} product={r.payload.get('product_name', '')}")
# Sparse-only: client-side BM25 keyword scoring
sparse_results = await bm25_search(query, top_k=5)
print("\n=== Sparse Results (BM25) ===")
for r in sparse_results:
print(f" score={r.score:.4f} product={r.payload.get('product_name', '')}")
# Hybrid RRF: rank-based fusion with equal weight between dense and sparse
hybrid_results = await hybrid_search(
query_text=query, alpha=0.5, top_k=5, fusion_method="rrf"
)
print("\n=== Hybrid Results (RRF, alpha=0.5) ===")
for r in hybrid_results:
print(f" score={r.score:.4f} product={r.payload.get('product_name', '')}")
# Hybrid DBSF: score-normalized fusion (alpha is ignored for DBSF)
hybrid_dbsf = await hybrid_search(
query_text=query, alpha=0.7, top_k=5, fusion_method="dbsf"
)
print("\n=== Hybrid Results (DBSF) ===")
for r in hybrid_dbsf:
print(f" score={r.score:.4f} product={r.payload.get('product_name', '')}")
asyncio.run(run_hybrid_demo())
```
The block runs the query `"dark blue french connection jeans for men"` through all four search modes in sequence. Dense-only search encodes the query as a CLIP text vector and runs server-side cosine similarity against stored image embeddings. Sparse-only search scores every point's `product_text` payload field using client-side BM25, rewarding exact token matches for terms like "french connection" and "jeans". The RRF-fused mode combines both ranked lists with equal weight (`alpha=0.5`), merging by rank position regardless of raw score scale. The DBSF-fused mode normalizes scores by their distribution before averaging. Hybrid search ranks "Dark Blue French Connection Jeans" highest across both fusion methods because it satisfies both the CLIP semantic similarity and the BM25 exact-keyword match.
#### Expected output
Exact scores depend on the dataset and the products registered. The values below are illustrative. Notice that BM25 scores are on a different scale from CLIP cosine similarity scores — RRF handles this by merging on rank position rather than raw values.
```text theme={null}
=== Dense Results (CLIP) ===
score=0.8521 product=Dark Blue French Connection Jeans
score=0.7834 product=Slim Fit Blue Denim
score=0.7102 product=Navy Casual Trousers
=== Sparse Results (BM25) ===
score=4.2310 product=Dark Blue French Connection Jeans
score=3.1205 product=French Connection Formal Shirt
score=2.8901 product=Blue Denim Jeans Men
=== Hybrid Results (RRF, alpha=0.5) ===
score=0.0323 product=Dark Blue French Connection Jeans
score=0.0294 product=Blue Denim Jeans Men
score=0.0256 product=Slim Fit Blue Denim
=== Hybrid Results (DBSF) ===
score=0.8100 product=Dark Blue French Connection Jeans
score=0.6543 product=Slim Fit Blue Denim
score=0.5982 product=Blue Denim Jeans Men
```
### Step 11: Collection administration
The block below demonstrates three VDE operations: retrieving the current vector count, flushing buffered writes to disk, and deleting the collection (shown as a comment). Running it prints the total number of stored product vectors and a confirmation that the flush completed successfully.
```python theme={null}
async def admin_operations():
async with AsyncVectorAIClient(url=SERVER) as client:
# get_vector_count returns the total number of indexed points
count = await client.vde.get_vector_count(COLLECTION)
print(f"Total products in collection: {count}")
# flush() persists any buffered in-memory writes to disk
await client.vde.flush(COLLECTION)
print("Collection flushed to disk.")
# Uncomment to permanently delete the collection and all its vectors
# await client.collections.delete(COLLECTION)
asyncio.run(admin_operations())
```
#### Expected output
Running this block prints the current vector count followed by a confirmation that the flush completed. The count reflects how many product points have been registered in the collection.
```text theme={null}
Total products in collection: 42
Collection flushed to disk.
```
***
## How images are returned after retrieval
Vector databases store embeddings, not raw image files. A common question when building retrieval systems is how to get images back from search results.
The answer is payload metadata. When registering a product, store the `image_filename` in the payload alongside the embedding. The payload dictionary below shows what a complete point entry looks like:
```python theme={null}
payload = {
"point_id": point_id,
"image_filename": "abc123.jpg", # Path or key used to load the image from storage
"product_text": "dark blue jeans men apparel",
"product_name": "Dark Blue Jeans",
...
}
```
During retrieval, Actian VectorAI returns the full payload including `image_filename`. The application uses that value to load the image from disk or object storage and render it to the user. The vector database stores representations, not the raw images.
***
## Fusion methods compared
The hybrid search pipeline produces two separate ranked lists — one from dense CLIP search and one from sparse BM25 scoring. A fusion algorithm merges these lists into a single ranking. The table below compares the two built-in options provided by the Actian VectorAI SDK:
| Method | How it works | When to use |
| -------------------------------------- | ------------------------------------------------------------------- | ----------------------------------------------------- |
| RRF (Reciprocal Rank Fusion) | Merges by rank position, ignores raw scores. | When dense and sparse scores are on different scales. |
| DBSF (Distribution-Based Score Fusion) | Normalizes scores using mean and standard deviation, then averages. | When you want score-aware blending. |
Both functions accept the same input — a list of `ScoredPoint` lists — and return a single merged and ranked list. The example below shows how to call each one directly:
```python theme={null}
from actian_vectorai import reciprocal_rank_fusion, distribution_based_score_fusion
# RRF: weights=[0.7, 0.3] gives 70% influence to dense results, 30% to sparse
fused = reciprocal_rank_fusion([dense_results, sparse_results], limit=10, weights=[0.7, 0.3])
# DBSF: no weights — scoring is based on score distribution across both lists
fused = distribution_based_score_fusion([dense_results, sparse_results], limit=10)
```
***
## Actian VectorAI features used
The system in this tutorial relies on the following Actian VectorAI SDK features. The table below lists each feature, the corresponding API call, and its role in the pipeline:
| Feature | API | Purpose |
| ------------------- | ------------------------------------ | ---------------------------------------- |
| Collection creation | `client.collections.get_or_create()` | Create vector space with HNSW config. |
| Point upsert | `client.points.upsert()` | Store CLIP vectors with product payload. |
| Dense search | `client.points.search()` | Server-side CLIP similarity search. |
| Point retrieval | `client.points.get()` | Fetch points by ID for BM25 scoring. |
| Vector count | `client.vde.get_vector_count()` | Return total number of indexed points. |
| Flush | `client.vde.flush()` | Persist buffered writes to disk. |
| Delete collection | `client.collections.delete()` | Remove collection and all its vectors. |
| RRF fusion | `reciprocal_rank_fusion()` | Rank-based result merging. |
| DBSF fusion | `distribution_based_score_fusion()` | Score-normalized result merging. |
***
## Benefits of hybrid search
Using dense and sparse retrieval together produces results that neither approach achieves alone. The three main advantages are outlined below.
### Better ranking
Hybrid search improves result ranking by combining two complementary signals. Semantic meaning allows CLIP to match "casual dark denim" against "jeans". Exact token matching allows BM25 to surface brand names like "French Connection" that CLIP embeddings may not distinguish from other text.
### Multimodal query support
The pipeline accepts three types of query input through a single `hybrid_search` call, making it straightforward to support different client interfaces without changing the search logic:
* Text queries, processed through the CLIP text encoder.
* Image queries, processed through the CLIP image encoder.
* Metadata keywords, scored by BM25 against stored product text.
### Tunable balance
The `alpha` parameter lets you shift the retrieval balance between visual similarity and keyword precision without changing any code — only the value passed to `hybrid_search` changes.
***
## Next steps
This tutorial covered a complete multimodal hybrid search pipeline — from CLIP embeddings and BM25 scoring through to RRF and DBSF fusion. The tutorials below cover additional retrieval patterns that can be layered on top of what was built here:
Improve relevance with cross-encoder and reciprocal rank fusion re-ranking.
Learn the core retrieval workflow.
Combine vector search with structured payload constraints.
Measure and optimize search accuracy using precision, recall, and MRR.
# AI legal contract intelligence agent
Source: https://docs.vectoraidb.actian.com/academy/articles/building-a-scalable-agent-memory-with-Actian-vector-AI-database
Build an AI-powered legal contract analysis system using Actian VectorAI DB with cross-collection lookup, payload-sorted retrieval, connection pooling, and quantization-aware search.
In this tutorial, you build an AI legal contract intelligence agent that uses Actian VectorAI DB. The agent maintains two collections — one for individual clauses and one for full contract summaries — and demonstrates a range of advanced database features: cross-collection lookup, payload-sorted retrieval, connection pooling, durability tuning, quantization-aware search, and strict deletion.
Legal teams deal with thousands of contracts — vendor agreements, NDAs, employment contracts, licensing terms, partnership deals — spread across business units, jurisdictions, and time periods. When a new contract arrives for review, the fundamental question is: how does this compare to what has been signed before?
A lawyer reviewing a new vendor agreement needs to find prior contracts with similar indemnification language, comparable liability caps, or matching force majeure clauses. They need to search across both a clause-level collection (granular) and a contract-level collection (holistic), rank results by recency, and compare the actual embeddings of similar clauses side by side.
Traditional contract management systems rely on manual tagging and folder hierarchies. They cannot surface semantically similar clauses across contracts that use different legal phrasing for the same concept. A clause stating "Party A shall indemnify and hold harmless Party B" should match "Vendor assumes full liability for third-party claims against Client" — but keyword search will miss this entirely.
This tutorial is a technical demonstration of vector database features using legal contract data as an example domain. The system built here is not a substitute for qualified legal advice. Do not use the output of this system to make real legal decisions. Always consult a licensed attorney for contract review and legal guidance.
***
## Architecture
The system splits contracts into two parallel embedding pipelines. Individual clauses are stored in a clause collection using Dot distance, while full contract summaries are stored in a contract collection using Cosine distance. At query time, clause search runs with OrderBy date ordering, and `lookup_from` enriches each result with context from the contract collection before the analysis engine generates a risk report.
```mermaid theme={null}
flowchart LR
Contracts["Legal Contracts"]
subgraph embed [Clause + Contract Embedding]
ClauseEmbed["Clause\nEmbeddings"]
ContractEmbed["Contract Summary\nEmbeddings"]
end
ClauseDB[("Clause Collection\nDot distance")]
ContractDB[("Contract Collection\nCosine distance")]
subgraph retrieve [Cross-Collection Retrieval]
Search["Clause Search\n+ OrderBy date"]
Lookup["lookup_from\nenrich with contract context"]
Search --> Lookup
end
Engine["Contract\nAnalysis Engine"]
Report["Risk Report\nclauses · precedents · alerts"]
Contracts --> ClauseEmbed & ContractEmbed
ClauseEmbed --> ClauseDB
ContractEmbed --> ContractDB
ClauseDB --> Search
ContractDB -.-> Lookup
Lookup --> Engine --> Report
```
***
## Environment setup
Before running any of the code in this tutorial, install the required Python packages. The setup uses two libraries: the Actian VectorAI SDK for database operations and Sentence Transformers for local embedding generation.
Run the following command to install both packages:
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
The two packages cover everything needed to connect to Actian VectorAI DB and produce embeddings locally:
* `actian-vectorai-client` — Official Python SDK for Actian VectorAI DB (connection pooling, cross-collection lookup, OrderBy retrieval, advanced rebuild management, gRPC transport).
* `sentence-transformers` — For generating text embeddings with `all-MiniLM-L6-v2`.
***
## Implementation
The following steps build the agent end-to-end: setting up collections with custom distance metrics and durability config, ingesting clause and contract data, running cross-collection lookups and payload-sorted queries, and operating a risk analysis engine over the results.
### Step 1: Import dependencies and configure
The first step imports all required types, including connection pooling types, WAL and optimizer configs, quantization search params, rebuild management types, `OrderBy`, and alternative distance metrics. It also sets the server address, collection names, and loads the embedding model. Running this block prints the active server address, collection names, and embedding model so the rest of the steps can be verified against a known configuration.
```python theme={null}
import asyncio
import math
from datetime import datetime, timezone
from actian_vectorai import (
AsyncVectorAIClient,
ConnectionPool, # production connection pool
PoolConfig, # pool configuration settings
Distance, # vector distance metrics (Cosine, Dot, Euclid, Manhattan)
Field,
FieldType,
FilterBuilder,
KeywordIndexParams,
DatetimeIndexParams,
FloatIndexParams,
PointStruct,
PrefetchQuery,
SearchParams,
QuantizationSearchParams, # quantization-aware search control
VectorParams,
OrderBy, # payload-field-sorted retrieval
Direction, # sort direction (Asc / Desc)
is_null,
)
from actian_vectorai.models.collections import (
HnswConfigDiff,
WalConfigDiff, # Write-Ahead Log durability tuning
OptimizersConfigDiff, # background optimization control
)
from actian_vectorai.models.points import ScoredPoint
from actian_vectorai.models.vde import (
RebuildDataSourceConfig, # source data for index rebuild
RebuildTargetConfig, # target index type for rebuild
RebuildRunConfig, # batch size and runtime settings
CompactOptions, # fine-grained compaction control
)
from actian_vectorai.models.enums import (
RebuildDataSourceType, # enum: current index, storage, snapshot
RebuildTargetIndexType, # enum: HNSW, flat, auto
)
from sentence_transformers import SentenceTransformer
# gRPC server address
SERVER = "localhost:6574"
# collection names for clause-level and contract-level data
CLAUSE_COLLECTION = "Legal-Clauses"
CONTRACT_COLLECTION = "Legal-Contracts"
EMBED_MODEL = "all-MiniLM-L6-v2"
EMBED_DIM = 384 # output dimension of all-MiniLM-L6-v2
model = SentenceTransformer(EMBED_MODEL)
print(f"VectorAI Server: {SERVER}")
print(f"Clause collection: {CLAUSE_COLLECTION}")
print(f"Contract collection: {CONTRACT_COLLECTION}")
print(f"Embedding model: {EMBED_MODEL} ({EMBED_DIM}-dim)")
```
This block imports all required SDK types — including connection pooling, WAL and optimizer configuration, quantization search parameters, rebuild management types, `OrderBy`, and alternative distance metrics — then sets the server address, collection names, and loads the `all-MiniLM-L6-v2` embedding model. The four `print` statements confirm the active configuration values so every subsequent step can be verified against a known baseline.
**Expected Output**
```text theme={null}
VectorAI Server: localhost:6574
Clause collection: Legal-Clauses
Contract collection: Legal-Contracts
Embedding model: all-MiniLM-L6-v2 (384-dim)
```
### Step 2: Define embedding helpers
These two helper functions wrap the Sentence Transformers model to convert text into 384-dimensional vectors. The single-text version is used for queries, while the batch version is used during data ingestion. Defining these functions produces no output; they are called in later steps.
```python theme={null}
def embed_text(text: str) -> list[float]:
"""Generate a 384-dimensional embedding for a text string."""
return model.encode(text).tolist()
def embed_texts(texts: list[str]) -> list[list[float]]:
"""Batch-embed multiple text strings."""
return model.encode(texts).tolist()
```
### Step 3: Create collections with alternative distance metrics, WAL, and optimizer config
Two collections are created using different distance metrics. The clause collection uses Dot product distance, while the contract collection uses Cosine distance. WAL and optimizer settings are configured on the clause collection for production workloads. Running this block creates both collections and prints a readiness message for each.
```python theme={null}
async def create_collections():
async with AsyncVectorAIClient(url=SERVER) as client:
# Clause collection uses Dot distance for magnitude-sensitive similarity
await client.collections.get_or_create(
name=CLAUSE_COLLECTION,
vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Dot),
hnsw_config=HnswConfigDiff(m=16, ef_construct=200),
# WAL durability: rotate segments at 64 MB, keep 2 ahead for crash recovery
wal_config=WalConfigDiff(
wal_capacity_mb=64,
wal_segments_ahead=2,
),
# Optimizer: delay HNSW build until 10k points, flush every 30 s
optimizers_config=OptimizersConfigDiff(
indexing_threshold=10000,
flush_interval_sec=30,
max_optimization_threads=2,
),
)
print(f"Collection '{CLAUSE_COLLECTION}' ready (Dot distance, WAL + optimizer tuned).")
# Contract collection uses Cosine distance for direction-based summary similarity
await client.collections.get_or_create(
name=CONTRACT_COLLECTION,
vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine),
hnsw_config=HnswConfigDiff(m=32, ef_construct=256),
)
print(f"Collection '{CONTRACT_COLLECTION}' ready (Cosine distance).")
asyncio.run(create_collections())
```
This block calls `get_or_create` twice — once for the clause collection using `Distance.Dot` with WAL durability and optimizer settings tuned for large ingest workloads, and once for the contract collection using `Distance.Cosine` with a higher HNSW `m` value for broader graph connectivity. Both collections use 384-dimensional vectors to match the embedding model configured in Step 1. Each call prints a readiness message confirming the distance metric and any additional configuration applied.
**Expected Output**
```text theme={null}
Collection 'Legal-Clauses' ready (Dot distance, WAL + optimizer tuned).
Collection 'Legal-Contracts' ready (Cosine distance).
```
This step introduces two configuration objects not covered in previous tutorials. `WalConfigDiff` controls Write-Ahead Log durability. Setting `wal_capacity_mb=64` rotates WAL segments at 64 MB, and `wal_segments_ahead=2` keeps two segments ahead for crash recovery. `OptimizersConfigDiff` controls background optimization: `indexing_threshold=10000` delays HNSW index construction until 10,000 points are present, `flush_interval_sec=30` sets the automatic flush interval, and `max_optimization_threads=2` limits concurrent optimization threads.
Alternative distance metrics — previous tutorials used `Distance.Cosine`. Actian VectorAI DB supports four metrics:
| Metric | Value | Best for |
| -------------------- | ----- | -------------------------------------------------- |
| `Distance.Cosine` | 1 | Normalized embeddings, direction-based similarity. |
| `Distance.Dot` | 3 | Raw dot product, magnitude-sensitive. |
| `Distance.Euclid` | 2 | Absolute distance in embedding space. |
| `Distance.Manhattan` | 4 | L1 norm, robust to outliers. |
For legal clauses, Dot product is useful when embedding magnitude carries meaning — longer, more detailed clause descriptions produce higher-magnitude vectors, and Dot product preserves this signal.
### Step 4: Create payload indexes on both collections
Payload indexes accelerate filtered queries by creating lookup structures on specific fields. This step creates indexes on `contract_date` and `contract_type` on both collections, then adds `clause_type` and `risk_score` indexes on the clause collection to support type-filtered and risk-sorted retrieval. Running this block creates all indexes and prints a confirmation.
```python theme={null}
async def create_indexes():
async with AsyncVectorAIClient(url=SERVER) as client:
for coll in [CLAUSE_COLLECTION, CONTRACT_COLLECTION]:
await client.points.create_field_index(
coll, field_name="contract_date",
field_type=FieldType.FieldTypeDatetime,
field_index_params=DatetimeIndexParams(on_disk=False, is_principal=True),
)
await client.points.create_field_index(
coll, field_name="contract_type",
field_type=FieldType.FieldTypeKeyword,
field_index_params=KeywordIndexParams(is_tenant=False),
)
await client.points.create_field_index(
CLAUSE_COLLECTION, field_name="clause_type",
field_type=FieldType.FieldTypeKeyword,
)
await client.points.create_field_index(
CLAUSE_COLLECTION, field_name="risk_score",
field_type=FieldType.FieldTypeFloat,
field_index_params=FloatIndexParams(is_principal=True),
)
print("Payload indexes created on both collections.")
asyncio.run(create_indexes())
```
This block iterates over both collections and creates a datetime index on `contract_date` (marked as principal for optimized `OrderBy` queries) and a keyword index on `contract_type`. It then creates two additional indexes on the clause collection only: a keyword index on `clause_type` for type-filtered retrieval, and a float index on `risk_score` (also marked as principal) to support fast range queries against numeric risk values. All six index operations run sequentially and a single confirmation message is printed once every index is in place.
**Expected Output**
```text theme={null}
Payload indexes created on both collections.
```
### Step 5: Prepare sample contract and clause data
Two datasets are defined: contract-level summaries and clause-level extracts. Each clause references a contract by ID, which enables cross-collection lookup in later steps. Running this block loads the data into memory and prints a count of each dataset.
```python theme={null}
contracts = [
{
"contract_id": "CTR-2025-001",
"title": "Master Services Agreement with TechVendor Inc.",
"summary": "Three-year master services agreement for cloud infrastructure and managed services. Includes SLA guarantees, data processing terms, and quarterly business reviews.",
"contract_type": "vendor_agreement",
"counterparty": "TechVendor Inc.",
"jurisdiction": "Delaware",
"contract_date": "2025-06-15T00:00:00Z",
"expiry_date": "2028-06-14T00:00:00Z",
"total_value": 2400000.00,
"status": "active",
},
{
"contract_id": "CTR-2025-002",
"title": "Non-Disclosure Agreement with DataPartner Ltd.",
"summary": "Mutual NDA covering proprietary algorithms, customer data, and business strategies. Two-year term with automatic renewal.",
"contract_type": "nda",
"counterparty": "DataPartner Ltd.",
"jurisdiction": "California",
"contract_date": "2025-09-01T00:00:00Z",
"expiry_date": "2027-08-31T00:00:00Z",
"total_value": 0.0,
"status": "active",
},
{
"contract_id": "CTR-2026-003",
"title": "Software Licensing Agreement with CloudStack Corp.",
"summary": "Enterprise license for AI/ML platform with unlimited seats. Includes source code escrow, uptime SLA, and priority support.",
"contract_type": "license",
"counterparty": "CloudStack Corp.",
"jurisdiction": "New York",
"contract_date": "2026-01-10T00:00:00Z",
"expiry_date": "2029-01-09T00:00:00Z",
"total_value": 850000.00,
"status": "active",
},
{
"contract_id": "CTR-2024-004",
"title": "Employment Agreement — Senior Counsel",
"summary": "Employment contract for senior legal counsel. Includes non-compete, IP assignment, and severance terms.",
"contract_type": "employment",
"counterparty": "Jane Doe",
"jurisdiction": "Massachusetts",
"contract_date": "2024-03-01T00:00:00Z",
"expiry_date": None,
"total_value": 280000.00,
"status": "active",
},
]
clauses = [
{
"clause_text": "Vendor shall indemnify, defend, and hold harmless Client from and against any third-party claims, damages, losses, and expenses arising from Vendor's breach of this Agreement or negligent acts.",
"clause_type": "indemnification",
"contract_id": "CTR-2025-001",
"contract_type": "vendor_agreement",
"contract_date": "2025-06-15T00:00:00Z",
"section": "Section 8.1",
"risk_score": 3.2,
},
{
"clause_text": "Client's total aggregate liability under this Agreement shall not exceed the fees paid by Client during the twelve-month period preceding the claim.",
"clause_type": "liability_cap",
"contract_id": "CTR-2025-001",
"contract_type": "vendor_agreement",
"contract_date": "2025-06-15T00:00:00Z",
"section": "Section 9.2",
"risk_score": 5.8,
},
{
"clause_text": "Neither party shall be liable for failure to perform obligations due to events beyond reasonable control, including natural disasters, acts of war, pandemics, government actions, or infrastructure failures lasting more than thirty days.",
"clause_type": "force_majeure",
"contract_id": "CTR-2025-001",
"contract_type": "vendor_agreement",
"contract_date": "2025-06-15T00:00:00Z",
"section": "Section 12.3",
"risk_score": 2.5,
},
{
"clause_text": "Receiving Party shall not disclose, publish, or disseminate Confidential Information to any third party without prior written consent. All confidential materials must be returned or destroyed upon termination.",
"clause_type": "confidentiality",
"contract_id": "CTR-2025-002",
"contract_type": "nda",
"contract_date": "2025-09-01T00:00:00Z",
"section": "Section 3.1",
"risk_score": 1.5,
},
{
"clause_text": "Licensee acknowledges that all intellectual property rights in the Software remain with Licensor. Licensee receives a non-exclusive, non-transferable, revocable license to use the Software.",
"clause_type": "ip_rights",
"contract_id": "CTR-2026-003",
"contract_type": "license",
"contract_date": "2026-01-10T00:00:00Z",
"section": "Section 4.1",
"risk_score": 4.0,
},
{
"clause_text": "Licensor guarantees 99.9% uptime availability measured monthly. If uptime falls below the guaranteed level, Licensee receives service credits equal to 10% of monthly fees per percentage point below the SLA.",
"clause_type": "sla",
"contract_id": "CTR-2026-003",
"contract_type": "license",
"contract_date": "2026-01-10T00:00:00Z",
"section": "Section 6.2",
"risk_score": 3.8,
},
{
"clause_text": "Employee agrees to a twelve-month non-compete restriction within a fifty-mile radius of Company headquarters. Employee shall not solicit Company clients or employees for twenty-four months following termination.",
"clause_type": "non_compete",
"contract_id": "CTR-2024-004",
"contract_type": "employment",
"contract_date": "2024-03-01T00:00:00Z",
"section": "Section 7.1",
"risk_score": 7.5,
},
{
"clause_text": "All inventions, discoveries, and works of authorship created by Employee during employment and related to Company business shall be the exclusive property of Company. Employee assigns all rights therein.",
"clause_type": "ip_assignment",
"contract_id": "CTR-2024-004",
"contract_type": "employment",
"contract_date": "2024-03-01T00:00:00Z",
"section": "Section 5.3",
"risk_score": 6.0,
},
]
print(f"{len(contracts)} contracts and {len(clauses)} clauses loaded.")
```
This block defines two in-memory Python lists: `contracts`, which contains four contract-level records spanning vendor agreements, an NDA, a software license, and an employment contract; and `clauses`, which contains eight clause-level extracts covering indemnification, liability cap, force majeure, confidentiality, IP rights, SLA, non-compete, and IP assignment clause types. Each clause includes a `contract_id` field that links it to its parent contract, enabling cross-collection enrichment in later steps. The final `print` statement reports how many records of each type were loaded.
**Expected Output**
```text theme={null}
4 contracts and 8 clauses loaded.
```
### Step 6: Ingest data into both collections
This step embeds all contract summaries and clause texts, then upserts them into their respective collections. After upserting, it flushes both collections to disk and prints the final point counts. Running this block populates both collections and confirms the ingested totals.
```python theme={null}
async def ingest_data():
async with AsyncVectorAIClient(url=SERVER) as client:
# Embed all contract summaries and upsert them into the contract collection
contract_texts = [c["summary"] for c in contracts]
contract_vectors = embed_texts(contract_texts)
contract_points = [
# Store all fields except "summary" as payload; keep the text as "summary_text"
PointStruct(id=i, vector=contract_vectors[i], payload={k: v for k, v in c.items() if k != "summary"} | {"summary_text": c["summary"]})
for i, c in enumerate(contracts)
]
await client.points.upsert(CONTRACT_COLLECTION, points=contract_points)
await client.vde.flush(CONTRACT_COLLECTION) # persist to disk immediately
# Embed all clause texts and upsert them into the clause collection
clause_texts = [c["clause_text"] for c in clauses]
clause_vectors = embed_texts(clause_texts)
clause_points = [
PointStruct(id=i, vector=clause_vectors[i], payload=c)
for i, c in enumerate(clauses)
]
await client.points.upsert(CLAUSE_COLLECTION, points=clause_points)
await client.vde.flush(CLAUSE_COLLECTION) # persist to disk immediately
# Read back the ingested counts to confirm both writes succeeded
clause_count = await client.vde.get_vector_count(CLAUSE_COLLECTION)
contract_count = await client.vde.get_vector_count(CONTRACT_COLLECTION)
print(f"Ingested {contract_count} contracts and {clause_count} clauses.")
asyncio.run(ingest_data())
```
This block batch-encodes all contract summaries and clause texts using the `embed_texts` helper, constructs `PointStruct` objects that pair each numeric ID with its vector and payload, and upserts them into the corresponding collections. After each upsert, `vde.flush` is called to persist the writes to disk immediately rather than waiting for the background flush interval. The final two `get_vector_count` calls read back the indexed totals from each collection to confirm all points were successfully written.
**Expected Output**
```text theme={null}
Ingested 4 contracts and 8 clauses.
```
### Step 7: OrderBy — payload-sorted retrieval
The `query` endpoint supports `OrderBy` for sorting results by a payload field instead of vector similarity. The function below accepts a clause type, filters to matching clauses, and returns them sorted by `contract_date` descending so the most recent clauses appear first. Running this block queries for indemnification and IP rights clauses and prints their dates and contract IDs.
```python theme={null}
async def get_recent_clauses(clause_type: str, limit: int = 5):
filter_obj = FilterBuilder().must(Field("clause_type").eq(clause_type)).build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
CLAUSE_COLLECTION,
query={"order_by": OrderBy(key="contract_date", direction=Direction.Desc)},
filter=filter_obj,
limit=limit,
with_payload=True,
)
return results
results = asyncio.run(get_recent_clauses("indemnification"))
print("=== Most Recent Indemnification Clauses (OrderBy date DESC) ===")
for r in results:
print(f" id={r.id} date={r.payload.get('contract_date')} contract={r.payload.get('contract_id')}")
print(f" {r.payload.get('clause_text', '')[:100]}...")
results = asyncio.run(get_recent_clauses("ip_rights"))
print("\n=== Most Recent IP Rights Clauses ===")
for r in results:
print(f" id={r.id} date={r.payload.get('contract_date')} contract={r.payload.get('contract_id')}")
```
This block calls `get_recent_clauses` twice — first filtering for clauses where `clause_type` equals `"indemnification"`, then for `"ip_rights"`. Each call issues a `query` against the clause collection with an `OrderBy` directive on `contract_date` in descending order, so the most recently dated clause of each type is returned first. No vector is provided; the query relies entirely on payload filtering and date sorting. Each result is printed with its point ID, contract date, contract ID, and the first 100 characters of the clause text.
**Expected Output**
```text theme={null}
=== Most Recent Indemnification Clauses (OrderBy date DESC) ===
id=0 date=2025-06-15T00:00:00Z contract=CTR-2025-001
Vendor shall indemnify, defend, and hold harmless Client from and against any third-party claims,...
=== Most Recent IP Rights Clauses ===
id=4 date=2026-01-10T00:00:00Z contract=CTR-2026-003
```
`OrderBy` replaces vector similarity ranking with payload-field sorting. The `query` endpoint accepts it through a dict: `{"order_by": OrderBy(key="contract_date", direction=Direction.Desc)}`. `Direction.Desc` sorts newest first; `Direction.Asc` sorts oldest first. The `is_principal=True` flag set on the datetime index in step 4 optimizes this ordering. This retrieval mode is essential for legal workflows where recency matters — the most recent version of an indemnification clause is more relevant than one from five years ago.
### Step 8: Cross-collection lookup with lookup\_from
The `lookup_from` parameter on `query` enriches results from one collection with data from another. The function below searches the clause collection by vector similarity and uses `lookup_from` to draw in contract-level vectors during scoring. Running this block searches for clauses related to liability limitation and prints the top matches with their scores.
```python theme={null}
async def search_clauses_with_contract_context(query: str, top_k: int = 5):
query_vector = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
CLAUSE_COLLECTION,
query=query_vector,
limit=top_k,
with_payload=True,
lookup_from={
"collection": CONTRACT_COLLECTION,
"vector_name": "",
},
)
return results
results = asyncio.run(search_clauses_with_contract_context(
"Liability limitation and cap on damages"
))
print("=== Clause Search with Cross-Collection Lookup ===")
for r in results:
print(f" id={r.id} score={r.score:.4f} type={r.payload.get('clause_type')} contract={r.payload.get('contract_id')}")
print(f" {r.payload.get('clause_text', '')[:100]}...")
```
This block encodes the query string `"Liability limitation and cap on damages"` into a 384-dimensional vector and searches the clause collection for the top five most similar points. The `lookup_from` parameter instructs the query to draw in vectors from the contract collection during scoring, enriching clause-level results with contract-level embedding context. Each result is printed with its point ID, similarity score, clause type, parent contract ID, and the first 100 characters of the clause text.
**Expected Output**
```text theme={null}
=== Clause Search with Cross-Collection Lookup ===
id=1 score=0.8200 type=liability_cap contract=CTR-2025-001
Client's total aggregate liability under this Agreement shall not exceed the fees paid by Client ...
id=0 score=0.6100 type=indemnification contract=CTR-2025-001
Vendor shall indemnify, defend, and hold harmless Client from and against any third-party claims,...
```
The `lookup_from` parameter accepts two keys: `"collection"` for the name of the external collection to look up vectors from, and `"vector_name"` for the named vector to use (an empty string selects the default vector). The result is clause-level granularity for matching combined with contract-level embeddings for contextual scoring — useful when you need both.
### Step 9: Retrieve vectors alongside payloads
By default, `points.get` and `points.search` return only payloads and scores. Setting `with_vectors=True` includes the actual embedding data in the response, enabling client-side similarity analysis. The code below retrieves the embeddings for three specific clauses by ID, computes their pairwise cosine similarity, and then runs a search that also returns vectors. Running this block prints the vector dimension and first three values for each retrieved point, followed by the similarity score between clause 0 and clause 1.
```python theme={null}
async def get_clause_vectors(clause_ids: list[int]):
async with AsyncVectorAIClient(url=SERVER) as client:
# Retrieve points by ID; with_vectors=True includes the raw embedding arrays
points = await client.points.get(
CLAUSE_COLLECTION,
ids=clause_ids,
with_payload=True,
with_vectors=True,
)
return points
async def search_with_vectors(query: str, top_k: int = 3):
query_vector = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
# Standard similarity search; with_vectors=True appends embedding to each result
results = await client.points.search(
CLAUSE_COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
with_vectors=True,
) or []
return results
def cosine_similarity(a: list[float], b: list[float]) -> float:
# Manual cosine similarity used to compare two clause embeddings client-side
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(x * x for x in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
points = asyncio.run(get_clause_vectors([0, 1, 2]))
print("=== Clause Vectors Retrieved (with_vectors=True) ===")
for p in points:
vec = p.vector if isinstance(p.vector, list) else []
print(f" id={p.id} type={p.payload.get('clause_type')} vector_dim={len(vec)} first_3={vec[:3]}")
if len(points) >= 2:
v0 = points[0].vector if isinstance(points[0].vector, list) else []
v1 = points[1].vector if isinstance(points[1].vector, list) else []
sim = cosine_similarity(v0, v1)
print(f"\nCosine similarity between clause 0 ({points[0].payload.get('clause_type')}) and clause 1 ({points[1].payload.get('clause_type')}): {sim:.4f}")
results = asyncio.run(search_with_vectors("limitation of liability"))
print("\n=== Search with Vectors ===")
for r in results:
vec = r.vector if isinstance(r.vector, list) else []
print(f" id={r.id} score={r.score:.4f} vector_dim={len(vec)} type={r.payload.get('clause_type')}")
```
This block retrieves clause points 0, 1, and 2 by ID with `with_vectors=True`, then computes the cosine similarity between clause 0 (indemnification) and clause 1 (liability cap) using a manual dot-product calculation. It also runs a similarity search for `"limitation of liability"` with `with_vectors=True` to show that search results can carry their full embedding arrays. Each retrieved point is printed with its clause type, vector dimension (confirming the 384-dim model), and the first three float values of the embedding. The pairwise similarity and the ranked search results follow.
**Expected Output**
```text theme={null}
=== Clause Vectors Retrieved (with_vectors=True) ===
id=0 type=indemnification vector_dim=384 first_3=[0.0234, -0.0891, 0.0456]
id=1 type=liability_cap vector_dim=384 first_3=[-0.0123, 0.0567, 0.0789]
id=2 type=force_majeure vector_dim=384 first_3=[0.0345, -0.0234, 0.0123]
Cosine similarity between clause 0 (indemnification) and clause 1 (liability_cap): 0.6234
=== Search with Vectors ===
id=1 score=0.8200 vector_dim=384 type=liability_cap
id=0 score=0.6100 vector_dim=384 type=indemnification
id=5 score=0.4500 vector_dim=384 type=sla
```
Returning vectors increases response size significantly, so use `with_vectors=True` selectively. The primary use cases are client-side pairwise comparison, visualization (t-SNE or UMAP projections), debugging embedding quality, and exporting vectors for use in other systems.
### Step 10: Approximate count for fast dashboards
`points.count` supports an `exact` flag. When set to `False`, it returns an approximate count using index metadata rather than scanning all segments. The function below counts clauses and contracts both ways, then runs two filtered approximate counts — one for high-risk clauses and one for vendor agreement clauses. Running this block prints all four counts.
```python theme={null}
async def dashboard_counts():
async with AsyncVectorAIClient(url=SERVER) as client:
# exact=True scans all segments; exact=False reads index metadata for speed
exact_clause = await client.points.count(CLAUSE_COLLECTION, exact=True)
approx_clause = await client.points.count(CLAUSE_COLLECTION, exact=False)
print(f"Clauses — exact: {exact_clause}, approximate: {approx_clause}")
exact_contract = await client.points.count(CONTRACT_COLLECTION, exact=True)
approx_contract = await client.points.count(CONTRACT_COLLECTION, exact=False)
print(f"Contracts — exact: {exact_contract}, approximate: {approx_contract}")
# Count only clauses with risk_score >= 5.0 using a float filter
high_risk_filter = FilterBuilder().must(Field("risk_score").gte(5.0)).build()
high_risk_approx = await client.points.count(
CLAUSE_COLLECTION,
filter=high_risk_filter,
exact=False,
)
print(f"High-risk clauses (risk >= 5.0, approximate): {high_risk_approx}")
# Count only clauses belonging to vendor_agreement contracts
vendor_filter = FilterBuilder().must(Field("contract_type").eq("vendor_agreement")).build()
vendor_approx = await client.points.count(
CLAUSE_COLLECTION,
filter=vendor_filter,
exact=False,
)
print(f"Vendor agreement clauses (approximate): {vendor_approx}")
asyncio.run(dashboard_counts())
```
This block runs four count operations against the two collections. The first two pairs call `points.count` with `exact=True` and `exact=False` on both the clause and contract collections to illustrate the speed-accuracy trade-off between full segment scans and index-metadata reads. The third count applies a float filter (`risk_score >= 5.0`) on the clause collection to return an approximate tally of high-risk clauses. The fourth applies a keyword filter (`contract_type == "vendor_agreement"`) to count vendor agreement clauses only. All results are printed in sequence.
**Expected Output**
```text theme={null}
Clauses — exact: 8, approximate: 8
Contracts — exact: 4, approximate: 4
High-risk clauses (risk >= 5.0, approximate): 3
Vendor agreement clauses (approximate): 3
```
The two modes behave differently in terms of speed and accuracy:
| Mode | Speed | Accuracy |
| ------------- | ---------------------------- | ------------------------------- |
| `exact=True` | Slower (scans all segments). | 100% accurate. |
| `exact=False` | Fast (uses index metadata). | May differ slightly from exact. |
For dashboards showing "\~12,450 clauses in 340 contracts", approximate counts are sufficient and much faster at scale. Exact counts are needed for billing, compliance reporting, or data integrity checks.
### Step 11: Strict deletion — validate before removing
`strict=True` on `points.delete` validates that all specified IDs exist before performing the deletion. If any ID is missing, the entire operation is rejected without deleting anything. The code below inserts a temporary test clause, deletes it successfully with `strict=True`, then attempts a second deletion that includes a non-existent ID to demonstrate the error behavior. Running this block prints the result of each operation.
```python theme={null}
async def strict_deletion_demo():
async with AsyncVectorAIClient(url=SERVER) as client:
vector = embed_text("Temporary test clause for strict deletion demo.")
await client.points.upsert_single(
CLAUSE_COLLECTION, id=999,
vector=vector,
payload={"clause_type": "test", "clause_text": "Temporary test clause."},
)
print("Inserted test clause (id=999).")
result = await client.points.delete(
CLAUSE_COLLECTION,
ids=[999],
strict=True,
)
print(f"Strict delete of id=999: status={result.status}")
try:
await client.points.delete(
CLAUSE_COLLECTION,
ids=[999, 9999],
strict=True,
)
except Exception as e:
print(f"Strict delete of [999, 9999] failed as expected: {type(e).__name__}")
print(f" {e}")
await client.vde.flush(CLAUSE_COLLECTION)
asyncio.run(strict_deletion_demo())
```
This block first inserts a temporary clause at point ID 999 using `upsert_single`, then deletes it immediately with `strict=True` to confirm that a valid deletion succeeds and returns `UpdateStatus.Completed`. It then attempts a second deletion that includes both ID 999 (now removed) and ID 9999 (never existed) to trigger the strict validation failure. Because at least one ID in the batch cannot be found, the operation is rejected entirely and a `PointNotFoundError` is raised listing all missing IDs. The collection is flushed at the end to persist the final state.
**Expected Output**
```text theme={null}
Inserted test clause (id=999).
Strict delete of id=999: status=UpdateStatus.Completed
Strict delete of [999, 9999] failed as expected: PointNotFoundError
Points not found: [999, 9999]
```
The two deletion modes handle missing IDs differently:
| Mode | Behavior |
| ------------------------ | ---------------------------------------------------------------------- |
| `strict=False` (default) | Silently ignores non-existent IDs. |
| `strict=True` | Raises `PointNotFoundError` listing every missing ID, deletes nothing. |
For legal contract management, strict deletion prevents accidental data loss. If a batch delete includes IDs that do not exist — possibly already deleted or mistyped — the entire operation is rejected. This is essential for audit trails and compliance.
### Step 12: Quantization-aware search with SearchParams
When a collection uses quantization (scalar, product, or binary), search can use the compressed vectors for speed and optionally rescore with full-precision vectors for accuracy. The `QuantizationSearchParams` object inside `SearchParams` controls this behavior. The function below runs a clause search with oversampling and rescoring enabled to maximize accuracy. Running this block searches for liability cap provisions and prints the top matching clauses.
```python theme={null}
async def quantization_aware_search(query: str, top_k: int = 5):
query_vector = embed_text(query)
params = SearchParams(
hnsw_ef=200,
exact=False,
quantization=QuantizationSearchParams(
ignore=False, # use quantized vectors during initial retrieval
rescore=True, # rescore candidates with full-precision vectors
oversampling=2.0, # retrieve 2x candidates before rescoring
),
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
CLAUSE_COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
params=params,
) or []
return results
results = asyncio.run(quantization_aware_search(
"Limitation of damages and liability cap provisions"
))
print("=== Quantization-Aware Search ===")
for r in results:
print(f" id={r.id} score={r.score:.4f} type={r.payload.get('clause_type')} risk={r.payload.get('risk_score')}")
```
This block encodes the query `"Limitation of damages and liability cap provisions"` and searches the clause collection using a `SearchParams` object that configures quantization-aware retrieval. `ignore=False` allows the initial HNSW scan to use compressed quantized vectors for speed, `oversampling=2.0` doubles the candidate pool to 10 before final ranking, and `rescore=True` re-evaluates the top candidates using full-precision vectors to recover accuracy lost during compression. Each result is printed with its point ID, similarity score, clause type, and numeric risk score from the payload.
**Expected Output**
```text theme={null}
=== Quantization-Aware Search ===
id=1 score=0.8200 type=liability_cap risk=5.8
id=0 score=0.6100 type=indemnification risk=3.2
id=5 score=0.4500 type=sla risk=3.8
```
The three `QuantizationSearchParams` settings interact as follows:
| Parameter | Effect |
| ------------------ | ----------------------------------------------------------------------------------------------- |
| `ignore=False` | Use quantized vectors during search (fast). |
| `ignore=True` | Skip quantization, use full-precision vectors. |
| `rescore=True` | After initial retrieval with quantized vectors, rescore candidates with full-precision vectors. |
| `oversampling=2.0` | Retrieve 2x candidates before rescoring (higher recall at cost of latency). |
The combination `ignore=False, rescore=True, oversampling=2.0` provides the best accuracy-speed trade-off: fast initial search with quantized vectors, then precise rescoring with original vectors over a 2x candidate pool.
### Step 13: Update specific payload keys with the key parameter
The `key` parameter on `set_payload` places the new payload data under a specific nested key path rather than merging it at the top level. The code below adds review metadata to three clauses under the key `"review_metadata"`, then retrieves clause 0 to confirm the nested structure. Running this block prints the updated payload keys and the content of the new nested field.
```python theme={null}
async def targeted_payload_update():
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.set_payload(
CLAUSE_COLLECTION,
payload={"reviewed": True, "reviewer": "Sarah Chen", "review_date": "2026-03-23T10:00:00Z"},
ids=[0, 1, 2],
key="review_metadata",
)
print("Updated clauses 0-2 with review metadata under 'review_metadata' key.")
point = await client.points.get(CLAUSE_COLLECTION, ids=[0], with_payload=True)
if point:
print(f" Clause 0 payload keys: {list(point[0].payload.keys())}")
print(f" review_metadata: {point[0].payload.get('review_metadata')}")
asyncio.run(targeted_payload_update())
```
This block calls `set_payload` with `key="review_metadata"` to write three review fields — `reviewed`, `reviewer`, and `review_date` — as a nested object under that key for clause points 0, 1, and 2. Without the `key` parameter, these fields would be merged at the top level alongside `clause_type` and `clause_text`. After the update, `points.get` retrieves clause 0 to confirm that `review_metadata` appears as a new top-level key and that its nested content matches what was written.
**Expected Output**
```text theme={null}
Updated clauses 0-2 with review metadata under 'review_metadata' key.
Clause 0 payload keys: ['clause_text', 'clause_type', 'contract_id', 'contract_type', 'contract_date', 'section', 'risk_score', 'review_metadata']
review_metadata: {'reviewed': True, 'reviewer': 'Sarah Chen', 'review_date': '2026-03-23T10:00:00Z'}
```
The following JSON shows how the updated payload looks after the write. The review fields are nested under `"review_metadata"` rather than added at the top level alongside `clause_type` and `clause_text`:
```json theme={null}
{
"clause_type": "indemnification",
"clause_text": "...",
"review_metadata": {
"reviewed": true,
"reviewer": "Sarah Chen",
"review_date": "2026-03-23T10:00:00Z"
}
}
```
This keeps the payload organized and prevents key collisions when multiple systems update the same point.
### Step 14: Connection pooling for production workloads
`pool_size` on `AsyncVectorAIClient` creates multiple concurrent gRPC connections for high-throughput scenarios. The code below initializes a client with four connections and issues four clause searches in parallel using `asyncio.gather`, distributing the queries across the connection pool. Running this block prints the top results for each of the four queries.
```python theme={null}
async def pooled_connection_demo():
client = AsyncVectorAIClient(
url=SERVER,
pool_size=4, # open four gRPC channels to distribute concurrent requests
timeout=30.0,
max_retries=3,
)
async with client:
queries = [
"indemnification and hold harmless",
"limitation of liability and damages cap",
"force majeure and acts of god",
"intellectual property assignment",
]
async def search_one(q: str):
vec = embed_text(q)
return await client.points.search(
CLAUSE_COLLECTION, vector=vec, limit=3, with_payload=True,
) or []
tasks = [search_one(q) for q in queries]
all_results = await asyncio.gather(*tasks)
for query, results in zip(queries, all_results):
print(f"\nQuery: {query[:50]}...")
for r in results:
print(f" id={r.id} score={r.score:.4f} type={r.payload.get('clause_type')}")
asyncio.run(pooled_connection_demo())
```
This block initializes `AsyncVectorAIClient` with `pool_size=4`, which opens four concurrent gRPC channels to the server. Four search queries are defined — covering indemnification, liability cap, force majeure, and IP assignment — and submitted simultaneously using `asyncio.gather`, which distributes each query across the available pool connections. Each `search_one` coroutine embeds its query string independently and retrieves the top three matching clauses. The results for all four queries are printed in order once all concurrent searches complete.
**Expected Output**
```text theme={null}
Query: indemnification and hold harmless...
id=0 score=0.8900 type=indemnification
id=1 score=0.5800 type=liability_cap
Query: limitation of liability and damages cap...
id=1 score=0.8500 type=liability_cap
id=0 score=0.6200 type=indemnification
Query: force majeure and acts of god...
id=2 score=0.8700 type=force_majeure
Query: intellectual property assignment...
id=7 score=0.8800 type=ip_assignment
id=4 score=0.7200 type=ip_rights
```
The three client parameters interact as follows:
| Parameter | Default | Purpose |
| ------------- | ------- | -------------------------------------- |
| `pool_size` | 1 | Number of concurrent gRPC channels. |
| `timeout` | 30.0 | Per-call timeout in seconds. |
| `max_retries` | 3 | Automatic retry on transient failures. |
With `pool_size=4`, four concurrent searches run simultaneously over separate gRPC channels. A single connection would serialize all requests, which is unacceptable for production workloads with high query concurrency.
### Step 15: Advanced rebuild management
`trigger_rebuild` with a full configuration object provides fine-grained control over index rebuilds. The code below triggers a rebuild on the clause collection, immediately checks its status with `get_rebuild_task`, and then lists all rebuild tasks for the collection. Running this block prints the task ID, initial state, and a summary of all active tasks.
```python theme={null}
async def advanced_rebuild_demo():
async with AsyncVectorAIClient(url=SERVER) as client:
task_id, stats = await client.vde.trigger_rebuild(
CLAUSE_COLLECTION,
source=RebuildDataSourceConfig(
source_type=RebuildDataSourceType.SOURCE_CURRENT_INDEX,
),
target=RebuildTargetConfig(
target_type=RebuildTargetIndexType.TARGET_HNSW,
),
run_config=RebuildRunConfig(
batch_size=1000,
),
wait=False,
priority=10,
)
print(f"Rebuild triggered. Task ID: {task_id}")
task_info = await client.vde.get_rebuild_task(task_id)
print(f"Task state: {task_info.state}")
print(f"Task info: {task_info}")
tasks, total = await client.vde.list_rebuild_tasks(
collection_name=CLAUSE_COLLECTION,
limit=5,
)
print(f"\nRebuild tasks for {CLAUSE_COLLECTION}: {total} total")
for t in tasks:
print(f" Task {t.task_id}: state={t.state}")
# Cancel if still running (uncomment for production use):
# cancelled = await client.vde.cancel_rebuild_task(task_id)
# print(f"Task cancelled: {cancelled}")
asyncio.run(advanced_rebuild_demo())
```
This block triggers an index rebuild on the clause collection by providing a full `RebuildDataSourceConfig` (reading from the current index), a `RebuildTargetConfig` (rebuilding to HNSW), and a `RebuildRunConfig` with a batch size of 1,000 vectors. `wait=False` means the call returns immediately after submitting the task rather than blocking until completion. The assigned task ID is then used with `get_rebuild_task` to fetch the initial task state, and `list_rebuild_tasks` retrieves all rebuild tasks for the collection so their states can be compared. Because the collection is small, the task may already be in `TASK_COMPLETED` state by the time `list_rebuild_tasks` is called.
**Expected Output**
```text theme={null}
Rebuild triggered. Task ID: rebuild-abc123
Task state: RebuildTaskState.TASK_RUNNING
Task info: RebuildTaskInfo(...)
Rebuild tasks for Legal-Clauses: 1 total
Task rebuild-abc123: state=RebuildTaskState.TASK_COMPLETED
```
The `trigger_rebuild` parameters and monitoring methods are as follows:
| Parameter | Type | Purpose |
| ------------ | ------------------------- | -------------------------------------------------------------- |
| `source` | `RebuildDataSourceConfig` | Where to read vectors from (current index, storage, snapshot). |
| `target` | `RebuildTargetConfig` | What index type to build (HNSW, flat, auto). |
| `run_config` | `RebuildRunConfig` | Batch size, catchup rounds, and other runtime settings. |
| `wait` | `bool` | Block until rebuild completes. |
| `priority` | `int` | Higher priority tasks execute first. |
Once a rebuild is running, three methods are available for monitoring and control:
| Method | Purpose |
| ------------------------------ | ----------------------------------------------------------- |
| `get_rebuild_task(task_id)` | Get status and progress of a specific task. |
| `list_rebuild_tasks(...)` | List all tasks, optionally filtered by collection or state. |
| `cancel_rebuild_task(task_id)` | Cancel a running rebuild. |
This is essential for production maintenance windows — trigger a rebuild, monitor progress, and cancel if it runs too long.
### Step 16: Compaction with CompactOptions
`CompactOptions` provides fine-grained control over collection compaction — merging segments, purging deleted vector tombstones, and reclaiming storage. The code below triggers a compaction on the clause collection with `wait=True` so the call blocks until completion, then retrieves the collection state to confirm it is ready. Running this block prints the task ID, any available stats, and the post-compaction collection state.
```python theme={null}
async def compaction_demo():
async with AsyncVectorAIClient(url=SERVER) as client:
task_id, stats = await client.vde.compact_collection(
CLAUSE_COLLECTION,
options=CompactOptions(),
wait=True,
wait_timeout=120,
)
print(f"Compaction completed. Task: {task_id}")
if stats:
print(f" Stats: {stats}")
state = await client.vde.get_state(CLAUSE_COLLECTION)
print(f" Collection state after compaction: {state}")
asyncio.run(compaction_demo())
```
This block calls `compact_collection` on the clause collection with `wait=True` and a 120-second timeout, which blocks the calling coroutine until all compaction work finishes. Compaction merges small segments, purges deleted vector tombstones accumulated from earlier operations such as the strict deletion demo in Step 11, and reclaims the freed disk space. Once compaction completes, `vde.get_state` is called to confirm the collection has returned to `CollectionState.READY` and is ready to serve queries.
**Expected Output**
```text theme={null}
Compaction completed. Task: compact-xyz789
Stats: CompactStats(...)
Collection state after compaction: CollectionState.READY
```
Compaction is critical for collections with frequent deletions and updates. Over time, deleted vectors leave tombstones that occupy space and slow searches. Compaction merges small segments into larger ones, purges those tombstones, reclaims disk space, and improves query performance. Setting `wait=True` blocks until compaction finishes, and `wait_timeout=120` sets a maximum wait time in seconds.
### Step 17: Build the contract analysis engine
The analysis engine takes a query clause and a list of retrieved precedent clauses, scores each precedent by risk level and match quality, and returns a structured risk report. This function does not call the database — it operates entirely on the `ScoredPoint` objects returned by earlier search steps. Defining this function produces no output; it is called in step 18 where the full analysis runs.
```python theme={null}
def analyze_clauses(query_clause: str, precedents: list[ScoredPoint]) -> dict:
"""Analyze retrieved precedent clauses and generate risk assessment."""
if not precedents:
return {
"risk_level": "unknown",
"message": "No precedent clauses found.",
"precedents": 0,
"findings": [],
}
findings = []
risk_scores = []
clause_types_seen = set()
for p in precedents:
payload = p.payload or {}
clause_type = payload.get("clause_type", "unknown")
risk = payload.get("risk_score", 0.0)
contract_id = payload.get("contract_id", "unknown")
similarity = float(p.score or 0.0) # vector similarity score from the search
clause_types_seen.add(clause_type)
risk_scores.append(risk)
finding = {
"clause_id": p.id,
"clause_type": clause_type,
"contract_id": contract_id,
"similarity": round(similarity, 4),
"risk_score": risk,
"section": payload.get("section", ""),
}
# Assign a risk alert based on the numeric risk_score payload field
if risk >= 7.0:
finding["alert"] = "HIGH RISK — review with senior counsel"
elif risk >= 5.0:
finding["alert"] = "MODERATE RISK — standard review recommended"
else:
finding["alert"] = "LOW RISK — precedent exists"
# Classify match quality by similarity score thresholds
if similarity >= 0.8:
finding["match_quality"] = "strong"
elif similarity >= 0.5:
finding["match_quality"] = "moderate"
else:
finding["match_quality"] = "weak"
findings.append(finding)
avg_risk = sum(risk_scores) / len(risk_scores) if risk_scores else 0.0
max_risk = max(risk_scores) if risk_scores else 0.0
# Determine overall risk level from the worst single clause and the average
if max_risk >= 7.0 or avg_risk >= 5.0:
risk_level = "high"
elif max_risk >= 5.0 or avg_risk >= 3.0:
risk_level = "medium"
else:
risk_level = "low"
return {
"risk_level": risk_level,
"average_risk_score": round(avg_risk, 2),
"max_risk_score": max_risk,
"precedents": len(findings),
"clause_types": sorted(clause_types_seen),
"findings": findings,
"message": f"Found {len(findings)} precedent(s) across {len(clause_types_seen)} clause type(s). "
f"Overall risk: {risk_level} (avg={avg_risk:.1f}, max={max_risk:.1f}).",
}
```
### Step 18: Run the end-to-end contract analysis
`analyze_new_clause` embeds an incoming clause, searches the clause collection for the top five most similar precedents above a score threshold of 0.3, and passes the results to `analyze_clauses` to produce a risk report. Running this block analyzes two new clauses — a liability limitation clause and a non-compete clause — and prints a formatted risk report for each.
```python theme={null}
async def analyze_new_clause(clause_text: str):
query_vector = embed_text(clause_text)
async with AsyncVectorAIClient(url=SERVER) as client:
precedents = await client.points.search(
CLAUSE_COLLECTION,
vector=query_vector,
limit=5,
with_payload=True,
score_threshold=0.3,
) or []
report = analyze_clauses(clause_text, precedents)
print(f"\n{'='*60}")
print(f"CONTRACT CLAUSE ANALYSIS REPORT")
print(f"{'='*60}")
print(f"Clause: {clause_text[:100]}...")
print(f"\n{report['message']}")
print(f"Risk Level: {report['risk_level'].upper()}")
print(f"Average Risk: {report['average_risk_score']}")
print(f"Max Risk: {report['max_risk_score']}")
if report["findings"]:
print(f"\nPrecedent Findings:")
for f in report["findings"]:
print(f" [{f['match_quality'].upper()}] Clause {f['clause_id']} ({f['clause_type']}) — {f['contract_id']} {f['section']}")
print(f" Similarity: {f['similarity']} Risk: {f['risk_score']} {f['alert']}")
print()
asyncio.run(analyze_new_clause(
"The Service Provider's total liability for all claims arising under this agreement "
"shall be limited to the total fees paid in the preceding six-month period. In no event "
"shall either party be liable for indirect, consequential, or punitive damages."
))
asyncio.run(analyze_new_clause(
"Employee agrees that for a period of eighteen months following termination, Employee "
"shall not directly or indirectly engage in any business that competes with the Company "
"within a one-hundred-mile radius of any Company office."
))
```
This block runs the end-to-end analysis pipeline twice. The first call submits a liability limitation clause; the second submits a non-compete clause. For each clause, `analyze_new_clause` embeds the text, searches the clause collection for the top five precedents with a minimum similarity threshold of 0.3, and passes the results to `analyze_clauses`. The analysis engine scores each precedent by risk level and match quality, computes the average and maximum risk scores across all findings, and determines an overall risk level of `low`, `medium`, or `high`. Each report is printed with a header separator, the truncated clause text, the summary message, risk metrics, and a ranked list of precedent findings annotated with match quality labels and risk alerts.
**Expected Output**
```text theme={null}
```
# Overview
Source: https://docs.vectoraidb.actian.com/academy/articles/index
Browse and choose from articles on building AI agents and intelligent applications with Actian VectorAI DB.
These articles cover real-world AI agent architectures, multimodal systems, and industry-specific applications built with Actian VectorAI DB. Each article walks through an implementation covering data modeling, vector ingestion, semantic retrieval, filtering, and reasoning.
## Choose your focus area
Use the flowchart below to navigate to the article category that matches your interest. Each branch leads to a group of articles organized by theme.
```mermaid theme={null}
flowchart TD
Start[Start here] --> Q{What interests you?}
Q --> |AI agents| Agents[AI agent architectures]
Q --> |Multimodal & RAG| Multi[Multimodal & retrieval]
Q --> |Industry solutions| Industry[Industry applications]
Agents --> Memory[Scalable agent memory]
Agents --> Recipe[Recipe recommendation]
Multi --> Visual[Visual RAG]
Multi --> Product[Multimodal product discovery]
Industry --> Supply[Supply chain risk]
```
## AI agent architectures
These articles show how to build intelligent agents that combine semantic retrieval with domain-specific reasoning.
Build a scalable agent memory system with cross-collection lookup, retrieval sorted with OrderBy, WAL and optimizer tuning, and strict deletion.
Build a recipe recommendation agent that matches cravings through semantic search, filters by dietary restrictions and ingredients, and learns preferences over time.
## Multimodal and retrieval
These articles cover how to combine text, image, and document embeddings for rich retrieval experiences.
Build a multimodal document intelligence system that embeds PDF pages as images with CLIP and generates answers using GPT-4o vision.
Build a multimodal hybrid search system combining CLIP dense embeddings and BM25 sparse scoring for semantic and keyword product retrieval.
## Industry applications
These articles apply vector search to solve real-world problems across specific industries.
Build a supply chain risk intelligence workflow with semantic retrieval, payload filters, and a lightweight reasoning layer for stockout prediction.
## Article summary
The table below lists every article alongside its domain and the specific VectorAI DB features it covers, so you can find an article based on the capability you want to learn.
| Article | Domain | Key VectorAI DB features |
| ---------------------------------------------------------------------------------------------------------- | -------------- | --------------------------------------------------------------- |
| [Scalable agent memory](/academy/articles/building-a-scalable-agent-memory-with-Actian-vector-AI-database) | Infrastructure | Cross-collection, WAL tuning, optimizer config, strict deletion |
| [Recipe recommendation](/academy/articles/AI-recipe-recommendation-agent) | Consumer | Semantic search, payload filters, preference learning |
| [Visual RAG](/academy/articles/Multivector-Document-Intelligence-with-Visual-RAG) | Document AI | CLIP embeddings, multimodal retrieval, GPT-4o vision |
| [Multimodal product discovery](/academy/articles/Next-Gen-Product-Discovery-with-Multimodal-AI) | E-commerce | CLIP + BM25 hybrid search, sparse/dense fusion |
| [Supply chain risk](/academy/articles/supply-chain-inventory-management-agent) | Logistics | Semantic retrieval, payload filters, risk reasoning |
Each article is self-contained — pick the one that matches your use case and follow along. If you are new to VectorAI DB, then start with the [tutorials](/academy/tutorials/index) first to build foundational skills.
# AI supply chain inventory risk intelligence agent
Source: https://docs.vectoraidb.actian.com/academy/articles/supply-chain-inventory-management-agent
Build an AI-powered supply chain risk intelligence workflow using Actian VectorAI DB, semantic retrieval, payload filters, and lightweight reasoning.
Supply chains rarely fail because of a single obvious signal. More often, disruptions build gradually and combine into a hidden risk pattern:
* Supplier delays slow replenishment before stock runs out.
* Rising demand accelerates inventory depletion.
* Regional logistics issues block inbound shipments.
* Warehouse imbalances leave some locations critically understocked.
Traditional inventory systems track structured data well, but they are not designed to recognize semantically similar incidents across messy operational events.
In this tutorial, you will build an AI supply chain inventory risk intelligence agent. It uses Actian VectorAI DB to detect inventory risk patterns before they become stockouts. The system will:
* Ingest supply chain events and convert them into embeddings.
* Store events with structured metadata in Actian VectorAI DB.
* Retrieve similar historical incidents using semantic search and payload filters.
* Generate risk alerts using a lightweight reasoning layer.
## Architecture overview
The agent is built around four connected stages:
* **Ingestion pipeline** — Converts raw supply chain events into embeddings and stores them in Actian VectorAI DB.
* **Query pipeline** — Embeds incoming natural-language questions.
* **Retrieval layer** — Combines semantic search with payload filters to surface relevant historical incidents.
* **Risk reasoning layer** — Evaluates each result against five rule-based checks to generate actionable alerts.
The diagram below shows how supply chain events flow through embeddings, vector storage, semantic retrieval with payload filters, and the risk reasoning layer.
```mermaid theme={null}
flowchart TB
subgraph ingest [Ingestion Pipeline]
Events["Supply Chain Events - text + metadata"]
Embed["all-MiniLM-L6-v2 - 384-dim Embeddings"]
Store["Actian VectorAI DB - vector + payload"]
Events --> Embed --> Store
end
subgraph query [Query Pipeline]
UserQuery["User Query - natural language"]
QueryEmbed["all-MiniLM-L6-v2 - Query Embedding"]
UserQuery --> QueryEmbed
end
subgraph retrieval [Retrieval Layer]
SemanticSearch["Semantic Search - points.search"]
FilterDSL["Filter DSL - Field / FilterBuilder"]
QueryEmbed --> SemanticSearch
Store --> SemanticSearch
FilterDSL --> SemanticSearch
end
subgraph reasoning [Risk Reasoning Layer]
Results["Scored Results - ScoredPoint + payload"]
RiskEngine["Risk Engine - 5 rule-based checks"]
Alerts["Risk Alerts - severity + message + action"]
SemanticSearch --> Results --> RiskEngine --> Alerts
end
subgraph rules [Risk Rules]
R1["stock_below_reorder"]
R2["demand_spike"]
R3["supplier_delay"]
R4["quality_issue"]
R5["logistics_disruption"]
end
RiskEngine -.-> rules
```
## Environment setup
This tutorial requires Python, a sentence embedding model, and the Actian VectorAI Python SDK. Run the following command to install both required packages:
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
Each package serves a specific role in the pipeline:
* `actian-vectorai-client` — Official Python SDK for Actian VectorAI DB (async/sync clients, Filter DSL, gRPC transport).
* `sentence-transformers` — For generating text embeddings with `all-MiniLM-L6-v2`.
## Implementation
This section walks through the implementation steps for building the inventory risk intelligence workflow using Actian VectorAI DB.
### Step 1: Import dependencies and configure
The following block imports the Actian VectorAI client, the embedding model, and defines the connection endpoint and collection settings. Running it prints a confirmation that the configuration was loaded correctly.
```python theme={null}
# Core Actian VectorAI SDK — client, distance metric, filter helpers, and data types
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
Field,
FilterBuilder,
PointStruct,
VectorParams,
)
# HNSW index tuning parameters
from actian_vectorai.models.collections import HnswConfigDiff
# Sentence embedding model for converting event text to dense vectors
from sentence_transformers import SentenceTransformer
# gRPC endpoint for the Actian VectorAI server
SERVER = "localhost:6574"
# Name of the vector collection that stores supply chain events
COLLECTION = "Supply-Chain-Risk"
# Lightweight model that produces 384-dimensional embeddings
EMBED_MODEL = "all-MiniLM-L6-v2"
EMBED_DIM = 384
# Maximum number of similar events to retrieve per query
TOP_K = 5
# Load the embedding model once at startup to avoid repeated initialization overhead
model = SentenceTransformer(EMBED_MODEL)
print(f"VectorAI Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"Embedding model: {EMBED_MODEL} ({EMBED_DIM}-dim)")
```
Every component is configured upfront. The key settings are:
* VectorAI server - The Actian VectorAI gRPC endpoint (default port `6574`).
* Collection name — Identifies the vector collection for supply chain events.
* Embedding model — Converts event text into 384-dimensional dense vectors.
Running this configuration block prints the following confirmation:
```text theme={null}
VectorAI Server: localhost:6574
Collection: Supply-Chain-Risk
Embedding model: all-MiniLM-L6-v2 (384-dim)
```
### Step 2: Define embedding helpers
The following functions wrap the embedding model so the rest of the pipeline can convert event text to vectors with a single call. `embed_text` handles single strings; `embed_texts` processes a list of strings in one model pass, which is more efficient when ingesting batches.
```python theme={null}
def embed_text(text: str) -> list[float]:
"""Generate a 384-dimensional embedding for a text string."""
return model.encode(text).tolist()
def embed_texts(texts: list[str]) -> list[list[float]]:
"""Batch-embed multiple text strings in a single model pass."""
return model.encode(texts).tolist()
```
The embedding model turns natural-language event summaries into numerical representations that preserve semantic meaning. This allows the vector database to retrieve related incidents even when the wording is different. For example, these two event descriptions map to nearby vectors:
* "Supplier delay caused battery shortage risk."
* "Low battery stock after repeated replenishment delays."
### Step 3: Initialize the vector database collection
Collections in Actian VectorAI DB define the vector dimensionality, distance metric, and index parameters. The following code calls `get_or_create`, which is idempotent. It creates the collection if it does not exist and skips creation if it already does. Running this block prints a confirmation that the collection is ready.
```python theme={null}
import asyncio
async def ensure_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
# get_or_create is idempotent — safe to run on every startup
await client.collections.get_or_create(
name=COLLECTION,
# 384-dim cosine space matches the all-MiniLM-L6-v2 output
vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine),
# Higher m and ef_construct improve recall at the cost of build time
hnsw_config=HnswConfigDiff(m=32, ef_construct=256),
)
print(f"Collection '{COLLECTION}' ready.")
asyncio.run(ensure_collection())
```
This step creates a dedicated vector space for supply chain risk events. The configuration tells VectorAI DB the following:
* Vectors will have 384 dimensions.
* Similarity is computed with cosine distance.
* The HNSW index uses `m=32` connections and `ef_construct=256` for high recall.
Once the collection is ready, the following message is printed:
```text theme={null}
Collection 'Supply-Chain-Risk' ready.
```
### Step 4: Prepare sample supply chain events
The following block defines a dataset of realistic supply chain incidents. Each event includes an `event_text` field for semantic meaning and structured fields for payload filtering. Running it prints the number of events loaded.
```python theme={null}
events = [
{
# Natural-language summary used for embedding and semantic search
"event_text": "Supplier Alpha delayed two laptop battery shipments to Warehouse West while demand increased by 22 percent.",
"event_type": "supplier_delay",
"product": "laptop_battery",
"category": "electronics",
"supplier": "Supplier Alpha",
"warehouse": "Warehouse West",
"region": "Southeast Asia",
# Current and minimum acceptable stock thresholds
"stock_level": 18,
"reorder_point": 40,
"risk_level": "high",
"demand_change_pct": 22.0,
"created_at": "2026-03-10T09:30:00Z",
"location": {"lat": 13.7563, "lon": 100.5018},
},
{
"event_text": "Port congestion slowed inbound electronics shipments to Warehouse South.",
"event_type": "logistics_disruption",
"product": "microcontroller",
"category": "electronics",
"supplier": "Supplier Beta",
"warehouse": "Warehouse South",
"region": "South Asia",
"stock_level": 55,
"reorder_point": 35,
"risk_level": "medium",
"demand_change_pct": 8.0,
"created_at": "2026-03-08T12:00:00Z",
"location": {"lat": 6.9271, "lon": 79.8612},
},
{
"event_text": "Warehouse West reported critically low battery safety stock after repeated supplier delays.",
"event_type": "inventory_alert",
"product": "laptop_battery",
"category": "electronics",
"supplier": "Supplier Alpha",
"warehouse": "Warehouse West",
"region": "Southeast Asia",
"stock_level": 9,
"reorder_point": 40,
"risk_level": "high",
"demand_change_pct": 25.0,
"created_at": "2026-03-11T08:45:00Z",
"location": {"lat": 13.7563, "lon": 100.5018},
},
]
print(f"{len(events)} events loaded.")
```
Actian VectorAI DB stores rich payload metadata alongside vectors, making it possible to combine semantic similarity with operational filtering. Each event carries both unstructured text (for embeddings) and structured fields (for filters).
### Step 5: Embed and ingest events into VectorAI DB
The following code embeds each event description, packages it as a `PointStruct` with payload, and upserts all points into the collection in a single operation. Running it prints the number of events ingested and the updated total stored in the collection.
```python theme={null}
async def ingest_events(events):
texts = [e["event_text"] for e in events]
# Batch-embed all event texts in one model call for efficiency
vectors = embed_texts(texts)
async with AsyncVectorAIClient(url=SERVER) as client:
# Offset new IDs by the existing count to avoid collisions on repeated runs
existing = await client.vde.get_vector_count(COLLECTION)
points = []
for i, (event, vector) in enumerate(zip(events, vectors)):
payload = {**event}
# VectorAI DB stores geo coordinates as flat fields, not nested objects
if "location" in payload and payload["location"] is not None:
loc = payload.pop("location")
payload["lat"] = loc["lat"]
payload["lon"] = loc["lon"]
points.append(
PointStruct(
id=existing + i,
vector=vector,
payload=payload,
)
)
await client.points.upsert(COLLECTION, points=points)
# Flush immediately so vectors are queryable without waiting for background persistence
await client.vde.flush(COLLECTION)
total = await client.vde.get_vector_count(COLLECTION)
print(f"Ingested {len(points)} events. Total in collection: {total}")
asyncio.run(ingest_events(events))
```
Each supply chain event becomes a searchable point in VectorAI DB. The three fields that make up each point are:
* `id` — Sequential integer identifier.
* `vector` — 384-dim dense embedding from `all-MiniLM-L6-v2`.
* `payload` — All structured metadata (category, supplier, stock level, region, etc.).
The `vde.flush()` call ensures vectors are persisted to disk immediately. After ingestion, the pipeline prints the following confirmation:
```text theme={null}
Ingested 3 events. Total in collection: 3
```
### Step 6: Run basic semantic search
The following code embeds a natural-language inventory risk query and uses `points.search` to retrieve the most semantically similar events from the collection. Running it prints each result's ID, similarity score, and a preview of the event text.
```python theme={null}
async def semantic_search(query: str, top_k: int = TOP_K):
# Embed the query using the same model used during ingestion
query_vector = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
# Return the full payload so the caller can inspect event metadata
with_payload=True,
) or []
return results
query = "Laptop battery stock is falling while the supplier is delayed and demand is rising."
results = asyncio.run(semantic_search(query))
for r in results:
print(f" id={r.id} score={r.score:.4f} event={r.payload['event_text'][:80]}...")
```
This is the semantic search core of the system. The `points.search` method accepts the following parameters:
* `vector` — The query embedding.
* `limit` — Number of results.
* `with_payload` — Whether to return metadata.
Results are ranked by cosine similarity. The search returns the three ingested events in the following order:
```text theme={null}
id=0 score=0.9200 event=Supplier Alpha delayed two laptop battery shipments to Warehouse West whi...
id=2 score=0.8900 event=Warehouse West reported critically low battery safety stock after repeated...
id=1 score=0.5100 event=Port congestion slowed inbound electronics shipments to Warehouse South...
```
### Step 7: Apply structured payload filters using the Filter DSL
Actian VectorAI provides a type-safe `Field` / `FilterBuilder` API for payload filtering. The following code adds a server-side filter that restricts results to electronics events with stock below 20 before ranking by similarity. Running it prints only the two events that pass both the category and stock-level filters.
```python theme={null}
async def filtered_search(query: str, category: str, stock_below: int, top_k: int = TOP_K):
query_vector = embed_text(query)
# Build a server-side filter: only return electronics with stock below the threshold
filter_obj = (
FilterBuilder()
.must(Field("category").eq(category))
.must(Field("stock_level").lt(float(stock_below)))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
filter=filter_obj,
) or []
return results
results = asyncio.run(filtered_search(query, category="electronics", stock_below=20))
for r in results:
print(f" id={r.id} score={r.score:.4f} stock={r.payload['stock_level']} product={r.payload['product']}")
```
This combines semantic search with Actian VectorAI's Filter DSL. The three filter expressions used here are:
* `Field("category").eq("electronics")` — Exact match filter.
* `Field("stock_level").lt(20.0)` — Numeric range filter.
* `FilterBuilder().must(...)` — AND logic.
The filter is applied server-side before ranking, so only matching points are considered. With the category and stock filters applied, only the two low-stock laptop battery events are returned:
```text theme={null}
id=0 score=0.9200 stock=18 product=laptop_battery
id=2 score=0.8900 stock=9 product=laptop_battery
```
### Step 8: Add boolean logic with must, should, and must\_not
The Filter DSL supports `must` (AND), `should` (OR/preference), and `must_not` (exclusion) for complex business queries. The following code demonstrates all three clause types in a single filter. Running it returns only the Supplier Alpha events that match the low-stock electronics criteria, with the deprecated-region events excluded.
```python theme={null}
async def boolean_filtered_search(query: str, top_k: int = TOP_K):
query_vector = embed_text(query)
filter_obj = (
FilterBuilder()
# must: ALL of these conditions must match
.must(Field("category").eq("electronics"))
.must(Field("stock_level").lt(20.0))
# should: prefer events from Supplier Alpha (boosts their score)
.should(Field("supplier").eq("Supplier Alpha"))
# must_not: exclude events from regions marked as deprecated
.must_not(Field("region").eq("Deprecated Region"))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
filter=filter_obj,
) or []
return results
results = asyncio.run(boolean_filtered_search(query))
for r in results:
print(f" id={r.id} score={r.score:.4f} risk={r.payload['risk_level']} supplier={r.payload['supplier']}")
```
The Filter DSL supports three clause types, each with different matching behavior:
* `.must()` — All conditions must match (AND logic). Used here to require `category = electronics` and `stock_level < 20`.
* `.should()` — Preference boost. Events from `Supplier Alpha` are ranked higher but not excluded if absent.
* `.must_not()` — Hard exclusion. Events from `Deprecated Region` are removed from results entirely.
This lets the agent answer realistic business questions such as: find low-stock electronics events, prefer Supplier Alpha, and exclude deprecated regions. The boolean filter keeps only the high-risk Supplier Alpha events:
```text theme={null}
id=0 score=0.9200 risk=high supplier=Supplier Alpha
id=2 score=0.8900 risk=high supplier=Supplier Alpha
```
### Step 9: Build the hybrid inventory risk query
This step combines semantic search with multiple filter dimensions — category, stock level, risk, and supplier — into a single reusable function. Each filter parameter is optional, so the function adapts to different query scenarios without code changes. Running the example call returns the high-risk, low-stock electronics events that are semantically closest to the query.
```python theme={null}
async def hybrid_risk_search(
query: str,
category: str = None,
supplier: str = None,
risk_level: str = None,
stock_below: int = None,
event_type: str = None,
top_k: int = TOP_K,
):
query_vector = embed_text(query)
# Build filter dynamically — only add clauses for parameters that were provided
fb = FilterBuilder()
if category:
fb = fb.must(Field("category").eq(category))
if supplier:
fb = fb.must(Field("supplier").eq(supplier))
if risk_level:
fb = fb.must(Field("risk_level").eq(risk_level))
if stock_below is not None:
fb = fb.must(Field("stock_level").lt(float(stock_below)))
if event_type:
fb = fb.must(Field("event_type").eq(event_type))
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
filter=filter_obj,
) or []
return results
results = asyncio.run(hybrid_risk_search(
query,
category="electronics",
stock_below=20,
risk_level="high",
))
for r in results:
print(f" id={r.id} score={r.score:.4f} product={r.payload['product']} stock={r.payload['stock_level']}")
```
Hybrid retrieval combines vector similarity with structured constraints to deliver results that are both semantically relevant and operationally valid. The hybrid query narrows results to high-risk, low-stock electronics events matching the query:
```text theme={null}
id=0 score=0.9200 product=laptop_battery stock=18
id=2 score=0.8900 product=laptop_battery stock=9
```
### Step 10: Build the risk reasoning layer
Retrieval alone is not enough. The following function adds a rule-based reasoning layer that evaluates each retrieved event's payload and returns a list of structured risk alerts. Each alert includes a rule name, severity, recommended message, and action.
```python theme={null}
def assess_risk(payload: dict) -> list[dict]:
"""Run all risk rules against an event payload."""
alerts = []
stock = payload.get("stock_level")
reorder = payload.get("reorder_point")
if stock is not None and reorder is not None and stock < reorder:
pct_below = round((1 - stock / reorder) * 100, 1)
severity = "critical" if pct_below >= 50 else "warning"
alerts.append({
"rule": "stock_below_reorder",
"severity": severity,
"message": f"Stock ({stock}) is {pct_below}% below reorder point ({reorder}).",
"action": "Expedite reorder or activate backup supplier.",
})
change = payload.get("demand_change_pct", 0.0)
if change >= 20:
severity = "critical" if change >= 30 else "warning"
alerts.append({
"rule": "demand_spike",
"severity": severity,
"message": f"Demand surged by {change}%.",
"action": "Increase safety stock and review forecast.",
})
if payload.get("event_type") == "supplier_delay":
risk = payload.get("risk_level", "low")
severity = "critical" if risk == "high" else "warning"
alerts.append({
"rule": "supplier_delay",
"severity": severity,
"message": f"Supplier '{payload.get('supplier', 'unknown')}' has reported delays.",
"action": "Contact supplier for revised ETA or switch to alternate source.",
})
if payload.get("event_type") == "quality_issue":
alerts.append({
"rule": "quality_issue",
"severity": "warning",
"message": f"Quality issue for '{payload.get('product')}' from '{payload.get('supplier')}'.",
"action": "Hold incoming batch and schedule re-inspection.",
})
if payload.get("event_type") == "logistics_disruption":
severity = "critical" if payload.get("risk_level") == "high" else "warning"
alerts.append({
"rule": "logistics_disruption",
"severity": severity,
"message": f"Logistics disruption affecting '{payload.get('product')}' at {payload.get('warehouse')}.",
"action": "Reroute shipments or activate contingency logistics partner.",
})
return alerts
```
This is where the system becomes an agent rather than a search tool. The risk engine runs five rules against each event, as shown in the table below:
| Rule | Trigger | Severity |
| ---------------------- | ------------------------------------ | -------------------------------------- |
| `stock_below_reorder` | Stock \< reorder point | Critical if >= 50% below, else warning |
| `demand_spike` | Demand change >= 20% | Critical if >= 30%, else warning |
| `supplier_delay` | Event type is `supplier_delay` | Based on risk level |
| `quality_issue` | Event type is `quality_issue` | Always warning |
| `logistics_disruption` | Event type is `logistics_disruption` | Based on risk level |
### Step 11: Run the end-to-end flow
The following code connects all the pieces into a single pipeline function and runs it with a sample query. Calling `run_risk_intelligence` performs a hybrid semantic search, then runs the risk reasoning layer on every result and prints all triggered alerts with their severity and recommended action.
```python theme={null}
async def run_risk_intelligence(query: str, **filters):
"""Full pipeline: search + risk assessment."""
results = await hybrid_risk_search(query, **filters)
print(f"\nQuery: {query}")
print(f"Filters: {filters}")
print(f"Results found: {len(results)}\n")
for r in results:
payload = r.payload or {}
# Run all five risk rules against this event's payload
alerts = assess_risk(payload)
print(f"--- Event id={r.id} score={r.score:.4f} ---")
print(f" Text: {payload.get('event_text', '')[:100]}")
print(f" Product: {payload.get('product')} | Stock: {payload.get('stock_level')} | Risk: {payload.get('risk_level')}")
if alerts:
print(f" Alerts ({len(alerts)}):")
for a in alerts:
print(f" [{a['severity'].upper()}] {a['rule']}: {a['message']}")
print(f" -> {a['action']}")
else:
print(" No alerts triggered.")
print()
asyncio.run(run_risk_intelligence(
"Laptop battery stock is falling while the supplier is delayed and demand is rising.",
category="electronics",
stock_below=20,
))
```
The end-to-end pipeline prints the query, applied filters, and all risk alerts for each matched event:
```text theme={null}
Query: Laptop battery stock is falling while the supplier is delayed and demand is rising.
Filters: {'category': 'electronics', 'stock_below': 20}
Results found: 2
--- Event id=0 score=0.9200 ---
Text: Supplier Alpha delayed two laptop battery shipments to Warehouse West while demand increased by
Product: laptop_battery | Stock: 18 | Risk: high
Alerts (3):
[CRITICAL] stock_below_reorder: Stock (18) is 55.0% below reorder point (40).
-> Expedite reorder or activate backup supplier.
[WARNING] demand_spike: Demand surged by 22.0%.
-> Increase safety stock and review forecast.
[CRITICAL] supplier_delay: Supplier 'Supplier Alpha' has reported delays.
-> Contact supplier for revised ETA or switch to alternate source.
--- Event id=2 score=0.8900 ---
Text: Warehouse West reported critically low battery safety stock after repeated supplier delays.
Product: laptop_battery | Stock: 9 | Risk: high
Alerts (2):
[CRITICAL] stock_below_reorder: Stock (9) is 77.5% below reorder point (40).
-> Expedite reorder or activate backup supplier.
[WARNING] demand_spike: Demand surged by 25.0%.
-> Increase safety stock and review forecast.
```
### Step 12: Retrieve a specific event by ID for risk assessment
Actian VectorAI DB supports retrieving points by ID using `points.get`, which is useful for inspecting individual events without running a vector search. The following code fetches event ID `0` and runs the risk reasoning layer against it, printing the event summary and any triggered alerts.
```python theme={null}
async def assess_event_by_id(event_id: int):
"""Retrieve a specific event and run risk assessment."""
async with AsyncVectorAIClient(url=SERVER) as client:
# Fetch a single point by its integer ID without a vector search
points = await client.points.get(
COLLECTION,
ids=[event_id],
with_payload=True,
)
if not points:
print(f"Event {event_id} not found.")
return
payload = points[0].payload or {}
alerts = assess_risk(payload)
print(f"Event {event_id}: {payload.get('event_text', '')[:100]}")
print(f"Risk Level: {payload.get('risk_level')}")
for a in alerts:
print(f" [{a['severity'].upper()}] {a['rule']}: {a['message']}")
asyncio.run(assess_event_by_id(0))
```
Direct point retrieval via `points.get` allows the system to inspect specific events without a vector search, which is useful for dashboards and audit trails.
### Step 13: Collection administration
Actian VectorAI provides Vector Data Engine (VDE) operations for managing collections. Use `get_vector_count` to check collection size and `flush` to persist data to disk (already shown in step 5). To remove a collection entirely, call `client.collections.delete(COLLECTION)`.
## Actian VectorAI features used
The following table summarizes every Actian VectorAI DB API used in this tutorial and the role each one plays:
| Feature | API | Purpose |
| ------------------- | ------------------------------------------------- | ------------------------------------------- |
| Collection creation | `client.collections.get_or_create()` | Create vector space with HNSW config |
| Point upsert | `client.points.upsert()` | Store vectors with payload metadata |
| Semantic search | `client.points.search()` | Nearest-neighbour retrieval |
| Filtered search | `client.points.search(filter=...)` | Combine similarity with payload constraints |
| Filter DSL | `Field().eq()`, `.lt()`, `FilterBuilder().must()` | Type-safe filter construction |
| Point retrieval | `client.points.get()` | Fetch specific events by ID |
| Vector count | `client.vde.get_vector_count()` | Collection statistics |
| Flush | `client.vde.flush()` | Persist vectors to disk |
| Delete collection | `client.collections.delete()` | Clean up |
## Conclusion
This tutorial built an AI supply chain inventory risk intelligence agent using Actian VectorAI DB as the retrieval engine.
The full pipeline covered the following steps:
* Create a collection with `VectorParams` and `HnswConfigDiff`.
* Embed supply chain events with `all-MiniLM-L6-v2` (384-dim).
* Store vectors with rich payload metadata via `PointStruct`.
* Run semantic search with `points.search`.
* Refine results with the type-safe `Field` / `FilterBuilder` DSL.
* Retrieve specific events by ID with `points.get`.
* Apply a rule-based risk reasoning layer.
* Generate actionable inventory risk alerts.
This pattern is a strong fit for vector databases. Supply chain failures are rarely caused by one keyword or one threshold crossing. They emerge from combinations of semantically related events, metadata, recency, and location. Actian VectorAI's semantic retrieval and payload filter DSL let you detect these risk patterns before they turn into costly disruptions.
## Next steps
Explore these related tutorials to deepen your understanding of the Actian VectorAI DB features used in this workflow:
Improve relevance with cross-encoder and reciprocal rank fusion reranking
Learn the core similarity search workflow
Combine vector search with structured payload constraints
Measure and optimize search accuracy using precision, recall, and MRR
# Overview
Source: https://docs.vectoraidb.actian.com/academy/index
Learn Actian VectorAI DB through hands-on tutorials, deep-dive articles, and ready-to-run examples.
The VectorAI DB Academy covers building vector search applications and AI agents. Whether you are getting started with your first collection or building an AI agent, the Academy has a path for you.
## Choose your path
The diagram below shows three learning paths branching from a single entry point: tutorials for step-by-step guidance, articles for real-world architectures, and examples for runnable code. Follow the branch that matches your current goal.
```mermaid theme={null}
%% Learning path: each branch leads to a section of the Academy
flowchart TD
Start[Start here] --> Q{What do you need?}
Q --> |Step-by-step guidance| T["Tutorials\n8 hands-on walkthroughs"]
Q --> |Real-world architectures| A["Articles\n5 deep-dive implementations"]
T --> T1[Build your first app]
T --> T2[Search, filters, RAG]
T --> T3[Reranking, multimodal, adaptive RAG]
A --> A1[AI agent architectures]
A --> A2[Multimodal & retrieval]
A --> A3[Industry applications]
```
***
## Tutorials
Structured, step-by-step walkthroughs that teach VectorAI DB skills progressively. Each tutorial builds on the last, taking you from basic operations to advanced retrieval architectures.
Learn how to connect to VectorAI DB, store your first vectors, and run a semantic search query.
Learn how to search, score, batch, and paginate vector query results effectively.
Learn how to combine vector search with structured payload filters to narrow results.
Learn how to integrate open-source models like Sentence Transformers and BGE into your pipeline.
Learn how to fuse text, image, and metadata embeddings using named vectors.
Learn how to improve relevance with cross-encoder and reciprocal rank fusion reranking.
Learn how to measure and optimize search accuracy using precision, recall, and MRR.
Build RAG pipelines that automatically adapt their retrieval strategy based on query complexity.
See the full tutorial overview with a recommended learning order and time estimates.
***
## Articles
Deep-dive implementations of AI agents and real-world applications. Each article walks through a complete architecture, covering topics such as data modeling, retrieval strategies, and agent reasoning.
Build persistent agent memory with cross-collection lookup, WAL tuning, optimizer configuration, and strict deletion.
Build a visual document intelligence system using CLIP embeddings, multimodal retrieval, and GPT-4o vision.
Build a personalized recipe recommendation agent using semantic search, payload filters, and preference learning.
Build a product discovery system using CLIP and BM25 hybrid search with sparse and dense score fusion.
Build a supply chain risk agent using semantic retrieval, payload filters, and a reasoning layer.
See the full article overview organized by category with a feature summary table.
***
## Where to start
The table below maps common goals to the most relevant starting point in the Academy. Each link takes you directly to the tutorial, article, or example that best fits that goal.
| Your goal | Start here |
| ---------------------------- | ---------------------------------------------------------------------------------------------------------- |
| New to VectorAI DB | [Build your first application](/academy/tutorials/first-application) |
| Need to add search to an app | [Similarity search](/academy/tutorials/similarity-search) |
| Designing an AI agent | [Scalable agent memory](/academy/articles/building-a-scalable-agent-memory-with-Actian-vector-AI-database) |
| Working with images and text | [Multimodal systems](/academy/tutorials/multimodel-system) |
| Optimizing search quality | [Retrieval quality](/academy/tutorials/retrieval-quality) |
If you are new to vector databases, start with the tutorials — they build skills progressively from beginner to advanced. Articles are best when you have a specific use case in mind and want to see a complete implementation. Use examples when you need runnable code you can clone and adapt right away.
# Adaptive RAG systems
Source: https://docs.vectoraidb.actian.com/academy/tutorials/adaptive-rag
Learn how to build a Retrieval-Augmented Generation system that adapts its retrieval strategy at runtime based on query type, confidence signals, and user feedback—using Actian VectorAI DB's multistage prefetch, fusion, score thresholds, payload-driven routing, and feedback loops.
Standard RAG is a fixed pipeline: embed the query, search the vector database, stuff the top-K results into a prompt, and call the LLM. This works for simple factual questions but fails in practice because:
* Not all queries need retrieval. "What is 2 + 2?" should skip the vector database entirely. Sending it through retrieval wastes latency and may inject irrelevant context.
* Different queries need different retrieval strategies. A factual lookup ("What is the capital of France?") needs high-precision single-pass search. An exploratory question ("How does authentication work in the system?") needs broad multistage retrieval across multiple document types.
* Retrieval confidence varies. If the top result has a score of 0.92, the LLM probably has enough context. If the best score is 0.35, the system should either try a different search strategy or tell the user it does not know.
* User feedback should improve future retrieval. When a user marks a response as unhelpful, the system should learn which documents were not relevant.
An *Adaptive RAG* system solves these problems by making the retrieval pipeline dynamic. Instead of one fixed strategy, the system classifies each query, selects the appropriate retrieval approach, evaluates result quality, and adapts based on feedback.
This tutorial builds a complete adaptive RAG pipeline on Actian VectorAI DB. By the end, you will have:
* A knowledge base collection with payload indexes for routing, feedback, and analytics.
* A keyword-signal query classifier that maps queries to four retrieval strategies.
* Three retrieval strategies (precise, broad multistage, and nested troubleshooting prefetch) plus an automatic fallback.
* A confidence evaluator that decides whether results are good enough or a fallback is needed.
* A user feedback loop that updates per-document usefulness scores over time.
* A feedback-aware retrieval function that boosts historically helpful documents.
* An analytics function that shows which documents are most retrieved and most useful.
* A prompt-assembly step that packages context and confidence instructions for any LLM.
***
## Architecture overview
The following diagram shows how queries flow through the adaptive RAG pipeline, from classification through strategy selection, confidence evaluation, and the feedback loop back into the knowledge base:
```mermaid theme={null}
flowchart LR
Query[User Query] --> Classify[Query Classifier]
Classify --> Router{Route}
Router -->|factual| Precise[Precise Search - hnsw_ef=256, threshold=0.6]
Router -->|exploratory| Broad[Broad Multi-Stage - prefetch + fusion]
Router -->|no_retrieval| Direct[Direct LLM Response]
Precise --> Evaluate[Confidence Evaluator]
Broad --> Evaluate
Evaluate -->|confident| Generate[Generate Answer]
Evaluate -->|low_confidence| Fallback[Fallback Strategy]
Fallback --> Generate
Generate --> Feedback[User Feedback Loop]
Feedback --> KB[(Actian VectorAI DB - Knowledge Base)]
KB --> Precise
KB --> Broad
```
***
## Environment setup
Run the following command to install the Actian VectorAI SDK and the sentence-transformers library used for embedding:
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
***
## Step 1: Import dependencies and configure the environment
The following block imports all SDK symbols used throughout the tutorial, loads the `all-MiniLM-L6-v2` embedding model, and defines three constants (`SERVER`, `COLLECTION`, `EMBED_DIM`) that every subsequent step shares. Running it prints the active configuration so you can confirm the setup before proceeding:
```python theme={null}
import asyncio
from datetime import datetime, timezone
from enum import Enum
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
Field,
FieldType,
FilterBuilder,
FloatIndexParams,
IntegerIndexParams,
DatetimeIndexParams,
PointStruct,
PrefetchQuery,
SearchParams,
VectorParams,
reciprocal_rank_fusion,
)
from actian_vectorai.models.collections import HnswConfigDiff
from actian_vectorai.models.enums import Direction, Fusion, Sample
from actian_vectorai.models.points import (
OrderBy,
ScoredPoint,
WithPayloadSelector,
)
SERVER = "localhost:6574"
COLLECTION = "Adaptive-RAG"
EMBED_DIM = 384
model = SentenceTransformer("all-MiniLM-L6-v2")
def embed_text(text: str) -> list[float]:
return model.encode(text).tolist()
def embed_texts(texts: list[str]) -> list[list[float]]:
return model.encode(texts).tolist()
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
print(f"Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"Embedding: all-MiniLM-L6-v2 ({EMBED_DIM}-dim)")
```
### Expected output
This block loads the `all-MiniLM-L6-v2` sentence-transformer model and defines the three shared constants—`SERVER`, `COLLECTION`, and `EMBED_DIM`—that every subsequent step references. The three `print` calls confirm the active server address, the target collection name, and the embedding model with its vector dimensionality, so you can verify the configuration is correct before proceeding.
```text theme={null}
Server: localhost:6574
Collection: Adaptive-RAG
Embedding: all-MiniLM-L6-v2 (384-dim)
```
***
## Step 2: Create the knowledge base collection
The following block creates the `Adaptive-RAG` collection with a cosine-distance HNSW index and registers six payload field indexes. Running it prints a confirmation message when the collection and all indexes are ready:
```python theme={null}
async def create_knowledge_base():
async with AsyncVectorAIClient(url=SERVER) as client:
await client.collections.get_or_create(
name=COLLECTION,
vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine),
hnsw_config=HnswConfigDiff(m=16, ef_construct=128),
)
# Keyword indexes for routing and source filtering
await client.points.create_field_index(
COLLECTION, field_name="doc_type",
field_type=FieldType.FieldTypeKeyword,
)
await client.points.create_field_index(
COLLECTION, field_name="source",
field_type=FieldType.FieldTypeKeyword,
)
await client.points.create_field_index(
COLLECTION, field_name="section",
field_type=FieldType.FieldTypeKeyword,
)
# Numeric indexes for analytics and feedback-aware boosting
await client.points.create_field_index(
COLLECTION, field_name="retrieval_count",
field_type=FieldType.FieldTypeInteger,
field_index_params=IntegerIndexParams(range=True),
)
await client.points.create_field_index(
COLLECTION, field_name="usefulness_score",
field_type=FieldType.FieldTypeFloat,
field_index_params=FloatIndexParams(is_principal=True),
)
# Datetime index for time-based range queries
await client.points.create_field_index(
COLLECTION, field_name="created_at",
field_type=FieldType.FieldTypeDatetime,
field_index_params=DatetimeIndexParams(is_principal=True),
)
print(f"Knowledge base '{COLLECTION}' ready.")
asyncio.run(create_knowledge_base())
```
Each index serves a specific role in the adaptive pipeline. The table below explains what each field enables:
| Field | Purpose in adaptive RAG |
| ------------------ | -------------------------------------------------------------- |
| `doc_type` | Route different query types to different document categories. |
| `source` | Filter by origin (API docs vs. tutorials vs. changelogs). |
| `section` | Narrow retrieval to specific parts of the documentation. |
| `retrieval_count` | Track which documents are retrieved frequently. |
| `usefulness_score` | Boost or demote documents based on user feedback. |
| `created_at` | Enable time-based range queries and filtering on document age. |
The combination of keyword, integer, float, and datetime indexes means the adaptive router can filter, sort, and range-query on any payload field without a full collection scan.
***
## Step 3: Ingest documents into the knowledge base
The following block defines 20 sample documents across five categories (API reference, tutorials, conceptual guides, troubleshooting, and changelog) and upserts them into the collection. Each point is assigned initial metadata values: `retrieval_count: 0`, `usefulness_score: 0.5`, and a UTC timestamp. Running it prints the total number of documents confirmed in the collection:
```python theme={null}
documents = [
# API reference
{"text": "The create_collection method accepts vectors_config, hnsw_config, wal_config, and quantization_config parameters to initialize a new collection.", "doc_type": "api_reference", "source": "sdk_docs", "section": "collections"},
{"text": "points.search performs approximate nearest-neighbour search. It accepts vector, limit, filter, params, score_threshold, and offset parameters.", "doc_type": "api_reference", "source": "sdk_docs", "section": "search"},
{"text": "points.query is the universal endpoint supporting vector search, fusion, order_by, and multistage prefetch queries.", "doc_type": "api_reference", "source": "sdk_docs", "section": "search"},
{"text": "FilterBuilder supports must, should, must_not, and min_should for combining conditions. Field provides eq, any_of, except_of, gt, gte, lt, lte, between, and text methods.", "doc_type": "api_reference", "source": "sdk_docs", "section": "filters"},
{"text": "SearchParams allows setting hnsw_ef for accuracy tuning, exact for brute-force search, and quantization for compressed vector search.", "doc_type": "api_reference", "source": "sdk_docs", "section": "search"},
# Tutorials
{"text": "To build a RAG pipeline, first create a collection, embed your documents with a sentence transformer, upsert the vectors with payload metadata, and search at query time.", "doc_type": "tutorial", "source": "academy", "section": "getting_started"},
{"text": "Hybrid search combines dense vector similarity with sparse keyword matching. Use reciprocal_rank_fusion or distribution_based_score_fusion to merge results.", "doc_type": "tutorial", "source": "academy", "section": "hybrid_search"},
{"text": "Named vectors allow storing multiple embedding spaces per collection. Use vectors_config as a dictionary to define each space with its own dimensionality and distance metric.", "doc_type": "tutorial", "source": "academy", "section": "multimodal"},
{"text": "Prefetch queries retrieve candidates from multiple vector spaces or filter conditions, then a fusion stage merges and reranks the results.", "doc_type": "tutorial", "source": "academy", "section": "prefetch"},
{"text": "Score thresholds discard low-confidence results. Set score_threshold on search or query to filter out results below a minimum similarity.", "doc_type": "tutorial", "source": "academy", "section": "search_tuning"},
# Conceptual guides
{"text": "HNSW is a graph-based index where each node connects to M neighbours. Higher M improves recall at the cost of memory. ef_construct controls build-time search width.", "doc_type": "concept", "source": "guides", "section": "indexing"},
{"text": "Cosine distance measures the angle between vectors and is ideal for normalized embeddings. Dot product is equivalent to cosine for unit vectors.", "doc_type": "concept", "source": "guides", "section": "distance_metrics"},
{"text": "Scalar quantization compresses 32-bit floats to 8-bit integers, reducing memory by 4x. Use rescore=True and oversampling to recover accuracy.", "doc_type": "concept", "source": "guides", "section": "quantization"},
{"text": "Payload indexes accelerate filtered searches. Keyword indexes support exact match and any_of. Integer and float indexes support range queries.", "doc_type": "concept", "source": "guides", "section": "payload_indexes"},
# Troubleshooting
{"text": "If search returns empty results, check that the collection has vectors (vde.get_vector_count), that your filter is not too restrictive, and that you flushed after upserting.", "doc_type": "troubleshooting", "source": "faq", "section": "empty_results"},
{"text": "If recall is low, increase hnsw_ef at search time or rebuild the index with higher m and ef_construct values.", "doc_type": "troubleshooting", "source": "faq", "section": "low_recall"},
{"text": "If latency is high, reduce hnsw_ef, enable quantization, or decrease the limit parameter. Use connection pooling for concurrent access.", "doc_type": "troubleshooting", "source": "faq", "section": "high_latency"},
# Changelog
{"text": "Version 2.5 added the universal query endpoint with prefetch, fusion, and order_by support.", "doc_type": "changelog", "source": "releases", "section": "v2.5"},
{"text": "Version 2.4 introduced SmartBatcher for streaming ingestion with automatic size, byte, and time-based flush triggers.", "doc_type": "changelog", "source": "releases", "section": "v2.4"},
{"text": "Version 2.3 added scalar quantization with rescore and oversampling for memory-efficient search.", "doc_type": "changelog", "source": "releases", "section": "v2.3"},
]
async def ingest_documents():
texts = [d["text"] for d in documents]
vectors = embed_texts(texts)
points = []
for i, (doc, vector) in enumerate(zip(documents, vectors)):
points.append(PointStruct(
id=i,
vector=vector,
payload={
**doc,
"created_at": now_iso(),
"retrieval_count": 0,
"usefulness_score": 0.5, # Neutral starting score
"feedback_count": 0,
},
))
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.upsert(COLLECTION, points=points)
await client.vde.flush(COLLECTION)
count = await client.vde.get_vector_count(COLLECTION)
print(f"Ingested {len(points)} documents. Total: {count}")
asyncio.run(ingest_documents())
```
### Expected output
This block embeds all 20 document texts in a single batch call and upserts them as `PointStruct` objects, each carrying its source metadata alongside the initial tracking fields (`retrieval_count: 0`, `usefulness_score: 0.5`, `feedback_count: 0`). After upserting, it calls `flush` to persist the writes to disk and then queries `get_vector_count` to confirm the exact number of vectors now stored in the collection.
```text theme={null}
Ingested 20 documents. Total: 20
```
***
## Step 4: Build the query classifier
The classifier inspects keyword signals in a query and returns a `ClassifiedQuery` that names the query type and the target document categories to search. The following block defines the `QueryType` enum, the `ClassifiedQuery` dataclass, and the `classify_query` function, then runs it against four test queries and prints the assigned type and confidence for each:
```python theme={null}
class QueryType(Enum):
FACTUAL = "factual"
EXPLORATORY = "exploratory"
TROUBLESHOOTING = "troubleshooting"
NO_RETRIEVAL = "no_retrieval"
@dataclass
class ClassifiedQuery:
original: str
query_type: QueryType
target_doc_types: list[str]
confidence: float
def classify_query(query: str) -> ClassifiedQuery:
"""Classify a query to determine retrieval strategy.
In production, replace this keyword-signal approach with an LLM-based
classifier or a fine-tuned text classification model for higher accuracy.
"""
q = query.lower()
# Greetings and arithmetic do not benefit from document retrieval
no_retrieval_signals = [
"hello", "hi ", "thanks", "thank you",
"what is 2", "calculate", "what time",
]
if any(sig in q for sig in no_retrieval_signals):
return ClassifiedQuery(query, QueryType.NO_RETRIEVAL, [], 0.95)
# Error reports and "why/how to fix" patterns indicate a troubleshooting intent
troubleshooting_signals = [
"error", "not working", "empty results", "slow",
"fails", "issue", "problem", "bug", "fix",
"why is", "how to fix", "doesn't work",
]
if any(sig in q for sig in troubleshooting_signals):
return ClassifiedQuery(
query, QueryType.TROUBLESHOOTING,
["troubleshooting", "api_reference"], 0.85,
)
# "What is / what does / how to use" patterns point to a factual API lookup
factual_signals = [
"what is", "what does", "how to use", "what parameters",
"which method", "api for", "syntax for", "default value",
]
if any(sig in q for sig in factual_signals):
return ClassifiedQuery(
query, QueryType.FACTUAL,
["api_reference", "concept"], 0.80,
)
# Everything else is treated as an open-ended exploratory question
return ClassifiedQuery(
query, QueryType.EXPLORATORY,
["tutorial", "concept", "api_reference"], 0.70,
)
test_queries = [
"How to use the search method?",
"How does the prefetch pipeline work with hybrid search?",
"My search returns empty results, what's wrong?",
"Hello, how are you?",
]
for q in test_queries:
c = classify_query(q)
print(f" {c.query_type.value:>17} conf={c.confidence:.2f} {q}")
```
### Expected output
The classifier inspects each query for keyword signals and maps it to one of four `QueryType` values. The four test queries are designed to exercise every branch: a "how to use" phrase triggers `factual`, an open-ended "how does" triggers `exploratory`, an error-related phrase triggers `troubleshooting`, and a greeting triggers `no_retrieval`. Each line of output shows the assigned type right-aligned, the classifier's confidence score, and the original query text.
```text theme={null}
factual conf=0.80 How to use the search method?
exploratory conf=0.70 How does the prefetch pipeline work with hybrid search?
troubleshooting conf=0.85 My search returns empty results, what's wrong?
no_retrieval conf=0.95 Hello, how are you?
```
***
## Step 5: Strategy 1—Precise retrieval for factual queries
Factual queries require high precision. The following block defines `precise_retrieval`, which searches only within the specified document type categories, applies a `score_threshold` of 0.5 to discard low-similarity results, and uses `hnsw_ef=256` to maximise recall accuracy. Running the test query prints each result's score, document type, and a truncated text preview:
```python theme={null}
async def precise_retrieval(query: str, doc_types: list[str], top_k: int = 3) -> list[ScoredPoint]:
vec = embed_text(query)
fb = FilterBuilder()
if doc_types:
# Restrict results to the document categories appropriate for factual queries
fb = fb.must(Field("doc_type").any_of(doc_types))
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=vec,
limit=top_k,
filter=fb.build(),
score_threshold=0.5, # Drop results below cosine 0.5
params=SearchParams(hnsw_ef=256), # High ef for maximum accuracy
with_payload=True,
) or []
return results
query = "What parameters does the search method accept?"
results = asyncio.run(precise_retrieval(query, ["api_reference", "concept"]))
print(f"Query: {query}")
print(f"Strategy: PRECISE (hnsw_ef=256, threshold=0.5)\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} [{p['doc_type']}] {p['text'][:70]}...")
```
The table below explains why each parameter is configured this way for factual queries:
| Parameter | Setting | Rationale |
| --------------------- | ------- | ------------------------------------------------------------------- |
| `hnsw_ef=256` | High | Factual queries need the *right* answer, not just a plausible one. |
| `score_threshold=0.5` | Strict | Drops results below cosine 0.5—better to return nothing than noise. |
| `doc_types` filter | Focused | Searches only API reference and concepts for factual questions. |
| `top_k=3` | Small | Factual answers are usually found in one or two documents. |
With these settings the search either returns a small number of highly confident matches or nothing at all—both are useful signals. An empty result set tells the router to invoke the fallback strategy rather than hallucinate an answer.
***
## Step 6: Strategy 2—Broad multistage retrieval for exploratory queries
Exploratory queries need breadth across multiple document types. The following block defines `broad_retrieval`, which creates one prefetch stream per document type plus an unfiltered catch-all stream, then merges all candidates with RRF fusion. Running the test query prints each result's score, document type, and a text preview:
```python theme={null}
async def broad_retrieval(query: str, doc_types: list[str], top_k: int = 5) -> list[ScoredPoint]:
vec = embed_text(query)
prefetch_stages = []
# One prefetch stream per requested document type
for dtype in doc_types:
f = FilterBuilder().must(Field("doc_type").eq(dtype)).build()
prefetch_stages.append(
PrefetchQuery(
query=vec,
filter=f,
limit=10,
params=SearchParams(hnsw_ef=128), # Lower ef per stream; breadth matters more than per-stream precision
)
)
# Add an unfiltered stream to catch documents that span multiple types
prefetch_stages.append(
PrefetchQuery(query=vec, limit=15)
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
COLLECTION,
query={"fusion": Fusion.RRF}, # Merge all streams by reciprocal rank
prefetch=prefetch_stages,
limit=top_k,
with_payload=True,
)
return list(results or [])
query = "How does the prefetch pipeline work with hybrid search and fusion?"
results = asyncio.run(broad_retrieval(query, ["tutorial", "concept", "api_reference"]))
print(f"Query: {query}")
print(f"Strategy: BROAD (4 prefetch streams, RRF fusion)\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} [{p['doc_type']:>15}] {p['text'][:65]}...")
```
The following diagram shows the four prefetch streams and how RRF fusion merges them into a single ranked result set:
```text theme={null}
Prefetch 1: tutorial docs → 10 candidates (how-to context)
Prefetch 2: concept docs → 10 candidates (theory/explanation)
Prefetch 3: api_reference docs → 10 candidates (exact API details)
Prefetch 4: unfiltered → 15 candidates (catch-all)
RRF fusion: merge all by rank → top 5 (diverse, multiperspective results)
```
Documents that appear across multiple prefetch streams rank higher, giving the LLM a well-rounded context. The lower `hnsw_ef=128` per stream is a deliberate trade-off: the four parallel streams compensate for any individual miss, so per-stream precision matters less than overall breadth.
***
## Step 7: Strategy 3—Troubleshooting retrieval with nested prefetch
Troubleshooting queries benefit from a wide net across both FAQ-style documents and changelogs, which often contain relevant fixes. The following block defines `troubleshooting_retrieval`, which uses a nested prefetch pipeline: inner prefetch stages gather candidates from troubleshooting docs and changelogs, DBSF fusion merges them, and a final rerank pass uses the query vector to surface the most relevant results. Running the test query prints each result's score, document type, and a text preview:
```python theme={null}
async def troubleshooting_retrieval(query: str, top_k: int = 5) -> list[ScoredPoint]:
vec = embed_text(query)
# Target troubleshooting docs and API reference for error-related queries
trouble_filter = FilterBuilder().must(
Field("doc_type").any_of(["troubleshooting", "api_reference"])
).build()
# Also gather changelog entries, which often document bug fixes
changelog_filter = FilterBuilder().must(
Field("doc_type").eq("changelog")
).build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
COLLECTION,
query=vec, # Final rerank by query vector
prefetch=[
PrefetchQuery(
query={"fusion": Fusion.DBSF}, # Normalize scores across inner streams before merging
prefetch=[
PrefetchQuery(query=vec, filter=trouble_filter, limit=10),
PrefetchQuery(query=vec, filter=changelog_filter, limit=5),
],
limit=12,
),
],
limit=top_k,
with_payload=True,
)
return list(results or [])
query = "My search returns empty results, what's wrong?"
results = asyncio.run(troubleshooting_retrieval(query))
print(f"Query: {query}")
print(f"Strategy: TROUBLESHOOTING (nested prefetch: FAQ + changelog, DBSF → re-rank)\n")
for r in results:
p = r.payload
print(f" score={r.score:.4f} [{p['doc_type']:>17}] {p['text'][:65]}...")
```
The troubleshooting strategy uses three stages to progressively narrow candidates before the final rerank:
```text theme={null}
Inner prefetch 1: troubleshooting + api_reference → 10 candidates
Inner prefetch 2: changelog → 5 candidates
Middle stage: DBSF fusion → 12 candidates
Outer query: rerank by query vector → top 5
```
DBSF normalizes the scores from both inner streams before merging, giving a fair comparison between troubleshooting tips and changelog notes. The final rerank with the query vector ensures the most relevant results surface at the top.
***
## Step 8: Build the confidence evaluator
After retrieval, the pipeline needs to decide whether the results are strong enough to pass to the LLM or whether a fallback is needed. The following block defines the `RetrievalResult` dataclass and the `evaluate_confidence` function, then runs it against a test query and prints the confidence level, top score, average score, and document count:
```python theme={null}
@dataclass
class RetrievalResult:
results: list[ScoredPoint]
strategy: str
confidence: str # "high", "medium", "low", or "no_results"
top_score: float
avg_score: float
coverage: int # Number of documents returned
def evaluate_confidence(
results: list[ScoredPoint],
strategy: str,
high_threshold: float = 0.6,
low_threshold: float = 0.35,
) -> RetrievalResult:
"""Classify retrieval quality into four confidence levels.
Uses the top score as the primary signal and average score as a secondary
check to avoid cases where one strong result masks several weak ones.
"""
if not results:
return RetrievalResult(results, strategy, "no_results", 0.0, 0.0, 0)
scores = [r.score for r in results]
top_score = max(scores)
avg_score = sum(scores) / len(scores)
if top_score >= high_threshold and avg_score >= low_threshold:
confidence = "high"
elif top_score >= low_threshold:
confidence = "medium"
else:
confidence = "low"
return RetrievalResult(
results=results,
strategy=strategy,
confidence=confidence,
top_score=top_score,
avg_score=avg_score,
coverage=len(results),
)
query = "What parameters does the search method accept?"
results = asyncio.run(precise_retrieval(query, ["api_reference"]))
evaluation = evaluate_confidence(results, "precise")
print(f"Query: {query}")
print(f"Strategy: {evaluation.strategy}")
print(f"Confidence: {evaluation.confidence}")
print(f"Top score: {evaluation.top_score:.4f}")
print(f"Avg score: {evaluation.avg_score:.4f}")
print(f"Coverage: {evaluation.coverage} documents")
```
The evaluator maps score thresholds to one of four confidence levels, each of which drives a different downstream action:
| Confidence | Condition | Action |
| ------------ | --------------------------------- | ---------------------------------------------------------------- |
| `high` | Top score >= 0.6 and avg >= 0.35. | Proceed to LLM with full confidence. |
| `medium` | Top score >= 0.35. | Proceed but add a caveat: "Based on available information..." |
| `low` | Top score \< 0.35. | Try the fallback strategy or respond with "I don't know." |
| `no_results` | Empty result set. | Skip retrieval, answer directly or say "No relevant docs found." |
***
## Step 9: Fallback strategy—Widen the search
When initial retrieval has low confidence, the fallback strategy removes all filters, raises the candidate pool size, and merges the original results with an unfiltered search using client-side RRF. The following block defines `fallback_retrieval`, simulates a low-confidence query, and prints the fallback confidence level and the top results returned:
```python theme={null}
async def fallback_retrieval(query: str, original_results: list[ScoredPoint], top_k: int = 5) -> list[ScoredPoint]:
"""Widen the search when initial retrieval has low confidence.
Uses a single client connection to run both fallback queries together,
avoiding an extra network round-trip.
"""
vec = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
# Remove all filters and lower the threshold to cast the widest possible net
unfiltered = await client.points.search(
COLLECTION,
vector=vec,
limit=top_k * 3,
with_payload=True,
params=SearchParams(hnsw_ef=256),
) or []
# Sample random documents as a last-resort "did you mean?" backup
random_sample = list(await client.points.query(
COLLECTION,
query={"sample": Sample.Random},
limit=5,
with_payload=WithPayloadSelector(include=["text", "doc_type", "section"]),
) or [])
# Merge original filtered results with the unfiltered widening pass
if original_results and unfiltered:
merged = reciprocal_rank_fusion(
[original_results, unfiltered],
limit=top_k,
)
else:
merged = unfiltered[:top_k]
# If both passes return nothing, surface random documents so the user
# can see what the knowledge base contains and reformulate the query.
if not merged and random_sample:
return random_sample[:top_k]
return merged or random_sample[:top_k]
query = "How does the quantum flux capacitor module work?"
results = asyncio.run(precise_retrieval(query, ["api_reference"]))
evaluation = evaluate_confidence(results, "precise")
print(f"Initial: confidence={evaluation.confidence}, top_score={evaluation.top_score:.4f}")
if evaluation.confidence == "low" or evaluation.confidence == "no_results":
fallback_results = asyncio.run(fallback_retrieval(query, results))
fallback_eval = evaluate_confidence(fallback_results, "fallback")
print(f"Fallback: confidence={fallback_eval.confidence}, top_score={fallback_eval.top_score:.4f}")
for r in fallback_results:
p = r.payload
print(f" score={r.score:.4f} [{p.get('doc_type', '')}] {p.get('text', '')[:60]}...")
```
`Sample.Random` returns random points from the collection. In the fallback function above, it acts as a last-resort "did you mean?" response: if neither the original filtered search nor the unfiltered widening returns any results, the function returns these random documents so the user can see what is in the knowledge base and reformulate the query. Both fallback queries run inside a single client connection to avoid an extra round-trip.
***
## Step 10: Build the adaptive router
The router is the central coordinator. It classifies the incoming query, dispatches it to the appropriate retrieval strategy, evaluates the result confidence, invokes the fallback when needed, and increments a retrieval counter on every returned document. The following block defines the `AdaptiveRAGRouter` class:
```python theme={null}
class AdaptiveRAGRouter:
"""Routes queries to the appropriate retrieval strategy."""
async def retrieve(self, query: str) -> RetrievalResult:
"""Run the full adaptive retrieval pipeline for a single query."""
classified = classify_query(query)
# Short-circuit for queries that do not benefit from retrieval
if classified.query_type == QueryType.NO_RETRIEVAL:
return RetrievalResult([], "no_retrieval", "high", 0.0, 0.0, 0)
# Dispatch to the strategy that matches the query type
if classified.query_type == QueryType.FACTUAL:
results = await precise_retrieval(query, classified.target_doc_types)
evaluation = evaluate_confidence(results, "precise")
elif classified.query_type == QueryType.TROUBLESHOOTING:
results = await troubleshooting_retrieval(query)
evaluation = evaluate_confidence(results, "troubleshooting")
else:
results = await broad_retrieval(query, classified.target_doc_types)
evaluation = evaluate_confidence(results, "broad")
# Widen the search if the primary strategy did not produce confident results
if evaluation.confidence in ("low", "no_results"):
fallback_results = await fallback_retrieval(query, results)
evaluation = evaluate_confidence(fallback_results, f"{evaluation.strategy}+fallback")
await self._track_retrieval(evaluation.results)
return evaluation
async def _track_retrieval(self, results: list[ScoredPoint]):
"""Increment the retrieval counter on each returned document."""
if not results:
return
async with AsyncVectorAIClient(url=SERVER) as client:
for r in results:
count = (r.payload or {}).get("retrieval_count", 0) + 1
await client.points.set_payload(
COLLECTION,
payload={"retrieval_count": count},
ids=[r.id],
)
```
The following block runs the router against five representative queries and prints the assigned strategy, confidence level, top score, and document count for each. It covers all four query types including the fallback path:
```python theme={null}
async def demo_router():
router = AdaptiveRAGRouter()
queries = [
"What parameters does the search method accept?",
"How does hybrid search work with fusion and prefetch?",
"My search returns empty results, what's wrong?",
"Hi there!",
"How does the quantum flux capacitor module work?",
]
for query in queries:
result = await router.retrieve(query)
classified = classify_query(query)
print(
f" [{classified.query_type.value:>17}] strategy={result.strategy:<25} "
f"confidence={result.confidence:<10} top={result.top_score:.4f} "
f"docs={result.coverage} | {query[:50]}"
)
asyncio.run(demo_router())
```
### Expected output
The `demo_router` function passes five representative queries through the full adaptive pipeline. Each query is first classified, then dispatched to the appropriate strategy—`precise` for factual, `broad` for exploratory, `troubleshooting` for error queries, and `no_retrieval` for the greeting. The final query about a non-existent "quantum flux capacitor" does not match any document closely enough, so the primary broad search scores poorly and the router automatically invokes the fallback strategy, producing the `broad+fallback` label with a low confidence rating and a reduced top score.
```text theme={null}
[ factual] strategy=precise confidence=high top=0.7234 docs=3 | What parameters does the search method accept?
[ exploratory] strategy=broad confidence=high top=0.6890 docs=5 | How does hybrid search work with fusion and prefe
[ troubleshooting] strategy=troubleshooting confidence=high top=0.7123 docs=5 | My search returns empty results, what's wrong?
[ no_retrieval] strategy=no_retrieval confidence=high top=0.0000 docs=0 | Hi there!
[ exploratory] strategy=broad+fallback confidence=low top=0.2345 docs=5 | How does the quantum flux capacitor module work?
```
***
## Step 11: User feedback loop
When a user marks a response as helpful or unhelpful, the `usefulness_score` of every retrieved document is updated. The following block defines `record_feedback`, simulates a helpful feedback event on a real retrieval result, and prints a confirmation with the number of documents updated:
```python theme={null}
async def record_feedback(result: RetrievalResult, helpful: bool):
"""Update usefulness scores based on user feedback.
Applies an exponential moving-average adjustment so that a single
feedback event does not dominate the score history.
"""
if not result.results:
return
async with AsyncVectorAIClient(url=SERVER) as client:
for r in result.results:
payload = r.payload or {}
current_score = payload.get("usefulness_score", 0.5)
feedback_count = payload.get("feedback_count", 0) + 1
if helpful:
# Nudge score toward 1.0; converges asymptotically
new_score = min(current_score + (1.0 - current_score) * 0.1, 1.0)
else:
# Penalize faster to deprioritize persistently unhelpful documents
new_score = max(current_score - current_score * 0.15, 0.0)
await client.points.set_payload(
COLLECTION,
payload={
"usefulness_score": round(new_score, 4),
"feedback_count": feedback_count,
"last_feedback": now_iso(),
"last_feedback_type": "helpful" if helpful else "unhelpful",
},
ids=[r.id],
)
label = "helpful" if helpful else "unhelpful"
print(f"Recorded '{label}' feedback for {len(result.results)} documents.")
router = AdaptiveRAGRouter()
result = asyncio.run(router.retrieve("What parameters does the search method accept?"))
asyncio.run(record_feedback(result, helpful=True))
```
Each feedback event nudges a document's score toward 1.0 (helpful) or toward 0.0 (unhelpful) using an exponential moving-average formula so that no single event dominates the history:
| Scenario | Formula | Effect |
| ------------------- | ------------------------------ | -------------------------------------------- |
| Helpful feedback. | `score += (1.0 - score) * 0.1` | Score rises asymptotically toward 1.0. |
| Unhelpful feedback. | `score -= score * 0.15` | Score drops faster, penalizing poor results. |
| No feedback. | Score unchanged. | Stays at the default of 0.5. |
After many feedback cycles, frequently helpful documents accumulate high scores while unhelpful ones sink. The feedback-aware retrieval function in the next step uses these scores to boost useful documents.
***
## Step 12: Feedback-aware retrieval
The `usefulness_score` accumulated in step 11 can be used to bias future retrieval toward documents that users have consistently found helpful. The following block defines `feedback_aware_retrieval`, runs a test query, and prints each result's score, usefulness score, retrieval count, document type, and a text preview:
```python theme={null}
async def feedback_aware_retrieval(query: str, top_k: int = 5) -> list[ScoredPoint]:
"""Retrieve documents, biasing toward those with high usefulness scores."""
vec = embed_text(query)
# Restrict the second prefetch stream to documents above the usefulness threshold
useful_filter = FilterBuilder().must(
Field("usefulness_score").gte(0.4)
).build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.query(
COLLECTION,
query={"fusion": Fusion.RRF},
prefetch=[
PrefetchQuery(query=vec, limit=15), # Semantic relevance stream
PrefetchQuery(query=vec, filter=useful_filter, limit=15), # Proven-helpful stream
],
limit=top_k,
with_payload=True,
)
return list(results or [])
query = "How to perform filtered search?"
results = asyncio.run(feedback_aware_retrieval(query))
print(f"Query: {query}")
print(f"Strategy: feedback-aware (RRF: unfiltered + usefulness>=0.4)\n")
for r in results:
p = r.payload
print(
f" score={r.score:.4f} useful={p.get('usefulness_score', 0.5):.2f} "
f"retrievals={p.get('retrieval_count', 0)} [{p['doc_type']}] "
f"{p['text'][:55]}..."
)
```
The function runs two prefetch streams in parallel and merges them with RRF, so documents that satisfy both criteria rank above those that satisfy only one:
```text theme={null}
Prefetch 1: Unfiltered search → 15 candidates (semantic relevance)
Prefetch 2: Usefulness-filtered → 15 candidates (proven helpful)
RRF fusion: documents in both lists rank higher
```
A document that is both semantically relevant *and* historically useful gets a double boost. A document that is semantically relevant but has been marked unhelpful appears in only one stream and ranks lower.
***
## Step 13: Analytics—what is the system learning?
As the system accumulates retrieval events and feedback, payload fields like `retrieval_count` and `usefulness_score` reflect its usage patterns. The following block queries the collection for the five most-retrieved documents, the five most-useful documents, and any documents that are frequently retrieved but consistently rated unhelpful, then prints all three groups:
```python theme={null}
async def retrieval_analytics():
async with AsyncVectorAIClient(url=SERVER) as client:
total = await client.vde.get_vector_count(COLLECTION)
# Order by retrieval_count descending to find the most-used documents
most_retrieved = list(await client.points.query(
COLLECTION,
query={"order_by": OrderBy(key="retrieval_count", direction=Direction.Desc)},
limit=5,
with_payload=WithPayloadSelector(
include=["text", "doc_type", "retrieval_count", "usefulness_score"],
),
) or [])
# Order by usefulness_score descending to find the highest-rated documents
most_useful = list(await client.points.query(
COLLECTION,
query={"order_by": OrderBy(key="usefulness_score", direction=Direction.Desc)},
limit=5,
with_payload=WithPayloadSelector(
include=["text", "doc_type", "usefulness_score", "feedback_count"],
),
) or [])
# Find documents retrieved often but rated poorly—candidates for review or removal
frequently_bad = await client.points.search(
COLLECTION,
vector=embed_text("general documentation"),
limit=20,
filter=(
FilterBuilder()
.must(Field("retrieval_count").gte(3))
.must(Field("usefulness_score").lt(0.3))
.build()
),
with_payload=WithPayloadSelector(
include=["text", "doc_type", "retrieval_count", "usefulness_score"],
),
) or []
# Count documents per type to understand collection composition
for dtype in ["api_reference", "tutorial", "concept", "troubleshooting", "changelog"]:
result = await client.points.count(
COLLECTION,
filter=FilterBuilder().must(Field("doc_type").eq(dtype)).build(),
exact=True,
)
print(f" {dtype:>17}: {result.count} documents")
print(f"\nTotal documents: {total}")
print(f"\n--- Most retrieved ---")
for r in most_retrieved:
p = r.payload
print(f" retrievals={p.get('retrieval_count', 0):>3} useful={p.get('usefulness_score', 0.5):.2f} {p['text'][:55]}...")
print(f"\n--- Most useful ---")
for r in most_useful:
p = r.payload
print(f" useful={p.get('usefulness_score', 0.5):.2f} feedback={p.get('feedback_count', 0)} {p['text'][:55]}...")
if frequently_bad:
print(f"\n--- Frequently retrieved but not useful (candidates for review) ---")
for r in frequently_bad:
p = r.payload
print(f" retrievals={p.get('retrieval_count', 0)} useful={p.get('usefulness_score', 0.5):.2f} {p['text'][:55]}...")
asyncio.run(retrieval_analytics())
```
***
## Step 14: Prepare the prompt for LLM integration
The final pipeline step assembles the retrieved context chunks and a confidence-adjusted instruction into a prompt string. The following block defines `adaptive_rag_answer`, runs it against four test queries, and prints the strategy, confidence level, and source document count for each. The actual LLM call is left as a stub (`# answer = await llm.generate(prompt)`) so any provider can be plugged in:
```python theme={null}
async def adaptive_rag_answer(query: str) -> dict:
"""Classify the query, retrieve context, and assemble an LLM prompt.
Returns a dictionary containing the strategy used, confidence level,
source document metadata, and a truncated preview of the assembled prompt.
The prompt itself is ready to pass to any LLM; replace the stub comment
with your provider's generate call.
"""
router = AdaptiveRAGRouter()
classified = classify_query(query)
# Skip retrieval entirely for queries that do not need document context
if classified.query_type == QueryType.NO_RETRIEVAL:
return {
"query": query,
"strategy": "no_retrieval",
"confidence": "high",
"answer": f"[Direct response — no retrieval needed for: '{query}']",
"sources": [],
}
result = await router.retrieve(query)
context_chunks = []
sources = []
for r in result.results:
p = r.payload or {}
context_chunks.append(p.get("text", ""))
sources.append({
"id": r.id,
"score": r.score,
"doc_type": p.get("doc_type"),
"section": p.get("section"),
})
context = "\n\n".join(context_chunks)
# Adjust the instruction based on how confident the retrieval was
if result.confidence == "high":
instruction = "Answer based on the provided context."
elif result.confidence == "medium":
instruction = "Answer based on available information. Note that context may be incomplete."
else:
instruction = "Limited context was found. Provide a best-effort answer and suggest the user consult the full documentation."
prompt = f"""Context:
{context}
Instruction: {instruction}
Question: {query}
Answer:"""
# Send `prompt` to your LLM here, for example:
# answer = await llm.generate(prompt)
return {
"query": query,
"strategy": result.strategy,
"confidence": result.confidence,
"top_score": result.top_score,
"sources": sources,
"prompt_preview": prompt[:200] + "...",
}
queries = [
"What parameters does the search method accept?",
"How do I build a hybrid search pipeline?",
"My search is returning empty results",
"Thanks!",
]
for q in queries:
response = asyncio.run(adaptive_rag_answer(q))
print(f"\nQuery: {q}")
print(f" Strategy: {response['strategy']}")
print(f" Confidence: {response['confidence']}")
print(f" Sources: {len(response.get('sources', []))} documents")
```
***
## Step 15: Collection cleanup
The following block retrieves the current document count, flushes all pending writes to disk, and prints a confirmation. Uncomment the delete lines to remove the collection entirely:
```python theme={null}
async def cleanup():
async with AsyncVectorAIClient(url=SERVER) as client:
count = await client.vde.get_vector_count(COLLECTION)
print(f"Collection '{COLLECTION}' contains {count} documents.")
await client.vde.flush(COLLECTION)
print("Flushed to disk.")
# Uncomment to delete the collection:
# await client.collections.delete(COLLECTION)
# print("Collection deleted.")
asyncio.run(cleanup())
```
***
## Adaptive strategies summary
The following table summarizes the retrieval strategy, search configuration, and fusion method used for each query type:
| Query type | Strategy | Search config | Prefetch | Fusion | Threshold |
| ---------------- | ----------------- | ------------- | ------------------------------------------ | -------------------- | ----------------- |
| Factual. | Precise. | `hnsw_ef=256` | None. | None. | 0.5 |
| Exploratory. | Broad multistage. | `hnsw_ef=128` | 4 streams (per doc\_type + unfiltered). | `Fusion.RRF` | None. |
| Troubleshooting. | Nested prefetch. | Default. | 2 inner (FAQ + changelog) → DBSF → rerank. | `Fusion.DBSF` inner. | None. |
| Low confidence. | Fallback. | `hnsw_ef=256` | None (unfiltered) + `Sample.Random`. | Client-side RRF. | None. |
| Feedback-aware. | Boosted fusion. | Default. | 2 streams (all + useful). | `Fusion.RRF` | usefulness >= 0.4 |
***
## APIs and features used in this tutorial
The following table lists every VectorAI DB API and feature demonstrated across the fifteen steps:
| Feature | API | Purpose |
| ------------------------ | ------------------------------------------------- | -------------------------------------------------------------- |
| Collection creation. | `collections.get_or_create(hnsw_config=...)` | Knowledge base setup. |
| Semantic search. | `points.search(params=SearchParams(hnsw_ef=256))` | Precise factual retrieval. |
| Score threshold. | `points.search(score_threshold=0.5)` | Cut low-confidence results. |
| Multi-stage prefetch. | `PrefetchQuery(query=..., filter=..., limit=...)` | Per-doc-type retrieval streams. |
| Nested prefetch. | `PrefetchQuery(prefetch=[...])` | Three-stage troubleshooting pipeline. |
| Server-side RRF. | `query={"fusion": Fusion.RRF}` | Broad exploratory fusion. |
| Server-side DBSF. | `query={"fusion": Fusion.DBSF}` | Troubleshooting score-normalized fusion. |
| Random sampling. | `query={"sample": Sample.Random}` | Fallback discovery. |
| Client-side RRF. | `reciprocal_rank_fusion(results, limit=...)` | Fallback merge. |
| Payload updates. | `points.set_payload(payload=...)` | Feedback tracking, retrieval counters. |
| Payload ordering. | `query(query={"order_by": OrderBy(...)})` | Most-retrieved, most-useful analytics. |
| Selective payload. | `WithPayloadSelector(include=[...])` | Return only needed fields. |
| Keyword index. | `FieldType.FieldTypeKeyword` | Doc type and source filtering. |
| Float index (principal). | `FloatIndexParams(is_principal=True)` | Usefulness score ordering. |
| Integer index (range). | `IntegerIndexParams(range=True)` | Retrieval count range queries. |
| Datetime index. | `DatetimeIndexParams(is_principal=True)` | Index `created_at` for range queries and time-based filtering. |
| `any_of` filter. | `Field("doc_type").any_of([...])` | Multi-value doc type matching. |
| `gte` / `lt` filters. | `Field("usefulness_score").gte(0.4)` | Feedback-based boosting. |
| Filtered count. | `points.count(filter=..., exact=True)` | Analytics per doc type. |
| Vector count. | `vde.get_vector_count()` | Collection statistics. |
| Flush. | `vde.flush()` | Persist pending writes. |
***
## Next steps
Improve relevance with multistage reranking
Add image search to your RAG pipeline
Tune HNSW, quantization, and search parameters
Master the full Filter DSL
# Build your first application
Source: https://docs.vectoraidb.actian.com/academy/tutorials/first-application
Step-by-step guide to building a semantic search app with Actian VectorAI DB: install, connect, create collections, embed, store, search, filter, update, and delete.
In this tutorial, you build a complete application on Actian VectorAI DB from scratch. By the end, you have a working movie recommendation engine that can store movie descriptions as dense vectors, find semantically similar movies using natural language queries, filter results by genre, year, or rating, update movie information after ingestion, delete outdated records, and inspect collection health and statistics.
No prior vector database experience is required. Each step introduces a concept, explains why it matters, and shows the code you need.
***
## What you build
A user describes what they want to watch in natural language — "a suspenseful space movie" — and the system finds the best matches from the database, optionally filtered by genre, year, or minimum rating. The diagram below shows how data flows from raw movie records through embedding and into a searchable vector store.
```mermaid theme={null}
flowchart LR
Movies[Movie Data - title, plot, genre, year, rating] --> Embed[Sentence Transformer - 384-dim vectors]
Embed --> Store[(Actian VectorAI DB)]
Query[User Query - natural language] --> QEmbed[Embed Query]
QEmbed --> Search[Semantic Search + Filters]
Store --> Search
Search --> Results[Ranked Movies]
```
***
## Prerequisites
Before starting, make sure the following are in place.
* Python 3.10 or later.
* `pip` available in your environment (verify with `pip --version`).
* A virtual environment activated (recommended: `python -m venv .venv && source .venv/bin/activate`).
* An Actian VectorAI DB server running (default: `localhost:6574`).
* Internet access on first run — `sentence-transformers` downloads the embedding model (`all-MiniLM-L6-v2`, approximately 90 MB) from Hugging Face when you first call `SentenceTransformer(EMBED_MODEL)`.
* At least 512 MB of free memory to load the embedding model.
***
## Step 1: Install dependencies
The following command installs the Actian VectorAI SDK and the sentence embedding library. Run it inside your virtual environment.
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
The two packages serve distinct roles in the application.
| Package | Purpose |
| ------------------------ | --------------------------------------------------------------------- |
| `actian-vectorai-client` | Official Python SDK — async/sync clients, Filter DSL, gRPC transport. |
| `sentence-transformers` | Open-source library for generating text embeddings. |
***
## Step 2: Import libraries and configure
The following snippet imports every class needed for this tutorial and sets three constants that identify the server address, collection name, and embedding model. Running it loads the model into memory and prints the resolved configuration so you can confirm the values before proceeding.
```python theme={null}
import asyncio
from sentence_transformers import SentenceTransformer
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
Field,
FilterBuilder,
PointStruct,
VectorParams,
)
from actian_vectorai.models.collections import HnswConfigDiff
# Connection and collection settings
SERVER = "localhost:6574"
COLLECTION = "Movies"
# Embedding model settings — model name and its output dimension must match
EMBED_MODEL = "all-MiniLM-L6-v2"
EMBED_DIM = 384
# Load the embedding model into memory (downloads on first run)
model = SentenceTransformer(EMBED_MODEL)
print(f"Server: {SERVER}")
print(f"Collection: {COLLECTION}")
print(f"Model: {EMBED_MODEL} ({EMBED_DIM} dimensions)")
```
The table below describes what each import provides.
| Import | Purpose |
| --------------------- | ------------------------------------------------------------- |
| `AsyncVectorAIClient` | Manages the gRPC connection to VectorAI DB. |
| `Distance` | Enum for similarity metrics (Cosine, Dot, Euclid, Manhattan). |
| `Field` | Builds type-safe conditions on payload fields. |
| `FilterBuilder` | Combines conditions with boolean logic (AND / OR / NOT). |
| `PointStruct` | A data point: ID + vector + payload (metadata). |
| `VectorParams` | Configuration for the vector space: dimension + distance. |
| `HnswConfigDiff` | Tuning parameters for the HNSW search index. |
### Expected output
The three constants are printed in order — server address, collection name, and the model name with its output dimension.
```text theme={null}
Server: localhost:6574
Collection: Movies
Model: all-MiniLM-L6-v2 (384 dimensions)
```
***
## Step 3: Connect to the server
The following snippet opens a gRPC connection to the server, calls `health_check()`, and prints the server's version information. If the connection fails, an exception is raised inside the `async with` block and the error message identifies the problem.
```python theme={null}
async def check_connection():
# async with manages the connection lifecycle — opens on entry, closes on exit
async with AsyncVectorAIClient(url=SERVER) as client:
health = await client.health_check()
print(f"Server health: {health}")
asyncio.run(check_connection())
```
### Expected output
When the server is reachable, health information similar to the following is printed.
```text theme={null}
Server health: {'title': 'actian-vectorai', 'version': '2.5.0'}
```
If you see a connection error, then verify that the VectorAI DB server is running on `localhost:6574`.
When `check_connection()` runs, the `async with AsyncVectorAIClient(...)` block manages the gRPC connection lifecycle. The client opens a channel to `SERVER`, runs the coroutine body including `health_check()`, and closes the channel when the block exits, so resources are released even if something fails. The sequence is as follows.
1. `AsyncVectorAIClient(url=SERVER)` creates a client instance.
2. `async with` opens a gRPC channel and verifies the server is reachable.
3. `health_check()` pings the server and returns status information.
4. When the `async with` block exits, the connection is closed cleanly.
***
## Step 4: Create a collection
A *collection* is a named container for vectors. Think of it as a table in a relational database, but optimized for similarity search.
The following snippet calls `get_or_create`, which creates the collection if it does not already exist. On first run it prints `created`; on subsequent runs it prints `already exists`. The function returns a boolean indicating whether a new collection was provisioned.
```python theme={null}
async def create_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
created = await client.collections.get_or_create(
name=COLLECTION,
vectors_config=VectorParams(
size=EMBED_DIM, # vector dimension must match the embedding model
distance=Distance.Cosine, # cosine similarity is recommended for sentence transformers
),
hnsw_config=HnswConfigDiff(m=16, ef_construct=128),
)
print(f"Collection '{COLLECTION}' {'created' if created else 'already exists'}.")
asyncio.run(create_collection())
```
The arguments to `get_or_create` define the vector dimension, how similarity is measured, and how the HNSW index is built. The table below explains each parameter.
| Parameter | Value | Meaning |
| -------------------------- | ----------------------- | ---------------------------------------------------------------- |
| `size=384` | Vector dimension | Must match the embedding model's output dimension. |
| `distance=Distance.Cosine` | Similarity metric | Cosine similarity is ideal for sentence transformers. |
| `m=16` | HNSW graph connections | Each node connects to 16 neighbours — balances speed and recall. |
| `ef_construct=128` | Build-time search width | Higher values improve index quality at the cost of build time. |
### Why use `get_or_create`
`get_or_create` is safe to call repeatedly. When the collection does not yet exist, the SDK creates it and returns `True`. When the collection already exists, the SDK skips creation and returns `False`. This boolean return value lets you log whether a new collection was provisioned, and your scripts become idempotent — safe to re-run without side effects.
### Expected output
`get_or_create` prints whether it provisioned a new collection or found one that already existed.
```text theme={null}
Collection 'Movies' created.
```
***
## Step 5: Create embedding helpers
The following two functions wrap the sentence transformer model. `embed_text` encodes a single string; `embed_texts` encodes a list of strings in one forward pass and is significantly faster when processing multiple items.
```python theme={null}
def embed_text(text: str) -> list[float]:
"""Convert a single text string to a 384-dimensional vector."""
return model.encode(text).tolist()
def embed_texts(texts: list[str]) -> list[list[float]]:
"""Convert a batch of text strings to vectors in a single forward pass."""
return model.encode(texts).tolist()
# Encode a test string and print the dimension and first five values to verify the model loaded correctly
test_vec = embed_text("A thrilling adventure in space")
print(f"Vector dimension: {len(test_vec)}")
print(f"First 5 values: {[round(v, 4) for v in test_vec[:5]]}")
```
### Expected output
The vector dimension confirms the model loaded correctly. The five sample values will differ slightly between runs because the model weights are fixed but floating-point precision varies across platforms.
```text theme={null}
Vector dimension: 384
First 5 values: [-0.0234, 0.0891, -0.0567, 0.0123, -0.0456]
```
Batching matters for three reasons.
* Speed: `embed_texts` processes all texts in a single forward pass through the model, which is significantly faster than calling `embed_text` in a loop.
* Efficiency: Batching reduces CPU and memory overhead compared to encoding one string at a time.
* Best practice: Always batch when embedding more than a few texts.
***
## Step 6: Prepare your data
Each movie becomes a point in the collection. A point has three parts.
* ID — A unique identifier (integer or UUID string).
* Vector — An embedding of the movie's plot description.
* Payload — Structured metadata (genre, year, rating, and so on).
The following list defines ten movies that will be embedded and stored in the next step. Each entry includes a plot description that the embedding model will encode into a 384-dimensional vector.
```python theme={null}
movies = [
{
"title": "Interstellar",
"plot": "A team of explorers travel through a wormhole in space to ensure humanity's survival on a dying Earth.",
"genre": "sci-fi",
"year": 2014,
"rating": 8.7,
"director": "Christopher Nolan",
},
{
"title": "The Shawshank Redemption",
"plot": "A banker sentenced to life in prison forms an unlikely friendship and finds hope through acts of common decency.",
"genre": "drama",
"year": 1994,
"rating": 9.3,
"director": "Frank Darabont",
},
{
"title": "Inception",
"plot": "A thief who steals corporate secrets through dream-sharing technology is given the task of planting an idea in a target's mind.",
"genre": "sci-fi",
"year": 2010,
"rating": 8.8,
"director": "Christopher Nolan",
},
{
"title": "The Dark Knight",
"plot": "Batman faces the Joker, a criminal mastermind who plunges Gotham City into anarchy and forces the Dark Knight to confront his beliefs.",
"genre": "action",
"year": 2008,
"rating": 9.0,
"director": "Christopher Nolan",
},
{
"title": "Pulp Fiction",
"plot": "The lives of two mob hitmen, a boxer, a gangster, and his wife intertwine in four tales of violence and redemption.",
"genre": "crime",
"year": 1994,
"rating": 8.9,
"director": "Quentin Tarantino",
},
{
"title": "The Matrix",
"plot": "A computer hacker discovers that reality is a simulation created by machines and joins a rebellion to free humanity.",
"genre": "sci-fi",
"year": 1999,
"rating": 8.7,
"director": "The Wachowskis",
},
{
"title": "Forrest Gump",
"plot": "A slow-witted but kind-hearted man from Alabama witnesses and unwittingly influences several historical events in the 20th century.",
"genre": "drama",
"year": 1994,
"rating": 8.8,
"director": "Robert Zemeckis",
},
{
"title": "Alien",
"plot": "The crew of a commercial spaceship encounters a deadly extraterrestrial creature that begins hunting them one by one.",
"genre": "horror",
"year": 1979,
"rating": 8.5,
"director": "Ridley Scott",
},
{
"title": "Goodfellas",
"plot": "The story of Henry Hill and his life in the mob, covering his relationship with his wife and his mob partners.",
"genre": "crime",
"year": 1990,
"rating": 8.7,
"director": "Martin Scorsese",
},
{
"title": "Blade Runner 2049",
"plot": "A young blade runner discovers a long-buried secret that leads him to track down a former blade runner who has been missing for thirty years.",
"genre": "sci-fi",
"year": 2017,
"rating": 8.0,
"director": "Denis Villeneuve",
},
]
print(f"Loaded {len(movies)} movies.")
```
***
## Step 7: Embed and store the data
The following snippet embeds every plot in a single batch, wraps each movie as a `PointStruct`, sends all ten points to the server in one `upsert` call, flushes the data to disk, and then reads back the total vector count to confirm the write succeeded.
```python theme={null}
async def ingest_movies():
# Embed all plots in one batch for efficiency
plots = [m["plot"] for m in movies]
vectors = embed_texts(plots)
# Build a PointStruct for each movie: integer ID, plot vector, and full metadata as payload
points = []
for i, (movie, vector) in enumerate(zip(movies, vectors)):
points.append(PointStruct(
id=i,
vector=vector,
payload={
"title": movie["title"],
"plot": movie["plot"],
"genre": movie["genre"],
"year": movie["year"],
"rating": movie["rating"],
"director": movie["director"],
},
))
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.upsert(COLLECTION, points=points) # insert-or-update
await client.vde.flush(COLLECTION) # persist to disk immediately
count = await client.vde.get_vector_count(COLLECTION) # confirm stored count
print(f"Stored {len(points)} movies. Total in collection: {count}")
asyncio.run(ingest_movies())
```
### Expected output
After a successful upsert and flush, the stored count matches the number of points sent. The total reported by `get_vector_count` confirms all ten movies were persisted.
```text theme={null}
Stored 10 movies. Total in collection: 10
```
The ingestion pipeline runs through five stages.
1. `embed_texts` converts all 10 plots into 384-dimensional vectors in one batch.
2. Each movie becomes a `PointStruct` with an integer ID, the plot vector, and the full metadata as payload.
3. `points.upsert` sends the points to the server ("upsert" means insert-or-update).
4. `vde.flush` ensures the data is persisted to disk immediately.
5. `vde.get_vector_count` confirms how many vectors are stored.
***
## Step 8: Run your first semantic search
The following snippet embeds a natural-language query string, sends the query vector to the server, and prints the top five most similar movies ranked by cosine similarity score.
```python theme={null}
async def search_movies(query: str, top_k: int = 5):
query_vector = embed_text(query)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
with_payload=True,
) or []
return results
query = "a suspenseful movie set in outer space"
results = asyncio.run(search_movies(query))
print(f"Query: \"{query}\"\n")
for r in results:
p = r.payload
print(f" {r.score:.4f} {p['title']} ({p['year']}) — {p['genre']} — ★{p['rating']}")
```
The `search` call accepts three key parameters that control what is returned.
| Parameter | Value | Purpose |
| ------------------- | ---------------- | -------------------------------------------------------------- |
| `vector` | Query embedding | The search finds vectors closest to this one. |
| `limit=5` | Top 5 results | Number of results to return. |
| `with_payload=True` | Include metadata | Returns title, genre, year, and other fields with each result. |
### Example output
Your scores will vary. The embedding model surfaces space-themed films even when the exact query words do not appear in their plot descriptions.
```text theme={null}
Query: "a suspenseful movie set in outer space"
0.7823 Alien (1979) — horror — ★8.5
0.7156 Interstellar (2014) — sci-fi — ★8.7
0.5934 Blade Runner 2049 (2017) — sci-fi — ★8.0
0.5412 The Matrix (1999) — sci-fi — ★8.7
0.3201 Inception (2010) — sci-fi — ★8.8
```
In this example, the embedding model captures semantic similarity rather than exact keyword matching. The query "suspenseful movie set in outer space" returns "Alien" (a horror film about a creature on a spaceship) and "Interstellar" (a space exploration film), even though none of the exact query words appear in their plot descriptions. Search quality depends on the model and dataset.
***
## Step 9: Filter by metadata
Filters restrict the candidate set before vector ranking, so similarity scores are only compared within the matching subset. Actian VectorAI DB provides the `Field` and `FilterBuilder` classes for this purpose. The examples below show how to filter by genre, by a minimum rating, and by a combination of both.
### Filter by genre
The following snippet defines a `search_by_genre` function that builds a `must` condition on the `genre` field. Only points where `genre` equals the provided value are considered during ranking. Calling the function with `"sci-fi"` returns the top sci-fi matches for the query.
```python theme={null}
async def search_by_genre(query: str, genre: str, top_k: int = 5):
query_vector = embed_text(query)
# must() applies the condition before vector ranking — only matching points are scored
filter_obj = (
FilterBuilder()
.must(Field("genre").eq(genre))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
filter=filter_obj,
with_payload=True,
) or []
return results
results = asyncio.run(search_by_genre("an exciting adventure", "sci-fi"))
print("Genre filter: sci-fi\n")
for r in results:
p = r.payload
print(f" {r.score:.4f} {p['title']} ({p['year']})")
```
#### Example output
Your scores will vary. Only sci-fi movies are scored and returned. Non-matching genres are excluded before ranking.
```text theme={null}
Genre filter: sci-fi
0.6234 Interstellar (2014)
0.5890 The Matrix (1999)
0.5678 Inception (2010)
0.4512 Blade Runner 2049 (2017)
```
`Field("genre").eq("sci-fi")` creates a condition that passes only movies where `genre` equals `"sci-fi"`. The filter is applied before ranking, so the search only scores matching points.
### Filter by minimum rating
The following snippet uses `.gte()` on the numeric `rating` field to restrict results to movies at or above a minimum quality threshold.
```python theme={null}
async def search_highly_rated(query: str, min_rating: float, top_k: int = 5):
query_vector = embed_text(query)
filter_obj = (
FilterBuilder()
.must(Field("rating").gte(min_rating))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
filter=filter_obj,
with_payload=True,
) or []
return results
results = asyncio.run(search_highly_rated("intense crime story", 8.8))
print("Filter: rating >= 8.8\n")
for r in results:
p = r.payload
print(f" {r.score:.4f} {p['title']} — ★{p['rating']}")
```
#### Example output
Your scores will vary. Only movies with a rating of 8.8 or above are included. Results are ordered by semantic similarity, not by rating.
```text theme={null}
Filter: rating >= 8.8
0.6123 Pulp Fiction — ★8.9
0.5234 The Dark Knight — ★9.0
0.4678 The Shawshank Redemption — ★9.3
0.4012 Inception — ★8.8
0.3890 Forrest Gump — ★8.8
```
The filter passes only points whose `rating` payload value is greater than or equal to the threshold.
***
## Step 10: Combine multiple filters
`FilterBuilder` supports three types of boolean logic. Each method narrows or expands the candidate set in a different way.
| Method | Meaning | SQL equivalent |
| ------------- | ------------------------------------ | -------------- |
| `.must()` | All conditions must match. | `AND` |
| `.should()` | At least one condition should match. | `OR` |
| `.must_not()` | Exclude any points that match. | `NOT` |
The following snippet chains three conditions: movies released after 2000, with a rating of at least 8.5, and not in the drama genre. Running it with the query "mind-bending thriller" returns only films that satisfy all three conditions, ranked by similarity.
```python theme={null}
async def advanced_search(query: str, top_k: int = 5):
query_vector = embed_text(query)
# Chain multiple must() and must_not() calls — all conditions apply simultaneously
filter_obj = (
FilterBuilder()
.must(Field("year").gte(2000))
.must(Field("rating").gte(8.5))
.must_not(Field("genre").eq("drama"))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
filter=filter_obj,
with_payload=True,
) or []
return results
results = asyncio.run(advanced_search("mind-bending thriller"))
print("Filters: year >= 2000, rating >= 8.5, NOT drama\n")
for r in results:
p = r.payload
print(f" {r.score:.4f} {p['title']} ({p['year']}) — {p['genre']} — ★{p['rating']}")
```
### Example output
Your scores will vary. All three conditions are applied simultaneously. Only films released after 2000, rated at least 8.5, and not in the drama genre are considered for ranking.
```text theme={null}
Filters: year >= 2000, rating >= 8.5, NOT drama
0.6234 Inception (2010) — sci-fi — ★8.8
0.5890 The Dark Knight (2008) — action — ★9.0
0.5234 Interstellar (2014) — sci-fi — ★8.7
```
This query finds mind-bending thrillers released after 2000 with a rating of at least 8.5, excluding dramas.
***
## Step 11: Retrieve a specific movie by ID
The following snippet fetches movie ID `0` directly from the collection by passing the integer ID to `points.get()`. No search is performed — the server returns the exact point and its payload.
```python theme={null}
async def get_movie(movie_id: int):
async with AsyncVectorAIClient(url=SERVER) as client:
points = await client.points.get(
COLLECTION,
ids=[movie_id],
with_payload=True,
)
if not points:
print(f"Movie {movie_id} not found.")
return None
p = points[0].payload
print(f"ID {movie_id}: {p['title']} ({p['year']}) — {p['genre']} — ★{p['rating']}")
print(f" Plot: {p['plot']}")
return points[0]
asyncio.run(get_movie(0))
```
### Expected output
Point `0` is the first movie ingested in this tutorial, so the output shows Interstellar's full payload.
```text theme={null}
ID 0: Interstellar (2014) — sci-fi — ★8.7
Plot: A team of explorers travel through a wormhole in space to ensure humanity's survival on a dying Earth.
```
This code passes a single integer ID to `points.get()` with `with_payload=True`, so the server returns the exact point and its complete metadata without performing any similarity search. The function checks whether any points were returned, then prints the title, year, genre, rating, and full plot description of the matching record.
***
## Step 12: Update movie metadata
After ingestion, payload fields can be updated without re-embedding the vector. The following snippet calls `set_payload` to change the rating for movie ID `0` to `8.8`, then calls `get_movie` to confirm the change was applied.
```python theme={null}
async def update_movie_rating(movie_id: int, new_rating: float):
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.set_payload(
COLLECTION,
payload={"rating": new_rating},
ids=[movie_id],
)
print(f"Updated movie {movie_id} rating to ★{new_rating}")
asyncio.run(update_movie_rating(0, 8.8))
asyncio.run(get_movie(0))
```
### Expected output
The second call to `get_movie(0)` confirms the rating was updated from `8.7` to `8.8` while all other fields remain unchanged.
```text theme={null}
Updated movie 0 rating to ★8.8
ID 0: Interstellar (2014) — sci-fi — ★8.8
Plot: A team of explorers travel through a wormhole in space to ensure humanity's survival on a dying Earth.
```
`set_payload` merges the provided fields into the existing payload. Three properties define its behaviour.
* Merge behaviour: Only the specified fields are updated. All other fields in the existing payload remain unchanged.
* No re-embedding: The vector stays the same — only the metadata is modified, so there is no reprocessing cost.
* Immediate effect: Subsequent searches and retrievals reflect the updated values right away.
### Add new fields
`set_payload` can also add entirely new keys to a point. The following snippet adds a `tags` list to movie ID `0`. Because `set_payload` merges rather than replaces, the title, plot, genre, and all other existing fields are preserved.
```python theme={null}
async def add_tags(movie_id: int, tags: list[str]):
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.set_payload(
COLLECTION,
payload={"tags": tags},
ids=[movie_id],
)
print(f"Added tags to movie {movie_id}: {tags}")
asyncio.run(add_tags(0, ["space", "wormhole", "survival", "time-dilation"]))
asyncio.run(get_movie(0))
```
**Expected Output**
The follow-up `get_movie(0)` call confirms the new `tags` field was merged into the payload. All previously stored fields — title, plot, genre, year, and rating — remain intact.
```text theme={null}
Added tags to movie 0: ['space', 'wormhole', 'survival', 'time-dilation']
ID 0: Interstellar (2014) — sci-fi — ★8.8
Plot: A team of explorers travel through a wormhole in space to ensure humanity's survival on a dying Earth.
```
This code calls `add_tags` with movie ID `0` and a list of four descriptive tags: `space`, `wormhole`, `survival`, and `time-dilation`. The `set_payload` call merges the new `tags` field into the existing payload for that point, leaving all previously stored fields — title, plot, genre, year, rating, and director — unchanged. The follow-up call to `get_movie(0)` reads the point back from the collection so you can confirm the tags were stored correctly.
***
## Step 13: Delete points
Points can be removed individually by ID or in bulk by filter.
### Delete by ID
The following snippet removes movie ID `9` by passing an explicit ID list to `points.delete()`, then reads back the vector count to confirm the deletion.
```python theme={null}
async def delete_movie(movie_id: int):
async with AsyncVectorAIClient(url=SERVER) as client:
await client.points.delete(COLLECTION, ids=[movie_id])
count = await client.vde.get_vector_count(COLLECTION)
print(f"Deleted movie {movie_id}. Remaining: {count}")
asyncio.run(delete_movie(9))
```
#### Expected output
The vector count drops from 10 to 9, confirming that movie ID `9` (Blade Runner 2049) was removed from the collection.
```text theme={null}
Deleted movie 9. Remaining: 9
```
This code passes ID `9` — corresponding to "Blade Runner 2049", the last movie in the dataset — to `points.delete()`. After the deletion, `vde.get_vector_count` reads the updated total and prints it so you can confirm the point was removed.
### Delete by filter
The following snippet deletes all movies whose rating falls below a given threshold. The `filter_obj` uses `.lt()` (less than) to identify matching points. The vector count is read before and after the operation so the result is visible.
```python theme={null}
async def delete_low_rated(min_rating: float):
# Build a filter that matches any point with a rating below the threshold
filter_obj = (
FilterBuilder()
.must(Field("rating").lt(min_rating))
.build()
)
async with AsyncVectorAIClient(url=SERVER) as client:
count_before = await client.vde.get_vector_count(COLLECTION)
await client.points.delete(COLLECTION, filter=filter_obj)
await client.vde.flush(COLLECTION)
count_after = await client.vde.get_vector_count(COLLECTION)
print(f"Deleted movies with rating < {min_rating}. Before: {count_before}, After: {count_after}")
# Uncomment the line below to run — this permanently removes points from the collection
# asyncio.run(delete_low_rated(8.6))
```
***
## Step 14: Count points
The following snippet counts the total number of points in the collection, then runs three filtered counts to check how many sci-fi movies exist, how many have a rating of 8.8 or higher, and how many were directed by Christopher Nolan.
```python theme={null}
async def count_movies():
async with AsyncVectorAIClient(url=SERVER) as client:
# Total count — no filter applied
total = await client.vde.get_vector_count(COLLECTION)
print(f"Total movies: {total}")
# Filtered counts using exact=True for a precise scan
sci_fi = await client.points.count(
COLLECTION,
filter=FilterBuilder().must(Field("genre").eq("sci-fi")).build(),
exact=True,
)
print(f"Sci-fi movies: {sci_fi}")
highly_rated = await client.points.count(
COLLECTION,
filter=FilterBuilder().must(Field("rating").gte(8.8)).build(),
exact=True,
)
print(f"Movies with rating >= 8.8: {highly_rated}")
nolan = await client.points.count(
COLLECTION,
filter=FilterBuilder().must(Field("director").eq("Christopher Nolan")).build(),
exact=True,
)
print(f"Christopher Nolan movies: {nolan}")
asyncio.run(count_movies())
```
The `exact` parameter controls whether the count is precise or approximate. The table below explains the trade-off.
| Value | Behaviour |
| ------------- | ------------------------------------------------------------------------ |
| `exact=True` | Scans all points and returns the precise count. |
| `exact=False` | Uses an approximate count from the index (faster for large collections). |
For small collections, always use `exact=True`. For millions of points, `exact=False` avoids a full scan.
`client.points.count()` returns a count response object. The integer count is accessed via the `.count` attribute (for example, `sci_fi.count`). The code samples above print the response object directly for readability; update them to access `.count` if your SDK version returns a structured object rather than a raw integer.
### Expected output
Counts reflect the tutorial dataset after the earlier deletion of movie ID `9`.
```text theme={null}
Total movies: 9
Sci-fi movies: 3
Movies with rating >= 8.8: 5
Christopher Nolan movies: 3
```
The code runs four separate counts: one for the full collection and three with filters applied using `exact=True`. The first count returns the total number of points currently in the collection. The second filters by `genre == "sci-fi"`, the third by `rating >= 8.8`, and the fourth by `director == "Christopher Nolan"`. The results reflect the dataset state after movie ID `9` was deleted in Step 13.
***
## Step 15: Inspect collection status
The following snippet retrieves the collection's status, configuration, and current VDE lifecycle state, then prints them alongside the vector count. Run this at any point to verify the collection is healthy before running searches.
```python theme={null}
async def inspect_collection():
async with AsyncVectorAIClient(url=SERVER) as client:
info = await client.collections.get_info(COLLECTION)
print(f"Collection: {COLLECTION}")
print(f" Status: {info.status}")
print(f" Config: {info.config}")
state = await client.vde.get_state(COLLECTION)
print(f" VDE state: {state}")
count = await client.vde.get_vector_count(COLLECTION)
print(f" Vector count: {count}")
asyncio.run(inspect_collection())
```
### Expected output
This code connects to the server, calls `collections.get_info` to retrieve the collection's operational status and vector configuration, then calls `vde.get_state` to read the current VDE lifecycle state, and finally calls `vde.get_vector_count` to confirm the number of stored vectors. A `green` status and `active` VDE state indicate the collection is healthy and ready for searches.
```text theme={null}
Collection: Movies
Status: green
Config: {'params': {'vectors': {'size': 384, 'distance': 'Cosine'}, 'shard_number': 1, 'replication_factor': 1, 'write_consistency_factor': 1, 'on_disk_payload': True}, 'hnsw_config': {'m': 16, 'ef_construct': 128, 'full_scan_threshold': 10000, 'max_indexing_threads': 0, 'on_disk': False}, 'optimizer_config': {'deleted_threshold': 0.2, 'vacuum_min_vector_number': 1000, 'default_segment_number': 0, 'max_segment_size': None, 'memmap_threshold': None, 'indexing_threshold': 20000, 'flush_interval_sec': 5, 'max_optimization_threads': 1}, 'wal_config': {'wal_capacity_mb': 32, 'wal_segments_ahead': 0}, 'quantization_config': None}
VDE state: active
Vector count: 9
```
This code connects to the server, calls `collections.get_info` to retrieve the collection's operational status and vector configuration, then calls `vde.get_state` to read the current VDE lifecycle state, and finally calls `vde.get_vector_count` to confirm the number of stored vectors. All three values are printed together so you can verify the collection is healthy and correctly configured before running searches.
***
## Step 16: List all collections
The following snippet retrieves the names of every collection on the server and prints them as a numbered list. This is useful for confirming which collections are available before connecting a client.
```python theme={null}
async def list_collections():
async with AsyncVectorAIClient(url=SERVER) as client:
names = await client.collections.list()
print(f"Collections on server ({len(names)}):")
for name in names:
print(f" - {name}")
asyncio.run(list_collections())
```
### Expected output
Because only one collection was created in this tutorial, `collections.list()` returns a single entry. The count in the header updates automatically as collections are added or removed.
```text theme={null}
Collections on server (1):
- Movies
```
This code calls `collections.list()`, which returns the names of all collections currently provisioned on the server. The result is printed as a numbered list with the total count shown in the header. In this tutorial only one collection has been created, so the output lists `Movies` as the single entry.
***
## Step 17: Put it all together — a complete search function
The previous steps introduced each operation individually. This section consolidates them into a single reusable `recommend_movies` function that accepts optional filters and applies only the ones provided.
The function below accepts a natural-language query and four optional filter parameters. For each filter that is not `None`, the corresponding condition is added to the `FilterBuilder`. Running the three example calls prints results for an unfiltered sci-fi query, a crime story filtered to high-rated movies, and a feel-good query that excludes crime films made before 1990.
```python theme={null}
async def recommend_movies(
query: str,
genre: str | None = None,
min_year: int | None = None,
min_rating: float | None = None,
exclude_genre: str | None = None,
top_k: int = 5,
):
"""Recommend movies using semantic search with optional filters."""
query_vector = embed_text(query)
# Build filters conditionally — only add a condition when the parameter is provided
fb = FilterBuilder()
if genre:
fb = fb.must(Field("genre").eq(genre))
if min_year is not None:
fb = fb.must(Field("year").gte(min_year))
if min_rating is not None:
fb = fb.must(Field("rating").gte(min_rating))
if exclude_genre:
fb = fb.must_not(Field("genre").eq(exclude_genre))
filter_obj = fb.build()
async with AsyncVectorAIClient(url=SERVER) as client:
results = await client.points.search(
COLLECTION,
vector=query_vector,
limit=top_k,
filter=filter_obj,
with_payload=True,
) or []
# Print the query, active filters, and ranked results
print(f"\n Query: \"{query}\"")
filters_desc = []
if genre: filters_desc.append(f"genre={genre}")
if min_year is not None: filters_desc.append(f"year>={min_year}")
if min_rating is not None: filters_desc.append(f"rating>={min_rating}")
if exclude_genre: filters_desc.append(f"NOT {exclude_genre}")
print(f" Filters: {', '.join(filters_desc) or 'none'}")
print(f" Results: {len(results)}\n")
for r in results:
p = r.payload
print(f" {r.score:.4f} {p['title']} ({p['year']}) — {p['genre']} — ★{p['rating']}")
print(f" {p['plot'][:80]}...")
print()
asyncio.run(recommend_movies("a mind-bending sci-fi movie"))
asyncio.run(recommend_movies(
"an intense crime story",
min_rating=8.8,
))
asyncio.run(recommend_movies(
"a feel-good movie about life",
exclude_genre="crime",
min_year=1990,
))
```
### Example output
Your scores will vary. Three calls are made with different queries and filter combinations.
```text theme={null}
Query: "a mind-bending sci-fi movie"
Filters: none
Results: 5
0.7234 The Matrix (1999) — sci-fi — ★8.7
A computer hacker discovers that reality is a simulation created by machines and j...
0.7012 Inception (2010) — sci-fi — ★8.8
A thief who steals corporate secrets through dream-sharing technology is given the...
0.5890 Interstellar (2014) — sci-fi — ★8.8
A team of explorers travel through a wormhole in space to ensure humanity's surviv...
0.4234 The Dark Knight (2008) — action — ★9.0
Batman faces the Joker, a criminal mastermind who plunges Gotham City into anarchy...
0.3012 Alien (1979) — horror — ★8.5
The crew of a commercial spaceship encounters a deadly extraterrestrial creature t...
Query: "an intense crime story"
Filters: rating>=8.8
Results: 2
0.6890 Pulp Fiction (1994) — crime — ★8.9
The lives of two mob hitmen, a boxer, a gangster, and his wife intertwine in four ...
0.4123 The Dark Knight (2008) — action — ★9.0
Batman faces the Joker, a criminal mastermind who plunges Gotham City into anarchy...
Query: "a feel-good movie about life"
Filters: NOT crime, year>=1990
Results: 5
0.6345 Forrest Gump (1994) — drama — ★8.8
A slow-witted but kind-hearted man from Alabama witnesses and unwittingly influenc...
0.5678 The Shawshank Redemption (1994) — drama — ★9.3
A banker sentenced to life in prison forms an unlikely friendship and finds hope t...
...
```
The first call searches without any filters and returns the top five semantically similar movies for "a mind-bending sci-fi movie". The second call applies a `min_rating >= 8.8` filter, narrowing results to only highly rated movies that match "an intense crime story". The third call combines an `exclude_genre="crime"` exclusion with a `min_year=1990` lower bound, so the search for "a feel-good movie about life" returns only non-crime films from 1990 onwards. Each call prints the query, active filters, result count, and ranked movies with truncated plot descriptions.
***
## Step 18: Cleanup
The following snippet flushes any pending writes to disk and prints the current movie count. The two lines that delete the collection are commented out so the data is preserved by default — uncomment them only when the collection is no longer needed.
```python theme={null}
async def cleanup():
async with AsyncVectorAIClient(url=SERVER) as client:
count = await client.vde.get_vector_count(COLLECTION)
print(f"Collection '{COLLECTION}' contains {count} movies.")
await client.vde.flush(COLLECTION)
print("Data flushed to disk.")
# Uncomment the next two lines to permanently delete the collection:
# await client.collections.delete(COLLECTION)
# print(f"Collection '{COLLECTION}' deleted.")
asyncio.run(cleanup())
```
### Expected output
The vector count reflects the state of the collection after all previous steps. The flush confirmation line indicates that any pending writes have been safely persisted to disk.
```text theme={null}
Collection 'Movies' contains 9 movies.
Data flushed to disk.
```
This code reads the current vector count from the collection, prints it, then calls `vde.flush` to ensure any pending writes are persisted to disk. The two lines that delete the collection are commented out — they are safe to uncomment when the tutorial data is no longer needed, but the collection is preserved by default so the data remains available for further experimentation.
***
## What you learned
The table below summarises every concept and API used in this tutorial.
| Concept | API | What it does |
| ----------------- | ------------------------------------------------------------- | --------------------------------------------------------- |
| Connect | `AsyncVectorAIClient(url=...)` | Open a gRPC connection to VectorAI DB. |
| Health check | `client.health_check()` | Verify the server is reachable. |
| Create collection | `collections.get_or_create(vectors_config=VectorParams(...))` | Define a vector space with dimension and distance metric. |
| Embed text | `SentenceTransformer.encode()` | Convert text to a numerical vector. |
| Store data | `points.upsert(collection, points=[PointStruct(...)])` | Insert or update points with vectors and metadata. |
| Persist | `vde.flush(collection)` | Write pending data to disk. |
| Semantic search | `points.search(collection, vector=..., limit=5)` | Find the most similar vectors. |
| Filter (equality) | `Field("genre").eq("sci-fi")` | Match a specific value. |
| Filter (range) | `Field("rating").gte(8.5)` | Numeric comparison. |
| Filter (exclude) | `FilterBuilder().must_not(...)` | Exclude matching points. |
| Combine filters | `FilterBuilder().must(...).must(...).build()` | Boolean AND/OR/NOT logic. |
| Get by ID | `points.get(collection, ids=[0])` | Retrieve specific points. |
| Update metadata | `points.set_payload(collection, payload={...}, ids=[0])` | Merge new fields into existing payloads. |
| Delete by ID | `points.delete(collection, ids=[0])` | Remove specific points. |
| Delete by filter | `points.delete(collection, filter=...)` | Remove points matching conditions. |
| Count | `points.count(collection, filter=..., exact=True)` | Count matching points. |
| Vector count | `vde.get_vector_count(collection)` | Total vectors in the collection. |
| Collection info | `collections.get_info(collection)` | Status and configuration. |
| Collection state | `vde.get_state(collection)` | VDE lifecycle state. |
| List collections | `collections.list()` | All collection names on the server. |
| Delete collection | `collections.delete(collection)` | Remove a collection entirely. |
***
## Common patterns quick reference
The patterns below capture the idioms used most often when building applications with Actian VectorAI DB.
### Pattern 1: Search with optional filters
Build the filter conditionally so the same function works with or without constraints. Using `is not None` rather than a truthiness check prevents valid falsy values such as `0.0` from being silently skipped.
```python theme={null}
fb = FilterBuilder()
if genre:
fb = fb.must(Field("genre").eq(genre))
if min_rating is not None:
fb = fb.must(Field("rating").gte(min_rating))
filter_obj = fb.build()
```
### Pattern 2: Upsert is idempotent
Calling `upsert` with the same ID replaces the existing point, so ingestion scripts can be re-run safely without creating duplicates.
### Pattern 3: Always flush after writes
Call `vde.flush()` immediately after `points.upsert()` to ensure data survives server restarts. Without it, recent writes may be lost if the server crashes.
```python theme={null}
await client.points.upsert(COLLECTION, points=points)
await client.vde.flush(COLLECTION)
```
### Pattern 4: Use `get_or_create` for collections
`get_or_create` is safe to run on every application startup. It creates the collection if it does not exist and does nothing if it already does, so startup code does not need a separate existence check.
```python theme={null}
await client.collections.get_or_create(name=COLLECTION, vectors_config=...)
```
***
## Next steps
Master the full Filter DSL with all field types and operators.
Explore search parameters, score thresholds, and pagination.
Choose the right model and configure quantization for production.
Tune HNSW parameters, quantization, and search settings.
# Overview
Source: https://docs.vectoraidb.actian.com/academy/tutorials/index
Hands-on tutorials to build vector search applications with Actian VectorAI DB.
Learn Actian VectorAI DB through practical, task-focused tutorials. Each tutorial teaches specific skills you can apply immediately to your projects.
## Choose your learning path
Use this flowchart to find the tutorial track that matches your goals:
```mermaid theme={null}
flowchart TD
Start[Start here] --> Q{What do you want to do?}
Q --> |Learn basics| GS[Getting started]
Q --> |Build features| Core[Core features]
Q --> |Go deeper| Adv[Advanced topics]
GS --> App[Build your first application]
Core --> Search[Similarity search]
Core --> Filters[Predicate filters]
Adv --> Embed[Open-source embedding models]
Adv --> Multi[Multi-modal systems]
Adv --> Rerank[Reranking]
Adv --> Quality[Retrieval quality]
Adv --> Adaptive[Adaptive RAG]
```
## Getting started
Build foundational skills by creating your first VectorAI DB application.
Create a complete semantic search application from scratch. Learn to connect, store vectors, and query data.
## Core features
Master the essential features for production vector search applications.
Learn the core vector search workflow — from embedding and storing vectors to searching, scoring, batching, and paginating results.
Combine vector search with structured payload filters using the type-safe Filter DSL and logical operators.
## Advanced topics
Take your skills further with advanced techniques and architectures.
Choose, configure, and integrate Sentence Transformers, BGE, and other open-source models. Covers dimensionality trade-offs, quantization, and re-embedding workflows.
Store, search, and fuse text, image, and metadata embeddings in a single collection using named vectors, multistage prefetch, and server-side fusion.
Improve search relevance with multistage prefetch pipelines, cross-encoder scoring, payload-based boosting, and fusion reranking.
Measure and improve search accuracy by tuning HNSW parameters, distance metrics, quantization, score thresholds, and payload indexes.
Create RAG pipelines that adapt retrieval strategy at runtime based on query type, confidence signals, and user feedback.
## Recommended learning order
Follow this sequence to build skills progressively. Start with the beginner tutorials to build a strong foundation — each tutorial builds on concepts from previous ones, so following the recommended order helps you learn efficiently.
| Stage | Tutorial | Skills learned |
| ----- | -------------------------------------------------------------------------------------------- | -------------------------------------------------------- |
| 1 | [Build your first application](/academy/tutorials/first-application) | Connection, basic operations, search fundamentals |
| 2 | [Similarity search fundamentals](/academy/tutorials/similarity-search) | Search patterns, score thresholds, batch queries |
| 3 | [Predicate filters](/academy/tutorials/predicate-filters) | Metadata filtering, logical operators, combined queries |
| 4 | [Use open-source embedding models](/academy/tutorials/leverage-open-source-embedding-models) | Model selection, dimensionality, quantization |
| 5 | [Build multimodal systems](/academy/tutorials/multimodel-system) | Named vectors, multistage prefetch, fusion |
| 6 | [Rerank search results](/academy/tutorials/re-ranking) | Two-stage retrieval, cross-encoders, result optimization |
| 7 | [Optimize retrieval quality](/academy/tutorials/retrieval-quality) | Evaluation metrics, HNSW tuning, benchmarking |
| 8 | [Build adaptive RAG systems](/academy/tutorials/adaptive-rag) | Query classification, dynamic retrieval, self-correction |
## Time estimates
Use these estimates to plan your learning sessions and choose tutorials that fit your available time.
| Tutorial | Duration | Difficulty |
| -------------------------------- | -------- | ------------ |
| Build your first application | 45 min | Beginner |
| Similarity search fundamentals | 20 min | Beginner |
| Predicate filters | 25 min | Intermediate |
| Use open-source embedding models | 25 min | Intermediate |
| Build multimodal systems | 35 min | Advanced |
| Rerank search results | 30 min | Advanced |
| Optimize retrieval quality | 30 min | Advanced |
| Build adaptive RAG systems | 40 min | Advanced |
# Use open-source embedding models
Source: https://docs.vectoraidb.actian.com/academy/tutorials/leverage-open-source-embedding-models
Learn how to choose, configure, and integrate open-source embedding models with Actian VectorAI DB—covering model selection, dimensionality trade-offs, distance metrics, batch ingestion, quantization for large models, named vectors for multimodel search, and re-embedding workflows.
This is a hands-on tutorial. You will run Python against a local Actian VectorAI DB instance and step through choosing models, ingesting embeddings, and comparing search behavior. By the end, you will have a working multimodel search pipeline that lets you compare retrieval quality across different embedding architectures side by side.
Embedding models convert text (or images, audio, code) into dense numerical vectors that capture semantic meaning. Actian VectorAI DB stores these vectors and retrieves similar ones at scale — but the quality of your search depends entirely on the quality of your embeddings.
Choosing the right open-source model is one of the most impactful decisions you will make when building a vector search application. The wrong model wastes storage on unhelpful dimensions, produces low-recall results, and adds unnecessary latency.
Keep your server URL in `SERVER` aligned with your environment as you follow along.
***
## Architecture overview
The diagram below shows how documents flow through model selection, embedding, and storage in Actian VectorAI DB. Each model produces vectors of a different size, stored in separate collections so you can compare retrieval quality across configurations.
```mermaid theme={null}
flowchart LR
Data[Documents] --> Model{Choose Model}
Model -->|small & fast| MiniLM[MiniLM-L6 - 384-dim]
Model -->|balanced| MPNET[MPNet-base - 768-dim]
Model -->|high quality| E5[E5-large - 1024-dim]
MiniLM --> Collection[(Actian VectorAI DB)]
MPNET --> Collection
E5 --> Collection
Collection --> Search[Search & Compare]
```
***
## Environment setup
Run the following command to install the two packages this tutorial depends on.
```bash theme={null}
pip install actian-vectorai-client sentence-transformers
```
### What this installs
Both packages are required: one communicates with the database, and the other loads and runs embedding models on your machine. The list below maps each dependency to the role it plays in later steps.
* `actian-vectorai-client` — Official Python SDK for Actian VectorAI DB; provides async/sync clients, Filter DSL, and gRPC transport.
* `sentence-transformers` — Framework for loading and running open-source embedding models; downloads and caches models from Hugging Face.
***
## Step 1: Understand the model landscape
Before writing any code, review the table below to understand how the available models differ in dimension count, speed, and quality. The model you choose determines the shape of every vector stored in the database.
| Model | Dimensions | Speed | Quality | Best for |
| -------------------------------------------------- | ---------- | --------- | ------------ | --------------------------------- |
| `sentence-transformers/all-MiniLM-L6-v2` | 384 | Very fast | Good | Prototyping, low-latency apps |
| `sentence-transformers/all-MiniLM-L12-v2` | 384 | Fast | Better | Production with speed constraints |
| `sentence-transformers/all-mpnet-base-v2` | 768 | Moderate | High | General production use |
| `sentence-transformers/multi-qa-mpnet-base-dot-v1` | 768 | Moderate | High (QA) | Question-answering systems |
| `sentence-transformers/all-distilroberta-v1` | 768 | Moderate | High | Diverse text types |
| `intfloat/e5-large-v2` | 1024 | Slow | Very high | Maximum quality, offline indexing |
| `BAAI/bge-large-en-v1.5` | 1024 | Slow | Very high | Benchmarks, academic use |
| `sentence-transformers/clip-ViT-B-32` | 512 | Moderate | High (multi) | Text + image multimodal |
### Key trade-offs
Keep the following trade-offs in mind before choosing a model.
* More dimensions means more storage and slower search, but better semantic resolution.
* Fewer dimensions means less RAM and faster search, but may lose subtle meaning.
* Model architecture matters more than dimension count — a well-trained 384-dim model can outperform a poorly trained 768-dim one.
***
## Step 2: Import dependencies and configure
The block below imports every module used across all steps of this tutorial and sets the server address. Run it once at the top of your script or notebook. If the import succeeds and the server address prints, your environment is ready.
```python theme={null}
import asyncio
import time
from sentence_transformers import SentenceTransformer
from actian_vectorai import (
AsyncVectorAIClient,
Distance,
PointStruct,
VectorParams,
)
from actian_vectorai.models.collections import (
HnswConfigDiff,
ScalarQuantization,
QuantizationConfig,
)
from actian_vectorai.models.enums import Datatype, QuantizationType
from actian_vectorai.models.points import (
SearchParams,
QuantizationSearchParams,
WithPayloadSelector,
)
SERVER = "localhost:6574"
print(f"VectorAI Server: {SERVER}")
```
### Expected output
This block imports all SDK classes and utility modules needed throughout the tutorial — including the async client, distance enums, point structures, vector parameters, quantization types, and search parameter models — and sets `SERVER` to the local gRPC address. The final `print` statement confirms that the configuration loaded without errors and that the server address is set correctly.
```text theme={null}
VectorAI Server: localhost:6574
```
***
## Step 3: Load multiple models and compare embedding output
The code below loads three models — small, medium, and large — and encodes the same sample sentence with each one. Running it prints each model's load time, dimension count, and the first five values of the resulting vector, confirming that each model produces a vector of a different size.
```python theme={null}
# Three models spanning small, medium, and large architectures
models = {
"minilm": {
"name": "all-MiniLM-L6-v2",
"dim": 384,
"description": "Small, fast, good for prototyping",
},
"mpnet": {
"name": "all-mpnet-base-v2",
"dim": 768,
"description": "Balanced quality and speed",
},
"e5-large": {
"name": "intfloat/e5-large-v2",
"dim": 1024,
"description": "High quality, slower, needs quantization at scale",
},
}
# Load each model and report its dimension count and load time
loaded_models = {}
for key, info in models.items():
print(f"Loading {info['name']}...")
t0 = time.time()
loaded_models[key] = SentenceTransformer(info["name"])
elapsed = time.time() - t0
print(f" Loaded in {elapsed:.1f}s — {info['dim']} dimensions — {info['description']}")
# Encode one sentence with each model to confirm output dimensions
sample_text = "Vector databases store high-dimensional embeddings for similarity search."
print(f"\nSample: \"{sample_text}\"\n")
for key, m in loaded_models.items():
vec = m.encode(sample_text)
print(f" {key:>10}: dim={len(vec)}, first 5 values={vec[:5].round(4).tolist()}")
```
### Expected output
This block iterates over the three model definitions — MiniLM-L6 (384 dimensions), MPNet-base (768 dimensions), and E5-large-v2 (1024 dimensions) — loads each one from the Hugging Face cache via `SentenceTransformer`, and records the load time. It then encodes the same sample sentence with every loaded model and prints each model's actual output dimension and the first five vector values to confirm that the models are producing embeddings of the expected shape and are ready for ingestion.
```text theme={null}
Loading all-MiniLM-L6-v2...
Loaded in