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