From f7e8ee1dd066e7ed68ba92560440cf699fca30b3 Mon Sep 17 00:00:00 2001 From: KeyCOrigin Date: Wed, 4 Nov 2026 22:26:06 +0800 Subject: [PATCH 1/2] hybrid retriever: reuse query embedding across retries --- .../capabilities/kb/providers/hybrid_retriever.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/datamind/capabilities/kb/providers/hybrid_retriever.py b/datamind/capabilities/kb/providers/hybrid_retriever.py index 66a5f4a..bf05d86 100644 --- a/datamind/capabilities/kb/providers/hybrid_retriever.py +++ b/datamind/capabilities/kb/providers/hybrid_retriever.py @@ -145,16 +145,20 @@ async def _retrieve(self, query, *, top_k, filters, version): bm25, ids, texts, metas = await self._ensure_lexical_at_version(version) k_inner = top_k * self._cm stage = "kb.embedding" + # Reuse a completed embedding across vector-search retries within this + # retrieval only. Failed embedding attempts leave the cache empty. + query_vector: list[float] | None = None async def vector_search(): - nonlocal stage - stage = "kb.embedding" - vec = await self._embed.embed_query(query) + nonlocal stage, query_vector + if query_vector is None: + stage = "kb.embedding" + query_vector = await self._embed.embed_query(query) stage = "kb.vector_query" kwargs = {"top_k": k_inner, "where": filters} if version is not None: kwargs["version"] = version - return await self._store.query(vec, **kwargs) + return await self._store.query(query_vector, **kwargs) def rank_lexical(): if bm25 is not None and ids: From f48017f407c5f3ffdd85ed1770fd4b76da2f72c7 Mon Sep 17 00:00:00 2001 From: KeyCOrigin Date: Wed, 4 Nov 2026 22:43:32 +0800 Subject: [PATCH 2/2] test: expect single embedding call when vector read retries --- datamind/tests/test_read_recovery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datamind/tests/test_read_recovery.py b/datamind/tests/test_read_recovery.py index f52d6be..8893ea1 100644 --- a/datamind/tests/test_read_recovery.py +++ b/datamind/tests/test_read_recovery.py @@ -58,7 +58,7 @@ async def test_transient_failure_falls_back_after_bounded_retries(stage): hits = await retriever.aretrieve("needle", filters={"tenant": "a"}) assert [h.id for h in hits] == ["new"] assert hits[0].metadata["degraded"] is True - assert embed.calls == 2 + assert embed.calls == (2 if stage == "embedding" else 1) summary = recovery_summary(ctx.extra["recovery_events"]) assert summary["retries"] == 1 and summary["fallbacks"] == 1 assert summary["events"][0]["component"] == f"kb.{'embedding' if stage == 'embedding' else 'vector_query'}"