This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch investigate-and-fix-camel-23953-camel-langchain4j in repository https://gitbox.apache.org/repos/asf/camel.git
commit 35f8eedf693141453a3d6412bae6e32bbefe2f80 Author: Guillaume Nodet <[email protected]> AuthorDate: Fri Jul 31 11:29:52 2026 +0200 CAMEL-23953: Add batch operations for langchain4j embeddings and embeddingstore - LangChain4jEmbeddingsProducer: support List<TextSegment> body for embedAll() batch embedding in a single API call - LangChain4jEmbeddingStoreProducer ADD: support addAll() batch variants, caller-supplied IDs via EMBEDDING_ID/EMBEDDING_IDS headers - LangChain4jEmbeddingStoreProducer REMOVE: support removeAll(Collection), removeAll(Filter) via existing FILTER header, and removeAll() to clear store - Add EMBEDDINGS header constant for batch embedding results - Add EMBEDDING_ID and EMBEDDING_IDS header constants for caller-supplied IDs - Add unit tests for all batch operations (11 new tests) - Update documentation for both components Co-Authored-By: Claude Opus 4.6 <[email protected]> --- .../embeddings/langchain4j-embeddings.json | 5 +- .../docs/langchain4j-embeddings-component.adoc | 70 +++++ .../embeddings/LangChain4jEmbeddingsHeaders.java | 4 + .../embeddings/LangChain4jEmbeddingsProducer.java | 43 ++- .../embeddings/LangChain4jEmbeddingsBatchTest.java | 99 ++++++ .../embeddingstore/langchain4j-embeddingstore.json | 4 +- .../docs/langchain4j-embeddingstore-component.adoc | 70 ++++- .../LangChain4jEmbeddingStoreHeaders.java | 13 +- .../LangChain4jEmbeddingStoreProducer.java | 119 ++++++-- ...ngChain4jEmbeddingStoreBatchOperationsTest.java | 335 +++++++++++++++++++++ 10 files changed, 734 insertions(+), 28 deletions(-) diff --git a/components/camel-ai/camel-langchain4j-embeddings/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddings/langchain4j-embeddings.json b/components/camel-ai/camel-langchain4j-embeddings/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddings/langchain4j-embeddings.json index e307b3c21cd4..0d25ae090ff6 100644 --- a/components/camel-ai/camel-langchain4j-embeddings/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddings/langchain4j-embeddings.json +++ b/components/camel-ai/camel-langchain4j-embeddings/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddings/langchain4j-embeddings.json @@ -35,8 +35,9 @@ "CamelLangChain4jEmbeddingsOutputTokenCount": { "index": 2, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Output Token Count.", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#OUTPUT_TOKEN_COUNT" }, "CamelLangChain4jEmbeddingsTotalTokenCount": { "index": 3, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Total Token Count.", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#TOTAL_TOKEN_COUNT" }, "CamelLangChain4jEmbeddingsEmbedding": { "index": 4, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "dev.langchain4j.data.embedding.Embedding", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Embedding representation of a text", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#EMBEDDING" }, - "CamelLangChain4jEmbeddingsVector": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "float[]", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A dense vector embedding of a text", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#VECTOR" }, - "CamelLangChain4jEmbeddingsTextSegment": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": " dev.langchain4j.data.segment.TextSegment", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A TextSegment representation of the vector embedding input text", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#TEXT_SEGMENT" } + "CamelLangChain4jEmbeddingsEmbeddings": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "java.util.List<dev.langchain4j.data.embedding.Embedding>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "List of embeddings from a batch embedAll operation", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#EMBEDDINGS" }, + "CamelLangChain4jEmbeddingsVector": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "float[]", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A dense vector embedding of a text", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#VECTOR" }, + "CamelLangChain4jEmbeddingsTextSegment": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": " dev.langchain4j.data.segment.TextSegment", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A TextSegment representation of the vector embedding input text", "constantName": "org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders#TEXT_SEGMENT" } }, "properties": { "embeddingId": { "index": 0, "kind": "path", "displayName": "Embedding Id", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The id" }, diff --git a/components/camel-ai/camel-langchain4j-embeddings/src/main/docs/langchain4j-embeddings-component.adoc b/components/camel-ai/camel-langchain4j-embeddings/src/main/docs/langchain4j-embeddings-component.adoc index a145699c1524..38b456a868ce 100644 --- a/components/camel-ai/camel-langchain4j-embeddings/src/main/docs/langchain4j-embeddings-component.adoc +++ b/components/camel-ai/camel-langchain4j-embeddings/src/main/docs/langchain4j-embeddings-component.adoc @@ -34,6 +34,76 @@ include::partial$component-endpoint-headers.adoc[] == Usage +=== Single vs Batch Embedding + +The component supports both single and batch embedding operations: + +* **Single**: Send a `String` or `TextSegment` body to compute a single embedding. The result is available via the `CamelLangChain4jEmbeddingsEmbedding` header. +* **Batch**: Send a `List<String>` or `List<TextSegment>` body to embed all items in one API call using `EmbeddingModel.embedAll()`. The resulting `List<Embedding>` is set as the message body and also available via the `CamelLangChain4jEmbeddingsEmbeddings` header. + +Batch embedding is significantly more efficient for RAG ingestion pipelines where many text segments need to be embedded at once. + +==== Single Embedding Example + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:embed") + .setBody(constant("Hello world")) + .to("langchain4j-embeddings:embed"); +// Result: Embedding in the CamelLangChain4jEmbeddingsEmbedding header +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:embed + steps: + - setBody: + constant: "Hello world" + - to: + uri: langchain4j-embeddings:embed +---- +==== + +==== Batch Embedding Example + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:embed-batch") + .setBody(constant(List.of("Hello", "World", "Test"))) + .to("langchain4j-embeddings:embed"); +// Result: List<Embedding> as body and in the CamelLangChain4jEmbeddingsEmbeddings header +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:embed-batch + steps: + - setBody: + constant: + - "Hello" + - "World" + - "Test" + - to: + uri: langchain4j-embeddings:embed +---- +==== + === Using Embedding Models The Camel LangChain4j embeddings component provides support for generating embeddings using various embedding models supported by https://docs.langchain4j.dev/[LangChain4j]. diff --git a/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsHeaders.java b/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsHeaders.java index 396808aaa691..f52d827751e5 100644 --- a/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsHeaders.java +++ b/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsHeaders.java @@ -35,6 +35,10 @@ public class LangChain4jEmbeddingsHeaders { @Metadata(description = "Embedding representation of a text", javaType = "dev.langchain4j.data.embedding.Embedding") public static final String EMBEDDING = CamelLangchain4jAttributes.CAMEL_LANGCHAIN4J_EMBEDDING; + @Metadata(description = "List of embeddings from a batch embedAll operation", + javaType = "java.util.List<dev.langchain4j.data.embedding.Embedding>") + public static final String EMBEDDINGS = "CamelLangChain4jEmbeddingsEmbeddings"; + @Metadata(description = "A dense vector embedding of a text", javaType = "float[]") public static final String VECTOR = CamelLangchain4jAttributes.CAMEL_LANGCHAIN4J_EMBEDDING_VECTOR; diff --git a/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsProducer.java b/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsProducer.java index d1ad52fbf023..e42ca1d83501 100644 --- a/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsProducer.java +++ b/components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsProducer.java @@ -16,6 +16,8 @@ */ package org.apache.camel.component.langchain4j.embeddings; +import java.util.List; + import dev.langchain4j.data.embedding.Embedding; import dev.langchain4j.data.segment.TextSegment; import dev.langchain4j.model.embedding.EmbeddingModel; @@ -35,11 +37,22 @@ public class LangChain4jEmbeddingsProducer extends DefaultProducer { } @Override + @SuppressWarnings("unchecked") public void process(Exchange exchange) throws Exception { - final TextSegment in = exchange.getMessage().getMandatoryBody(TextSegment.class); final EmbeddingModel model = getEndpoint().getConfiguration().getEmbeddingModel(); - final Response<Embedding> result = model.embed(in); final Message message = exchange.getMessage(); + Object body = message.getBody(); + + if (body instanceof List) { + processBatch(exchange, model, message, (List<Object>) body); + } else { + processSingle(exchange, model, message); + } + } + + private void processSingle(Exchange exchange, EmbeddingModel model, Message message) throws Exception { + final TextSegment in = exchange.getMessage().getMandatoryBody(TextSegment.class); + final Response<Embedding> result = model.embed(in); if (result.finishReason() != null) { message.setHeader(LangChain4jEmbeddingsHeaders.FINISH_REASON, result.finishReason()); @@ -55,4 +68,30 @@ public class LangChain4jEmbeddingsProducer extends DefaultProducer { message.setHeader(LangChain4jEmbeddingsHeaders.TEXT_SEGMENT, in); message.setHeader(LangChain4jEmbeddingsHeaders.EMBEDDING, result.content()); } + + private void processBatch(Exchange exchange, EmbeddingModel model, Message message, List<Object> bodyList) + throws Exception { + // Convert each element to TextSegment using the type converter + List<TextSegment> segments = new java.util.ArrayList<>(bodyList.size()); + for (Object item : bodyList) { + TextSegment segment = exchange.getContext().getTypeConverter().mandatoryConvertTo(TextSegment.class, item); + segments.add(segment); + } + + final Response<List<Embedding>> result = model.embedAll(segments); + + if (result.finishReason() != null) { + message.setHeader(LangChain4jEmbeddingsHeaders.FINISH_REASON, result.finishReason()); + } + + if (result.tokenUsage() != null) { + message.setHeader(LangChain4jEmbeddingsHeaders.INPUT_TOKEN_COUNT, result.tokenUsage().inputTokenCount()); + message.setHeader(LangChain4jEmbeddingsHeaders.OUTPUT_TOKEN_COUNT, result.tokenUsage().outputTokenCount()); + message.setHeader(LangChain4jEmbeddingsHeaders.TOTAL_TOKEN_COUNT, result.tokenUsage().totalTokenCount()); + } + + List<Embedding> embeddings = result.content(); + message.setHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, embeddings); + message.setBody(embeddings); + } } diff --git a/components/camel-ai/camel-langchain4j-embeddings/src/test/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsBatchTest.java b/components/camel-ai/camel-langchain4j-embeddings/src/test/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsBatchTest.java new file mode 100644 index 000000000000..4862d2055809 --- /dev/null +++ b/components/camel-ai/camel-langchain4j-embeddings/src/test/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsBatchTest.java @@ -0,0 +1,99 @@ +/* + * 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.camel.component.langchain4j.embeddings; + +import java.util.Arrays; +import java.util.List; + +import dev.langchain4j.data.embedding.Embedding; +import dev.langchain4j.model.embedding.onnx.allminilml6v2.AllMiniLmL6V2EmbeddingModel; +import org.apache.camel.CamelContext; +import org.apache.camel.Message; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class LangChain4jEmbeddingsBatchTest extends CamelTestSupport { + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + + LangChain4jEmbeddingsComponent component + = context.getComponent(LangChain4jEmbeddings.SCHEME, LangChain4jEmbeddingsComponent.class); + + component.getConfiguration().setEmbeddingModel(new AllMiniLmL6V2EmbeddingModel()); + + return context; + } + + @Test + @DisplayName("Batch embedAll with List<String> body produces List<Embedding> result") + void batchEmbedAllWithStringList() { + List<String> texts = Arrays.asList("hello", "world", "test"); + + Message result = fluentTemplate.to("langchain4j-embeddings:batch") + .withBody(texts) + .request(Message.class); + + @SuppressWarnings("unchecked") + List<Embedding> embeddings = result.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, List.class); + assertThat(embeddings).hasSize(3); + assertThat(embeddings.get(0).vector()).hasSize(384); + assertThat(embeddings.get(1).vector()).hasSize(384); + assertThat(embeddings.get(2).vector()).hasSize(384); + + // Body should also be set to the list of embeddings + assertThat(result.getBody()).isInstanceOf(List.class); + } + + @Test + @DisplayName("Single string body still works as before") + void singleStringBodyStillWorks() { + Message result = fluentTemplate.to("langchain4j-embeddings:single") + .withBody("hello") + .request(Message.class); + + Embedding embedding = result.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDING, Embedding.class); + assertThat(embedding).isNotNull(); + assertThat(embedding.vector()).hasSize(384); + + // EMBEDDINGS header should NOT be set for single operation + assertThat(result.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS)).isNull(); + } + + @Test + @DisplayName("Batch embedAll preserves token usage headers") + void batchEmbedAllPreservesTokenUsage() { + List<String> texts = Arrays.asList("hi", "bye"); + + Message result = fluentTemplate.to("langchain4j-embeddings:batch-tokens") + .withBody(texts) + .request(Message.class); + + @SuppressWarnings("unchecked") + List<Embedding> embeddings = result.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, List.class); + assertThat(embeddings).hasSize(2); + + // Token usage may or may not be provided by the model, but headers should be set if available + // AllMiniLmL6V2 provides token usage, so verify it's there + Integer inputTokens = result.getHeader(LangChain4jEmbeddingsHeaders.INPUT_TOKEN_COUNT, Integer.class); + assertThat(inputTokens).isNotNull(); + } +} diff --git a/components/camel-ai/camel-langchain4j-embeddingstore/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddingstore/langchain4j-embeddingstore.json b/components/camel-ai/camel-langchain4j-embeddingstore/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddingstore/langchain4j-embeddingstore.json index f28886c6612e..4484969e848f 100644 --- a/components/camel-ai/camel-langchain4j-embeddingstore/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddingstore/langchain4j-embeddingstore.json +++ b/components/camel-ai/camel-langchain4j-embeddingstore/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/embeddingstore/langchain4j-embeddingstore.json @@ -38,7 +38,9 @@ "CamelLangchain4jEmbeddingStoreAction": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "enum": [ "ADD", "REMOVE", "SEARCH" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The action to be performed.", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#ACTION" }, "CamelLangchain4jEmbeddingStoreMaxResults": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "5", "description": "Maximum number of search results to return", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#MAX_RESULTS" }, "CamelLangchain4jEmbeddingStoreMinScore": { "index": 2, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Minimum similarity score for search results", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#MIN_SCORE" }, - "CamelLangchain4jEmbeddingStoreFilter": { "index": 3, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "dev.langchain4j.store.embedding.filter.Filter", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Search filter for metadata-based constraints", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#FILTER" } + "CamelLangchain4jEmbeddingStoreFilter": { "index": 3, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "dev.langchain4j.store.embedding.filter.Filter", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Filter for metadata-based constraints (used by SEARCH and REMOVE operations)", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#FILTER" }, + "CamelLangchain4jEmbeddingStoreEmbeddingId": { "index": 4, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Caller-supplied embedding ID for single ADD operations", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#EMBEDDING_ID" }, + "CamelLangchain4jEmbeddingStoreEmbeddingIds": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "java.util.List<java.lang.String>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Caller-supplied embedding IDs for batch ADD operations", "constantName": "org.apache.camel.component.langchain4j.embeddingstore.LangChain4jEmbeddingStoreHeaders#EMBEDDING_IDS" } }, "properties": { "embeddingStoreId": { "index": 0, "kind": "path", "displayName": "Embedding Store Id", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The id of the embedding store" }, diff --git a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/docs/langchain4j-embeddingstore-component.adoc b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/docs/langchain4j-embeddingstore-component.adoc index 942444fb4208..ebd833950634 100644 --- a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/docs/langchain4j-embeddingstore-component.adoc +++ b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/docs/langchain4j-embeddingstore-component.adoc @@ -30,8 +30,8 @@ The LangChain4j Embedding Store component offers the following key features: The component supports three main operations controlled by the `CamelLangchain4jEmbeddingStoreAction` header: -* **ADD** - Store embeddings with optional text segments and metadata -* **REMOVE** - Delete embeddings by their unique identifier +* **ADD** - Store embeddings with optional text segments and metadata. Supports single and batch operations, as well as caller-supplied IDs. +* **REMOVE** - Delete embeddings by single ID, collection of IDs, metadata filter, or clear the entire store. * **SEARCH** - Perform similarity search with configurable filters and scoring == URI format @@ -106,6 +106,34 @@ YAML:: The response body contains the generated embedding ID. +==== ADD with Caller-Supplied ID + +You can provide your own embedding ID using the `CamelLangchain4jEmbeddingStoreEmbeddingId` header: + +[source,java] +---- +from("direct:store-with-id") + .to("langchain4j-embeddings:embed") + .setHeader("CamelLangchain4jEmbeddingStoreEmbeddingId", constant("my-custom-id")) + .to("langchain4j-embeddingstore:myStore?action=ADD"); +---- + +==== Batch ADD (addAll) + +For RAG ingestion pipelines, batch adding is significantly more efficient. Send a `List<Embedding>` via the `CamelLangChain4jEmbeddingsEmbeddings` header: + +[source,java] +---- +from("direct:batch-store") + // Assuming the body is a List<String> or List<TextSegment> + .to("langchain4j-embeddings:embed") + // After batch embed, the CamelLangChain4jEmbeddingsEmbeddings header contains List<Embedding> + .to("langchain4j-embeddingstore:myStore?action=ADD"); +// Body: List<String> of generated IDs +---- + +You can also supply caller-supplied IDs for the batch via the `CamelLangchain4jEmbeddingStoreEmbeddingIds` header. + === Searching Embeddings (SEARCH Operation) Perform similarity search to find relevant content: @@ -181,7 +209,11 @@ YAML:: === Removing Embeddings (REMOVE Operation) -Delete embeddings by their ID: +The REMOVE operation supports multiple strategies: + +==== Remove by Single ID + +Delete a single embedding by its ID (body is a `String`): [tabs] ==== @@ -213,6 +245,38 @@ YAML:: ---- ==== +==== Remove by Collection of IDs + +Delete multiple embeddings by passing a `Collection<String>` as the body: + +[source,java] +---- +from("direct:remove-batch") + .setBody(constant(List.of("id-1", "id-2", "id-3"))) + .to("langchain4j-embeddingstore:myStore?action=REMOVE"); +---- + +==== Remove by Metadata Filter + +Delete all embeddings matching a metadata filter using the `CamelLangchain4jEmbeddingStoreFilter` header: + +[source,java] +---- +from("direct:remove-by-filter") + .setHeader("CamelLangchain4jEmbeddingStoreFilter", constant(metadataFilter)) + .to("langchain4j-embeddingstore:myStore?action=REMOVE"); +---- + +==== Clear All Embeddings + +Remove all embeddings from the store by sending a REMOVE with no body and no filter: + +[source,java] +---- +from("direct:clear-store") + .to("langchain4j-embeddingstore:myStore?action=REMOVE"); +---- + === Complete RAG Pipeline Example A complete example showing document ingestion and retrieval: diff --git a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreHeaders.java b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreHeaders.java index e40318d9db5c..6fbbee082117 100644 --- a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreHeaders.java +++ b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreHeaders.java @@ -41,8 +41,17 @@ public class LangChain4jEmbeddingStoreHeaders { @Metadata(description = "Minimum similarity score for search results", javaType = "Double") public static final String MIN_SCORE = "CamelLangchain4jEmbeddingStoreMinScore"; - /** Filter to apply to search operations for metadata-based filtering */ - @Metadata(description = "Search filter for metadata-based constraints", + /** Filter to apply to search and remove operations for metadata-based filtering */ + @Metadata(description = "Filter for metadata-based constraints (used by SEARCH and REMOVE operations)", javaType = "dev.langchain4j.store.embedding.filter.Filter") public static final String FILTER = "CamelLangchain4jEmbeddingStoreFilter"; + + /** Caller-supplied embedding ID for single ADD operations */ + @Metadata(description = "Caller-supplied embedding ID for single ADD operations", javaType = "String") + public static final String EMBEDDING_ID = "CamelLangchain4jEmbeddingStoreEmbeddingId"; + + /** Caller-supplied embedding IDs for batch ADD operations */ + @Metadata(description = "Caller-supplied embedding IDs for batch ADD operations", + javaType = "java.util.List<java.lang.String>") + public static final String EMBEDDING_IDS = "CamelLangchain4jEmbeddingStoreEmbeddingIds"; } diff --git a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java index 257583509ed9..24498727bfc6 100644 --- a/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java +++ b/components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java @@ -16,6 +16,7 @@ */ package org.apache.camel.component.langchain4j.embeddingstore; +import java.util.Collection; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.stream.Collectors; @@ -109,26 +110,50 @@ public class LangChain4jEmbeddingStoreProducer extends DefaultProducer { } /** - * Adds an embedding to the store with optional text segment. + * Adds embeddings to the store with optional text segments and caller-supplied IDs. * * <p> - * Expects the following headers: + * Supports both single and batch operations: + * </p> + * + * <p> + * <b>Single operation</b> - when the {@code CamelLangChain4jEmbeddingsEmbedding} header contains a single + * {@link Embedding}: * </p> * <ul> - * <li>{@code CamelLangchain4jEmbeddingEmbedding} - The embedding vector (required)</li> - * <li>{@code CamelLangchain4jEmbeddingTextSegment} - Associated text segment (optional)</li> + * <li>With caller-supplied ID header ({@code CamelLangchain4jEmbeddingStoreEmbeddingId}): calls + * {@code add(id, embedding)}</li> + * <li>With text segment header: calls {@code add(embedding, textSegment)}</li> + * <li>Without text segment: calls {@code add(embedding)}</li> * </ul> * * <p> - * Returns the generated embedding ID in the message body. + * <b>Batch operation</b> - when the {@code CamelLangChain4jEmbeddingsEmbeddings} header contains a + * {@code List<Embedding>}: * </p> + * <ul> + * <li>With IDs header ({@code CamelLangchain4jEmbeddingStoreEmbeddingIds}) and text segments body: calls + * {@code addAll(ids, embeddings, textSegments)}</li> + * <li>With text segments body: calls {@code addAll(embeddings, textSegments)}</li> + * <li>Without text segments: calls {@code addAll(embeddings)}</li> + * </ul> * * @param exchange the Camel exchange containing the embedding data * @throws Exception if the add operation fails */ + @SuppressWarnings("unchecked") private void add(Exchange exchange) throws Exception { final Message in = exchange.getMessage(); + EmbeddingStore<TextSegment> store = getEndpoint().getConfiguration().getEmbeddingStore(); + + // Check for batch embeddings header first + List<Embedding> embeddings = in.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, List.class); + if (embeddings != null) { + addBatch(in, store, embeddings); + return; + } + // Single embedding path if (in.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDING) == null) { throw new NoSuchHeaderException( "The embedding is a required header for ADD operations", exchange, @@ -136,36 +161,94 @@ public class LangChain4jEmbeddingStoreProducer extends DefaultProducer { } Embedding embedding = in.getHeader(LangChain4jEmbeddingsHeaders.EMBEDDING, Embedding.class); + + // Check for caller-supplied ID + String callerId = in.getHeader(LangChain4jEmbeddingStoreHeaders.EMBEDDING_ID, String.class); String id; - if (in.getHeader(LangChain4jEmbeddingsHeaders.TEXT_SEGMENT) != null) { + if (callerId != null) { + store.add(callerId, embedding); + id = callerId; + } else if (in.getHeader(LangChain4jEmbeddingsHeaders.TEXT_SEGMENT) != null) { TextSegment text = in.getHeader(LangChain4jEmbeddingsHeaders.TEXT_SEGMENT, TextSegment.class); - id = getEndpoint().getConfiguration().getEmbeddingStore().add(embedding, text); + id = store.add(embedding, text); } else { - id = getEndpoint().getConfiguration().getEmbeddingStore().add(embedding); + id = store.add(embedding); } - Message out = exchange.getMessage(); - out.setBody(id); + in.setBody(id); + } + + @SuppressWarnings("unchecked") + private void addBatch(Message in, EmbeddingStore<TextSegment> store, List<Embedding> embeddings) { + List<String> callerIds = in.getHeader(LangChain4jEmbeddingStoreHeaders.EMBEDDING_IDS, List.class); + Object body = in.getBody(); + List<TextSegment> textSegments = null; + + if (body instanceof List && !((List<?>) body).isEmpty() && ((List<?>) body).get(0) instanceof TextSegment) { + textSegments = (List<TextSegment>) body; + } + + List<String> ids; + if (callerIds != null && textSegments != null) { + store.addAll(callerIds, embeddings, textSegments); + ids = callerIds; + } else if (textSegments != null) { + ids = store.addAll(embeddings, textSegments); + } else { + ids = store.addAll(embeddings); + } + + in.setBody(ids); } /** - * Removes an embedding from the store by its ID. + * Removes embeddings from the store. Supports multiple removal strategies: * - * <p> - * Expects the embedding ID as the message body (String). - * </p> + * <ul> + * <li><b>By filter</b>: when the {@code CamelLangchain4jEmbeddingStoreFilter} header is set, removes all embeddings + * matching the filter via {@code removeAll(Filter)}</li> + * <li><b>By ID list</b>: when the body is a {@code Collection<String>}, removes all specified embeddings via + * {@code removeAll(Collection)}</li> + * <li><b>By single ID</b>: when the body is a single {@code String}, removes that embedding via + * {@code remove(id)}</li> + * <li><b>Clear all</b>: when the body is null/empty and no filter is set, removes all embeddings from the store via + * {@code removeAll()}</li> + * </ul> * - * @param exchange the Camel exchange containing the embedding ID to remove + * @param exchange the Camel exchange containing removal parameters * @throws Exception if the remove operation fails */ + @SuppressWarnings("unchecked") private void remove(Exchange exchange) throws Exception { final Message in = exchange.getMessage(); - String id = in.getBody(String.class); + EmbeddingStore<TextSegment> store = getEndpoint().getConfiguration().getEmbeddingStore(); + + // Check for filter-based removal first + Filter filter = in.getHeader(LangChain4jEmbeddingStoreHeaders.FILTER, Filter.class); + if (filter != null) { + store.removeAll(filter); + return; + } - getEndpoint().getConfiguration().getEmbeddingStore().remove(id); + Object body = in.getBody(); - Message out = exchange.getMessage(); + // Batch removal by collection of IDs + if (body instanceof Collection) { + Collection<String> ids = (Collection<String>) body; + store.removeAll(ids); + return; + } + + // Single ID removal + String id = in.getBody(String.class); + if (id != null && !id.isEmpty()) { + store.remove(id); + return; + } + + // No filter, no IDs — clear the entire store + store.removeAll(); } /** diff --git a/components/camel-ai/camel-langchain4j-embeddingstore/src/test/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreBatchOperationsTest.java b/components/camel-ai/camel-langchain4j-embeddingstore/src/test/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreBatchOperationsTest.java new file mode 100644 index 000000000000..97be9a9e3322 --- /dev/null +++ b/components/camel-ai/camel-langchain4j-embeddingstore/src/test/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreBatchOperationsTest.java @@ -0,0 +1,335 @@ +/* + * 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.camel.component.langchain4j.embeddingstore; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +import dev.langchain4j.data.embedding.Embedding; +import dev.langchain4j.data.segment.TextSegment; +import dev.langchain4j.store.embedding.filter.Filter; +import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore; +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.component.langchain4j.embeddings.LangChain4jEmbeddingsHeaders; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class LangChain4jEmbeddingStoreBatchOperationsTest extends CamelTestSupport { + + private RecordingEmbeddingStore embeddingStore; + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + embeddingStore = new RecordingEmbeddingStore(); + + LangChain4jEmbeddingStoreComponent component = context.getComponent( + LangChain4jEmbeddingStore.SCHEME, LangChain4jEmbeddingStoreComponent.class); + component.getConfiguration().setEmbeddingStore(embeddingStore); + + return context; + } + + // ---- ADD: caller-supplied ID ---- + + @Test + @DisplayName("ADD with caller-supplied ID uses add(id, embedding)") + void addWithCallerSuppliedId() { + Embedding embedding = Embedding.from(new float[] { 0.1f, 0.2f, 0.3f }); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.ADD) + .withHeader(LangChain4jEmbeddingsHeaders.EMBEDDING, embedding) + .withHeader(LangChain4jEmbeddingStoreHeaders.EMBEDDING_ID, "my-custom-id") + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getBody(String.class)).isEqualTo("my-custom-id"); + assertThat(embeddingStore.getAddWithIdInvocations()).isEqualTo(1); + assertThat(embeddingStore.getLastCallerSuppliedId()).isEqualTo("my-custom-id"); + } + + // ---- ADD: batch operations ---- + + @Test + @DisplayName("ADD with EMBEDDINGS header and no body calls addAll(embeddings)") + void addBatchEmbeddingsOnly() { + List<Embedding> embeddings = Arrays.asList( + Embedding.from(new float[] { 0.1f, 0.2f }), + Embedding.from(new float[] { 0.3f, 0.4f })); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.ADD) + .withHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, embeddings) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getAddAllInvocations()).isEqualTo(1); + + @SuppressWarnings("unchecked") + List<String> ids = result.getMessage().getBody(List.class); + assertThat(ids).hasSize(2); + } + + @Test + @DisplayName("ADD with EMBEDDINGS header and TextSegment body calls addAll(embeddings, textSegments)") + void addBatchEmbeddingsWithTextSegments() { + List<Embedding> embeddings = Arrays.asList( + Embedding.from(new float[] { 0.1f, 0.2f }), + Embedding.from(new float[] { 0.3f, 0.4f })); + List<TextSegment> segments = Arrays.asList( + TextSegment.from("hello"), + TextSegment.from("world")); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.ADD) + .withHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, embeddings) + .withBody(segments) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getAddAllWithSegmentsInvocations()).isEqualTo(1); + + @SuppressWarnings("unchecked") + List<String> ids = result.getMessage().getBody(List.class); + assertThat(ids).hasSize(2); + } + + @Test + @DisplayName("ADD with EMBEDDINGS header, caller IDs, and TextSegment body calls addAll(ids, embeddings, textSegments)") + void addBatchWithCallerIdsAndTextSegments() { + List<String> callerIds = Arrays.asList("id-1", "id-2"); + List<Embedding> embeddings = Arrays.asList( + Embedding.from(new float[] { 0.1f, 0.2f }), + Embedding.from(new float[] { 0.3f, 0.4f })); + List<TextSegment> segments = Arrays.asList( + TextSegment.from("hello"), + TextSegment.from("world")); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.ADD) + .withHeader(LangChain4jEmbeddingsHeaders.EMBEDDINGS, embeddings) + .withHeader(LangChain4jEmbeddingStoreHeaders.EMBEDDING_IDS, callerIds) + .withBody(segments) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getAddAllWithIdsInvocations()).isEqualTo(1); + + @SuppressWarnings("unchecked") + List<String> ids = result.getMessage().getBody(List.class); + assertThat(ids).containsExactly("id-1", "id-2"); + } + + // ---- REMOVE: batch by ID list ---- + + @Test + @DisplayName("REMOVE with Collection<String> body calls removeAll(ids)") + void removeBatchByIds() { + // Pre-populate + String id1 = embeddingStore.add(Embedding.from(new float[] { 0.1f })); + String id2 = embeddingStore.add(Embedding.from(new float[] { 0.2f })); + embeddingStore.resetCounters(); + + List<String> ids = Arrays.asList(id1, id2); + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.REMOVE) + .withBody(ids) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getRemoveAllByIdsInvocations()).isEqualTo(1); + } + + // ---- REMOVE: by filter ---- + + @Test + @DisplayName("REMOVE with FILTER header calls removeAll(filter)") + void removeByFilter() { + Filter filter = new Filter() { + @Override + public boolean test(Object object) { + return true; + } + }; + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.REMOVE) + .withHeader(LangChain4jEmbeddingStoreHeaders.FILTER, filter) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getRemoveAllByFilterInvocations()).isEqualTo(1); + } + + // ---- REMOVE: clear all ---- + + @Test + @DisplayName("REMOVE with null body and no filter calls removeAll()") + void removeClearAll() { + // Pre-populate + embeddingStore.add(Embedding.from(new float[] { 0.1f })); + embeddingStore.resetCounters(); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.REMOVE) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getRemoveAllInvocations()).isEqualTo(1); + } + + // ---- REMOVE: single ID still works ---- + + @Test + @DisplayName("REMOVE with single String body still calls remove(id)") + void removeSingleIdStillWorks() { + String id = embeddingStore.add(Embedding.from(new float[] { 0.9f })); + embeddingStore.resetCounters(); + + Exchange result = fluentTemplate.to("langchain4j-embeddingstore:test") + .withHeader(LangChain4jEmbeddingStoreHeaders.ACTION, LangChain4jEmbeddingStoreAction.REMOVE) + .withBody(id) + .request(Exchange.class); + + assertThat(result.getException()).isNull(); + assertThat(embeddingStore.getRemoveSingleInvocations()).isEqualTo(1); + } + + /** + * Extended recording store that tracks which method variants are called. + */ + private static final class RecordingEmbeddingStore extends InMemoryEmbeddingStore<TextSegment> { + + private int addWithIdInvocations; + private int addAllInvocations; + private int addAllWithSegmentsInvocations; + private int addAllWithIdsInvocations; + private int removeSingleInvocations; + private int removeAllByIdsInvocations; + private int removeAllByFilterInvocations; + private int removeAllInvocations; + private String lastCallerSuppliedId; + + @Override + public String add(Embedding embedding) { + return super.add(embedding); + } + + @Override + public void add(String id, Embedding embedding) { + addWithIdInvocations++; + lastCallerSuppliedId = id; + super.add(id, embedding); + } + + @Override + public List<String> addAll(List<Embedding> embeddings) { + addAllInvocations++; + return super.addAll(embeddings); + } + + @Override + public List<String> addAll(List<Embedding> embeddings, List<TextSegment> embedded) { + addAllWithSegmentsInvocations++; + return super.addAll(embeddings, embedded); + } + + @Override + public void addAll(List<String> ids, List<Embedding> embeddings, List<TextSegment> embedded) { + addAllWithIdsInvocations++; + super.addAll(ids, embeddings, embedded); + } + + @Override + public void remove(String id) { + removeSingleInvocations++; + super.remove(id); + } + + @Override + public void removeAll(Collection<String> ids) { + removeAllByIdsInvocations++; + super.removeAll(ids); + } + + @Override + public void removeAll(Filter filter) { + removeAllByFilterInvocations++; + super.removeAll(filter); + } + + @Override + public void removeAll() { + removeAllInvocations++; + super.removeAll(); + } + + void resetCounters() { + addWithIdInvocations = 0; + addAllInvocations = 0; + addAllWithSegmentsInvocations = 0; + addAllWithIdsInvocations = 0; + removeSingleInvocations = 0; + removeAllByIdsInvocations = 0; + removeAllByFilterInvocations = 0; + removeAllInvocations = 0; + lastCallerSuppliedId = null; + } + + int getAddWithIdInvocations() { + return addWithIdInvocations; + } + + int getAddAllInvocations() { + return addAllInvocations; + } + + int getAddAllWithSegmentsInvocations() { + return addAllWithSegmentsInvocations; + } + + int getAddAllWithIdsInvocations() { + return addAllWithIdsInvocations; + } + + int getRemoveSingleInvocations() { + return removeSingleInvocations; + } + + int getRemoveAllByIdsInvocations() { + return removeAllByIdsInvocations; + } + + int getRemoveAllByFilterInvocations() { + return removeAllByFilterInvocations; + } + + int getRemoveAllInvocations() { + return removeAllInvocations; + } + + String getLastCallerSuppliedId() { + return lastCallerSuppliedId; + } + } +}
