Documentation
GitHubDownloads← Home

Memory Graph

Graph traversal over the fact store plus ChromaDB-backed episodic memory — links facts with typed edges and stores conversation episodes for long-range recall.

Overview

MemoryGraph sits on top of MemoryManager and adds two capabilities the flat fact store doesn't have:

  • Typed edges — facts can be linked with relationships likeCONFIRMS, CONTRADICTS, CAUSED_BY, etc. Graph traversal (BFS) lets you retrieve chains of related facts that a pure semantic search would miss.
  • Episodic memory — whole conversations are compressed into episodes with a summary, key facts, and an embedding. Retrieving episodes gives Luna cross-session narrative context.
ModuleContents
memory_graph/constants.pyREL_* constants, ALL_REL_TYPES list.
memory_graph/models.pyFactNode, GraphEdge dataclasses.
memory_graph/graph.pyFull MemoryGraph class.

Instantiation

example.py
from backend.services.memory_graph import MemoryGraph
from backend.models.database import SessionLocal

db = SessionLocal()
graph = MemoryGraph(db)

# ... use graph ...

db.close()
💡
Use the same SQLAlchemy session for both MemoryManager andMemoryGraph within the same request — they share the same database and the session tracks uncommitted state.

Creates or updates a typed directed edge between two fact IDs. Upserts — calling it twice on the same pair updates the relationship type and confidence.

signature
def link_facts(
    fact_a: int,
    fact_b: int,
    relation: str,          # one of the REL_* constants
    confidence: float = 0.8,
    note: str = "",
) -> None
example.py
from backend.services.memory_graph import MemoryGraph, REL_UPDATES

graph = MemoryGraph(db)

# fact 42 updates (supersedes) fact 17
graph.link_facts(42, 17, REL_UPDATES, confidence=0.95, note="user corrected their job title")

BFS from a seed fact ID through the fact_relationships table. Returns all reachable active FactNode objects up tomax_depth hops away. Can be filtered to specific relationship types.

signature
def get_related_facts(
    fact_id: int,
    max_depth: int = 2,
    rel_types: list[str] | None = None,   # filter by relationship type
) -> list[FactNode]
example.py
from backend.services.memory_graph import MemoryGraph, REL_CONFIRMS, REL_RELATED

graph = MemoryGraph(db)

# All facts within 2 hops of fact 5
nodes = graph.get_related_facts(5, max_depth=2)

# Only confirmation/related edges
nodes = graph.get_related_facts(5, rel_types=[REL_CONFIRMS, REL_RELATED])
for n in nodes:
    print(f"[{n.category}] {n.content}  (conf={n.confidence})")

get_subgraph()

Returns a JSON-serialisable dict of nodes and edges for a set of seed fact IDs. Useful for rendering a visual memory graph or for feeding structured context to the LLM.

signature
def get_subgraph(
    seed_fact_ids: list[int],
    max_depth: int = 2,
) -> dict  # {"nodes": [...], "edges": [...]}
example.py
subgraph = graph.get_subgraph([5, 12, 33])
print(subgraph["nodes"][0])
# {"id": 5, "content": "...", "category": "preference", "confidence": 0.9, "importance": 0.8}
print(subgraph["edges"][0])
# {"a": 5, "b": 12, "rel": "RELATED", "conf": 0.75}

find_clusters()

Returns the connected components of the active fact graph using Union-Find. Each component is a list of fact IDs. Only components with 2 or more nodes are returned.

example.py
clusters = graph.find_clusters()
for cluster in clusters:
    print(f"Cluster of {len(cluster)} facts: {cluster}")

Useful for visualisation and for detecting isolated facts that have never been connected to anything else.

Episodic memory

Episodic memory stores a compressed summary of each conversation alongside the IDs of the key facts extracted from it. Episodes are embedded into ChromaDB (luna_episodes collection) for semantic retrieval.

store_episode()

signature
def store_episode(
    conversation_id: int,
    summary: str,
    key_fact_ids: list[int],
    key_entities: list[str],
    embed_fn: Callable[[str], list[float]] | None = None,
) -> None
example.py
from backend.services.llm import ollama

graph.store_episode(
    conversation_id=42,
    summary="User and Luna discussed the Rust borrow checker for 20 minutes. Key insight: lifetime annotations.",
    key_fact_ids=[5, 12, 33],
    key_entities=["Rust", "borrow checker", "lifetime"],
    embed_fn=lambda text: asyncio.run(ollama.embed(text)),
)
# importance = min(1.0, 0.3 + 3 * 0.07) = 0.51

retrieve_episodes()

Returns relevant past episodes ordered by semantic similarity to a query. Falls back to importance-ordered SQL retrieval if embeddings aren't available.

signature
def retrieve_episodes(
    query: str,
    limit: int = 5,
    embed_fn: Callable[[str], list[float]] | None = None,
) -> list[dict]  # [{"summary", "conversation_id", "importance", "entities"}]
example.py
episodes = graph.retrieve_episodes(
    "what did we talk about with Rust?",
    embed_fn=lambda t: asyncio.run(ollama.embed(t)),
)
for ep in episodes:
    print(ep["summary"])

get_episode_facts()

example.py
fact_ids = graph.get_episode_facts(conversation_id=42)
# [5, 12, 33]
# Use with get_subgraph() to pull the full context for an episode
subgraph = graph.get_subgraph(fact_ids)

Relationship types

ConstantValueMeaning
REL_CONTRADICTS"CONTRADICTS"Fact A conflicts with fact B.
REL_CONFIRMS"CONFIRMS"Fact A reinforces fact B.
REL_UPDATES"UPDATES"Fact A supersedes fact B (more recent).
REL_RELATED"RELATED"Topically related, no causal direction.
REL_CAUSED_BY"CAUSED_BY"Fact A is a consequence of fact B.
REL_PRECEDES"PRECEDES"Fact A happened before fact B.
REL_PART_OF"PART_OF"Fact A is a component of fact B.
import
from backend.services.memory_graph import (
    REL_CONTRADICTS, REL_CONFIRMS, REL_UPDATES,
    REL_RELATED, REL_CAUSED_BY, REL_PRECEDES, REL_PART_OF,
    ALL_REL_TYPES,   # list of all seven strings
)

FactNode and GraphEdge

memory_graph/models.py
@dataclass
class FactNode:
    id: int
    content: str
    category: str
    confidence: float = 0.8
    importance: float = 0.5

@dataclass
class GraphEdge:
    fact_id_a: int
    fact_id_b: int
    relationship: str
    confidence: float = 0.8
    note: str = ""