This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/main by this push:
new ae4ef2f4de Fixes #9280. Add filter.* and modality=media to
langchain4j-ingest
ae4ef2f4de is described below
commit ae4ef2f4dec34977d7f46f1b4abe130b59878070
Author: jiriOndrusek-agent <[email protected]>
AuthorDate: Tue Oct 27 12:28:37 2026 +0100
Fixes #9280. Add filter.* and modality=media to langchain4j-ingest
filter.include-id, filter.exclude-id, filter.min-document-size and
filter.document-filter go to the sink Kamelet and are enforced by the
component (CAMEL-24753), for configured and @Ingest pipelines alike.
With a parser, the id patterns are also checked before the parse, so
an excluded document is never parsed. A missing document-filter bean
fails the start naming it. On a directory pipeline, a filtered file's
key is held in memory rather than in a named, possibly persistent
register, so a changed filter applies to the file after a restart.
modality=media, with an optional content-type, embeds each document
whole through the sink's media path (CAMEL-25027); IngestPipeline gets
modality() and contentType(). A configured pipeline's modality and
content-type are checked at build time, while the splitter sizes and
batch size, which do not apply, are not; the builder setters reject
bad values.
modality, contentType and the filter options replace the neutral
values #9152 pins on the sink, so app-wide camel.kamelet.<id>.*
properties still cannot change them. The file source's charset has no
neutral value, so parser and media pipelines drop an app-wide one
before the raw bytes are read.
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../reference/extensions/langchain4j-ingest.adoc | 105 +++++++++-
.../deployment/Langchain4jIngestProcessor.java | 37 +++-
.../deployment/IngestBuilderFilterTypoTest.java | 57 ++++++
.../deployment/IngestMediaParserConflictTest.java | 71 +++++++
.../IngestMediaSplitterSettingsTest.java | 90 +++++++++
.../IngestMissingDocumentFilterTest.java | 44 +++++
.../deployment/IngestModalityConfigTest.java | 50 +++++
.../IngestPipelineModalityValidationTest.java | 56 ++++++
.../runtime/src/main/doc/usage.adoc | 39 +++-
.../langchain4j/ingest/IngestBuildTimeConfig.java | 33 +++-
.../langchain4j/ingest/IngestPipeline.java | 82 +++++++-
.../component/langchain4j/ingest/IngestRoutes.java | 219 ++++++++++++++++++---
.../langchain4j/ingest/IngestRunTimeConfig.java | 41 ++++
.../it/DeterministicAudioEmbeddingModel.java | 72 +++++++
.../langchain4j/ingest/it/IngestItProducers.java | 38 ++++
.../langchain4j/ingest/it/IngestResource.java | 59 ++++++
.../langchain4j/ingest/it/ItBuilderPipelines.java | 14 ++
.../src/main/resources/application.properties | 44 ++++-
.../ingest/it/Langchain4jIngestAudioIT.java | 23 +++
.../ingest/it/Langchain4jIngestAudioTest.java | 126 ++++++++++++
.../ingest/it/Langchain4jIngestFilterIT.java | 23 +++
.../ingest/it/Langchain4jIngestFilterTest.java | 148 ++++++++++++++
.../ingest/it/Langchain4jIngestTikaTest.java | 10 +-
23 files changed, 1433 insertions(+), 48 deletions(-)
diff --git
a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index 201059975b..697fc8e707 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -62,7 +62,7 @@
quarkus.camel.langchain4j.ingest.products.embedding-model=my-model
Each file is read as UTF-8 text — unless a `parser` turns binary formats into
text first, see below — split into overlapping segments (`max-segment-size`,
`max-overlap-size`; or a `DocumentSplitter` CDI bean named by
`document-splitter`, which then replaces the sizes), embedded in batches
(`embedding-batch-size`, 32 by default) and written to the store; a document is
held in memory whole while it is split, so `max-document-size` can cap it in
characters — 0, the default, means no limit, [...]
`embedding-store` and `embedding-model` name CDI beans and may be omitted when
the application has exactly one of each.
Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id`
metadata, so retrieval can cite which document an answer came from (up to
3.39.x the keys were `camel_quarkus_pipeline` and `camel_quarkus_document_id` —
see the migration guide).
-Apart from `enabled` and the `source.*` settings shown here, properties are
fixed at build time.
+Apart from `enabled`, `filter.++*++` and the `source.++*++` settings shown
here, properties are fixed at build time.
A pipeline declared through runtime properties alone — nothing but a
`source.directory`, say — is invisible to build-time validation; its checks,
including the clash with an equally named `@Ingest` pipeline, report at startup
instead.
Each pipeline is composed from the catalog's `langchain4j-ingest-file-source`,
parser action and `langchain4j-ingest-sink` Kamelets: its route
`langchain4j-ingest-<name>` plus one route per Kamelet it uses,
`langchain4j-ingest-<name>-source` (directory pipelines), `-parser` (with a
`parser`) and `-sink`.
The pipelines always load these Kamelets from the classpath, whatever
`camel.component.kamelet.location` says; shipping an equally named Kamelet
replaces internal steps — including the identity capture — and is advanced
surgery.
@@ -77,6 +77,24 @@ Replace and delete arrive with the synchronising engine in a
later release.
What is remembered is which documents were already ingested — the idempotent
repository below.
====
+[id="extensions-langchain4j-ingest-usage-filtering-deliveries"]
+=== Filtering deliveries
+
+Per pipeline, `filter.include-id` and `filter.exclude-id` take comma-separated
Ant-style patterns matched against the document id — exclusion wins, and a
rejected delivery never takes the sink's dedup claim nor reads the body.
+Matching is case-sensitive, and an id starting with `/` (an absolute path,
say) only matches a pattern that starts with `/` too.
+With a `parser` they are checked before the parse, so an excluded document is
never parsed nor sent to Docling Serve.
+`filter.min-document-size` rejects documents too short to carry retrievable
content, and `filter.document-filter` names a Camel `Predicate` bean evaluated
with the body available.
+With a `parser` these two see the parsed text, so a document they reject has
been parsed first — with `docling`, by the remote Docling Serve.
+A rejected delivery is answered with a `filtered` outcome; on a directory
pipeline it is logged at DEBUG and the file is not read again until it changes,
its key held in memory even with a persistent register: after a restart, a
changed `filter.++*++` applies to it.
+An `@Ingest` pipeline is filtered the same way: the builder has no filter
methods, and the `filter.*` properties apply to it too.
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.filter.include-id=**/*.pdf,**/*.md
+quarkus.camel.langchain4j.ingest.products.filter.exclude-id=**/draft-*
+quarkus.camel.langchain4j.ingest.products.filter.min-document-size=100
+----
+
[id="extensions-langchain4j-ingest-usage-the-idempotent-repository"]
=== The idempotent repository
@@ -171,6 +189,27 @@ A missing parser extension fails the build naming the
artifact, the same way a m
The Java twin is `IngestPipeline.parser("tika")`, and a consumer-fed pipeline
parses the same way — the payload the consumer delivers goes to the parser
first.
With a parser, `max-document-size` also rejects a raw payload larger than the
cap (in bytes) before it reaches the parser, on top of capping the extracted
text (in characters).
+[id="extensions-langchain4j-ingest-usage-ingesting-media-images-audio-video-and-pdf"]
+=== Ingesting media: images, audio, video and PDF
+
+`modality=media` embeds each document whole, as one vector, instead of reading
it as text: the bytes go to a multimodal embedding model as audio, an image,
video or a PDF, told apart by the MIME type, and the vector is stored with a
placeholder segment whose text is the document id, stamped with the usual
identity metadata, so retrieval cites it like any other.
+The model must declare the matching content type in its
`supportedContentTypes()`: the pipeline fails to start with a text-only model,
and a document whose medium the model does not declare fails before its bytes
are read.
+`parser` and `document-splitter` must not be set, the splitter sizes and
`embedding-batch-size` do not apply, and `max-document-size` and
`filter.min-document-size` count bytes.
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.photos.source.directory=/var/data/photos
+quarkus.camel.langchain4j.ingest.photos.modality=media
+quarkus.camel.langchain4j.ingest.photos.embedding-model=my-multimodal-model
+quarkus.camel.langchain4j.ingest.photos.embedding-store=photos
+quarkus.camel.langchain4j.ingest.photos.filter.include-id=**/*.png,**/*.jpg
+----
+
+The MIME type handed to the model comes from `content-type` or, when unset,
from the file extension through Camel's MIME table (wav, mp3, flac, ogg, opus,
m4a, aac and aiff for audio; png, jpg, gif and webp for images; mp4, mov and
webm for video; pdf); a file whose type cannot be determined fails the
exchange, and `content-type` without `modality=media` is rejected like the
parser and splitter conflicts.
+On a directory pipeline such a file — `notes.txt`, `Thumbs.db` — fails again
on every poll, so restrict the ids with `filter.include-id`, as above: it is
checked before the media type.
+Images are embedded today by several LangChain4j providers; no provider embeds
audio out of the box yet, so an audio model is a CDI bean of your own that
declares `AUDIO` — typically a client of an audio embedding service — and
either way the same bean embeds the query image or clip at retrieval time.
+The Java twin is `IngestPipeline.modality("media")`, and a consumer-fed
pipeline embeds the delivered payload the same way.
+
[id="extensions-langchain4j-ingest-usage-other-sources-declared-in-java"]
=== Other sources, declared in Java
@@ -282,6 +321,28 @@ value needs its extension on the classpath:
`camel-quarkus-tika` or
| `string`
|
+a|icon:lock[title=Fixed at build time]
[[quarkus-camel-langchain4j-ingest-pipeline-name-modality]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-modality[quarkus.camel.langchain4j.ingest."pipeline-name".modality]`
+
+What the consumed payload is. `text`, the default, is read as a String, split
into
+segments and embedded segment by segment. `media` is read as bytes and
embedded whole,
+as one vector: audio, an image, video or a PDF, told apart by the MIME type,
each
+needing an embedding model that declares the matching content type — the
pipeline
+fails to start with a text-only model. In media mode `parser` and
`document-splitter`
+must not be set, the splitter sizes and `embedding-batch-size` do not apply,
and
+`max-document-size` and `filter.min-document-size` count bytes.
+| `string`
+| `text`
+
+a|icon:lock[title=Fixed at build time]
[[quarkus-camel-langchain4j-ingest-pipeline-name-content-type]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-content-type[quarkus.camel.langchain4j.ingest."pipeline-name".content-type]`
+
+MIME type of a media payload, such as `audio/wav` or `image/png`, handed to the
+embedding model. When not set, it is derived from the document id's file
extension
+through 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`: set without it, the pipeline is rejected.
+| `string`
+|
+
a|icon:lock[title=Fixed at build time]
[[quarkus-camel-langchain4j-ingest-pipeline-name-embedding-store]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-embedding-store[quarkus.camel.langchain4j.ingest."pipeline-name".embedding-store]`
Name of the `EmbeddingStore` bean to write to. When not set, the only one
present is
@@ -320,10 +381,11 @@ together against the provider's token limits.
a|icon:lock[title=Fixed at build time]
[[quarkus-camel-langchain4j-ingest-pipeline-name-max-document-size]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-max-document-size[quarkus.camel.langchain4j.ingest."pipeline-name".max-document-size]`
-Maximum size of one document in characters, applied to the text about to be
split;
-0, the default, means no limit. The pipeline holds a document in memory whole,
so the
-cap is the protection against oversized — on a consumer-fed pipeline,
attacker-sized —
-payloads. An oversized document fails the exchange cleanly.
+Maximum size of one document in characters, applied to the text about to be
split, or
+in bytes with `modality=media`; 0, the default, means no limit. The pipeline
holds a
+document in memory whole, so the cap is the protection against oversized — on a
+consumer-fed pipeline, attacker-sized — payloads. An oversized document fails
the
+exchange cleanly.
| `int`
| `0`
@@ -386,6 +448,39 @@ uses the file name, and one consuming from a component
uses the
`CamelIngestDocumentId` is still read as a fallback).
| `string`
|
+
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-filter-include-id]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-filter-include-id[quarkus.camel.langchain4j.ingest."pipeline-name".filter.include-id]`
+
+Comma-separated Ant-style patterns the document id must match to be ingested,
+for example `++**++/++*++.pdf,++**++/++*++.md` (`++*++.pdf` alone misses files
in
+subdirectories). Matching is case-sensitive, and an id starting with `/` needs
a
+pattern starting with `/`. A non-matching delivery is rejected before the
sink's dedup
+claim, without reading the body and, with a `parser`, before the parse. When
not set,
+every id is accepted.
+| `string`
+|
+
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-filter-exclude-id]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-filter-exclude-id[quarkus.camel.langchain4j.ingest."pipeline-name".filter.exclude-id]`
+
+Comma-separated Ant-style patterns for document ids to skip, for example
+`++**++/draft-++*++`. Exclusion wins over `include-id`.
+| `string`
+|
+
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-filter-min-document-size]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-filter-min-document-size[quarkus.camel.langchain4j.ingest."pipeline-name".filter.min-document-size]`
+
+Minimum size of one document in characters, in bytes with `modality=media`; 0,
the
+default, means no minimum. A shorter document is answered `filtered` instead
of being
+written.
+| `int`
+| `0`
+
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-filter-document-filter]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-filter-document-filter[quarkus.camel.langchain4j.ingest."pipeline-name".filter.document-filter]`
+
+Name of a Camel `Predicate` bean deciding whether a delivery is ingested,
+evaluated with the body available. Looked up by name only.
+| `string`
+|
|===
[.configuration-legend]
diff --git
a/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
b/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
index 7455e47c44..d90226088f 100644
---
a/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
+++
b/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
@@ -290,16 +290,18 @@ class Langchain4jIngestProcessor {
void validatePipelines(IngestBuildTimeConfig config,
BuildProducer<ValidationErrorBuildItem> validationErrors) {
for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig>
entry : config.pipelines().entrySet()) {
IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline =
entry.getValue();
+ // as in the component, the splitter sizes and the batch size are
not validated where they do not apply
+ boolean media = "media".equalsIgnoreCase(pipeline.modality());
- if (pipeline.maxSegmentSize() <= 0 || pipeline.maxOverlapSize() < 0
- || pipeline.maxOverlapSize() >= pipeline.maxSegmentSize())
{
+ if (!media && (pipeline.maxSegmentSize() <= 0 ||
pipeline.maxOverlapSize() < 0
+ || pipeline.maxOverlapSize() >=
pipeline.maxSegmentSize())) {
validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
"Ingestion pipeline '" + entry.getKey() + "':
max-segment-size must be positive and "
+ "max-overlap-size must be smaller than it
(got " + pipeline.maxSegmentSize()
+ " / " + pipeline.maxOverlapSize() + ")")));
}
- if (pipeline.embeddingBatchSize() < 1) {
+ if (!media && pipeline.embeddingBatchSize() < 1) {
validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
"Ingestion pipeline '" + entry.getKey() + "':
embedding-batch-size must be positive (got "
+ pipeline.embeddingBatchSize() + ")")));
@@ -310,6 +312,35 @@ class Langchain4jIngestProcessor {
"Ingestion pipeline '" + entry.getKey() + "':
max-document-size must not be negative, 0 "
+ "meaning no limit (got " +
pipeline.maxDocumentSize() + ")")));
}
+
+ // the component and IngestRoutes check these at startup; a
configured pipeline fails the build instead
+ if (!media && !"text".equalsIgnoreCase(pipeline.modality())) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "': modality
must be 'text' or 'media' (got '"
+ + pipeline.modality() + "')")));
+ }
+ if (media && pipeline.parser().isPresent()) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' sets
modality 'media' together with a parser. A "
+ + "media document is embedded whole and never
parsed; remove one of them.")));
+ }
+ if (media && pipeline.documentSplitter().isPresent()) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' sets
modality 'media' together with a "
+ + "document-splitter. A media document is
embedded whole and never split; remove one "
+ + "of them.")));
+ }
+ if (!media && pipeline.contentType().isPresent()) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "':
content-type only applies to modality 'media' "
+ + "(got '" + pipeline.contentType().get() +
"'). Set modality=media, or remove it.")));
+ }
+ // the component drops the parameters after ';', so a value with
nothing before them would silently be unset
+ if (pipeline.contentType().isPresent() &&
pipeline.contentType().get().split(";", 2)[0].isBlank()) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "':
content-type must not be blank or only parameters "
+ + "(got '" + pipeline.contentType().get() +
"')")));
+ }
}
}
}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderFilterTypoTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderFilterTypoTest.java
new file mode 100644
index 0000000000..3733962a24
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderFilterTypoTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Source;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * filter.* meant for an {@code @Ingest} pipeline under a misspelled name
declares a pipeline of its own; the start
+ * fails naming the Java-declared pipelines, not just the missing directory.
+ */
+class IngestBuilderFilterTypoTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(Pipelines.class,
TestEmbeddingBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.datasheet.filter.exclude-id",
"secret-*")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "Ingestion pipeline 'datasheet' has no source.directory,
and no @Ingest method declares it",
+ "(declared in Java: [datasheets])"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ @Ingest("datasheets")
+ IngestPipeline datasheets() {
+ return IngestPipeline.from(Source.file("target/filter-typo-docs"))
+ .embeddingStore("store")
+ .embeddingModel("model");
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaParserConflictTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaParserConflictTest.java
new file mode 100644
index 0000000000..bdd55c54f2
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaParserConflictTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+import org.apache.camel.Component;
+import org.apache.camel.component.file.FileComponent;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Source;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * A media document is embedded whole and never parsed, so modality media
together with a parser
+ * fails the start.
+ */
+class IngestMediaParserConflictTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(Pipelines.class,
TestEmbeddingBeans.class))
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "Ingestion pipeline 'docs' sets modality 'media' together
with a parser"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ // stands in for camel-quarkus-tika, absent here, so the start gets
past the parser
+ // presence check to the conflict
+ @Produces
+ @Singleton
+ @Named("tika")
+ Component tika() {
+ return new FileComponent();
+ }
+
+ @Ingest("docs")
+ IngestPipeline docs() {
+ return IngestPipeline.from(Source.file("target/media-parser-docs"))
+ .parser("tika")
+ .modality("media")
+ .embeddingStore("store")
+ .embeddingModel("model");
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaSplitterSettingsTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaSplitterSettingsTest.java
new file mode 100644
index 0000000000..b93604b295
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMediaSplitterSettingsTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import dev.langchain4j.data.embedding.Embedding;
+import dev.langchain4j.data.message.ContentType;
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.model.output.Response;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import dev.langchain4j.store.embedding.inmemory.InMemoryEmbeddingStore;
+import io.quarkus.test.QuarkusExtensionTest;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.apache.camel.CamelContext;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * A media document is embedded whole, so the splitter sizes and the batch
size do not apply: settings copied from a
+ * text pipeline are not validated, neither at build time nor by the
component, and the pipeline starts.
+ */
+class IngestMediaSplitterSettingsTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(MediaBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.clips.source.directory",
"target/media-clips")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.clips.modality", "media")
+ // both invalid for text: an overlap above the default segment
size of 500, and no batch
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.clips.max-overlap-size",
"600")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.clips.embedding-batch-size",
"0");
+
+ @Inject
+ CamelContext context;
+
+ @Test
+ void mediaPipelineStarts() {
+ Assertions.assertNotNull(context.getRoute("langchain4j-ingest-clips"));
+ }
+
+ @ApplicationScoped
+ public static class MediaBeans {
+
+ @Produces
+ @Singleton
+ EmbeddingStore<TextSegment> store() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ @Produces
+ @Singleton
+ EmbeddingModel model() {
+ return new EmbeddingModel() {
+ @Override
+ public Response<List<Embedding>> embedAll(List<TextSegment>
segments) {
+ return Response.from(segments.stream()
+ .map(segment -> Embedding.from(new float[] { 1f }))
+ .collect(Collectors.toList()));
+ }
+
+ @Override
+ public Set<ContentType> supportedContentTypes() {
+ return Set.of(ContentType.AUDIO);
+ }
+ };
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingDocumentFilterTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingDocumentFilterTest.java
new file mode 100644
index 0000000000..a966d830c0
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingDocumentFilterTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * The document filter is resolved by name when the route starts, like the
idempotent repository, and a missing bean
+ * fails the start naming the pipeline and the bean.
+ */
+class IngestMissingDocumentFilterTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.directory",
"target/filter-docs")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.filter.document-filter",
"no-such-filter")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "Ingestion pipeline 'docs' references document filter
'no-such-filter'", "no such bean exists"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestModalityConfigTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestModalityConfigTest.java
new file mode 100644
index 0000000000..69c84b6769
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestModalityConfigTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** The modality and content-type rules are validated at build time on the
configuration path. */
+class IngestModalityConfigTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> {
+ })
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.unknown.modality", "sound")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.parsed.modality", "media")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.parsed.parser", "tika")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.split.modality", "MEDIA")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.split.document-splitter",
"splitter")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.typed.content-type",
"audio/wav")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.params.modality", "media")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.params.content-type", ";
codecs=opus")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "'unknown': modality must be 'text' or 'media' (got
'sound')",
+ "'parsed' sets modality 'media' together with a parser",
+ "'split' sets modality 'media' together with a
document-splitter",
+ "'typed': content-type only applies to modality 'media'
(got 'audio/wav')",
+ "'params': content-type must not be blank or only
parameters (got '; codecs=opus')"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineModalityValidationTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineModalityValidationTest.java
new file mode 100644
index 0000000000..f0d706ef79
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineModalityValidationTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Source;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** The builder rejects a bad modality or a blank content type eagerly, like
the configuration path. */
+class IngestPipelineModalityValidationTest {
+
+ @Test
+ void unknownOrMissingModalityRejected() {
+ for (String modality : new String[] { "sound", null }) {
+ IllegalArgumentException e =
assertThrows(IllegalArgumentException.class,
+ () ->
IngestPipeline.from(Source.file("target/x")).modality(modality));
+ assertTrue(e.getMessage().contains("modality must be 'text' or
'media'"), e.getMessage());
+ }
+ }
+
+ @Test
+ void blankContentTypeRejected() {
+ // parameters alone would be dropped by the component, leaving the
content type unset
+ for (String contentType : new String[] { " ", null, "; codecs=opus" })
{
+ IllegalArgumentException e =
assertThrows(IllegalArgumentException.class,
+ () ->
IngestPipeline.from(Source.file("target/x")).contentType(contentType));
+ assertTrue(e.getMessage().contains("content-type must not be blank
or only parameters"), e.getMessage());
+ }
+ }
+
+ @Test
+ void validValuesAccepted() {
+ assertDoesNotThrow(() -> IngestPipeline.from(Source.file("target/x"))
+ .modality("text")
+ .modality("Media")
+ .contentType("audio/wav"));
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
index 532a4eb694..ef5383976e 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
+++ b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
@@ -13,7 +13,7 @@
quarkus.camel.langchain4j.ingest.products.embedding-model=my-model
Each file is read as UTF-8 text — unless a `parser` turns binary formats into
text first, see below — split into overlapping segments (`max-segment-size`,
`max-overlap-size`; or a `DocumentSplitter` CDI bean named by
`document-splitter`, which then replaces the sizes), embedded in batches
(`embedding-batch-size`, 32 by default) and written to the store; a document is
held in memory whole while it is split, so `max-document-size` can cap it in
characters — 0, the default, means no limit, [...]
`embedding-store` and `embedding-model` name CDI beans and may be omitted when
the application has exactly one of each.
Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id`
metadata, so retrieval can cite which document an answer came from (up to
3.39.x the keys were `camel_quarkus_pipeline` and `camel_quarkus_document_id` —
see the migration guide).
-Apart from `enabled` and the `source.*` settings shown here, properties are
fixed at build time.
+Apart from `enabled`, `filter.++*++` and the `source.++*++` settings shown
here, properties are fixed at build time.
A pipeline declared through runtime properties alone — nothing but a
`source.directory`, say — is invisible to build-time validation; its checks,
including the clash with an equally named `@Ingest` pipeline, report at startup
instead.
Each pipeline is composed from the catalog's `langchain4j-ingest-file-source`,
parser action and `langchain4j-ingest-sink` Kamelets: its route
`langchain4j-ingest-<name>` plus one route per Kamelet it uses,
`langchain4j-ingest-<name>-source` (directory pipelines), `-parser` (with a
`parser`) and `-sink`.
The pipelines always load these Kamelets from the classpath, whatever
`camel.component.kamelet.location` says; shipping an equally named Kamelet
replaces internal steps — including the identity capture — and is advanced
surgery.
@@ -28,6 +28,23 @@ Replace and delete arrive with the synchronising engine in a
later release.
What is remembered is which documents were already ingested — the idempotent
repository below.
====
+=== Filtering deliveries
+
+Per pipeline, `filter.include-id` and `filter.exclude-id` take comma-separated
Ant-style patterns matched against the document id — exclusion wins, and a
rejected delivery never takes the sink's dedup claim nor reads the body.
+Matching is case-sensitive, and an id starting with `/` (an absolute path,
say) only matches a pattern that starts with `/` too.
+With a `parser` they are checked before the parse, so an excluded document is
never parsed nor sent to Docling Serve.
+`filter.min-document-size` rejects documents too short to carry retrievable
content, and `filter.document-filter` names a Camel `Predicate` bean evaluated
with the body available.
+With a `parser` these two see the parsed text, so a document they reject has
been parsed first — with `docling`, by the remote Docling Serve.
+A rejected delivery is answered with a `filtered` outcome; on a directory
pipeline it is logged at DEBUG and the file is not read again until it changes,
its key held in memory even with a persistent register: after a restart, a
changed `filter.++*++` applies to it.
+An `@Ingest` pipeline is filtered the same way: the builder has no filter
methods, and the `filter.*` properties apply to it too.
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.filter.include-id=**/*.pdf,**/*.md
+quarkus.camel.langchain4j.ingest.products.filter.exclude-id=**/draft-*
+quarkus.camel.langchain4j.ingest.products.filter.min-document-size=100
+----
+
=== The idempotent repository
A directory pipeline registers what it ingested, keyed on the file's path,
modification time and size: unchanged files are skipped, edited files re-ingest
(their previous segments remain, see the note above).
@@ -120,6 +137,26 @@ A missing parser extension fails the build naming the
artifact, the same way a m
The Java twin is `IngestPipeline.parser("tika")`, and a consumer-fed pipeline
parses the same way — the payload the consumer delivers goes to the parser
first.
With a parser, `max-document-size` also rejects a raw payload larger than the
cap (in bytes) before it reaches the parser, on top of capping the extracted
text (in characters).
+=== Ingesting media: images, audio, video and PDF
+
+`modality=media` embeds each document whole, as one vector, instead of reading
it as text: the bytes go to a multimodal embedding model as audio, an image,
video or a PDF, told apart by the MIME type, and the vector is stored with a
placeholder segment whose text is the document id, stamped with the usual
identity metadata, so retrieval cites it like any other.
+The model must declare the matching content type in its
`supportedContentTypes()`: the pipeline fails to start with a text-only model,
and a document whose medium the model does not declare fails before its bytes
are read.
+`parser` and `document-splitter` must not be set, the splitter sizes and
`embedding-batch-size` do not apply, and `max-document-size` and
`filter.min-document-size` count bytes.
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.photos.source.directory=/var/data/photos
+quarkus.camel.langchain4j.ingest.photos.modality=media
+quarkus.camel.langchain4j.ingest.photos.embedding-model=my-multimodal-model
+quarkus.camel.langchain4j.ingest.photos.embedding-store=photos
+quarkus.camel.langchain4j.ingest.photos.filter.include-id=**/*.png,**/*.jpg
+----
+
+The MIME type handed to the model comes from `content-type` or, when unset,
from the file extension through Camel's MIME table (wav, mp3, flac, ogg, opus,
m4a, aac and aiff for audio; png, jpg, gif and webp for images; mp4, mov and
webm for video; pdf); a file whose type cannot be determined fails the
exchange, and `content-type` without `modality=media` is rejected like the
parser and splitter conflicts.
+On a directory pipeline such a file — `notes.txt`, `Thumbs.db` — fails again
on every poll, so restrict the ids with `filter.include-id`, as above: it is
checked before the media type.
+Images are embedded today by several LangChain4j providers; no provider embeds
audio out of the box yet, so an audio model is a CDI bean of your own that
declares `AUDIO` — typically a client of an audio embedding service — and
either way the same bean embeds the query image or clip at retrieval time.
+The Java twin is `IngestPipeline.modality("media")`, and a consumer-fed
pipeline embeds the delivered payload the same way.
+
=== Other sources, declared in Java
Any Camel consumer can feed a pipeline — the roughly 300 components, each with
its own options and its own documentation.
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
index 98742f8d64..985c966bf6 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
@@ -46,6 +46,9 @@ public interface IngestBuildTimeConfig {
/** Mirrors the {@code @WithDefault} below, which can only carry a
literal; 0 means no limit. */
int DEFAULT_MAX_DOCUMENT_SIZE = 0;
+ /** Mirrors the {@code @WithDefault} below, which can only carry a
literal. */
+ String DEFAULT_MODALITY = "text";
+
/**
* Ingestion pipelines by name.
*/
@@ -70,6 +73,27 @@ public interface IngestBuildTimeConfig {
*/
Optional<String> parser();
+ /**
+ * What the consumed payload is. `text`, the default, is read as a
String, split into
+ * segments and embedded segment by segment. `media` is read as bytes
and embedded whole,
+ * as one vector: audio, an image, video or a PDF, told apart by the
MIME type, each
+ * needing an embedding model that declares the matching content type
— the pipeline
+ * fails to start with a text-only model. In media mode `parser` and
`document-splitter`
+ * must not be set, the splitter sizes and `embedding-batch-size` do
not apply, and
+ * `max-document-size` and `filter.min-document-size` count bytes.
+ */
+ @WithDefault("text")
+ String modality();
+
+ /**
+ * MIME type of a media payload, such as `audio/wav` or `image/png`,
handed to the
+ * embedding model. When not set, it is derived from the document id's
file extension
+ * through 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`: set without it, the pipeline is rejected.
+ */
+ Optional<String> contentType();
+
/**
* Name of the `EmbeddingStore` bean to write to. When not set, the
only one present is
* used.
@@ -105,10 +129,11 @@ public interface IngestBuildTimeConfig {
int embeddingBatchSize();
/**
- * Maximum size of one document in characters, applied to the text
about to be split;
- * 0, the default, means no limit. The pipeline holds a document in
memory whole, so the
- * cap is the protection against oversized — on a consumer-fed
pipeline, attacker-sized —
- * payloads. An oversized document fails the exchange cleanly.
+ * Maximum size of one document in characters, applied to the text
about to be split, or
+ * in bytes with `modality=media`; 0, the default, means no limit. The
pipeline holds a
+ * document in memory whole, so the cap is the protection against
oversized — on a
+ * consumer-fed pipeline, attacker-sized — payloads. An oversized
document fails the
+ * exchange cleanly.
*/
@WithDefault("0")
int maxDocumentSize();
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
index 42492f393e..4f0b847736 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
@@ -19,6 +19,8 @@ package org.apache.camel.quarkus.component.langchain4j.ingest;
import java.util.Optional;
import java.util.Set;
+import
org.apache.camel.quarkus.component.langchain4j.ingest.IngestRunTimeConfig.PipelineRunTimeConfig.FilterRunTimeConfig;
+
/**
* A pipeline declared in Java rather than in configuration, returned from an
{@link Ingest}
* method. Every property has a configuration twin, and both paths share the
same runtime.
@@ -28,10 +30,36 @@ public final class IngestPipeline {
/** The values {@link #parser(String)} and the {@code parser}
configuration property accept. */
public static final Set<String> SUPPORTED_PARSERS = Parser.NAMES;
+ /** The filters of a pipeline configuration says nothing about: every
delivery is accepted. */
+ static final FilterRunTimeConfig NO_FILTER = new FilterRunTimeConfig() {
+
+ @Override
+ public Optional<String> includeId() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> excludeId() {
+ return Optional.empty();
+ }
+
+ @Override
+ public int minDocumentSize() {
+ return 0;
+ }
+
+ @Override
+ public Optional<String> documentFilter() {
+ return Optional.empty();
+ }
+ };
+
private final Source source;
private String embeddingStoreName;
private String embeddingModelName;
private String parser;
+ private String modality = IngestBuildTimeConfig.DEFAULT_MODALITY;
+ private String contentType;
private int maxSegmentSize =
IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE;
private int maxOverlapSize =
IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE;
private int embeddingBatchSize =
IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE;
@@ -100,8 +128,8 @@ public final class IngestPipeline {
}
/**
- * Maximum size of one document in characters; 0, the default, means no
limit. The twin of
- * the {@code max-document-size} configuration property.
+ * Maximum size of one document in characters, in bytes with {@code
modality("media")}; 0, the
+ * default, means no limit. The twin of the {@code max-document-size}
configuration property.
*/
public IngestPipeline maxDocumentSize(int maxDocumentSize) {
// the same rule the configuration path is held to at build time
@@ -123,6 +151,36 @@ public final class IngestPipeline {
return this;
}
+ /**
+ * What the consumed payload is: {@code text}, the default, is split and
embedded segment by
+ * segment; {@code media} (audio, an image, video or a PDF, told apart by
the MIME type) is
+ * embedded whole, as one vector, by a model that declares the matching
content type. The
+ * twin of the {@code modality} configuration property.
+ */
+ public IngestPipeline modality(String modality) {
+ // the same rule the configuration path is held to at build time
+ if (!"text".equalsIgnoreCase(modality) &&
!"media".equalsIgnoreCase(modality)) {
+ throw new IllegalArgumentException("modality must be 'text' or
'media' (got '" + modality + "')");
+ }
+ this.modality = modality;
+ return this;
+ }
+
+ /**
+ * MIME type of a media payload, such as {@code audio/wav}; unset, it is
derived from the
+ * document id's file extension. The twin of the {@code content-type}
configuration property.
+ */
+ public IngestPipeline contentType(String contentType) {
+ // the same rule the configuration path is held to at build time: the
component drops the
+ // parameters after ';', so a value with nothing before them would
silently be unset
+ if (contentType == null || contentType.split(";", 2)[0].isBlank()) {
+ throw new IllegalArgumentException(
+ "content-type must not be blank or only parameters (got '"
+ contentType + "')");
+ }
+ this.contentType = contentType;
+ return this;
+ }
+
String sourceType() {
return source.type();
}
@@ -143,6 +201,14 @@ public final class IngestPipeline {
return Optional.ofNullable(parser);
}
+ String modality() {
+ return modality;
+ }
+
+ Optional<String> contentType() {
+ return Optional.ofNullable(contentType);
+ }
+
int maxSegmentSize() {
return maxSegmentSize;
}
@@ -163,8 +229,11 @@ public final class IngestPipeline {
return Optional.ofNullable(documentSplitterName);
}
- /** The configuration view, so a builder pipeline reuses every
configuration path verbatim. */
- IngestRunTimeConfig.PipelineRunTimeConfig asRunTimeConfig() {
+ /**
+ * The configuration view, so a builder pipeline reuses the configuration
paths verbatim. Its
+ * filter is the configured one, or none: the builder has no filter API.
+ */
+ IngestRunTimeConfig.PipelineRunTimeConfig
asRunTimeConfig(IngestRunTimeConfig.PipelineRunTimeConfig configured) {
IngestRunTimeConfig.PipelineRunTimeConfig.SourceRunTimeConfig
sourceConfig = source.asRunTimeConfig();
return new IngestRunTimeConfig.PipelineRunTimeConfig() {
@@ -177,6 +246,11 @@ public final class IngestPipeline {
public SourceRunTimeConfig source() {
return sourceConfig;
}
+
+ @Override
+ public FilterRunTimeConfig filter() {
+ return configured == null ? NO_FILTER : configured.filter();
+ }
};
}
}
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
index 0c895937b2..46e8699a09 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
@@ -39,9 +39,12 @@ import jakarta.enterprise.inject.literal.NamedLiteral;
import jakarta.inject.Inject;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Expression;
+import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.file.GenericFile;
import org.apache.camel.component.langchain4j.ingest.IngestResult;
import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders;
@@ -50,6 +53,9 @@ import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.builder.ExpressionBuilder;
import org.apache.camel.support.builder.PredicateBuilder;
import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.AntPathMatcher;
import org.apache.camel.util.URISupport;
import org.jboss.logging.Logger;
@@ -123,6 +129,13 @@ public class IngestRoutes extends RouteBuilder {
LOG.infof("Ingestion pipeline '%s' is disabled", name);
continue;
}
+ // runtime keys alone name it and none is its source: most likely
filter.* meant for an
+ // @Ingest pipeline under a misspelled name
+ if (pipeline == null && !builderDeclared.isEmpty() &&
runtime.source().directory().isEmpty()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ "' has no source.directory, and no"
+ + " @Ingest method declares it (declared in Java: " +
new TreeSet<>(builderDeclared)
+ + "). Fix the name, or set
quarkus.camel.langchain4j.ingest." + name + ".source.directory");
+ }
// a consumer URI says "consume from this"; its absence says "read
that directory"
String uri = pipeline == null ? null :
pipeline.source().uri().orElse(null);
@@ -140,7 +153,9 @@ public class IngestRoutes extends RouteBuilder {
pipeline == null ?
IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE :
pipeline.embeddingBatchSize(),
pipeline == null ?
IngestBuildTimeConfig.DEFAULT_MAX_DOCUMENT_SIZE : pipeline.maxDocumentSize(),
pipeline == null ? null :
pipeline.documentSplitter().orElse(null),
- pipeline == null ? null : pipeline.parser().orElse(null));
+ pipeline == null ? null : pipeline.parser().orElse(null),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MODALITY
: pipeline.modality(),
+ pipeline == null ? null :
pipeline.contentType().orElse(null));
}
for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
@@ -182,7 +197,7 @@ public class IngestRoutes extends RouteBuilder {
LOG.infof("Ingestion pipeline '%s' (builder) is disabled", name);
return;
}
- // enabled is the one thing configuration may say about a builder
pipeline; anything about
+ // enabled and filter.* are what configuration may say about a builder
pipeline; anything about
// its source would be quietly overruled by the @Ingest method, so it
is an error instead
// (source.recursive cannot be told apart from its default, so it
alone goes undetected -
// Source.recursive() is its builder twin)
@@ -197,7 +212,7 @@ public class IngestRoutes extends RouteBuilder {
IngestPipeline definition = builderPipelines.definition(entry);
String uri = "file".equals(definition.sourceType()) ? null :
definition.sourceUri();
- compositionRoute(name, uri, definition.asRunTimeConfig(),
+ compositionRoute(name, uri, definition.asRunTimeConfig(external),
resolveStore(name,
definition.embeddingStoreName().orElse(null)),
resolveModel(name,
definition.embeddingModelName().orElse(null)),
definition.maxSegmentSize(),
@@ -205,7 +220,9 @@ public class IngestRoutes extends RouteBuilder {
definition.embeddingBatchSize(),
definition.maxDocumentSize(),
definition.documentSplitterName().orElse(null),
- definition.parser().orElse(null));
+ definition.parser().orElse(null),
+ definition.modality(),
+ definition.contentType().orElse(null));
}
/**
@@ -216,7 +233,7 @@ public class IngestRoutes extends RouteBuilder {
IngestRunTimeConfig.PipelineRunTimeConfig runtime,
EmbeddingStore<TextSegment> store, EmbeddingModel model,
int maxSegmentSize, int maxOverlapSize, int embeddingBatchSize,
int maxDocumentSize,
- String documentSplitterName, String parser) {
+ String documentSplitterName, String parser, String modality,
String contentType) {
// the name is substituted into Kamelet URIs and registry references,
so both declaration
// styles are held to one charset; @Ingest names are already checked
at build time
@@ -224,6 +241,15 @@ public class IngestRoutes extends RouteBuilder {
throw new IllegalStateException(
"Ingestion pipeline name '" + name + "' may only contain
letters, digits, '.', '_' and '-'");
}
+ // the component validates modality and content type; that a media
document is never
+ // parsed is a rule of this composition, where the parser action sits
before the sink.
+ // A configured pipeline fails the build on both already
+ boolean media = "media".equalsIgnoreCase(modality);
+ if (media && parser != null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "' sets modality 'media' together with a parser. A media
document is embedded whole and"
+ + " never parsed; remove one of them.");
+ }
String storeRef = bindInstance(name, "store", store);
String modelRef = bindInstance(name, "model", model);
String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
@@ -241,17 +267,35 @@ public class IngestRoutes extends RouteBuilder {
sink.put("maxSegmentSize", String.valueOf(maxSegmentSize));
sink.put("maxOverlapSize", String.valueOf(maxOverlapSize));
sink.put("embeddingBatchSize", String.valueOf(embeddingBatchSize));
- sink.put("modality", "text");
- sink.put("contentType", "");
+ // lower-cased, as the Kamelet declares it, rather than relying on a
case-insensitive enum conversion
+ sink.put("modality", modality.toLowerCase(Locale.ROOT));
+ sink.put("contentType", contentType == null ? "" : contentType);
// 0, no limit, is the endpoint's default too
sink.put("maxDocumentSize", String.valueOf(maxDocumentSize));
- sink.put("minDocumentSize", "0");
if (documentSplitterName != null) {
sink.put("documentSplitter", "#bean:" + documentSplitterName);
}
- sink.put("includeId", "");
- sink.put("excludeId", "");
- sink.put("documentFilter", "#bean:" + bindInstance(name,
"document-filter", PredicateBuilder.constant(true)));
+ // filter.* comes from configuration for both declaration styles - a
builder pipeline's view
+ // delegates to it, the builder having no filter API - and the
component enforces it
+ IngestRunTimeConfig.PipelineRunTimeConfig.FilterRunTimeConfig filter =
runtime == null
+ ? IngestPipeline.NO_FILTER
+ : runtime.filter();
+ String includeId = filter.includeId().orElse("");
+ String excludeId = filter.excludeId().orElse("");
+ sink.put("includeId", includeId);
+ sink.put("excludeId", excludeId);
+ sink.put("minDocumentSize", String.valueOf(filter.minDocumentSize()));
+ String documentFilter = filter.documentFilter().orElse(null);
+ // checked up front, like the idempotent repository: the Kamelet's
property binding would
+ // fail naming neither the pipeline nor the property
+ if (documentFilter != null
+ &&
getContext().getRegistry().lookupByNameAndType(documentFilter, Predicate.class)
== null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "'
references document filter '"
+ + documentFilter + "' but no such bean exists");
+ }
+ sink.put("documentFilter", "#bean:" + (documentFilter != null
+ ? documentFilter
+ : bindInstance(name, "document-filter",
PredicateBuilder.constant(true))));
sink.put("embeddingStore", "#bean:" + storeRef);
sink.put("embeddingModel", "#bean:" + modelRef);
@@ -269,28 +313,38 @@ public class IngestRoutes extends RouteBuilder {
source.put("recursive",
String.valueOf(runtime.source().recursive()));
// the file consumer's own default poll delay
source.put("delay", "500");
- if (parser == null) {
- // text is read as UTF-8; a parser receives the raw bytes
instead - the format is
- // its business, and a charset conversion would corrupt a
binary document
+ if (parser == null && !media) {
+ // text is read as UTF-8; a parser or a media model receives
the raw bytes instead -
+ // the format is its business, and a charset conversion would
corrupt a binary document
source.put("charset", "UTF-8");
}
source.put("idempotentRepository", "#bean:" + repositoryRef);
ProcessorDefinition<?> route = from(kameletUri(name,
FILE_SOURCE_KAMELET, "source", source))
.routeId(id(name));
+ if (parser != null || media) {
+ // the file endpoint has no neutral charset to pin, so an
application-wide
+ // camel.kamelet.langchain4j-ingest-file-source.charset still
reaches it; dropped
+ // before any step reads the file, it can neither fail nor
silently corrupt the bytes
+ route = route.process(IngestRoutes::dropCharset);
+ }
if (documentId != null) {
// override the source's file-name default; captured before
any further step
route = route.setHeader(LangChain4jIngestHeaders.DOCUMENT_ID,
documentIdExpression(documentId));
}
// no duplicate pre-check: the source register already filtered
duplicates out
- route = parseSteps(route, name, parser, maxDocumentSize, true,
null);
+ route = parseSteps(route, name, parser, maxDocumentSize, true,
null, includeId, excludeId);
// the file consumer discards the reply and the source register
already keeps the
// same file version from being ingested twice, so no repository
goes to the sink.
// The discarded reply would hide an empty outcome, so it is
logged: warned with a
// parser, since a parse to nothing typically means a missing Tika
parser module or an
- // image-only document, and the file's register key is committed
+ // image-only document, and the file's register key is committed.
A filtered one is
+ // logged at DEBUG, its key kept in memory only
route.to(kameletUri(name, SINK_KAMELET, "sink",
sink)).process(exchange -> {
IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
+ if (result != null && result.outcome() ==
IngestResult.Outcome.FILTERED) {
+ fileFiltered(exchange, name, result.documentId());
+ }
if (result == null || result.outcome() !=
IngestResult.Outcome.EMPTY) {
return;
}
@@ -332,7 +386,7 @@ public class IngestRoutes extends RouteBuilder {
IdempotentRepository register = repositoryRef == null
? null
:
getContext().getRegistry().lookupByNameAndType(repositoryRef,
IdempotentRepository.class);
- route = parseSteps(route, name, parser, maxDocumentSize, false,
register);
+ route = parseSteps(route, name, parser, maxDocumentSize, false,
register, includeId, excludeId);
if (repositoryRef != null) {
// deduplication by document id happens inside the sink's
producer: a duplicate
// is answered SKIPPED, a blank delivery releases its claim
@@ -344,13 +398,13 @@ public class IngestRoutes extends RouteBuilder {
}
/**
- * The optional parse stage: the id guard, the advisory duplicate check
and the raw-size
- * guard, then the parser action Kamelet, which captures the document id
from the canonical
- * header into the exchange property before the parse. The route is
returned unchanged when
- * the pipeline has no parser.
+ * The optional parse stage: the id guard, the advisory id-pattern and
duplicate checks and the
+ * raw-size guard, then the parser action Kamelet, which captures the
document id from the
+ * canonical header into the exchange property before the parse. The route
is returned
+ * unchanged when the pipeline has no parser.
*/
private static ProcessorDefinition<?> parseSteps(ProcessorDefinition<?>
route, String name, String parser,
- int maxDocumentSize, boolean directory, IdempotentRepository
register) {
+ int maxDocumentSize, boolean directory, IdempotentRepository
register, String includeId, String excludeId) {
if (parser == null) {
return route;
}
@@ -366,6 +420,29 @@ public class IngestRoutes extends RouteBuilder {
+ " the parse - a parsed document must not supply its
own identity");
}
});
+ String[] include = patterns(includeId);
+ String[] exclude = patterns(excludeId);
+ if (include != null || exclude != null) {
+ // advisory too: the sink's id patterns, applied before the parse
so an excluded
+ // document is never parsed, nor fails the parse or the size guard
on every poll.
+ // patterns() and this match mirror the private parsePatterns and
idAccepted of the
+ // component's LangChain4jIngestProducer, which stay authoritative
- keep them in step
+ route = route.choice()
+ .when(exchange -> {
+ String id =
exchange.getMessage().getHeader(documentIdHeader, String.class);
+ return AntPathMatcher.INSTANCE.anyMatch(exclude, id)
+ || (include != null &&
!AntPathMatcher.INSTANCE.anyMatch(include, id));
+ })
+ .process(exchange -> {
+ String id =
exchange.getMessage().getHeader(documentIdHeader, String.class);
+ if (directory) {
+ fileFiltered(exchange, name, id);
+ }
+ exchange.getMessage().setBody(new IngestResult(name,
id, 0, IngestResult.Outcome.FILTERED));
+ })
+ .stop()
+ .end();
+ }
if (register != null) {
// advisory: a known duplicate is answered SKIPPED before the
(possibly remote) parse
// is paid for; the sink producer's eager claim stays
authoritative, so a duplicate
@@ -434,6 +511,37 @@ public class IngestRoutes extends RouteBuilder {
};
}
+ /** Comma-separated id patterns, parsed as the component parses them; null
when none remain. */
+ private static String[] patterns(String value) {
+ String[] patterns = Arrays.stream(value.split(","))
+ .map(String::trim)
+ .filter(pattern -> !pattern.isEmpty())
+ .toArray(String[]::new);
+ return patterns.length == 0 ? null : patterns;
+ }
+
+ /**
+ * A directory pipeline discards the reply, so a filtered document is
logged instead; a named register does not
+ * keep its file's key, see {@link SourceRegister}.
+ */
+ private static void fileFiltered(Exchange exchange, String name, String
documentId) {
+ LOG.debugf("Ingestion pipeline '%s': document '%s' was filtered out;
it is not read again until the file"
+ + " changes or the application restarts", name, documentId);
+ String key = exchange.getProperty(Exchange.FILE_IDEMPOTENT_KEY,
String.class);
+ if (key != null &&
exchange.getContext().getRegistry().lookupByNameAndType(id(name, "register"),
+ IdempotentRepository.class) instanceof SourceRegister
register) {
+ register.filtered(key);
+ }
+ }
+
+ /** As if the file consumer had no charset: a {@code GenericFile} body
then converts to the file's own bytes. */
+ private static void dropCharset(Exchange exchange) {
+ exchange.removeProperty(ExchangePropertyKey.CHARSET_NAME);
+ if (exchange.getMessage().getBody() instanceof GenericFile<?> file) {
+ file.setCharset(null);
+ }
+ }
+
/**
* Built through {@code createQueryString} rather than concatenated, so no
Kamelet property can be injected.
* Every value travels as a {@code RAW(...)} token: {@code
createQueryString} leaves those unencoded and the
@@ -481,7 +589,7 @@ public class IngestRoutes extends RouteBuilder {
* checked up front, with the configuration-level message), an
auto-created in-memory register
* bound under the configured name, or - for a directory pipeline naming
none - a generated
* built-in one, sized above the file endpoint's default so eviction does
not re-ingest large
- * directories.
+ * directories. A directory pipeline's named register is wrapped in a
{@link SourceRegister}.
*/
private String repositoryRef(String name,
IngestRunTimeConfig.PipelineRunTimeConfig runtime, boolean directory) {
String repositoryName = runtime == null ? null :
runtime.source().idempotentRepository().orElse(null);
@@ -499,14 +607,14 @@ public class IngestRoutes extends RouteBuilder {
throw new IllegalStateException("Ingestion pipeline '" +
name
+ "' references idempotent repository '" +
repositoryName + "' but no such bean exists");
}
- getContext().getRegistry().bind(repositoryName,
-
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
+ repository =
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY);
+ getContext().getRegistry().bind(repositoryName, repository);
} else {
// a CDI-produced repository does not pass through the
registry's bind hook, so a
// CamelContextAware implementation would otherwise run
contextless
CamelContextAware.trySetCamelContext(repository, getContext());
}
- return repositoryName;
+ return directory ? bindInstance(name, "register", new
SourceRegister(repository)) : repositoryName;
}
if (directory) {
String ref = id(name, "register");
@@ -618,4 +726,63 @@ public class IngestRoutes extends RouteBuilder {
}
return handles.get(0).get();
}
+
+ /**
+ * A directory pipeline's named register, which may be persistent. The key
of a file the pipeline answers
+ * filtered is held in memory instead: the file is skipped while the
application runs and checked again after a
+ * restart, so a changed {@code filter.*} applies to it. Releasing the key
would read the file on every poll.
+ */
+ private static final class SourceRegister extends ServiceSupport
implements IdempotentRepository {
+
+ private final IdempotentRepository register;
+ private final IdempotentRepository filtered =
MemoryIdempotentRepository
+ .memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY);
+
+ SourceRegister(IdempotentRepository register) {
+ this.register = register;
+ }
+
+ /** Moves a key the file consumer added when it picked the file up;
its commit then finds it here. */
+ void filtered(String key) {
+ filtered.add(key);
+ register.remove(key);
+ }
+
+ @Override
+ public boolean add(String key) {
+ return !filtered.contains(key) && register.add(key);
+ }
+
+ @Override
+ public boolean contains(String key) {
+ return filtered.contains(key) || register.contains(key);
+ }
+
+ @Override
+ public boolean remove(String key) {
+ // from both, never short-circuited
+ return filtered.remove(key) | register.remove(key);
+ }
+
+ @Override
+ public boolean confirm(String key) {
+ return filtered.contains(key) || register.confirm(key);
+ }
+
+ @Override
+ public void clear() {
+ filtered.clear();
+ register.clear();
+ }
+
+ @Override
+ protected void doStart() {
+ ServiceHelper.startService(register);
+ }
+
+ @Override
+ protected void doStop() {
+ ServiceHelper.stopService(register);
+ }
+ }
}
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
index f6320321d5..9fbbabd448 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
@@ -54,6 +54,47 @@ public interface IngestRunTimeConfig {
*/
SourceRunTimeConfig source();
+ /**
+ * Filters deciding which deliveries are ingested. A rejected delivery
is answered with
+ * a `filtered` outcome and never keeps the sink's dedup claim; a
directory pipeline does not
+ * read the file again until it changes, its key held in memory even
with a persistent
+ * register, so after a restart a changed filter applies to it.
+ */
+ FilterRunTimeConfig filter();
+
+ interface FilterRunTimeConfig {
+
+ /**
+ * Comma-separated Ant-style patterns the document id must match
to be ingested,
+ * for example `++**++/++*++.pdf,++**++/++*++.md` (`++*++.pdf`
alone misses files in
+ * subdirectories). Matching is case-sensitive, and an id starting
with `/` needs a
+ * pattern starting with `/`. A non-matching delivery is rejected
before the sink's dedup
+ * claim, without reading the body and, with a `parser`, before
the parse. When not set,
+ * every id is accepted.
+ */
+ Optional<String> includeId();
+
+ /**
+ * Comma-separated Ant-style patterns for document ids to skip,
for example
+ * `++**++/draft-++*++`. Exclusion wins over `include-id`.
+ */
+ Optional<String> excludeId();
+
+ /**
+ * Minimum size of one document in characters, in bytes with
`modality=media`; 0, the
+ * default, means no minimum. A shorter document is answered
`filtered` instead of being
+ * written.
+ */
+ @WithDefault("0")
+ int minDocumentSize();
+
+ /**
+ * Name of a Camel `Predicate` bean deciding whether a delivery is
ingested,
+ * evaluated with the body available. Looked up by name only.
+ */
+ Optional<String> documentFilter();
+ }
+
interface SourceRunTimeConfig {
/**
diff --git
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicAudioEmbeddingModel.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicAudioEmbeddingModel.java
new file mode 100644
index 0000000000..d70cbf5019
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicAudioEmbeddingModel.java
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Set;
+
+import dev.langchain4j.data.embedding.Embedding;
+import dev.langchain4j.data.message.AudioContent;
+import dev.langchain4j.data.message.Content;
+import dev.langchain4j.data.message.ContentType;
+import dev.langchain4j.model.embedding.request.EmbeddingInput;
+import dev.langchain4j.model.embedding.request.EmbeddingRequest;
+import dev.langchain4j.model.embedding.response.EmbeddingResponse;
+
+/**
+ * Deterministic fake that also accepts audio: same bytes, same vector — no
network, no model download, native-friendly.
+ */
+public class DeterministicAudioEmbeddingModel extends
DeterministicEmbeddingModel {
+
+ public DeterministicAudioEmbeddingModel(int dimension) {
+ super(dimension);
+ }
+
+ @Override
+ public Set<ContentType> supportedContentTypes() {
+ return Set.of(ContentType.TEXT, ContentType.AUDIO);
+ }
+
+ @Override
+ public EmbeddingResponse doEmbed(EmbeddingRequest request) {
+ List<Embedding> embeddings = new ArrayList<>(request.inputs().size());
+ for (EmbeddingInput input : request.inputs()) {
+ AudioContent audio = audioOf(input);
+ embeddings.add(embeddingFor(audio == null ? input.text() :
audio.audio().base64Data()));
+ }
+ return EmbeddingResponse.builder().embeddings(embeddings).build();
+ }
+
+ /** The vector this fake produces for the given bytes: what a query by the
same clip embeds to. */
+ public Embedding embeddingOf(byte[] audio) {
+ return embeddingFor(Base64.getEncoder().encodeToString(audio));
+ }
+
+ private static AudioContent audioOf(EmbeddingInput input) {
+ if (input.contents() == null) {
+ return null;
+ }
+ for (Content content : input.contents()) {
+ if (content instanceof AudioContent audio) {
+ return audio;
+ }
+ }
+ return null;
+ }
+}
diff --git
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
index d407e2b599..6909881956 100644
---
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
@@ -111,6 +111,29 @@ public class IngestItProducers {
return new InMemoryEmbeddingStore<>();
}
+ @Produces
+ @Singleton
+ @Named("filtered-store")
+ EmbeddingStore<TextSegment> filteredStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ /** The documentFilter predicate of the filtered pipeline: content marked
confidential stays out. */
+ @Produces
+ @Singleton
+ @Named("confidentialFilter")
+ org.apache.camel.Predicate confidentialFilter() {
+ return exchange ->
!exchange.getMessage().getBody(String.class).contains("CONFIDENTIAL");
+ }
+
+ /** Named only by the application-wide documentFilter property the
pipelines must not pick up. */
+ @Produces
+ @Singleton
+ @Named("rejectAll")
+ org.apache.camel.Predicate rejectAll() {
+ return exchange -> false;
+ }
+
@Produces
@Singleton
@Named("test-model")
@@ -118,6 +141,21 @@ public class IngestItProducers {
return new DeterministicEmbeddingModel(64);
}
+ @Produces
+ @Singleton
+ @Named("audio-store")
+ EmbeddingStore<TextSegment> audioStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ // the concrete type, so the REST resource can embed a query clip with it
+ @Produces
+ @Singleton
+ @Named("audio-model")
+ DeterministicAudioEmbeddingModel audioModel() {
+ return new DeterministicAudioEmbeddingModel(64);
+ }
+
// the custom pipeline's register; auto-create is also set, so this
existing bean must win
@Produces
@Singleton
diff --git
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
index 5db526584b..528f0a3d0a 100644
---
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+import java.io.ByteArrayInputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
@@ -94,6 +95,18 @@ public class IngestResource {
@Named("capped-store")
EmbeddingStore<TextSegment> cappedStore;
+ @Inject
+ @Named("filtered-store")
+ EmbeddingStore<TextSegment> filteredStore;
+
+ @Inject
+ @Named("audio-store")
+ EmbeddingStore<TextSegment> audioStore;
+
+ @Inject
+ @Named("audio-model")
+ DeterministicAudioEmbeddingModel audioModel;
+
@Inject
ProducerTemplate producerTemplate;
@@ -112,6 +125,9 @@ public class IngestResource {
@ConfigProperty(name = "ingest.capped.directory")
String cappedDirectory;
+ @ConfigProperty(name = "ingest.audio.directory")
+ String audioDirectory;
+
/** Asserts a key was committed; registry lookup by name, the same way the
pipelines resolve. */
@GET
@jakarta.ws.rs.Path("/register-contains")
@@ -141,6 +157,7 @@ public class IngestResource {
Path dir = Path.of(switch (pipeline) {
case "scans" -> scansDirectory;
case "capped" -> cappedDirectory;
+ case "audio" -> audioDirectory;
default -> reportsDirectory;
});
Files.createDirectories(dir);
@@ -161,6 +178,7 @@ public class IngestResource {
case "jdbc" -> jdbcStore;
case "reports" -> reportsStore;
case "capped" -> cappedStore;
+ case "filtered" -> filteredStore;
case "scans" -> scansStore;
default -> productsStore;
};
@@ -180,6 +198,25 @@ public class IngestResource {
.toList();
}
+ /** Query by audio: the clip is embedded with the audio model and the
audio store searched for exact hits. */
+ @POST
+ @jakarta.ws.rs.Path("/search/audio")
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ @Produces(MediaType.APPLICATION_JSON)
+ public List<SearchHit> searchAudio(byte[] clip) {
+ var result = audioStore.search(EmbeddingSearchRequest.builder()
+ .queryEmbedding(audioModel.embeddingOf(clip))
+ .maxResults(10)
+ .minScore(0.99)
+ .build());
+ return result.matches().stream()
+ .map(match -> new SearchHit(
+ match.embedded().text(),
+
match.embedded().metadata().getString(LangChain4jIngest.METADATA_PIPELINE),
+
match.embedded().metadata().getString(LangChain4jIngest.METADATA_DOCUMENT_ID)))
+ .toList();
+ }
+
/** One stored segment with the metadata the pipeline stamped on it. */
public record SearchHit(String text, String pipeline, String documentId) {
}
@@ -201,6 +238,28 @@ public class IngestResource {
return result.outcome().label();
}
+ /**
+ * Feeds a media pipeline the raw bytes as a stream, as a streaming
consumer would, optionally announcing a (forged)
+ * CamelFileLength; answers the outcome or the failure message.
+ */
+ @POST
+ @jakarta.ws.rs.Path("/feed-binary/{pipeline}/{documentId:.+}")
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ @Produces(MediaType.TEXT_PLAIN)
+ public String feedBinary(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
+ @QueryParam("fileLength") Long fileLength, byte[] content) {
+ Exchange exchange = producerTemplate.request("direct:" + pipeline +
"-feed", e -> {
+ e.getMessage().setHeader(IngestHeaders.DOCUMENT_ID, documentId);
+ if (fileLength != null) {
+ e.getMessage().setHeader(Exchange.FILE_LENGTH, fileLength);
+ }
+ e.getMessage().setBody(new ByteArrayInputStream(content));
+ });
+ return exchange.getException() != null
+ ? exchange.getException().getMessage()
+ :
exchange.getMessage().getBody(IngestResult.class).outcome().label();
+ }
+
/** Feeds a pipeline carrying different ids in the current and the
deprecated header, so tests can assert precedence. */
@POST
@jakarta.ws.rs.Path("/feed-both/{pipeline}/{documentId}/{legacyDocumentId}")
diff --git
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
index 512853f563..fb47fbe527 100644
---
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
@@ -57,6 +57,20 @@ public class ItBuilderPipelines {
.splitter(120, 20);
}
+ /**
+ * Media with a fixed content type: the ids it is fed carry no extension
to derive one from. The component reads
+ * a stream only one byte past the byte cap.
+ */
+ @Ingest("clips")
+ IngestPipeline clips() {
+ return IngestPipeline.from(Source.endpoint(direct("clips-feed")))
+ .modality("media")
+ .contentType("audio/wav")
+ .maxDocumentSize(20_000)
+ .embeddingStore("audio-store")
+ .embeddingModel("audio-model");
+ }
+
/**
* The lambda form of the Endpoint DSL — nothing imported, the IDE lists
every component off
* {@code dsl.}. The object key identifies the document, and the store
addresses buckets by
diff --git
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
index 2440083c87..1a407c9f8d 100644
---
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
+++
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
@@ -21,10 +21,17 @@ ingest.test.directory=target/ingest-docs
# per test class, and fail the whole module on platforms without a container
environment
quarkus.kafka.devservices.enabled=false
-# meant for an application's own use of these Kamelets: the pipelines pin the
cap, else every
-# parse would exceed it
+# meant for an application's own use of these Kamelets: the pipelines pin
these options, else the
+# text pipelines would fail the start, every parse would exceed the cap, every
document be filtered
+# and media and parsed files be decoded as text
+camel.kamelet.langchain4j-ingest-sink.modality=media
+camel.kamelet.langchain4j-ingest-sink.contentType=audio/wav
+camel.kamelet.langchain4j-ingest-sink.includeId=no-such-id
+camel.kamelet.langchain4j-ingest-sink.excludeId=**
+camel.kamelet.langchain4j-ingest-sink.documentFilter=#bean:rejectAll
camel.kamelet.tika-extract-text-action.maxDocumentSize=1
camel.kamelet.docling-convert-action.maxDocumentSize=1
+camel.kamelet.langchain4j-ingest-file-source.charset=UTF-16
quarkus.camel.langchain4j.ingest.products.source.directory=${ingest.test.directory}
quarkus.camel.langchain4j.ingest.products.embedding-store=products-store
@@ -36,6 +43,8 @@ quarkus.camel.langchain4j.ingest.products.max-overlap-size=20
camel.beans.propertiesRegister=\#class\:org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository
camel.beans.propertiesRegister.cacheSize=5000
quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=propertiesRegister
+# a filtered file's key must stay out of the named register, which could be
persistent
+quarkus.camel.langchain4j.ingest.products.filter.exclude-id=**/draft-*
# parser pipelines: reports extracts PDF text in-process (tika); scans
converts through a
# Docling Serve instance (docling), which the tests stub with WireMock - see
@@ -48,11 +57,13 @@
quarkus.camel.langchain4j.ingest.reports.embedding-model=test-model
quarkus.camel.langchain4j.ingest.reports.max-segment-size=120
quarkus.camel.langchain4j.ingest.reports.max-overlap-size=20
-# parser pipeline with a tight raw-byte cap: exercises the pre-parse size guard
+# parser pipeline with a tight raw-byte cap: exercises the pre-parse size
guard, and with the
+# exclusion, the id patterns applied before it on a directory pipeline
ingest.capped.directory=target/ingest-capped
quarkus.camel.langchain4j.ingest.capped.source.directory=${ingest.capped.directory}
quarkus.camel.langchain4j.ingest.capped.parser=tika
quarkus.camel.langchain4j.ingest.capped.max-document-size=200
+quarkus.camel.langchain4j.ingest.capped.filter.exclude-id=**/*.mp4
quarkus.camel.langchain4j.ingest.capped.embedding-store=capped-store
quarkus.camel.langchain4j.ingest.capped.embedding-model=test-model
quarkus.camel.langchain4j.ingest.capped.max-segment-size=120
@@ -71,9 +82,12 @@ camel.component.docling.use-docling-serve=true
camel.component.docling.docling-serve-url=${docling.serve.url:http://localhost:5001}
# consumer-fed pipeline with a parser: the endpoint+parser combination, and
the pre-parse
-# document-id capture (a parsed document cannot forge its own id through Tika
metadata headers)
+# document-id capture (a parsed document cannot forge its own id through Tika
metadata headers).
+# The cap and the exclusion show that the id patterns apply before the size
guard and the parse
quarkus.camel.langchain4j.ingest.htmlfeed.source.uri=direct:htmlfeed-feed
quarkus.camel.langchain4j.ingest.htmlfeed.parser=tika
+quarkus.camel.langchain4j.ingest.htmlfeed.max-document-size=4096
+quarkus.camel.langchain4j.ingest.htmlfeed.filter.exclude-id=**/*.mp4
quarkus.camel.langchain4j.ingest.htmlfeed.embedding-store=htmlfeed-store
quarkus.camel.langchain4j.ingest.htmlfeed.embedding-model=test-model
quarkus.camel.langchain4j.ingest.htmlfeed.max-segment-size=120
@@ -110,6 +124,28 @@
quarkus.camel.langchain4j.ingest.jdbcdocs.embedding-model=test-model
quarkus.camel.langchain4j.ingest.jdbcdocs.max-segment-size=120
quarkus.camel.langchain4j.ingest.jdbcdocs.max-overlap-size=20
+# the filter options, forwarded to the sink Kamelet and enforced by the
component
+quarkus.camel.langchain4j.ingest.filtered.source.uri=direct:filtered-feed
+quarkus.camel.langchain4j.ingest.filtered.embedding-store=filtered-store
+quarkus.camel.langchain4j.ingest.filtered.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.filtered.max-segment-size=120
+quarkus.camel.langchain4j.ingest.filtered.max-overlap-size=20
+quarkus.camel.langchain4j.ingest.filtered.filter.include-id=**/*.md,*.md
+quarkus.camel.langchain4j.ingest.filtered.filter.exclude-id=**/draft-*
+quarkus.camel.langchain4j.ingest.filtered.filter.min-document-size=25
+quarkus.camel.langchain4j.ingest.filtered.filter.document-filter=confidentialFilter
+
+# filters on a builder (@Ingest) pipeline: the datasheets pipeline is declared
in Java, the
+# exclusion comes from configuration
+quarkus.camel.langchain4j.ingest.datasheets.filter.exclude-id=secret-*
+
+# audio pipeline: each file is embedded whole, as one vector, by the
audio-capable test model
+ingest.audio.directory=target/ingest-audio
+quarkus.camel.langchain4j.ingest.audio.source.directory=${ingest.audio.directory}
+quarkus.camel.langchain4j.ingest.audio.modality=media
+quarkus.camel.langchain4j.ingest.audio.embedding-store=audio-store
+quarkus.camel.langchain4j.ingest.audio.embedding-model=audio-model
+
# s3docs and events are declared in Java (ItBuilderPipelines) through the
Endpoint DSL; they stay
# off until S3TestResource / IngestKafkaTestResource provide a store and a
broker
quarkus.camel.langchain4j.ingest.s3docs.enabled=false
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioIT.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioIT.java
new file mode 100644
index 0000000000..aa88cc51e6
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioIT.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+@QuarkusIntegrationTest
+class Langchain4jIngestAudioIT extends Langchain4jIngestAudioTest {
+}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioTest.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioTest.java
new file mode 100644
index 0000000000..5dfe9f4f63
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestAudioTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@code media} modality with audio: a WAV dropped into the watched
directory is embedded whole, as one
+ * vector, by the audio-capable test model, the file name as the document id.
The fixture is
+ * generated, so no binary is checked in.
+ */
+@QuarkusTest
+class Langchain4jIngestAudioTest {
+
+ @Test
+ void audioFileIsEmbeddedWholeAndFoundByTheSameClip() throws Exception {
+ byte[] tone = wav(200);
+
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(tone)
+ .post("/langchain4j-ingest/binary/audio/tone.wav")
+ .then().statusCode(204);
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500,
TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ List<Map<String, String>> hits = searchByClip(tone);
+ assertEquals(1, hits.size(), "one vector per audio file");
+ assertEquals("audio", hits.get(0).get("pipeline"));
+ assertEquals("tone.wav", hits.get(0).get("documentId"));
+ // the placeholder segment carries the id as its text
+ assertEquals("tone.wav", hits.get(0).get("text"));
+ });
+
+ // a different clip embeds to a different vector, so it is no exact hit
+ assertTrue(searchByClip(wav(120)).isEmpty(), "another clip must not
match the stored one");
+ }
+
+ /** The {@code clips} builder pipeline's content type stands in for the
extension its ids lack. */
+ @Test
+ void contentTypeAppliesToIdsWithoutExtension() {
+ byte[] clip = wav(300);
+
+ String outcome = RestAssured.given().contentType(ContentType.BINARY)
+ .body(clip)
+ .post("/langchain4j-ingest/feed-binary/clips/clip-7")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", outcome);
+
+ List<Map<String, String>> hits = searchByClip(clip);
+ assertEquals(1, hits.size());
+ assertEquals("clips", hits.get(0).get("pipeline"));
+ assertEquals("clip-7", hits.get(0).get("documentId"));
+ }
+
+ /**
+ * A consumer-fed media stream over the cap is refused after max + 1
bytes, whatever CamelFileLength it announces:
+ * the component reads it no further (CAMEL-25463), read whole it would
report all 32044 bytes.
+ */
+ @Test
+ void oversizedClipIsRefusedBeforeItIsReadWhole() {
+ byte[] clip = wav(1000);
+
+ String outcome = RestAssured.given().contentType(ContentType.BINARY)
+ .queryParam("fileLength", 10)
+ .body(clip)
+ .post("/langchain4j-ingest/feed-binary/clips/big-clip")
+ .then().statusCode(200).extract().asString();
+ assertTrue(outcome.contains("exceeds maxDocumentSize (20001 > 20000
bytes)"), outcome);
+ assertTrue(searchByClip(clip).isEmpty(), "the oversized clip must not
be stored");
+ }
+
+ /** Query by audio: the clip is embedded with the same model and the store
searched for exact hits. */
+ private static List<Map<String, String>> searchByClip(byte[] clip) {
+ return RestAssured.given().contentType(ContentType.BINARY)
+ .body(clip)
+ .post("/langchain4j-ingest/search/audio")
+ .then()
+ .statusCode(200)
+ .extract().jsonPath().getList("");
+ }
+
+ /** A mono 16 kHz PCM WAV of the given length holding a 440 Hz tone, the
RIFF header written by hand. */
+ static byte[] wav(int millis) {
+ int rate = 16_000;
+ int frames = rate * millis / 1000;
+ ByteBuffer wav = ByteBuffer.allocate(44 + frames *
2).order(ByteOrder.LITTLE_ENDIAN);
+ wav.put("RIFF".getBytes(StandardCharsets.US_ASCII)).putInt(36 + frames
* 2)
+ .put("WAVE".getBytes(StandardCharsets.US_ASCII))
+ .put("fmt ".getBytes(StandardCharsets.US_ASCII)).putInt(16)
+ .putShort((short) 1).putShort((short)
1).putInt(rate).putInt(rate * 2).putShort((short) 2)
+ .putShort((short) 16)
+ .put("data".getBytes(StandardCharsets.US_ASCII)).putInt(frames
* 2);
+ for (int i = 0; i < frames; i++) {
+ wav.putShort((short) (Math.sin(2 * Math.PI * 440 * i / rate) *
Short.MAX_VALUE / 4));
+ }
+ return wav.array();
+ }
+}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterIT.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterIT.java
new file mode 100644
index 0000000000..5f8078d5b0
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterIT.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+@QuarkusIntegrationTest
+class Langchain4jIngestFilterIT extends Langchain4jIngestFilterTest {
+}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterTest.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterTest.java
new file mode 100644
index 0000000000..16e9145f20
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestFilterTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The filter configuration, forwarded through the sink Kamelet to the
component: id patterns,
+ * the size floor and the documentFilter predicate each answer {@code
filtered}.
+ */
+@QuarkusTest
+class Langchain4jIngestFilterTest {
+
+ @Test
+ void filtersGovernIngestion() {
+ assertEquals("filtered", feed("notes.txt", "A text file the id
patterns must reject."));
+ assertEquals("filtered", feed("draft-plan.md", "Excluded although the
include patterns match."));
+ assertEquals("filtered", feed("stub.md", "too short"));
+ assertEquals("filtered", feed("brief.md", "The CONFIDENTIAL relay
specification stays out."));
+
+ assertEquals("ingested", feed("camels.md", "Camels are resilient
desert animals with two rows of eyelashes."));
+
+ // only the accepted document reached the store
+ assertNotNull(Langchain4jIngestTest.hit("What is resilient?",
"filtered", "Camels"));
+ assertTrue(Langchain4jIngestTest.hits("What is resilient?",
"filtered").stream()
+ .allMatch(hit -> hit.get("documentId").equals("camels.md")),
+ "no filtered delivery may have been written");
+ }
+
+ /** A builder ({@code @Ingest}) pipeline is filtered through configuration
the same way. */
+ @Test
+ void configurationFiltersApplyToBuilderPipelines() {
+ String excluded = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The OMEGA-3 valve datasheet is not for the knowledge
base.")
+ .post("/langchain4j-ingest/feed/datasheets/secret-omega.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("filtered", excluded);
+
+ String accepted = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The OMEGA-3 valve datasheet lists coolant tolerances.")
+ .post("/langchain4j-ingest/feed/datasheets/sheet-omega.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", accepted);
+ }
+
+ /**
+ * With a parser, the id patterns are checked before the size guard and
the parse: on the
+ * {@code htmlfeed} pipeline an excluded document over the cap is answered
filtered, while an
+ * included one still meets the guard.
+ */
+ @Test
+ void excludedDocumentIsFilteredBeforeTheParse() {
+ String oversized = "<html><body><p>" + "x".repeat(5000) +
"</p></body></html>";
+ assertEquals("filtered", feedProperty("htmlfeed", "clips/clip.mp4",
oversized));
+ assertTrue(feedProperty("htmlfeed", "big.html",
oversized).contains("exceeds maxDocumentSize"));
+ }
+
+ /**
+ * The same on a directory pipeline: on {@code capped} an excluded file
over the cap is filtered,
+ * so the file register records it as done instead of the guard failing it
on every poll.
+ */
+ @Test
+ void excludedFileIsFilteredBeforeTheParse() throws Exception {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(new byte[1000])
+ .post("/langchain4j-ingest/binary/capped/clip.mp4")
+ .then().statusCode(204);
+
+ // the file source's register key: absolute path, modification time
and size. The register
+ // adds it when a poll picks the file up and drops it when the
exchange fails, so only a key
+ // that stays across poll cycles shows the file is done with
+ Path file = Path.of("target/ingest-capped/clip.mp4").toAbsolutePath();
+ String key = file + ":" + Files.getLastModifiedTime(file).toMillis() +
":" + Files.size(file);
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).during(3,
TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .until(() -> Boolean.parseBoolean(RestAssured.given()
+ .queryParam("repo",
"langchain4j-ingest-capped-register")
+ .queryParam("key", key)
+ .get("/langchain4j-ingest/register-contains")
+ .then().statusCode(200).extract().asString()));
+ }
+
+ /**
+ * A named register may be persistent, so a filtered file's key stays out
of it: on {@code products} the excluded
+ * file is still skipped across polls, its key held in memory, and a
filter changed for the next start applies to it.
+ */
+ @Test
+ void filteredFileStaysOutOfTheNamedRegister() throws Exception {
+ Langchain4jIngestTest.write("draft-pumps.txt", "The ZETA-9 pump draft
is not final.");
+
+ Path file =
Path.of("target/ingest-docs/draft-pumps.txt").toAbsolutePath();
+ String key = file + ":" + Files.getLastModifiedTime(file).toMillis() +
":" + Files.size(file);
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).during(3,
TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .until(() ->
registerContains("langchain4j-ingest-products-register", key));
+ assertFalse(registerContains("propertiesRegister", key), "the named
register must not hold the key");
+ }
+
+ private static boolean registerContains(String repo, String key) {
+ return Boolean.parseBoolean(RestAssured.given()
+ .queryParam("repo", repo)
+ .queryParam("key", key)
+ .get("/langchain4j-ingest/register-contains")
+ .then().statusCode(200).extract().asString());
+ }
+
+ /** Answers the outcome, or the failure message. */
+ private static String feedProperty(String pipeline, String documentId,
String body) {
+ return RestAssured.given().contentType(ContentType.TEXT)
+ .body(body)
+ .post("/langchain4j-ingest/feed-property/" + pipeline + "/" +
documentId)
+ .then().statusCode(200).extract().asString();
+ }
+
+ private static String feed(String documentId, String body) {
+ return RestAssured.given().contentType(ContentType.TEXT)
+ .body(body)
+ .post("/langchain4j-ingest/feed/filtered/" + documentId)
+ .then().statusCode(200).extract().asString();
+ }
+}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
index 5e06dc33c0..1f01c7c566 100644
---
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
@@ -138,10 +138,18 @@ class Langchain4jIngestTikaTest {
.post("/langchain4j-ingest/binary/capped/pump.pdf")
.then().statusCode(204);
+ // a comment carries the bytes, not the text: parsed, it fits the cap,
so only the raw-byte
+ // cap keeps it out
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(("<html><body><!-- " + "x".repeat(300) + "
--><p>LAMBDA-7 hides in markup.</p></body></html>")
+ .getBytes(java.nio.charset.StandardCharsets.UTF_8))
+ .post("/langchain4j-ingest/binary/capped/markup.html")
+ .then().statusCode(204);
+
// the absence must hold across several poll cycles, not just at one
instant
Awaitility.await().during(3, TimeUnit.SECONDS).atMost(10,
TimeUnit.SECONDS)
.until(() -> Langchain4jIngestTest.hits("What does the pump
tolerate?", "capped").stream()
- .noneMatch(hit ->
hit.get("text").contains("DELTA-5")));
+ .noneMatch(hit -> hit.get("text").contains("DELTA-5")
|| hit.get("text").contains("LAMBDA-7")));
}
/**
