Skip to content

teff.memory

teff.memory

Long-term memory: cross-session facts stored over a vector store.

Modules:

Name Description
base

Long-term memory: a namespace store over a vector store.
context

Context injection helpers for long-term memory.
extract

LLM-based fact extraction for long-term memory.
tool

Agent-facing tool for long-term memory (remember / recall / forget).

Classes:

Name Description
MemoryConfig

Declarative memory injection for agent / llm nodes.
MemoryExtractor

Extract durable facts from a conversation using an LLM.
MemoryItem

A single stored memory.
MemoryStore

Namespace-scoped semantic memory over a :class:VectorStore.
MemoryTool

Tool that lets an agent read and write long-term memory.

Functions:

Name Description
last_user_text

Return the most recent non-empty user message text.
memory_context

Return a formatted block of recalled memories, or "" if none.
memory_context_from_config

Recall block for a node's memory config, or "" when off.

MemoryConfig dataclass

Declarative memory injection for agent / llm nodes.

Passed to :class:~teff.node.agent.ReActAgent, :class:~teff.node.llm.LLM (and the flow.react() / flow.harness() / flow.llm() helpers) via the memory parameter. A plain config dict is accepted too — that is what YAML workflows deserialize to.

Attributes:

Name Type Description
store MemoryStore | dict | None

A ready :class:~teff.memory.base.MemoryStore, or a store config dict ({"type": "sqlite", ...}) to build with the node's provider registry.
namespace tuple[str, ...] | list[str] | str | None

Namespace subtree to recall from (a string becomes a single-segment namespace). Each segment may reference ${owner}, ${session_id} or ${checkpoint_id}, which are resolved from the enclosing run — so a shared graph serves per-user memory via ["users", "${owner}"].
k int

Maximum number of memories recalled per turn.
header str

First line of the injected block.

Methods:

Name Description
to_dict

Config-dict form (used internally; YAML round-trips as this).
Source code in teff/memory/context.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class MemoryConfig:
    """Declarative memory injection for ``agent`` / ``llm`` nodes.

    Passed to :class:`~teff.node.agent.ReActAgent`,
    :class:`~teff.node.llm.LLM` (and the ``flow.react()`` /
    ``flow.harness()`` / ``flow.llm()`` helpers) via the ``memory``
    parameter.  A plain config dict is accepted too — that is what YAML
    workflows deserialize to.

    Attributes:
        store: A ready :class:`~teff.memory.base.MemoryStore`, or a store
            config dict (``{"type": "sqlite", ...}``) to build with the
            node's provider registry.
        namespace: Namespace subtree to recall from (a string becomes a
            single-segment namespace).  Each segment may reference
            ``${owner}``, ``${session_id}`` or ``${checkpoint_id}``, which
            are resolved from the enclosing run — so a shared graph serves
            per-user memory via ``["users", "${owner}"]``.
        k: Maximum number of memories recalled per turn.
        header: First line of the injected block.
    """

    store: MemoryStore | dict | None = None
    namespace: tuple[str, ...] | list[str] | str | None = None
    k: int = 5
    header: str = DEFAULT_HEADER

    def to_dict(self) -> dict:
        """Config-dict form (used internally; YAML round-trips as this)."""
        return {
            "store": self.store,
            "namespace": self.namespace,
            "k": self.k,
            "header": self.header,
        }

to_dict 

to_dict()

Config-dict form (used internally; YAML round-trips as this).

Source code in teff/memory/context.py 
54
55
56
57
58
59
60
61
def to_dict(self) -> dict:
    """Config-dict form (used internally; YAML round-trips as this)."""
    return {
        "store": self.store,
        "namespace": self.namespace,
        "k": self.k,
        "header": self.header,
    }

MemoryExtractor

Extract durable facts from a conversation using an LLM.

Parameters:

Name Type Description Default
harness Harness | None

A :class:~teff.harness.loop.Harness used for the single extraction call. When omitted, one is built from model and provider.

 None
model str | None

Model name for a self-built harness (ignored when harness is given).

 None
provider str | None

Provider key for a self-built harness.

 None
system_prompt str | None

Overrides the default extraction prompt.

 None
temperature float

Sampling temperature for the extraction call.

 0.0

Methods:

Name Description
extract

Return the durable facts found in conversation.
save

