šŸ§¬ Semantic Memory

Persistent knowledge storage and retrieval.

Overview

Semantic Memory is the long-term knowledge store that maintains the enterpriseā€™s semantic understanding across sessions, enabling AI agents to access accumulated knowledge.

Layer Purpose
Knowledge Graph Structured entity relationships
Vector Store Semantic similarity search
Temporal Store Time-aware knowledge
Debt Ledger Semantic debt tracking

Knowledge Sources

Explicit Knowledge

Directly defined in DomainForgeā„¢ models:

Entity "Customer.Premium" in sales
  @description "High-value recurring customer"
  @attributes { tier: "gold", since: 2020 }

Derived Knowledge

Computed through inference and projections:

1
2
3
4
5
6

# Find implicit relationships
SELECT ?entity ?connects_to
WHERE {
  ?flow sea:from ?entity ;
        sea:to ?connects_to .
}

Learned Knowledge

Captured from interactions and feedback:

1
2
3
4
5
6
7

{
  "type": "learned_association",
  "source": "user_feedback",
  "entities": ["Customer.X", "Product.Y"],
  "relationship": "prefers",
  "confidence": 0.85
}

Memory Operations

Store

1
2
3
4
5
6
7
8
9
10

async def store_knowledge(
    subject: str,
    predicate: str,
    obj: str | dict,
    metadata: dict = None
) -> str:
    """Store a new knowledge triple."""
    triple = RDFTriple(subject, predicate, obj)
    await kg_service.insert(triple, metadata)
    return triple.id

Query

1
2
3
4
5
6

async def query_knowledge(
    sparql: str,
    snapshot_id: str = None
) -> list[dict]:
    """Query knowledge graph."""
    return await kg_service.query(sparql, snapshot_id)

Update

1
2
3
4
5
6
7
8
9
10

async def update_knowledge(
    triple_id: str,
    new_value: str | dict
) -> None:
    """Update existing knowledge."""
    changeset = ChangeSet([
        DeleteOp(triple_id),
        AddOp(triple_id, new_value)
    ])
    await kg_service.apply_changeset(changeset)

Vector Semantic Memory

For similarity-based retrieval:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

from semantic_memory import VectorStore

# Store with embedding
await vector_store.upsert(
    id="concept:payment-processing",
    content="Payment processing handles financial transactions...",
    metadata={"domain": "finance", "type": "concept"}
)

# Semantic search
results = await vector_store.search(
    query="How do we handle refunds?",
    top_k=5,
    filter={"domain": "finance"}
)

Temporal Knowledge

Track knowledge changes over time:

Store Temporal Fact

1
2
3
4
5
6
7

await temporal_store.store(
    subject="policy:vendor-limit",
    predicate="value",
    obj=5000,
    valid_from=datetime(2025, 1, 1),
    valid_to=datetime(2025, 12, 31)
)

Query At Point In Time

1
2
3
4
5
6

# What was the vendor limit on March 15?
result = await temporal_store.query_at(
    subject="policy:vendor-limit",
    predicate="value",
    at_time=datetime(2025, 3, 15)
)

Semantic Debt Integration

Track knowledge quality issues:

1
2
3
4
5
6
7
8
9
10

# Record semantic debt
await debt_ledger.open(
    title="Missing vendor classification",
    concept="entity:vendor-x",
    severity="warning",
    evidence="No tier attribute defined"
)

# Query debt by concept
debts = await debt_ledger.query_by_concept("entity:vendor-x")

Memory Consistency

Snapshot Isolation

1
2
3
4
5
6
7
8
9
10
11

# Create consistent snapshot
snapshot = await memory.create_snapshot(
    version="1.2.0",
    description="Q1 knowledge base"
)

# Query against snapshot
results = await memory.query(
    sparql="SELECT * WHERE { ?s ?p ?o }",
    snapshot_id=snapshot.id
)

Change Tracking

All changes are tracked as immutable change sets:

1
2
3

changeset = await memory.get_changeset(changeset_id)
print(f"Operations: {len(changeset.operations)}")
print(f"Applied to: {changeset.source_snapshot_id}")