Vector Stores & Retrieval: Semantic Search

Overview

Vector stores enable semantic search over memories—finding similar items based on meaning, not just keywords.

This is critical for agents to retrieve relevant past experiences and knowledge.

---

The Vector Store Concept

How It Works

Step 1: Embedding
  Text: "I analyzed quarterly sales data"
  → Embed → [0.12, 0.89, -0.34, ...]

Step 2: Store
  Save embedding in vector DB
  Index for fast retrieval

Step 3: Query
  Query: "What reports did we analyze?"
  → Embed → [0.15, 0.91, -0.30, ...]

Step 4: Retrieve
  Find closest embeddings
  Return: "I analyzed quarterly sales data"

Why Embeddings?

Embeddings capture semantic meaning:

Semantic similarity:
  "I analyzed sales data" ✓ Similar
  "I examined revenue reports" ✓ Similar
  "The weather was nice today" ✗ Different

vs Keyword matching:
  "I analyzed sales data" ✓ Exact match
  "I examined revenue reports" ✗ No match
  "Sales data analysis" ✗ Different order

---

Popular Vector Stores (2025-2026)

1. Chroma

import chromadb

client = chromadb.Client()
collection = client.create_collection("memories")

# Add documents
collection.add(
    documents=["I analyzed quarterly sales"],
    ids=["doc1"],
    embeddings=[0.1, 0.2, 0.3, ...](/0.1,-0.2,-0.3,-.../)
)

# Query
results = collection.query(
    query_texts=["What reports did we analyze?"],
    n_results=5
)

# Pros: Simple, local, no server
# Cons: Limited scale

2. Weaviate

from weaviate import Client

client = Client("http://localhost:8080")

# Store
client.data_object.create(
    data_object={
        "content": "I analyzed quarterly sales",
        "timestamp": "2025-01-15"
    },
    class_name="Memory"
)

# Query
response = client.query.get("Memory").with_near_text({
    "concepts": ["reports analysis"]
}).do()

# Pros: GraphQL, metadata, scalable
# Cons: Complex to set up

3. Milvus

from pymilvus import Collection, connections

connections.connect("default", host="localhost", port="19530")
collection = Collection("memories")

# Insert
collection.insert([embeddings])

# Search
results = collection.search(
    query_embeddings,
    "embedding_field",
    search_params={"metric_type": "L2", "params": {"nprobe": 10}},
    limit=5
)

# Pros: High performance, scalable to billions
# Cons: Operational overhead

4. LanceDB

import lancedb

db = lancedb.connect("./data")
table = db.create_table("memories")

# Add data
table.add([
    {"text": "I analyzed sales", "embedding": [...]}
])

# Query
results = table.search([query_embedding]).limit(5).to_list()

# Pros: Simple, modern, good performance
# Cons: Newer (less production battle-tested)

---

Implementing Semantic Search

Complete Example

from sentence_transformers import SentenceTransformer
import chromadb

class SemanticMemoryStore:
    def __init__(self):
        # Initialize embedder
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

        # Initialize vector store
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("episodic")

    def add_memory(self, text: str, metadata: dict = None):
        """Add memory with automatic embedding"""

        # Embed text
        embedding = self.encoder.encode(text).tolist()

        # Store with metadata
        memory_id = str(uuid.uuid4())

        self.collection.add(
            ids=[memory_id],
            documents=[text],
            embeddings=[embedding],
            metadatas=[metadata or {}]
        )

        return memory_id

    def retrieve(self, query: str, k: int = 5):
        """Retrieve similar memories"""

        # Embed query
        query_embedding = self.encoder.encode(query).tolist()

        # Search
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=k
        )

        # Format results
        return self._format_results(results)

    def _format_results(self, results):
        """Format raw results nicely"""
        formatted = []

        for i in range(len(results['documents'][0])):
            formatted.append({
                'text': results['documents'][0][i],
                'metadata': results['metadatas'][0][i],
                'distance': results['distances'][0][i]
            })

        # Sort by relevance (lower distance = more similar)
        formatted.sort(key=lambda x: x['distance'])

        return formatted

# Usage
memory_store = SemanticMemoryStore()

# Add memories
memory_store.add_memory(
    "Analyzed quarterly revenue report for Q1 2025",
    {"type": "analysis", "date": "2025-01-15"}
)

# Retrieve
results = memory_store.retrieve("What reports have we analyzed?")
for result in results:
    print(f"Found: {result['text']}")

---

Retrieval Strategies

Strategy 1: Direct Similarity

# Simple: Find most similar
query = "How do we analyze data?"
top_5 = memory_store.retrieve(query, k=5)

Strategy 2: Multi-Query Retrieval

# Query from multiple angles
queries = [
    "How do we analyze data?",
    "Data analysis methodology",
    "Steps for data processing"
]

all_results = []
for q in queries:
    results = memory_store.retrieve(q, k=3)
    all_results.extend(results)

# Deduplicate and rank
unique_results = deduplicate(all_results)
ranked = rank_by_relevance(unique_results)