Extract facts and write them into memory.
Source code in teff/memory/extract.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class MemoryExtractor:
    """Extract durable facts from a conversation using an LLM.

    Args:
        harness: A :class:`~teff.harness.loop.Harness` used for the single
            extraction call.  When omitted, one is built from *model* and
            *provider*.
        model: Model name for a self-built harness (ignored when *harness*
            is given).
        provider: Provider key for a self-built harness.
        system_prompt: Overrides the default extraction prompt.
        temperature: Sampling temperature for the extraction call.
    """

    def __init__(
        self,
        harness: Harness | None = None,
        *,
        model: str | None = None,
        provider: str | None = None,
        system_prompt: str | None = None,
        temperature: float = 0.0,
    ):
        if harness is None:
            if not model:
                raise ValueError(
                    "MemoryExtractor needs a harness or a model to build one"
                )
            harness = Harness(model=model, provider=provider, temperature=temperature)
        self._harness = harness
        self._system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT

    async def extract(self, conversation: list[dict]) -> list[str]:
        """Return the durable facts found in *conversation*.

        Args:
            conversation: OpenAI-style messages (``{"role", "content"}``);
                the text of every message is included in the prompt.
        """
        transcript = _transcript(conversation)
        reply = await self._harness.call(
            [
                {"role": "system", "content": self._system_prompt},
                {"role": "user", "content": transcript},
            ]
        )
        return _parse_facts(reply.content)

    async def save(
        self,
        memory: Any,
        conversation: list[dict],
        namespace: tuple[str, ...],
        *,
        ttl: float | None = None,
    ) -> list[tuple[str, str]]:
        """Extract facts and write them into *memory*.

        Each fact is stored under a stable key derived from its text (a
        short SHA-1), so re-extracting the same fact updates it in place.

        Args:
            memory: A :class:`~teff.memory.base.MemoryStore`.
            conversation: The messages to extract facts from.
            namespace: Namespace to store the facts under.
            ttl: Per-item TTL in seconds, or ``None`` for no expiry.

        Returns:
            The ``(key, fact)`` pairs that were written.
        """
        written: list[tuple[str, str]] = []
        for fact in await self.extract(conversation):
            key = _fact_key(fact)
            await memory.put(
                namespace, key, {"text": fact, "source": "extractor"}, ttl=ttl
            )
            written.append((key, fact))
        return written

extract async 

extract(conversation)

Return the durable facts found in conversation.

Parameters:

Name Type Description Default
conversation list[dict]

OpenAI-style messages ({"role", "content"}); the text of every message is included in the prompt.

 required
Source code in teff/memory/extract.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def extract(self, conversation: list[dict]) -> list[str]:
    """Return the durable facts found in *conversation*.

    Args:
        conversation: OpenAI-style messages (``{"role", "content"}``);
            the text of every message is included in the prompt.
    """
    transcript = _transcript(conversation)
    reply = await self._harness.call(
        [
            {"role": "system", "content": self._system_prompt},
            {"role": "user", "content": transcript},
        ]
    )
    return _parse_facts(reply.content)

save async 

save(memory, conversation, namespace, *, ttl=None)

Extract facts and write them into memory.

Each fact is stored under a stable key derived from its text (a short SHA-1), so re-extracting the same fact updates it in place.

Parameters:

Name Type Description Default
memory Any

A :class:~teff.memory.base.MemoryStore.

 required
conversation list[dict]

The messages to extract facts from.

 required
namespace tuple[str, ...]

Namespace to store the facts under.

 required
ttl float | None

Per-item TTL in seconds, or None for no expiry.

 None

Returns:

Type Description
list[tuple[str, str]]

The (key, fact) pairs that were written.
Source code in teff/memory/extract.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def save(
    self,
    memory: Any,
    conversation: list[dict],
    namespace: tuple[str, ...],
    *,
    ttl: float | None = None,
) -> list[tuple[str, str]]:
    """Extract facts and write them into *memory*.

    Each fact is stored under a stable key derived from its text (a
    short SHA-1), so re-extracting the same fact updates it in place.

    Args:
        memory: A :class:`~teff.memory.base.MemoryStore`.
        conversation: The messages to extract facts from.
        namespace: Namespace to store the facts under.
        ttl: Per-item TTL in seconds, or ``None`` for no expiry.

    Returns:
        The ``(key, fact)`` pairs that were written.
    """
    written: list[tuple[str, str]] = []
    for fact in await self.extract(conversation):
        key = _fact_key(fact)
        await memory.put(
            namespace, key, {"text": fact, "source": "extractor"}, ttl=ttl
        )
        written.append((key, fact))
    return written

