Memory API Reference¶

Agent memory integration layer with auto-injection. Provides a MemoryProvider protocol and adapters for external memory backends (Mem0, ChromaDB) plus a lightweight InMemoryProvider for testing. The key feature is auto-injection: MemoryAgent wraps any LangGraph agent and transparently prepends relevant memories to every invocation.

MemoryResult¶

`promptise.memory.MemoryResult` `dataclass` ¶

A single memory search result.

Attributes:

Name	Type	Description
`content`	`str`	The stored text.
`score`	`float`	Relevance score (0.0 = no match, 1.0 = perfect). Clamped to `[0.0, 1.0]` on construction.
`memory_id`	`str`	Unique identifier for this memory entry.
`metadata`	`dict[str, Any]`	Provider-specific metadata.

MemoryProvider¶

`promptise.memory.MemoryProvider` ¶

Bases: Protocol

Interface for memory backends.

All methods are async. Implementations must handle their own thread-safety and connection management. Callers should call :meth:close when the provider is no longer needed.

Multi-user isolation is controlled by the scope passed to the concrete provider (see :class:MemoryScope). When scope is PER_USER, every mutating / querying call must carry a user_id so the provider can partition entries. When scope is SHARED the user_id parameter is ignored.

`add(content, *, metadata=None, user_id=None)` `async` ¶

Store a new memory.

Parameters:

Name	Type	Description	Default
`content`	`str`	Text to remember.	required
`metadata`	`dict[str, Any] \| None`	Optional metadata (e.g. source, tags).	`None`
`user_id`	`str \| None`	Owning user for per-user scoped providers. Ignored when `scope=SHARED`. Required when `scope=PER_USER`.	`None`

Returns:

Type	Description
`str`	The `memory_id` of the stored entry.

`close()` `async` ¶

Release resources (connections, file handles).

`delete(memory_id, *, user_id=None)` `async` ¶

Delete a memory by ID.

Per-user scoped providers refuse to delete entries that belong to a different user_id and return False.

Returns:

Type	Description
`bool`	`True` if the entry existed, was owned by `user_id`
`bool`	(when scoped), and was deleted.

`purge_user(user_id)` `async` ¶

Delete every entry owned by user_id.

Use it to delete one user's data on request. For SHARED scope providers this is a no-op that returns 0 — there is no per-user ownership to honor.

Returns:

Type	Description
`int`	The number of entries removed.

`search(query, *, limit=5, user_id=None)` `async` ¶

Search for memories relevant to query.

Parameters:

Name	Type	Description	Default
`query`	`str`	Natural-language search query.	required
`limit`	`int`	Maximum results to return.	`5`
`user_id`	`str \| None`	Owning user for per-user scoped providers. Ignored when `scope=SHARED`. Required when `scope=PER_USER` — raises :class:`MemoryIsolationError` when missing.	`None`

Returns:

Type	Description
`list[MemoryResult]`	Ranked list of :class:`MemoryResult` (best match first).

InMemoryProvider¶

`promptise.memory.InMemoryProvider` ¶

Keyword-search memory provider for testing.

Stores entries in a dictionary. Search ranks entries that contain the whole query first, then entries by the share of the query's words they contain (case-insensitive, common stop words ignored). Entries that share no word with the query are not returned. There are no embeddings, so synonyms do not match — not suitable for production.

Parameters:

Name	Type	Description	Default
`max_entries`	`int`	Maximum stored entries. When exceeded, oldest entries are evicted (FIFO).	`10000`
`scope`	`MemoryScope`	Isolation mode. `SHARED` (default) lets every caller see every entry. `PER_USER` partitions entries by `user_id` — search / delete only operate on rows the caller owns.	`SHARED`

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

List stored entries whose metadata matches metadata exactly.

Unlike :meth:search this does not rank by relevance; it is how callers enumerate entries they tagged (the adaptive strategy system uses it to count failures and cap stored lessons). Scoping follows :meth:search. Entries are returned oldest first, score=1.0.

ChromaProvider¶

`promptise.memory.ChromaProvider` ¶

Adapter for ChromaDB <https://www.trychroma.com/>_.

Requires pip install chromadb (or pip install "promptise[all]"). Provides local vector similarity search with automatic embedding generation. ChromaDB handles its own input validation internally.

Scores. New collections are created with cosine distance (metadata={"hnsw:space": "cosine"}), and a result's :attr:MemoryResult.score is its cosine similarity (1 - distance, clamped to [0, 1]). A collection that already exists keeps the distance function it was created with — Chroma cannot change it — and the score is derived from that function: ip uses 1 - distance, and l2 (Chroma's default, used by collections created by Promptise 1.2.1 and earlier) uses 1 - distance / 2, which equals cosine similarity for unit-length embeddings such as Chroma's default model. A warning is logged for an l2 collection; re-create it to get cosine distances (see the memory guide).

