symfonic.memory.embeddings.auto_embed

auto_embed

Auto-embed helper for graph-backed memory layers.

v7.3 Roadmap Item 13.1 (PR-C). When a caller writes a node without pre-computing an embedding, the graph-backed layers (semantic / procedural / prospective) defer to :func:maybe_embed which:

1. Caller-provided embeddings always win -- if node.embedding is already set the node is returned unchanged (no provider call, no cache touch).
2. Per-kind text selection -- the text fed to EmbeddingProvider.embed is picked by :func:resolve_embedding_text (see its docstring for the rule).
3. Shared cache -- the orchestrator hands an :class:EmbeddingCache to every layer; identical text from different layers reuses one embedding instead of paying embed() twice.
4. Failure mode (load-bearing) -- if provider.embed raises, the exception is caught, a single WARNING is logged ("embedding_provider failed: %s") and the original node is returned with embedding=None. The write proceeds. This is the risk-register §12.8 default: a flaky embedder must NEVER block a memory write.

v7.3 ships per-call sync embeds; batched embeds across an add_node fan-out are deferred to v7.4 (see TODO at the call sites in the layer wrappers).

DEFAULT_CACHE_CAP module-attribute

DEFAULT_CACHE_CAP = 1024

Maximum entries retained by the shared :class:EmbeddingCache.

Sized for ~12-15 active tenants emitting ~70 unique entity surfaces each per consolidation cycle. Past the cap the oldest entry is evicted (OrderedDict.popitem(last=False)). The cache is in-memory only -- process restart drops it.

EmbeddingCache

EmbeddingCache(max_entries: int = DEFAULT_CACHE_CAP)

sha256(text) -> embedding LRU cache shared across layers.

Backed by :class:collections.OrderedDict so eviction is O(1) and deterministic. The cache is intentionally simple: no TTL, no persistence, no thread-locking (the symfonic memory pipeline is single-event-loop per process; concurrent writes from the same loop serialise on the await boundary inside maybe_embed).

Source code in src/symfonic/memory/embeddings/auto_embed.py

def __init__(self, max_entries: int = DEFAULT_CACHE_CAP) -> None:
    if max_entries < 1:
        msg = f"max_entries must be >= 1, got {max_entries}"
        raise ValueError(msg)
    self._max = max_entries
    self._store: OrderedDict[str, list[float]] = OrderedDict()

get

get(text: str) -> list[float] | None

Return the cached embedding for text or None.

Touching an entry marks it as most-recently-used (moves to end).

Source code in src/symfonic/memory/embeddings/auto_embed.py

def get(self, text: str) -> list[float] | None:
    """Return the cached embedding for *text* or ``None``.

    Touching an entry marks it as most-recently-used (moves to end).
    """
    key = self.key_for(text)
    if key not in self._store:
        return None
    self._store.move_to_end(key)
    return self._store[key]

key_for staticmethod

key_for(text: str) -> str

Return the sha256 cache key for text.

Source code in src/symfonic/memory/embeddings/auto_embed.py