MemoryItem dataclass

A single stored memory.

Attributes:

Name Type Description
key str

Item key within its namespace.
value dict

The stored {text, ...} dict (namespace metadata stripped).
namespace tuple[str, ...]

The namespace the item lives under.
updated_at float

Unix timestamp of the last write.
score float | None

Similarity score from a semantic search, or None for a recency-only lookup.
Source code in teff/memory/base.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass
class MemoryItem:
    """A single stored memory.

    Attributes:
        key: Item key within its namespace.
        value: The stored ``{text, ...}`` dict (namespace metadata stripped).
        namespace: The namespace the item lives under.
        updated_at: Unix timestamp of the last write.
        score: Similarity score from a semantic search, or ``None`` for a
            recency-only lookup.
    """

    key: str
    value: dict
    namespace: tuple[str, ...]
    updated_at: float
    score: float | None = None

MemoryStore

Namespace-scoped semantic memory over a :class:VectorStore.

Parameters:

Name Type Description Default
store VectorStore

Backing vector store (in-memory, sqlite, qdrant, ...).

 required
embedder Embedder

Embedding service used for put and search.

 required
ttl float | None

Default seconds an item lives unless overridden at put time. None (default) means items never expire.

 None

Methods:

Name Description
cleanup

Delete expired items; return how many were removed.
delete

Remove the item under namespace::key (no-op if absent).
get

Return the item under namespace::key, or None.
list

Return the keys stored under namespace (recency order).
put

Upsert value under namespace::key.
search

Return the k most relevant items under namespace.

Attributes:

Name Type Description
store VectorStore