Parameters:

Name	Type	Description	Default
`collection_name`	`str`	Name of the ChromaDB collection.	`'agent_memory'`
`persist_directory`	`str \| None`	Path for persistent storage. When `None`, uses an ephemeral in-memory client.	`None`
`embedding_function`	`Any`	Custom ChromaDB embedding function. When `None`, uses ChromaDB's default (all-MiniLM-L6-v2 via Sentence Transformers).	`None`
`scope`	`MemoryScope`	Isolation mode (see :class:`MemoryScope`).	`SHARED`

Raises:

Type	Description
`ImportError`	If `chromadb` is not installed.

provider = ChromaProvider(
    collection_name="agent_memory",
    persist_directory=".promptise/chroma",
)
await provider.add("User prefers dark mode")
results = await provider.search("theme preferences")

`close()` `async` ¶

Release ChromaDB client resources.

After calling close(), any further operations will raise RuntimeError. ChromaDB's PersistentClient does not expose an explicit close method, so we null references to allow garbage collection.

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

List entries whose metadata matches metadata exactly.

Runs a ChromaDB get(where=...); see :meth:InMemoryProvider.list_entries. Order is ChromaDB's.

`purge_user(user_id)` `async` ¶

Delete every Chroma entry owned by user_id.

Uses ChromaDB's delete(where=...) filter to drop all rows in a single round-trip. Returns 0 for SHARED-scope providers.

Mem0Provider¶

`promptise.memory.Mem0Provider` ¶

Adapter for mem0 <https://github.com/mem0ai/mem0>_.

Requires pip install mem0ai (or pip install "promptise[all]"). Supported releases: mem0ai>=0.1.118,<3 (the 0.1, 1.x and 2.x lines). Wraps Mem0's hybrid retrieval (vector + optional graph search) behind the :class:MemoryProvider protocol.

Mem0 changed its Python API between major versions: from 2.0, Memory.search() and Memory.get_all() take the entity ids in a filters dict and the result count as top_k (and reject user_id= / limit=), while 0.1 and 1.x take user_id= / agent_id= / limit=. The adapter reads the installed client's search() signature once and calls whichever form it accepts. Memory.add() takes the text as messages in every supported release.

Mem0 can run fully local (with Ollama) or via their cloud platform. Pass a config dict to from_config() for self-hosted setups.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	Scopes memories to a user (required by Mem0).	`'default'`
`agent_id`	`str \| None`	Optional agent identifier for multi-agent scoping.	`None`
`config`	`dict[str, Any] \| None`	Optional Mem0 configuration dict. Passed directly to `mem0.Memory.from_config()`. If `None`, uses Mem0 defaults.	`None`
`scope`	`MemoryScope`	Isolation mode (see :class:`MemoryScope`).	`SHARED`

Raises:

Type	Description
`ImportError`	If `mem0ai` is not installed.

provider = Mem0Provider(user_id="user-42")
await provider.add("User prefers dark mode")
results = await provider.search("theme preferences")

`close()` `async` ¶

Release Mem0 client resources. Stored memories are kept.

Calls the client's close() when it has one, then drops the reference. After calling close(), any further operations will raise RuntimeError.

In 1.4.3 and earlier this called Mem0's Memory.reset(), which deletes every stored memory (the vector store collection and the history database), so closing a provider (AgentProcess.close(), async with, removing a runtime process) wiped the memories it was meant to keep. To delete memories on purpose, use :meth:delete / :meth:purge_user, or Mem0's own API.

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

List the user's entries whose metadata matches metadata exactly.

Uses Mem0's get_all (up to limit entries, in the API form the installed release accepts -- see :meth:search) and filters client-side. See :meth:InMemoryProvider.list_entries.

`purge_user(user_id)` `async` ¶

Delete every Mem0 entry owned by user_id.

Counts the user's entries with get_all and removes them with Mem0's delete_all(user_id=...); when delete_all is missing or fails, deletes the listed entries one by one. Returns 0 for SHARED-scope providers and when no entries existed.

`search(query, *, limit=5, user_id=None)` `async` ¶

Search Mem0 for memories relevant to query.

Raises:

Type	Description
`MemoryIsolationError`	`scope=PER_USER` and no `user_id`.
`Exception`	Whatever the Mem0 client raises. Errors are not turned into an empty result, so an incompatible Mem0 release or an unreachable vector store is visible.

