From 0684553223e28f41a71ff300a859b9dd3de6154b Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Thu, 27 Aug 2026 22:26:07 +0800 Subject: [PATCH] fix(local_embedding_store): retry batch computation on vector space changes - Add up to 3 attempts to recompute embedding batch if vector space changes during processing - Log warnings when maximum retries reached and discard stale results - Prevent caching results from outdated vector spaces to maintain consistency - Add tests to verify retry behavior and abort after continuous vector space churn fix(daily_paper): update digest search logic and tests - Change search to query existing memory, not only previous articles in daily_dir - Allow multiple searches outside daily_dir but limit links to dated markdown in daily_dir before today - Update test assertions to reflect revised search and linking rules --- .../src/reme_daily_paper/digest.yaml | 7 +-- plugins/daily_paper/tests/test_daily_paper.py | 4 +- .../embedding_store/local_embedding_store.py | 25 ++++++++-- tests/unit/test_local_embedding_store.py | 48 ++++++++++++++++--- 4 files changed, 69 insertions(+), 15 deletions(-) diff --git a/plugins/daily_paper/src/reme_daily_paper/digest.yaml b/plugins/daily_paper/src/reme_daily_paper/digest.yaml index bb8c1a20..9bd8f1a9 100644 --- a/plugins/daily_paper/src/reme_daily_paper/digest.yaml +++ b/plugins/daily_paper/src/reme_daily_paper/digest.yaml @@ -4,9 +4,10 @@ digest_user: | 内容只能依据输入文档，不得补充文档中没有提供的事实。 保留技术准确性，同时解释三篇论文为什么值得关注，以及它们之间有什么联系。 - 在写作前，先调用 `search` 检索以前的文章：围绕三篇论文的核心问题、方法、关键词和同义表达组织查询。 - 主题跨度较大时可以多次检索。只把 `{daily_dir}/` 下日期早于今天、 - 且与本期内容确实相似或互补的 Markdown 文章作为候选；必要时调用 `read` 核验全文，不要仅凭标题判断。 + 在写作前，先调用 `search` 检索已有记忆：围绕三篇论文的核心问题、方法、关键词和同义表达组织查询。 + 主题跨度较大时可以多次检索，搜索结果不必局限于 `{daily_dir}/`。只有 `{daily_dir}/` 下日期早于今天、 + 且与本期内容确实相似或互补的 Markdown 文章才可作为正文中的历史链接候选；必要时调用 `read` 核验全文， + 不要仅凭标题判断。 将确认相关的旧文章以 Wikilink 自然织入正文，并用句子说明关联（延续、对比、补充或方法相似）； 链接必须采用带 `.md` 的完整 workspace-relative 路径，例如 `[[{daily_dir}/2026-07-01/旧文章.md|此前的相关解读]]`。不要输出裸链接、独立关系字段，也不要虚构搜索未命中的路径。 diff --git a/plugins/daily_paper/tests/test_daily_paper.py b/plugins/daily_paper/tests/test_daily_paper.py index 29ed9cb6..dcf6ecc6 100644 --- a/plugins/daily_paper/tests/test_daily_paper.py +++ b/plugins/daily_paper/tests/test_daily_paper.py @@ -643,6 +643,7 @@ def test_digest_prompt_uses_configured_daily_directory(tmp_path: Path): ) assert "`memory/`" in prompt + assert "搜索结果不必局限于 `memory/`" in prompt assert "[[memory/2026-07-01/旧文章.md" in prompt assert "[[daily/2026-07-01/" not in prompt @@ -932,7 +933,8 @@ async def test_pipeline_filters_strict_yesterday_and_writes_outputs( assert "调用 Read" not in digest_prompt assert "daily/2026-07-21" not in digest_prompt assert "长期记忆" not in digest_prompt - assert "先调用 `search` 检索以前的文章" in digest_prompt + assert "先调用 `search` 检索已有记忆" in digest_prompt + assert "搜索结果不必局限于 `daily/`" in digest_prompt assert "end_date" not in digest_prompt assert "limit=" not in digest_prompt assert "Wikilink" in digest_prompt diff --git a/reme/components/embedding_store/local_embedding_store.py b/reme/components/embedding_store/local_embedding_store.py index 81167560..795a81eb 100644 --- a/reme/components/embedding_store/local_embedding_store.py +++ b/reme/components/embedding_store/local_embedding_store.py @@ -12,6 +12,7 @@ from ..component_registry import R from ..as_embedding import BaseAsEmbedding Miss = tuple[int, str, str] # (result_index, text, cache_key) +_MAX_VECTOR_SPACE_ATTEMPTS = 3 @R.register("local") @@ -125,14 +126,30 @@ class LocalEmbeddingStore(BaseEmbeddingStore): return results, misses async def _fill_misses(self, misses: list[Miss], results: list[np.ndarray | None], **kwargs) -> None: - vector_space_id = self._cache_space size = self.max_batch_size for start in range(0, len(misses), size): batch = misses[start : start + size] - for idx, key, emb in await self._compute_batch(batch, **kwargs): - results[idx] = emb - if vector_space_id == self.vector_space_id == self._cache_space: + for attempt in range(1, _MAX_VECTOR_SPACE_ATTEMPTS + 1): + await self._sync_cache_space() + vector_space_id = self._cache_space + computed = await self._compute_batch(batch, **kwargs) + if vector_space_id != self.vector_space_id or vector_space_id != self._cache_space: + if attempt == _MAX_VECTOR_SPACE_ATTEMPTS: + self.logger.warning( + f"Embedding vector space kept changing while computing a batch; " + f"discarding {len(computed)} stale result(s) after {attempt} attempts", + ) + else: + self.logger.info( + f"Embedding vector space changed while computing a batch; " + f"discarding {len(computed)} stale result(s) and retrying " + f"({attempt}/{_MAX_VECTOR_SPACE_ATTEMPTS})", + ) + continue + for idx, key, emb in computed: + results[idx] = emb self._cache_put(key, emb) + break async def _compute_batch(self, batch: list[Miss], **kwargs) -> list[tuple[int, str, np.ndarray]]: texts = [text for _, text, _ in batch] diff --git a/tests/unit/test_local_embedding_store.py b/tests/unit/test_local_embedding_store.py index 713afaa8..f0d449d1 100644 --- a/tests/unit/test_local_embedding_store.py +++ b/tests/unit/test_local_embedding_store.py @@ -509,24 +509,58 @@ def test_cache_space_is_rechecked_after_async_load(monkeypatch, tmp_path): run(go()) -def test_completed_request_only_writes_to_its_active_cache_space(): - """A v3 request must not populate v4 after the provider switches back to v3.""" +def test_completed_request_retries_after_vector_space_changes(): + """A request completed by the old provider must not escape into the new vector space.""" async def go(): embedding = OpenAIAsEmbedding(name="t_space_write_race", backend="openai", model="v3", dimensions=2) store = LocalEmbeddingStore(name="t_local_write_race") store.as_embedding = embedding store._cache_space = embedding.vector_space_id + calls = 0 async def compute_after_round_trip(_batch, **_kwargs): - embedding.model = FakeProviderModel("v4") - store._cache_space = embedding.vector_space_id - embedding.model = FakeProviderModel("v3") - return [(0, "key", np.array([3.0, 0.0], dtype=np.float16))] + nonlocal calls + calls += 1 + if calls == 1: + embedding.model = FakeProviderModel("v4") + return [(0, "key", np.array([3.0, 0.0], dtype=np.float16))] + return [(0, "key", np.array([4.0, 0.0], dtype=np.float16))] store._compute_batch = compute_after_round_trip - await store._fill_misses([(0, "text", "key")], [None]) + results = [None] + await store._fill_misses([(0, "text", "key")], results) + assert calls == 2 + assert store._cache_space == embedding.vector_space_id + np.testing.assert_array_equal(results[0], np.array([4.0, 0.0], dtype=np.float16)) + np.testing.assert_array_equal(store._cache["key"], np.array([4.0, 0.0], dtype=np.float16)) + + run(go()) + + +def test_completed_request_stops_retrying_when_vector_space_keeps_changing(): + """Continuous configuration churn must leave the batch empty instead of blocking forever.""" + + async def go(): + embedding = OpenAIAsEmbedding(name="t_space_write_churn", backend="openai", model="v3", dimensions=2) + store = LocalEmbeddingStore(name="t_local_write_churn") + store.as_embedding = embedding + store._cache_space = embedding.vector_space_id + calls = 0 + + async def change_space_every_time(_batch, **_kwargs): + nonlocal calls + calls += 1 + embedding.model = FakeProviderModel(f"v{calls + 3}") + return [(0, "key", np.array([float(calls), 0.0], dtype=np.float16))] + + store._compute_batch = change_space_every_time + results = [None] + await store._fill_misses([(0, "text", "key")], results) + + assert calls == 3 + assert results == [None] assert "key" not in store._cache run(go())