The backing vector store (exposed for lifecycle tools).
Source code in teff/memory/base.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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
149
150
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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class MemoryStore:
    """Namespace-scoped semantic memory over a :class:`VectorStore`.

    Args:
        store: Backing vector store (in-memory, sqlite, qdrant, ...).
        embedder: Embedding service used for ``put`` and ``search``.
        ttl: Default seconds an item lives unless overridden at ``put``
            time.  ``None`` (default) means items never expire.
    """

    def __init__(
        self,
        store: VectorStore,
        embedder: Embedder,
        *,
        ttl: float | None = None,
    ):
        self._store = store
        self._embedder = embedder
        self._ttl = ttl

    @property
    def store(self) -> VectorStore:
        """The backing vector store (exposed for lifecycle tools)."""
        return self._store

    async def put(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict,
        *,
        ttl: float | None | object = _UNSET,
    ) -> None:
        """Upsert *value* under ``namespace::key``.

        Args:
            namespace: Hierarchical path (``("users", "u1")``).
            key: Unique key within the namespace; writing to an existing
                key overwrites it.
            value: The memory dict; must contain a non-empty ``text``
                field (the part that is embedded).
            ttl: ``None`` disables expiry for this item; a number overrides
                the store-level TTL; omitted uses the store default.
        """
        text = value.get("text")
        if not text:
            raise ValueError("memory value requires a non-empty 'text' field")
        vec = await self._embedder.embed(text)
        now = time.time()
        meta: dict[str, Any] = {
            **value,
            **_ns_filter(namespace),
            "updated_at": now,
        }
        eff_ttl: float | None
        if ttl is _UNSET:
            eff_ttl = self._ttl
        elif isinstance(ttl, (int, float)):
            eff_ttl = ttl
        else:
            eff_ttl = None
        if eff_ttl is not None:
            meta["expires_at"] = now + eff_ttl
        await self._store.add([(_item_id(namespace, key), vec, meta)])

    async def get(self, namespace: tuple[str, ...], key: str) -> MemoryItem | None:
        """Return the item under ``namespace::key``, or ``None``.

        Expired items are reported as missing.
        """
        rows = await self._store.get([_item_id(namespace, key)])
        if not rows:
            return None
        _id, meta = rows[0]
        if self._is_expired(meta):
            return None
        return self._to_item(namespace, key, meta)

    async def delete(self, namespace: tuple[str, ...], key: str) -> None:
        """Remove the item under ``namespace::key`` (no-op if absent)."""
        await self._store.delete([_item_id(namespace, key)])

    async def search(
        self,
        namespace: tuple[str, ...],
        *,
        query: str | None = None,
        k: int = 10,
        filter: dict | None = None,
    ) -> list[MemoryItem]:
        """Return the *k* most relevant items under *namespace*.

        With *query*, items are ranked by semantic similarity to it; a
        ``namespace`` match also covers deeper sub-namespaces (prefix
        semantics).  Without *query*, the most recently written items are
        returned instead.

        Args:
            namespace: Namespace subtree to search.
            query: Natural-language query; when ``None`` the search falls
                back to recency order.
            k: Maximum number of results.
            filter: Extra metadata filter DSL (see
                :func:`~teff.rag.base.match_filter`).
        """
        eff_filter = {**_ns_filter(namespace), **(filter or {})}
        if query:
            qvec = await self._embedder.embed(query)
            results = await self._store.search(
                qvec,
                k=max(k, k * 2),
                filter=eff_filter,
                query_text=query,
            )
        else:
            rows = await self._store.entries(limit=100_000)
            results = [
                (_id, 0.0, meta) for _id, meta in rows if match_filter(meta, eff_filter)
            ]
            results.sort(key=lambda r: r[2].get("updated_at", 0.0), reverse=True)

        items: list[MemoryItem] = []
        for _id, score, meta in results:
            if self._is_expired(meta):
                continue
            ns, key = self._split_id(_id)
            item = self._to_item(ns, key, meta)
            item.score = score if query else None
            items.append(item)
            if len(items) >= k:
                break
        return items

    async def list(
        self, namespace: tuple[str, ...], limit: int = 100, offset: int = 0
    ) -> list[str]:
        """Return the keys stored under *namespace* (recency order)."""
        items = await self.search(namespace, k=limit + offset)
        keys = [i.key for i in items if i.key]
        return keys[offset : offset + limit]

    async def cleanup(self, *, max_age: float | None = None) -> int:
        """Delete expired items; return how many were removed.

        ``max_age`` additionally removes items whose ``updated_at`` is
        older than that many seconds.  Expired items are removed from the
        backing store (not merely hidden).
        """
        rows = await self._store.entries(limit=100_000)
        now = time.time()
        to_delete: list[str] = []
        for _id, meta in rows:
            if self._is_expired(meta):
                to_delete.append(_id)
                continue
            if max_age is not None:
                updated = meta.get("updated_at", 0.0)
                if updated and now - updated > max_age:
                    to_delete.append(_id)
        if to_delete:
            await self._store.delete(to_delete)
        return len(to_delete)

    def _is_expired(self, meta: dict) -> bool:
        expires = meta.get("expires_at")
        return isinstance(expires, (int, float)) and expires <= time.time()

    def _to_item(self, namespace: tuple[str, ...], key: str, meta: dict) -> MemoryItem:
        value = {k: v for k, v in meta.items() if not _is_meta_key(k)}
        return MemoryItem(
            key=key,
            value=value,
            namespace=namespace,
            updated_at=float(meta.get("updated_at", 0.0)),
        )

    def _split_id(self, item_id: str) -> tuple[tuple[str, ...], str]:
        parts = item_id.split("::")
        if len(parts) < 2:
            return (), parts[0]
        return tuple(parts[:-1]), parts[-1]

store property 

store

The backing vector store (exposed for lifecycle tools).

cleanup async 

cleanup(*, max_age=None)

Delete expired items; return how many were removed.

max_age additionally removes items whose updated_at is older than that many seconds. Expired items are removed from the backing store (not merely hidden).

Source code in teff/memory/base.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def cleanup(self, *, max_age: float | None = None) -> int:
    """Delete expired items; return how many were removed.

    ``max_age`` additionally removes items whose ``updated_at`` is
    older than that many seconds.  Expired items are removed from the
    backing store (not merely hidden).
    """
    rows = await self._store.entries(limit=100_000)
    now = time.time()
    to_delete: list[str] = []
    for _id, meta in rows:
        if self._is_expired(meta):
            to_delete.append(_id)
            continue
        if max_age is not None:
            updated = meta.get("updated_at", 0.0)
            if updated and now - updated > max_age:
                to_delete.append(_id)
    if to_delete:
        await self._store.delete(to_delete)
    return len(to_delete)

delete async 

delete(namespace, key)

Remove the item under namespace::key (no-op if absent).

Source code in teff/memory/base.py 
146
147
148
async def delete(self, namespace: tuple[str, ...], key: str) -> None:
    """Remove the item under ``namespace::key`` (no-op if absent)."""
    await self._store.delete([_item_id(namespace, key)])

