gruenstifter/artenschutz-router
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: align router with memory-api's actual payload shape (smoke findings)

Browse source 
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()
This commit is contained in:
m2 (AI Agent) 2026-05-14 23:30:57 +02:00
parent 6da874b2e6
commit 04d2f5b9e4
4 changed files with 124 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
scripts/smoke.sh
View file

@ -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 <qdrant>: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'\"}}]}}'"

76
src/artenschutz_router/clients/bge.py
View file

@ -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}

20
src/artenschutz_router/clients/qdrant.py
View file

@ -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()),

39
tests/test_filter_builder.py
View file

@ -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"