This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 542f006d0055 CAMEL-26206: camel-langchain4j-ingest - text mode ingests
the text of a TextSegment or Document body, and fails a body that is not text
(#29030)
542f006d0055 is described below
commit 542f006d0055fc7b94eb2ebc5f689c1939d0417e
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 07:45:45 2026 +0100
CAMEL-26206: camel-langchain4j-ingest - text mode ingests the text of a
TextSegment or Document body, and fails a body that is not text (#29030)
* CAMEL-26206: camel-langchain4j-ingest - text mode ingests the text of…
… a TextSegment or Document body, and fails a body that is not text
The text modality read the body with getBody(String.class), which for an
object
without a real text form falls back to its toString(). A TextSegment was
ingested
as "TextSegment { text = ... }", a Map or a POJO as its toString(), without
any error.
A TextSegment or Document body now gives its text. Text, bytes, a stream, a
Reader
or a file is read as before. Any other body fails with an
InvalidPayloadException,
as a Map or a POJO already did in media mode.
* CAMEL-26206: camel-langchain4j-ingest - test that an oversized TextSe…
…gment or Document body fails maxDocumentSize like a String
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../catalog/components/langchain4j-ingest.json | 2 +-
.../catalog/docs/langchain4j-ingest-component.adoc | 7 ++
.../langchain4j/ingest/langchain4j-ingest.json | 2 +-
.../main/docs/langchain4j-ingest-component.adoc | 7 ++
.../ingest/LangChain4jIngestEndpoint.java | 10 +--
.../ingest/LangChain4jIngestProducer.java | 46 ++++++++++--
.../ingest/LangChain4jIngestProducerTest.java | 81 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 ++
8 files changed, 151 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/langchain4j-ingest.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/langchain4j-ingest.json
index 5ee640ee5c5e..d555e1b70039 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/langchain4j-ingest.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/langchain4j-ingest.json
@@ -47,7 +47,7 @@
"CamelLangChain4jIngestDocumentId": { "index": 0, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The stable document id of the ingested
payload, read when the documentIdHeader endpoint option does not name another
header. The CamelLangChain4jIngestDocumentId exchange property, when set, takes
precedence over any header.", "c [...]
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"byOption": [ { "option": "modality", "value": "MEDIA", "carriers": [ "bytes",
"stream", "file", "null" ] } ], "description": "The document read as a String,
then split and embedded; another body, such as a Map or a TextSegment, is
ingested as its toString(). With modality=MEDIA it is read as bytes and
embedded whole, and a Map or a POJO fails. No body or a blank one writes
nothing. The document id comes from [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"byOption": [ { "option": "modality", "value": "MEDIA", "carriers": [ "bytes",
"stream", "file", "null" ] } ], "description": "The document read as text, then
split and embedded; a LangChain4j TextSegment or Document gives its text, and
another body, such as a Map or a POJO, fails. With modality=MEDIA it is read as
bytes and embedded whole, and a Map or a POJO fails. No body or a blank one
writes nothing. The [...]
"producerOut": { "carriers": [ "pojo" ], "javaType":
"org.apache.camel.component.langchain4j.ingest.IngestResult", "description":
"An IngestResult with the pipeline, the document id, the number of segments
written and the outcome: INGESTED, EMPTY, SKIPPED or FILTERED" }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
index cc719532b02c..7a0f04c0a89c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
@@ -142,6 +142,13 @@ the `CamelLangChain4jIngestDocumentId` header and fails
the exchange when no id
what update and delete semantics are built on, so a generated fallback would
silently break
replacement.
+The body is read as text: a String, bytes, a stream, a `Reader` or a file. A
LangChain4j
+`TextSegment` or `Document` gives its text; its metadata is not ingested, the
written segments
+carry the identity metadata above. Any other body, such as a `Map` or a POJO,
fails the exchange
+with an `InvalidPayloadException` instead of being ingested as its
`toString()`: turn it into
+text first, with a marshal step, say. With `modality=media` the body is read
as bytes instead
+(see below).
+
After ingestion the message body is replaced with the `IngestResult`, so a
request-reply
caller receives the outcome of its call: the pipeline, the document id, the
number of segments
written and the outcome — `ingested`, `empty` (blank document, nothing
written), `skipped`
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/ingest/langchain4j-ingest.json
b/components/camel-ai/camel-langchain4j-ingest/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/ingest/langchain4j-ingest.json
index 5ee640ee5c5e..d555e1b70039 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/ingest/langchain4j-ingest.json
+++
b/components/camel-ai/camel-langchain4j-ingest/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/ingest/langchain4j-ingest.json
@@ -47,7 +47,7 @@
"CamelLangChain4jIngestDocumentId": { "index": 0, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The stable document id of the ingested
payload, read when the documentIdHeader endpoint option does not name another
header. The CamelLangChain4jIngestDocumentId exchange property, when set, takes
precedence over any header.", "c [...]
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"byOption": [ { "option": "modality", "value": "MEDIA", "carriers": [ "bytes",
"stream", "file", "null" ] } ], "description": "The document read as a String,
then split and embedded; another body, such as a Map or a TextSegment, is
ingested as its toString(). With modality=MEDIA it is read as bytes and
embedded whole, and a Map or a POJO fails. No body or a blank one writes
nothing. The document id comes from [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"byOption": [ { "option": "modality", "value": "MEDIA", "carriers": [ "bytes",
"stream", "file", "null" ] } ], "description": "The document read as text, then
split and embedded; a LangChain4j TextSegment or Document gives its text, and
another body, such as a Map or a POJO, fails. With modality=MEDIA it is read as
bytes and embedded whole, and a Map or a POJO fails. No body or a blank one
writes nothing. The [...]
"producerOut": { "carriers": [ "pojo" ], "javaType":
"org.apache.camel.component.langchain4j.ingest.IngestResult", "description":
"An IngestResult with the pipeline, the document id, the number of segments
written and the outcome: INGESTED, EMPTY, SKIPPED or FILTERED" }
},
"properties": {
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
b/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
index cc719532b02c..7a0f04c0a89c 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
@@ -142,6 +142,13 @@ the `CamelLangChain4jIngestDocumentId` header and fails
the exchange when no id
what update and delete semantics are built on, so a generated fallback would
silently break
replacement.
+The body is read as text: a String, bytes, a stream, a `Reader` or a file. A
LangChain4j
+`TextSegment` or `Document` gives its text; its metadata is not ingested, the
written segments
+carry the identity metadata above. Any other body, such as a `Map` or a POJO,
fails the exchange
+with an `InvalidPayloadException` instead of being ingested as its
`toString()`: turn it into
+text first, with a marshal step, say. With `modality=media` the body is read
as bytes instead
+(see below).
+
After ingestion the message body is replaced with the `IngestResult`, so a
request-reply
caller receives the outcome of its call: the pipeline, the document id, the
number of segments
written and the outcome — `ingested`, `empty` (blank document, nothing
written), `skipped`
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestEndpoint.java
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestEndpoint.java
index 3cadcb98c265..ecf364813fa4 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestEndpoint.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestEndpoint.java
@@ -48,11 +48,11 @@ import org.apache.camel.support.DefaultEndpoint;
carriers = {
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.NULL }),
- description = "The document read as a
String, then split and embedded; another body,"
- + " such as a Map or a
TextSegment, is ingested as its toString(). With"
- + " modality=MEDIA it is read
as bytes and embedded whole, and a Map or a POJO"
- + " fails. No body or a blank
one writes nothing. The document id comes from"
- + " the
CamelLangChain4jIngestDocumentId header or exchange property"),
+ description = "The document read as text,
then split and embedded; a LangChain4j TextSegment or"
+ + " Document gives its text,
and another body, such as a Map or a POJO, fails."
+ + " With modality=MEDIA it is
read as bytes and embedded whole, and a Map or a"
+ + " POJO fails. No body or a
blank one writes nothing. The document id comes"
+ + " from the
CamelLangChain4jIngestDocumentId header or exchange property"),
producerOut = @BodyType(value = Carrier.POJO,
javaType =
"org.apache.camel.component.langchain4j.ingest.IngestResult",
description = "An IngestResult with the
pipeline, the document id, the number of segments"
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
index 1124ee761461..848ca1f9c4fc 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
@@ -17,12 +17,17 @@
package org.apache.camel.component.langchain4j.ingest;
import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
import java.util.Locale;
import java.util.Set;
+import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.message.ContentType;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
@@ -31,6 +36,8 @@ import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Predicate;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.support.service.ServiceHelper;
@@ -306,6 +313,38 @@ public class LangChain4jIngestProducer extends
DefaultProducer {
return textService.ingest(documentId, textBody(exchange));
}
+ /**
+ * The text of the body: a LangChain4j {@code TextSegment} or {@code
Document} gives its text, and text, bytes, a
+ * stream, a Reader or a file is read as text. Any other body, such as a
Map or a POJO, fails as an
+ * InvalidPayloadException instead of being ingested as its toString().
+ */
+ private String textBody(Exchange exchange) throws InvalidPayloadException,
IOException {
+ Object body = exchange.getMessage().getBody();
+ if (body instanceof TextSegment segment) {
+ return segment.text();
+ }
+ if (body instanceof Document document) {
+ return document.text();
+ }
+ if (body != null && !isReadableAsText(body)) {
+ throw new InvalidPayloadException(
+ exchange, String.class, exchange.getMessage(),
+ "ingestion pipeline '" + endpoint.getPipelineName() + "'
reads text, bytes, a stream, a Reader, a"
+ + " file, a
TextSegment or a Document; convert the body"
+ + " to text
before the endpoint");
+ }
+ return readText(exchange, body);
+ }
+
+ /** A body Camel converts to its text, rather than to the toString() of an
object. */
+ private static boolean isReadableAsText(Object body) {
+ return body instanceof CharSequence || body instanceof byte[] || body
instanceof ByteBuffer
+ || body instanceof InputStream || body instanceof Reader ||
body instanceof StreamCache
+ // the output of a Tika parse
+ || body instanceof ByteArrayOutputStream
+ || body instanceof File || body instanceof Path || body
instanceof WrappedFile;
+ }
+
/**
* The media body as bytes, not a String - a charset conversion would
corrupt them. A null body stays EMPTY;
* anything else must convert, so a wrong body type (a POJO, say) fails as
an InvalidPayloadException instead of
@@ -326,11 +365,10 @@ public class LangChain4jIngestProducer extends
DefaultProducer {
}
/**
- * The text body. With maxDocumentSize set, a body not yet a String is
read only one character past the cap, enough
- * for the size check to refuse it.
+ * The body read as text. With maxDocumentSize set, a body not yet a
String is read only one character past the cap,
+ * enough for the size check to refuse it.
*/
- private String textBody(Exchange exchange) throws IOException {
- Object body = exchange.getMessage().getBody();
+ private String readText(Exchange exchange, Object body) throws IOException
{
int limit = readLimit();
Reader reader = body == null || body instanceof String || limit < 0
? null : readerOf(exchange);
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
index df44131aa81f..80e6bef2cc59 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
@@ -17,16 +17,20 @@
package org.apache.camel.component.langchain4j.ingest;
import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.InputStream;
+import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import dev.langchain4j.data.document.Document;
import dev.langchain4j.data.document.DocumentSplitter;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.store.embedding.EmbeddingMatch;
@@ -266,6 +270,83 @@ class LangChain4jIngestProducerTest extends
CamelTestSupport {
assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
}
+ @Test
+ void textSegmentAndDocumentBodiesIngestTheirText() {
+ String text = "Camel rides through the desert.";
+ for (Object body : new Object[] { TextSegment.from(text),
Document.from(text) }) {
+ String documentId = "doc-" + body.getClass().getSimpleName();
+ IngestResult result =
template.requestBodyAndHeader("direct:ingest", body,
+ LangChain4jIngestHeaders.DOCUMENT_ID, documentId,
IngestResult.class);
+
+
assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
+ // the text itself, not "TextSegment { text = ... }" or "Document
{ text = ... }"
+ assertThat(storedTexts(text, documentId)).containsExactly(text);
+ }
+ }
+
+ @Test
+ void oversizedTextSegmentAndDocumentFailLikeAnOversizedString() {
+ // the text of a TextSegment or a Document meets the same cap as a
String body
+ String oversized = "exactly thirty characters long";
+ for (Object body : new Object[] { oversized,
TextSegment.from(oversized), Document.from(oversized) }) {
+ assertThatThrownBy(() ->
template.requestBodyAndHeader("direct:capped", body, "MyDocId",
"doc-oversized"))
+ .as(body.getClass().getName())
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("exceeds maxDocumentSize (30 > 20
characters)");
+ }
+ assertThat(search(oversized)).noneMatch(m ->
m.embedded().text().contains("thirty"));
+
+ String atCap = "twenty characters ok";
+ for (Object body : new Object[] { TextSegment.from(atCap),
Document.from(atCap) }) {
+ IngestResult result =
template.requestBodyAndHeader("direct:capped", body, "MyDocId",
+ "doc-at-cap-" + body.getClass().getSimpleName(),
IngestResult.class);
+
assertThat(result.outcome()).as(body.getClass().getName()).isEqualTo(IngestResult.Outcome.INGESTED);
+ }
+ }
+
+ @Test
+ void textCarriersAreReadAsText() throws Exception {
+ String text = "Camel rides through the desert.";
+ ByteArrayOutputStream tikaOutput = new ByteArrayOutputStream();
+ tikaOutput.write(text.getBytes(StandardCharsets.UTF_8));
+ Object[] bodies = {
+ new StringBuilder(text),
text.getBytes(StandardCharsets.UTF_8), new StringReader(text), tikaOutput };
+ for (int i = 0; i < bodies.length; i++) {
+ Object body = bodies[i];
+ String documentId = "doc-" + i;
+ IngestResult result =
template.requestBodyAndHeader("direct:ingest", body,
+ LangChain4jIngestHeaders.DOCUMENT_ID, documentId,
IngestResult.class);
+
+
assertThat(result.outcome()).as(body.getClass().getName()).isEqualTo(IngestResult.Outcome.INGESTED);
+ assertThat(storedTexts(text,
documentId)).as(body.getClass().getName()).containsExactly(text);
+ }
+ }
+
+ @Test
+ void bodyThatIsNotTextFails() {
+ // a Map or a POJO used to be ingested as its toString(); with a cap
or without, it fails
+ for (Object body : new Object[] { Map.of("title", "Camel"), new
Object() }) {
+ assertThatThrownBy(() ->
template.requestBodyAndHeader("direct:ingest", body,
+ LangChain4jIngestHeaders.DOCUMENT_ID, "doc-pojo"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("InvalidPayloadException")
+ .hasStackTraceContaining("convert the body to text before
the endpoint");
+ assertThatThrownBy(() ->
template.requestBodyAndHeader("direct:capped", body, "MyDocId", "doc-pojo"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("InvalidPayloadException");
+ }
+ assertThat(search("Camel")).isEmpty();
+ }
+
+ /** The texts written for the document id, among the segments similar to
the given text. */
+ private List<String> storedTexts(String text, String documentId) {
+ return search(text).stream()
+ .map(EmbeddingMatch::embedded)
+ .filter(segment ->
documentId.equals(segment.metadata().getString(LangChain4jIngest.METADATA_DOCUMENT_ID)))
+ .map(TextSegment::text)
+ .toList();
+ }
+
@Test
void customBatchSizeControlsTheEmbeddingRequests() {
// 10 paragraphs of ~19 characters against maxSegmentSize=30: one
segment each, and a
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 158a61a0e8ab..0e829726f1d4 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2072,6 +2072,13 @@ On Jira Cloud the Jira consumers now read the pages of a
search with the next pa
start index: with `maxResults` above 50, a search with more than 50 results
failed on the second page with
`UnsupportedOperationException`.
+=== camel-langchain4j-ingest - a body that is not text fails
+
+In the default text modality, the `langchain4j-ingest` producer now ingests
the text of a LangChain4j `TextSegment` or
+`Document` body, and fails the exchange with an `InvalidPayloadException` for
a body that is not text, bytes, a stream,
+a `Reader` or a file, such as a `Map` or a POJO. Before, such a body was
ingested as its `toString()`, for example
+`TextSegment { text = ... }`. Convert the body to text before the endpoint.
+
=== camel-netty-http - an ambiguous request path is rejected when a security
constraint is configured
A `netty-http` consumer configured with a `securityConstraint` now rejects,
with HTTP 400, a request whose path