get async 

get(namespace, key)

Return the item under namespace::key, or None.

Expired items are reported as missing.

Source code in teff/memory/base.py 
133
134
135
136
137
138
139
140
141
142
143
144
async def get(self, namespace: tuple[str, ...], key: str) -> MemoryItem | None:
    """Return the item under ``namespace::key``, or ``None``.

    Expired items are reported as missing.
    """
    rows = await self._store.get([_item_id(namespace, key)])
    if not rows:
        return None
    _id, meta = rows[0]
    if self._is_expired(meta):
        return None
    return self._to_item(namespace, key, meta)

list async 

list(namespace, limit=100, offset=0)

Return the keys stored under namespace (recency order).

Source code in teff/memory/base.py 
201
202
203
204
205
206
207
async def list(
    self, namespace: tuple[str, ...], limit: int = 100, offset: int = 0
) -> list[str]:
    """Return the keys stored under *namespace* (recency order)."""
    items = await self.search(namespace, k=limit + offset)
    keys = [i.key for i in items if i.key]
    return keys[offset : offset + limit]

put async 

put(namespace, key, value, *, ttl=_UNSET)

Upsert value under namespace::key.

Parameters:

Name Type Description Default
namespace tuple[str, ...]

Hierarchical path (("users", "u1")).

 required
key str

Unique key within the namespace; writing to an existing key overwrites it.

 required
value dict

The memory dict; must contain a non-empty text field (the part that is embedded).

 required
ttl float | None | object

None disables expiry for this item; a number overrides the store-level TTL; omitted uses the store default.

 _UNSET
Source code in teff/memory/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def put(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: dict,
    *,
    ttl: float | None | object = _UNSET,
) -> None:
    """Upsert *value* under ``namespace::key``.

    Args:
        namespace: Hierarchical path (``("users", "u1")``).
        key: Unique key within the namespace; writing to an existing
            key overwrites it.
        value: The memory dict; must contain a non-empty ``text``
            field (the part that is embedded).
        ttl: ``None`` disables expiry for this item; a number overrides
            the store-level TTL; omitted uses the store default.
    """
    text = value.get("text")
    if not text:
        raise ValueError("memory value requires a non-empty 'text' field")
    vec = await self._embedder.embed(text)
    now = time.time()
    meta: dict[str, Any] = {
        **value,
        **_ns_filter(namespace),
        "updated_at": now,
    }
    eff_ttl: float | None
    if ttl is _UNSET:
        eff_ttl = self._ttl
    elif isinstance(ttl, (int, float)):
        eff_ttl = ttl
    else:
        eff_ttl = None
    if eff_ttl is not None:
        meta["expires_at"] = now + eff_ttl
    await self._store.add([(_item_id(namespace, key), vec, meta)])

search async 

search(namespace, *, query=None, k=10, filter=None)

Return the k most relevant items under namespace.

With query, items are ranked by semantic similarity to it; a namespace match also covers deeper sub-namespaces (prefix semantics). Without query, the most recently written items are returned instead.

Parameters:

Name Type Description Default
namespace tuple[str, ...]

Namespace subtree to search.

 required
query str | None

Natural-language query; when None the search falls back to recency order.

 None
k int

Maximum number of results.

 10
filter dict | None

Extra metadata filter DSL (see :func:~teff.rag.base.match_filter).

 None
Source code in teff/memory/base.py 
150
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
async def search(
    self,
    namespace: tuple[str, ...],
    *,
    query: str | None = None,
    k: int = 10,
    filter: dict | None = None,
) -> list[MemoryItem]:
    """Return the *k* most relevant items under *namespace*.

    With *query*, items are ranked by semantic similarity to it; a
    ``namespace`` match also covers deeper sub-namespaces (prefix
    semantics).  Without *query*, the most recently written items are
    returned instead.

    Args:
        namespace: Namespace subtree to search.
        query: Natural-language query; when ``None`` the search falls
            back to recency order.
        k: Maximum number of results.
        filter: Extra metadata filter DSL (see
            :func:`~teff.rag.base.match_filter`).
    """
    eff_filter = {**_ns_filter(namespace), **(filter or {})}
    if query:
        qvec = await self._embedder.embed(query)
        results = await self._store.search(
            qvec,
            k=max(k, k * 2),
            filter=eff_filter,
            query_text=query,
        )
    else:
        rows = await self._store.entries(limit=100_000)
        results = [
            (_id, 0.0, meta) for _id, meta in rows if match_filter(meta, eff_filter)
        ]
        results.sort(key=lambda r: r[2].get("updated_at", 0.0), reverse=True)

    items: list[MemoryItem] = []
    for _id, score, meta in results:
        if self._is_expired(meta):
            continue
        ns, key = self._split_id(_id)
        item = self._to_item(ns, key, meta)
        item.score = score if query else None
        items.append(item)
        if len(items) >= k:
            break
    return items

