Skip to content

symfonic.memory.embeddings

embeddings

Embedding helpers shared by the graph-backed memory layers.

v7.3 Roadmap Item 13.1 surface. Provides:

  • :class:EmbeddingCache -- a shared sha256-keyed LRU mapping (cap 1024) that the orchestrator hands to every graph-backed layer so identical text from different layers reuses a single embed() call.
  • :func:maybe_embed -- the auto-embed wrapper invoked by :class:~symfonic.memory.layers.semantic.SemanticLayer, :class:~symfonic.memory.layers.procedural.layer.ProceduralLayer, and :class:~symfonic.memory.layers.prospective.ProspectiveLayer before delegating to GraphMemoryStore.add_node/update_node.
  • :func:make_embedding_provider -- the string-identifier factory consumed by the scaffold's EMBEDDING_PROVIDER env-var path so callers can opt into a provider without importing the symfonic-core internals directly.

The protocol itself (EmbeddingProvider) lives in symfonic.memory.protocols and is re-exported here for convenience.

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.

SUPPORTED_PROVIDER_IDENTIFIERS module-attribute 

SUPPORTED_PROVIDER_IDENTIFIERS = (
    "sentence-transformers",
    "openai",
)

String identifiers accepted by :func:make_embedding_provider.

The literal "none" and the empty string are also accepted and map to None (no auto-embed; caller still populates manually).

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 
108
109
110
111
112
113
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 
120
121
122
123
124
125
126
127
128
129
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 
115
116
117
118
@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 
131
132
133
134
135
136
137
138
139
140
141
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)

EmbeddingProvider

Bases: Protocol

Protocol for embedding text into vector representations.

embed async 

embed(text: str) -> list[float]

Embed a single text string into a vector.

Source code in src/symfonic/memory/protocols.py 
121
122
123
async def embed(self, text: str) -> list[float]:
    """Embed a single text string into a vector."""
    ...

embed_batch async 

embed_batch(texts: list[str]) -> list[list[float]]

Embed multiple texts into vectors.

Source code in src/symfonic/memory/protocols.py 
125
126
127
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
    """Embed multiple texts into vectors."""
    ...

make_embedding_provider 

make_embedding_provider(
    identifier: str | None,
) -> EmbeddingProvider | None

Build an :class:EmbeddingProvider from a string identifier.

Returns None for the default-off path (identifier in {None, "", "none"}). Raises ValueError on an unknown string so a typo in .env does not silently degrade to keyword-only retrieval.

The actual provider instance is constructed lazily -- importing this module does NOT pull in sentence_transformers or the OpenAI SDK. Each branch handles its own ImportError and re-raises as ValueError with the required pip-install hint.

Source code in src/symfonic/memory/embeddings/factory.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def make_embedding_provider(
    identifier: str | None,
) -> EmbeddingProvider | None:
    """Build an :class:`EmbeddingProvider` from a string *identifier*.

    Returns ``None`` for the default-off path (``identifier in
    {None, "", "none"}``). Raises ``ValueError`` on an unknown string so
    a typo in ``.env`` does not silently degrade to keyword-only
    retrieval.

    The actual provider instance is constructed lazily -- importing this
    module does NOT pull in ``sentence_transformers`` or the OpenAI SDK.
    Each branch handles its own ImportError and re-raises as
    ``ValueError`` with the required pip-install hint.
    """
    if identifier is None:
        return None
    name = identifier.strip().lower()
    if name in {"", "none"}:
        return None

    if name == "sentence-transformers":
        return _build_sentence_transformers_provider()
    if name == "openai":
        return _build_openai_provider()

    raise ValueError(
        f"Unknown EMBEDDING_PROVIDER identifier: {identifier!r}. "
        f"Supported: {', '.join(SUPPORTED_PROVIDER_IDENTIFIERS)} "
        f"or 'none' / empty for off.",
    )

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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
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 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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