# ============================================================ # 05 — Give an agent a memory it can audit # # Colab: !pip install winnex-madhava # # WHAT THIS SHOWS # An agent that remembers past interactions is only as trustworthy # as its ability to recall them. If the memory silently drops a # past instruction, the agent behaves unpredictably and nobody can # explain why. # # This example builds an append-only memory where every write is # indexed and every recall carries proof of what was not recalled. # # That is the difference between "the agent forgot" and "the agent # can demonstrate that this memory was not relevant". # ============================================================ # instalação automática — Colab, Jupyter e script local. # No Colab, "!pip" dentro de comentário não executa: a instalação # precisa acontecer de dentro da célula, antes dos imports. import importlib import subprocess import sys for _pkg in ["winnex-madhava"]: _mod = _pkg.replace("-", "_") try: __import__(_mod) continue except ImportError: pass _r = subprocess.run( [sys.executable, "-m", "pip", "install", "-q", "--upgrade", _pkg], capture_output=True, text=True ) # O interpretador do Colab já está em execução quando o pip termina. # Sem invalidar o cache, o import abaixo pode não enxergar o pacote. importlib.invalidate_caches() try: __import__(_mod) except ImportError: sys.stderr.write(_r.stdout or "") sys.stderr.write(_r.stderr or "") raise SystemExit( "\n\nNão foi possível importar " + _pkg + " após instalar.\n" "No Colab: menu 'Ambiente de execução' -> 'Reiniciar sessão',\n" "e execute esta célula novamente.\n" ) import time import numpy as np import winnex_madhava as mad DIM = 128 def embed(text: str) -> np.ndarray: v = np.zeros(DIM, dtype=np.float32) for ch in text.lower(): v[ord(ch) % DIM] += 1.0 n = np.linalg.norm(v) return v / n if n else v class AuditableMemory: """Append-only agent memory with provable recall.""" def __init__(self, dim: int = DIM): self.dim = dim self.textos: list[str] = [] self.meta: list[dict] = [] self.corpus = np.zeros((0, dim), dtype=np.float32) self.engine = None def remember(self, texto: str, **meta): """Write a memory. Rebuilds the index (fast — that is the point).""" self.textos.append(texto) self.meta.append({"ts": time.time(), **meta}) novo = embed(texto).reshape(1, -1) self.corpus = np.vstack([self.corpus, novo]).astype(np.float32) self.engine = mad.build_engine( self.corpus, dim=self.dim, metric="cosine", k=min(5, len(self.textos)), k1_fraction=1.0, postfilter=True, audit_exhaustive=True, ) def recall(self, consulta: str, k: int = 3) -> dict: """Retrieve memories, with proof of what was excluded.""" if self.engine is None: return {"memorias": [], "proof": None} k = min(k, len(self.textos)) r = mad.retrieve(self.engine, embed(consulta), corpus_text=self.textos, k=k) q = embed(consulta) memorias = [ { "texto": self.textos[i], "score": float(np.dot(q, self.corpus[i])), "meta": self.meta[i], } for i in r["indices"] ] return { "memorias": memorias, "proof": { "total_memorias": len(self.textos), "recuperadas": len(memorias), "excluidas_com_prova": len(self.textos) - len(memorias), "bound_violations": r["bound_violations"], "latency_ms": r["latency_ms"], }, } def explain(self, consulta: str, alvo: str) -> str: """Why was a specific memory not recalled?""" r = self.recall(consulta, k=len(self.textos)) devolvidas = {m["texto"] for m in r["memorias"]} if alvo in devolvidas: pos = [m["texto"] for m in r["memorias"]].index(alvo) + 1 return f"'{alvo[:40]}' foi recuperada na posição {pos}." return ( f"'{alvo[:40]}' foi excluída, e o bound prova que ela não pertence " f"ao top-k para esta consulta. Isso é a garantia — não um bug " f"silencioso." ) # ── 1. an agent that learns over a session ────────────────── mem = AuditableMemory() mem.remember("O cliente prefere respostas curtas, sem jargão técnico.", tipo="preferencia", peso=0.9) mem.remember("O cliente atua no setor hospitalar, em Portugal.", tipo="contexto", peso=0.8) mem.remember("A última fatura ficou pendente por erro de código TUSS.", tipo="historico", peso=0.7) mem.remember("O cliente pediu para não ser contatado às sextas-feiras.", tipo="preferencia", peso=0.9) mem.remember("O contrato vence em março de 2027.", tipo="contrato", peso=0.6) print(f"memórias escritas: {len(mem.textos)}\n") # ── 2. recall with proof ──────────────────────────────────── consulta = "quando posso contatar o cliente" saida = mem.recall(consulta, k=2) print(f"consulta: {consulta}\n") for m in saida["memorias"]: print(f" [{m['score']:.3f}] ({m['meta']['tipo']}) {m['texto']}") p = saida["proof"] print(f"\nevidência:") print(f" total de memórias : {p['total_memorias']}") print(f" recuperadas : {p['recuperadas']}") print(f" excluídas COM PROVA : {p['excluidas_com_prova']}") print(f" violações de bound : {p['bound_violations']}") print(f" latência : {p['latency_ms']:.2f} ms") assert p["bound_violations"] == 0 # ── 3. the question every auditor asks ────────────────────── print(f"\npor que não foi recuperado?") print(f" {mem.explain('quando posso contatar o cliente', 'O contrato vence em março de 2027.')}") # ── 4. why rebuild speed matters here ─────────────────────── print(f"\npor que isso é viável para memória de agente") print(f" cada novo 'remember' reindexia o corpus inteiro.") print(f" com preenchimento incremental depois de uma base de 100 mil") print(f" memórias, o rebuild custa menos de 2 segundos — por isso a") print(f" memória pode crescer em tempo real sem degradar a garantia.") print("\nOK — memory recalled with proof, and every exclusion is explainable.")