MemoryTool

Bases: Tool

Tool that lets an agent read and write long-term memory.

Usage::

memory = MemoryTool(
    store=SQLiteVectorStore(path="./memory.db", dim=768),
    embedder=Embedder(provider="ollama", model="nomic-embed-text"),
    namespace=("users", "u1"),
)
await memory.arun(action="remember", text="prefers email over Slack")
result = await memory.arun(action="recall", query="how to reach them?")

Actions (passed as action):

  • remember — upsert a fact (text plus optional metadata). When similarity_threshold is set and a semantically close item already exists in the namespace, the new text overwrites that item instead of creating a duplicate.
  • recall — return top-k memories for a query (or the most recent if no query is given), formatted for a prompt.
  • forget — delete the memory at key.
  • list — enumerate stored keys.

Can be built from a config dict (e.g. a tools: entry in a workflow YAML)::

{
  "name": "memory",
  "store": {"type": "sqlite", "path": "./memory.db", "dim": 768},
  "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
  "namespace": ["users", "${USER_ID}"],
  "default_k": 5,
  "similarity_threshold": 0.6,
}

Supported store types match RAGTool: in_memory (default), sqlite, chroma, qdrant, pgvector, faiss, lance, milvus, weaviate, pinecone.

Methods:

Name Description
arun

Run a memory operation and return a human-readable result.
Source code in teff/memory/tool.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class MemoryTool(Tool):
    """Tool that lets an agent read and write long-term memory.

    Usage::

        memory = MemoryTool(
            store=SQLiteVectorStore(path="./memory.db", dim=768),
            embedder=Embedder(provider="ollama", model="nomic-embed-text"),
            namespace=("users", "u1"),
        )
        await memory.arun(action="remember", text="prefers email over Slack")
        result = await memory.arun(action="recall", query="how to reach them?")

    Actions (passed as ``action``):

    - ``remember`` — upsert a fact (``text`` plus optional ``metadata``).
      When ``similarity_threshold`` is set and a semantically close item
      already exists in the namespace, the new text overwrites that item
      instead of creating a duplicate.
    - ``recall`` — return top-*k* memories for a ``query`` (or the most
      recent if no query is given), formatted for a prompt.
    - ``forget`` — delete the memory at ``key``.
    - ``list`` — enumerate stored keys.

    Can be built from a config dict (e.g. a ``tools:`` entry in a
    workflow YAML)::

        {
          "name": "memory",
          "store": {"type": "sqlite", "path": "./memory.db", "dim": 768},
          "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
          "namespace": ["users", "${USER_ID}"],
          "default_k": 5,
          "similarity_threshold": 0.6,
        }

    Supported store types match ``RAGTool``: ``in_memory`` (default),
    ``sqlite``, ``chroma``, ``qdrant``, ``pgvector``, ``faiss``, ``lance``,
    ``milvus``, ``weaviate``, ``pinecone``.
    """

    name = "memory"
    description = (
        "Long-term memory: remember facts, recall relevant memories, "
        "forget, and list what is stored."
    )

    def __init__(
        self,
        config: dict | None = None,
        *,
        store: VectorStore | None = None,
        embedder: Embedder | None = None,
        namespace: tuple[str, ...] | list[str] = (),
        default_k: int = 5,
        similarity_threshold: float | None = None,
        ttl: float | None = None,
    ):
        self._memory: MemoryStore | None = None
        self._namespace = tuple(namespace)
        self._default_k = default_k
        self._threshold = similarity_threshold
        self._ttl = ttl
        if isinstance(config, dict):
            self._apply_config(config)
        elif store is not None and embedder is not None:
            self.memory = MemoryStore(store=store, embedder=embedder, ttl=ttl)

    @property
    def memory(self) -> MemoryStore:
        if self._memory is None:
            raise RuntimeError("memory store not initialised")
        return self._memory

    @memory.setter
    def memory(self, value: MemoryStore) -> None:
        self._memory = value

    def _apply_config(self, config: dict) -> None:
        self.memory = memory_from_config(config, default_ttl=self._ttl)
        ns = config.get("namespace")
        if ns:
            self._namespace = tuple(str(part) for part in ns)
        if config.get("default_k") is not None:
            self._default_k = int(config["default_k"])
        if config.get("similarity_threshold") is not None:
            self._threshold = float(config["similarity_threshold"])

    async def arun(  # type: ignore[override]
        self,
        action: str = "recall",
        key: str = "",
        text: str = "",
        value: dict | None = None,
        query: str = "",
        metadata: dict | None = None,
        k: int | None = None,
    ) -> str:
        """Run a memory operation and return a human-readable result.

        The namespace is fixed at construction time and can never be
        overridden by the caller — an agent cannot address another owner's
        memories by passing a namespace.  Per-owner isolation is achieved by
        building one tool per owner (``namespace=("users", owner)``).
        """
        ns = self._namespace
        eff_k = int(k) if k is not None else self._default_k
        mem = self.memory

        if action == "remember":
            return await self._remember(ns, key, text, value, metadata)
        if action == "recall":
            items = await mem.search(ns, query=query or None, k=eff_k)
            return _format_recall(items)
        if action == "forget":
            if not key:
                return "forget requires a `key`"
            await mem.delete(ns, key)
            return f"forgotten {key!r}"
        if action == "list":
            keys = await mem.list(ns, limit=1000)
            return "\n".join(keys) if keys else "(no memories)"
        raise ValueError(f"unknown memory action: {action!r}")

    async def _remember(
        self,
        ns: tuple[str, ...],
        key: str,
        text: str,
        value: dict | None,
        metadata: dict | None,
    ) -> str:
        if value is None:
            if not text:
                return "remember requires `text`"
            value = {"text": text, **(metadata or {})}
        elif "text" not in value:
            return "remember `value` requires a 'text' field"

        if self._threshold is not None:
            similar = await self.memory.search(ns, query=value["text"], k=1)
            if (
                similar
                and similar[0].score is not None
                and similar[0].score >= self._threshold
            ):
                key = similar[0].key

        final_key = key or uuid.uuid4().hex[:12]
        await self.memory.put(ns, final_key, value)
        return f"remembered {final_key!r}"

