Build a lightweight vector database for AI context sharing
Welcome to another official UVF IT build guide. Today we are creating a minimal clone of FastRecall, a service that allows different AI models to share context seamlessly.
We will build a Python-based memory store using SQLite and simple vector math. This project demonstrates how to store, retrieve, and manage semantic memory without relying on expensive external APIs.
pip installed. We use faiss-cpu for vector similarity, which may take a moment to install.Initialize the project directory and install the core libraries: FastAPI for the API, FAISS for vector search, and SQLite3 (built-in) for metadata.
We will create a simple structure with a main.py for the API and a memory.py for the core logic.
mkdir -p uvf-memory-store/src
cd uvf-memory-store
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows use: venv\Scripts\activate
# Install dependencies
pip install fastapi uvicorn faiss-cpu numpy
# Initialize git
git init
touch src/__init__.py
echo "__pycache__/\nvenv/\n*.pyc" > .gitignoreImplement the MemoryStore class in src/memory.py.
This class handles two main tasks: storing text with a generated vector and searching for similar text using cosine similarity via FAISS.
We use a simple hash-based embedding function for this demo to avoid API costs. In a real app, you'd replace embed_text with a call to an LLM.
src/memory.py
import hashlib
import sqlite3
import numpy as np
import faiss
from typing import List, Dict, Any, Optional
import uuid
class MemoryStore:
def __init__(self, db_path: str = "memory.db", dim: int = 128):
self.db_path = db_path
self.dim = dim
self.index = faiss.IndexFlatIP(dim) # Inner Product for cosine similarity (normalized vectors)
self._init_db()
def _init_db(self):
"""Initialize SQLite database with metadata table."""
self.conn = sqlite3.connect(self.db_path)
self.cursor = self.conn.cursor()
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS memories (
id TEXT PRIMARY KEY,
text TEXT NOT NULL,
metadata TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
def embed_text(self, text: str) -> np.ndarray:
"""
Generate a deterministic vector from text using hashing.
This is a placeholder for real LLM embeddings.
"""
vector = np.zeros(self.dim, dtype=np.float32)
# Use multiple hashes to fill the vector space
for i in range(self.dim):
h = hashlib.md5(f"{text}_{i}".encode('utf-8')).hexdigest()
# Convert hex to float in range [-1, 1]
val = (int(h[:8], 16) / 0xFFFFFFFF) * 2 - 1
vector[i] = val
# Normalize vector for cosine similarity (FAISS IndexFlatIP expects normalized vectors for cosine)
norm = np.linalg.norm(vector)
if norm > 0:
vector = vector / norm
return vector
def add_memory(self, text: str, metadata: Optional[Dict[str, Any]] = None) -> str:
"""Store a new memory and return its ID."""
memory_id = str(uuid.uuid4())
vector = self.embed_text(text)
# Store in FAISS
self.index.add(vector.reshape(1, -1))
# Store metadata in SQLite
import json
metadata_json = json.dumps(metadata) if metadata else None
self.cursor.execute(
"INSERT INTO memories (id, text, metadata) VALUES (?, ?, ?)",
(memory_id, text, metadata_json)
)
self.conn.commit()
return memory_id
def search_memories(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
"""Search for similar memories and return with metadata."""
query_vector = self.embed_text(query).reshape(1, -1)
# Search FAISS for top_k similar vectors
scores, indices = self.index.search(query_vector, top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx == -1: # Invalid index
continue
# Get the ID from FAISS index (FAISS stores IDs sequentially, but we need to map to our UUIDs)
# For this simple demo, we assume the order in FAISS matches insertion order.
# In a production system, you'd maintain a mapping table between FAISS index and UUID.
# Here, we fetch all IDs from SQLite in insertion order to map.
self.cursor.execute("SELECT id FROM memories ORDER BY rowid LIMIT ?", (idx + 1,))
rows = self.cursor.fetchall()
if not rows:
continue
memory_id = rows[-1][0] # Last row is the one at index idx
# Fetch full memory details
self.cursor.execute(
"SELECT text, metadata FROM memories WHERE id = ?",
(memory_id,)
)
row = self.cursor.fetchone()
if row:
import json
meta = json.loads(row[1]) if row[1] else {}
results.append({
"id": memory_id,
"text": row[0],
"metadata": meta,
"score": float(score)
})
return results
def close(self):
"""Close database connection."""
if self.conn:
self.conn.close()Create the FastAPI application in src/main.py.
Expose two endpoints: POST /memories to store new context and GET /memories/search to retrieve relevant context.
The API wraps the MemoryStore class and handles JSON serialization.
src/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
from src.memory import MemoryStore
app = FastAPI(title="UVF IT Memory Store")
# Initialize the memory store
store = MemoryStore()
class MemoryCreate(BaseModel):
text: str
metadata: Optional[Dict[str, Any]] = None
class MemorySearch(BaseModel):
query: str
top_k: int = 5
class MemoryResponse(BaseModel):
id: str
text: str
metadata: Optional[Dict[str, Any]]
score: float
@app.get("/health")
def health_check():
return {"status": "ok"}
@app.post("/memories", response_model=Dict[str, str])
def create_memory(memory: MemoryCreate):
"""Store a new memory."""
if not memory.text.strip():
raise HTTPException(status_code=400, detail="Text cannot be empty")
try:
memory_id = store.add_memory(memory.text, memory.metadata)
return {"id": memory_id, "status": "created"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/memories/search", response_model=List[MemoryResponse])
def search_memories(query: str, top_k: int = 5):
"""Search for similar memories."""
if not query.strip():
raise HTTPException(status_code=400, detail="Query cannot be empty")
try:
results = store.search_memories(query, top_k)
return results
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.on_event("shutdown")
def shutdown_event():
store.close()Now that the code is written, it is time to bring the application to life. We will use Uvicorn, the ASGI server, to run our FastAPI application.
Run the command below from your project root directory. This will start the server on port 8000, which is the standard port for local development.
uvicorn src.main:app --reload --port 8000With the server running, we can verify that the memory store works as expected. We will use curl to simulate the requests that a client application would make.
First, we store a specific memory. Then, we search for it using a semantically similar query. If the system is working, the search should return the original memory with a high similarity score.
curl -X POST http://localhost:8000/memories \
-H "Content-Type: application/json" \
-d '{"text": "The capital of France is Paris", "metadata": {"category": "geography"}}'
curl -X GET "http://localhost:8000/memories/search?query=What is the capital of France?&limit=1"| Check | Command / Why it matters |
|---|---|
| Server starts without errors | Run curl http://localhost:8000/health and expect {"status": "ok"} |
| Memory can be stored | POST to /memories returns a JSON object with an id field |
| Search returns relevant results | GET to /memories/search with a matching query returns the stored text in results[0].text |
| Metadata is preserved | The metadata field in the search response matches what was sent in the POST request |
You now have a working cross-model memory store. To extend this, replace the hash-based embedding with a real LLM embedding API (like OpenAI or HuggingFace), add authentication, and implement distributed storage for production use.