This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new fd5debb90 [runtime][python] Handle missing long-term memory IDs (#1314)
fd5debb90 is described below
commit fd5debb9092bda1c56dd8126c64ff1d9afa4b7c2
Author: Wenjin Xie <[email protected]>
AuthorDate: Thu Oct 22 13:57:40 2026 +0800
[runtime][python] Handle missing long-term memory IDs (#1314)
Generated-by: Codex CLI 0.153.4 (GPT-6)
Co-authored-by: Codex <[email protected]>
---
.../apache/flink/agents/api/memory/MemorySet.java | 9 ++-
.../docs/development/memory/long_term_memory.md | 32 +++++++++-
python/flink_agents/api/memory/long_term_memory.py | 15 ++++-
.../vector_stores/mem0/mem0_vector_store.py | 9 ++-
.../mem0/tests/test_mem0_vector_store.py | 26 ++++++++
.../memory/mem0/flink_agents_mem0_adapters.py | 6 +-
.../runtime/memory/mem0/mem0_long_term_memory.py | 2 +
.../mem0/tests/test_mem0_long_term_memory.py | 74 ++++++++++++++++++++++
.../memory/mem0/tests/test_mem0_op_recording.py | 13 ++++
9 files changed, 175 insertions(+), 11 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/memory/MemorySet.java
b/api/src/main/java/org/apache/flink/agents/api/memory/MemorySet.java
index 0af97ae63..8e14bec17 100644
--- a/api/src/main/java/org/apache/flink/agents/api/memory/MemorySet.java
+++ b/api/src/main/java/org/apache/flink/agents/api/memory/MemorySet.java
@@ -62,7 +62,9 @@ public class MemorySet {
/**
* Retrieves memory items. When {@code ids} is provided, {@code filters}
and {@code limit} are
- * ignored.
+ * ignored. IDs directly address records in the backing collection without
checking job,
+ * partition, or set ownership. Use IDs returned by this set's {@link
#add} operation. IDs that
+ * do not exist are skipped.
*
* @param ids optional list of item IDs to retrieve
* @param filters optional metadata filters
@@ -98,6 +100,11 @@ public class MemorySet {
/**
* Deletes memory items. If {@code ids} is null, all items in the set are
deleted.
*
+ * <p>Explicit IDs directly address records in the backing collection
without checking job,
+ * partition, or set ownership. Use IDs returned by this set's {@link
#add} operation. An ID
+ * that does not exist causes a backend not-found exception. IDs are
deleted sequentially;
+ * earlier deletions are not rolled back when a later ID fails.
+ *
* @param ids optional list of item IDs to delete
*/
public void delete(@Nullable List<String> ids) throws Exception {
diff --git a/docs/content/docs/development/memory/long_term_memory.md
b/docs/content/docs/development/memory/long_term_memory.md
index 591adb7d0..f282257db 100644
--- a/docs/content/docs/development/memory/long_term_memory.md
+++ b/docs/content/docs/development/memory/long_term_memory.md
@@ -141,7 +141,7 @@ Represents a single memory item stored in Long-Term Memory:
### MemorySet
-A named collection of memory items. Memory sets provide logical grouping and
isolation of memories. See [Context Isolation](#context-isolation) for details
on how memories are scoped and isolated.
+A named collection of memory items. Memory sets provide logical grouping and
context-scoped operations. Explicit ID reads and deletes directly address
records in the backing collection. See [Context Isolation](#context-isolation)
for the scope rules and caller responsibilities.
## Operations
@@ -190,6 +190,8 @@ A memory set is scoped to the key of the action that
obtained it. Call `get_memo
`getMemorySet` inside each action that needs one, rather than caching a set
and reusing it
in a later action. Reusing a set would apply another key's operations to the
key it was
originally obtained for, and operating on a set that carries no scope raises
an error.
+Explicit ID reads and deletes do not check record ownership against that
scope; see
+[Context Isolation](#context-isolation).
Call `getMemorySet` / `get_memory_set` and `deleteMemorySet` /
`delete_memory_set` in
the Action body, outside functions submitted for async or durable execution.
Operations
@@ -254,6 +256,12 @@ ids = memorySet.add(
### Retrieving Items
+When IDs are supplied, records are read directly from the backing collection
without
+checking job, partition-key, or memory-set ownership. Use IDs returned by the
intended
+set's `add` operation; see [Context Isolation](#context-isolation). IDs that
do not
+exist are skipped, so the result is empty if none are found. Without IDs,
`get` lists
+items within the calling set's scope.
+
{{< tabs "Retrieving Items" >}}
{{< tab "Python" >}}
@@ -360,6 +368,13 @@ results = memorySet.search(
### Deleting Items
+When IDs are supplied, records are deleted directly from the backing
collection without
+checking job, partition-key, or memory-set ownership. Use IDs returned by the
intended
+set's `add` operation; see [Context Isolation](#context-isolation). A
nonexistent ID
+raises a not-found error (`ValueError` in Python, propagated through the
Python bridge
+in Java). IDs are deleted sequentially: processing stops at the first error,
and earlier
+deletions are not rolled back. Without IDs, deletion is limited to the calling
set's scope.
+
{{< tabs "Deleting Items" >}}
{{< tab "Python" >}}
@@ -565,14 +580,25 @@ public static void processEvent(Event event,
RunnerContext ctx) throws Exception
## Context Isolation
-Long-Term Memory automatically provides context isolation through Flink's
keyed partition model. Each keyed partition normally maintains its own isolated
set of memories.
+Long-Term Memory scopes additions, listing, semantic search, and whole-set
deletion
+by job identifier, partition key, and memory-set name. Explicit ID reads and
deletes
+are an exception to these scope rules.
The isolation hierarchy works as follows:
- **Job-level** (`JOB_IDENTIFIER`): Separates memories by job identifier
- **Partition-level** (keyed partition key): Separates memories between
different keys within the same job
- **Set-level** (memory set name): Separates memories between different
logical categories within the same partition
-This means you can reuse the same memory set name across different partitions,
and each partition will normally access only its own memories.
+This means you can reuse the same memory set name across different partitions
for
+context-scoped operations.
+
+When you explicitly supply IDs to `get` or `delete`, those IDs directly
address records
+in the configured backing collection. Flink Agents does not verify that the
records
+belong to the calling set's job identifier, partition key, or set name. The
caller is
+responsible for using IDs returned by the intended set's `add` operation and
keeping
+them associated with that context. Passing IDs from another context can read
or delete
+that context's records if they share the same backing collection. This
behavior is the
+same in Java and Python and avoids adding ownership-validation reads before
deletion.
> **Note:** Partition-level isolation uses a textual identity derived from the
> logical key instead of `key.hashCode()`. Java keys use
> `String.valueOf(key)`. Default-serialized PyFlink keys are deserialized and
> use Python `str`; explicitly typed PyFlink keys use `String.valueOf`, except
> byte-array keys, which use Python's bytes representation. This avoids hash
> collisions, but distinct keys may still share a memory context if they
> produce the same text, for example when custom key types ha [...]
diff --git a/python/flink_agents/api/memory/long_term_memory.py
b/python/flink_agents/api/memory/long_term_memory.py
index c3ecd8921..92f837100 100644
--- a/python/flink_agents/api/memory/long_term_memory.py
+++ b/python/flink_agents/api/memory/long_term_memory.py
@@ -118,12 +118,14 @@ class MemorySet(BaseModel):
Args:
ids: The ids of the items to retrieve. If provided, ``filters``
- and ``limit`` are ignored.
+ and ``limit`` are ignored. IDs directly address records in the
+ backing collection without checking job, partition, or set
+ ownership. Use IDs returned by this set's ``add`` operation.
filters: Optional metadata filters applied when listing items.
limit: Maximum number of items to return. Defaults to 100.
Returns:
- The memory items retrieved.
+ The memory items retrieved. IDs that do not exist are skipped.
"""
return self.ltm.get(memory_set=self, ids=ids, filters=filters,
limit=limit)
@@ -151,7 +153,14 @@ class MemorySet(BaseModel):
Args:
ids: The ids of items to be deleted. If not provided, all items
- will be deleted.
+ in this set will be deleted. Explicit IDs directly address
+ records in the backing collection without checking job,
+ partition, or set ownership. Use IDs returned by this set's
+ ``add`` operation.
+
+ Raises:
+ ValueError: If an ID does not exist. IDs are deleted sequentially;
+ earlier deletions are not rolled back when a later ID fails.
"""
self.ltm.delete(memory_set=self, ids=ids)
diff --git
a/python/flink_agents/integrations/vector_stores/mem0/mem0_vector_store.py
b/python/flink_agents/integrations/vector_stores/mem0/mem0_vector_store.py
index a5ad3dd4e..f0bde635d 100644
--- a/python/flink_agents/integrations/vector_stores/mem0/mem0_vector_store.py
+++ b/python/flink_agents/integrations/vector_stores/mem0/mem0_vector_store.py
@@ -174,7 +174,14 @@ class Mem0VectorStore(CollectionManageableVectorStore):
if ids is not None:
docs: List[Document] = []
for id_ in _maybe_cast_to_list(ids):
- output = store.get(vector_id=id_)
+ if self.provider == "chroma":
+ # Mem0's Chroma get indexes the first parsed result even
+ # when no record exists. Keep its payload conversion, but
+ # handle the empty result before indexing it.
+ outputs =
store._parse_output(store.collection.get(ids=[id_]))
+ output = outputs[0] if outputs else None
+ else:
+ output = store.get(vector_id=id_)
if output is None or getattr(output, "id", None) is None:
continue
docs.append(_mem0_output_to_document(output))
diff --git
a/python/flink_agents/integrations/vector_stores/mem0/tests/test_mem0_vector_store.py
b/python/flink_agents/integrations/vector_stores/mem0/tests/test_mem0_vector_store.py
index 27253b964..bfb9f58a4 100644
---
a/python/flink_agents/integrations/vector_stores/mem0/tests/test_mem0_vector_store.py
+++
b/python/flink_agents/integrations/vector_stores/mem0/tests/test_mem0_vector_store.py
@@ -190,6 +190,32 @@ def test_add_and_get_without_ids_returns_all(tmp_path) ->
None:
assert {d.id for d in got} == {"a", "b"}
+def test_get_skips_missing_ids(tmp_path) -> None:
+ vs = _make_store(tmp_path)
+ vs.add(documents=[_doc("found", "value")], collection_name="missing_ids")
+
+ assert vs.get(ids="missing", collection_name="missing_ids") == []
+ got = vs.get(
+ ids=["missing", "found", "missing"], collection_name="missing_ids"
+ )
+ assert [doc.id for doc in got] == ["found"]
+ assert got[0].content == "value"
+
+
+def test_get_propagates_chroma_read_errors(tmp_path, monkeypatch) -> None:
+ vs = _make_store(tmp_path)
+ vs.create_collection_if_not_exists("read_errors")
+ store = vs._stores["read_errors"]
+
+ def fail_get(*args: Any, **kwargs: Any) -> None:
+ msg = "Storage unavailable"
+ raise RuntimeError(msg)
+
+ monkeypatch.setattr(type(store.collection), "get", fail_get)
+ with pytest.raises(RuntimeError, match="Storage unavailable"):
+ vs.get(ids="id", collection_name="read_errors")
+
+
def test_get_respects_explicit_limit(tmp_path) -> None:
vs = _make_store(tmp_path)
vs.add(
diff --git
a/python/flink_agents/runtime/memory/mem0/flink_agents_mem0_adapters.py
b/python/flink_agents/runtime/memory/mem0/flink_agents_mem0_adapters.py
index f0b40f118..350666a10 100644
--- a/python/flink_agents/runtime/memory/mem0/flink_agents_mem0_adapters.py
+++ b/python/flink_agents/runtime/memory/mem0/flink_agents_mem0_adapters.py
@@ -310,11 +310,11 @@ class FlinkAgentsMem0VectorStore(VectorStoreBase):
**self._vs.store_kwargs,
)
- def get(self, vector_id: str) -> _OutputData:
- """Return the document with ``vector_id``, or an empty OutputData."""
+ def get(self, vector_id: str) -> _OutputData | None:
+ """Return the document with ``vector_id``, or None if it does not
exist."""
docs = self._vs.get(ids=[vector_id], collection_name=self._col)
if not docs:
- return _OutputData(id=None, score=None, payload=None)
+ return None
d = docs[0]
return _OutputData(
id=d.id,
diff --git a/python/flink_agents/runtime/memory/mem0/mem0_long_term_memory.py
b/python/flink_agents/runtime/memory/mem0/mem0_long_term_memory.py
index 03a14f9e3..cd8cf1832 100644
--- a/python/flink_agents/runtime/memory/mem0/mem0_long_term_memory.py
+++ b/python/flink_agents/runtime/memory/mem0/mem0_long_term_memory.py
@@ -654,6 +654,8 @@ class Mem0LongTermMemory(InternalBaseLongTermMemory):
items = []
for memory_id in ids:
result = self._mem0_instance.get(memory_id=memory_id)
+ if result is None:
+ continue
items.append(self._convert_mem0_result(memory_set.name,
result))
item = items[-1]
self._record_ltm_op(
diff --git
a/python/flink_agents/runtime/memory/mem0/tests/test_mem0_long_term_memory.py
b/python/flink_agents/runtime/memory/mem0/tests/test_mem0_long_term_memory.py
index adf79bcc1..449b7b3eb 100644
---
a/python/flink_agents/runtime/memory/mem0/tests/test_mem0_long_term_memory.py
+++
b/python/flink_agents/runtime/memory/mem0/tests/test_mem0_long_term_memory.py
@@ -24,6 +24,7 @@ import pytest
from flink_agents.api.chat_message import ChatMessage, MessageRole, TextBlock
from flink_agents.api.chat_result import ChatResult, TokenUsage
from flink_agents.api.memory.long_term_memory import (
+ MemorySet,
MemorySetItem,
)
from flink_agents.api.resource import ResourceType
@@ -308,6 +309,79 @@ def test_delete_by_id(ltm) -> None:
assert ids[0] not in deleted_ids
+def test_get_skips_missing_ids(ltm) -> None:
+ memory_set = ltm.get_memory_set(name="missing_get_set")
+ missing_id = "00000000-0000-0000-0000-000000000000"
+ ids = memory_set.add(items="A memory that still exists.")
+ assert len(ids) == 1
+
+ assert memory_set.get(ids=missing_id) == []
+ assert memory_set.get(ids=[missing_id]) == []
+ assert memory_set.get(ids=[]) == []
+ items = memory_set.get(ids=[missing_id, ids[0], missing_id])
+ assert [item.id for item in items] == ids
+ assert items[0].value == "A memory that still exists."
+
+ memory_set.delete(ids=ids)
+ assert memory_set.get(ids=ids) == []
+
+
+def test_delete_missing_id_raises_not_found(ltm) -> None:
+ memory_set = ltm.get_memory_set(name="missing_delete_set")
+ missing_id = "00000000-0000-0000-0000-000000000000"
+ with pytest.raises(ValueError, match=f"Memory with id {missing_id} not
found"):
+ memory_set.delete(ids=missing_id)
+
+ ids = memory_set.add(items="A memory to delete only once.")
+ assert len(ids) == 1
+ memory_set.delete(ids=ids)
+ with pytest.raises(ValueError, match=f"Memory with id {ids[0]} not found"):
+ memory_set.delete(ids=ids)
+
+
+def test_delete_batch_stops_at_missing_id_without_rollback(ltm) -> None:
+ memory_set = ltm.get_memory_set(name="partial_delete_set")
+ first = memory_set.add(items="The first memory is deleted.")
+ last = memory_set.add(items="The last memory is retained.")
+ assert len(first) == len(last) == 1
+ missing_id = "00000000-0000-0000-0000-000000000000"
+
+ with pytest.raises(ValueError, match=f"Memory with id {missing_id} not
found"):
+ memory_set.delete(ids=[first[0], missing_id, last[0]])
+
+ assert [item.id for item in memory_set.get()] == last
+
+
[email protected]("scope", ["job", "key", "set"])
+def test_explicit_ids_address_records_outside_memory_set(mock_ctx, scope) ->
None:
+ def get_set(job: str, key: str, name: str) -> MemorySet:
+ memory = Mem0LongTermMemory(
+ ctx=mock_ctx,
+ job_id=job,
+ chat_model_name="test_chat_model",
+ embedding_model_name="test_embedding_model",
+ vector_store_name="test_vector_store",
+ mailbox_thread_checker=lambda: None,
+ )
+ memory.switch_context(key, observation_id="id-scope-action")
+ return memory.get_memory_set(name)
+
+ name = f"explicit_ids_{scope}"
+ owner = get_set("id-job", "id-key", name)
+ caller = get_set(
+ "other-job" if scope == "job" else "id-job",
+ "other-key" if scope == "key" else "id-key",
+ "other-set" if scope == "set" else name,
+ )
+ ids = owner.add(items="The owner prefers tea.")
+ assert len(ids) == 1
+ assert caller.get() == []
+ assert [item.id for item in caller.get(ids=ids)] == ids
+
+ caller.delete(ids=ids)
+ assert owner.get() == []
+
+
def test_delete_all(ltm) -> None:
memory_set = ltm.get_memory_set(name="delete_all_set")
diff --git
a/python/flink_agents/runtime/memory/mem0/tests/test_mem0_op_recording.py
b/python/flink_agents/runtime/memory/mem0/tests/test_mem0_op_recording.py
index fa729ecc6..456fddff1 100644
--- a/python/flink_agents/runtime/memory/mem0/tests/test_mem0_op_recording.py
+++ b/python/flink_agents/runtime/memory/mem0/tests/test_mem0_op_recording.py
@@ -173,6 +173,19 @@ def test_get_delete_and_search_keep_structured_identity()
-> None:
assert records[2]["query"] == "refund"
+def test_missing_ids_do_not_produce_get_observations() -> None:
+ mem0 = MagicMock()
+ mem0.get.side_effect = [None, {"id": "found", "memory": "value"}, None]
+ ltm = _make_ltm(mem0)
+
+ items = ltm.get_memory_set("prefs").get(ids=["missing", "found",
"missing"])
+
+ assert [item.id for item in items] == ["found"]
+ assert [(record["op"], record["id"]) for record in _drain(ltm)] == [
+ ("GET", "found"),
+ ]
+
+
def test_context_switch_changes_observation_owner_and_current_suppression() ->
None:
mem0 = MagicMock()
mem0.add.return_value = {
