`symfonic.memory.layers.semantic`¶

semantic ¶

SemanticLayer -- permanent facts and graph entities.

Wraps GraphMemoryStore for fact storage following the User -> Project -> Task -> Tool -> Result graph pattern. Implements BaseMemoryStore with link as a native operation.

SemanticLayer ¶

SemanticLayer(
    graph_store: GraphMemoryStore,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
)

Semantic memory layer for permanent facts and graph entities.

BaseMemoryStore contract

retrieve: queries facts by content similarity
write: stores a fact as a graph node
summarize: returns empty string (facts are not summarizable in isolation)
link: fully implemented (graph-native operation)
delete: removes a fact node from the graph

Wrap graph_store with optional auto-embedding (v7.3 Item 13.1).

embedding_provider (default None) preserves byte-identical pre-Item-13.1 behaviour -- the caller is still responsible for populating node.embedding before write. When supplied, the provider is invoked through :func:maybe_embed before every store_fact / write so nodes the caller leaves with embedding=None auto-embed. Caller-provided embeddings always win.

embedding_cache is the shared sha256-keyed LRU cache the orchestrator hands to every graph-backed layer; passing None is supported but means this layer never benefits from a sibling layer having already embedded the same text.

Per-kind embedding-text rule lives in :func:symfonic.memory.embeddings.auto_embed.resolve_embedding_text -- semantic entity nodes (label prefix Entity:) prefer node.label; all other semantic nodes prefer node.properties['content'].

Source code in src/symfonic/memory/layers/semantic.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
) -> None:
    """Wrap *graph_store* with optional auto-embedding (v7.3 Item 13.1).

    ``embedding_provider`` (default ``None``) preserves byte-identical
    pre-Item-13.1 behaviour -- the caller is still responsible for
    populating ``node.embedding`` before write. When supplied, the
    provider is invoked through :func:`maybe_embed` before every
    ``store_fact`` / write so nodes the caller leaves with
    ``embedding=None`` auto-embed. Caller-provided embeddings always
    win.

    ``embedding_cache`` is the shared sha256-keyed LRU cache the
    orchestrator hands to every graph-backed layer; passing ``None``
    is supported but means this layer never benefits from a sibling
    layer having already embedded the same text.

    Per-kind embedding-text rule lives in
    :func:`symfonic.memory.embeddings.auto_embed.resolve_embedding_text`
    -- semantic entity nodes (label prefix ``Entity:``) prefer
    ``node.label``; all other semantic nodes prefer
    ``node.properties['content']``.
    """
    self._graph = graph_store
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache
    # v7.22 T-7.22.7 -- ``(tenant_id, sub_tenant_id)`` pairs that have
    # already received a MultiScopeIgnoredWarning from this layer
    # instance.  Non-atomic check-then-add is acceptable here
    # (T-7.22.11); a duplicate warning under race is benign and avoids
    # imposing a lock on the per-query hot path.
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

delete `async` ¶

delete(scope: TenantScope, node_id: NodeId) -> None

Delete a semantic fact node and its edges.

Source code in src/symfonic/memory/layers/semantic.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a semantic fact node and its edges."""
    await self._graph.delete_node(scope, node_id)

link `async` ¶

link(
    scope: TenantScope,
    source: NodeId,
    target: NodeId,
    relationship: str,
) -> None

Create a relationship edge between two semantic nodes.

Source code in src/symfonic/memory/layers/semantic.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """Create a relationship edge between two semantic nodes."""
    edge = MemoryEdge(
        source=source,
        target=target,
        relationship=relationship,
        tenant_id=scope.tenant_id,
    )
    await self._graph.add_edge(scope, edge)

query_facts `async` ¶

query_facts(
    scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]

Query semantic facts by keyword matching.

For production use, this should be combined with embedding-based retrieval via the RetrievalEngine. This method provides basic keyword filtering as a fallback.

On a no-keyword-match it returns ONLY the always-on persona node (the AGENT_IDENTITY allowlist) -- never the full importance-sorted node set. The old "top-k by importance" fallback recited unrelated high-importance facts on irrelevant queries (e.g. a bare "Hello"); the allowlist still serves "who are you?"-style identity turns without that leak.

Source code in src/symfonic/memory/layers/semantic.py

async def query_facts(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Query semantic facts by keyword matching.

    For production use, this should be combined with embedding-based
    retrieval via the RetrievalEngine. This method provides basic
    keyword filtering as a fallback.

    On a no-keyword-match it returns ONLY the always-on persona node
    (the ``AGENT_IDENTITY`` allowlist) -- never the full
    importance-sorted node set. The old "top-k by importance"
    fallback recited unrelated high-importance facts on irrelevant
    queries (e.g. a bare "Hello"); the allowlist still serves
    "who are you?"-style identity turns without that leak.
    """
    nodes = await self._graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    query_words = [w for w in query.lower().split() if len(w) > 2]

    def _to_entry(node: MemoryNode) -> MemoryEntry:
        return MemoryEntry(
            layer=MemoryLayer.SEMANTIC,
            tenant_id=node.tenant_id,
            content=str(node.properties.get("content", node.label)),
            node_id=node.id,
            importance=node.importance,
            created_at=node.created_at,
        )

    # Try keyword matching first.
    if query_words:
        matched = [
            node for node in nodes
            if any(
                word in str(node.properties.get("content", node.label)).lower()
                for word in query_words
            )
        ]
        if matched:
            return [_to_entry(n) for n in matched[:top_k]]

    # No keyword match (or an empty query): surface only the
    # always-on persona node(s) via an explicit allowlist instead of
    # the unbounded importance-sorted fallback.
    persona = [n for n in nodes if _is_persona_node(n)]
    return [_to_entry(n) for n in persona[:top_k]]

