Vector Stores & Semantic Search

How Embeddings Work

An embedding model converts text into a dense vector, a list of floating-point numbers (typically 768–3072 dimensions depending on the model). The key property: texts with similar meanings produce vectors that are close together in high-dimensional space.

"Python async programming" → [0.12, -0.89, 0.34, ... 1536 numbers]
"Concurrent Python code"   → [0.14, -0.87, 0.31, ... very similar!]
"Cooking pasta recipes"    → [0.67, 0.21, -0.45, ... very different]

Cosine similarity between two vectors measures how semantically related they are:

• 1.0 = identical meaning
• 0.8+ = highly related
• 0.5 = loosely related
• 0.0 = unrelated

This is the foundation of all semantic search and RAG pipelines.

Generating Embeddings with OpenAI

The two functions below cover the two most common embedding patterns: single-text for real-time queries, and batched for indexing large document collections efficiently.

# src/memory/embeddings.py
from openai import AsyncOpenAI
import numpy as np

client = AsyncOpenAI()

async def embed_text(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """Generate an embedding for a single text."""
    response = await client.embeddings.create(
        model=model,
        # text-embedding-3-small: 1536 dimensions, best price/quality ratio for most RAG use cases
        input=text,
        encoding_format="float",  # return raw floats, not base64
    )
    return response.data[0].embedding

async def embed_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """Embed multiple texts in one API call (more efficient)."""
    response = await client.embeddings.create(
        model=model,
        input=texts,  # up to 2048 texts per call, batching reduces API round-trips significantly
        encoding_format="float",
    )
    # response.data preserves the same order as the input list
    return [item.embedding for item in response.data]

# Embedding model comparison
EMBEDDING_MODELS = {
    "text-embedding-3-small": {
        "dimensions": 1536,
        "cost_per_1M_tokens": 0.02,
        "use_case": "General purpose, cost-effective"
    },
    "text-embedding-3-large": {
        "dimensions": 3072,           # 2x the dimensions means finer semantic distinctions
        "cost_per_1M_tokens": 0.13,   # 6.5x more expensive, only worth it for high-accuracy retrieval
        "use_case": "High accuracy retrieval"
    },
}

Setting Up Qdrant as Long-term Memory

This class wraps Qdrant to give agents a simple store/search/delete interface, hiding the vector database details behind a memory-like API.

# src/memory/vector_store.py
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue, Range,
)
import uuid
from datetime import datetime

class AgentMemoryStore:
    """Vector-based long-term memory for AI agents."""
    
    def __init__(self, url: str, collection_name: str = "agent_memory"):
        self.client = AsyncQdrantClient(url=url)
        self.collection = collection_name
        self.embedding_dim = 1536  # must match the embedding model's output size
    
    async def initialize(self):
        """Create the collection if it doesn't exist."""
        collections = await self.client.get_collections()
        exists = any(c.name == self.collection for c in collections.collections)
        
        if not exists:
            await self.client.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(
                    size=self.embedding_dim,
                    distance=Distance.COSINE,  # cosine similarity is standard for text embeddings
                ),
            )
            print(f"Created collection: {self.collection}")
    
    async def store_memory(
        self,
        content: str,
        metadata: dict,
        memory_id: str | None = None,
    ) -> str:
        """Store a memory with its embedding and metadata."""
        embedding = await embed_text(content)
        memory_id = memory_id or str(uuid.uuid4())  # generate a unique ID if not provided
        
        await self.client.upsert(
            collection_name=self.collection,
            points=[
                PointStruct(
                    id=memory_id,
                    vector=embedding,       # the semantic fingerprint of the content
                    payload={
                        "content": content,
                        "created_at": datetime.utcnow().isoformat(),
                        **metadata,  # user_id, topic, source, etc., used for filtering later
                    }
                )
            ]
        )
        return memory_id
    
    async def search(
        self,
        query: str,
        top_k: int = 5,
        score_threshold: float = 0.7,
        filter_by: dict | None = None,
    ) -> list[dict]:
        """Semantic search with optional metadata filtering."""
        query_embedding = await embed_text(query)  # embed the query the same way as stored content
        
        # Build metadata filter
        qdrant_filter = None
        if filter_by:
            conditions = [
                # FieldCondition filters by metadata: only returns documents matching these key/value pairs
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filter_by.items()
            ]
            qdrant_filter = Filter(must=conditions)  # must = AND logic; use should for OR
        
        results = await self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding,
            limit=top_k,                    # top_k=5 limits results to keep context window manageable
            score_threshold=score_threshold, # cosine_similarity > 0.7 filters out loosely related results
            query_filter=qdrant_filter,
            with_payload=True,              # include the stored metadata alongside scores
        )
        
        return [
            {
                "id": r.id,
                "content": r.payload["content"],
                "score": r.score,           # cosine similarity score between 0 and 1
                "metadata": {k: v for k, v in r.payload.items() if k != "content"},
            }
            for r in results
        ]
    
    async def delete_memory(self, memory_id: str):
        await self.client.delete(
            collection_name=self.collection,
            points_selector=[memory_id],
        )