arun async 

arun(action='recall', key='', text='', value=None, query='', metadata=None, k=None)

Run a memory operation and return a human-readable result.

The namespace is fixed at construction time and can never be overridden by the caller — an agent cannot address another owner's memories by passing a namespace. Per-owner isolation is achieved by building one tool per owner (namespace=("users", owner)).

Source code in teff/memory/tool.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def arun(  # type: ignore[override]
    self,
    action: str = "recall",
    key: str = "",
    text: str = "",
    value: dict | None = None,
    query: str = "",
    metadata: dict | None = None,
    k: int | None = None,
) -> str:
    """Run a memory operation and return a human-readable result.

    The namespace is fixed at construction time and can never be
    overridden by the caller — an agent cannot address another owner's
    memories by passing a namespace.  Per-owner isolation is achieved by
    building one tool per owner (``namespace=("users", owner)``).
    """
    ns = self._namespace
    eff_k = int(k) if k is not None else self._default_k
    mem = self.memory

    if action == "remember":
        return await self._remember(ns, key, text, value, metadata)
    if action == "recall":
        items = await mem.search(ns, query=query or None, k=eff_k)
        return _format_recall(items)
    if action == "forget":
        if not key:
            return "forget requires a `key`"
        await mem.delete(ns, key)
        return f"forgotten {key!r}"
    if action == "list":
        keys = await mem.list(ns, limit=1000)
        return "\n".join(keys) if keys else "(no memories)"
    raise ValueError(f"unknown memory action: {action!r}")

last_user_text 

last_user_text(messages, fallback='')

Return the most recent non-empty user message text.

Source code in teff/memory/context.py 
145
146
147
148
149
150
151
152
153
def last_user_text(messages: list[dict], fallback: str = "") -> str:
    """Return the most recent non-empty user message text."""
    for msg in reversed(messages or []):
        if msg.get("role") != "user":
            continue
        text = str(msg.get("content", "")).strip()
        if text:
            return text
    return fallback

memory_context async 

