From 09f72afd66bc208777fc7d266b69da6c9de902e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Mon, 10 Aug 2026 17:41:22 +0800 Subject: [PATCH 1/2] fix(ai): keep text without digits embeddable The value filter feeding the embedding store dropped ordinary prose and kept digit-only noise, so an embedding store silently produced nothing for any text that happened to contain no digit. No request was issued, no error was raised, and the entity was still registered as indexed with an empty vector list, which makes it unrecallable. SearchUtils.isAllAllowedChars returned on the first character *inside* the ignorable set instead of the first one outside it, which is the negation of both its name and its Javadoc. Reproduced on master with an unusable ModelConfig: "no digits here at all" completes without contacting a model and writes zero index lines while logging "Successfully added 1 new index items", whereas the same text with a digit appended does attempt the request. - Correct the predicate and rename it to isAllIgnorableChars. Renaming rather than fixing in place is deliberate: a caller depending on the old meaning now fails to compile instead of silently flipping behaviour. Null and empty become ignorable, since callers use this to decide what to skip and there is nothing in them to index. Both call sites are in SubgraphSemanticPromptFunction and read the same as before, `!isAllIgnorableChars(value)`. - Stop reporting entities with no embeddable text as indexed. indexBatch now warns with a count and an example key, and the summary line reports how many entities actually hold vectors rather than how many were queued. Silence was half of this defect. Measured on the LDBC test dataset before changing anything: of 168 entities, the set wanting vectors is the same before and after the fix, and none of them are absent from the committed index file. LDBC values mix letters and digits, so both the old and the new predicate keep them; the two differ only on values that are entirely digits or punctuation, and on values that are entirely letters. The committed embedding index therefore needs no regeneration and GraphMemoryTest passes unchanged. Tests in IgnorableTextFilterTest, 7 cases: the predicate over values that do and do not carry meaning, the property that keeping a value must not depend on it containing a digit, verbalization of digit free vertex and edge text, and the store either attempting a request or correctly declining to. All 7 fail against the previous behaviour. --- .../geaflow/ai/index/EmbeddingIndexStore.java | 23 ++- .../geaflow/ai/operator/SearchUtils.java | 40 ++-- .../SubgraphSemanticPromptFunction.java | 4 +- .../ai/operator/IgnorableTextFilterTest.java | 194 ++++++++++++++++++ 4 files changed, 243 insertions(+), 18 deletions(-) create mode 100644 geaflow-ai/src/test/java/org/apache/geaflow/ai/operator/IgnorableTextFilterTest.java diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java index 8fae2e1f0..e67329d81 100644 --- a/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java +++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java @@ -172,8 +172,16 @@ public void initStore(GraphAccessor graphAccessor, VerbalizationFunction func, batchEntitiesBuffer.clear(); } - LOGGER.info("Successfully added {} new index items. Total indexed: {}", - addedCount, indexStoreMap.size()); + // Count entries that actually carry vectors, not entities that were queued: an entity with + // no embeddable text is registered with an empty list and must not be reported as indexed. + long withVectors = 0; + for (List vectors : indexStoreMap.values()) { + if (vectors != null && !vectors.isEmpty()) { + withVectors++; + } + } + LOGGER.info("Successfully added {} new index items. Entities holding vectors: {} of {}", + addedCount, withVectors, indexStoreMap.size()); } private List indexBatch(EmbeddingService service, List pendingEntities) { @@ -182,14 +190,25 @@ private List indexBatch(EmbeddingService service, List pend } List pendingTexts = new ArrayList<>(pendingEntities.size()); Map> entity2StartEndPair = new HashMap<>(); + List withoutText = new ArrayList<>(); for (GraphEntity e : pendingEntities) { Integer start = pendingTexts.size(); pendingTexts.addAll(ModelUtils.splitLongText( Constants.EMBEDDING_INDEX_STORE_SPLIT_TEXT_CHUNK_SIZE, verbFunc.verbalize(e).toArray(new String[0]))); Integer end = pendingTexts.size(); + if (start.equals(end)) { + withoutText.add(e); + } entity2StartEndPair.put(e, Pair.of(start, end)); } + if (!withoutText.isEmpty()) { + // Say so rather than reporting these as indexed. An entity whose values are all + // ignorable yields no text, so it gets an empty vector list and can never be recalled. + LOGGER.warn("{} of {} entities have no embeddable text and will hold no vectors, " + + "for example {}", withoutText.size(), pendingEntities.size(), + ModelUtils.getGraphEntityKey(withoutText.get(0))); + } Gson gson = new Gson(); int batchSize = pendingEntities.size(); diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/operator/SearchUtils.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/operator/SearchUtils.java index c60d21af3..350eaaa94 100644 --- a/geaflow-ai/src/main/java/org/apache/geaflow/ai/operator/SearchUtils.java +++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/operator/SearchUtils.java @@ -31,17 +31,17 @@ public class SearchUtils { '*', '#', '-', '?', '`', '{', '}', '[', ']', '(', ')', '>', '<', ':', '/', '.' )); - // Set of allowed characters for validation in isAllAllowedChars - // Includes: digits (0-9), and some common safe symbols - private static final Set IGNORE_CHARS = buildIgnoredChars(); + // Characters that carry no meaning on their own: digits and a few common symbols. + // A value made up entirely of these is not worth indexing or embedding. + private static final Set IGNORABLE_CHARS = buildIgnorableChars(); /** - * Builds the set of allowed characters for input validation. - * Includes alphanumeric characters and selected common symbols. + * Builds the set of characters that carry no meaning on their own. + * Digits and a few common symbols. * - * @return an unmodifiable set of ignored characters + * @return an unmodifiable set of ignorable characters */ - private static Set buildIgnoredChars() { + private static Set buildIgnorableChars() { Set ignored = new HashSet<>(32); // Add digits for (char c = '0'; c <= '9'; c++) { @@ -87,18 +87,30 @@ public static String formatQuery(String query) { } /** - * Checks whether all characters in the given string are within the allowed character set. - * Useful for validating usernames, identifiers, or safe input formats. + * Whether the given value consists entirely of characters that carry no meaning on their own, + * and therefore has nothing worth indexing or embedding. A bare id, a date or a run of + * punctuation is ignorable; anything containing a letter or a CJK character is not. * - * @param str the string to validate - * @return true if all characters are allowed; false otherwise + *

