UVF IT: Build a Basic Agent Memory Store with Hindsight in Python

Welcome to another official UVF IT guide! In this quick build you’ll create a tiny agent‑memory service inspired by Hindsight, the open‑source memory store for smarter agents. We’ll spin up a local FastAPI server, persist embeddings in SQLite, and expose a tiny REST API – all in under an hour.

The clone focuses on the core concepts: storing a vector (as a JSON list) with a key, and retrieving the most similar entries via a simple cosine similarity query. No external cloud services or paid APIs are required.

Before you start…

You need Python 3.10+ and pip. The real Hindsight supports massive vector indexes; our SQLite version is only for demo purposes and won’t scale.
Make sure no other process is using port 8000 – the server will bind there.
If you see sqlite3.OperationalError: database is locked, stop the server and retry – the simple demo doesn’t handle concurrent writes.

Component Overview

Phase 1: Scaffold the project

We start by isolating our work in a virtual environment so dependencies don’t clash with other Python projects. Then we install FastAPI for the web layer, Uvicorn as the ASGI server, and the built‑in SQLite driver (via the standard library) – nothing extra is required.

All files will live in a fresh folder called hindsight_clone; this keeps the demo tidy and makes it easy to delete later.

mkdir hindsight_clone && cd hindsight_clone
python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install fastapi uvicorn

Phase 2: Initialise the SQLite store

Next we create a tiny SQLite database with a single table to hold each memory entry. The table stores an auto‑increment id, the vector as a JSON‑encoded list, and an optional metadata JSON column for future extensions.

Running this script will create memory.db in the project root; the FastAPI app will later reuse the same file for reads and writes.

db_init.py

import sqlite3, json

conn = sqlite3.connect('memory.db')
cur = conn.cursor()
cur.execute('''
CREATE TABLE IF NOT EXISTS memories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    vector TEXT NOT NULL,          -- JSON array of floats
    metadata TEXT                  -- optional JSON object
)
''')
conn.commit()
conn.close()
print('SQLite store initialised: memory.db')

Phase 3: Implement FastAPI endpoints

In this step we wire up a tiny FastAPI app that talks to a local SQLite file. The /store endpoint will accept a JSON payload containing an id and a numeric vector, then persist it. The /search endpoint reads the query vector from the q query‑string, computes cosine similarity against all stored vectors, and returns the three most similar records as JSON. Keeping everything in a single file makes the clone easy to run and extend.

main.py

import json, math, sqlite3
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel

app = FastAPI()
DB = 'memory.db'

# Ensure table exists
with sqlite3.connect(DB) as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS vectors (
        id TEXT PRIMARY KEY,
        vec TEXT NOT NULL
    )''')

class StoreItem(BaseModel):
    id: str
    vector: list[float]

def cosine(a, b):
    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(y*y for y in b))
    return dot/(norm_a*norm_b) if norm_a and norm_b else 0.0

@app.post('/store')
def store(item: StoreItem):
    vec_json = json.dumps(item.vector)
    try:
        with sqlite3.connect(DB) as conn:
            conn.execute('INSERT OR REPLACE INTO vectors (id, vec) VALUES (?,?)',
                         (item.id, vec_json))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {'status':'ok'}

@app.get('/search')
def search(q: str = Query(..., description='comma‑separated floats')):
    try:
        query_vec = [float(x) for x in q.split(',')]
    except ValueError:
        raise HTTPException(status_code=400, detail='Invalid vector format')
    results = []
    with sqlite3.connect(DB) as conn:
        for row in conn.execute('SELECT id, vec FROM vectors'):
            stored_vec = json.loads(row[1])
            sim = cosine(query_vec, stored_vec)
            results.append({'id': row[0], 'score': sim})
    top3 = sorted(results, key=lambda x: x['score'], reverse=True)[:3]
    return top3

Phase 4: Run the server locally

Now that the API code is ready, we launch it with Uvicorn, the recommended ASGI server for FastAPI. The --reload flag watches source changes so you can iterate quickly. The server will be reachable at http://localhost:8000.

uvicorn main:app --reload

Phase 5: Verify and extend

With the server running, we can use curl to store a sample vector and then query for similar ones. The response should be a JSON array containing the stored record with a high similarity score. From here you might add a tiny HTML front‑end, batch upserts, or swap SQLite for PostgreSQL for a more robust clone.

# Store a vector
curl -X POST http://localhost:8000/store \
  -H 'Content-Type: application/json' \
  -d '{"id":"vec1","vector":[0.1,0.2,0.3]}'

# Query for similar vectors
curl 'http://localhost:8000/search?q=0.1,0.2,0.3' | jq '.'

Verifying the Stack

Check	Command / Why it matters
Server starts without errors	uvicorn main:app --reload shows 'Uvicorn running on http://127.0.0.1:8000'
POST /store returns status ok	curl -X POST http://localhost:8000/store -H 'Content-Type: application/json' -d '{\"id\":\"vec1\",\"vector\":[0.1,0.2,0.3]}' \| grep ok
GET /search returns a JSON array with at least one entry	curl 'http://localhost:8000/search?q=0.1,0.2,0.3' \| jq '.'

What’s next?

You now have a working local memory store. Try adding batch upserts, a simple front‑end UI, or switching to PostgreSQL for a more production‑ready clone of Hindsight.