From c34bae0ac0759cd18563c36863260d7083a74233 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 15:50:46 +0000 Subject: [PATCH 1/2] DOC-7228 Port the Lettuce semantic cache to the typed Redis Search API The demo now pins Lettuce 7.9.0 (was 6.7.1) and uses ftCreate, ftSearch and ftDropindex instead of dispatch() with hand-built CommandArgs, and SearchArgs instead of an alternating reply parsed by position. FT.INFO is the one search command with no typed method in 7.9.0, so indexInfo() still dispatches it. The page said Lettuce 6.7 has no FT.* bindings; that paragraph, the query snippet and the connection snippet are rewritten, and the class javadoc now says which commands are typed. The explicit RESP2 pin is removed. It existed so the hand-written parser saw the flat RESP2 shape, and the typed API parses either protocol. I ran the same driver under RESP2, RESP3 and the default protocol and the results were identical, including the FT.INFO dispatch, which NestedMultiOutput handles in both. MONITOR shows the typed calls send the same FT.SEARCH as before (same RETURN, SORTBY, LIMIT, PARAMS and DIALECT 2). FT.CREATE sends the vector attributes in a different order, which is equivalent. dropIndex tolerates a missing index by message text, and Redis 8.10 words it "SEARCH_INDEX_NOT_FOUND Index not found", so isMissingIndex now covers that wording too. Verified with mvn compile (no deprecation warnings) and a throwaway driver against Redis 8.10.2 with synthetic vectors: dropIndex on a missing index, idempotent createIndex, put, hit, hit_count increment, tenant isolation, far-vector miss, empty-tenant miss, listEntries, indexInfo and clear. The embedding model was not available offline, so LocalEmbedder and the DemoServer HTTP endpoints have not run on 7.9.0. Learned: Lettuce 7.9.0's typed ftSearch replaces the hand-parsed FT.SEARCH reply and works under RESP2 and RESP3, so the RESP2 pin is unnecessary Constraint: FT.INFO has no typed method in Lettuce 7.9.0, so indexInfo() must keep dispatch() with FtCommand.FT_INFO Gaps: LocalEmbedder and the DemoServer endpoints have not run on 7.9.0; only RedisSemanticCache was exercised Recheck: when Lettuce adds a typed ftInfo, replace the FT.INFO dispatch Ticket: DOC-7228 Co-Authored-By: Claude Sonnet 5.5 --- .../semantic-cache/java-lettuce/_index.md | 41 ++-- .../semantic-cache/java-lettuce/pom.xml | 2 +- .../java/com/redis/semcache/DemoServer.java | 12 - .../redis/semcache/RedisSemanticCache.java | 228 ++++++++---------- 4 files changed, 120 insertions(+), 163 deletions(-) diff --git a/content/develop/use-cases/semantic-cache/java-lettuce/_index.md b/content/develop/use-cases/semantic-cache/java-lettuce/_index.md index 12a66c5097..9dcdc532ae 100644 --- a/content/develop/use-cases/semantic-cache/java-lettuce/_index.md +++ b/content/develop/use-cases/semantic-cache/java-lettuce/_index.md @@ -56,20 +56,15 @@ The `RedisSemanticCache` class wraps the Redis Search index and the lookup / wri ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/RedisSemanticCache.java)): ```java -import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.protocol.ProtocolVersion; import com.redis.semcache.RedisSemanticCache; import com.redis.semcache.LocalEmbedder; import com.redis.semcache.LookupResult; import com.redis.semcache.CacheHit; RedisClient client = RedisClient.create("redis://localhost:6379"); -client.setOptions(ClientOptions.builder() - .protocolVersion(ProtocolVersion.RESP2) - .build()); StatefulRedisConnection connection = client.connect(ByteArrayCodec.INSTANCE); @@ -120,7 +115,7 @@ if (result instanceof CacheHit hit) { } ``` -The connection uses Lettuce's `ByteArrayCodec` so the binary float-32 bytes of the embedding can share an `HSET` mapping with the UTF-8 text fields without a second connection. RESP2 is pinned explicitly: Lettuce 6.7 negotiates RESP3 by default, which returns the [`FT.SEARCH`](/content/commands/ft.search.md) reply as a map keyed by `results` / `total_results` instead of the flat alternating list the demo's parser expects, and pinning RESP2 keeps the wire format identical to what the Python, Node, and Jedis ports speak. +The connection uses Lettuce's `ByteArrayCodec` so the binary float-32 bytes of the embedding can share an `HSET` mapping with the UTF-8 text fields without a second connection. ### Data model @@ -158,30 +153,32 @@ FT.CREATE semcache:idx ### The query -The lookup is a hybrid query: a TAG pre-filter expression in parentheses, then `=>[KNN 1 @embedding $vec]`. With `DIALECT 2`, Redis applies the filter first and KNN-ranks only the matching documents. Lettuce 6.7 doesn't yet ship first-class [`FT.SEARCH`](/content/commands/ft.search.md) bindings, so the helper sends the command through `RedisCommands.dispatch()` with a custom `ProtocolKeyword` — the same machinery Lettuce uses internally for any command it does support natively, just spelled out by the caller. The wire bytes are identical to typing the command in `redis-cli`: +The lookup is a hybrid query: a TAG pre-filter expression in parentheses, then `=>[KNN 1 @embedding $vec]`. With `DIALECT 2`, Redis applies the filter first and KNN-ranks only the matching documents. Lettuce has a typed [`FT.SEARCH`](/content/commands/ft.search.md) method, `ftSearch`, so the helper builds the query with `SearchArgs` and sends it on the same `byte[]` connection. The binary `$vec` parameter goes through `SearchArgs.Builder.param(...)`: ```java -CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add("(@tenant:{acme} @locale:{en} @model_version:{gpt\\-4\\.5\\-2026} @safety:{ok})" - + "=>[KNN 1 @embedding $vec AS distance]") - .add("RETURN").add(7) - .add("prompt").add("response").add("tenant").add("locale") - .add("model_version").add("hit_count").add("distance") - .add("SORTBY").add("distance").add("ASC") - .add("LIMIT").add(0).add(1) - .add("PARAMS").add(2).add("vec".getBytes()).add(LocalEmbedder.toBytes(queryVec)) - .add("DIALECT").add(2); - -List raw = sync.dispatch( - FtCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), +SearchArgs args = SearchArgs.builder() + .param("vec", LocalEmbedder.toBytes(queryVec)) + .returnField("prompt").returnField("response") + .returnField("tenant").returnField("locale") + .returnField("model_version").returnField("hit_count") + .returnField("distance") + .sortBy(SortByArgs.builder().attribute("distance").build()) + .limit(0, 1) + .dialect(QueryDialects.DIALECT2) + .build(); + +SearchReply reply = sync.ftSearch( + "semcache:idx", + "(@tenant:{acme} @locale:{en} @model_version:{gpt\\-4\\.5\\-2026} @safety:{ok})" + + "=>[KNN 1 @embedding $vec AS distance]", args ); ``` `distance` is the cosine *distance* (0 means identical, 2 means opposite). The result is sorted ascending, so the top row is the closest candidate. The application inspects `distance` against the threshold and decides hit or miss in user code — Redis returns the row either way, and treating it as a hit or a miss is a policy decision the cache helper owns, not a server-side filter. +The index itself is created with the typed `ftCreate` method and `dropIndex` uses `ftDropindex`. `FT.INFO` is the one Redis Search command Lettuce has no typed method for, so the helper's `indexInfo()` sends it with `dispatch()` and a custom `ProtocolKeyword`. + ## The mock LLM To make the latency and token savings visible without requiring an API key, `MockLLM.java` provides a deterministic stand-in diff --git a/content/develop/use-cases/semantic-cache/java-lettuce/pom.xml b/content/develop/use-cases/semantic-cache/java-lettuce/pom.xml index f28828fcbd..24efba43d3 100644 --- a/content/develop/use-cases/semantic-cache/java-lettuce/pom.xml +++ b/content/develop/use-cases/semantic-cache/java-lettuce/pom.xml @@ -21,7 +21,7 @@ 17 17 UTF-8 - 6.7.1.RELEASE + 7.9.0.RELEASE 0.33.0 20240303 diff --git a/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/DemoServer.java b/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/DemoServer.java index c5006e2a3d..dca0b285fa 100644 --- a/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/DemoServer.java +++ b/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/DemoServer.java @@ -3,12 +3,10 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; -import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.protocol.ProtocolVersion; import org.json.JSONArray; import org.json.JSONObject; @@ -65,16 +63,6 @@ public static void main(String[] argv) throws Exception { .withTimeout(Duration.ofSeconds(2)) .build(); RedisClient client = RedisClient.create(uri); - // Pin the connection to RESP2 so FT.SEARCH and FT.INFO come - // back in the flat alternating-list shape the cache parses. - // Lettuce 6.7 negotiates RESP3 by default, which wraps the - // search reply in a map keyed by "results", "total_results", - // etc. — a richer shape, but one the demo's parser doesn't - // need; RESP2 keeps the wire format identical to what the - // Python, Node, Go, and Jedis ports already speak. - client.setOptions(ClientOptions.builder() - .protocolVersion(ProtocolVersion.RESP2) - .build()); StatefulRedisConnection connection; try { diff --git a/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/RedisSemanticCache.java b/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/RedisSemanticCache.java index d3da18d9bd..7d79f97233 100644 --- a/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/RedisSemanticCache.java +++ b/content/develop/use-cases/semantic-cache/java-lettuce/src/main/java/com/redis/semcache/RedisSemanticCache.java @@ -6,12 +6,23 @@ import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.codec.ByteArrayCodec; import io.lettuce.core.output.NestedMultiOutput; -import io.lettuce.core.output.StatusOutput; import io.lettuce.core.protocol.CommandArgs; import io.lettuce.core.protocol.ProtocolKeyword; +import io.lettuce.core.search.FieldValue; +import io.lettuce.core.search.SearchReply; +import io.lettuce.core.search.arguments.CreateArgs; +import io.lettuce.core.search.arguments.FieldArgs; +import io.lettuce.core.search.arguments.NumericFieldArgs; +import io.lettuce.core.search.arguments.QueryDialects; +import io.lettuce.core.search.arguments.SearchArgs; +import io.lettuce.core.search.arguments.SortByArgs; +import io.lettuce.core.search.arguments.TagFieldArgs; +import io.lettuce.core.search.arguments.TextFieldArgs; +import io.lettuce.core.search.arguments.VectorFieldArgs; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -48,14 +59,16 @@ * {@code allkeys-lfu} eviction policy on the database to cap memory * under pressure too. * - *

