The Write Path & the Read Path

# Colab cell 1 — run once. Set your key in the 🔑 panel (name it
# ANTHROPIC_API_KEY) or just paste it when prompted.
!pip install -q anthropic sentence-transformers

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
except Exception:
    from getpass import getpass
    os.environ.setdefault("ANTHROPIC_API_KEY", getpass("Anthropic API key: "))

import sqlite3, json, time
import numpy as np
import anthropic
from sentence_transformers import SentenceTransformer

client = anthropic.Anthropic()
encoder = SentenceTransformer("all-MiniLM-L6-v2")

# Lesson 3's MemoryStore, unchanged, so this notebook stands alone.
SCHEMA = """
CREATE TABLE IF NOT EXISTS memories (
    id            INTEGER PRIMARY KEY,
    fact          TEXT NOT NULL,
    provenance    TEXT NOT NULL,
    created_at    REAL NOT NULL,
    importance    REAL NOT NULL DEFAULT 0.5,
    superseded_by INTEGER,
    embedding     TEXT NOT NULL
);
"""

class MemoryStore:
    def __init__(self, path: str = "memory.db"):
        self.db = sqlite3.connect(path)
        self.db.executescript(SCHEMA)

    def add(self, fact: str, provenance: dict, importance: float = 0.5) -> int:
        vec = encoder.encode(fact, normalize_embeddings=True)
        cur = self.db.execute(
            "INSERT INTO memories (fact, provenance, created_at, importance, embedding) "
            "VALUES (?, ?, ?, ?, ?)",
            (fact, json.dumps(provenance), time.time(), importance,
             json.dumps(vec.tolist())),
        )
        self.db.commit()
        return cur.lastrowid

    def all_active(self) -> list[dict]:
        rows = self.db.execute(
            "SELECT id, fact, provenance, created_at, importance, embedding "
            "FROM memories WHERE superseded_by IS NULL").fetchall()
        return [{"id": r[0], "fact": r[1], "provenance": json.loads(r[2]),
                 "created_at": r[3], "importance": r[4],
                 "vec": np.array(json.loads(r[5]))} for r in rows]

def log_conflict(old: dict, new_id: int) -> None:
    print(f"[conflict] memory #{old['id']} superseded by #{new_id}")

DUP_THRESHOLD = 0.90      # near-identical: skip
TOPIC_THRESHOLD = 0.70    # same topic: check for contradiction

def judge_contradiction(new_fact: str, old_fact: str) -> str:
    resp = client.messages.create(
        model="claude-sonnet-5", max_tokens=10,
        messages=[{"role": "user", "content":
            "Do these two statements contradict each other? "
            "Answer only CONTRADICTS or COMPATIBLE.\n"
            f"A: {old_fact}\nB: {new_fact}"}],
    )
    verdict = next(b.text for b in resp.content if b.type == "text")
    return verdict.strip().upper()

def write_fact(store: MemoryStore, candidate: dict) -> str:
    vec = encoder.encode(candidate["fact"], normalize_embeddings=True)
    for mem in store.all_active():
        sim = float(vec @ mem["vec"])
        if sim >= DUP_THRESHOLD:
            return f"skipped duplicate of #{mem['id']}"
        if sim >= TOPIC_THRESHOLD:
            if judge_contradiction(candidate["fact"], mem["fact"]) == "CONTRADICTS":
                new_id = store.add(candidate["fact"], candidate["provenance"],
                                   candidate["importance"])
                store.db.execute(
                    "UPDATE memories SET superseded_by = ? WHERE id = ?",
                    (new_id, mem["id"]))
                store.db.commit()
                log_conflict(old=mem, new_id=new_id)   # surface, don't hide
                return f"stored #{new_id}, superseded #{mem['id']} (conflict flagged)"
    new_id = store.add(candidate["fact"], candidate["provenance"],
                       candidate["importance"])
    return f"stored #{new_id}"

# demo: the three zones — store, skip-duplicate, supersede-on-contradiction
store = MemoryStore(":memory:")
prov = {"type": "session_extraction", "source_type": "user_direct", "session": "demo"}
for fact in ["User deploys to production on Fridays.",
             "User deploys to production on Fridays.",
             "User no longer deploys on Fridays; deploys moved to Wednesdays."]:
    print(write_fact(store, {"fact": fact, "provenance": prov, "importance": 0.7}))

# Colab cell 2 — run cell 1 first (it defines store, encoder, and the
# demo's three writes — recall below draws from that store).
W_RELEVANCE, W_RECENCY, W_IMPORTANCE = 0.60, 0.25, 0.15
HALF_LIFE_DAYS = 30.0

def recall(store: MemoryStore, query_text: str, k: int = 5,
           min_score: float = 0.35) -> list[dict]:
    q = encoder.encode(query_text, normalize_embeddings=True)
    now = time.time()
    scored = []
    for mem in store.all_active():
        relevance = float(q @ mem["vec"])                     # [-1, 1]
        age_days = (now - mem["created_at"]) / 86_400
        recency = 0.5 ** (age_days / HALF_LIFE_DAYS)          # (0, 1]
        score = (W_RELEVANCE * relevance
                 + W_RECENCY * recency
                 + W_IMPORTANCE * mem["importance"])
        scored.append((score, mem))
    scored.sort(key=lambda pair: pair[0], reverse=True)
    return [m for s, m in scored[:k] if s >= min_score]       # floor matters

# injected via assemble_window() from Lesson 1 — fenced, labeled untrusted
memories = [m["fact"] for m in recall(store, "plan this week's production deploy")]
print(memories)   # the Wednesday fact wins; the superseded Friday fact never appears