docs / agent-memory / examples

Examples

Runnable recipes. The repo has no examples/ directory — these snippets mirror the Recipes below mirror the repo's test suite (tests/test_ghost_memory_demo.py et al.); the wiring example composes documented APIs. tests/test_ghost_memory_demo.py).

Quickstart — two calls

from fg_agent_memory import Memory

memory = Memory("./memory")          # readable directory you can commit to git
memory.remember("John's favorite editor is Zed.")
print(memory.recall("what editor does John use?").as_prompt_block())

Ghost memory — contradiction, resolve, history

from fg_agent_memory import Memory

memory = Memory("./memory")
a = memory.remember("The staging database is Postgres.")
b = memory.remember("The staging database is not Postgres.")

memory.consolidate()                 # detects contradiction; neither side dropped
print(memory.recall("staging database").as_prompt_block())
# Relevant memories (query: ...):
# - The staging database is Postgres. [disputed: one side negates the other]
# - The staging database is not Postgres. [disputed: one side negates the other]

memory.resolve(a.record_id, b.record_id,
               "checked infra: staging moved off Postgres in March")

print(memory.recall("staging database").as_prompt_block())
# Relevant memories (query: ...):
# - The staging database is not Postgres.

print(memory.recall("staging database", include_history=True).as_prompt_block())
# Relevant memories (query: ...):
# - The staging database is not Postgres.
# - The staging database is Postgres. [superseded — kept for history]

The loser keeps its full version trail — the test asserts ["active", "transitional", "superseded"] across history, with superseded_by == b.record_id and the resolve reason recorded as state_reason.

Value-swap detection — no negation word needed

Contradictions aren't only "X" vs "not X". Inferred subject→value slots catch "X is A" vs "X is B":

from fg_agent_memory import Memory, RecordState

memory = Memory(tmp_path / "memory", auto_consolidate=False)
first  = memory.remember("The staging database is Postgres.")
second = memory.remember("The staging database is MySQL.")
memory.consolidate()
assert memory.store.get(first.record_id).state is RecordState.TRANSITIONAL

Also covered by the heuristics: morphological negation (requires vs does not require) and polarity frames (no restart needed vs requires a restart). Non-disputes stay untouched — likes coffee vs likes tea is not a contradiction.

Auto-consolidation cadence

By default the porcelain consolidates every 4 remembers. Tighten the cadence and disputes surface without ever calling consolidate():

memory = Memory(tmp_path / "memory", auto_consolidate_every=2)
a = memory.remember("The deploy pipeline is green.")
b = memory.remember("The deploy pipeline is not green.")
assert b.consolidation is not None   # cadence hit; dispute already surfaced

MCP server

fg-agent-memory-mcp --path ~/agent-memory

{ "mcpServers": { "memory": { "command": "fg-agent-memory-mcp",
                              "args": ["--path", "/home/me/agent-memory"] } } }

Tools exposed: remember, recall, consolidate, redact, status.

Bring your own store, index, and operators

Every seam is a constructor argument — the porcelain composes ports:

from fg_agent_id import KeyPair
from fg_agent_memory import (
    Memory, SQLiteRecordStore, SQLiteSearchIndex,
    ConsolidationOperators, TrigramNearDupOperator,
    HeuristicContradictionOperator,
)

memory = Memory(
    store=SQLiteRecordStore("memory.db"),
    indexes=(SQLiteSearchIndex("index.sqlite"),),
    extractor=my_llm_extractor,          # any Operator: propose(text, provenance)
    operators=ConsolidationOperators(
        near_dup=TrigramNearDupOperator(threshold=0.7),
        contradiction=HeuristicContradictionOperator(),
        resolver=None,                   # keep contradictions human-resolved
    ),
    identity=KeyPair.generate(),
)

An LLM extractor or consolidation operator still only emits typed Proposals — the lifecycle validation gate is identical, and invalid proposals are rejected, never repaired.

Export and load a signed memory file

from fg_agent_id import KeyPair
from fg_agent_memory import Memory

keys = KeyPair.generate()
memory = Memory("./memory", identity=keys)
memory.remember("Deploys go out on Tuesdays.")

memory.export("agent.memory.json")     # one canonical-JSON MemoryFile, file-signed

# elsewhere — verification is on by default and REQUIRED:
restored = Memory.load("agent.memory.json", path="./restored-memory")
# verify=True demands a valid file signature and verifies every
# record that names a signer; it never silently downgrades.

Redaction — the governed escape hatch

record = memory.redact(record_id, "contained an API token")
assert record.redacted
assert record.body == "[redacted]"

Every stored version is replaced by one tombstone: id, type, timestamps, and the link graph survive; body, slots, tags, and provenance are destroyed; the full birth_digest is recorded. Distinct from decay — archival keeps content. Redaction is unreachable by operators and deliberately absent from the RecordStore port.

PreviousAPI reference