Lettuce 6.7 doesn't yet ship first-class {@code FT.*} bindings, - * so the cache uses {@code dispatch()} with a custom - * {@link ProtocolKeyword} for {@code FT.CREATE}, {@code FT.SEARCH}, - * {@code FT.INFO}, and {@code FT.DROPINDEX}. Everything else — - * {@code HSET}, {@code HINCRBY}, {@code EXPIRE}, {@code TTL}, - * {@code MULTI}/{@code EXEC} — goes through the built-in synchronous - * API on a {@code byte[]}-codec connection so binary embedding bytes - * and UTF-8 text can share the same hash without separate connections. + *

{@code FT.CREATE}, {@code FT.SEARCH}, and {@code FT.DROPINDEX} go + * through Lettuce's typed {@code RediSearchCommands} API + * ({@code ftCreate}, {@code ftSearch}, {@code ftDropindex}). Lettuce + * has no typed {@code FT.INFO} helper, so that one command is sent + * with {@code dispatch()} and a custom {@link ProtocolKeyword}. + * Everything else — {@code HSET}, {@code HINCRBY}, {@code EXPIRE}, + * {@code TTL}, {@code MULTI}/{@code EXEC} — goes through the built-in + * synchronous API on a {@code byte[]}-codec connection so binary + * embedding bytes and UTF-8 text can share the same hash without + * separate connections. */ public final class RedisSemanticCache { @@ -69,22 +82,14 @@ public final class RedisSemanticCache { private static final String TAG_SPECIAL = "\\,.<>{}[]\"':;!@#$%^&*()-+=~| "; /** - * Custom {@link ProtocolKeyword}s for the Redis Search - * subcommands we send via {@code dispatch()}. Lettuce 6.7 has no - * native {@code FT.*} bindings, but {@code dispatch()} accepts - * any keyword whose {@code getBytes()} returns the raw command - * name. We deliberately spell out each one as its own keyword - * (rather than {@code add("CREATE")} on a single {@code FT} - * keyword) so the wire bytes match the standard Redis Search - * tooling — {@code MONITOR}, {@code LATENCY HISTORY}, server-side - * ACL rules — exactly as if the commands had been typed in + * Custom {@link ProtocolKeyword} for {@code FT.INFO}, the one Redis + * Search command Lettuce has no typed method for. {@code dispatch()} + * accepts any keyword whose {@code getBytes()} returns the raw + * command name, so the wire bytes match what you would type in * {@code redis-cli}. */ private enum FtCommand implements ProtocolKeyword { - FT_CREATE("FT.CREATE"), - FT_SEARCH("FT.SEARCH"), - FT_INFO("FT.INFO"), - FT_DROPINDEX("FT.DROPINDEX"); + FT_INFO("FT.INFO"); private final byte[] bytes; private final String wire; @@ -191,29 +196,28 @@ private byte[] entryKeyBytes(String entryId) { * but the cache lookup itself is vector-only. */ public void createIndex() { - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add("ON").add("HASH") - .add("PREFIX").add(1).add(keyPrefix.getBytes(StandardCharsets.UTF_8)) - .add("SCHEMA") - .add("prompt").add("TEXT") - .add("response").add("TEXT") - .add("tenant").add("TAG") - .add("locale").add("TAG") - .add("model_version").add("TAG") - .add("safety").add("TAG") - .add("created_ts").add("NUMERIC").add("SORTABLE") - .add("hit_count").add("NUMERIC").add("SORTABLE") - .add("embedding").add("VECTOR").add("HNSW").add(6) - .add("TYPE").add("FLOAT32") - .add("DIM").add(vectorDim) - .add("DISTANCE_METRIC").add("COSINE"); + List schema = Arrays.asList( + TextFieldArgs.builder().name("prompt").build(), + TextFieldArgs.builder().name("response").build(), + TagFieldArgs.builder().name("tenant").build(), + TagFieldArgs.builder().name("locale").build(), + TagFieldArgs.builder().name("model_version").build(), + TagFieldArgs.builder().name("safety").build(), + NumericFieldArgs.builder().name("created_ts").sortable().build(), + NumericFieldArgs.builder().name("hit_count").sortable().build(), + VectorFieldArgs.builder() + .name("embedding") + .hnsw() + .type(VectorFieldArgs.VectorType.FLOAT32) + .dimensions(vectorDim) + .distanceMetric(VectorFieldArgs.DistanceMetric.COSINE) + .build()); + CreateArgs createArgs = CreateArgs.builder() + .on(CreateArgs.TargetType.HASH) + .withPrefix(keyPrefix) + .build(); try { - sync.dispatch( - FtCommand.FT_CREATE, - new StatusOutput<>(ByteArrayCodec.INSTANCE), - args - ); + sync.ftCreate(indexName, createArgs, schema); } catch (RedisException ex) { if (!String.valueOf(ex.getMessage()).contains("Index already exists")) { throw ex; @@ -223,25 +227,26 @@ public void createIndex() { /** Drop the search index. Optionally also delete cached entries. */ public void dropIndex(boolean deleteDocuments) { - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes); - if (deleteDocuments) { - args.add("DD"); - } try { - sync.dispatch( - FtCommand.FT_DROPINDEX, - new StatusOutput<>(ByteArrayCodec.INSTANCE), - args - ); + sync.ftDropindex(indexName, deleteDocuments); } catch (RedisException ex) { - String msg = String.valueOf(ex.getMessage()).toLowerCase(Locale.ROOT); - if (!msg.contains("no such index") && !msg.contains("unknown index name")) { + if (!isMissingIndex(ex)) { throw ex; } } } + /** + * Redis Search phrases the missing-index error differently across + * versions ("Unknown index name", "no such index", and, from Redis + * 8.10, "SEARCH_INDEX_NOT_FOUND Index not found"). + */ + private static boolean isMissingIndex(RedisException ex) { + String msg = String.valueOf(ex.getMessage()).toLowerCase(Locale.ROOT); + return msg.contains("no such index") || msg.contains("unknown index name") + || msg.contains("index not found"); + } + // ------------------------------------------------------------------ // Lookup // ------------------------------------------------------------------ @@ -287,24 +292,20 @@ public LookupResult lookup( String knnQuery = filterClause + "=>[KNN 1 @embedding $vec AS distance]"; byte[] vecBytes = LocalEmbedder.toBytes(queryVec); - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add(knnQuery) - .add("RETURN").add(7) - .add("prompt").add("response").add("tenant").add("locale") - .add("model_version").add("hit_count").add("distance") - .add("SORTBY").add("distance").add("ASC") - .add("LIMIT").add(0).add(1) - .add("PARAMS").add(2).add("vec".getBytes(StandardCharsets.UTF_8)).add(vecBytes) - .add("DIALECT").add(2); - - List raw = sync.dispatch( - FtCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), - args - ); + SearchArgs args = SearchArgs.builder() + .param("vec", vecBytes) + .returnField("prompt").returnField("response") + .returnField("tenant").returnField("locale") + .returnField("model_version").returnField("hit_count") + .returnField("distance") + .sortBy(SortByArgs.builder().attribute("distance").build()) + .limit(0, 1) + .dialect(QueryDialects.DIALECT2) + .build(); - SearchHit hit = parseFirstHit(raw); + SearchReply reply = sync.ftSearch(indexName, knnQuery, args); + + SearchHit hit = firstHit(reply); if (hit == null) { return new CacheMiss(null, null); } @@ -535,23 +536,19 @@ public Map indexInfo() { /** Return every cached entry (no embedding) for the admin UI. */ public List> listEntries(int limit) { - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add("*") - .add("RETURN").add(8) - .add("prompt").add("response").add("tenant").add("locale") - .add("model_version").add("safety").add("created_ts").add("hit_count") - .add("SORTBY").add("created_ts").add("DESC") - .add("LIMIT").add(0).add(limit) - .add("DIALECT").add(2); - - List raw = sync.dispatch( - FtCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), - args - ); - - List hits = parseAllHits(raw); + SearchArgs args = SearchArgs.builder() + .returnField("prompt").returnField("response") + .returnField("tenant").returnField("locale") + .returnField("model_version").returnField("safety") + .returnField("created_ts").returnField("hit_count") + .sortBy(SortByArgs.builder().attribute("created_ts").descending().build()) + .limit(0, limit) + .dialect(QueryDialects.DIALECT2) + .build(); + + SearchReply reply = sync.ftSearch(indexName, "*", args); + + List hits = allHits(reply); List> out = new ArrayList<>(hits.size()); for (SearchHit hit : hits) { String entryId = hit.id.startsWith(keyPrefix) @@ -608,47 +605,22 @@ private static final class SearchHit { } } - /** - * Parse the first hit from an FT.SEARCH reply. The RESP2 shape - * is [count, key1, [field, value, field, value, ...], key2, ...]. - * Returns {@code null} when count is 0 or the reply is shorter - * than expected. - */ - private static SearchHit parseFirstHit(List raw) { - if (raw == null || raw.isEmpty()) return null; - long count = parseLong(raw.get(0), 0L); - if (count <= 0 || raw.size() < 3) return null; - String id = decode(raw.get(1)); - Map fields = fieldsToMap(raw.get(2)); - return new SearchHit(id, fields); + /** The first hit of a typed FT.SEARCH reply, or {@code null} if it had none. */ + private static SearchHit firstHit(SearchReply reply) { + List hits = allHits(reply); + return hits.isEmpty() ? null : hits.get(0); } - /** Parse every hit from an FT.SEARCH reply, preserving order. */ - private static List parseAllHits(List raw) { + /** Every hit of a typed FT.SEARCH reply, preserving order. */ + private static List allHits(SearchReply reply) { List out = new ArrayList<>(); - if (raw == null || raw.size() < 3) return out; - // index 0 holds the total count; entries follow as - // (key, fields)* pairs. - for (int i = 1; i + 1 < raw.size(); i += 2) { - String id = decode(raw.get(i)); - Map fields = fieldsToMap(raw.get(i + 1)); - out.add(new SearchHit(id, fields)); - } - return out; - } - - /** - * Turn the [field, value, field, value, ...] array inside an - * FT.SEARCH document into a {@code Map}, decoding - * every entry as UTF-8. - */ - private static Map fieldsToMap(Object array) { - Map out = new HashMap<>(); - if (!(array instanceof List list)) return out; - for (int i = 0; i + 1 < list.size(); i += 2) { - String field = decode(list.get(i)); - String value = decode(list.get(i + 1)); - out.put(field, value); + if (reply == null) return out; + for (SearchReply.SearchResult result : reply.getResults()) { + Map fields = new HashMap<>(); + for (Map.Entry field : result.getFields().entrySet()) { + fields.put(field.getKey(), field.getValue().asString()); + } + out.add(new SearchHit(decode(result.getId()), fields)); } return out; } From 3433472a20d90fd06309e4c9224f246f9ad80fa4 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 15:57:21 +0000 Subject: [PATCH 2/2] DOC-7228 Port the Lettuce agent memory demo to the typed search and JSON APIs The demo now pins Lettuce 7.9.0 (was 6.7.1). LongTermMemory uses ftCreate, ftSearch and ftDropindex for Redis Search and jsonSet and jsonNumincrby for Redis JSON, instead of dispatch() with hand-built CommandArgs and a parser that walked the nested reply. JSON.SET still runs inside MULTI/EXEC under the shared txLock, and the typed call queues like any other command. FT.INFO has no typed method in 7.9.0, so indexInfo() still dispatches it. The page claimed Lettuce 6.7 has no FT.* or JSON.* bindings; the overview, the recall and remember steps, the connection snippet and the "Sending FT.* / JSON.* commands" section are rewritten around the typed calls. The explicit RESP2 pin is removed from DemoServer and the page: the typed parser handles both protocols, which I confirmed by running the same driver under RESP2, RESP3 and the default protocol with identical results, and by recording the wire commands with MONITOR (FT.CREATE on a JSON index with the same aliased schema, FT.SEARCH with the same KNN query, RETURN list, SORTBY, LIMIT and DIALECT 2, JSON.SET with the document and JSON.NUMINCRBY on $.hit_count). dropIndex tolerated a missing index only by two message texts, and Redis 8.10 words it "SEARCH_INDEX_NOT_FOUND Index not found", so isMissingIndex now covers that wording too. Verified with mvn compile (no deprecation warnings) and a throwaway driver against Redis 8.10.2 with synthetic vectors: dropIndex on a missing index, idempotent createIndex, remember (new write, same-vector dedupe that bumps hit_count, other user not deduped), recall with its TTL, scope isolation, listMemories, indexInfo, deleteMemory and clear. The embedding model was not available offline, so LocalEmbedder, AgentSession, AgentEventLog and the DemoServer endpoints have not run on 7.9.0. Learned: Lettuce 7.9.0 has typed ftSearch, ftCreate, ftDropindex, jsonSet and jsonNumincrby that work inside MULTI/EXEC and under RESP2 and RESP3 Constraint: FT.INFO has no typed method in Lettuce 7.9.0, so indexInfo() must keep dispatch() with ModuleCommand.FT_INFO Gaps: only LongTermMemory was exercised; LocalEmbedder, AgentSession, AgentEventLog and the DemoServer endpoints have not run on 7.9.0 Recheck: when Lettuce adds a typed ftInfo, replace the FT.INFO dispatch Ticket: DOC-7228 Co-Authored-By: Claude Sonnet 5.5 --- .../memory-layer/java-lettuce/_index.md | 69 +++--- .../memory-layer/java-lettuce/pom.xml | 2 +- .../java/com/redis/agentmem/DemoServer.java | 12 - .../com/redis/agentmem/LongTermMemory.java | 212 ++++++++---------- 4 files changed, 126 insertions(+), 169 deletions(-) diff --git a/content/develop/use-cases/memory-layer/java-lettuce/_index.md b/content/develop/use-cases/memory-layer/java-lettuce/_index.md index 843366ae46..194f7e8f4e 100644 --- a/content/develop/use-cases/memory-layer/java-lettuce/_index.md +++ b/content/develop/use-cases/memory-layer/java-lettuce/_index.md @@ -19,7 +19,7 @@ This guide shows you how to build a small Redis-backed memory layer in Java with The embedder is [DJL](https://djl.ai/) (`ai.djl.huggingface.tokenizers` + `ai.djl.pytorch.pytorch-model-zoo`) running the canonical `sentence-transformers/all-MiniLM-L6-v2` PyTorch checkpoint — the same library and model the existing [Lettuce vector-search example](/content/develop/clients/lettuce/vecsearch.md) uses, and the same encoder the [Python](/content/develop/use-cases/memory-layer/redis-py/_index.md) example loads. DJL drives libtorch through the same C++ runtime as Python's PyTorch bindings, so the vectors produced here are numerically identical to the Python ones to within rounding noise, and the distance bands the Python walkthrough quotes carry over to this demo without recalibration. A memory written by one demo can be recalled by the other against the same Redis instance. -The big shape difference from the [sister Jedis port](/content/develop/use-cases/memory-layer/java-jedis/_index.md) is that Lettuce 6.7 doesn't ship first-class `FT.*` or `JSON.*` bindings yet. The helper classes send those commands through Lettuce's generic `dispatch()` API with a custom `ProtocolKeyword`, and the FT.SEARCH/FT.INFO replies are parsed by walking the resulting `List` — see [`LongTermMemory.java`](https://github.com/redis/docs/blob/main/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java) for the parser. The same connection is pinned to RESP2 at startup so the reply shape stays a flat array (Lettuce 6.7 negotiates RESP3 by default, which wraps the data in a map structure the parser would have to special-case). +The big shape difference from the [sister Jedis port](/content/develop/use-cases/memory-layer/java-jedis/_index.md) is that the helpers use Lettuce's typed `RediSearchCommands` and `RedisJsonCommands` APIs (`ftCreate`, `ftSearch`, `ftDropindex`, `jsonSet`, `jsonNumincrby`) on a `byte[]`-codec connection, where the Jedis port builds the commands by hand. `FT.INFO` is the one exception: Lettuce has no typed method for it, so the helper sends it with `dispatch()` and a custom `ProtocolKeyword` — see [`LongTermMemory.java`](https://github.com/redis/docs/blob/main/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java). ## Overview @@ -44,8 +44,8 @@ Each turn through the agent loop touches all three tiers in one pass: append to 1. The application calls `embedder.encodeOne(text)` to turn the incoming turn into a 384-element `float[]`. 2. `session.appendTurn(threadId, role, content, user, agent, null)` reads the per-thread Hash with [`HGETALL`](/content/commands/hgetall.md), appends the new turn to the rolling window in application code, trims it back to the configured maximum, and writes the Hash back with [`HSET`](/content/commands/hset.md) + [`EXPIRE`](/content/commands/expire.md) inside Lettuce's `multi()` / `exec()` transaction. The session TTL refreshes on every write so an active thread stays alive. -3. `memory.recall(vec, user, namespace, null, 5, threshold)` issues `FT.SEARCH` via `dispatch()` with a TAG pre-filter and a `KNN 5` clause. Redis returns the closest matching memories together with their cosine distances; memories beyond the recall threshold are dropped before they reach the agent so an unrelated query doesn't surface confident-looking false positives. -4. `memory.remember(text, vec, user, namespace, kind, threadId, null)` runs the same KNN with a tighter dedup threshold. If an existing memory is within the threshold, the new write is skipped and the existing memory's `hit_count` is incremented with `JSON.NUMINCRBY` (also via `dispatch()`) — best-effort: if the memory's TTL has elapsed between the recall and the bump, the increment quietly fails and the hit count for that recall is lost. Otherwise a fresh JSON document is written with `JSON.SET` and a per-kind [`EXPIRE`](/content/commands/expire.md) inside the same `multi()` transaction. +3. `memory.recall(vec, user, namespace, null, 5, threshold)` issues `FT.SEARCH` through the typed `ftSearch` method with a TAG pre-filter and a `KNN 5` clause. Redis returns the closest matching memories together with their cosine distances; memories beyond the recall threshold are dropped before they reach the agent so an unrelated query doesn't surface confident-looking false positives. +4. `memory.remember(text, vec, user, namespace, kind, threadId, null)` runs the same KNN with a tighter dedup threshold. If an existing memory is within the threshold, the new write is skipped and the existing memory's `hit_count` is incremented with `JSON.NUMINCRBY` (`jsonNumincrby`) — best-effort: if the memory's TTL has elapsed between the recall and the bump, the increment quietly fails and the hit count for that recall is lost. Otherwise a fresh JSON document is written with `JSON.SET` and a per-kind [`EXPIRE`](/content/commands/expire.md) inside the same `multi()` transaction. 5. `events.record(threadId, action, detail)` appends one entry to the per-thread Stream with [`XADD MAXLEN ~`](/content/commands/xadd.md) (`XAddArgs.Builder.maxlen(N).approximateTrimming()`), bounding retention to roughly a thousand entries per thread without an explicit cleanup job. The embedding is computed once and reused for steps 3 and 4 — there's no point encoding the same text twice. Recall runs before the write, so the agent doesn't see its own just-written turn echoed back as a recalled memory. @@ -57,21 +57,11 @@ The embedding is computed once and reused for steps 3 and 4 — there's no point ```java import com.redis.agentmem.AgentSession; import com.redis.agentmem.SessionState; -import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.protocol.ProtocolVersion; RedisClient client = RedisClient.create("redis://localhost:6379"); -// Pin to RESP2 so FT.SEARCH / FT.INFO reply with the flat arrays -// the helper's parser expects (Lettuce 6.7 otherwise negotiates -// RESP3 against Redis 7+ and wraps the same data in a map shape — -// see the "Sending FT.* / JSON.* commands through Lettuce" section -// below for the longer story). -client.setOptions(ClientOptions.builder() - .protocolVersion(ProtocolVersion.RESP2) - .build()); StatefulRedisConnection connection = client.connect(ByteArrayCodec.INSTANCE); // Shared lock across helpers so concurrent MULTI/EXEC spans on the @@ -150,41 +140,38 @@ for (MemoryRecord h : memory.recall(q, "alice", "default", null, 5, null)) { } ``` -### Sending FT.* / JSON.* commands through Lettuce +### Using the typed search and JSON commands -Lettuce 6.7 doesn't ship first-class `FT.*` or `JSON.*` bindings, so the helper sends them through `dispatch()` with a custom `ProtocolKeyword`. Each command is its own keyword whose `getBytes()` returns the literal command name; the rest is built up with `CommandArgs`: +Lettuce has typed methods for the Redis Search and Redis JSON commands the helper needs. `ftSearch` takes the query string and a `SearchArgs` builder, and the binary `$vec` parameter goes through `SearchArgs.Builder.param(...)`: ```java -private enum ModuleCommand implements ProtocolKeyword { - FT_CREATE("FT.CREATE"), - FT_SEARCH("FT.SEARCH"), - FT_INFO("FT.INFO"), - FT_DROPINDEX("FT.DROPINDEX"), - JSON_SET("JSON.SET"), - JSON_NUMINCRBY("JSON.NUMINCRBY"); - // ...implementation elided -} - -CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add(filterClause + "=>[KNN 5 @embedding $vec AS distance]") - .add("PARAMS").add(2).add("vec".getBytes(UTF_8)).add(vecBytes) - .add("RETURN").add(8).add("user").add("namespace").add("kind") - .add("source_thread").add("text").add("created_ts") - .add("hit_count").add("distance") - .add("SORTBY").add("distance").add("ASC") - .add("LIMIT").add(0).add(5) - .add("DIALECT").add(2); - -List raw = sync.dispatch( - ModuleCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), +SearchArgs args = SearchArgs.builder() + .param("vec", vecBytes) + .returnField("user").returnField("namespace").returnField("kind") + .returnField("source_thread").returnField("text") + .returnField("created_ts").returnField("hit_count") + .returnField("distance") + .sortBy(SortByArgs.builder().attribute("distance").build()) + .limit(0, 5) + .dialect(QueryDialects.DIALECT2) + .build(); + +SearchReply reply = sync.ftSearch( + indexName, + filterClause + "=>[KNN 5 @embedding $vec AS distance]", args); ``` -The `NestedMultiOutput` collects the RESP reply into a `List` — a flat `[count, key1, fields1, key2, fields2, ...]` array under RESP2, where each `fields` is itself a nested list of alternating field/value pairs. The helper walks that with `parseAllHits()` and `fieldsToMap()` (both private to `LongTermMemory`). +Each result's `getFields()` is a `Map`, and the helper reads the values with `asString()`. The index is created with `ftCreate`, using `CreateArgs` for the JSON target and prefix and one `FieldArgs` builder per schema field (`.name("$.text").as("text")` maps a JSON path to the short alias the queries use). The JSON writes use `jsonSet` and `jsonNumincrby`: + +```java +sync.jsonSet(keyBytes, JsonPath.ROOT_PATH, doc.toString()); +sync.jsonNumincrby(keyBytes, JsonPath.of("$.hit_count"), 1); +``` + +`FT.INFO` is the one command Lettuce has no typed method for. The helper sends it with `dispatch()` and a custom `ProtocolKeyword` whose `getBytes()` returns the literal command name, and reads the alternating key and value reply that `NestedMultiOutput` collects. -The connection is pinned to RESP2 at startup with `ClientOptions.builder().protocolVersion(ProtocolVersion.RESP2)`. Lettuce 6.7 otherwise negotiates RESP3 against Redis 7+, which wraps `FT.SEARCH` and `FT.INFO` results in map / set shapes that the flat-array parser would have to special-case across every reply path. +The typed methods parse both RESP2 and RESP3 replies, so the demo doesn't pin a protocol version. ### Data model diff --git a/content/develop/use-cases/memory-layer/java-lettuce/pom.xml b/content/develop/use-cases/memory-layer/java-lettuce/pom.xml index bcafb79856..e3362fce4d 100644 --- a/content/develop/use-cases/memory-layer/java-lettuce/pom.xml +++ b/content/develop/use-cases/memory-layer/java-lettuce/pom.xml @@ -21,7 +21,7 @@ 17 17 UTF-8 - 6.7.1.RELEASE + 7.9.0.RELEASE 0.33.0 20240303 diff --git a/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/DemoServer.java b/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/DemoServer.java index 026b167ce6..c13c65504a 100644 --- a/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/DemoServer.java +++ b/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/DemoServer.java @@ -3,12 +3,10 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; -import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.protocol.ProtocolVersion; import org.json.JSONArray; import org.json.JSONObject; @@ -62,16 +60,6 @@ public static void main(String[] argv) throws Exception { RedisClient client = RedisClient.create( RedisURI.Builder.redis(args.redisHost, args.redisPort).build()); - // Pin the connection to RESP2 so FT.SEARCH and FT.INFO come - // back as the flat arrays our `NestedMultiOutput` parser - // expects. Lettuce 6.7 negotiates RESP3 by default with - // Redis 7+, which wraps the same data in a map/set structure - // we'd have to special-case across both the recall query - // and the FT.INFO snapshot; RESP2 keeps the wire format - // identical to what `redis-cli` shows on the command line. - client.setOptions(ClientOptions.builder() - .protocolVersion(ProtocolVersion.RESP2) - .build()); StatefulRedisConnection connection; try { connection = client.connect(ByteArrayCodec.INSTANCE); diff --git a/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java b/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java index 5dc4476669..b1a7476e1e 100644 --- a/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java +++ b/content/develop/use-cases/memory-layer/java-lettuce/src/main/java/com/redis/agentmem/LongTermMemory.java @@ -5,15 +5,26 @@ import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.json.JsonPath; import io.lettuce.core.output.NestedMultiOutput; -import io.lettuce.core.output.StatusOutput; -import io.lettuce.core.output.ValueOutput; import io.lettuce.core.protocol.CommandArgs; import io.lettuce.core.protocol.ProtocolKeyword; +import io.lettuce.core.search.FieldValue; +import io.lettuce.core.search.SearchReply; +import io.lettuce.core.search.arguments.CreateArgs; +import io.lettuce.core.search.arguments.FieldArgs; +import io.lettuce.core.search.arguments.NumericFieldArgs; +import io.lettuce.core.search.arguments.QueryDialects; +import io.lettuce.core.search.arguments.SearchArgs; +import io.lettuce.core.search.arguments.SortByArgs; +import io.lettuce.core.search.arguments.TagFieldArgs; +import io.lettuce.core.search.arguments.TextFieldArgs; +import io.lettuce.core.search.arguments.VectorFieldArgs; import org.json.JSONObject; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -49,12 +60,14 @@ * TTL by default. * * - *

