Skip to content

Memory

EpisodicMemory 

EpisodicMemory(embeddings: Embeddings | None = None, collection_name: str = 'episodic_memory')

Vector-backed episodic store with timestamped retrieval.

Source code in src/agentic_architectures/memory/episodic.py 
36
37
38
39
40
41
42
def __init__(
    self,
    embeddings: Embeddings | None = None,
    collection_name: str = "episodic_memory",
) -> None:
    self._vector = VectorMemory(embeddings=embeddings, collection_name=collection_name)
    self._episodes: list[Episode] = []

record 

record(episode: Episode | str, **metadata: Any) -> None

Add an episode to memory (string is converted to a default Episode).

Source code in src/agentic_architectures/memory/episodic.py 
44
45
46
47
48
49
50
def record(self, episode: Episode | str, **metadata: Any) -> None:
    """Add an episode to memory (string is converted to a default Episode)."""
    from langchain_core.documents import Document

    ep = episode if isinstance(episode, Episode) else Episode(content=episode, metadata=metadata)
    self._episodes.append(ep)
    self._vector.add([Document(page_content=ep.content, metadata={**asdict(ep), **ep.metadata})])

recall 

recall(query: str, k: int = 5) -> list[Episode]

Retrieve the k most semantically similar past episodes.

Source code in src/agentic_architectures/memory/episodic.py 
52
53
54
55
56
57
58
59
60
61
62
63
def recall(self, query: str, k: int = 5) -> list[Episode]:
    """Retrieve the k most semantically similar past episodes."""
    docs = self._vector.search(query, k=k)
    return [
        Episode(
            content=d.page_content,
            role=d.metadata.get("role", "user"),
            timestamp=d.metadata.get("timestamp", ""),
            metadata={k: v for k, v in d.metadata.items() if k not in {"role", "timestamp", "content"}},
        )
        for d in docs
    ]

Episode dataclass 

Episode(content: str, role: str = 'user', timestamp: str = (lambda: datetime.now(timezone.utc).isoformat())(), metadata: dict[str, Any] = dict())

A single recorded episode.

SemanticMemory 

SemanticMemory(graph: BaseGraphMemory | None = None, backend: GraphBackend | None = None)

Triple-store wrapper on top of whichever graph backend is configured.

Source code in src/agentic_architectures/memory/semantic.py 
21
22
23
24
25
26
def __init__(
    self,
    graph: BaseGraphMemory | None = None,
    backend: GraphBackend | None = None,
) -> None:
    self._memory: BaseGraphMemory = graph if graph is not None else get_graph_memory(backend)

add_fact 

add_fact(subject: str, predicate: str, obj: str) -> None

Record a (subject, predicate, object) triple.

Source code in src/agentic_architectures/memory/semantic.py 
28
29
30
def add_fact(self, subject: str, predicate: str, obj: str) -> None:
    """Record a (subject, predicate, object) triple."""
    self._memory.add_triple(subject, predicate, obj)

add_facts 

add_facts(facts: list[tuple[str, str, str]]) -> None

Bulk-add triples.

Source code in src/agentic_architectures/memory/semantic.py 
32
33
34
35
def add_facts(self, facts: list[tuple[str, str, str]]) -> None:
    """Bulk-add triples."""
    for s, p, o in facts:
        self._memory.add_triple(s, p, o)

facts_about 

facts_about(entity: str, depth: int = 1) -> list[dict[str, str]]

Return facts within depth hops of the given entity.

Source code in src/agentic_architectures/memory/semantic.py 
37
38
39
40
41
42
43
44
45
def facts_about(self, entity: str, depth: int = 1) -> list[dict[str, str]]:
    """Return facts within `depth` hops of the given entity."""
    cypher = (
        "MATCH path = (e:Entity {name: $name})-[r:RELATES*1.." + str(depth) + "]-(other) "
        "UNWIND relationships(path) AS rel "
        "RETURN startNode(rel).name AS subject, rel.predicate AS predicate, "
        "endNode(rel).name AS object LIMIT 100"
    )
    return self._memory.query(cypher, params={"name": entity})

VectorMemory 

VectorMemory(embeddings: Embeddings | None = None, backend: VectorBackend | None = None, collection_name: str = 'agentic_architectures')

Thin convenience wrapper used by RAG architectures.

Source code in src/agentic_architectures/memory/vector.py 
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    embeddings: Embeddings | None = None,
    backend: VectorBackend | None = None,
    collection_name: str = "agentic_architectures",
) -> None:
    self._store = get_vector_store(
        embeddings=embeddings,
        backend=backend,
        collection_name=collection_name,
    )