Skip to content

Python API Reference

The main async API. All public methods are coroutines.

from cognitive_memory import CognitiveMemory
CognitiveMemory(
config: CognitiveMemoryConfig | None = None,
embedder: EmbeddingProvider | Literal["openai", "hash"] | None = None,
adapter: MemoryAdapter | None = None,
)
ParameterTypeDefaultDescription
configCognitiveMemoryConfigCognitiveMemoryConfig()All tunable parameters
embedderEmbeddingProvider | "openai" | "hash""openai"Embedding backend
adapterMemoryAdapterInMemoryAdapter()Storage backend

add(content, category, importance, session_id, timestamp) -> Memory

Section titled “add(content, category, importance, session_id, timestamp) -> Memory”

Add a single memory directly, bypassing LLM extraction.

memory = await mem.add(
content="User is allergic to peanuts",
category=MemoryCategory.CORE, # default: EPISODIC
importance=0.95, # default: 0.5
session_id="session-1", # optional
timestamp=datetime(2024, 3, 15), # default: now
)

Returns the created Memory object with generated ID and embedding.

Add a pre-built Memory object. Embeds the content if memory.embedding is None.

from cognitive_memory.types import Memory
mem_obj = Memory(content="User likes coffee", category=MemoryCategory.SEMANTIC)
await mem.add_memory_object(mem_obj)

extract_and_store(conversation_text, session_id, timestamp, run_tick) -> list[Memory]

Section titled “extract_and_store(conversation_text, session_id, timestamp, run_tick) -> list[Memory]”

Extract memories from conversation text using LLM, embed them, check for conflicts, create associations, and store.

memories = await mem.extract_and_store(
conversation_text="User: I just moved to Seattle.\nAssistant: How exciting!",
session_id="s1",
timestamp=datetime(2024, 3, 15), # default: now
run_tick=True, # run maintenance every 5 ingestions
)

Returns list of created Memory objects.

search(query, top_k, timestamp, session_id, deep_recall, trace) -> SearchResponse

Section titled “search(query, top_k, timestamp, session_id, deep_recall, trace) -> SearchResponse”

Search memories with full retention-weighted scoring, associative expansion, and retrieval boosting.

response = await mem.search(
query="what is the user allergic to?",
top_k=10, # default: 10
timestamp=datetime(2024, 3, 20), # default: now
session_id="s2", # optional, for boost tracking
deep_recall=False, # default: False
trace=False, # default: False
)
results = response.results
ParameterTypeDefaultDescription
traceboolFalseWhen True, the response includes a trace field with scoring breakdown and pipeline timing details

Returns a SearchResponse object containing a results list of SearchResult objects sorted by combined_score descending, and an optional trace field when tracing is enabled.

Run periodic maintenance: cold migration, TTL expiry, and consolidation.

await mem.tick()
await mem.tick(now=datetime(2024, 6, 1)) # simulate future time

Return current memory system statistics.

stats = await mem.get_stats()
# {
# "total_memories": 142,
# "hot_memories": 98,
# "cold_memories": 38,
# "stub_memories": 6,
# "core_memories": 12,
# "faint_memories": 23,
# "avg_retention": 0.71,
# }

Delete all memories from the adapter.

await mem.clear()
PropertyTypeDescription
adapterMemoryAdapterThe storage adapter
engineCognitiveEngineThe scoring/decay engine
embedderEmbeddingProviderThe embedding provider

Synchronous wrapper around CognitiveMemory. Same interface, no await.

from cognitive_memory import SyncCognitiveMemory
mem = SyncCognitiveMemory(embedder="hash")
mem.add("User likes coffee", importance=0.5)
results = mem.search("coffee")
stats = mem.get_stats()
mem.clear()

All methods have identical signatures to CognitiveMemory but run synchronously. Works in Jupyter notebooks and scripts.


Core memory dataclass.

from cognitive_memory import Memory
FieldTypeDefaultDescription
idstruuid4()Unique identifier
contentstr""Memory text
categoryMemoryCategoryEPISODICMemory type
importancefloat0.5LLM-assessed importance [0,1]
stabilityfloat0.1Resistance to decay [0,1]
access_countint0Retrieval count
last_accessed_atdatetime | NoneNoneLast retrieval timestamp
created_atdatetime | NoneNoneCreation timestamp
embeddinglist[float] | NoneNoneVector embedding
associationsdict[str, Association]{}Association links
session_idsset[str]set()Sessions that accessed this
is_coldboolFalseIn cold storage
cold_sincedatetime | NoneNoneWhen moved to cold
days_at_floorint0Consecutive days at floor
is_supersededboolFalseReplaced by consolidation
superseded_bystr | NoneNoneSummary ID
contradicted_bystr | NoneNoneContradicting memory ID
is_stubboolFalseArchived stub
PropertyTypeDescription
floorfloatRetention floor (0.60 for core, 0.02 for regular)
base_decay_ratefloatBase decay in days for this category
is_faintboolLow stability, non-core
is_core_memoryboolCategory is CORE

Returned by search().

from cognitive_memory import SearchResult
FieldTypeDescription
memoryMemoryThe matched memory
relevance_scorefloatCosine similarity to query
retention_scorefloatR(m) from decay model
combined_scorefloatrelevance * retention^alpha
is_associativeboolCame via association link
via_deep_recallboolFrom superseded memory

Enum for memory types.

from cognitive_memory import MemoryCategory
MemoryCategory.EPISODIC # "episodic"
MemoryCategory.SEMANTIC # "semantic"
MemoryCategory.PROCEDURAL # "procedural"
MemoryCategory.CORE # "core"

Abstract interface for custom embedding backends.

from cognitive_memory import EmbeddingProvider
class MyEmbedder(EmbeddingProvider):
def embed(self, text: str) -> list[float]:
return my_model.encode(text).tolist()
def embed_batch(self, texts: list[str]) -> list[list[float]]:
return [self.embed(t) for t in texts]

Built-in implementations:

  • OpenAIEmbeddings(model="text-embedding-3-small", dimensions=1536)
  • HashEmbeddings(dimensions=384) — offline, deterministic, testing only