CAMEL-25027: langchain4j-ingest-sink: add modality and contentType

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

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}}"

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

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}}"

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"

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}}"