Lettuce 6.7 doesn't ship first-class {@code FT.*} or - * {@code JSON.*} bindings, so the helper sends them through - * {@code dispatch()} with custom {@link ProtocolKeyword}s. Everything - * else — {@code EXPIRE}, {@code TTL}, {@code DEL}, - * {@code MULTI}/{@code EXEC} — goes through the built-in synchronous - * API on a {@code byte[]}-codec connection. + *

{@code FT.CREATE}, {@code FT.SEARCH}, and {@code FT.DROPINDEX} go + * through Lettuce's typed {@code RediSearchCommands} API, and + * {@code JSON.SET} and {@code JSON.NUMINCRBY} through its typed + * {@code RedisJsonCommands} API. Lettuce has no typed {@code FT.INFO} + * helper, so that one command is sent with {@code dispatch()} and a + * custom {@link ProtocolKeyword}. Everything else — {@code EXPIRE}, + * {@code TTL}, {@code DEL}, {@code MULTI}/{@code EXEC} — goes through + * the built-in synchronous API on a {@code byte[]}-codec connection. */ public final class LongTermMemory { @@ -78,20 +91,14 @@ public static Map defaultTtlByKind() { private static final String TAG_SPECIAL = "\\,.<>{}[]\"':;!@#$%^&*()-+=~| "; /** - * Custom {@link ProtocolKeyword}s for the Redis Search and Redis - * JSON subcommands the helper sends via {@code dispatch()}. - * Lettuce 6.7 has no native bindings; {@code dispatch()} accepts - * any keyword whose {@code getBytes()} returns the raw command - * name, so spelling each command out as its own keyword keeps - * the wire bytes matching what {@code redis-cli} would send. + * Custom {@link ProtocolKeyword} for {@code FT.INFO}, the one + * command the helper uses that Lettuce has no typed method for. + * {@code dispatch()} accepts any keyword whose {@code getBytes()} + * returns the raw command name, so the wire bytes match what + * {@code redis-cli} would send. */ private enum ModuleCommand implements ProtocolKeyword { - FT_CREATE("FT.CREATE"), - FT_SEARCH("FT.SEARCH"), - FT_INFO("FT.INFO"), - FT_DROPINDEX("FT.DROPINDEX"), - JSON_SET("JSON.SET"), - JSON_NUMINCRBY("JSON.NUMINCRBY"); + FT_INFO("FT.INFO"); private final byte[] bytes; private final String wire; @@ -167,31 +174,30 @@ private byte[] memoryKeyBytes(String memoryId) { // ------------------------------------------------------------------ public void createIndex() { - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add("ON").add("JSON") - .add("PREFIX").add(1).add(keyPrefixBytes) - .add("SCHEMA") - .add("$.text").add("AS").add("text").add("TEXT") - .add("$.user").add("AS").add("user").add("TAG") - .add("$.namespace").add("AS").add("namespace").add("TAG") - .add("$.kind").add("AS").add("kind").add("TAG") - .add("$.source_thread").add("AS").add("source_thread").add("TAG") - .add("$.created_ts").add("AS").add("created_ts") - .add("NUMERIC").add("SORTABLE") - .add("$.hit_count").add("AS").add("hit_count") - .add("NUMERIC").add("SORTABLE") - .add("$.embedding").add("AS").add("embedding") - .add("VECTOR").add("HNSW").add(6) - .add("TYPE").add("FLOAT32") - .add("DIM").add(vectorDim) - .add("DISTANCE_METRIC").add("COSINE"); + // The index reads JSON paths and exposes each under a short + // alias, so queries and RETURN lists use "text", "user", ... + List schema = Arrays.asList( + TextFieldArgs.builder().name("$.text").as("text").build(), + TagFieldArgs.builder().name("$.user").as("user").build(), + TagFieldArgs.builder().name("$.namespace").as("namespace").build(), + TagFieldArgs.builder().name("$.kind").as("kind").build(), + TagFieldArgs.builder().name("$.source_thread").as("source_thread").build(), + NumericFieldArgs.builder().name("$.created_ts").as("created_ts").sortable().build(), + NumericFieldArgs.builder().name("$.hit_count").as("hit_count").sortable().build(), + VectorFieldArgs.builder() + .name("$.embedding").as("embedding") + .hnsw() + .type(VectorFieldArgs.VectorType.FLOAT32) + .dimensions(vectorDim) + .distanceMetric(VectorFieldArgs.DistanceMetric.COSINE) + .build()); + CreateArgs createArgs = CreateArgs.builder() + .on(CreateArgs.TargetType.JSON) + .withPrefix(keyPrefix) + .build(); try { synchronized (txLock) { - sync.dispatch( - ModuleCommand.FT_CREATE, - new StatusOutput<>(ByteArrayCodec.INSTANCE), - args); + sync.ftCreate(indexName, createArgs, schema); } } catch (RedisException ex) { if (!String.valueOf(ex.getMessage()).contains("Index already exists")) { @@ -201,24 +207,28 @@ public void createIndex() { } public void dropIndex(boolean deleteDocuments) { - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes); - if (deleteDocuments) args.add("DD"); try { synchronized (txLock) { - sync.dispatch( - ModuleCommand.FT_DROPINDEX, - new StatusOutput<>(ByteArrayCodec.INSTANCE), - args); + sync.ftDropindex(indexName, deleteDocuments); } } catch (RedisException ex) { - String msg = String.valueOf(ex.getMessage()).toLowerCase(Locale.ROOT); - if (!msg.contains("no such index") && !msg.contains("unknown index name")) { + if (!isMissingIndex(ex)) { throw ex; } } } + /** + * Redis Search phrases the missing-index error differently across + * versions ("Unknown index name", "no such index", and, from Redis + * 8.10, "SEARCH_INDEX_NOT_FOUND Index not found"). + */ + private static boolean isMissingIndex(RedisException ex) { + String msg = String.valueOf(ex.getMessage()).toLowerCase(Locale.ROOT); + return msg.contains("no such index") || msg.contains("unknown index name") + || msg.contains("index not found"); + } + // ------------------------------------------------------------------ // Write // ------------------------------------------------------------------ @@ -277,7 +287,6 @@ public WriteResult remember( doc.put("embedding", embedding); doc.put("created_ts", now); doc.put("hit_count", 0); - byte[] docBytes = doc.toString().getBytes(StandardCharsets.UTF_8); Long ttl = ttlSeconds != null ? ttlSeconds : ttlByKind.get(kind); @@ -290,13 +299,7 @@ public WriteResult remember( TransactionResult txResult; synchronized (txLock) { sync.multi(); - sync.dispatch( - ModuleCommand.JSON_SET, - new StatusOutput<>(ByteArrayCodec.INSTANCE), - new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(keyBytes) - .add("$") - .add(docBytes)); + sync.jsonSet(keyBytes, JsonPath.ROOT_PATH, doc.toString()); if (ttl != null && ttl > 0) { sync.expire(keyBytes, ttl); } @@ -364,30 +367,27 @@ public List listMemories( if (limit <= 0) limit = 100; String filterClause = buildFilterClause(user, namespace, kind); - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add(filterClause) - .add("RETURN").add(7) - .add("user").add("namespace").add("kind").add("source_thread") - .add("text").add("created_ts").add("hit_count") - .add("SORTBY").add("created_ts").add("DESC") - .add("LIMIT").add(0).add(limit) - .add("DIALECT").add(2); + SearchArgs args = SearchArgs.builder() + .returnField("user").returnField("namespace") + .returnField("kind").returnField("source_thread") + .returnField("text").returnField("created_ts") + .returnField("hit_count") + .sortBy(SortByArgs.builder().attribute("created_ts").descending().build()) + .limit(0, limit) + .dialect(QueryDialects.DIALECT2) + .build(); // Hold `txLock` for the whole FT.SEARCH + per-row TTL fetch // so this batch of commands lands together on the connection // and can't get tangled with another helper's MULTI/EXEC. synchronized (txLock) { - List raw; + SearchReply reply; try { - raw = sync.dispatch( - ModuleCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), - args); + reply = sync.ftSearch(indexName, filterClause, args); } catch (RedisException ex) { return new ArrayList<>(); } - List hits = parseAllHits(raw); + List hits = allHits(reply); List out = new ArrayList<>(hits.size()); for (SearchHit hit : hits) { String memoryId = stripPrefix(hit.id); @@ -427,26 +427,23 @@ private List nearest( String knnQuery = filterClause + "=>[KNN " + k + " @embedding $vec AS distance]"; byte[] vecBytes = LocalEmbedder.toBytes(embedding); - CommandArgs args = new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(indexNameBytes) - .add(knnQuery) - .add("RETURN").add(8) - .add("user").add("namespace").add("kind").add("source_thread") - .add("text").add("created_ts").add("hit_count").add("distance") - .add("SORTBY").add("distance").add("ASC") - .add("LIMIT").add(0).add(k) - .add("PARAMS").add(2).add("vec".getBytes(StandardCharsets.UTF_8)).add(vecBytes) - .add("DIALECT").add(2); + SearchArgs args = SearchArgs.builder() + .param("vec", vecBytes) + .returnField("user").returnField("namespace") + .returnField("kind").returnField("source_thread") + .returnField("text").returnField("created_ts") + .returnField("hit_count").returnField("distance") + .sortBy(SortByArgs.builder().attribute("distance").build()) + .limit(0, k) + .dialect(QueryDialects.DIALECT2) + .build(); // Same connection-discipline as `listMemories`: hold the // shared lock for FT.SEARCH + the per-row TTL fetches so the // batch can't tangle with a MULTI/EXEC on another helper. synchronized (txLock) { - List raw = sync.dispatch( - ModuleCommand.FT_SEARCH, - new NestedMultiOutput<>(ByteArrayCodec.INSTANCE), - args); - List hits = parseAllHits(raw); + SearchReply reply = sync.ftSearch(indexName, knnQuery, args); + List hits = allHits(reply); List out = new ArrayList<>(hits.size()); for (SearchHit hit : hits) { // `hit.id` is the full Redis key (e.g. `agent:mem:abc123`). @@ -471,13 +468,7 @@ private void bumpHitCount(String memoryId) { // JSON.NUMINCRBY from landing inside another helper's // open MULTI on the connection. synchronized (txLock) { - sync.dispatch( - ModuleCommand.JSON_NUMINCRBY, - new ValueOutput<>(ByteArrayCodec.INSTANCE), - new CommandArgs<>(ByteArrayCodec.INSTANCE) - .add(memoryKeyBytes(memoryId)) - .add("$.hit_count") - .add(1)); + sync.jsonNumincrby(memoryKeyBytes(memoryId), JsonPath.of("$.hit_count"), 1); } } catch (RedisException ignored) { // memory expired or path not found @@ -544,25 +535,16 @@ private static final class SearchHit { } } - /** Parse every hit from an FT.SEARCH reply, preserving order. */ - private static List parseAllHits(List raw) { + /** Every hit of a typed FT.SEARCH reply, preserving order. */ + private static List allHits(SearchReply reply) { List out = new ArrayList<>(); - if (raw == null || raw.size() < 3) return out; - // index 0 holds the total count; entries follow as - // (key, fields)* pairs. - for (int i = 1; i + 1 < raw.size(); i += 2) { - String id = decode(raw.get(i)); - Map fields = fieldsToMap(raw.get(i + 1)); - out.add(new SearchHit(id, fields)); - } - return out; - } - - private static Map fieldsToMap(Object array) { - Map out = new HashMap<>(); - if (!(array instanceof List list)) return out; - for (int i = 0; i + 1 < list.size(); i += 2) { - out.put(decode(list.get(i)), decode(list.get(i + 1))); + if (reply == null) return out; + for (SearchReply.SearchResult result : reply.getResults()) { + Map fields = new HashMap<>(); + for (Map.Entry field : result.getFields().entrySet()) { + fields.put(field.getKey(), field.getValue().asString()); + } + out.add(new SearchHit(decode(result.getId()), fields)); } return out; }