How to add persistent memory to a chatbot agent

Most LLM chatbots reset to a blank slate on every restart because the transcript lives only in process memory. A production persistent memory chatbot agent stores user facts, prior turns, and derived context in an external store, then pulls the relevant slice back before each inference. This article gives you a concrete Python implementation using Postgres and vector search that you can drop into an existing agent loop.

Step 1: Provision a Postgres instance with pgvector

You need a relational store that can also do cosine similarity without bolting on a separate vector database. pgvector adds a vector type to Postgres and handles millions of rows with an ivfflat or HNSW index. It is the lowest-friction option if you already run Postgres.

Spin up a local container for dev:

docker run -d --name memdb -p 5432:5432 -e POSTGRES_PASSWORD=secret postgres:16
docker exec -it memdb psql -U postgres -c "CREATE EXTENSION IF NOT EXISTS vector;"

On a managed cloud, enable the extension via the console SQL editor. Pick embedding dimension up front—1536 matches OpenAI’s text-embedding-3-small and text-embedding-3-large (smaller variant). Changing dimensions later requires a schema migration and re-embedding.

Step 2: Define the memory schema

Separate raw conversation turns from extracted facts. Raw turns give replayability and debugging; facts give low-latency retrieval without scanning every message.

CREATE TABLE IF NOT EXISTS memories (
    id BIGSERIAL PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    session_id TEXT NOT NULL,
    content TEXT NOT NULL,
    embedding vector(1536),
    kind TEXT NOT NULL DEFAULT 'turn',
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS memories_embedding_idx
    ON memories USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

CREATE INDEX IF NOT EXISTS memories_session_idx
    ON memories (tenant_id, session_id, kind);

The tenant_id column lets you serve multiple bots or customers from one table with hard isolation. The ivfflat index with lists = 100 works well up to ~1M rows; for larger datasets switch to HNSW (USING hnsw). Always filter by tenant_id in queries to keep the index scan narrow.

Step 3: Implement the memory store interface

Write a small class that handles inserts, recent turns, and similarity queries. Use psycopg2 for sync code; swap to asyncpg if your agent is async.

import psycopg2

class MemoryStore:
    def __init__(self, dsn: str):
        self.conn = psycopg2.connect(dsn)

    def add(self, tenant_id, session_id, content, embedding, kind="turn"):
        with self.conn.cursor() as cur:
            cur.execute(
                "INSERT INTO memories (tenant_id, session_id, content, embedding, kind) "
                "VALUES (%s, %s, %s, %s, %s)",
                (tenant_id, session_id, content, embedding, kind)
            )
        self.conn.commit()

    def recent_turns(self, tenant_id, session_id, limit=4):
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT content FROM memories "
                "WHERE tenant_id = %s AND session_id = %s AND kind = 'turn' "
                "ORDER BY created_at DESC LIMIT %s",
                (tenant_id, session_id, limit)
            )
            return list(reversed([r[0] for r in cur.fetchall()]))

    def similar(self, tenant_id, embedding, limit=5, kind=None):
        with self.conn.cursor() as cur:
            if kind:
                cur.execute(
                    "SELECT content FROM memories "
                    "WHERE tenant_id = %s AND kind = %s "
                    "ORDER BY embedding <=> %s LIMIT %s",
                    (tenant_id, kind, embedding, limit)
                )
            else:
                cur.execute(
                    "SELECT content FROM memories "
                    "WHERE tenant_id = %s "
                    "ORDER BY embedding <=> %s LIMIT %s",
                    (tenant_id, embedding, limit)
                )
            return [r[0] for r in cur.fetchall()]

The <=> operator is pgvector’s cosine distance. Lower is closer. Wrap these calls in a retry decorator if you run against a flaky network.

Step 4: Generate embeddings for each memory

You need a consistent embedding model. text-embedding-3-small outputs 1536 dims at low cost. Call it via any OpenAI-compatible endpoint.

import openai

client = openai.OpenAI(
    base_url="https://api.n4n.ai/v1",  # OpenAI-compatible, 240+ models, auto fallback
    api_key="YOUR_KEY"
)

def embed(text: str) -> list[float]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return resp.data[0].embedding

