atiaomar1978-hub commented on code in PR #25273:
URL: https://github.com/apache/camel/pull/25273#discussion_r3741028093
##########
components/camel-ai/camel-langchain4j-embeddings/src/main/java/org/apache/camel/component/langchain4j/embeddings/LangChain4jEmbeddingsProducer.java:
##########
@@ -55,4 +69,30 @@ public void process(Exchange exchange) throws Exception {
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 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);
Review Comment:
**Grok:** Batch path sets `EMBEDDINGS` header + embedding list body but
omits parallel text segments. For embed→store RAG flows, callers must retain
original `List<TextSegment>` separately for `addAll(embeddings, textSegments)`.
Doc example should show a `.process()` preserving segments or add a
`TEXT_SEGMENTS` header.
_AI-generated Grok inline comment on behalf of atiaomar1978-hub._
##########
components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java:
##########
@@ -109,63 +110,150 @@ public void process(Exchange exchange) throws Exception {
}
/**
- * 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,
LangChain4jEmbeddingsHeaders.EMBEDDING);
}
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) {
Review Comment:
**Bugbot:** Before `addAll(callerIds, embeddings, textSegments)`, validate
all three lists are non-null and equal size. Langchain4j may throw opaque
errors otherwise.
_AI-generated Bugbot inline comment on behalf of atiaomar1978-hub._
##########
components/camel-ai/camel-langchain4j-embeddingstore/src/main/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreProducer.java:
##########
@@ -109,63 +110,150 @@ public void process(Exchange exchange) throws Exception {
}
/**
- * 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,
LangChain4jEmbeddingsHeaders.EMBEDDING);
}
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 (callerIds != null) {
+ // No addAll(ids, embeddings) overload in langchain4j, so loop
with add(id, embedding)
+ for (int i = 0; i < embeddings.size(); i++) {
+ store.add(callerIds.get(i), embeddings.get(i));
+ }
+ 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>
+ * </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();
- getEndpoint().getConfiguration().getEmbeddingStore().remove(id);
+ // Check for filter-based removal first
+ Filter filter = in.getHeader(LangChain4jEmbeddingStoreHeaders.FILTER,
Filter.class);
+ if (filter != null) {
+ store.removeAll(filter);
+ return;
+ }
- Message out = exchange.getMessage();
+ Object body = in.getBody();
+
+ // Batch removal by collection of IDs
+ if (body instanceof Collection) {
Review Comment:
**Grok:** `body instanceof Collection` accepts any collection type at
compile time but casts to `Collection<String>` unchecked. Non-string elements
fail at runtime inside langchain4j — consider validating element types or
documenting constraint.
_AI-generated Grok inline comment on behalf of atiaomar1978-hub._
##########
components/camel-ai/camel-langchain4j-embeddingstore/src/test/java/org/apache/camel/component/langchain4j/embeddingstore/LangChain4jEmbeddingStoreBatchOperationsTest.java:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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");
+ }
+
+ @Test
+ @DisplayName("ADD with EMBEDDINGS header and caller IDs but no text
segments loops with add(id, embedding)")
+ void addBatchWithCallerIdsNoTextSegments() {
Review Comment:
**Grok (tests):** Strong `RecordingEmbeddingStore` pattern — verifies exact
SDK dispatch. Missing tests: size mismatch on `EMBEDDING_IDS`, single ADD with
both `EMBEDDING_ID` + `TEXT_SEGMENT`, end-to-end embed-batch→store-batch route.
_AI-generated Grok test-coverage inline comment on behalf of
atiaomar1978-hub._
##########
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")
Review Comment:
**Bugbot (tests):** Good AssertJ coverage for batch/single/token usage.
Missing: `List<TextSegment>` body path, empty list behavior, integration test
chaining to embeddingstore ADD with segments.
_AI-generated Bugbot test-coverage inline comment on behalf of
atiaomar1978-hub._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