render_for_prompt ¶

render_for_prompt(
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]]
    | None = None,
) -> str

Render a semantic entry for MEMORY_CONTEXT.

v7.7: explicit override delegating to :func:render_legacy so the per-layer contract test catches a missing override on a future subclass. The pre-v7.7 [layer] content (k=v) shape is the correct semantic-fact render; layer-specific shaping would change cache-keyed prompt bytes for every adopter.

Source code in src/symfonic/memory/layers/semantic.py

def render_for_prompt(
    self,
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> str:
    """Render a semantic entry for ``MEMORY_CONTEXT``.

    v7.7: explicit override delegating to :func:`render_legacy` so
    the per-layer contract test catches a missing override on a
    future subclass.  The pre-v7.7 ``[layer] content (k=v)`` shape
    is the correct semantic-fact render; layer-specific shaping
    would change cache-keyed prompt bytes for every adopter.
    """
    return render_legacy(entry, scrubber)

retrieve `async` ¶

retrieve(
    scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]

Retrieve semantic facts matching the query.

v7.22 T-7.22.7: emits :class:MultiScopeIgnoredWarning on first encounter of a scope carrying inherits_from. The SemanticLayer does NOT honour the field (facts are tenant-bound by entity construction); the warning surfaces the silent-ignore class so adopters know their inheritance setup is not propagating here.

Source code in src/symfonic/memory/layers/semantic.py

async def retrieve(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Retrieve semantic facts matching the query.

    v7.22 T-7.22.7: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope carrying ``inherits_from``.  The SemanticLayer
    does NOT honour the field (facts are tenant-bound by entity
    construction); the warning surfaces the silent-ignore class so
    adopters know their inheritance setup is not propagating here.
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    return await self.query_facts(scope, query, top_k)

store_fact `async` ¶

store_fact(
    scope: TenantScope, fact: MemoryEntry
) -> MemoryNode

Store a fact as a MemoryNode in the graph.

Parameters:

Name	Type	Description	Default
`scope`	`TenantScope`	Tenant scope for isolation.	required
`fact`	`MemoryEntry`	The memory entry representing the fact.	required

Returns:

Type	Description
`MemoryNode`	The persisted MemoryNode.

Source code in src/symfonic/memory/layers/semantic.py

async def store_fact(self, scope: TenantScope, fact: MemoryEntry) -> MemoryNode:
    """Store a fact as a MemoryNode in the graph.

    Args:
        scope: Tenant scope for isolation.
        fact: The memory entry representing the fact.

    Returns:
        The persisted MemoryNode.
    """
    # Prefer the LLM-provided label (passed via metadata) over the
    # rich_content string so that labels like "SOUL" are preserved
    # instead of becoming "profession: developer, ...".
    node_label = fact.metadata.get("label") or fact.content[:100]
    node = MemoryNode(
        layer=MemoryLayer.SEMANTIC,
        tenant_id=scope.tenant_id,
        label=str(node_label)[:100],
        properties={"content": fact.content, **fact.metadata},
        importance=fact.importance,
    )
    if fact.node_id:
        node.id = fact.node_id
    # v7.3 Item 13.1: auto-embed when a provider is configured and the
    # caller didn't pre-populate node.embedding. Caller-provided
    # embeddings always win; provider failures degrade to embedding=None
    # (see auto_embed.maybe_embed docstring).
    # TODO(v7.4): batch embeddings when add_node is called in a tight
    # loop (EntityLinker mints 5-10 entity nodes per cycle).
    node = await maybe_embed(
        node, self._embedding_provider, self._embedding_cache,
    )
    return await self._graph.add_node(scope, node)

summarize `async` ¶

summarize(
    scope: TenantScope, entries: list[MemoryEntry]
) -> str

Not meaningful for semantic facts. Returns empty string.

Source code in src/symfonic/memory/layers/semantic.py

async def summarize(self, scope: TenantScope, entries: list[MemoryEntry]) -> str:
    """Not meaningful for semantic facts. Returns empty string."""
    return ""

write `async` ¶

write(scope: TenantScope, entry: MemoryEntry) -> None

Write a memory entry as a semantic fact node.

Source code in src/symfonic/memory/layers/semantic.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write a memory entry as a semantic fact node."""
    await self.store_fact(scope, entry)

symfonic.memory.layers.semantic¶

semantic ¶

SemanticLayer ¶

delete async ¶

link async ¶

query_facts async ¶

render_for_prompt ¶

retrieve async ¶

store_fact async ¶

summarize async ¶

write async ¶

`symfonic.memory.layers.semantic`¶

delete `async` ¶

link `async` ¶

query_facts `async` ¶

retrieve `async` ¶

store_fact `async` ¶

summarize `async` ¶

write `async` ¶