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-kamelets.git
The following commit(s) were added to refs/heads/main by this push:
new 9692c9448 CAMEL-25027: langchain4j-ingest-sink: add modality and
contentType (#3085)
9692c9448 is described below
commit 9692c9448dafbbc05e538021505d411132e51937
Author: jiriOndrusek-agent <[email protected]>
AuthorDate: Sun Oct 11 16:30:51 2026 +0200
CAMEL-25027: langchain4j-ingest-sink: add modality and contentType (#3085)
* CAMEL-25027: langchain4j-ingest-sink: add modality and contentType
modality=media hands the payload to the component's media path, which
embeds audio, an image, video or a PDF whole through an embedding model
that declares the matching content type; contentType overrides the MIME
type derived from the document id. Both are forwarded only when set, so
a sink without them keeps working on a Camel without the options. One
Citrus test runs the file source -> sink composition in media mode over
the PNG signature as fixture under an id without extension, the stub
model failing unless the bytes arrive unchanged. Related: CAMEL-25027.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* CAMEL-25027: langchain4j-ingest-sink: anchor the contentType pattern
JSON Schema patterns match partially, so the unanchored pattern accepted
any value that contained a MIME type, such as one with stray spaces or
surrounding text. Anchored, like the catalog's other patterns. Review
feedback.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Jiří Ondrušek <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../langchain4j-ingest-file-source.kamelet.yaml | 5 +-
kamelets/langchain4j-ingest-sink.kamelet.yaml | 41 +++++++++++--
.../langchain4j-ingest-file-source.kamelet.yaml | 5 +-
.../kamelets/langchain4j-ingest-sink.kamelet.yaml | 41 +++++++++++--
...gchain4j-ingest-file-media-route.citrus.it.yaml | 68 ++++++++++++++++++++++
.../langchain4j-ingest-file-media-route.yaml | 68 ++++++++++++++++++++++
6 files changed, 212 insertions(+), 16 deletions(-)
diff --git a/kamelets/langchain4j-ingest-file-source.kamelet.yaml
b/kamelets/langchain4j-ingest-file-source.kamelet.yaml
index 0fdc1434a..c133d9a56 100644
--- a/kamelets/langchain4j-ingest-file-source.kamelet.yaml
+++ b/kamelets/langchain4j-ingest-file-source.kamelet.yaml
@@ -76,8 +76,9 @@ spec:
title: Charset
description: Character set for reading text documents, for example
UTF-8. Leave
unset when the source feeds a parser action
(tika-extract-text-action or
- docling-convert-action), which must receive the raw bytes - a charset
- conversion would corrupt a binary document such as a PDF.
+ docling-convert-action) or the langchain4j-ingest-sink in media
mode, which must
+ receive the raw bytes - a charset conversion would corrupt a binary
document
+ such as a PDF.
type: string
idempotentRepository:
title: Idempotent Repository
diff --git a/kamelets/langchain4j-ingest-sink.kamelet.yaml
b/kamelets/langchain4j-ingest-sink.kamelet.yaml
index 1e43e351a..66b9c6bce 100644
--- a/kamelets/langchain4j-ingest-sink.kamelet.yaml
+++ b/kamelets/langchain4j-ingest-sink.kamelet.yaml
@@ -46,6 +46,11 @@ spec:
version-aware id. The filter options (`includeId`, `excludeId`,
`minDocumentSize`,
`documentFilter`) answer rejected deliveries with a filtered outcome
that never keeps
a dedup claim.
+
+ With `modality` set to `media` the payload (audio, an image, video or a
PDF) is embedded
+ whole, as one vector; the source must deliver raw bytes. A file source
re-polls any
+ document that fails the exchange on every run, so restrict the ids with
`includeId` or
+ the source's `include`.
type: object
properties:
pipelineName:
@@ -61,6 +66,25 @@ spec:
CamelLangChain4jIngestDocumentId exchange property, when set, takes
precedence.
type: string
default: CamelLangChain4jIngestDocumentId
+ modality:
+ title: Modality
+ description: What the payload is; text when unset. `text` is split
into segments and
+ embedded segment by segment; `media` (audio, an image, video or a
PDF) is embedded
+ whole, as one vector, by a model that declares the matching content
type; a text-only
+ model fails the start, a medium the model lacks fails the exchange.
In media mode the
+ splitter options and embeddingBatchSize do not apply,
documentSplitter must not be
+ set, and maxDocumentSize and minDocumentSize count bytes.
+ type: string
+ enum: ["text", "media"]
+ contentType:
+ title: Content Type
+ description: MIME type of a media payload, such as audio/wav or
image/png, handed to
+ the embedding model. When unset, it is derived from the document
id's file extension
+ (Camel's MIME table); a document whose type cannot be determined, or
whose medium the
+ model does not declare, fails the exchange. Only valid with modality
media, else the
+ sink fails to start.
+ type: string
+ pattern: "^[^/ ;]+/[^/ ;]+(;.*)?$"
maxSegmentSize:
title: Max Segment Size
description: Maximum size of one segment, in characters.
@@ -78,9 +102,10 @@ spec:
default: 32
maxDocumentSize:
title: Max Document Size
- description: Maximum size of one document in characters; unset means
no limit. The
- pipeline holds a document in memory whole, so set the cap when the
source can
- deliver oversized payloads. An oversized document fails the exchange
cleanly.
+ description: Maximum size of one document in characters, bytes with
modality media;
+ unset means no limit. The pipeline holds a document in memory whole,
so set the cap
+ when the source can deliver oversized payloads. An oversized
document fails the
+ exchange cleanly.
type: integer
includeId:
title: Include Id
@@ -95,8 +120,9 @@ spec:
type: string
minDocumentSize:
title: Min Document Size
- description: Minimum size of one document in characters; unset means
no minimum. A
- shorter document is answered filtered and releases its dedup claim.
+ description: Minimum size of one document in characters, bytes with
modality media;
+ unset means no minimum. A shorter document is answered filtered and
releases its
+ dedup claim.
type: integer
documentFilter:
title: Document Filter
@@ -107,7 +133,8 @@ spec:
documentSplitter:
title: Document Splitter
description: A DocumentSplitter bean replacing the default recursive
splitting, as a
- `#bean:name` reference; maxSegmentSize and maxOverlapSize are then
ignored.
+ `#bean:name` reference; maxSegmentSize and maxOverlapSize are then
ignored. With
+ modality media the sink fails to start.
type: string
embeddingStore:
title: Embedding Store
@@ -136,6 +163,7 @@ spec:
uri: langchain4j-ingest
parameters:
pipelineName: "{{pipelineName}}"
+ contentType: "{{?contentType}}"
documentFilter: "{{?documentFilter}}"
documentIdHeader: "{{documentIdHeader}}"
documentSplitter: "{{?documentSplitter}}"
@@ -149,3 +177,4 @@ spec:
maxOverlapSize: "{{maxOverlapSize}}"
maxSegmentSize: "{{maxSegmentSize}}"
minDocumentSize: "{{?minDocumentSize}}"
+ modality: "{{?modality}}"
diff --git
a/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-file-source.kamelet.yaml
b/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-file-source.kamelet.yaml
index 0fdc1434a..c133d9a56 100644
---
a/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-file-source.kamelet.yaml
+++
b/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-file-source.kamelet.yaml
@@ -76,8 +76,9 @@ spec:
title: Charset
description: Character set for reading text documents, for example
UTF-8. Leave
unset when the source feeds a parser action
(tika-extract-text-action or
- docling-convert-action), which must receive the raw bytes - a charset
- conversion would corrupt a binary document such as a PDF.
+ docling-convert-action) or the langchain4j-ingest-sink in media
mode, which must
+ receive the raw bytes - a charset conversion would corrupt a binary
document
+ such as a PDF.
type: string
idempotentRepository:
title: Idempotent Repository
diff --git
a/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-sink.kamelet.yaml
b/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-sink.kamelet.yaml
index 1e43e351a..66b9c6bce 100644
---
a/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-sink.kamelet.yaml
+++
b/library/camel-kamelets/src/main/resources/kamelets/langchain4j-ingest-sink.kamelet.yaml
@@ -46,6 +46,11 @@ spec:
version-aware id. The filter options (`includeId`, `excludeId`,
`minDocumentSize`,
`documentFilter`) answer rejected deliveries with a filtered outcome
that never keeps
a dedup claim.
+
+ With `modality` set to `media` the payload (audio, an image, video or a
PDF) is embedded
+ whole, as one vector; the source must deliver raw bytes. A file source
re-polls any
+ document that fails the exchange on every run, so restrict the ids with
`includeId` or
+ the source's `include`.
type: object
properties:
pipelineName:
@@ -61,6 +66,25 @@ spec:
CamelLangChain4jIngestDocumentId exchange property, when set, takes
precedence.
type: string
default: CamelLangChain4jIngestDocumentId
+ modality:
+ title: Modality
+ description: What the payload is; text when unset. `text` is split
into segments and
+ embedded segment by segment; `media` (audio, an image, video or a
PDF) is embedded
+ whole, as one vector, by a model that declares the matching content
type; a text-only
+ model fails the start, a medium the model lacks fails the exchange.
In media mode the
+ splitter options and embeddingBatchSize do not apply,
documentSplitter must not be
+ set, and maxDocumentSize and minDocumentSize count bytes.
+ type: string
+ enum: ["text", "media"]
+ contentType:
+ title: Content Type
+ description: MIME type of a media payload, such as audio/wav or
image/png, handed to
+ the embedding model. When unset, it is derived from the document
id's file extension
+ (Camel's MIME table); a document whose type cannot be determined, or
whose medium the
+ model does not declare, fails the exchange. Only valid with modality
media, else the
+ sink fails to start.
+ type: string
+ pattern: "^[^/ ;]+/[^/ ;]+(;.*)?$"
maxSegmentSize:
title: Max Segment Size
description: Maximum size of one segment, in characters.
@@ -78,9 +102,10 @@ spec:
default: 32
maxDocumentSize:
title: Max Document Size
- description: Maximum size of one document in characters; unset means
no limit. The
- pipeline holds a document in memory whole, so set the cap when the
source can
- deliver oversized payloads. An oversized document fails the exchange
cleanly.
+ description: Maximum size of one document in characters, bytes with
modality media;
+ unset means no limit. The pipeline holds a document in memory whole,
so set the cap
+ when the source can deliver oversized payloads. An oversized
document fails the
+ exchange cleanly.
type: integer
includeId:
title: Include Id
@@ -95,8 +120,9 @@ spec:
type: string
minDocumentSize:
title: Min Document Size
- description: Minimum size of one document in characters; unset means
no minimum. A
- shorter document is answered filtered and releases its dedup claim.
+ description: Minimum size of one document in characters, bytes with
modality media;
+ unset means no minimum. A shorter document is answered filtered and
releases its
+ dedup claim.
type: integer
documentFilter:
title: Document Filter
@@ -107,7 +133,8 @@ spec:
documentSplitter:
title: Document Splitter
description: A DocumentSplitter bean replacing the default recursive
splitting, as a
- `#bean:name` reference; maxSegmentSize and maxOverlapSize are then
ignored.
+ `#bean:name` reference; maxSegmentSize and maxOverlapSize are then
ignored. With
+ modality media the sink fails to start.
type: string
embeddingStore:
title: Embedding Store
@@ -136,6 +163,7 @@ spec:
uri: langchain4j-ingest
parameters:
pipelineName: "{{pipelineName}}"
+ contentType: "{{?contentType}}"
documentFilter: "{{?documentFilter}}"
documentIdHeader: "{{documentIdHeader}}"
documentSplitter: "{{?documentSplitter}}"
@@ -149,3 +177,4 @@ spec:
maxOverlapSize: "{{maxOverlapSize}}"
maxSegmentSize: "{{maxSegmentSize}}"
minDocumentSize: "{{?minDocumentSize}}"
+ modality: "{{?modality}}"
diff --git
a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.citrus.it.yaml
b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.citrus.it.yaml
new file mode 100644
index 000000000..907df21db
--- /dev/null
+++
b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.citrus.it.yaml
@@ -0,0 +1,68 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+
+name: langchain4j-ingest-file-media-route-test
+actions:
+ # drain any request a previous test may have left on the shared server
+ - purge:
+ endpoints:
+ - name: "httpServer"
+ timeout: 100
+ - createVariables:
+ variables:
+ - name: "report.url"
+ value: "http://localhost:${http.server.port}/result/file-media";
+ # a fresh directory per test run, see the file-source test
+ - name: "ingest.dir"
+ value: "ingest-media-citrus:randomNumber(5)"
+
+ # Run the file-source -> sink composition in media mode over a byte fixture
under an id
+ # without extension; the sink's IngestResult is reported back to the test
+ - camel:
+ cli:
+ run:
+ waitForRunningState: false
+ args:
+ - "--dep=camel:groovy,camel:langchain4j-ingest"
+ - "--max-seconds=60"
+ integration:
+ file: "langchain4j-ingest/langchain4j-ingest-file-media-route.yaml"
+ systemProperties:
+ properties:
+ - name: "report.url"
+ value: "${report.url}"
+ - name: "ingest.dir"
+ value: "${ingest.dir}"
+
+ # one vector for the whole file, under the file name as the id; the stub
model has
+ # already failed the exchange if the bytes did not arrive unchanged
+ - http:
+ server: "httpServer"
+ receiveRequest:
+ POST:
+ path: "/result/file-media"
+ type: "plaintext"
+ body:
+ data: "@contains('documentId=img-42, segmentsWritten=1,
outcome=INGESTED')@"
+
+ - http:
+ server: "httpServer"
+ sendResponse:
+ response:
+ status: 200
+ reasonPhrase: "OK"
+ version: "HTTP/1.1"
diff --git
a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.yaml
b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.yaml
new file mode 100644
index 000000000..c9244ef04
--- /dev/null
+++
b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/langchain4j-ingest-file-media-route.yaml
@@ -0,0 +1,68 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+
+# The media composition: the file source feeds the sink with modality=media.
The seed
+# route writes the eight-byte PNG signature, whose first byte is invalid
UTF-8, under a
+# name without extension; the source delivers it raw (charset unset, its
default), only
+# the forwarded contentType can type it, and the stub model fails unless the
bytes arrive
+# unchanged - a charset conversion anywhere on the way fails the test instead
of passing.
+- beans:
+ - name: store
+ type: dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore
+ # an in-process stub accepting images only, so the media path is the only
one that can succeed;
+ # it keys doEmbed, the extension point the default embed(request)
delegates to
+ - name: model
+ type: dev.langchain4j.model.embedding.EmbeddingModel
+ scriptLanguage: groovy
+ script: |
+ def expected = "iVBORw0KGgo=" // base64 of the PNG signature 89 50 4E
47 0D 0A 1A 0A
+ return [
+ supportedContentTypes: {
+ [dev.langchain4j.data.message.ContentType.IMAGE] as Set
+ },
+ doEmbed: { request ->
+ def image = request.inputs()[0].contents()
+ .find { it instanceof
dev.langchain4j.data.message.ImageContent }
+ def received = image?.image()?.base64Data()
+ if (received != expected) {
+ throw new IllegalStateException("model received " +
received + ", expected " + expected)
+ }
+
dev.langchain4j.model.embedding.response.EmbeddingResponse.builder()
+
.embeddings([dev.langchain4j.data.embedding.Embedding.from([1f, 0f, 0f, 1f] as
float[])])
+ .build()
+ }
+ ] as dev.langchain4j.model.embedding.EmbeddingModel
+- route:
+ id: seed-media
+ from:
+ uri: "timer:seed"
+ parameters:
+ repeatCount: 1
+ steps:
+ - setBody:
+ groovy: "[0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A] as
byte[]"
+ - to: "file:{{ingest.dir}}?fileName=img-42"
+- route:
+ id: langchain4j-ingest-file-media-route
+ from:
+ uri: "kamelet:langchain4j-ingest-file-source?directory={{ingest.dir}}"
+ steps:
+ - to:
"kamelet:langchain4j-ingest-sink?modality=media&contentType=image/png&embeddingStore=#bean:store&embeddingModel=#bean:model"
+ - log: "INGEST ${body}"
+ - convertBodyTo:
+ type: "java.lang.String"
+ - to: "{{report.url}}"
