This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26206-ingest-text-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit 78e90b2417b4eb6b31a46b12f1a2bd2641edcee4 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 00:22:32 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 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. 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 | 61 ++++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +++ 8 files changed, 131 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..f2fd971c46fc 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,63 @@ 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 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 10f5a47c1b89..fcdbbcd535c3 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 @@ -2016,6 +2016,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
