From 04d2f5b9e44c8511c13270a92a5f5511ea84c6d4 Mon Sep 17 00:00:00 2001 From: "m2 (AI Agent)" Date: Thu, 14 May 2026 23:30:57 +0200 Subject: [PATCH] fix: align router with memory-api's actual payload shape (smoke findings) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end smoke against the live coolify memory stack surfaced three gaps between the spec and what memory-api actually writes: 1. memory-api wraps the /ingest payload under `metadata.*` in Qdrant — our domain fields (fact_type, scope.states, …) don't live at the top of the payload, they're nested. `build_qdrant_filter` now auto-prefixes any non-top-level key with `metadata.`. Callers can write `{"fact_type": "legal"}` and it just works; top-level fields (`agent_id`, `memory_type`, `source`, …) pass through unprefixed. 2. The deployed BGE-M3 TEI image doesn't serve /embed_sparse (returns 424 Failed Dependency). Stored points have `has_sparse=false`. `BGEClient.embed_sparse` now returns `{}` on 4xx and logs a warning; `QdrantReader.hybrid_query` falls back to a dense-only query when sparse is empty. RRF fusion is skipped in that path. 3. `scripts/smoke.sh` had a missing `shift` after reading the limit positional arg — caused arg leakage onto the CLI (`--limit 5 5`). The cleanup hint also referenced a stale agent_id; now reads from $ARTENSCHUTZ_AGENT_ID. Smoke is now green end-to-end: 18 records ingested across bnatschg / mhbasp / biotopwertliste(BY) / state-berlin under agent_id=gruenstifter_smoketest, hybrid + payload-filter searches all return hits. 24/24 unit tests pass. Companion fixes pushed to github.com/machine-machine/agent.memory.system: d0e9529 fix: /memory/store passed wrong kwarg name to AgentMemory ebc896c fix: don't close caller-supplied clients in AgentMemory.close() --- scripts/smoke.sh | 11 ++-- src/artenschutz_router/clients/bge.py | 76 +++++++++++++++++++++--- src/artenschutz_router/clients/qdrant.py | 20 ++++++- tests/test_filter_builder.py | 39 +++++++++--- 4 files changed, 124 insertions(+), 22 deletions(-) diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 7ddd11e..85b9961 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -17,8 +17,8 @@ pass "router can reach Qdrant + BGE-M3 + memory-api" ingest_source() { local source="$1" ; shift - local limit="${1:-5}" - step "ingest --source $source --limit $limit" + local limit="${1:-5}" ; shift || true + step "ingest --source $source --limit $limit ${*:-}" python -m artenschutz_router.ingest.cli \ --source "$source" \ --router-url "$ROUTER_URL" \ @@ -66,7 +66,8 @@ pass "payload filtering on nested keys works" printf '\n\033[1;32mALL SMOKE STEPS PASSED\033[0m\n' echo -echo "To clean up the staging data later:" -echo " curl -X POST :6333/collections/agent_memory/points/delete \\" +agent_id_used="${ARTENSCHUTZ_AGENT_ID:-gruenstifter}" +echo "To clean up the smoke data later (agent_id=$agent_id_used):" +echo " curl -X POST http://memory-qdrant:6333/collections/agent_memory/points/delete \\" echo " -H 'Content-Type: application/json' \\" -echo " -d '{\"filter\":{\"must\":[{\"key\":\"agent_id\",\"match\":{\"value\":\"gruenstifter_staging\"}}]}}'" +echo " -d '{\"filter\":{\"must\":[{\"key\":\"agent_id\",\"match\":{\"value\":\"'$agent_id_used'\"}}]}}'" diff --git a/src/artenschutz_router/clients/bge.py b/src/artenschutz_router/clients/bge.py index e04e3d5..6fca50b 100644 --- a/src/artenschutz_router/clients/bge.py +++ b/src/artenschutz_router/clients/bge.py @@ -33,11 +33,22 @@ class BGEClient: return data # type: ignore[return-value] async def embed_sparse(self, text: str) -> dict[int, float]: - resp = await self._client.post("/embed_sparse", json={"inputs": text}) - resp.raise_for_status() + """Returns sparse term weights, or {} if the TEI server doesn't expose + /embed_sparse (HTTP 4xx). The router uses {} as a signal to fall back + to dense-only retrieval. + """ + try: + resp = await self._client.post("/embed_sparse", json={"inputs": text}) + resp.raise_for_status() + except httpx.HTTPStatusError as exc: + if 400 <= exc.response.status_code < 500: + logger.warning( + "BGE TEI /embed_sparse unavailable (%s) — falling back to dense-only", + exc.response.status_code, + ) + return {} + raise data = resp.json() - # TEI returns either {"sparse": [{"index": int, "value": float}, ...]} - # or directly a list of those, depending on version. sparse_list = data[0] if isinstance(data, list) and data and isinstance(data[0], list) else data out: dict[int, float] = {} for item in sparse_list: @@ -46,17 +57,67 @@ class BGEClient: return out async def embed_hybrid(self, text: str) -> tuple[list[float], dict[int, float]]: + """Returns (dense, sparse). sparse may be empty when TEI doesn't + serve /embed_sparse — callers must handle that as dense-only.""" dense = await self.embed_dense(text) sparse = await self.embed_sparse(text) return dense, sparse +# Top-level Qdrant payload keys written by memory-api. Anything outside this +# set is assumed to be a domain-payload field and gets auto-prefixed with +# `metadata.` so callers can write `fact_type` instead of `metadata.fact_type`. +_TOP_LEVEL_PAYLOAD_KEYS = { + "id", + "content", + "memory_type", + "agent_id", + "session_id", + "user_id", + "importance", + "initial_importance", + "source", + "entities", + "language", + "metadata", + "timestamp", + "last_retrieved", + "last_utilized", + "last_boosted", + "retrieval_count", + "utilization_count", + "outcome_count", + "has_dense", + "has_sparse", + "has_colbert", + "consolidated", +} + + +def _normalise_key(key: str) -> str: + """Auto-prefix domain payload fields with `metadata.` so the public API + can pretend memory-api's wrapping doesn't exist. + + Rules: + - keys that already start with `metadata.` are passed through verbatim + - top-level memory-api keys (agent_id, memory_type, …) are passed through + - everything else gets the `metadata.` prefix prepended + """ + if key.startswith("metadata."): + return key + head = key.split(".", 1)[0] + if head in _TOP_LEVEL_PAYLOAD_KEYS: + return key + return f"metadata.{key}" + + def build_qdrant_filter(metadata_filter: dict[str, Any] | None) -> dict[str, Any] | None: """Translate a flat `{key: value | [values]}` dict into a Qdrant `must` filter. None values are skipped. List values become `match: {any: [...]}` clauses. Scalar values become `match: {value: x}` clauses. - Supports nested keys via dotted form, e.g. `scope.states`. + Nested keys (`scope.states`) are supported. Domain-payload keys are + auto-prefixed with `metadata.` — see `_normalise_key`. """ if not metadata_filter: return None @@ -65,10 +126,11 @@ def build_qdrant_filter(metadata_filter: dict[str, Any] | None) -> dict[str, Any for key, value in metadata_filter.items(): if value is None: continue + normalised = _normalise_key(key) if isinstance(value, list): - must.append({"key": key, "match": {"any": value}}) + must.append({"key": normalised, "match": {"any": value}}) else: - must.append({"key": key, "match": {"value": value}}) + must.append({"key": normalised, "match": {"value": value}}) if not must: return None return {"must": must} diff --git a/src/artenschutz_router/clients/qdrant.py b/src/artenschutz_router/clients/qdrant.py index 7ded094..7e4e6b3 100644 --- a/src/artenschutz_router/clients/qdrant.py +++ b/src/artenschutz_router/clients/qdrant.py @@ -54,13 +54,31 @@ class QdrantReader: limit: int = 10, prefetch_limit: int = 50, ) -> list[dict[str, Any]]: - """Dense + sparse hybrid with RRF fusion, scoped to agent_id.""" + """Dense + sparse hybrid with RRF fusion, scoped to agent_id. + + When `sparse` is empty (TEI doesn't serve /embed_sparse), falls back + to dense-only — sends a single dense query with no fusion. This + matches the shape of points stored by the current memory-api, which + only writes dense vectors. + """ combined: dict[str, Any] = {"must": [{"key": "agent_id", "match": {"value": agent_id}}]} if metadata_filter and "must" in metadata_filter: combined["must"].extend(metadata_filter["must"]) qfilter = self._to_filter(combined) + if not sparse: + result = await self._client.query_points( + collection_name=self._collection, + query=dense, + using="dense", + query_filter=qfilter, + limit=limit, + with_payload=True, + with_vectors=False, + ) + return [self._format_point(p) for p in result.points] + sparse_vec = qm.SparseVector( indices=list(sparse.keys()), values=list(sparse.values()), diff --git a/tests/test_filter_builder.py b/tests/test_filter_builder.py index 82f8614..90f5155 100644 --- a/tests/test_filter_builder.py +++ b/tests/test_filter_builder.py @@ -8,23 +8,44 @@ def test_returns_none_for_empty(): assert build_qdrant_filter({}) is None -def test_scalar_match_value(): - f = build_qdrant_filter({"kind": "fact"}) - assert f == {"must": [{"key": "kind", "match": {"value": "fact"}}]} +def test_domain_keys_get_metadata_prefix(): + """`fact_type` is a domain-payload field — memory-api wraps it under + `metadata.*` in Qdrant.""" + f = build_qdrant_filter({"fact_type": "legal"}) + assert f == {"must": [{"key": "metadata.fact_type", "match": {"value": "legal"}}]} -def test_list_match_any(): +def test_list_match_any_with_prefix(): f = build_qdrant_filter({"fact_type": ["legal", "mitigation"]}) - assert f == {"must": [{"key": "fact_type", "match": {"any": ["legal", "mitigation"]}}]} + assert f == { + "must": [ + {"key": "metadata.fact_type", "match": {"any": ["legal", "mitigation"]}} + ] + } def test_drops_none_values(): - f = build_qdrant_filter({"a": "x", "b": None, "c": [1, 2]}) + f = build_qdrant_filter({"fact_type": "legal", "skip": None, "states": ["BE"]}) must = f["must"] keys = {clause["key"] for clause in must} - assert keys == {"a", "c"} + assert keys == {"metadata.fact_type", "metadata.states"} -def test_nested_dotted_keys_pass_through(): +def test_top_level_keys_pass_through_unprefixed(): + """`agent_id`, `memory_type`, `source` etc. live at the top of the + Qdrant payload and must NOT be prefixed.""" + f = build_qdrant_filter({"agent_id": "x", "memory_type": "fact", "source": "document"}) + keys = {c["key"] for c in f["must"]} + assert keys == {"agent_id", "memory_type", "source"} + + +def test_nested_domain_keys_get_prefix(): + """`scope.states` is a domain key — it's nested INSIDE the metadata wrapper.""" f = build_qdrant_filter({"scope.states": ["BE", "BB"]}) - assert f["must"][0]["key"] == "scope.states" + assert f["must"][0]["key"] == "metadata.scope.states" + + +def test_already_prefixed_pass_through(): + """If the caller already wrote `metadata.foo`, don't double-prefix.""" + f = build_qdrant_filter({"metadata.fact_type": "legal"}) + assert f["must"][0]["key"] == "metadata.fact_type"