diff --git a/CLAUDE.md b/CLAUDE.md index 29378d2..496cf03 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,7 +8,8 @@ starting work; it captures the non-obvious decisions and gotchas that aren't vis **Everything runs on Upstash Redis. No vector database, ever.** All "semantic"/similarity features use **Upstash Redis Search** with the **`$smart` fuzzy operator** — via `@upstash/redis`'s `redis.search` API, **not** RediSearch `FT.*` and **not** `@upstash/vector`. Matching is lexical/fuzzy (BM25), not -embeddings — keep that in mind when naming/among scoring. +embeddings — keep that in mind when naming/among scoring. The one exception is `upstash-hermes-memory`, whose memory +`text` is a Redis Search `HYBRID` field with managed embeddings (still no separate vector database). ## Packages diff --git a/python/hermes-memory/CHANGELOG.md b/python/hermes-memory/CHANGELOG.md index 0c4556b..e5fc847 100644 --- a/python/hermes-memory/CHANGELOG.md +++ b/python/hermes-memory/CHANGELOG.md @@ -4,6 +4,22 @@ Written by hand: this package is not versioned by Changesets. To release, bump ` `pyproject.toml`, run `node scripts/sync-version.mjs` and `uv lock`, and add a `## ` section here. The release workflow publishes any version PyPI does not have yet. +## 0.3.0 + +### Minor Changes + +- **Breaking:** recall is hybrid only. The memory `text` field is a Redis Search `HYBRID` field and + recall merges keyword (`$smart`) and semantic (`$ann`) results with `$rrf`. An index from 0.2.x + (`text` as `TEXT`) is refused with an error: stop every client and run `SEARCH.DROP `. + Documents are kept and indexed again on the next start. Stores with their own `prefix` change too. +- `AgentMemory.recall` takes `mode="auto"|"keyword"`; records carry `retrieval`. A failed hybrid + query falls back to keyword once. `min_score` applies to keyword results only. +- Automatic recall injects 3 memories instead of 5. +- `upstash_memory_search` results include `retrieval`. +- Fix: an empty result from a retriever query no longer sends `SEARCH.COUNT` with that query + (Search rejects `$ann` in COUNT); the missing-index check counts `{}` instead. +- Docs: the shared `agentkit` index is Python-only; the TS SDK creates its own indexes. + ## 0.2.2 ### Patch Changes diff --git a/python/hermes-memory/README.md b/python/hermes-memory/README.md index aa599f4..41b6bf9 100644 --- a/python/hermes-memory/README.md +++ b/python/hermes-memory/README.md @@ -7,6 +7,7 @@ your AI SDK app saves is recalled by your Hermes agent, and the other way round. ## What it does in Hermes - **Recall:** before each turn, the memories closest to the user's message are added to the prompt. + By keyword and by meaning; see [Semantic recall](#semantic-recall). - **Chat history:** after each turn, the turn is appended in place to the session's searchable transcript document, so a write stays the size of one turn however long the session runs. - **Mirroring:** facts the built-in `memory` tool adds are also saved to Redis, and when Hermes @@ -40,6 +41,31 @@ these): | `UPSTASH_HERMES_NAMESPACE` | Key namespace. Default `agentkit`, so memories sit at the TS SDK's default keys and are shared with AgentKit apps on the same database. | | `UPSTASH_HERMES_TELEMETRY` | Set to `1` to send telemetry. Off by default; see [Telemetry](#telemetry). | +## Semantic recall + +Keyword recall (`$smart`: phrase, term and fuzzy matching) only finds a memory that shares words +with the message: "where is the assistant hosted" misses "Hermes runs on the ops Box". So the +memory `text` field is a `HYBRID` field, which Upstash Redis Search embeds itself (no separate +vector database), and recall merges a keyword query with a semantic `$ann` query using `$rrf` +([Hybrid Search](https://upstash.com/docs/redis/search/hybrid-search)). + +- **Upgrading from 0.2.x:** the old index has `text` as `TEXT` and is not supported. Recall fails + until you replace it: stop every client of the namespace (Hermes gateway, desktop app, cron jobs, + other profiles on it) and run `SEARCH.DROP agentkit` (or your namespace's index). Your memories + and chats are kept, and the next start creates the new index and indexes them again. +- Automatic recall injects up to **3** memories: the semantic side has no relevance + threshold and returns up to its limit even for unrelated text, and in a small-store test its + similarity scores did not separate related from unrelated memories well enough for a cutoff. `upstash_memory_search` keeps its limit. +- Results carry `retrieval: "hybrid"` or `"keyword"`. Hybrid scores are rank-fusion values that + only order one search's results; `min_score` applies to keyword results only. +- If a hybrid query fails, recall falls back to keyword search for that call (logged at most + every 10 minutes). +- Chat history stays keyword-only: a `HYBRID` field gets one embedding for its whole value, and a + chat document holds a whole session. +- Every memory in the index is embedded, whichever user or profile wrote it, and each recall + sends the message to Upstash Redis Search to be embedded. In a small-store test a hybrid query + took about 150 to 300 ms, against a few ms for keyword recall. + ## Telemetry Off by default. Set `UPSTASH_HERMES_TELEMETRY=1` to opt in: requests to your Upstash database then diff --git a/python/hermes-memory/plugin.yaml b/python/hermes-memory/plugin.yaml index e4512ac..b1c0f05 100644 --- a/python/hermes-memory/plugin.yaml +++ b/python/hermes-memory/plugin.yaml @@ -3,7 +3,7 @@ # https://hermes-agent.nousresearch.com/docs/developer-guide/plugins/catalog-submission name: upstash # Copied from pyproject.toml by scripts/sync-version.mjs. -version: "0.2.2" +version: "0.3.0" description: "Upstash Redis memory provider: long-term memory, searchable chat history and recall on Redis Search, sharing keys with @upstash/agentkit-sdk." # Also declared in the catalog entry; older Hermes releases only enforce this one (catalog rule 14). requires_hermes: ">=0.19" diff --git a/python/hermes-memory/pyproject.toml b/python/hermes-memory/pyproject.toml index bfc2537..3f6f25e 100644 --- a/python/hermes-memory/pyproject.toml +++ b/python/hermes-memory/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "upstash-hermes-memory" -version = "0.2.2" +version = "0.3.0" description = "Upstash Redis memory provider for Hermes Agent: long-term memory, searchable chat history and recall on Redis Search, with no separate vector database." readme = "README.md" license = "MIT" diff --git a/python/hermes-memory/src/upstash_hermes_memory/_search.py b/python/hermes-memory/src/upstash_hermes_memory/_search.py index 2c21feb..0b7442d 100644 --- a/python/hermes-memory/src/upstash_hermes_memory/_search.py +++ b/python/hermes-memory/src/upstash_hermes_memory/_search.py @@ -1,7 +1,9 @@ """A Redis Search index that is created on first use, shared by AgentMemory and ChatHistory. -The commands are the ones `ReactiveSearchIndex` sends from the TS SDK, argument for argument, so -an index created from either language is the same index. +A store with its own `prefix` sends the commands `ReactiveSearchIndex` sends from the TS SDK, +argument for argument, except that memory `text` is HYBRID here (the TS SDK still creates it as +TEXT), so the two languages cannot share an index over the same name. The default shared index is +Python-only. """ import json @@ -25,13 +27,18 @@ def dumps(value: Any) -> str: DEFAULT_NAMESPACE = "agentkit" -# The one index AgentMemory and ChatHistory share by default (the TS `SharedIndexSchema`, field -# for field and in the same order). Upstash's Free plan allows one search index per database. -# `kind` ("memory" or "chat") keeps the two apart in every query. +# Retrievers ($rrf, $ann) must be the only key of the query object and cannot be counted. +RETRIEVERS = frozenset({"$rrf", "$ann"}) + +# The one index AgentMemory and ChatHistory share by default (Python only: the TS SDK gives each +# store its own index). Upstash's Free plan allows one search index per database. +# `kind` ("memory" or "chat") keeps the two apart in every query. Only the memory `text` is HYBRID +# (full-text plus an embedding for `$ann`): chat transcripts stay TEXT, since a HYBRID field gets +# one embedding for its whole value and `userMessages` / `modelMessages` hold a whole session each. SHARED_SCHEMA: list[Field] = [ ("kind", "TEXT", True), ("userId", "TEXT", True), - ("text", "TEXT", False), + ("text", "HYBRID", False), ("sessionId", "TEXT", True), ("userMessages", "TEXT", False), ("modelMessages", "TEXT", False), @@ -42,7 +49,7 @@ def shared_index(redis: Any, namespace: str) -> "SearchIndex": """The shared index of a namespace, covering `:memory:` and `:chat:`. Both stores build this same definition, so whichever reads first creates it and the other's - EXISTSOK create is a no-op; the TS SDK builds it identically. + EXISTSOK create is a no-op. """ return SearchIndex( redis, @@ -52,6 +59,10 @@ def shared_index(redis: Any, namespace: str) -> "SearchIndex": ) +def _is_config_mismatch(error: Exception) -> bool: + return "different configuration" in str(error) + + class SearchIndex: def __init__( self, redis: Any, *, name: str, prefix: str | Sequence[str], schema: Sequence[Field] @@ -72,20 +83,38 @@ def create_command(self) -> list[str]: return cmd def ensure(self) -> None: - """Create the index if it does not exist yet. A no-op after the first call.""" + """Create the index if it does not exist yet. A no-op after the first call. + + An index of this name with another definition (one from 0.2.x has `text` as TEXT) is never + dropped here, since other clients may be using it; the error says how to replace it. + """ if self._ready: return with self._lock: if self._ready: return - self.redis.execute(self.create_command()) - self.redis.execute(["SEARCH.WAITINDEXING", self.name]) + try: + self.redis.execute(self.create_command()) + except Exception as error: + if not _is_config_mismatch(error): + raise + raise RuntimeError( + f"search index {self.name} exists with a different definition, for example " + f"from upstash-hermes-memory 0.2.x. Stop every client, run `SEARCH.DROP " + f"{self.name}` (documents are kept and indexed again) and start again." + ) from error + # 1 once pending updates are searchable; 0 means the index is gone (dropped right + # after the create): stay not-ready, so the next call creates it again. + if self.redis.execute(["SEARCH.WAITINDEXING", self.name]) != 1: + raise RuntimeError(f"search index {self.name} disappeared while it was created") self._ready = True def query(self, filter: dict[str, Any], limit: int | None = None) -> list[Any]: self.ensure() rows = self._query(filter, limit) - if not rows and self._raw_count(filter) < 0: + # COUNT cannot take a retriever, so a retriever query checks the index with `{}` instead. + probe = {} if RETRIEVERS & set(filter) else filter + if not rows and self._raw_count(probe) < 0: # The Python client turns a missing index into [], so an empty result is ambiguous. # COUNT says -1 when the index is gone (another client dropped it): recreate, retry once. self._ready = False @@ -114,9 +143,15 @@ def _raw_count(self, filter: dict[str, Any]) -> int: return int(getattr(result, "count", result)) def wait_indexing(self) -> None: - """Block until every write so far is searchable (and every delete has left the index).""" + """Block until every write so far is searchable (and every delete has left the index). + + WAITINDEXING answers 0 when the index is gone (another client dropped it): it is created + again (its documents are indexed by the create-time scan) and waited for once more. + """ self.ensure() - self.redis.execute(["SEARCH.WAITINDEXING", self.name]) + if self.redis.execute(["SEARCH.WAITINDEXING", self.name]) != 1: + self._ready = False + self.ensure() def drop(self) -> None: self.redis.execute(["SEARCH.DROP", self.name]) diff --git a/python/hermes-memory/src/upstash_hermes_memory/_version.py b/python/hermes-memory/src/upstash_hermes_memory/_version.py index 5228795..713288d 100644 --- a/python/hermes-memory/src/upstash_hermes_memory/_version.py +++ b/python/hermes-memory/src/upstash_hermes_memory/_version.py @@ -1,2 +1,2 @@ # Generated by scripts/sync-version.mjs from pyproject.toml — do not edit by hand. -VERSION = "0.2.2" +VERSION = "0.3.0" diff --git a/python/hermes-memory/src/upstash_hermes_memory/memory.py b/python/hermes-memory/src/upstash_hermes_memory/memory.py index f267133..3c383a2 100644 --- a/python/hermes-memory/src/upstash_hermes_memory/memory.py +++ b/python/hermes-memory/src/upstash_hermes_memory/memory.py @@ -1,7 +1,9 @@ """Python port of `AgentMemory` from `@upstash/agentkit-sdk` (packages/sdk/src/memory.ts). -Same keys, same documents, same index: a memory saved by the TS SDK is recalled here and the other -way round, as long as both sides use the same namespace (default `agentkit`). +Same keys and documents as the TS SDK: a memory saved by the TS SDK is recalled here and the other +way round, as long as both sides use the same namespace (default `agentkit`). The search index is +not shared: Python indexes memories and chats together in one index (`agentkit`), the TS SDK +creates its own (`agentkit_memory`), and both index the same keys. key :memory:: value JSON {"text", "userId", "createdAt" (epoch ms), ...metadata}, exactly what the TS SDK writes @@ -13,16 +15,31 @@ Passing `prefix` (or `index_name`) gives the store its own index over `:` instead. """ +import logging import time import uuid from dataclasses import dataclass, field -from typing import Any +from typing import Any, Literal + +import httpx +from upstash_redis.errors import UpstashError from ._search import DEFAULT_NAMESPACE, SearchIndex, dumps, index_name_for, shared_index from ._telemetry import add_telemetry +logger = logging.getLogger(__name__) + DEFAULT_PREFIX = f"{DEFAULT_NAMESPACE}:memory" +# How often a failing hybrid query is logged, per store. +_WARN_EVERY_S = 600.0 + +RecallMode = Literal["auto", "keyword"] + +# What a failed hybrid query can raise: a server error (e.g. the intermittent "dot metric only +# accepts unit-length vectors") or a transport failure. Anything else is a bug and propagates. +_FALLBACK_ERRORS = (UpstashError, httpx.HTTPError) + # The document's own fields. Metadata is spread next to them, so it must not replace them: a # `userId` in metadata would move the memory into another user's searches. RESERVED_FIELDS = frozenset({"kind", "text", "userId", "createdAt"}) @@ -47,6 +64,10 @@ class MemoryRecord: created_at: int score: float | None = None metadata: dict[str, Any] = field(default_factory=dict) + # How a recalled record was found: "keyword" ($smart, BM25 score) or "hybrid" ($rrf of $smart + # and $ann; the score is a rank-fusion value, only comparable within one query). None for + # records that were not searched for (get, list, an empty query). + retrieval: str | None = None class AgentMemory: @@ -60,9 +81,11 @@ def __init__( min_score: float = 0, enable_telemetry: bool = False, ): + """`min_score` applies to keyword results only; hybrid scores are not on the same scale.""" self.redis = redis self.min_score = min_score self.shared = prefix is None and index_name is None + self._warned_at: dict[str, float] = {} if self.shared: self.prefix = f"{namespace}:memory" self.index = shared_index(redis, namespace) @@ -72,7 +95,7 @@ def __init__( redis, name=index_name or index_name_for(self.prefix), prefix=f"{self.prefix}:", - schema=[("text", "TEXT", False), ("userId", "TEXT", True)], + schema=[("text", "HYBRID", False), ("userId", "TEXT", True)], ) add_telemetry(redis, enabled=enable_telemetry) @@ -110,18 +133,57 @@ def add( self.redis.execute(["JSON.SET", self._key(user_id, id), "$", dumps(doc)]) return MemoryRecord(id=id, text=text, created_at=created_at, metadata=metadata or {}) - def recall(self, query: str, *, user_id: str, top_k: int = 5) -> list[MemoryRecord]: - """The `top_k` memories closest to `query` ($smart: full-text plus fuzzy matching).""" + def recall( + self, query: str, *, user_id: str, top_k: int = 5, mode: RecallMode = "auto" + ) -> list[MemoryRecord]: + """The `top_k` memories closest to `query`. + + `mode="auto"` uses hybrid search ($rrf of $smart and $ann). A failed hybrid query falls + back to keyword search ($smart: full-text plus fuzzy matching) once. `mode="keyword"` + always uses keyword search, for callers that need COUNT and QUERY to describe the same + set. + """ + if mode not in ("auto", "keyword"): + raise ValueError(f"mode must be 'auto' or 'keyword' (got {mode!r})") _check_id(user_id, "user_id") - filter = self._filter(user_id) - has_query = bool(query and query.strip()) - if has_query: - filter["text"] = {"$smart": query} - rows = self._memories(self.index.query(filter, limit=top_k), user_id) - min_score = self.min_score if has_query else 0 - return [ - r for r in (self._record(row, user_id) for row in rows) if (r.score or 0) >= min_score - ] + base = self._filter(user_id) + if not (query and query.strip()): + rows = self._memories(self.index.query(base, limit=top_k), user_id) + return [self._record(row, user_id) for row in rows] + if mode == "auto": + hybrid = { + "$rrf": { + "retrievers": [ + {**base, "text": {"$smart": query}}, + { + "$ann": { + "field": "text", + "query": query, + "filter": self._filter(user_id), + } + }, + ] + } + } + try: + rows = self._memories(self.index.query(hybrid, limit=top_k), user_id) + except _FALLBACK_ERRORS as error: + self._warn( + "hybrid", "upstash memory hybrid recall failed, using keyword: %s", error + ) + else: + return [self._record(row, user_id, "hybrid") for row in rows] + rows = self._memories( + self.index.query({**base, "text": {"$smart": query}}, limit=top_k), user_id + ) + records = [self._record(row, user_id, "keyword") for row in rows] + return [r for r in records if (r.score or 0) >= self.min_score] + + def _warn(self, what: str, message: str, *args: Any) -> None: + now = time.monotonic() + if now - self._warned_at.get(what, -_WARN_EVERY_S) >= _WARN_EVERY_S: + self._warned_at[what] = now + logger.warning(message, *args) def list(self, *, user_id: str, limit: int = 100) -> list[MemoryRecord]: _check_id(user_id, "user_id") @@ -145,10 +207,11 @@ def forget(self, id: str, *, user_id: str) -> bool: _check_memory_id(id) return bool(self.redis.delete(self._key(user_id, id))) - def _record(self, row: Any, user_id: str) -> MemoryRecord: + def _record(self, row: Any, user_id: str, retrieval: str | None = None) -> MemoryRecord: id = row.key[len(self._key(user_id, "")) :] record = self._from_doc(id, row.data if isinstance(row.data, dict) else {}) record.score = row.score + record.retrieval = retrieval return record @staticmethod diff --git a/python/hermes-memory/src/upstash_hermes_memory/provider.py b/python/hermes-memory/src/upstash_hermes_memory/provider.py index 3d03f5a..8d3e058 100644 --- a/python/hermes-memory/src/upstash_hermes_memory/provider.py +++ b/python/hermes-memory/src/upstash_hermes_memory/provider.py @@ -52,7 +52,10 @@ def get_secret(name: str, default: str | None = None) -> str | None: # Hermes only lets the primary agent write; subagents, cron jobs and flushes read. _WRITING_CONTEXTS = {"", "primary"} -_PREFETCH_TOP_K = 5 +# Hybrid search always returns its `LIMIT` (the $ann side has no relevance threshold, and its +# similarity scores do not separate related from unrelated memories well enough for a cutoff), so +# automatic recall injects only a few. The search tool has its own limit. +_PREFETCH_TOP_K = 3 _EXCERPT_CHARS = 300 # Built-in memory tool actions Hermes forwards (a patch arrives as a replace). _MIRRORED_ACTIONS = {"add", "replace", "remove"} @@ -79,7 +82,10 @@ def _mirror_id(target: str, text: str) -> str: } SEARCH_SCHEMA = { "name": "upstash_memory_search", - "description": "Search long-term memory for facts related to a query. Returns ids and text.", + "description": ( + "Search long-term memory for facts related to a query. Returns ids, text and a score; " + "scores rank results within one search only (`retrieval` says how they were found)." + ), "parameters": { "type": "object", "properties": { @@ -554,7 +560,8 @@ def _all_matches(self, query: str) -> list[MemoryRecord]: ) if total <= 0: return [] - return self._memory.recall(query, user_id=self._user_id, top_k=total) + # Keyword mode: the COUNT above cannot take $ann, and both must describe the same set. + return self._memory.recall(query, user_id=self._user_id, top_k=total, mode="keyword") def _all_memories(self) -> list[MemoryRecord]: """Every memory of this user, for the substring match that search can't do.""" @@ -611,7 +618,12 @@ def _call(self, tool_name: str, args: dict[str, Any]) -> dict[str, Any]: user_id=self._user_id, top_k=int(args.get("limit") or 5), ) - return {"results": [{"id": h.id, "text": h.text, "score": h.score} for h in hits]} + return { + "results": [ + {"id": h.id, "text": h.text, "score": h.score, "retrieval": h.retrieval} + for h in hits + ] + } if tool_name == "upstash_memory_forget": if not self._write_enabled: raise RuntimeError("memory is read-only in this agent context") diff --git a/python/hermes-memory/tests/test_layout.py b/python/hermes-memory/tests/test_layout.py index 6ee5e30..051b461 100644 --- a/python/hermes-memory/tests/test_layout.py +++ b/python/hermes-memory/tests/test_layout.py @@ -28,7 +28,7 @@ def test_index_names_follow_the_ts_default(): SHARED_CREATE = ( "SEARCH.CREATE agentkit EXISTSOK ON JSON PREFIX 2 agentkit:memory: agentkit:chat: SCHEMA " - "kind TEXT NOTOKENIZE userId TEXT NOTOKENIZE text TEXT sessionId TEXT NOTOKENIZE " + "kind TEXT NOTOKENIZE userId TEXT NOTOKENIZE text HYBRID sessionId TEXT NOTOKENIZE " "userMessages TEXT modelMessages TEXT" ) @@ -45,7 +45,7 @@ def test_a_prefix_gives_a_store_its_own_index_like_the_ts_sdk(): memory = AgentMemory(Recorder(), prefix="agentkit:memory", enable_telemetry=False) assert " ".join(memory.index.create_command()) == ( "SEARCH.CREATE agentkit_memory EXISTSOK ON JSON PREFIX 1 agentkit:memory: " - "SCHEMA text TEXT userId TEXT NOTOKENIZE" + "SCHEMA text HYBRID userId TEXT NOTOKENIZE" ) chats = ChatHistory(Recorder(), prefix="agentkit:chat", enable_telemetry=False) assert " ".join(chats.index.create_command()) == ( diff --git a/python/hermes-memory/tests/test_redis.py b/python/hermes-memory/tests/test_redis.py index 327b243..9924bf0 100644 --- a/python/hermes-memory/tests/test_redis.py +++ b/python/hermes-memory/tests/test_redis.py @@ -41,7 +41,8 @@ def test_memory_lifecycle(memory): memory.add("Someone else likes coffee too", user_id="u2") memory.index.wait_indexing() - hits = memory.recall("coffee", user_id="u1") + # Keyword mode: hybrid search has no relevance threshold, so it would add the unrelated memory. + hits = memory.recall("coffee", user_id="u1", mode="keyword") assert [h.id for h in hits] == ["coffee"] assert hits[0].text == "Arda prefers dark roast coffee" assert hits[0].score > 0 @@ -55,7 +56,7 @@ def test_memory_lifecycle(memory): assert memory.get("coffee", user_id="u1") is None memory.index.wait_indexing() assert memory.count(user_id="u1") == 1 - assert memory.recall("coffee", user_id="u1") == [] + assert memory.recall("coffee", user_id="u1", mode="keyword") == [] def test_memory_reads_documents_written_by_the_ts_sdk(redis, memory): diff --git a/python/hermes-memory/tests/test_semantic.py b/python/hermes-memory/tests/test_semantic.py new file mode 100644 index 0000000..40d24be --- /dev/null +++ b/python/hermes-memory/tests/test_semantic.py @@ -0,0 +1,254 @@ +"""Semantic (hybrid) recall, against a scripted fake Redis.""" + +import json +from dataclasses import dataclass +from typing import Any + +import httpx +import pytest +from upstash_hermes_memory import AgentMemory, ChatHistory, UpstashMemoryProvider +from upstash_hermes_memory import provider as provider_module +from upstash_hermes_memory._search import SHARED_SCHEMA, shared_index +from upstash_redis.errors import UpstashError + +MISMATCH = "ERR Index agentkit already exists with a different configuration" +HYBRID = SHARED_SCHEMA +OLD_TEXT_INDEX = [(f, "TEXT", nt) for f, _, nt in SHARED_SCHEMA] +USER = {"userId": {"$eq": "u1"}, "$mustNot": [{"kind": {"$eq": "chat"}}]} + + +@dataclass +class Row: + key: str + score: float + data: dict[str, Any] + + +def memory_row(id, text="x", score=1.0): + return Row(f"agentkit:memory:u1:{id}", score, {"text": text, "userId": "u1", "createdAt": 1}) + + +class FakeRedis: + """A small model of Upstash Redis Search, enough for the index lifecycle and recall. + + Memory rows and a chat count are the index contents. Every command is recorded. + """ + + def __init__(self, index_schema=None, memories=(), chats=0): + self.index_schema = list(index_schema) if index_schema else None + self.memories = list(memories) + self.chats = chats + self.hybrid_error: Exception | None = None + self.keyword_error: Exception | None = None + self.wait_result: int | None = None + self.commands: list[list[Any]] = [] + self._headers: dict[str, str] = {} + + def execute(self, command): + self.commands.append(command) + op = command[0] + if op == "SEARCH.CREATE": + schema = _schema_of(command) + if self.index_schema is not None and self.index_schema != schema: + raise UpstashError(MISMATCH) + self.index_schema = schema + return 1 + if op == "SEARCH.DROP": + dropped, self.index_schema = self.index_schema is not None, None + return int(dropped) + if op == "SEARCH.WAITINDEXING": + if self.wait_result is not None: + return self.wait_result + return 1 if self.index_schema else 0 + if op == "SEARCH.QUERY": + if self.index_schema is None: + return [] + filter = json.loads(command[2]) + error = self.hybrid_error if "$rrf" in filter else self.keyword_error + if error: + raise error + limit = int(command[command.index("LIMIT") + 1]) if "LIMIT" in command else 10 + return self.memories[:limit] + if op == "SEARCH.COUNT": + if self.index_schema is None: + return -1 + filter = json.loads(command[2]) + if {"$rrf", "$ann"} & set(filter): + raise UpstashError("ERR $ann cannot be used with SEARCH.COUNT") + if filter == {"kind": {"$eq": "chat"}}: + return self.chats + if filter == {}: + return len(self.memories) + self.chats + return len(self.memories) + return 1 + + def delete(self, key): + self.commands.append(["DEL", key]) + return 1 + + def expire(self, key, seconds): + return True + + def ops(self): + return [c[0] for c in self.commands] + + def queries(self): + return [json.loads(c[2]) for c in self.commands if c[0] == "SEARCH.QUERY"] + + +def _schema_of(command): + i = command.index("SCHEMA") + 1 + out, rest = [], command[i:] + while rest: + field, type_ = rest[0], rest[1] + no_tokenize = len(rest) > 2 and rest[2] == "NOTOKENIZE" + out.append((field, type_, no_tokenize)) + rest = rest[3:] if no_tokenize else rest[2:] + return out + + +# -- index creation ------------------------------------------------------------------------------ + + +def test_text_is_the_only_hybrid_field(): + command = shared_index(FakeRedis(), "agentkit").create_command() + assert " ".join(command).endswith( + "SCHEMA kind TEXT NOTOKENIZE userId TEXT NOTOKENIZE text HYBRID sessionId TEXT NOTOKENIZE " + "userMessages TEXT modelMessages TEXT" + ) + + +def test_memory_and_chats_create_the_index_the_same_way(): + redis = FakeRedis() + assert AgentMemory(redis).index.create_command() == ChatHistory(redis).index.create_command() + + +def test_a_0_2_text_index_is_refused_with_the_fix_and_never_dropped(): + redis = FakeRedis(OLD_TEXT_INDEX) + with pytest.raises(RuntimeError, match="SEARCH.DROP agentkit"): + AgentMemory(redis).recall("x", user_id="u1") + assert "SEARCH.DROP" not in redis.ops() + assert redis.queries() == [] # no pointless keyword fallback against the same refusal + + +def test_an_index_that_vanishes_during_create_is_not_marked_ready(): + redis = FakeRedis() + redis.wait_result = 0 + memory = AgentMemory(redis) + with pytest.raises(RuntimeError): + memory.index.ensure() + assert not memory.index._ready + + +# -- recall -------------------------------------------------------------------------------------- + + +def test_hybrid_recall_fuses_keyword_and_meaning_for_this_user_only(): + redis = FakeRedis(HYBRID, [memory_row("1", "Weeks start Monday", 0.03)]) + memory = AgentMemory(redis, min_score=5) + hits = memory.recall("what day does the week begin", user_id="u1", top_k=3) + assert [(h.text, h.retrieval) for h in hits] == [("Weeks start Monday", "hybrid")] + (query,) = redis.queries() + keyword, ann = query["$rrf"]["retrievers"] + assert keyword == {**USER, "text": {"$smart": "what day does the week begin"}} + assert ann == { + "$ann": {"field": "text", "query": "what day does the week begin", "filter": USER} + } + + +@pytest.mark.parametrize( + "error", + [ + UpstashError("ERR dot metric only accepts unit-length vectors"), + httpx.ConnectError("connection reset"), + httpx.ReadTimeout("read timed out"), + ], +) +def test_a_failed_hybrid_query_falls_back_to_keyword_once_with_min_score(error): + redis = FakeRedis(HYBRID, [memory_row("1", "strong", 3.0), memory_row("2", "weak", 0.2)]) + redis.hybrid_error = error + memory = AgentMemory(redis, min_score=1) + hits = memory.recall("billing for big accounts", user_id="u1") + assert [(h.text, h.retrieval) for h in hits] == [("strong", "keyword")] + assert ["$rrf" in q for q in redis.queries()] == [True, False] + assert redis.queries()[1] == {**USER, "text": {"$smart": "billing for big accounts"}} + + +def test_a_failed_fallback_raises_for_the_caller_to_handle(): + redis = FakeRedis(HYBRID, [memory_row("1")]) + redis.hybrid_error = redis.keyword_error = httpx.ConnectError("down") + with pytest.raises(httpx.ConnectError): + AgentMemory(redis).recall("x", user_id="u1") + assert len(redis.queries()) == 2 # one hybrid attempt, one keyword attempt + + +def test_programming_errors_are_not_swallowed_by_the_fallback(): + redis = FakeRedis(HYBRID, [memory_row("1")]) + redis.hybrid_error = TypeError("bug") + with pytest.raises(TypeError): + AgentMemory(redis).recall("x", user_id="u1") + + +def test_keyword_mode_never_uses_a_retriever(): + redis = FakeRedis(HYBRID, [memory_row("1")]) + AgentMemory(redis).recall("x", user_id="u1", mode="keyword") + assert all("$rrf" not in q for q in redis.queries()) + + +def test_an_unknown_mode_is_refused(): + with pytest.raises(ValueError): + AgentMemory(FakeRedis()).recall("x", user_id="u1", mode="vector") + + +def test_an_empty_hybrid_result_checks_the_index_without_counting_a_retriever(): + redis = FakeRedis(HYBRID) + assert AgentMemory(redis).recall("nothing", user_id="u1") == [] + assert [c[2] for c in redis.commands if c[0] == "SEARCH.COUNT"] == ["{}"] + + +# -- provider ------------------------------------------------------------------------------------ + + +@pytest.fixture +def hybrid_provider(monkeypatch): + env = {"UPSTASH_HERMES_USER_ID": "u1"} + monkeypatch.setattr(provider_module, "get_secret", lambda n, d=None: env.get(n, d)) + redis = FakeRedis(HYBRID, [memory_row(str(i), f"m{i}") for i in range(5)]) + provider = UpstashMemoryProvider(redis) + provider.initialize("s1", platform="cli") + return provider, redis + + +def test_prefetch_injects_three_memories(hybrid_provider): + provider, redis = hybrid_provider + block = provider.prefetch("what do I prefer") + assert block.count("\n- ") == 3 + (limit,) = [c[c.index("LIMIT") + 1] for c in redis.commands if c[0] == "SEARCH.QUERY"] + assert limit == "3" + + +def test_prefetch_survives_a_redis_outage(hybrid_provider): + provider, redis = hybrid_provider + redis.hybrid_error = redis.keyword_error = httpx.ConnectError("down") + assert provider.prefetch("anything") == "" + + +def test_the_search_tool_says_how_results_were_found(hybrid_provider): + provider, _ = hybrid_provider + result = json.loads(provider.handle_tool_call("upstash_memory_search", {"query": "m"})) + assert {r["retrieval"] for r in result["results"]} == {"hybrid"} + + +def test_mirror_deletion_matches_by_keyword_only(hybrid_provider): + provider, redis = hybrid_provider + provider._all_matches("old entry") + assert all("$rrf" not in q for q in redis.queries()) + + +def test_wait_indexing_recreates_an_index_that_was_dropped_under_it(): + redis = FakeRedis(HYBRID, [memory_row(1)]) + memory = AgentMemory(redis) + memory.index.ensure() + redis.index_schema = None # another client dropped it + memory.index.wait_indexing() + assert redis.index_schema == HYBRID and memory.index._ready diff --git a/python/hermes-memory/tests/test_semantic_redis.py b/python/hermes-memory/tests/test_semantic_redis.py new file mode 100644 index 0000000..b5d265b --- /dev/null +++ b/python/hermes-memory/tests/test_semantic_redis.py @@ -0,0 +1,73 @@ +"""Semantic recall against a real Upstash Redis (skipped without secrets). + +Hybrid queries can fail intermittently on the server ("dot metric only accepts unit-length +vectors"); recall then falls back to keyword search, so assertions about meaning retry a few times +and only count results that were actually found by hybrid search. +""" + +import pytest +from upstash_hermes_memory import AgentMemory, ChatHistory + +FACTS = { + "week": "Weeks start Monday for him.", + "coffee": "His favorite coffee is a flat white from the place near the office.", + "drafts": "Customer-facing drafts on Slack go out as draft cards, never plain text.", + "invoices": "Invoices for enterprise customers are reviewed on the first business day.", + "host": "Hermes runs only on the ops Box since October; the Mac runs the desktop app.", + "dashes": "He writes without em dashes and leads with the answer.", + "churn": "The churn analysis runs monthly and he reviews it before publication.", + "keys": "1Password is his default credential store.", + "units": "Sizes are written in decimal MB and GB.", + "board": "Side ideas go to the hermes-queue Kanban board.", + "shell": "Shell snippets for him are written in fish.", + "mail": "Never delete or trash email; ask before sending.", +} + + +def hybrid_recall(memory, query, user_id="u1", top_k=3, tries=4): + """Recall that was answered by hybrid search. Results that came back by keyword are the + fallback after a server error, so they are retried; no results at all is a failure.""" + for _ in range(tries): + hits = memory.recall(query, user_id=user_id, top_k=top_k) + assert hits, f"no memories recalled for {query!r}" + if hits[0].retrieval == "hybrid": + return hits + pytest.skip("hybrid search kept failing on the server; recall fell back to keyword") + + +@pytest.fixture +def semantic(redis, prefix): + memory = AgentMemory(redis, namespace=prefix, enable_telemetry=False) + memory.index.ensure() + yield memory + memory.index.drop() + + +def ids(hits): + return [h.id for h in hits] + + +def test_recall_finds_a_memory_by_meaning_within_what_is_injected(semantic): + # Hermes injects the top 3. Equal-weight $rrf can rank a keyword-only match + # (here "first business day" for "day") level with the best meaning match, so the contract + # tested is "among the injected 3", not "first". + for id, text in FACTS.items(): + semantic.add(text, user_id="u1", id=id) + semantic.index.wait_indexing() + assert "week" in ids(hybrid_recall(semantic, "what day does the week begin")) + assert "coffee" in ids(hybrid_recall(semantic, "what hot drink does he enjoy")) + assert "drafts" in ids(hybrid_recall(semantic, "how should a message to a client be sent")) + + +def test_hybrid_recall_never_crosses_users_or_returns_chats(semantic, redis, prefix): + semantic.add("Hermes runs on the ops Box.", user_id="u1", id="mine") + semantic.add("Hermes runs on a Raspberry Pi.", user_id="u2", id="theirs") + chats = ChatHistory(redis, namespace=prefix, enable_telemetry=False) + chats.save_chat("s1", [{"role": "user", "content": "where does hermes run"}], user_id="u1") + semantic.index.wait_indexing() + hits = hybrid_recall(semantic, "where is the assistant hosted", top_k=5) + assert [h.id for h in hits] == ["mine"] + + +def test_an_empty_store_recalls_nothing_without_errors(semantic): + assert semantic.recall("anything at all", user_id="nobody") == [] diff --git a/uv.lock b/uv.lock index ed7d5fd..08f3c48 100644 --- a/uv.lock +++ b/uv.lock @@ -178,7 +178,7 @@ wheels = [ [[package]] name = "upstash-hermes-memory" -version = "0.2.2" +version = "0.3.0" source = { editable = "python/hermes-memory" } dependencies = [ { name = "upstash-redis" },