@staticmethod
def key_for(text: str) -> str:
    """Return the sha256 cache key for *text*."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

set

set(text: str, embedding: list[float]) -> None

Store embedding for text. Evicts oldest when over cap.

Source code in src/symfonic/memory/embeddings/auto_embed.py

def set(self, text: str, embedding: list[float]) -> None:
    """Store *embedding* for *text*. Evicts oldest when over cap."""
    key = self.key_for(text)
    if key in self._store:
        # Refresh recency without growing the dict.
        self._store.move_to_end(key)
        self._store[key] = embedding
        return
    self._store[key] = embedding
    while len(self._store) > self._max:
        self._store.popitem(last=False)

maybe_embed async

maybe_embed(
    node: MemoryNode,
    provider: EmbeddingProvider | None,
    cache: EmbeddingCache | None = None,
) -> MemoryNode

Auto-embed node unless the caller already populated node.embedding.

This is the single chokepoint every graph-backed layer's add_node / update_node calls through. The contract is:

• Caller-provided node.embedding is preserved verbatim -- the provider is NOT invoked.
• provider is None (no provider configured) returns the node unchanged. This preserves byte-identical pre-Item-13.1 behaviour so existing deployments are unaffected.
• The text to embed is picked by :func:resolve_embedding_text (entity nodes prefer label, everyone else prefers content).
• Empty text short-circuits -- nothing to embed.
• Cache hits skip the provider call entirely. Cache misses populate the cache on success.
• Provider exceptions are caught, logged at WARNING, and the node is returned with embedding=None. This is the load-bearing failure mode (risk-register §12.8 option b) -- a flaky embedder must not block a write.

The returned node is the same instance when no auto-embed happened (preserves Pydantic mutation expectations) and a model_copy with embedding set otherwise.

Source code in src/symfonic/memory/embeddings/auto_embed.py

async def maybe_embed(
    node: MemoryNode,
    provider: EmbeddingProvider | None,
    cache: EmbeddingCache | None = None,
) -> MemoryNode:
    """Auto-embed *node* unless the caller already populated ``node.embedding``.

    This is the single chokepoint every graph-backed layer's
    ``add_node`` / ``update_node`` calls through. The contract is:

    * Caller-provided ``node.embedding`` is preserved verbatim -- the
      provider is NOT invoked.
    * ``provider is None`` (no provider configured) returns the node
      unchanged. This preserves byte-identical pre-Item-13.1 behaviour
      so existing deployments are unaffected.
    * The text to embed is picked by :func:`resolve_embedding_text`
      (entity nodes prefer label, everyone else prefers content).
    * Empty text short-circuits -- nothing to embed.
    * Cache hits skip the provider call entirely. Cache misses populate
      the cache on success.
    * Provider exceptions are caught, logged at WARNING, and the node is
      returned with ``embedding=None``. This is the load-bearing failure
      mode (risk-register §12.8 option b) -- a flaky embedder must not
      block a write.

    The returned node is the same instance when no auto-embed happened
    (preserves Pydantic mutation expectations) and a ``model_copy`` with
    ``embedding`` set otherwise.
    """
    if node.embedding is not None:
        return node
    if provider is None:
        return node

    text = resolve_embedding_text(node)
    if not text:
        return node

    if cache is not None:
        cached = cache.get(text)
        if cached is not None:
            return node.model_copy(update={"embedding": list(cached)})

    try:
        vector = await provider.embed(text)
    except Exception as exc:  # noqa: BLE001 -- load-bearing catch-all
        logger.warning("embedding_provider failed: %s", exc)
        return node

    # Defensive: a misbehaving provider returning None falls under the
    # same failure-mode contract as a raised exception -- log and skip.
    if vector is None:
        logger.warning(
            "embedding_provider failed: returned None for text len=%d",
            len(text),
        )
        return node

    embedding_list = list(vector)
    if cache is not None:
        cache.set(text, embedding_list)
    return node.model_copy(update={"embedding": embedding_list})

resolve_embedding_text

resolve_embedding_text(node: MemoryNode) -> str

Return the text to feed into EmbeddingProvider.embed for node.

Per-kind rule (v7.3 Item 13.1):

• Entity nodes -- when node.label starts with "Entity:" or node.properties['label_prefix'] equals "Entity", embed node.label (the canonical surface form). This is what RetrievalScorer should fuzzy-match across runs so spelling drift collapses to one retrieval hit.
• Default -- embed node.properties['content'] when that key is non-empty; otherwise fall back to node.label. MemoryNode has no top-level content field -- the convention across all five layers is properties['content'] carries the rich text.

Returns the empty string when neither label nor content is usable (caller is expected to skip the embed call in that case).

Source code in src/symfonic/memory/embeddings/auto_embed.py

def resolve_embedding_text(node: MemoryNode) -> str:
    """Return the text to feed into ``EmbeddingProvider.embed`` for *node*.

    Per-kind rule (v7.3 Item 13.1):

    * **Entity nodes** -- when ``node.label`` starts with ``"Entity:"``
      or ``node.properties['label_prefix']`` equals ``"Entity"``, embed
      ``node.label`` (the canonical surface form). This is what
      ``RetrievalScorer`` should fuzzy-match across runs so spelling
      drift collapses to one retrieval hit.
    * **Default** -- embed ``node.properties['content']`` when that key
      is non-empty; otherwise fall back to ``node.label``. ``MemoryNode``
      has no top-level ``content`` field -- the convention across all
      five layers is ``properties['content']`` carries the rich text.

    Returns the empty string when neither label nor content is usable
    (caller is expected to skip the embed call in that case).
    """
    label = str(node.label or "").strip()
    props = node.properties or {}
    label_prefix = str(props.get("label_prefix", "")).strip()

    is_entity = (
        label.startswith(f"{_ENTITY_LABEL_PREFIX}:")
        or label_prefix == _ENTITY_LABEL_PREFIX
    )
    if is_entity and label:
        return label

    content = props.get("content")
    if isinstance(content, str) and content.strip():
        return content
    return label