Integrating an Engram.
Bind shared memory to a Neuron with EngramBinding. The Neuron calls recall() and imprint() to read and write the bound Engram without ever touching the protocol. Backed by InMemoryEngram here; swap for SqliteEngram or PostgresEngram without editing Neuron code.
InMemoryEngram and SqliteEngram ship with the core package. PostgresEngram needs the [postgres] extra (which pulls in asyncpg).
# Python 3.11+. The default InMemoryEngram needs no extras. $ pip install cosmonapse # For durable storage: $ pip install "cosmonapse[postgres]" # PostgresEngram # (SqliteEngram is in the stdlib — no extra needed.)
Pure function — plus two helpers.
The Neuron is still a plain async function. The only change is the signature: recall and imprint arrive as keyword-only parameters injected by the Axon. The Neuron addresses memory by the local binding name ("ctx"), not the wire-level engram_id.
# The Neuron gains two keyword-only parameters: recall and imprint. # The Axon injects them at call time because the Axon was constructed # with engrams=[EngramBinding(name="ctx", ...)] (see step 2). # # Under the hood, recall("ctx", ...) emits a RECALL Signal under the # current trace_id and awaits the matching RECALLED reply. The Neuron # stays pure — it never imports the protocol or touches the Synapse. async def researcher(input, context, *, recall, imprint): question = input["question"] # 1. Look in shared memory for a prior answer to this exact question. prior = await recall("ctx", query={"text": question}) if prior.hits: cached = prior.hits[0].content["answer"] return {"answer": cached, "source": "cache"} # 2. Compute a "fresh" answer (stubbed for the demo). answer = f"Answer to {question!r}: 42" # 3. Write it back so the next call hits the cache. # merge_key dedupes by question text so repeated imprints upsert # a single entry per question. await imprint( "ctx", op="upsert", entry={"question": question, "answer": answer, "tags": ["qa"]}, merge_key=f"q:{question}", await_ack=True, deadline_ms=500, ) return {"answer": answer, "source": "computed"}
Engram host · worker · orchestrator.
One Dendrite hosts the Engram, one hosts the Neuron with a declarative EngramBinding, and one dispatches TASKs. The host could be the same process as the worker — we split them here to make the routing explicit.
# Three Dendrites, one shared Synapse: # # host — owns the Engram backend (answers RECALL/IMPRINT) # worker — hosts the Neuron, declares the EngramBinding # orchestrator — dispatches TASKs from cosmonapse import ( Axon, Dendrite, EngramBinding, InMemoryEngram, MemorySynapse, ) synapse = MemorySynapse() await synapse.connect() # 1. Engram host. engram_id="ctx" is the wire address. host = Dendrite(synapse=synapse, namespace="demo", dendrite_id="engram-host", role="worker") host.attach_engram( InMemoryEngram(engram_id="ctx", engram_kind="context") ) # 2. Worker. The binding maps a local name ("ctx") to the wire # engram_id, so the Neuron addresses memory by a stable local # name — operations repoint the backend without editing Neuron code. worker = Dendrite(synapse=synapse, namespace="demo", dendrite_id="worker", role="worker") worker.attach_axon( Axon( neuron_id="researcher", neuron_fn=researcher, capabilities=["research"], engrams=[EngramBinding(name="ctx", engram_id="ctx")], ) ) orchestrator = Dendrite(synapse=synapse, namespace="demo")
First call computes, second call recalls.
Two back-to-back dispatches with the same question. The first sees an empty Engram and imprints the answer. The second finds the cached entry and short-circuits — proof the imprint landed.
# Call twice with the same input. The first call computes + imprints; # the second call recalls and short-circuits. async with host, worker, orchestrator: for label in ("first call ", "second call"): reply = await orchestrator.dispatch_and_wait( neuron="researcher", input={"question": "what is the meaning of life?"}, timeout_s=5.0, ) out = reply.payload["output"] print(f"[{label}] {out['source']:>8s} → {out['answer']}")
$ python main.py [first call ] computed → Answer to 'what is the meaning of life?': 42 [second call] cache → Answer to 'what is the meaning of life?': 42
Three Engram backends, one API.
The Engram interface is uniform. Mount whichever backend fits your deployment; the Neuron and the binding never change.
# Same Engram API, three backends. Swap the line where you mount it; # the Neuron and the binding never change. from cosmonapse import InMemoryEngram, SqliteEngram, PostgresEngram # 1 · In-process — for tests & single-process apps. host.attach_engram(InMemoryEngram(engram_id="ctx", engram_kind="context")) # 2 · SQLite — durable, single file, no server. host.attach_engram(SqliteEngram( engram_id="ctx", engram_kind="context", path="./engram.db", )) # 3 · Postgres — for production. Requires the [postgres] extra. host.attach_engram(PostgresEngram( engram_id="ctx", engram_kind="context", dsn="postgresql://user:pass@localhost/cosmo", ))
Five imprint ops, three recall modes.
imprint covers add / append / merge / upsert / delete. recall returns one entry, a merged view, or the full set — pick the mode per call, or set a default on the binding.
# imprint operations await imprint("ctx", op="add", entry={...}) # fail if exists await imprint("ctx", op="append", entry={...}) # always grow await imprint("ctx", op="merge", entry={...}, merge_key="q:42") # combine await imprint("ctx", op="upsert", entry={...}, merge_key="q:42") # insert or replace await imprint("ctx", op="delete", entry={...}) # remove # recall modes (configure default on the binding) EngramBinding(name="ctx", engram_id="ctx", default_recall_mode="merge") # "first" — return the best single match (default) # "merge" — combine matching entries across backends # "all" — return every match, partial flag if any backend timed out