Skip to content

Interfaces

Sync Protocols

emotional_memory.interfaces.Embedder

Bases: Protocol

Converts text to a dense vector representation.

emotional_memory.interfaces.MemoryStore

Bases: Protocol

Persistent storage for Memory objects.

emotional_memory.interfaces.AffectiveStateStore

Bases: Protocol

Persistent storage for the current affective state snapshot.

emotional_memory.interfaces.SequentialEmbedder

Base class for Embedder implementations that lack native batching.

Subclass this and implement embed(); embed_batch() is provided automatically by calling embed() for each item in sequence.

Example::

class MyEmbedder(SequentialEmbedder):
    def embed(self, text: str) -> list[float]:
        return my_model.encode(text).tolist()

Async Protocols

emotional_memory.interfaces_async.AsyncEmbedder

Bases: Protocol

Async variant of the Embedder protocol.

emotional_memory.interfaces_async.AsyncMemoryStore

Bases: Protocol

Async variant of the MemoryStore protocol.

count() replaces __len__ because __len__ cannot be declared async in Python.

emotional_memory.interfaces_async.AsyncAppraisalEngine

Bases: Protocol

Async variant of the AppraisalEngine protocol.

Async Adapters

emotional_memory.async_adapters.SyncToAsyncEmbedder(sync_embedder: Embedder)

Wraps a synchronous Embedder for use with AsyncEmotionalMemory.

Source code in src/emotional_memory/async_adapters.py 
49
50
def __init__(self, sync_embedder: Embedder) -> None:
    self._inner = sync_embedder

emotional_memory.async_adapters.SyncToAsyncStore(sync_store: MemoryStore)

Wraps a synchronous MemoryStore for use with AsyncEmotionalMemory.

Source code in src/emotional_memory/async_adapters.py 
67
68
def __init__(self, sync_store: MemoryStore) -> None:
    self._inner = sync_store

close() -> None async

Proxy close() to the wrapped store if it supports it.

Source code in src/emotional_memory/async_adapters.py 
91
92
93
94
95
async def close(self) -> None:
    """Proxy ``close()`` to the wrapped store if it supports it."""
    close = getattr(self._inner, "close", None)
    if callable(close):
        await asyncio.to_thread(close)

emotional_memory.async_adapters.SyncToAsyncAppraisalEngine(sync_engine: AppraisalEngine)

Wraps a synchronous AppraisalEngine for use with AsyncEmotionalMemory.

Source code in src/emotional_memory/async_adapters.py 
106
107
def __init__(self, sync_engine: AppraisalEngine) -> None:
    self._inner = sync_engine

emotional_memory.async_adapters.as_async(engine: EmotionalMemory) -> AsyncEmotionalMemory

Wrap a synchronous EmotionalMemory into an async one.

All I/O calls (embed, store, appraise) are delegated via asyncio.to_thread. The engine's affective state is copied at wrap time — the two engines are independent after this point and their states diverge as each processes new events. The underlying store and embedder are shared (same object references), so concurrent writes from both engines will interleave.

Source code in src/emotional_memory/async_adapters.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def as_async(engine: EmotionalMemory) -> AsyncEmotionalMemory:
    """Wrap a synchronous ``EmotionalMemory`` into an async one.

    All I/O calls (embed, store, appraise) are delegated via
    ``asyncio.to_thread``.  The engine's affective state is **copied** at
    wrap time — the two engines are independent after this point and their
    states diverge as each processes new events.  The underlying store and
    embedder are shared (same object references), so concurrent writes from
    both engines will interleave.
    """
    from emotional_memory.async_engine import AsyncEmotionalMemory

    appraisal_async = (
        SyncToAsyncAppraisalEngine(engine._appraisal_engine)
        if engine._appraisal_engine is not None
        else None
    )
    async_engine = AsyncEmotionalMemory(
        store=SyncToAsyncStore(engine._store),
        embedder=SyncToAsyncEmbedder(engine._embedder),
        appraisal_engine=appraisal_async,
        config=engine._config,
        state_store=engine._state_store,
        # Carry the classifier over: with query_classifier mode 'llm' the async
        # engine has no way to rebuild it from config alone, and dropping it
        # would silently disable per-query-type weight routing.
        query_classifier=engine._query_classifier,
    )
    # Share the current AffectiveState reference as the async engine's starting
    # point.  AffectiveState is always *replaced* (never mutated) on every
    # update, so both engines immediately diverge once either processes an event.
    # There is no aliasing hazard because the object is effectively immutable.
    async_engine._state = engine._state
    return async_engine