MemoryAgent¶

`promptise.memory.MemoryAgent` ¶

Wraps an agent with automatic memory context injection.

Before every :meth:ainvoke, searches the memory provider for content relevant to the user's query and injects matching results as a SystemMessage. The agent sees relevant context without needing explicit memory tools.

If the memory provider fails (timeout, connection error), the agent continues normally without memory context — memory never blocks execution.

Composable with :class:PromptiseAgent::

MemoryAgent(PromptiseAgent(graph), provider)

Parameters:

Name	Type	Description	Default
`inner`	`Any`	The wrapped LangGraph agent (Runnable).	required
`provider`	`MemoryProvider`	A :class:`MemoryProvider` implementation.	required
`max_memories`	`int`	Maximum results to inject per invocation.	`5`
`min_score`	`float`	Minimum relevance score threshold (0.0--1.0).	`0.0`
`timeout`	`float`	Maximum seconds to wait for memory search.	`5.0`
`auto_store`	`bool`	If `True`, automatically store each exchange (user input + agent output) after invocation completes.	`False`

`ainvoke(input, config=None, **kwargs)` `async` ¶

Invoke the agent with auto-injected memory context.

`astream(input, config=None, **kwargs)` `async` ¶

Stream the agent with auto-injected memory context.

`invoke(input, config=None, **kwargs)` ¶

Synchronous invoke with memory injection.

sanitize_memory_content¶

`promptise.memory.sanitize_memory_content(content)` ¶

Sanitize memory content before injection into agent messages.

Strips known prompt-injection patterns and fence markers so that recalled memory cannot hijack the agent's instruction context. Content is also truncated to a safe injection length.

Uses case-insensitive regex matching with whitespace tolerance to catch pattern variations like "SyStEm :" or "SYSTEM:".

Note

This only affects the injected representation — the stored content in the memory provider remains unmodified.

Parameters:

Name	Type	Description	Default
`content`	`str`	Raw memory content string.	required

Returns:

Type	Description
`str`	Sanitized content safe for injection into a prompt.

Memory API Reference¶

MemoryResult¶

promptise.memory.MemoryResult dataclass ¶

MemoryProvider¶

promptise.memory.MemoryProvider ¶

add(content, *, metadata=None, user_id=None) async ¶

close() async ¶

delete(memory_id, *, user_id=None) async ¶

purge_user(user_id) async ¶

search(query, *, limit=5, user_id=None) async ¶

InMemoryProvider¶

promptise.memory.InMemoryProvider ¶

list_entries(*, user_id=None, metadata=None, limit=1000) async ¶

ChromaProvider¶

promptise.memory.ChromaProvider ¶

close() async ¶

list_entries(*, user_id=None, metadata=None, limit=1000) async ¶

purge_user(user_id) async ¶

Mem0Provider¶

promptise.memory.Mem0Provider ¶

close() async ¶

list_entries(*, user_id=None, metadata=None, limit=1000) async ¶

purge_user(user_id) async ¶

search(query, *, limit=5, user_id=None) async ¶

MemoryAgent¶

promptise.memory.MemoryAgent ¶

ainvoke(input, config=None, **kwargs) async ¶

astream(input, config=None, **kwargs) async ¶

invoke(input, config=None, **kwargs) ¶

sanitize_memory_content¶

promptise.memory.sanitize_memory_content(content) ¶

`promptise.memory.MemoryResult` `dataclass` ¶

`promptise.memory.MemoryProvider` ¶

`add(content, *, metadata=None, user_id=None)` `async` ¶

`close()` `async` ¶

`delete(memory_id, *, user_id=None)` `async` ¶

`purge_user(user_id)` `async` ¶

`search(query, *, limit=5, user_id=None)` `async` ¶

`promptise.memory.InMemoryProvider` ¶

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

`promptise.memory.ChromaProvider` ¶

`close()` `async` ¶

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

`purge_user(user_id)` `async` ¶

`promptise.memory.Mem0Provider` ¶

`close()` `async` ¶

`list_entries(*, user_id=None, metadata=None, limit=1000)` `async` ¶

`purge_user(user_id)` `async` ¶

`search(query, *, limit=5, user_id=None)` `async` ¶

`promptise.memory.MemoryAgent` ¶

`ainvoke(input, config=None, **kwargs)` `async` ¶

`astream(input, config=None, **kwargs)` `async` ¶

`invoke(input, config=None, **kwargs)` ¶

`promptise.memory.sanitize_memory_content(content)` ¶