memory_context(store, query, *, namespace=(), k=5, header=DEFAULT_HEADER, bullet='-')

Return a formatted block of recalled memories, or "" if none.

Parameters:

Name Type Description Default
store Any

A :class:~teff.memory.base.MemoryStore.

 required
query str

Natural-language query used for the semantic recall.

 required
namespace tuple[str, ...]

Namespace subtree to recall from.

 ()
k int

Maximum number of memories to include.

 5
header str

First line of the block.

 DEFAULT_HEADER
bullet str

Per-item bullet prefix.

 '-'

The returned string is meant to be appended to a system prompt; it is empty when nothing matched, so callers can skip it entirely.

Source code in teff/memory/context.py 
110
111
112
113
114
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
async def memory_context(
    store: Any,
    query: str,
    *,
    namespace: tuple[str, ...] = (),
    k: int = 5,
    header: str = DEFAULT_HEADER,
    bullet: str = "-",
) -> str:
    """Return a formatted block of recalled memories, or ``""`` if none.

    Args:
        store: A :class:`~teff.memory.base.MemoryStore`.
        query: Natural-language query used for the semantic recall.
        namespace: Namespace subtree to recall from.
        k: Maximum number of memories to include.
        header: First line of the block.
        bullet: Per-item bullet prefix.

    The returned string is meant to be appended to a system prompt; it is
    empty when nothing matched, so callers can skip it entirely.
    """
    if not query or not str(query).strip():
        return ""
    items = await store.search(namespace, query=str(query), k=k)
    lines = [
        f"{bullet} {item.value.get('text', '')}"
        for item in items
        if item.value.get("text")
    ]
    if not lines:
        return ""
    return f"{header}\n" + "\n".join(lines)

memory_context_from_config async 

memory_context_from_config(cfg, *, state, ctx)

Recall block for a node's memory config, or "" when off.

Shared by :class:~teff.node.agent.ReActAgent and :class:~teff.node.llm.LLM. Reads the node's memory config ({store, namespace, k, header}), resolves the store — a :class:~teff.memory.base.MemoryStore instance, or a config dict built via memory_from_config using ctx's provider registry — and recalls memories for the most recent user message. Namespace segments may reference ${owner} / ${session_id} / ${checkpoint_id}, resolved from ctx — the building block for per-user memory behind a shared multi-tenant graph.

The returned block is meant to be prepended to the LLM messages as a system message; it is empty when memory is unconfigured or nothing matched.

Source code in teff/memory/context.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def memory_context_from_config(cfg: dict, *, state: dict, ctx: Any) -> str:
    """Recall block for a node's ``memory`` config, or ``""`` when off.

    Shared by :class:`~teff.node.agent.ReActAgent` and
    :class:`~teff.node.llm.LLM`.  Reads the node's ``memory`` config
    (``{store, namespace, k, header}``), resolves the store — a
    :class:`~teff.memory.base.MemoryStore` instance, or a config dict
    built via ``memory_from_config`` using *ctx*'s provider registry —
    and recalls memories for the most recent user message.  Namespace
    segments may reference ``${owner}`` / ``${session_id}`` /
    ``${checkpoint_id}``, resolved from *ctx* — the building block for
    per-user memory behind a shared multi-tenant graph.

    The returned block is meant to be prepended to the LLM messages as a
    ``system`` message; it is empty when memory is unconfigured or
    nothing matched.
    """
    memory_cfg = cfg.get("memory")
    if not memory_cfg:
        return ""
    if isinstance(memory_cfg, MemoryConfig):
        memory_cfg = memory_cfg.to_dict()
    mem_store = memory_cfg.get("store")
    if isinstance(mem_store, dict):
        from teff.memory.tool import memory_from_config

        mem_store = memory_from_config(
            memory_cfg,
            providers=getattr(ctx, "providers", None),
            default_provider=getattr(ctx, "default_provider", None),
        )
    if mem_store is None:
        return ""
    ns_raw = memory_cfg.get("namespace")
    namespace = _resolve_namespace(ns_raw, ctx)
    messages = list(state.get(cfg.get("messages_key", "messages"), []) or [])
    fallback = str(state.get(cfg.get("input_key", "input"), ""))
    return await memory_context(
        mem_store,
        last_user_text(messages, fallback=fallback),
        namespace=namespace,
        k=int(memory_cfg.get("k", 5)),
        header=str(memory_cfg.get("header", DEFAULT_HEADER)),
    )