Strategy 3: Hybrid Retrieval

# Combine semantic + keyword search
semantic_results = memory_store.retrieve(query, k=5)

keyword_results = memory_store.keyword_search(query, k=5)

# Merge and rank
combined = merge_results(semantic_results, keyword_results)
final = rank_and_deduplicate(combined)

Strategy 4: Metadata Filtering

# Find similar with constraints
results = memory_store.retrieve_with_filter(
    query="data analysis",
    filters={"type": "methodology"},
    date_range=("2024-01-01", "2025-01-01"),
    k=10
)

---

Advanced: Re-ranking

Sometimes first results aren't best. Re-rank them:

class ReRanker:
    def __init__(self):
        # Use cross-encoder for re-ranking
        self.model = CrossEncoder('ms-marco-MiniLM-L-12-v2')

    def rerank(self, query, initial_results, k=5):
        """Re-rank initial results"""

        # Score query-result pairs
        pairs = [(query, r['text']) for r in initial_results]
        scores = self.model.predict(pairs)

        # Re-rank by score
        ranked = sorted(
            zip(initial_results, scores),
            key=lambda x: x[1],
            reverse=True
        )

        # Return top k
        return [r[0] for r in ranked[:k]]

# Usage
initial_results = memory_store.retrieve("data analysis", k=20)
final_results = reranker.rerank("data analysis", initial_results, k=5)

---

Performance Optimization

Challenge: Latency

1000 memories: 10ms search ✓ Good
100,000 memories: 100ms search ✓ OK
1,000,000 memories: 1s search ✗ Too slow

Solutions

1. Indexing

# Build efficient index
collection.create_index("embeddings")
# Subsequent searches much faster

2. Partitioning

# Split by time
jan_2025 = collection_jan_2025
dec_2024 = collection_dec_2024

# Search recent first, then historical
recent_results = jan_2025.search(query)
if not enough_results:
    historical_results = dec_2024.search(query)

3. Approximate Search

# Use HNSW or IVF for approximate results
# Much faster, slightly less accurate
search_params = {
    "metric_type": "L2",
    "params": {"nprobe": 10}  # Approximate
}
results = collection.search(query, search_params)

---

Embedding Model Selection

Different embeddings for different needs:

General Purpose

# Sentence Transformers
model = SentenceTransformer('all-MiniLM-L6-v2')
# Fast, good quality, ~384 dims

Dense & High Quality

model = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
# Slower, higher quality, ~768 dims

Domain Specific

# Fine-tune on your domain
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
model = model.fine_tune_on_domain(your_data)

---

Practical Tips

Tip 1: Normalize Embeddings

import numpy as np

# Normalize to unit length
embedding = embedding / np.linalg.norm(embedding)
# Makes similarity calculations more efficient

Tip 2: Batch Processing

# Process in batches for efficiency
texts = ["memory1", "memory2", ..., "memory_1000"]

embeddings = []
for i in range(0, len(texts), 100):
    batch = texts[i:i+100]
    batch_embeddings = encoder.encode(batch)
    embeddings.extend(batch_embeddings)

Tip 3: Caching

# Cache common queries
@cache(ttl=3600)
def retrieve_cached(query, k=5):
    return memory_store.retrieve(query, k)

Tip 4: Metadata is Key

# Store rich metadata for filtering
memory_store.add_memory(
    text="Analyzed Q1 sales",
    metadata={
        "type": "financial",
        "period": "Q1",
        "year": 2025,
        "department": "sales",
        "confidence": 0.95,
        "source": "internal_report"
    }
)

# Later, filter by metadata
results = memory_store.retrieve_with_filter(
    query="sales",
    filters={"year": 2025, "department": "sales"},
    k=10
)

---

Monitoring Retrieval Quality

class RetrievalEvaluator:
    def evaluate(self, query, gold_results, retrieved_results):
        """Evaluate retrieval quality"""

        # Recall@k
        retrieved_ids = {r['id'] for r in retrieved_results}
        gold_ids = {r['id'] for r in gold_results}

        recall = len(retrieved_ids & gold_ids) / len(gold_ids)

        # MRR (Mean Reciprocal Rank)
        mrr = 0
        for i, result in enumerate(retrieved_results):
            if result in gold_results:
                mrr = 1 / (i + 1)
                break

        return {
            "recall": recall,
            "mrr": mrr,
            "ndcg": self.calculate_ndcg(gold_ids, retrieved_ids)
        }

---

Key Takeaways

1. Embeddings capture meaning - Not just keywords
2. Vector stores enable semantic search - Find similar memories
3. Multiple retrieval strategies exist - Similarity, hybrid, re-ranking
4. Performance matters - Indexing, partitioning, approximation
5. Metadata filtering powerful - Combine semantic + structured search
6. Monitoring quality critical - Track recall, MRR, NDCG

---

Next Steps

• Read Memory Compression - Making memories efficient
• Read Episodic & Procedural - Memory types

---

Last Updated: August 9, 2026