Using the Memory Store in an Agent

This function demonstrates the three-phase memory loop: retrieve relevant past context, call the LLM with that context injected, then store the new interaction for future retrieval.

# src/agents/memory_agent.py
memory_store = AgentMemoryStore(url="http://localhost:6333")

async def agent_with_memory(user_id: str, query: str) -> str:
    # 1. Retrieve relevant memories before calling the LLM
    relevant_memories = await memory_store.search(
        query=query,
        top_k=5,                           # fetch the 5 most semantically similar memories
        score_threshold=0.75,              # 0.75 is a good starting threshold, tune per domain
        filter_by={"user_id": user_id},   # scope search to this user's memories only
    )
    
    memory_context = ""
    if relevant_memories:
        memory_context = "Relevant past context:\n" + "\n".join([
            # include the date so the LLM can reason about recency
            f"- [{m['metadata']['created_at'][:10]}] {m['content']}"
            for m in relevant_memories
        ])
    
    # 2. Call the agent with memory context injected into the system prompt
    messages = [
        SystemMessage(content=f"""You are a helpful assistant with access to past interactions.
        
{memory_context}

Use the above context if relevant to the current query."""),
        HumanMessage(content=query),
    ]
    
    response = llm.invoke(messages)
    
    # 3. Store the new interaction so it's retrievable in future conversations
    await memory_store.store_memory(
        content=f"User asked: {query}\nAssistant answered: {response.content}",
        metadata={
            "user_id": user_id,
            "topic": classify_topic(query),  # tag by topic for more targeted filtering later
            "session_id": str(uuid.uuid4()),
        }
    )
    
    return response.content

Vector Store Comparison

Production recommendation: Default to Qdrant. It runs embedded on disk with zero infrastructure for local development (exactly like you’d reach for pgvector), then scales to a self-hosted or managed cluster without a migration when you need it, unlike pgvector, which hits a hard ceiling around 10M vectors and rising query latency well before that. Reach for pgvector specifically when a vector column inside your existing Postgres database, sharing transactions with your relational data, matters more than retrieval performance. Pinecone is worth it only when you want zero ops and are fine paying for it.

Hybrid Search: Combining Semantic + Keyword

For production retrieval, combining semantic and keyword search significantly improves recall:

from qdrant_client.models import SparseVector, SparseVectorParams

async def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    """BM25 + semantic search with Reciprocal Rank Fusion."""
    # Dense (semantic) results: captures conceptual similarity
    semantic_results = await memory_store.search(query, top_k=top_k * 2)  # fetch 2x to have room to fuse
    
    # Sparse (keyword) results via BM25: captures exact term matches
    # Qdrant natively supports sparse vectors for BM25
    sparse_results = await keyword_search(query, top_k=top_k * 2)
    
    # Reciprocal Rank Fusion: weight results by their rank position, not raw score
    # RRF is model-agnostic and handles score scale differences between dense and sparse results
    scores = {}
    for rank, result in enumerate(semantic_results):
        # 1 / (60 + rank) is the standard RRF formula; 60 is a smoothing constant
        scores[result["id"]] = scores.get(result["id"], 0) + 1 / (60 + rank)
    for rank, result in enumerate(sparse_results):
        scores[result["id"]] = scores.get(result["id"], 0) + 1 / (60 + rank)
    
    # Sort by fused score and return top K
    fused = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
    return [get_result_by_id(id) for id, _ in fused]

Hybrid search is standard in production retrieval pipelines: semantic search catches conceptual matches, BM25 catches exact term matches, and their combination significantly outperforms either alone.