Callers use this to skip values, so an empty or absent value is ignorable too: there is + * nothing in it to index. + * + *

This replaces {@code isAllAllowedChars}, whose loop returned on the first character + * inside the set rather than the first one outside it, making it the negation of both + * its own name and its own documentation. The practical effect was that ordinary prose was + * discarded while digit-only noise was kept, so an embedding store silently produced nothing + * for text that happened to contain no digit. The method is renamed rather than corrected in + * place, so that any caller depending on the previous meaning fails to compile instead of + * silently flipping behaviour. + * + * @param str the value to check + * @return true if every character is ignorable, or the value is null or empty */ - public static boolean isAllAllowedChars(String str) { + public static boolean isAllIgnorableChars(String str) { if (str == null || str.isEmpty()) { - return false; // Consider empty/null invalid; adjust based on use case + return true; } for (char c : str.toCharArray()) { - if (IGNORE_CHARS.contains(c)) { + if (!IGNORABLE_CHARS.contains(c)) { return false; } } diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/verbalization/SubgraphSemanticPromptFunction.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/verbalization/SubgraphSemanticPromptFunction.java index 0b3407c06..55dacc5a6 100644 --- a/geaflow-ai/src/main/java/org/apache/geaflow/ai/verbalization/SubgraphSemanticPromptFunction.java +++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/verbalization/SubgraphSemanticPromptFunction.java @@ -72,12 +72,12 @@ public List verbalize(GraphEntity entity) { if (entity instanceof GraphVertex) { GraphVertex graphVertex = (GraphVertex) entity; return graphVertex.getVertex().getValues().stream() - .filter(str -> !SearchUtils.isAllAllowedChars(str)) + .filter(str -> !SearchUtils.isAllIgnorableChars(str)) .map(SearchUtils::formatQuery).collect(Collectors.toList()); } else if (entity instanceof GraphEdge) { GraphEdge graphEdge = (GraphEdge) entity; return graphEdge.getEdge().getValues().stream() - .filter(str -> !SearchUtils.isAllAllowedChars(str)) + .filter(str -> !SearchUtils.isAllIgnorableChars(str)) .map(SearchUtils::formatQuery).collect(Collectors.toList()); } return new ArrayList<>(); diff --git a/geaflow-ai/src/test/java/org/apache/geaflow/ai/operator/IgnorableTextFilterTest.java b/geaflow-ai/src/test/java/org/apache/geaflow/ai/operator/IgnorableTextFilterTest.java new file mode 100644 index 000000000..1452d4cc5 --- /dev/null +++ b/geaflow-ai/src/test/java/org/apache/geaflow/ai/operator/IgnorableTextFilterTest.java @@ -0,0 +1,194 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.ai.operator; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.geaflow.ai.common.config.Constants; +import org.apache.geaflow.ai.common.model.ModelConfig; +import org.apache.geaflow.ai.graph.GraphEntity; +import org.apache.geaflow.ai.graph.LocalMemoryGraphAccessor; +import org.apache.geaflow.ai.graph.io.Edge; +import org.apache.geaflow.ai.graph.io.EdgeGroup; +import org.apache.geaflow.ai.graph.io.EdgeSchema; +import org.apache.geaflow.ai.graph.io.EntityGroup; +import org.apache.geaflow.ai.graph.io.GraphSchema; +import org.apache.geaflow.ai.graph.io.MemoryGraph; +import org.apache.geaflow.ai.graph.io.Vertex; +import org.apache.geaflow.ai.graph.io.VertexGroup; +import org.apache.geaflow.ai.graph.io.VertexSchema; +import org.apache.geaflow.ai.index.EmbeddingIndexStore; +import org.apache.geaflow.ai.verbalization.SubgraphSemanticPromptFunction; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * The value filter feeding the embedding store must drop values that carry no meaning and keep the + * ones that do. + * + *

It used to do the opposite, because the predicate returned on the first character + * inside the ignorable set rather than the first one outside it. Ordinary prose was + * therefore discarded and digit-only noise kept, and an embedding store silently produced nothing + * for any text that happened to contain no digit: no request was issued, no error was raised, and + * the entity was still registered as indexed with an empty vector list. + */ +public class IgnorableTextFilterTest { + + private static final String LABEL = "chunk"; + private static final String EDGE_LABEL = "rel"; + + @Test + public void testValuesCarryingMeaningAreNotIgnorable() { + for (String value : new String[] { + "the alpaca grazes on the hillside", + "chunk 7 of the report", + "a", + "\u4e2d\u6587\u5185\u5bb9", + "id-42 belongs to Alice"}) { + Assertions.assertFalse(SearchUtils.isAllIgnorableChars(value), + "value carries meaning and must be kept: " + value); + } + } + + @Test + public void testValuesWithoutMeaningAreIgnorable() { + for (String value : new String[] { + "12345", "2024-01-01", "----", "___", "42.0", "%%", null, ""}) { + Assertions.assertTrue(SearchUtils.isAllIgnorableChars(value), + "value carries no meaning and must be dropped: " + value); + } + } + + /** + * The property that matters, stated without reference to the implementation: whether a value is + * kept must not depend on it containing a digit. + */ + @Test + public void testKeepingAValueDoesNotDependOnContainingADigit() { + Assertions.assertEquals( + SearchUtils.isAllIgnorableChars("the alpaca grazes"), + SearchUtils.isAllIgnorableChars("the alpaca grazes 7"), + "adding a digit to prose must not change whether it is indexable"); + } + + @Test + public void testVerbalizationKeepsDigitFreeText() { + LocalMemoryGraphAccessor accessor = buildGraph("no digits here at all", "still none"); + SubgraphSemanticPromptFunction func = new SubgraphSemanticPromptFunction(accessor); + + GraphEntity vertex = accessor.getVertex(LABEL, "v1"); + Assertions.assertEquals(Collections.singletonList("no digits here at all"), + func.verbalize(vertex), + "digit free vertex text must survive verbalization"); + + List edges = new ArrayList<>(accessor.getEdge(EDGE_LABEL, "v1", "v2")); + Assertions.assertEquals(1, edges.size()); + Assertions.assertEquals(Collections.singletonList("plain edge text"), + func.verbalize(edges.get(0)), + "digit free edge text must survive verbalization"); + } + + @Test + public void testVerbalizationDropsValuesWithoutMeaning() { + LocalMemoryGraphAccessor accessor = buildGraph("2024-01-01", "still none"); + SubgraphSemanticPromptFunction func = new SubgraphSemanticPromptFunction(accessor); + Assertions.assertEquals(Collections.emptyList(), + func.verbalize(accessor.getVertex(LABEL, "v1")), + "a value that is only a date carries nothing to embed"); + } + + /** + * End to end on the store, offline: an entity with digit free text must be queued for + * embedding. The unusable model config makes that observable without a service, since reaching + * the request at all fails on the null url. + */ + @Test + public void testStoreRequestsEmbeddingsForDigitFreeText(@TempDir Path tempDir) { + LocalMemoryGraphAccessor accessor = buildGraph("no digits here at all", "still none"); + Path indexFile = tempDir.resolve("index.jsonl"); + EmbeddingIndexStore store = new EmbeddingIndexStore(); + + int retries = Constants.MODEL_CLIENT_RETRY_TIMES; + int interval = Constants.MODEL_CLIENT_RETRY_INTERVAL_MS; + // One attempt is enough to show a request was made, and keeps the test off the retry budget. + Constants.MODEL_CLIENT_RETRY_TIMES = 1; + Constants.MODEL_CLIENT_RETRY_INTERVAL_MS = 1; + try { + Assertions.assertThrows(Throwable.class, () -> store.initStore(accessor, + new SubgraphSemanticPromptFunction(accessor), indexFile.toString(), + new ModelConfig(null, null, null, null)), + "the store must try to embed this text; before the fix it silently did nothing"); + } finally { + Constants.MODEL_CLIENT_RETRY_TIMES = retries; + Constants.MODEL_CLIENT_RETRY_INTERVAL_MS = interval; + } + } + + /** + * The counterpart: when nothing in the graph carries meaning there is genuinely nothing to + * embed, so the store must complete without contacting a model. + */ + @Test + public void testStoreRequestsNothingWhenNoValueCarriesMeaning(@TempDir Path tempDir) + throws Exception { + // The edge value has to be ignorable as well, otherwise there is legitimately something + // to embed and the store is right to try. + LocalMemoryGraphAccessor accessor = buildGraph("2024-01-01", "1999", "42.0"); + Path indexFile = tempDir.resolve("index.jsonl"); + EmbeddingIndexStore store = new EmbeddingIndexStore(); + store.initStore(accessor, new SubgraphSemanticPromptFunction(accessor), + indexFile.toString(), new ModelConfig(null, null, null, null)); + + Assertions.assertEquals(Collections.emptyList(), Files.readAllLines(indexFile)); + Assertions.assertTrue(store.getEntityIndex(accessor.getVertex(LABEL, "v1")).isEmpty()); + Assertions.assertTrue(store.getEntityIndex(accessor.getVertex(LABEL, "v2")).isEmpty()); + } + + private LocalMemoryGraphAccessor buildGraph(String v1Text, String v2Text) { + return buildGraph(v1Text, v2Text, "plain edge text"); + } + + private LocalMemoryGraphAccessor buildGraph(String v1Text, String v2Text, String edgeText) { + GraphSchema schema = new GraphSchema(); + schema.setName("filter_graph"); + VertexSchema vs = new VertexSchema(LABEL, "id", Collections.singletonList("text")); + EdgeSchema es = new EdgeSchema(EDGE_LABEL, "srcId", "dstId", + Collections.singletonList("rel")); + schema.addVertex(vs); + schema.addEdge(es); + + List vertices = new ArrayList<>(Arrays.asList( + new Vertex(LABEL, "v1", Collections.singletonList(v1Text)), + new Vertex(LABEL, "v2", Collections.singletonList(v2Text)))); + List edges = new ArrayList<>(Collections.singletonList( + new Edge(EDGE_LABEL, "v1", "v2", Collections.singletonList(edgeText)))); + Map entities = new HashMap<>(); + entities.put(LABEL, new VertexGroup(vs, vertices)); + entities.put(EDGE_LABEL, new EdgeGroup(es, edges)); + return new LocalMemoryGraphAccessor(new MemoryGraph(schema, entities)); + } +} From b007b1a4dae21cbce2d3bcd6e696d364b05b6bd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Fri, 21 Aug 2026 17:14:09 +0800 Subject: [PATCH 2/2] test(ai): pin down the index over successive runs, and name the empty case Review on #832 raised the concern that an entity holding no embeddable text is registered in indexStoreMap with an empty list, and that a later run would skip it through the containsKey check, so an entity whose value later gains meaning would never be embedded. That does not happen, and the reason is worth a test rather than an argument. initStore assigns a fresh indexStoreMap on every call and repopulates it from the index file alone, and nothing is written to that file for an entity with no text. A later run therefore sees the entity as unseen and embeds it as soon as its value carries meaning. The registration is still doing work within a run: scanEdge returns an edge from both of its endpoints, so the empty entry is what stops the second visit from verbalizing it again. EmbeddingIndexLifecycleTest covers three successive runs against a local embeddings endpoint, so the request and response serialization, the on-disk file and the decision about what to embed are all real, with no model service involved. A value carrying meaning is embedded and persisted while a date-only value is not; an unchanged graph on a second run issues no request and restores its vectors from the file; and when the date is replaced by meaningful text, that entity alone is embedded, the request carries only its text, and the record is appended. Also from review: getEntityIndex returned an empty list both for an entity that was checked and held nothing and for an entity nobody has looked at yet. It still returns the same value, since callers want vectors either way, but the first case now says so at debug level instead of being silent. --- .../geaflow/ai/index/EmbeddingIndexStore.java | 27 ++- .../ai/index/EmbeddingIndexLifecycleTest.java | 225 ++++++++++++++++++ 2 files changed, 243 insertions(+), 9 deletions(-) create mode 100644 geaflow-ai/src/test/java/org/apache/geaflow/ai/index/EmbeddingIndexLifecycleTest.java diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java index e67329d81..ea9e99848 100644 --- a/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java +++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/index/EmbeddingIndexStore.java @@ -262,15 +262,24 @@ private void flushBatchIndex(List newItemStrings, boolean force) { @Override public List getEntityIndex(GraphEntity entity) { - if (entity != null && indexStoreMap.get(entity) != null) { - List resultList = indexStoreMap.get(entity); - List result = new ArrayList<>(); - for (EmbeddingService.EmbeddingResult res : resultList) { - double[] embedding = res.embedding; - result.add(new EmbeddingVector(embedding)); - } - return result; + if (entity == null) { + return Collections.emptyList(); + } + List resultList = indexStoreMap.get(entity); + if (resultList == null) { + return Collections.emptyList(); + } + if (resultList.isEmpty()) { + // An entity that was looked at but held no embeddable text is indistinguishable from an + // entity nobody has looked at yet, since both give an empty result. Say which it was. + LOGGER.debug("Entity {} was checked and holds no embeddable text", + ModelUtils.getGraphEntityKey(entity)); + return Collections.emptyList(); + } + List result = new ArrayList<>(resultList.size()); + for (EmbeddingService.EmbeddingResult res : resultList) { + result.add(new EmbeddingVector(res.embedding)); } - return Collections.emptyList(); + return result; } } diff --git a/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/EmbeddingIndexLifecycleTest.java b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/EmbeddingIndexLifecycleTest.java new file mode 100644 index 000000000..38c7ce9ef --- /dev/null +++ b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/EmbeddingIndexLifecycleTest.java @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.ai.index; + +import com.google.gson.Gson; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import org.apache.geaflow.ai.common.model.EmbeddingResponse; +import org.apache.geaflow.ai.common.model.ModelConfig; +import org.apache.geaflow.ai.graph.GraphEntity; +import org.apache.geaflow.ai.graph.LocalMemoryGraphAccessor; +import org.apache.geaflow.ai.graph.io.Edge; +import org.apache.geaflow.ai.graph.io.EdgeGroup; +import org.apache.geaflow.ai.graph.io.EdgeSchema; +import org.apache.geaflow.ai.graph.io.EntityGroup; +import org.apache.geaflow.ai.graph.io.GraphSchema; +import org.apache.geaflow.ai.graph.io.MemoryGraph; +import org.apache.geaflow.ai.graph.io.Vertex; +import org.apache.geaflow.ai.graph.io.VertexGroup; +import org.apache.geaflow.ai.graph.io.VertexSchema; +import org.apache.geaflow.ai.verbalization.SubgraphSemanticPromptFunction; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * The index over successive runs, against a local embeddings endpoint so the whole path is + * exercised without a model service: request and response serialisation, the on disk index file, + * and what a later run decides to embed. + * + *

Chiefly this pins down what happens to an entity that yields no embeddable text. Such an + * entity is registered with an empty vector list, which is also what makes it skippable within a + * run, since {@code scanEdge} returns an edge from both of its endpoints. Nothing is written to the + * index file for it, and {@code initStore} rebuilds its map from that file alone, so a later run + * treats it as unseen and embeds it once its value carries meaning. That last point is the one worth + * a test: the alternative, never revisiting it, would make the omission permanent. + */ +public class EmbeddingIndexLifecycleTest { + + private static final String LABEL = "chunk"; + private static final String EDGE_LABEL = "rel"; + private static final String MEANINGFUL = "learning without thought is labour lost"; + private static final String IGNORABLE = "2024-01-01"; + private static final String EDGE_TEXT = "the master teaches the disciple"; + private static final int DIMS = 4; + + private HttpServer server; + private final List requestBodies = new CopyOnWriteArrayList<>(); + + @BeforeEach + void startEndpoint() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/v1/embeddings", this::handle); + server.setExecutor(null); + server.start(); + } + + @AfterEach + void stopEndpoint() { + if (server != null) { + server.stop(0); + } + } + + private void handle(HttpExchange exchange) throws IOException { + byte[] requestBytes = readAll(exchange); + String body = new String(requestBytes, StandardCharsets.UTF_8); + requestBodies.add(body); + + String[] inputs = new Gson().fromJson(body, Request.class).input; + EmbeddingResponse response = new EmbeddingResponse(); + response.object = "list"; + response.model = "test-local"; + response.data = new ArrayList<>(); + for (int i = 0; i < inputs.length; i++) { + EmbeddingResponse.EmbeddingVector vector = new EmbeddingResponse.EmbeddingVector(); + vector.index = i; + vector.embedding = deterministicVector(inputs[i]); + response.data.add(vector); + } + + byte[] out = new Gson().toJson(response).getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json; charset=utf-8"); + exchange.sendResponseHeaders(200, out.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(out); + } + } + + private static byte[] readAll(HttpExchange exchange) throws IOException { + java.io.ByteArrayOutputStream buffer = new java.io.ByteArrayOutputStream(); + byte[] chunk = new byte[4096]; + int read; + while ((read = exchange.getRequestBody().read(chunk)) > 0) { + buffer.write(chunk, 0, read); + } + return buffer.toByteArray(); + } + + /** Stable and non-zero, so a stored vector can be told apart from an absent one. */ + private static double[] deterministicVector(String text) { + double[] vector = new double[DIMS]; + for (int i = 0; i < DIMS; i++) { + vector[i] = ((text.hashCode() >> i) & 0xFF) / 255.0 + 0.1; + } + return vector; + } + + private ModelConfig config() { + return new ModelConfig("test-local", + "http://127.0.0.1:" + server.getAddress().getPort(), "/v1/embeddings", "test-token"); + } + + @Test + public void testIndexLifecycleAcrossRuns(@TempDir Path tempDir) throws Exception { + String indexPath = tempDir.resolve("index.jsonl").toString(); + + // First run. One vertex and the edge carry meaning, the other vertex is only a date. + LocalMemoryGraphAccessor first = buildGraph(MEANINGFUL, IGNORABLE); + EmbeddingIndexStore store = new EmbeddingIndexStore(); + store.initStore(first, new SubgraphSemanticPromptFunction(first), indexPath, config()); + + Assertions.assertEquals(1, requestBodies.size(), + "the two values that carry meaning belong to one batch, so one request"); + Assertions.assertFalse(store.getEntityIndex(first.getVertex(LABEL, "v1")).isEmpty(), + "a value carrying meaning must hold a vector"); + Assertions.assertTrue(store.getEntityIndex(first.getVertex(LABEL, "v2")).isEmpty(), + "a value that is only a date has nothing to embed"); + Assertions.assertEquals(2, indexLines(indexPath), + "only the vertex and the edge that carry meaning are persisted"); + + // Second run over an unchanged graph. Everything with vectors comes back from the file, so + // the model is not consulted again. + requestBodies.clear(); + LocalMemoryGraphAccessor second = buildGraph(MEANINGFUL, IGNORABLE); + EmbeddingIndexStore reloaded = new EmbeddingIndexStore(); + reloaded.initStore(second, new SubgraphSemanticPromptFunction(second), indexPath, config()); + + Assertions.assertEquals(0, requestBodies.size(), + "an unchanged graph must not be re-embedded"); + Assertions.assertFalse(reloaded.getEntityIndex(second.getVertex(LABEL, "v1")).isEmpty(), + "vectors must survive a rebuild from the index file"); + Assertions.assertEquals(2, indexLines(indexPath), "no duplicate records appended"); + + // Third run, with the date replaced by text that carries meaning. Nothing was persisted for + // that entity, so this run must embed it, and must leave the rest alone. + requestBodies.clear(); + String nowMeaningful = "the master replied in the temple"; + LocalMemoryGraphAccessor third = buildGraph(MEANINGFUL, nowMeaningful); + EmbeddingIndexStore afterChange = new EmbeddingIndexStore(); + afterChange.initStore(third, new SubgraphSemanticPromptFunction(third), indexPath, config()); + + Assertions.assertEquals(1, requestBodies.size(), + "exactly the entity that gained meaning is embedded"); + Assertions.assertTrue(requestBodies.get(0).contains(nowMeaningful), + "the request must carry the newly meaningful text"); + Assertions.assertFalse(requestBodies.get(0).contains(MEANINGFUL), + "an entity already in the index file must not be embedded again"); + Assertions.assertFalse(afterChange.getEntityIndex(third.getVertex(LABEL, "v2")).isEmpty(), + "an entity that yielded nothing before must not be excluded for good"); + Assertions.assertEquals(3, indexLines(indexPath), "the new record is appended"); + } + + private long indexLines(String indexPath) throws IOException { + return Files.readAllLines(java.nio.file.Paths.get(indexPath), StandardCharsets.UTF_8) + .stream().filter(line -> !line.trim().isEmpty()).count(); + } + + private LocalMemoryGraphAccessor buildGraph(String v1Text, String v2Text) { + GraphSchema schema = new GraphSchema(); + schema.setName("lifecycle"); + VertexSchema vs = new VertexSchema(LABEL, "id", Collections.singletonList("text")); + EdgeSchema es = new EdgeSchema(EDGE_LABEL, "srcId", "dstId", + Collections.singletonList("rel")); + schema.addVertex(vs); + schema.addEdge(es); + + List vertices = new ArrayList<>(Arrays.asList( + new Vertex(LABEL, "v1", Collections.singletonList(v1Text)), + new Vertex(LABEL, "v2", Collections.singletonList(v2Text)))); + List edges = new ArrayList<>(Collections.singletonList( + new Edge(EDGE_LABEL, "v1", "v2", Collections.singletonList(EDGE_TEXT)))); + Map entities = new HashMap<>(); + entities.put(LABEL, new VertexGroup(vs, vertices)); + entities.put(EDGE_LABEL, new EdgeGroup(es, edges)); + return new LocalMemoryGraphAccessor(new MemoryGraph(schema, entities)); + } + + /** Mirrors the request the client sends, enough of it to read the inputs back. */ + private static class Request { + private String[] input; + } +}