From 6aaac5d55c0c69b864d8da70c928ddc6c1d1ca3e Mon Sep 17 00:00:00 2001 From: Alexander Gusak Date: Mon, 13 Jul 2026 18:07:07 +0100 Subject: [PATCH 1/2] xmemory: make the recall read mode configurable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `memory_recall` hardcoded xmemory's SINGLE_ANSWER read. Expose the mode as `xmemory.read_mode` so an instance can instead return its structured rows (`raw-tables`) or `xresponse`. Default stays `single-answer`, so recall output is unchanged for anyone who does not set it. Config values mirror the SDK's own wire values, so ReadMode resolves them directly rather than through a mapping table — a mode added to the SDK later needs no change here. Underscores are accepted as an alias, and an unknown value falls back to the default with a warning. Rendering is keyed off the mode rather than sniffed from the payload shape: only `single-answer` carries an `answer` envelope to unwrap, while the other modes return tables, rendered as JSON. Sniffing would misread a data row with an `answer` column as the synthesized answer and drop the rest of the row. The recall section label drops "synthesized answer" for a plain `[xmemory]`, since under the structured modes the payload is rows, not an answer. --- docs/config.md | 3 +- docs/memory.md | 2 +- nerve/agent/tools/handlers/memory.py | 6 +- nerve/config.py | 2 + nerve/memory/xmemory_bridge.py | 114 +++++++++++++++++++++------ tests/test_xmemory_bridge.py | 93 ++++++++++++++++++++-- 6 files changed, 185 insertions(+), 35 deletions(-) diff --git a/docs/config.md b/docs/config.md index 1ee5449..3b5f91f 100644 --- a/docs/config.md +++ b/docs/config.md @@ -240,7 +240,7 @@ Sources pull data from external services on a schedule. See [sources.md](sources When active: - The `memorize` tool **dual-writes**: memU (as always) plus an async `write_async` to xmemory. Failures on the xmemory side never fail the tool. -- `memory_recall` appends xmemory's single synthesized answer (its `SINGLE_ANSWER` read mode) to memU's N items, run concurrently so the dual lookup is one round-trip. +- `memory_recall` appends xmemory's read result to memU's N items, run concurrently so the dual lookup is one round-trip. Read behavior is controlled via `xmemory.read_mode` (defaults to `single-answer`). - The memorization **sweep** (session-close / cron) stays memU-only — it does not go through the `memorize` tool handler. | Key | Type | Default | Description | @@ -249,6 +249,7 @@ When active: | `xmemory.instance_id` | string | *(empty)* | The xmemory instance to bind. Both this and `api_key` are required to activate. | | `xmemory.api_url` | string | `https://api.xmemory.ai` | API base URL. | | `xmemory.extraction_logic` | string | `deep` | Write extraction mode: `deep` (accurate) or `fast` (high-volume). | +| `xmemory.read_mode` | string | `single-answer` | Read mode for recall: `single-answer` (synthesized natural-language answer), `raw-tables` (structured rows, JSON-rendered), or `xresponse`. | | `xmemory.timeout` | float | `60.0` | Per-request timeout in seconds. | ## Docker diff --git a/docs/memory.md b/docs/memory.md index 0ca8b46..cf8e089 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -183,7 +183,7 @@ It is inert unless both `xmemory.api_key` and `xmemory.instance_id` are configur When active: - **`memorize` dual-writes** — the fact goes to memU (file + extraction, as always) **and** is enqueued to xmemory via async `write_async`. xmemory failures are swallowed (logged) so the tool never fails on them. The result message shows `(+ xmemory)` when the xmemory enqueue succeeded. -- **`memory_recall` adds a synthesized answer** — alongside memU's N items + category breadcrumbs, recall runs xmemory's `SINGLE_ANSWER` read **concurrently** and appends a `[xmemory] synthesized answer` section. When xmemory is disabled or returns nothing, recall output is byte-for-byte the memU-only shape. +- **`memory_recall` adds xmemory read output** — alongside memU's N items + category breadcrumbs, recall runs xmemory **concurrently** and appends a `[xmemory]` section. Read behavior is controlled by `xmemory.read_mode` (`single-answer` by default, appending a synthesized natural-language answer; `raw-tables` instead appends the matching rows as JSON, and `xresponse` is also supported). When xmemory is disabled or returns nothing, recall output is byte-for-byte the memU-only shape. - **The memorization sweep stays memU-only** — session-close / cron indexing goes through the bridge directly, not the `memorize` tool handler, so it is unaffected. Implementation: `nerve/memory/xmemory_bridge.py` (`XmemoryBridge`), wired into `ToolContext.xmemory_bridge` next to `memory_bridge`. Backed by the `xmemory-ai` SDK (`AsyncXmemoryClient`). diff --git a/nerve/agent/tools/handlers/memory.py b/nerve/agent/tools/handlers/memory.py index dfdc905..d212cd5 100644 --- a/nerve/agent/tools/handlers/memory.py +++ b/nerve/agent/tools/handlers/memory.py @@ -106,7 +106,7 @@ async def memory_recall_handler(ctx: ToolContext, args: dict) -> ToolResult: logger.error("Memory recall failed: %s", e) memu_block = f"Memory recall error: {e}" - # Collect xmemory's synthesized answer (None when disabled/empty/error). + # Collect xmemory's read output (None when disabled/empty/error). xmem_answer: str | None = None if xmem_task is not None: try: @@ -121,10 +121,12 @@ async def memory_recall_handler(ctx: ToolContext, args: dict) -> ToolResult: return ToolResult.text(memu_block) # Both stores in play → label each source so the two are distinguishable. + # The label stays mode-agnostic: under raw-tables/xresponse the payload is + # structured rows, not a synthesized answer. memu_part = memu_block if memu_block is not None else "No relevant memories found." xmem_part = _clip_to_budget(xmem_answer, _MAX_XMEM_ANSWER_BYTES) return ToolResult.text( - f"[memU] {memu_part}\n\n---\n\n[xmemory] synthesized answer:\n\n{xmem_part}" + f"[memU] {memu_part}\n\n---\n\n[xmemory]\n\n{xmem_part}" ) diff --git a/nerve/config.py b/nerve/config.py index 7d5849f..a97e0d7 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -1304,6 +1304,7 @@ class XmemoryConfig: instance_id: str = "" api_url: str = "https://api.xmemory.ai" extraction_logic: str = "deep" # "deep" (default) or "fast" + read_mode: str = "single-answer" # "single-answer" | "raw-tables" | "xresponse" timeout: float = 60.0 @property @@ -1318,6 +1319,7 @@ def from_dict(cls, d: dict) -> "XmemoryConfig": instance_id=d.get("instance_id", ""), api_url=d.get("api_url", "https://api.xmemory.ai"), extraction_logic=d.get("extraction_logic", "deep"), + read_mode=d.get("read_mode", "single-answer"), timeout=float(d.get("timeout", 60.0)), ) diff --git a/nerve/memory/xmemory_bridge.py b/nerve/memory/xmemory_bridge.py index 300d0af..ee5745e 100644 --- a/nerve/memory/xmemory_bridge.py +++ b/nerve/memory/xmemory_bridge.py @@ -11,7 +11,10 @@ * ``memorize`` tool → dual-writes: memU (as today) **and** xmemory (async ``write_async``, fire-and-forget). * ``memory_recall`` tool → memU returns its N items/breadcrumbs **and** - this bridge appends xmemory's single synthesized answer to the query. + this bridge appends xmemory's read output for the query. The read mode + is configurable (``xmemory.read_mode``): a synthesized natural-language + answer by default (``single-answer``), or the structured ``raw-tables`` / + ``xresponse`` payloads. * The memorization *sweep* (session-close, cron) stays memU-only — it never goes through the ``memorize`` tool handler, so it's untouched. @@ -22,6 +25,7 @@ from __future__ import annotations +import json import logging from typing import TYPE_CHECKING, Any @@ -131,6 +135,25 @@ def _extraction_logic(self) -> Any: fast = (self._config.extraction_logic or "deep").strip().lower() == "fast" return self._ExtractionLogic.FAST if fast else self._ExtractionLogic.DEEP + def _read_mode(self) -> Any: + """Map the configured ``read_mode`` to the SDK enum. + + Config values mirror the SDK's own wire values (``single-answer``, + ``raw-tables``, ``xresponse``), so the enum resolves them directly and + any mode the SDK adds later needs no change here. Underscores are + accepted as an alias; unknown values fall back to ``single-answer``, + the configured default. + """ + mode = (self._config.read_mode or "").strip().lower().replace("_", "-") + try: + return self._ReadMode(mode) + except ValueError: + logger.warning( + "xmemory: unknown read_mode=%r, falling back to single-answer", + self._config.read_mode, + ) + return self._ReadMode.SINGLE_ANSWER + async def memorize(self, text: str) -> bool: """Async-write ``text`` to xmemory (fire-and-forget). @@ -149,40 +172,81 @@ async def memorize(self, text: str) -> bool: return False async def recall_answer(self, query: str) -> str | None: - """Query xmemory and return its single synthesized answer. + """Query xmemory and return its read output as text. + + The read mode comes from ``xmemory.read_mode``: ``single-answer`` (the + default) yields the synthesized natural-language answer, while + ``raw-tables`` and ``xresponse`` yield structured payloads, rendered + as JSON. - Uses ``SINGLE_ANSWER`` read mode — xmemory translates the question - to SQL over its knowledge graph and returns a natural-language - answer. Returns the answer string, or ``None`` when unavailable, - empty, or on any error (so recall always falls back to memU alone). + Returns ``None`` when unavailable, empty, or on any error (so recall + always falls back to memU alone). """ if not self.available or not query: return None try: - result = await self._instance.read( - query, read_mode=self._ReadMode.SINGLE_ANSWER, - ) - return _extract_answer(result) + read_mode = self._read_mode() + result = await self._instance.read(query, read_mode=read_mode) except Exception as e: logger.warning("xmemory read failed: %s", e) return None + return _extract_answer( + result, single_answer=read_mode == self._ReadMode.SINGLE_ANSWER, + ) + +def _extract_answer(result: Any, *, single_answer: bool = True) -> str | None: + """Render an SDK ReadResult as text, per the read mode it was fetched in.""" + return _render_payload( + getattr(result, "reader_result", result), single_answer=single_answer, + ) -def _extract_answer(result: Any) -> str | None: - """Pull the natural-language answer out of an SDK ReadResult. - SINGLE_ANSWER mode yields ``reader_result == {"answer": "..."}``; we - stay defensive about shape (dict, object, or bare string). +def _render_payload(payload: Any, *, single_answer: bool) -> str | None: + """Render one reader payload — a whole read's, or one sub-query's — as text. + + Only ``single-answer`` payloads carry an ``answer`` envelope. The other + modes return tables, which must not be searched for an ``answer`` key: a + data row with an ``answer`` column would otherwise be mistaken for the + synthesized answer and the rest of the row dropped. + """ + if single_answer: + payload = _unwrap_answer(payload) + return _as_text(payload) + + +def _unwrap_answer(payload: Any) -> Any: + """Pull the answer out of a ``single-answer`` payload. + + Arrives as ``{"answer": ...}``, an object with ``.answer``, or a bare + string, depending on server/SDK version. """ - reader = getattr(result, "reader_result", result) - answer: Any = None - if isinstance(reader, dict): - answer = reader.get("answer") - elif hasattr(reader, "answer"): - answer = reader.answer - elif isinstance(reader, str): - answer = reader - if answer is None: + if isinstance(payload, dict): + return payload.get("answer") + if isinstance(payload, str): + return payload + return getattr(payload, "answer", None) + + +def _as_text(value: Any) -> str | None: + """Render a value as text — JSON for structured (non-string) payloads.""" + if value is None: return None - text = str(answer).strip() - return text or None + if not isinstance(value, str): + try: + value = json.dumps( + value, ensure_ascii=False, sort_keys=True, default=_json_default, + ) + except Exception: + value = str(value) + return value.strip() or None + + +def _json_default(value: Any) -> Any: + for attr in ("model_dump", "dict"): + method = getattr(value, attr, None) + if callable(method): + return method() + if hasattr(value, "__dict__"): + return value.__dict__ + return str(value) diff --git a/tests/test_xmemory_bridge.py b/tests/test_xmemory_bridge.py index 8628c7e..9ed1a82 100644 --- a/tests/test_xmemory_bridge.py +++ b/tests/test_xmemory_bridge.py @@ -2,7 +2,7 @@ xmemory runs *alongside* memU, never replacing it: * ``memorize`` dual-writes (memU + xmemory async), and -* ``memory_recall`` appends xmemory's synthesized answer to memU's hits. +* ``memory_recall`` appends xmemory's read output to memU's hits. These tests lock in three contracts: (1) the bridge is inert unless both a token and an instance_id are configured, (2) every xmemory failure is @@ -45,6 +45,7 @@ def test_config_from_dict_defaults_and_overrides() -> None: assert c.api_key == "" and c.instance_id == "" assert c.api_url == "https://api.xmemory.ai" assert c.extraction_logic == "deep" + assert c.read_mode == "single-answer" assert c.timeout == 60.0 c2 = XmemoryConfig.from_dict({ @@ -52,10 +53,12 @@ def test_config_from_dict_defaults_and_overrides() -> None: "instance_id": "inst_1", "api_url": "https://example.test", "extraction_logic": "fast", + "read_mode": "raw-tables", "timeout": 30, }) assert c2.enabled and c2.api_url == "https://example.test" - assert c2.extraction_logic == "fast" and c2.timeout == 30.0 + assert c2.extraction_logic == "fast" and c2.read_mode == "raw-tables" + assert c2.timeout == 30.0 def test_nerveconfig_wires_xmemory_block() -> None: @@ -78,6 +81,30 @@ def test_extract_answer_shapes() -> None: assert _extract_answer(SimpleNamespace(reader_result=SimpleNamespace(answer="x"))) == "x" +def test_extract_answer_does_not_mine_answer_keys_from_table_rows() -> None: + """Under raw-tables the payload is data rows. A row with an ``answer`` + column must not be mistaken for the synthesized answer.""" + rows = [ + {"question": "What is our refund window?", "answer": "30 days"}, + {"question": "What is our SLA?", "answer": "99.9%"}, + ] + text = _extract_answer( + SimpleNamespace(reader_result=rows, reader_results=[]), single_answer=False, + ) + assert text is not None + # Whole rows survive — questions included, not just the 'answer' cells. + assert "refund window" in text and "30 days" in text + + +def test_extract_answer_preserves_non_ascii() -> None: + """JSON rendering must not escape non-ASCII into \\uXXXX noise.""" + text = _extract_answer( + SimpleNamespace(reader_result=[{"city": "Zürich"}], reader_results=[]), + single_answer=False, + ) + assert text is not None and "Zürich" in text + + # --------------------------------------------------------------------------- # # Bridge — disabled / missing-package paths # --------------------------------------------------------------------------- # @@ -107,10 +134,18 @@ async def test_bridge_disabled_when_package_missing(monkeypatch) -> None: # --------------------------------------------------------------------------- # -async def _enabled_bridge(extraction_logic: str = "deep") -> XmemoryBridge: +async def _enabled_bridge( + extraction_logic: str = "deep", + read_mode: str = "single-answer", # mirrors the production default +) -> XmemoryBridge: """Build a bridge bound to a (fake-token) real client, then mock the instance handle so reads/writes never hit the network.""" - cfg = XmemoryConfig(api_key="tok", instance_id="inst_1", extraction_logic=extraction_logic) + cfg = XmemoryConfig( + api_key="tok", + instance_id="inst_1", + extraction_logic=extraction_logic, + read_mode=read_mode, + ) bridge = XmemoryBridge(cfg) await bridge.initialize() # client + .instance() are network-free assert bridge.available @@ -120,7 +155,7 @@ async def _enabled_bridge(extraction_logic: str = "deep") -> XmemoryBridge: @pytest.mark.asyncio async def test_recall_answer_returns_single_answer() -> None: - bridge = await _enabled_bridge() + bridge = await _enabled_bridge(read_mode="single-answer") bridge._instance.read = AsyncMock( return_value=SimpleNamespace(reader_result={"answer": "alice@acme.com"}) ) @@ -132,6 +167,52 @@ async def test_recall_answer_returns_single_answer() -> None: await bridge.aclose() +@pytest.mark.asyncio +async def test_recall_defaults_to_single_answer() -> None: + """An unconfigured ``read_mode`` synthesizes an answer, not raw rows.""" + bridge = XmemoryBridge(XmemoryConfig(api_key="tok", instance_id="inst_1")) + await bridge.initialize() + bridge._instance = AsyncMock() + bridge._instance.read = AsyncMock( + return_value=SimpleNamespace(reader_result={"answer": "HQ is in Berlin."}) + ) + assert await bridge.recall_answer("Where is HQ?") == "HQ is in Berlin." + _, kwargs = bridge._instance.read.call_args + assert kwargs["read_mode"] == bridge._ReadMode.SINGLE_ANSWER + await bridge.aclose() + + +@pytest.mark.asyncio +async def test_recall_answer_honors_raw_tables_read_mode() -> None: + bridge = await _enabled_bridge(read_mode="raw-tables") + bridge._instance.read = AsyncMock( + return_value=SimpleNamespace(reader_result={"rows": [{"k": "v"}]}) + ) + ans = await bridge.recall_answer("Show rows") + assert ans is not None + assert '"rows"' in ans + _, kwargs = bridge._instance.read.call_args + assert kwargs["read_mode"] == bridge._ReadMode.RAW_TABLES + await bridge.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("configured", "expected_attr"), + [ + ("raw_tables", "RAW_TABLES"), # underscore alias still accepted + ("RAW-TABLES", "RAW_TABLES"), # case-insensitive + (" xresponse ", "XRESPONSE"), # surrounding whitespace tolerated + ("nonsense", "SINGLE_ANSWER"), # unknown → default + ("", "SINGLE_ANSWER"), # empty → default + ], +) +async def test_read_mode_resolution(configured: str, expected_attr: str) -> None: + bridge = await _enabled_bridge(read_mode=configured) + assert bridge._read_mode() == getattr(bridge._ReadMode, expected_attr) + await bridge.aclose() + + @pytest.mark.asyncio async def test_recall_answer_isolates_errors() -> None: bridge = await _enabled_bridge() @@ -206,7 +287,7 @@ async def test_recall_handler_combines_memu_and_xmemory() -> None: assert "[memU]" in text assert "Alice lives in Metropolis" in text - assert "[xmemory] synthesized answer" in text + assert "[xmemory]" in text assert "alice@acme.com" in text xmem.recall_answer.assert_awaited_once_with("alice") From 1cca2d2184601e7f58c819cab2ed1a5048abfa2d Mon Sep 17 00:00:00 2001 From: Alexander Gusak Date: Mon, 13 Jul 2026 18:07:58 +0100 Subject: [PATCH 2/2] xmemory: surface per-sub-query answers from reader_results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit xmemory-ai 0.10.0 decomposes a composite query ("who leads sales, and where is HQ?") into sub-queries and answers each, returning them as `reader_results` alongside the combined `reader_result`. Render each sub-answer under the sub-question it answers, so recall hands the agent an unambiguous mapping rather than a flat run of sentences. Requires the bump to xmemory-ai >=0.10.0; `reader_results` does not exist before it. Everything else here works on the old pin, hence the separate commit. A sub-query the server could not answer carries an `error` while its siblings still succeed. Report it as "(unavailable: ...)": an explicit "not known" is worth more to the agent than a silently missing sub-query, and it stops a partial failure from looking like a complete answer. Fall back to the combined `reader_result` when the query was not decomposed, when the server predates decomposition (the list is empty), or when no sub-query yielded anything usable — otherwise a read where every sub-query failed would discard an answer the server did produce. --- nerve/memory/xmemory_bridge.py | 28 ++++++++++++++++- pyproject.toml | 2 +- tests/test_xmemory_bridge.py | 55 ++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 2 deletions(-) diff --git a/nerve/memory/xmemory_bridge.py b/nerve/memory/xmemory_bridge.py index ee5745e..aa010b4 100644 --- a/nerve/memory/xmemory_bridge.py +++ b/nerve/memory/xmemory_bridge.py @@ -196,7 +196,33 @@ async def recall_answer(self, query: str) -> str | None: def _extract_answer(result: Any, *, single_answer: bool = True) -> str | None: - """Render an SDK ReadResult as text, per the read mode it was fetched in.""" + """Render an SDK ReadResult as text. + + Prefers ``reader_results`` — the per-sub-query answers the server returns + when it decomposes a composite query (xmemory-ai 0.10.0+) — so each answer + stays labelled with the sub-question it answers. Falls back to the combined + ``reader_result`` when the query was not decomposed, when the server + predates decomposition, or when no sub-query yielded anything usable. + """ + parts: list[str] = [] + for tagged in getattr(result, "reader_results", None) or (): + body = _render_payload( + getattr(tagged, "reader_result", None), single_answer=single_answer, + ) + if body is None: + # ``error`` is set only when this one sub-query could not be + # answered while the others still were. Surface it: an explicit + # "not known" beats dropping the sub-query silently. + error = _as_text(getattr(tagged, "error", None)) + body = f"(unavailable: {error})" if error else None + if body is None: + continue + sub_query = _as_text(getattr(tagged, "sub_query", None)) + parts.append(f"{sub_query}: {body}" if sub_query else body) + + if parts: + return "\n".join(parts) + return _render_payload( getattr(result, "reader_result", result), single_answer=single_answer, ) diff --git a/pyproject.toml b/pyproject.toml index 9dc6432..f139759 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "html2text>=2024.2.26", # Optional xmemory.ai structured-memory layer — only active when an # xmemory api_key + instance_id are configured (see XmemoryConfig). - "xmemory-ai>=0.5.3", + "xmemory-ai>=0.10.0", # Optional observability stack — only active when langfuse keys are # configured. All three are pure-Python wheels. "langfuse>=3.0.0", diff --git a/tests/test_xmemory_bridge.py b/tests/test_xmemory_bridge.py index 9ed1a82..8d17c54 100644 --- a/tests/test_xmemory_bridge.py +++ b/tests/test_xmemory_bridge.py @@ -81,6 +81,61 @@ def test_extract_answer_shapes() -> None: assert _extract_answer(SimpleNamespace(reader_result=SimpleNamespace(answer="x"))) == "x" +def test_extract_answer_labels_each_sub_query() -> None: + """Decomposed answers keep the sub-question they answer — otherwise the + agent cannot tell which answer belongs to which part of the query.""" + assert _extract_answer( + SimpleNamespace( + reader_result="Who leads sales? -> Ann\nWhere is HQ? -> Berlin", + reader_results=[ + SimpleNamespace( + sub_query="Who leads sales?", + reader_result={"answer": "Ann"}, + error=None, + ), + SimpleNamespace( + sub_query="Where is HQ?", + reader_result={"answer": "Berlin"}, + error=None, + ), + ], + ) + ) == "Who leads sales?: Ann\nWhere is HQ?: Berlin" + + +def test_extract_answer_surfaces_sub_query_error() -> None: + """A sub-query the server could not answer is reported, not dropped.""" + assert _extract_answer( + SimpleNamespace( + reader_result="Ann leads sales.", + reader_results=[ + SimpleNamespace( + sub_query="Who leads sales?", + reader_result={"answer": "Ann"}, + error=None, + ), + SimpleNamespace( + sub_query="What is churn?", reader_result="", error="no data", + ), + ], + ) + ) == "Who leads sales?: Ann\nWhat is churn?: (unavailable: no data)" + + +def test_extract_answer_falls_back_when_no_sub_query_answers() -> None: + """Every sub-query empty and error-free → fall back to the combined + ``reader_result`` rather than discarding a real answer.""" + assert _extract_answer( + SimpleNamespace( + reader_result="Partial: HQ is Berlin.", + reader_results=[ + SimpleNamespace(sub_query="q1", reader_result="", error=None), + SimpleNamespace(sub_query="q2", reader_result=None, error=None), + ], + ) + ) == "Partial: HQ is Berlin." + + def test_extract_answer_does_not_mine_answer_keys_from_table_rows() -> None: """Under raw-tables the payload is data rows. A row with an ``answer`` column must not be mistaken for the synthesized answer."""