API reference¶
Generated from the apogee-ai-memory source with mkdocstrings. Every symbol below is exported from apogee_ai_memory, so it is part of the supported public surface.
Application · DTOs¶
Bases: BaseModel
class-attribute
instance-attribute
¶
Cap on the number of consolidated memories produced.
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
types: list[MemoryType] = Field(default_factory=list)
Application · Use cases¶
Python
ConsolidateMemoryUseCase(store: IMemoryStore, reflection: IMemoryReflection)
Reads episodic records, runs reflection, optionally writes back.
Source code in apogee_ai_memory/application/use_cases/consolidate_memory_use_case.py
async
¶
Python
execute(dto: ConsolidateDTO) -> ReflectionResult
Source code in apogee_ai_memory/application/use_cases/consolidate_memory_use_case.py
Python
async def execute(self, dto: ConsolidateDTO) -> ReflectionResult:
candidates = await self._store.list(
tenant_id=dto.tenant_id,
user_id=dto.user_id,
)
episodic = [r for r in candidates if r.type == MemoryType.EPISODIC]
result = await self._reflection.reflect(
episodic, max_consolidations=dto.max_consolidations
)
if dto.persist and result.succeeded:
for record in result.consolidated:
await self._store.remember(record)
return result
Python
ForgetUseCase(store: IMemoryStore)
Source code in apogee_ai_memory/application/use_cases/forget_use_case.py
async
¶
Python
ListMemoriesUseCase(store: IMemoryStore)
Source code in apogee_ai_memory/application/use_cases/list_memories_use_case.py
async
¶
Python
execute(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
Source code in apogee_ai_memory/application/use_cases/list_memories_use_case.py
Python
ProfileUserUseCase(store: IMemoryStore)
Source code in apogee_ai_memory/application/use_cases/profile_user_use_case.py
async
¶
Python
execute(*, tenant_id: str | None = None, user_id: str | None = None) -> MemoryProfile
Source code in apogee_ai_memory/application/use_cases/profile_user_use_case.py
Python
async def execute(
self,
*,
tenant_id: str | None = None,
user_id: str | None = None,
) -> MemoryProfile:
records = await self._store.list(tenant_id=tenant_id, user_id=user_id)
if not records:
return MemoryProfile(tenant_id=tenant_id, user_id=user_id)
counter = Counter(r.type.value for r in records)
most_recent = max(r.created_at for r in records)
most_accessed = max(records, key=lambda r: (r.access_count, r.importance, r.created_at))
return MemoryProfile(
tenant_id=tenant_id,
user_id=user_id,
total_records=len(records),
by_type=dict(counter),
most_recent=most_recent,
most_accessed_text=most_accessed.text if most_accessed.access_count else None,
)
Python
RecallUseCase(store: IMemoryStore)
Source code in apogee_ai_memory/application/use_cases/recall_use_case.py
async
¶
Python
execute(dto_or_query: RecallDTO | MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/application/use_cases/recall_use_case.py
Python
async def execute(self, dto_or_query: RecallDTO | MemoryQuery) -> list[MemoryHit]:
if isinstance(dto_or_query, MemoryQuery):
return await self._store.recall(dto_or_query)
query = MemoryQuery(
text=dto_or_query.text,
types=tuple(dto_or_query.types),
tenant_id=dto_or_query.tenant_id,
user_id=dto_or_query.user_id,
agent_id=dto_or_query.agent_id,
tags=tuple(dto_or_query.tags),
top_k=dto_or_query.top_k,
min_score=dto_or_query.min_score,
include_decayed=dto_or_query.include_decayed,
)
return await self._store.recall(query)
Python
RememberUseCase(store: IMemoryStore)
Source code in apogee_ai_memory/application/use_cases/remember_use_case.py
async
¶
Python
execute(dto_or_record: RememberDTO | MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/application/use_cases/remember_use_case.py
Python
async def execute(self, dto_or_record: RememberDTO | MemoryRecord) -> MemoryRecord:
if isinstance(dto_or_record, MemoryRecord):
return await self._store.remember(dto_or_record)
record = MemoryRecord(
text=dto_or_record.text,
type=dto_or_record.type,
scope=dto_or_record.scope,
tenant_id=dto_or_record.tenant_id,
user_id=dto_or_record.user_id,
agent_id=dto_or_record.agent_id,
importance=dto_or_record.importance,
confidence=dto_or_record.confidence,
tags=tuple(dto_or_record.tags),
source=dto_or_record.source,
metadata=dict(dto_or_record.metadata),
)
return await self._store.remember(record)
Domain¶
dataclass
¶
Python
cosine(other: EmbeddingVector) -> float
Source code in apogee_ai_memory/domain/value_objects/embedding_vector.py
Python
def cosine(self, other: EmbeddingVector) -> float:
if self.dimension != other.dimension:
raise ValueError(
f"Cannot compare vectors of different dimensions: "
f"{self.dimension} vs {other.dimension}"
)
dot = sum(a * b for a, b in zip(self.values, other.values, strict=False))
norm_a = math.sqrt(sum(a * a for a in self.values))
norm_b = math.sqrt(sum(b * b for b in other.values))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
dataclass
¶
Python
MemoryHit(record: MemoryRecord, score: float, decayed_score: float, backend: str = 'memory')
A single match for a recall query.
instance-attribute
¶
Score after applying retention/decay; equal to score when curve == NONE.
dataclass
¶
Python
MemoryKey(tenant_id: str | None = None, user_id: str | None = None, agent_id: str | None = None, scope: MemoryScope = USER)
Identity tuple every memory record must carry to support isolation.
Python
matches(other: MemoryKey) -> bool
other is allowed to read self?
Hierarchy: USER ⊂ AGENT ⊂ TENANT ⊂ GLOBAL.
Source code in apogee_ai_memory/domain/value_objects/memory_key.py
Python
def matches(self, other: MemoryKey) -> bool:
"""``other`` is allowed to read ``self``?
Hierarchy: USER ⊂ AGENT ⊂ TENANT ⊂ GLOBAL.
"""
if self.scope == MemoryScope.GLOBAL:
return True
if self.tenant_id is not None and self.tenant_id != other.tenant_id:
return False
if self.scope == MemoryScope.TENANT:
return True
if self.scope == MemoryScope.AGENT:
return self.agent_id is None or self.agent_id == other.agent_id
# USER scope
if self.user_id is not None and self.user_id != other.user_id:
return False
if self.agent_id is not None and self.agent_id != other.agent_id:
return False
return True
dataclass
¶
Python
MemoryProfile(tenant_id: str | None, user_id: str | None, total_records: int = 0, by_type: dict[str, int] = dict(), most_recent: datetime | None = None, most_accessed_text: str | None = None, generated_at: datetime = (lambda: now(utc))())
dataclass
¶
Python
MemoryQuery(text: str = '', types: tuple[MemoryType, ...] = tuple(), tenant_id: str | None = None, user_id: str | None = None, agent_id: str | None = None, tags: tuple[str, ...] = tuple(), top_k: int = 5, min_score: float = 0.0, include_decayed: bool = False)
Search criteria for IMemoryStore.search.
class-attribute
instance-attribute
¶
Python
types: tuple[MemoryType, ...] = field(default_factory=tuple)
Empty = match any type.
class-attribute
instance-attribute
¶
dataclass
¶
Python
MemoryRecord(text: str, type: MemoryType = SEMANTIC, scope: MemoryScope = USER, id: str = (lambda: token_hex(16))(), tenant_id: str | None = None, user_id: str | None = None, agent_id: str | None = None, embedding: EmbeddingVector | None = None, confidence: float = 1.0, importance: float = 0.5, created_at: datetime = (lambda: now(utc))(), last_accessed_at: datetime = (lambda: now(utc))(), access_count: int = 0, tags: tuple[str, ...] = tuple(), source: str | None = None, metadata: dict[str, str] = dict())
One persisted memory item.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Reliability score [0,1] — judges/reflection adjust it over time.
class-attribute
instance-attribute
¶
[0,1] — bumps surfaced by reflection / explicit pinning.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Bases: str, Enum
Visibility of a memory record across the multi-tenant hierarchy.
dataclass
¶
Python
ReflectionResult(consolidated: tuple[MemoryRecord, ...] = tuple(), sources_considered: int = 0, sources_used: int = 0, backend: str = 'rule_based', error: str | None = None)
Outcome of consolidating episodic records into semantic memories.
class-attribute
instance-attribute
¶
Python
consolidated: tuple[MemoryRecord, ...] = field(default_factory=tuple)
dataclass
¶
Python
RetentionPolicy(curve: DecayCurve = NONE, half_life_days: float = 30.0, max_age_days: float | None = None)
Domain · Enums¶
Bases: str, Enum
Canonical memory categories — modelled after cognitive psychology.
class-attribute
instance-attribute
¶
Time-stamped events: 'user asked X at 10:00'.
class-attribute
instance-attribute
¶
Stable facts and preferences: 'user prefers PT-BR'.
class-attribute
instance-attribute
¶
Learned skills: 'when invoice arrives, run pipeline X'.
class-attribute
instance-attribute
¶
Short-lived buffer for the current conversation/turn.
Domain · Exceptions¶
Bases: Exception
Base class for apogee-ai-memory errors.
Domain · Protocols (ports)¶
Bases: Protocol
Computes a [0,1] retention multiplier from age.
Bases: Protocol
async
¶
Python
embed(text: str) -> EmbeddingVector
async
¶
Python
embed_many(texts: list[str]) -> list[EmbeddingVector]
Bases: Protocol
Consolidates a batch of episodic memories into semantic ones.
async
¶
Python
reflect(episodic_records: list[MemoryRecord], *, max_consolidations: int = 5) -> ReflectionResult
Bases: Protocol
Backend that stores and recalls :class:MemoryRecord objects.
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
async
¶
async
¶
Python
list(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
async
¶
Infrastructure¶
Python
BridgedEmbedder(encoder: Callable[[str], list[float] | Awaitable[list[float]]], *, model: str = 'bridged')
Wraps an arbitrary callable that returns list[float] per text.
Decouples apogee-ai-memory from any specific provider package while still allowing real embedding pipelines (sentence-transformers, OpenAI embeddings, Voyage, etc.) to be plugged in.
Source code in apogee_ai_memory/infrastructure/embeddings/bridged_embedder.py
async
¶
Python
embed(text: str) -> EmbeddingVector
Source code in apogee_ai_memory/infrastructure/embeddings/bridged_embedder.py
Python
async def embed(self, text: str) -> EmbeddingVector:
try:
result = self._encoder(text)
if hasattr(result, "__await__"):
values = await result # type: ignore[assignment]
else:
values = result
except Exception as exc: # noqa: BLE001
raise EmbedderUnavailableException(self.name, str(exc)) from exc
return EmbeddingVector(values=tuple(float(v) for v in values), model=self._model)
async
¶
Python
embed_many(texts: list[str]) -> list[EmbeddingVector]
Deterministic, dependency-free embedder.
Uses SHA-256 of the lowercased token to spread it into a fixed-size
vector. Decent for unit tests and CI where deterministic outputs
matter; for production replace with BridgedEmbedder.
Source code in apogee_ai_memory/infrastructure/embeddings/hashing_embedder.py
async
¶
Python
embed(text: str) -> EmbeddingVector
async
¶
Python
embed_many(texts: list[str]) -> list[EmbeddingVector]
Python
InMemoryMemoryStore(*, embedder: IMemoryEmbedder | None = None, decay: IMemoryDecay | None = None)
Bases: IMemoryStore
Thread-unsafe in-memory store; default for tests and dev.
Source code in apogee_ai_memory/infrastructure/stores/in_memory_memory_store.py
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/in_memory_memory_store.py
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/in_memory_memory_store.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
candidates = filter_by_query(self._records.values(), query)
if not candidates:
return []
query_vec = await self._embedder.embed(query.text) if query.text else None
now = datetime.now(timezone.utc)
hits: list[MemoryHit] = []
for record in candidates:
score = (
record.embedding.cosine(query_vec)
if record.embedding is not None and query_vec is not None
else 0.5
)
score = max(0.0, min(1.0, (score + 1.0) / 2.0)) if query_vec else score
retention = self._decay.retention(created_at=record.created_at, now=now)
decayed = max(0.0, min(1.0, score * retention))
if not query.include_decayed and retention == 0.0:
continue
if decayed < query.min_score:
continue
hits.append(
MemoryHit(
record=record,
score=max(0.0, min(1.0, score)),
decayed_score=decayed,
backend=self.name,
)
)
hits.sort(key=lambda h: h.decayed_score, reverse=True)
return hits[: query.top_k]
async
¶
async
¶
Python
list(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
Source code in apogee_ai_memory/infrastructure/stores/in_memory_memory_store.py
Python
async def list(
self,
*,
tenant_id: str | None = None,
user_id: str | None = None,
limit: int | None = None,
) -> list[MemoryRecord]:
items = list(self._records.values())
if tenant_id is not None:
items = [r for r in items if r.tenant_id == tenant_id or r.tenant_id is None]
if user_id is not None:
items = [r for r in items if r.user_id == user_id or r.user_id is None]
items.sort(key=lambda r: r.created_at, reverse=True)
if limit is not None:
items = items[:limit]
return items
async
¶
Python
JsonMemoryStore(root: str | Path, *, embedder: IMemoryEmbedder | None = None, decay: IMemoryDecay | None = None)
File-based memory store: <root>/<record_id>.json.
Designed for single-process workloads — concurrent processes should use the SQL backend instead.
Source code in apogee_ai_memory/infrastructure/stores/json_memory_store.py
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/json_memory_store.py
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/json_memory_store.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
records = await asyncio.to_thread(self._read_all)
candidates = filter_by_query(records, query)
if not candidates:
return []
query_vec = await self._embedder.embed(query.text) if query.text else None
now = datetime.now(timezone.utc)
hits: list[MemoryHit] = []
for record in candidates:
score = (
record.embedding.cosine(query_vec)
if record.embedding is not None and query_vec is not None
else 0.5
)
score = max(0.0, min(1.0, (score + 1.0) / 2.0)) if query_vec else score
retention = self._decay.retention(created_at=record.created_at, now=now)
decayed = max(0.0, min(1.0, score * retention))
if not query.include_decayed and retention == 0.0:
continue
if decayed < query.min_score:
continue
hits.append(
MemoryHit(
record=record,
score=max(0.0, min(1.0, score)),
decayed_score=decayed,
backend=self.name,
)
)
hits.sort(key=lambda h: h.decayed_score, reverse=True)
return hits[: query.top_k]
async
¶
async
¶
Python
list(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
Source code in apogee_ai_memory/infrastructure/stores/json_memory_store.py
Python
async def list(
self,
*,
tenant_id: str | None = None,
user_id: str | None = None,
limit: int | None = None,
) -> list[MemoryRecord]:
records = await asyncio.to_thread(self._read_all)
if tenant_id is not None:
records = [r for r in records if r.tenant_id in {tenant_id, None}]
if user_id is not None:
records = [r for r in records if r.user_id in {user_id, None}]
records.sort(key=lambda r: r.created_at, reverse=True)
if limit is not None:
records = records[:limit]
return records
async
¶
Python
LLMReflection(summarize: Callable[[str], str | Awaitable[str]], *, max_consolidations: int = 5)
LLM-driven consolidation.
Accepts a callable summarize(prompt) -> str to remain decoupled
from any specific provider package. Use apogee-ai-providers or
apogee-ai-prompt to back the callable.
Source code in apogee_ai_memory/infrastructure/reflection/llm_reflection.py
async
¶
Python
reflect(episodic_records: list[MemoryRecord], *, max_consolidations: int = 5) -> ReflectionResult
Source code in apogee_ai_memory/infrastructure/reflection/llm_reflection.py
Python
async def reflect(
self,
episodic_records: list[MemoryRecord],
*,
max_consolidations: int = 5,
) -> ReflectionResult:
budget = min(max_consolidations, self._max)
episodic = [r for r in episodic_records if r.type == MemoryType.EPISODIC]
if not episodic:
return ReflectionResult(backend=self.name, sources_considered=len(episodic_records))
prompt = _PROMPT.format(
max_consolidations=budget,
episodic="\n".join(f"- {r.text}" for r in episodic),
)
try:
response = self._summarize(prompt)
if hasattr(response, "__await__"):
text = await response # type: ignore[assignment]
else:
text = response
except Exception as exc: # noqa: BLE001
raise ReflectionFailureException(str(exc)) from exc
lines = [line.strip(" -*\t") for line in (text or "").splitlines() if line.strip()]
consolidated: list[MemoryRecord] = []
seed = episodic[0]
for line in lines[:budget]:
consolidated.append(
replace(
seed,
id=f"reflect-llm-{seed.id}-{len(consolidated)}",
text=line,
type=MemoryType.SEMANTIC,
confidence=min(1.0, seed.confidence + 0.1),
importance=min(1.0, max(0.6, seed.importance + 0.2)),
source=self.name,
tags=tuple({*seed.tags, "reflected"}),
)
)
return ReflectionResult(
consolidated=tuple(consolidated),
sources_considered=len(episodic_records),
sources_used=len(episodic),
backend=self.name,
)
Bridge to Letta (formerly MemGPT).
Lazy import: install with pip install 'apogee-ai-memory[letta]'.
Letta operates per-agent; this adapter expects record.agent_id to
point at a Letta agent already provisioned.
Source code in apogee_ai_memory/infrastructure/stores/letta_adapter.py
Python
def __init__(self, *, api_key: str | None = None, base_url: str | None = None) -> None:
try:
import letta_client # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"LettaAdapter requires `letta-client`. "
"Install with: pip install 'apogee-ai-memory[letta]'"
) from exc
from letta_client import Letta # type: ignore
try:
self._client = Letta(token=api_key, base_url=base_url)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/letta_adapter.py
Python
async def remember(self, record: MemoryRecord) -> MemoryRecord:
if record.agent_id is None:
raise MemoryStoreUnavailableException(
self.name, "Letta requires record.agent_id"
)
try:
self._client.agents.archival_memory.create(
agent_id=record.agent_id, text=record.text
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
return replace(record)
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/letta_adapter.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
if query.agent_id is None:
return []
try:
response = self._client.agents.archival_memory.search(
agent_id=query.agent_id, query=query.text, limit=query.top_k
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
hits: list[MemoryHit] = []
for item in response or []:
score = float(getattr(item, "score", 0.5))
score = max(0.0, min(1.0, score))
text = getattr(item, "text", "") or ""
record = MemoryRecord(
id=getattr(item, "id", "") or "",
text=text,
agent_id=query.agent_id,
user_id=query.user_id,
tenant_id=query.tenant_id,
) if text else None
if record is None:
continue
hits.append(
MemoryHit(record=record, score=score, decayed_score=score, backend=self.name)
)
return hits
async
¶
async
¶
Python
list(**_: object) -> list[MemoryRecord]
async
¶
Bridge to mem0ai/mem0.
Lazy import: install with pip install 'apogee-ai-memory[mem0]'.
Mem0 does not expose all our knobs, so this adapter focuses on the
common cases: remember + recall + forget. Decay/scope are mapped to
Mem0 metadata when available.
Source code in apogee_ai_memory/infrastructure/stores/mem0_adapter.py
Python
def __init__(self, *, api_key: str | None = None, version: str = "v1.1") -> None:
try:
import mem0 # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"Mem0Adapter requires `mem0ai`. "
"Install with: pip install 'apogee-ai-memory[mem0]'"
) from exc
from mem0 import Memory # type: ignore
try:
self._client = (
Memory(api_key=api_key) if api_key else Memory.from_config({"version": version})
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/mem0_adapter.py
Python
async def remember(self, record: MemoryRecord) -> MemoryRecord:
try:
response = self._client.add(
messages=[{"role": "user", "content": record.text}],
user_id=record.user_id or "default",
metadata={
"type": record.type.value,
"scope": record.scope.value,
"tenant_id": record.tenant_id,
"agent_id": record.agent_id,
"tags": list(record.tags),
**dict(record.metadata),
},
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
new_id = (
response.get("results", [{}])[0].get("id") if isinstance(response, dict) else None
) or record.id
return replace(record, id=str(new_id))
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/mem0_adapter.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
try:
response = self._client.search(
query=query.text,
user_id=query.user_id or "default",
limit=query.top_k,
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
results = response.get("results", []) if isinstance(response, dict) else response or []
hits: list[MemoryHit] = []
for item in results:
score = float(item.get("score", 0.0))
score = max(0.0, min(1.0, score))
metadata = item.get("metadata") or {}
try:
mem_type = MemoryType(metadata.get("type", "semantic"))
except ValueError:
mem_type = MemoryType.SEMANTIC
record = MemoryRecord(
id=str(item.get("id", "")),
text=str(item.get("memory") or item.get("text") or ""),
type=mem_type,
tenant_id=metadata.get("tenant_id"),
user_id=query.user_id,
created_at=_parse_dt(item.get("created_at")),
)
hits.append(
MemoryHit(record=record, score=score, decayed_score=score, backend=self.name)
)
return hits
async
¶
async
¶
Python
list(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
Source code in apogee_ai_memory/infrastructure/stores/mem0_adapter.py
Python
async def list(
self,
*,
tenant_id: str | None = None, # noqa: ARG002
user_id: str | None = None,
limit: int | None = None,
) -> list[MemoryRecord]:
try:
response = self._client.get_all(user_id=user_id or "default")
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
results = response.get("results", []) if isinstance(response, dict) else response or []
records = [
MemoryRecord(
id=str(item.get("id", "")),
text=str(item.get("memory") or item.get("text") or ""),
user_id=user_id,
created_at=_parse_dt(item.get("created_at")),
)
for item in results
]
if limit is not None:
records = records[:limit]
return records
async
¶
Python
MemoryStoreRegistry(stores: Mapping[str, IMemoryStore] | None = None)
Source code in apogee_ai_memory/infrastructure/registry/memory_store_registry.py
Python
register(store: IMemoryStore) -> None
Python
get(name: str) -> IMemoryStore
Python
find(name: str) -> IMemoryStore | None
Python
SqlMemoryStore(session_factory: Any, *, table_name: str = 'apogee_memory', embedder: IMemoryEmbedder | None = None, decay: IMemoryDecay | None = None)
SQLAlchemy 2 async store backed by SQLite or Postgres.
The schema is one row per record with the full payload in a JSON
column — keeps the surface area small while still allowing tenant_id
indexing for isolation. Lazy SQLAlchemy import via [sqlite] or
[pgvector].
Source code in apogee_ai_memory/infrastructure/stores/sql_memory_store.py
Python
def __init__(
self,
session_factory: Any,
*,
table_name: str = "apogee_memory",
embedder: IMemoryEmbedder | None = None,
decay: IMemoryDecay | None = None,
) -> None:
try:
import sqlalchemy # type: ignore # noqa: F401
except ImportError as exc: # pragma: no cover
raise ImportError(
"SqlMemoryStore requires SQLAlchemy. "
"Install with: pip install 'apogee-ai-memory[sqlite]' or [pgvector]"
) from exc
self._session_factory = session_factory
self._table_name = table_name
self._embedder = embedder or HashingEmbedder()
self._decay = decay or NoneDecay()
self._table = self._build_table()
async
¶
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/sql_memory_store.py
Python
async def remember(self, record: MemoryRecord) -> MemoryRecord:
from sqlalchemy import delete, insert
if record.embedding is None:
embedding = await self._embedder.embed(record.text)
record = replace(record, embedding=embedding)
payload = record_to_dict(record)
async with self._session_factory() as session:
await session.execute(self._table.delete().where(self._table.c.id == record.id))
del delete # silence ruff
await session.execute(
insert(self._table).values(
id=record.id,
tenant_id=record.tenant_id,
user_id=record.user_id,
data=payload,
)
)
await session.commit()
return record
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/sql_memory_store.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
from sqlalchemy import select
async with self._session_factory() as session:
stmt = select(self._table.c.data)
if query.tenant_id is not None:
stmt = stmt.where(
(self._table.c.tenant_id == query.tenant_id)
| (self._table.c.tenant_id.is_(None))
)
rows = (await session.execute(stmt)).all()
records = [record_from_dict(self._decode(r[0])) for r in rows]
candidates = filter_by_query(records, query)
if not candidates:
return []
query_vec = await self._embedder.embed(query.text) if query.text else None
now = datetime.now(timezone.utc)
hits: list[MemoryHit] = []
for record in candidates:
score = (
record.embedding.cosine(query_vec)
if record.embedding is not None and query_vec is not None
else 0.5
)
score = max(0.0, min(1.0, (score + 1.0) / 2.0)) if query_vec else score
retention = self._decay.retention(created_at=record.created_at, now=now)
decayed = max(0.0, min(1.0, score * retention))
if not query.include_decayed and retention == 0.0:
continue
if decayed < query.min_score:
continue
hits.append(
MemoryHit(
record=record,
score=max(0.0, min(1.0, score)),
decayed_score=decayed,
backend=self.name,
)
)
hits.sort(key=lambda h: h.decayed_score, reverse=True)
return hits[: query.top_k]
async
¶
Source code in apogee_ai_memory/infrastructure/stores/sql_memory_store.py
async
¶
Python
list(*, tenant_id: str | None = None, user_id: str | None = None, limit: int | None = None) -> list[MemoryRecord]
Source code in apogee_ai_memory/infrastructure/stores/sql_memory_store.py
Python
async def list(
self,
*,
tenant_id: str | None = None,
user_id: str | None = None,
limit: int | None = None,
) -> list[MemoryRecord]:
from sqlalchemy import select
async with self._session_factory() as session:
stmt = select(self._table.c.data).order_by(self._table.c.id)
if tenant_id is not None:
stmt = stmt.where(
(self._table.c.tenant_id == tenant_id)
| (self._table.c.tenant_id.is_(None))
)
if user_id is not None:
stmt = stmt.where(
(self._table.c.user_id == user_id)
| (self._table.c.user_id.is_(None))
)
rows = (await session.execute(stmt)).all()
records = [record_from_dict(self._decode(r[0])) for r in rows]
records.sort(key=lambda r: r.created_at, reverse=True)
if limit is not None:
records = records[:limit]
return records
async
¶
Rule-based consolidation: groups episodic records by user/tenant and emits a single semantic summary per group.
Doesn't call any LLM — useful for tests and deterministic pipelines.
Source code in apogee_ai_memory/infrastructure/reflection/summary_reflection.py
async
¶
Python
reflect(episodic_records: list[MemoryRecord], *, max_consolidations: int = 5) -> ReflectionResult
Source code in apogee_ai_memory/infrastructure/reflection/summary_reflection.py
Python
async def reflect(
self,
episodic_records: list[MemoryRecord],
*,
max_consolidations: int = 5,
) -> ReflectionResult:
sources_considered = len(episodic_records)
if not episodic_records:
return ReflectionResult(backend=self.name)
groups: dict[tuple[str | None, str | None, str | None], list[MemoryRecord]] = {}
for r in episodic_records:
if r.type != MemoryType.EPISODIC:
continue
groups.setdefault((r.tenant_id, r.user_id, r.agent_id), []).append(r)
consolidated: list[MemoryRecord] = []
sources_used = 0
for key, items in groups.items():
if len(items) < self._min_per_group:
continue
tenant_id, user_id, agent_id = key
phrases = Counter(self._split_keyphrases(items))
top = ", ".join(p for p, _ in phrases.most_common(5)) or items[0].text
summary_text = f"User often mentions: {top}"
base = items[0]
consolidated.append(
replace(
base,
id=f"reflect-{base.id}",
text=summary_text,
type=MemoryType.SEMANTIC,
confidence=min(1.0, base.confidence + 0.1),
importance=min(1.0, max(0.6, base.importance + 0.2)),
source=self.name,
tags=tuple({*base.tags, "reflected"}),
)
)
sources_used += len(items)
del tenant_id, user_id, agent_id
if len(consolidated) >= max_consolidations:
break
return ReflectionResult(
consolidated=tuple(consolidated),
sources_considered=sources_considered,
sources_used=sources_used,
backend=self.name,
)
Bridge to Zep cloud memory.
Lazy import: install with pip install 'apogee-ai-memory[zep]'.
Source code in apogee_ai_memory/infrastructure/stores/zep_adapter.py
Python
def __init__(self, *, api_key: str | None = None) -> None:
try:
import zep_cloud # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"ZepAdapter requires `zep-cloud`. "
"Install with: pip install 'apogee-ai-memory[zep]'"
) from exc
from zep_cloud.client import AsyncZep # type: ignore
try:
self._client = AsyncZep(api_key=api_key)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
async
¶
Python
remember(record: MemoryRecord) -> MemoryRecord
Source code in apogee_ai_memory/infrastructure/stores/zep_adapter.py
Python
async def remember(self, record: MemoryRecord) -> MemoryRecord:
if record.user_id is None:
raise MemoryStoreUnavailableException(self.name, "Zep requires user_id")
try:
await self._client.user.add(user_id=record.user_id)
except Exception: # noqa: BLE001 - already exists
pass
try:
response = await self._client.memory.add(
user_id=record.user_id,
content=record.text,
metadata={
"type": record.type.value,
"tenant_id": record.tenant_id,
"agent_id": record.agent_id,
**dict(record.metadata),
},
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
new_id = getattr(response, "memory_id", None) or record.id
return replace(record, id=str(new_id))
async
¶
Python
recall(query: MemoryQuery) -> list[MemoryHit]
Source code in apogee_ai_memory/infrastructure/stores/zep_adapter.py
Python
async def recall(self, query: MemoryQuery) -> list[MemoryHit]:
if query.user_id is None:
return []
try:
response = await self._client.memory.search(
user_id=query.user_id, text=query.text, limit=query.top_k
)
except Exception as exc: # noqa: BLE001
raise MemoryStoreUnavailableException(self.name, str(exc)) from exc
hits: list[MemoryHit] = []
for item in (response or []):
score = float(getattr(item, "score", 0.5))
score = max(0.0, min(1.0, score))
text = getattr(item, "content", "") or ""
if not text:
continue
record = MemoryRecord(
id=str(getattr(item, "memory_id", "") or ""),
text=text,
user_id=query.user_id,
tenant_id=query.tenant_id,
created_at=getattr(item, "created_at", datetime.now(timezone.utc)),
)
hits.append(
MemoryHit(record=record, score=score, decayed_score=score, backend=self.name)
)
return hits
async
¶
async
¶
Python
list(**_: object) -> list[MemoryRecord]