Batch embed when writing many facts at once: pass a list to input and iterate resp.data. Never embed the system prompt with user data unless you want cross-contamination of namespaces. Store the raw vector; if you later change models, add a model_version column and filter on it.

Step 5: Extract facts from conversations

Storing every turn is not enough; a persistent memory chatbot agent should distill durable facts so retrieval stays signal-rich. Use the LLM to summarize new info after each exchange.

EXTRACT_PROMPT = (
    "Extract user-specific durable facts (name, preferences, constraints, deadlines) "
    "as short bullet points. Ignore chit-chat and speculative statements."
)

def extract_facts(transcript: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": EXTRACT_PROMPT},
            {"role": "user", "content": transcript}
        ]
    )
    return resp.choices[0].message.content

After each user/assistant pair, call extract_facts, split into bullets, embed each, and store with kind='fact'. This keeps the working set small. A persistent memory chatbot agent that only appends turns will eventually blow up prompt size and latency.

Step 6: Retrieve and inject memory into the agent loop

Before calling the model for a response, pull recent turns and top similar facts, then prepend as context.

def build_context(store, tenant_id, session_id, query_embedding):
    recent = store.recent_turns(tenant_id, session_id, limit=4)
    facts = store.similar(tenant_id, query_embedding, limit=3, kind="fact")
    return "\n".join(recent + facts)

def handle_message(store, tenant_id, session_id, user_msg):
    emb = embed(user_msg)
    ctx = build_context(store, tenant_id, session_id, emb)
    messages = [
        {"role": "system", "content": f"Known memory:\n{ctx}"},
        {"role": "user", "content": user_msg}
    ]
    reply = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
    # persist both turns
    store.add(tenant_id, session_id, user_msg, emb, kind="turn")
    store.add(tenant_id, session_id, reply.choices[0].message.content,
              embed(reply.choices[0].message.content), kind="turn")
    # async fact extraction
    facts = extract_facts(f"{user_msg}\n{reply.choices[0].message.content}")
    for f in facts.splitlines():
        if f.strip():
            store.add(tenant_id, session_id, f, embed(f), kind="fact")
    return reply.choices[0].message.content

The agent now answers with awareness of prior context even after a process restart, because nothing lives in RAM.

Step 7: Manage memory lifecycle

Unbounded growth hurts latency and cost. Add a periodic job to prune old turns while keeping facts unless invalidated.

DELETE FROM memories
WHERE kind = 'turn'
  AND created_at < now() - interval '30 days';

When a user corrects a fact (“I’m vegetarian, not vegan”), store a new fact and mark the old stale. The simplest robust pattern is an invalidated_at timestamp and a query that picks the latest fact per content hash. For a persistent memory chatbot agent, treat facts as mutable state, not append-only log.

Also monitor index health: as rows grow, increase ivfflat lists or migrate to HNSW. Run EXPLAIN ANALYZE on your similar query monthly.

Step 8: Verify the persistent memory chatbot agent works

Write an integration test that simulates a restart by recreating the store client against the same database.

def test_memory_persistence():
    store = MemoryStore("postgresql://postgres:secret@localhost:5432/postgres")
    e1 = embed("My name is Ada")
    store.add("t1", "s1", "My name is Ada", e1, kind="fact")
    # simulate process restart: new client, same backing store
    store2 = MemoryStore("postgresql://postgres:secret@localhost:5432/postgres")
    query = embed("What's my name?")
    results = store2.similar("t1", query, limit=1, kind="fact")
    assert "Ada" in results[0]

    # verify turn recall
    store2.add("t1", "s1", "I like trains", embed("I like trains"), kind="turn")
    recent = store2.recent_turns("t1", "s1", limit=1)
    assert "trains" in recent[0]

Run with pytest. For end-to-end confidence, launch the bot, send “I’m working on a Rust project”, kill the process, restart, then ask “What language was my project in?” If the reply says Rust, the persistent memory chatbot agent is functioning.

Load test with 10k synthetic memories and confirm similar returns under 20 ms on a warmed index. If not, tune the index or reduce limit.

You now have a memory layer that survives restarts, scales across sessions, and retrieves context without stuffing the entire history into the prompt.