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 1a8c186786 Fixes #9152. Compose the langchain4j-ingest pipelines from
the catalog Kamelets
1a8c186786 is described below
commit 1a8c186786240bdef4b54d2f977fcacc4204a4e9
Author: jiriOndrusek-agent <[email protected]>
AuthorDate: Mon Oct 26 12:44:27 2026 +0100
Fixes #9152. Compose the langchain4j-ingest pipelines from the catalog
Kamelets
Replaces the extension's own route builder with a composition of the
catalog Kamelets `langchain4j-ingest-file-source`, `tika-extract-text-action`
or `docling-convert-action`, and `langchain4j-ingest-sink`, always loaded from
the classpath. Each Kamelet runs as its own route, and the extension registers
the four Kamelets for native itself. Ingestion results stay the same apart from
the edge cases listed in the 4.0 migration guide, which also covers the three
new dependencies and the T [...]
Requires camel-kamelets 4.23.0.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---
docs/modules/ROOT/pages/migration-guide/4.0.0.adoc | 10 +-
.../reference/extensions/langchain4j-ingest.adoc | 5 +
extensions/langchain4j-ingest/deployment/pom.xml | 8 +
.../deployment/Langchain4jIngestProcessor.java | 24 +-
.../IngestBuilderAmbiguousStoreTest.java | 71 ++++
.../deployment/IngestBuilderInvalidNameTest.java | 55 +++
.../deployment/IngestConsumerDocumentIdTest.java | 133 ++++++
.../deployment/IngestKameletLocationTest.java | 48 +++
.../ingest/deployment/IngestRouteIdClashTest.java | 58 +++
.../ingest/deployment/IngestUnnamedBeansTest.java | 8 +
extensions/langchain4j-ingest/runtime/pom.xml | 12 +
.../runtime/src/main/doc/usage.adoc | 5 +
.../langchain4j/ingest/IngestPipeline.java | 2 +-
.../ingest/IngestPipelineDefinition.java | 351 ----------------
.../ingest/IngestPipelineRouteBuilder.java | 458 --------------------
.../component/langchain4j/ingest/IngestRoutes.java | 460 ++++++++++++++++++---
.../component/langchain4j/ingest/Parser.java | 39 ++
.../langchain4j/ingest/it/IngestItProducers.java | 7 +
.../langchain4j/ingest/it/IngestResource.java | 33 +-
.../src/main/resources/application.properties | 13 +
.../ingest/it/Langchain4jIngestTikaTest.java | 37 ++
pom.xml | 2 +-
poms/bom/src/main/generated/flattened-full-pom.xml | 2 +-
.../src/main/generated/flattened-reduced-pom.xml | 2 +-
.../generated/flattened-reduced-verbose-pom.xml | 2 +-
25 files changed, 969 insertions(+), 876 deletions(-)
diff --git a/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
b/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
index 93910fb7d7..746c4ea2f5 100644
--- a/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
+++ b/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
@@ -89,9 +89,17 @@ The declaration surface is unchanged — the same
`quarkus.camel.langchain4j.ing
Java code reading the request-reply outcome as `IngestResult` now imports
`org.apache.camel.component.langchain4j.ingest.IngestResult` — the extension's
`org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult` is
gone with the engine, so this surfaces at compile time.
The same applies to the segment metadata keys
`IngestService.METADATA_DOCUMENT_ID` and `IngestService.METADATA_PIPELINE`: use
`org.apache.camel.component.langchain4j.ingest.LangChain4jIngest.METADATA_DOCUMENT_ID`
and `METADATA_PIPELINE` instead; the values are unchanged.
-A pipeline's route id is now `langchain4j-ingest-<pipeline-name>` rather than
`camel-quarkus-langchain4j-ingest-<pipeline-name>`.
+A pipeline's own route id is now `langchain4j-ingest-<pipeline-name>` rather
than `camel-quarkus-langchain4j-ingest-<pipeline-name>`.
+The pipeline is composed from Kamelets now, and each one it uses adds a route:
`langchain4j-ingest-<pipeline-name>-source`, `-parser` or `-sink`.
Tooling that addresses the routes by id — JMX, the management endpoints,
`camel.main.*` route policies keyed on ids — must use the new names.
The prefix is the component's scheme name, so it follows the component.
+To pause a directory pipeline, stop its
`langchain4j-ingest-<pipeline-name>-source` route, for example with
`camelContext.getRouteController().stopRoute(...)`: with
`langchain4j-ingest-<pipeline-name>` alone stopped, the directory is still
polled and each new file fails after the Kamelet's 30 s timeout.
+Camel registers the Kamelet routes in JMX only with
`camel.main.jmxManagementRegisterRoutesCreateByKamelet=true`.
+
+The extension now depends on `camel-kamelets`, `camel-quarkus-kamelet` and
`camel-quarkus-yaml-dsl`.
+
+With `parser=tika`, the reply to a request-reply caller of a consumer-fed
pipeline keeps only the `Camel*` headers; the previous route kept the caller's
headers.
+The extracted text also keeps Tika's surrounding whitespace, which now counts
toward both `max-document-size` and `max-segment-size`: a document near either
limit can fail or split differently, though the default splitter still trims
each segment.
A consumer-fed delivery whose document id resolves to nothing still fails the
exchange, but the message now comes from the component and names its
`documentIdHeader` endpoint option instead of the
`quarkus.camel.langchain4j.ingest.<name>.source.document-id` property.
Code matching on the message text must be adjusted; matching on the exception
type is unaffected.
diff --git
a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index 6a9adaa8e2..201059975b 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -64,6 +64,11 @@ Each file is read as UTF-8 text — unless a `parser` turns
binary formats into
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.
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.
+Application-wide `camel.kamelet.<kamelet>.*` properties and `CAMEL_KAMELET_*`
environment variables for these four Kamelets reach the pipelines too, for
every option the extension leaves unset, so do not set them.
+To pause a directory pipeline, stop its `langchain4j-ingest-<name>-source`
route, or both routes, for example with
`camelContext.getRouteController().stopRoute(...)`: with
`langchain4j-ingest-<name>` alone stopped, the directory is still polled and
each new file fails after the Kamelet's 30 s timeout.
+Camel registers the Kamelet routes in JMX only with
`camel.main.jmxManagementRegisterRoutesCreateByKamelet=true`.
[NOTE]
====
diff --git a/extensions/langchain4j-ingest/deployment/pom.xml
b/extensions/langchain4j-ingest/deployment/pom.xml
index be40e6accd..8957af335c 100644
--- a/extensions/langchain4j-ingest/deployment/pom.xml
+++ b/extensions/langchain4j-ingest/deployment/pom.xml
@@ -38,6 +38,14 @@
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-file-deployment</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kamelet-deployment</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-yaml-dsl-deployment</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-langchain4j-ingest</artifactId>
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 25f6554f7c..7455e47c44 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
@@ -36,6 +36,7 @@ import io.quarkus.deployment.annotations.Record;
import io.quarkus.deployment.builditem.ApplicationArchivesBuildItem;
import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
import io.quarkus.deployment.builditem.FeatureBuildItem;
+import
io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
import io.quarkus.runtime.configuration.ConfigurationException;
import jakarta.inject.Singleton;
@@ -73,6 +74,17 @@ class Langchain4jIngestProcessor {
.build();
}
+ /**
+ * The composition resolves its catalog Kamelets at runtime, so a native
image needs their
+ * YAML whatever {@code quarkus.camel.kamelet.identifiers} narrows the
kamelet extension to.
+ */
+ @BuildStep
+ NativeImageResourceBuildItem ingestKamelets() {
+ return new NativeImageResourceBuildItem(IngestRoutes.KAMELETS.stream()
+ .map(id -> "kamelets/" + id + ".kamelet.yaml")
+ .toList());
+ }
+
/**
* {@code #class:} beans are instantiated reflectively, which native mode
allows only for
* registered classes: the two camel-core repositories the documentation
recommends, plus
@@ -97,8 +109,8 @@ class Langchain4jIngestProcessor {
}
/**
- * Discovers {@code @Ingest} builder methods: validated here (return type,
no parameters,
- * unique names, no collision with configuration-declared pipelines),
invoked reflectively once
+ * Discovers {@code @Ingest} builder methods: validated here (name
charset, return type, no
+ * parameters, unique names, no collision with configuration-declared
pipelines), invoked reflectively once
* at startup. Violations are reported as {@link
ValidationErrorBuildItem}s — the channel every
* build-time check of this extension uses, so dev and test mode see them
too, and all of them
* at once.
@@ -134,6 +146,14 @@ class Langchain4jIngestProcessor {
"@Ingest on " + location + " has a blank pipeline
name")));
continue;
}
+ // the charset IngestRoutes holds configured names to at startup:
the name ends up in
+ // Kamelet URIs and registry references, and an @Ingest name is
known already here
+ if (!name.matches(IngestRoutes.NAME_PATTERN)) {
+ validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
+ "@Ingest on " + location + " has pipeline name '" +
name
+ + "', which may only contain letters, digits,
'.', '_' and '-'")));
+ continue;
+ }
if (!method.returnType().name().equals(pipelineType)) {
validationErrors.produce(new ValidationErrorBuildItem(new
ConfigurationException(
"@Ingest method " + location + " must return " +
IngestPipeline.class.getSimpleName())));
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderAmbiguousStoreTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderAmbiguousStoreTest.java
new file mode 100644
index 0000000000..34b5d0d36a
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderAmbiguousStoreTest.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 dev.langchain4j.data.segment.TextSegment;
+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.Named;
+import jakarta.inject.Singleton;
+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;
+
+/**
+ * With two store beans, a Java-declared pipeline must name one. The message
points at the
+ * builder method: the configuration key would be rejected at build time for
this name.
+ */
+class IngestBuilderAmbiguousStoreTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class, OtherStore.class, Pipelines.class))
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "Ingestion pipeline 'docs' found 2 embedding store beans",
+ "Name the one to use with
IngestPipeline.embeddingStore(...) in its @Ingest method"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class OtherStore {
+
+ @Produces
+ @Singleton
+ @Named("other-store")
+ EmbeddingStore<TextSegment> store() {
+ return new InMemoryEmbeddingStore<>();
+ }
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ @Ingest("docs")
+ IngestPipeline docs() {
+ return
IngestPipeline.from(Source.file("target/ambiguous-store")).embeddingModel("model");
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderInvalidNameTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderInvalidNameTest.java
new file mode 100644
index 0000000000..1ba490f73b
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderInvalidNameTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+/**
+ * A pipeline name is substituted into Kamelet URIs and registry references,
so a Java-declared
+ * one is held to the same charset as a configured one - checked at build
time, where it is known.
+ */
+class IngestBuilderInvalidNameTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(Pipelines.class))
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "has pipeline name 'docs/en', which may only contain
letters, digits"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ @Ingest("docs/en")
+ IngestPipeline docs() {
+ return IngestPipeline.from(Source.file("target/invalid-name"))
+ .embeddingStore("store")
+ .embeddingModel("model");
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestConsumerDocumentIdTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestConsumerDocumentIdTest.java
new file mode 100644
index 0000000000..6b2c4e4fa9
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestConsumerDocumentIdTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.Inject;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Predicate;
+import org.apache.camel.Processor;
+import org.apache.camel.component.langchain4j.ingest.IngestResult;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestHeaders;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * The document id on consumer-fed pipelines without a parser, as with the
previous route builder:
+ * the header wins and the exchange property is only the fallback, and a
missing id is reported
+ * under the configured header. Application-wide sink Kamelet properties do
not reach them either.
+ */
+class IngestConsumerDocumentIdTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class, ValidationTestSupport.class,
+ RejectAllFilter.class))
+ // any consumer the test classpath has: the tests hand their
exchanges to the routes directly
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.uri",
"file:target/consumer-document-id-docs")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.document-id",
"X-Doc-Id")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.plain.source.uri",
"file:target/consumer-document-id-plain")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.plain.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.plain.embedding-model",
"model")
+ // meant for an application's own use of the sink Kamelet: the
pipelines pin these options, else
+ // they would fail the start (modality, contentType) or filter
every document
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.documentIdHeader",
"Other-Id")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.minDocumentSize",
"1000")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.modality", "media")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.contentType",
"audio/wav")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.includeId",
"no-such-id")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.excludeId", "**")
+
.overrideConfigKey("camel.kamelet.langchain4j-ingest-sink.documentFilter",
"#bean:rejectAll");
+
+ @Inject
+ CamelContext context;
+
+ @Test
+ void missingIdNamesTheConfiguredHeader() throws Exception {
+ Exchange exchange = deliver("docs", e -> {
+ });
+ ValidationTestSupport.assertFailure(exchange.getException(), "no
document id. Set the X-Doc-Id header");
+ }
+
+ @Test
+ void idInThePropertyOfTheConfiguredHeaderIsIngested() throws Exception {
+ assertIngested(deliver("docs", e -> e.setProperty("X-Doc-Id",
"manual.txt")), "manual.txt");
+ }
+
+ @Test
+ void headerWinsOverStaleExchangeProperty() throws Exception {
+ assertIngested(deliver("plain", e -> {
+ e.getMessage().setHeader(IngestHeaders.DOCUMENT_ID, "fresh.txt");
+ e.setProperty(IngestHeaders.DOCUMENT_ID, "stale.txt");
+ }), "fresh.txt");
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void idOnlyInTheExchangePropertyIsIngested() throws Exception {
+ assertIngested(deliver("plain", e ->
e.setProperty(IngestHeaders.DOCUMENT_ID, "property.txt")), "property.txt");
+ assertIngested(deliver("plain", e ->
e.setProperty(IngestHeaders.LEGACY_DOCUMENT_ID, "legacy.txt")), "legacy.txt");
+ }
+
+ @Test
+ void appWideSinkPropertiesDoNotReachThePipeline() throws Exception {
+ // minDocumentSize, includeId, excludeId and documentFilter would each
filter this short document
+ assertIngested(deliver("plain", e ->
e.getMessage().setHeader(IngestHeaders.DOCUMENT_ID, "short.txt")),
+ "short.txt");
+ // documentIdHeader=Other-Id would be named here
+ ValidationTestSupport.assertFailure(deliver("plain", e -> {
+ }).getException(), "no document id. Set the
CamelLangChain4jIngestDocumentId header");
+ }
+
+ private static void assertIngested(Exchange exchange, String documentId) {
+ Assertions.assertNull(exchange.getException());
+ IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
+ Assertions.assertEquals(IngestResult.Outcome.INGESTED,
result.outcome());
+ Assertions.assertEquals(documentId, result.documentId());
+ }
+
+ /** Runs an exchange through the pipeline's route as if its consumer had
delivered it. */
+ private Exchange deliver(String pipeline, Processor customizer) throws
Exception {
+ Consumer consumer = context.getRoute("langchain4j-ingest-" +
pipeline).getConsumer();
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getMessage().setBody("The pump tolerates five bar.");
+ customizer.process(exchange);
+ consumer.getProcessor().process(exchange);
+ return exchange;
+ }
+
+ /** Named only by the application-wide documentFilter property the
pipelines must not pick up. */
+ @ApplicationScoped
+ public static class RejectAllFilter {
+
+ @Produces
+ @Singleton
+ @Named("rejectAll")
+ Predicate rejectAll() {
+ return exchange -> false;
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestKameletLocationTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestKameletLocationTest.java
new file mode 100644
index 0000000000..bd6a289ef7
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestKameletLocationTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.inject.Inject;
+import org.apache.camel.CamelContext;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * The pipelines load their Kamelets from the classpath whatever {@code
camel.component.kamelet.location}
+ * says, so an application keeping its own Kamelets elsewhere does not break
them.
+ */
+class IngestKameletLocationTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class))
+ .overrideConfigKey("camel.component.kamelet.location",
"classpath:my-kamelets")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.directory",
"target/kamelet-location-docs")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model");
+
+ @Inject
+ CamelContext context;
+
+ @Test
+ void pipelineStartsWithAnotherKameletLocation() {
+
Assertions.assertNotNull(context.getRoute("langchain4j-ingest-docs-source"));
+
Assertions.assertNotNull(context.getRoute("langchain4j-ingest-docs-sink"));
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestRouteIdClashTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestRouteIdClashTest.java
new file mode 100644
index 0000000000..4d082c6515
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestRouteIdClashTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+/**
+ * A pipeline named after a step of another one would share a route id with
it: the start fails
+ * naming both, whichever way each is declared.
+ */
+class IngestRouteIdClashTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class, Pipelines.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs-sink.source.directory",
"target/route-id-clash-sink")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs-sink.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs-sink.embedding-model",
"model")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "Ingestion pipelines 'docs' and 'docs-sink' would both use
the route id 'langchain4j-ingest-docs-sink'"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ @Ingest("docs")
+ IngestPipeline docs() {
+ return IngestPipeline.from(Source.file("target/route-id-clash"))
+ .embeddingStore("store")
+ .embeddingModel("model");
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
index 0bc6f94b10..f0957bbc24 100644
---
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
@@ -59,6 +59,14 @@ class IngestUnnamedBeansTest {
"the pipeline must start with the only store and model beans,
unnamed");
}
+ /** Each Kamelet a pipeline is composed from adds a route named after the
pipeline. */
+ @Test
+ void kameletRoutesAreNamedAfterThePipeline() {
+
Assertions.assertNotNull(context.getRoute("langchain4j-ingest-docs-source"));
+
Assertions.assertNotNull(context.getRoute("langchain4j-ingest-docs-sink"));
+ Assertions.assertEquals(3, context.getRoutes().size(), "the pipeline's
route and one per Kamelet");
+ }
+
@ApplicationScoped
public static class UnnamedEmbeddingBeans {
diff --git a/extensions/langchain4j-ingest/runtime/pom.xml
b/extensions/langchain4j-ingest/runtime/pom.xml
index bf77d71837..dbec116384 100644
--- a/extensions/langchain4j-ingest/runtime/pom.xml
+++ b/extensions/langchain4j-ingest/runtime/pom.xml
@@ -45,6 +45,18 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-ingest</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.kamelets</groupId>
+ <artifactId>camel-kamelets</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kamelet</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-yaml-dsl</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-file</artifactId>
diff --git a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
index 5cdcdd6b33..532a4eb694 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
+++ b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
@@ -15,6 +15,11 @@ Each file is read as UTF-8 text — unless a `parser` turns
binary formats into
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.
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.
+Application-wide `camel.kamelet.<kamelet>.*` properties and `CAMEL_KAMELET_*`
environment variables for these four Kamelets reach the pipelines too, for
every option the extension leaves unset, so do not set them.
+To pause a directory pipeline, stop its `langchain4j-ingest-<name>-source`
route, or both routes, for example with
`camelContext.getRouteController().stopRoute(...)`: with
`langchain4j-ingest-<name>` alone stopped, the directory is still polled and
each new file fails after the Kamelet's 30 s timeout.
+Camel registers the Kamelet routes in JMX only with
`camel.main.jmxManagementRegisterRoutesCreateByKamelet=true`.
[NOTE]
====
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 d5c9ad8c9e..42492f393e 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
@@ -26,7 +26,7 @@ import java.util.Set;
public final class IngestPipeline {
/** The values {@link #parser(String)} and the {@code parser}
configuration property accept. */
- public static final Set<String> SUPPORTED_PARSERS =
IngestPipelineDefinition.SUPPORTED_PARSERS;
+ public static final Set<String> SUPPORTED_PARSERS = Parser.NAMES;
private final Source source;
private String embeddingStoreName;
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
deleted file mode 100644
index e94a65684e..0000000000
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
+++ /dev/null
@@ -1,351 +0,0 @@
-/*
- * 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;
-
-import java.util.Arrays;
-import java.util.Locale;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import dev.langchain4j.data.segment.TextSegment;
-import dev.langchain4j.model.embedding.EmbeddingModel;
-import dev.langchain4j.store.embedding.EmbeddingStore;
-import org.apache.camel.builder.EndpointConsumerBuilder;
-
-/**
- * A declarative description of one ingestion pipeline, turned into a route by
{@link IngestPipelineRouteBuilder}: what
- * to consume, how to identify each document, whether to parse it, and where
to embed and store it.
- *
- * <p>
- * A pipeline reads either a directory ({@link #directory(String, String)}) or
any Camel consumer
- * ({@link #consumer(String, String)}); every other property has a matching
endpoint option on the
- * {@code langchain4j-ingest} producer the generated route ends in.
- *
- * <p>
- * An internal copy of the pipeline assembly proposed alongside the upstream
component — see
- * {@link IngestPipelineRouteBuilder}.
- */
-final class IngestPipelineDefinition {
-
- /** The supported document parsers; the configuration value is the
lower-case name. */
- public enum Parser {
- TIKA,
- DOCLING;
-
- static Parser of(String value) {
- try {
- return valueOf(value.toUpperCase(Locale.ROOT));
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- "parser must be one of " + SUPPORTED_PARSERS + " (got
'" + value + "')");
- }
- }
- }
-
- // derived from the enum, so a parser added there cannot diverge from the
validated set
- public static final Set<String> SUPPORTED_PARSERS =
Arrays.stream(Parser.values())
- .map(parser -> parser.name().toLowerCase(Locale.ROOT))
- .collect(Collectors.toUnmodifiableSet());
-
- private final String name;
- private final String directory;
- private final String uri;
-
- private boolean recursive = true;
- private String documentId;
- private Parser parser;
- private int maxSegmentSize =
IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE;
- private int maxOverlapSize =
IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE;
- private int embeddingBatchSize =
IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE;
- private int maxDocumentSize =
IngestBuildTimeConfig.DEFAULT_MAX_DOCUMENT_SIZE;
-
- private String embeddingStoreRef;
- private EmbeddingStore<TextSegment> embeddingStore;
- private String embeddingModelRef;
- private EmbeddingModel embeddingModel;
- private String documentSplitterRef;
-
- private String idempotentRepositoryRef;
- private boolean idempotentRepositoryAutoCreate;
-
- private IngestPipelineDefinition(String name, String directory, String
uri) {
- this.name = name;
- this.directory = directory;
- this.uri = uri;
- }
-
- /**
- * A pipeline reading the files of a directory: they are left in place (a
knowledge base reads its source, it does
- * not consume it), unchanged files are remembered and skipped, and a file
still being copied in is waited for
- * rather than half-ingested.
- */
- public static IngestPipelineDefinition directory(String name, String
directory) {
- requireText(directory, "directory");
- if (directory.contains("?") || directory.contains("#")) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + name + "': the directory must not
contain '?' or '#' (got '"
- + directory + "')");
- }
- return new IngestPipelineDefinition(requireName(name), directory,
null);
- }
-
- /**
- * A pipeline consuming from any Camel component, with the component's own
options — the escape hatch beside
- * {@link #directory(String, String)}. Which part of the exchange
identifies the document is the consumer's
- * business, so say it with {@link #documentId}: {@code CamelAwsS3Key} for
an S3 consumer, {@code CamelKafkaKey} for
- * a Kafka one.
- */
- public static IngestPipelineDefinition consumer(String name, String uri) {
- requireText(uri, "uri");
- if (!uri.contains(":")) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + name + "': '" + uri + "' is not a
consumer URI");
- }
- return new IngestPipelineDefinition(requireName(name), null, uri);
- }
-
- /**
- * The same, built with the Camel Endpoint DSL (add the {@code
camel-endpointdsl} dependency and use its static
- * builders or an {@code EndpointBuilderFactory}).
- */
- public static IngestPipelineDefinition consumer(String name,
EndpointConsumerBuilder uri) {
- return consumer(name, uri.getRawUri());
- }
-
- /** Whether subdirectories are ingested too; directory pipelines only. */
- public IngestPipelineDefinition recursive(boolean recursive) {
- this.recursive = recursive;
- return this;
- }
-
- /**
- * Where the document id lives in the exchange the consumer delivers: the
name of a header, or a simple-language
- * expression (a value containing {@code ${} or {@code $simple{} is parsed
as one). When not set, a directory
- * pipeline uses the file name and a consumer pipeline the {@code
CamelLangChain4jIngestDocumentId} header. The id
- * is captured before any parse, so a document cannot forge its own
identity.
- */
- public IngestPipelineDefinition documentId(String documentId) {
- this.documentId = requireText(documentId, "documentId");
- return this;
- }
-
- /**
- * Parses the consumed payload into text before splitting: {@code tika}
extracts plain text in-process,
- * {@code docling} converts to markdown through a Docling Serve instance.
The corresponding component must be on the
- * classpath: {@code camel-tika} or {@code camel-docling}.
- */
- public IngestPipelineDefinition parser(String parser) {
- this.parser = Parser.of(requireText(parser, "parser"));
- return this;
- }
-
- /**
- * The typed twin of {@link #parser(String)}, for Java callers; the String
form exists for configuration
- * translators, whose input is text by nature.
- */
- public IngestPipelineDefinition parser(Parser parser) {
- if (parser == null) {
- throw new IllegalArgumentException("Ingestion pipeline parser must
not be null");
- }
- this.parser = parser;
- return this;
- }
-
- /** The splitter bounds; the same rule the endpoint enforces, failing
early here. */
- public IngestPipelineDefinition splitter(int maxSegmentSize, int
maxOverlapSize) {
- if (maxSegmentSize <= 0 || maxOverlapSize < 0 || maxOverlapSize >=
maxSegmentSize) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + name + "': maxSegmentSize must be
positive and maxOverlapSize must be"
- + " non-negative and smaller than it (got " +
maxSegmentSize + " / "
- + maxOverlapSize + ")");
- }
- this.maxSegmentSize = maxSegmentSize;
- this.maxOverlapSize = maxOverlapSize;
- return this;
- }
-
- /**
- * How many segments are embedded per request to the embedding model; the
same rule the endpoint enforces, failing
- * early here. A batch carries at most {@code embeddingBatchSize x
maxSegmentSize} characters, so tune the two
- * together against the provider's token limits.
- */
- public IngestPipelineDefinition embeddingBatchSize(int embeddingBatchSize)
{
- if (embeddingBatchSize < 1) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + name + "': embeddingBatchSize
must be positive (got "
- + embeddingBatchSize + ")");
- }
- this.embeddingBatchSize = embeddingBatchSize;
- return this;
- }
-
- /**
- * Maximum size of one document; unset 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. It is enforced twice: a
- * pipeline with a {@link #parser(Parser)} rejects a larger raw payload
(in bytes) before the parse ever runs, and
- * the endpoint rejects longer extracted text (in characters) before
splitting — a small file can still parse into a
- * lot of text. An oversized document fails the exchange cleanly: a
directory pipeline quarantines it like any other
- * failure, a consumer pipeline propagates the error to its source, and a
dedup claim is released either way.
- */
- public IngestPipelineDefinition maxDocumentSize(int maxDocumentSize) {
- if (maxDocumentSize < 0) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + name + "': maxDocumentSize must
not be negative, 0 meaning no limit (got "
- + maxDocumentSize + ")");
- }
- this.maxDocumentSize = maxDocumentSize;
- return this;
- }
-
- /**
- * Name of the {@code DocumentSplitter} bean replacing the default
recursive splitting; {@code maxSegmentSize} and
- * {@code maxOverlapSize} are then ignored. Referenced by name on purpose
— an application may hold unrelated
- * splitters, so none is picked up by type. Segments returned without the
identity metadata are re-stamped, so a
- * custom splitter cannot break citation.
- */
- public IngestPipelineDefinition documentSplitter(String beanName) {
- this.documentSplitterRef = requireText(beanName, "documentSplitter");
- return this;
- }
-
- /** Name of the {@code EmbeddingStore} bean to write to; omitted, the
single bean of the type is used. */
- public IngestPipelineDefinition embeddingStore(String beanName) {
- this.embeddingStoreRef = requireText(beanName, "embeddingStore");
- return this;
- }
-
- /** The {@code EmbeddingStore} to write to, as an instance; it is bound to
the registry under a generated name. */
- public IngestPipelineDefinition embeddingStore(EmbeddingStore<TextSegment>
store) {
- this.embeddingStore = store;
- return this;
- }
-
- /** Name of the {@code EmbeddingModel} bean to embed with; omitted, the
single bean of the type is used. */
- public IngestPipelineDefinition embeddingModel(String beanName) {
- this.embeddingModelRef = requireText(beanName, "embeddingModel");
- return this;
- }
-
- /** The {@code EmbeddingModel} to embed with, as an instance; it is bound
to the registry under a generated name. */
- public IngestPipelineDefinition embeddingModel(EmbeddingModel model) {
- this.embeddingModel = model;
- return this;
- }
-
- /**
- * Name of the {@code IdempotentRepository} bean remembering ingested
documents. A directory pipeline uses it in its
- * file consumer, keyed on path, modification time and size, so a
persistent repository makes restarts free. A
- * consumer pipeline deduplicates deliveries by document id: first write
wins.
- */
- public IngestPipelineDefinition idempotentRepository(String beanName) {
- this.idempotentRepositoryRef = requireText(beanName,
"idempotentRepository");
- return this;
- }
-
- /**
- * When {@code true}, an in-memory register (100 000 keys, lost on
restart) is created and bound under the
- * {@link #idempotentRepository(String)} name, unless a bean with that
name exists — the existing bean wins.
- */
- public IngestPipelineDefinition idempotentRepositoryAutoCreate(boolean
autoCreate) {
- this.idempotentRepositoryAutoCreate = autoCreate;
- return this;
- }
-
- private static String requireName(String name) {
- requireText(name, "name");
- if (!name.matches("[A-Za-z0-9._-]+")) {
- throw new IllegalArgumentException(
- "Ingestion pipeline name '" + name + "' may only contain
letters, digits, '.', '_' and '-'");
- }
- return name;
- }
-
- private static String requireText(String value, String what) {
- if (value == null || value.isBlank()) {
- throw new IllegalArgumentException("Ingestion pipeline " + what +
" must not be null or blank");
- }
- return value;
- }
-
- String name() {
- return name;
- }
-
- String directory() {
- return directory;
- }
-
- String uri() {
- return uri;
- }
-
- boolean recursive() {
- return recursive;
- }
-
- String documentId() {
- return documentId;
- }
-
- Parser parser() {
- return parser;
- }
-
- int maxSegmentSize() {
- return maxSegmentSize;
- }
-
- int maxOverlapSize() {
- return maxOverlapSize;
- }
-
- int embeddingBatchSize() {
- return embeddingBatchSize;
- }
-
- int maxDocumentSize() {
- return maxDocumentSize;
- }
-
- String documentSplitterRef() {
- return documentSplitterRef;
- }
-
- String embeddingStoreRef() {
- return embeddingStoreRef;
- }
-
- EmbeddingStore<TextSegment> embeddingStore() {
- return embeddingStore;
- }
-
- String embeddingModelRef() {
- return embeddingModelRef;
- }
-
- EmbeddingModel embeddingModel() {
- return embeddingModel;
- }
-
- String idempotentRepositoryRef() {
- return idempotentRepositoryRef;
- }
-
- boolean idempotentRepositoryAutoCreate() {
- return idempotentRepositoryAutoCreate;
- }
-}
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
deleted file mode 100644
index d945eeee5e..0000000000
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
+++ /dev/null
@@ -1,458 +0,0 @@
-/*
- * 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;
-
-import java.io.InputStream;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.camel.Component;
-import org.apache.camel.Exchange;
-import org.apache.camel.Expression;
-import org.apache.camel.Processor;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.langchain4j.ingest.IngestResult;
-import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
-import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders;
-import org.apache.camel.component.langchain4j.ingest.TikaTextDecode;
-import org.apache.camel.model.ProcessorDefinition;
-import org.apache.camel.spi.IdempotentRepository;
-import org.apache.camel.support.ExpressionAdapter;
-import org.apache.camel.support.builder.ExpressionBuilder;
-import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
-import org.apache.camel.util.StringHelper;
-import org.apache.camel.util.URISupport;
-
-/**
- * Generates one Camel route per {@link IngestPipelineDefinition}: consume,
resolve the document id, optionally parse,
- * then split, embed and store through the {@code langchain4j-ingest} producer.
- *
- * <p>
- * A directory pipeline watches its folder with the safe file-consumer
defaults — documents are left in place, unchanged
- * files are remembered in a duplicate register keyed on path, modification
time and size, and a file still being copied
- * in is waited for. A consumer pipeline reads any component and deduplicates
by document id when a repository is
- * configured. The id is always captured into an exchange property
<em>before</em> the parse stage: a parser copies
- * document metadata over the headers, so a crafted document could otherwise
forge its own identity.
- *
- * <p>
- * An internal copy of the pipeline assembly proposed alongside the upstream
component: the delegation to Apache Camel
- * is engine-only, so the topology lives here, package-private — replaceable
by an upstream artifact or kamelets if the
- * community adopts one of them.
- *
- * <p>
- * The class is abstract on purpose: camel-quarkus routes discovery
instantiates every concrete public
- * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped
by construction.
- */
-abstract class IngestPipelineRouteBuilder extends RouteBuilder {
-
- /**
- * The built-in register capacity, sized above Camel's 1000-entry default
so eviction does not re-ingest large
- * directories during normal operation; in-memory, so lost on restart.
- */
- static final int DEFAULT_REGISTER_CAPACITY = 100_000;
-
- /**
- * The pipelines to build routes for; supplied by the subclass assembling
its definitions from another source — a
- * configuration model, say.
- */
- protected abstract List<IngestPipelineDefinition> pipelines();
-
- @Override
- public void configure() {
- Set<String> names = new HashSet<>();
- for (IngestPipelineDefinition pipeline : pipelines()) {
- if (!names.add(pipeline.name())) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + pipeline.name() + "' is
defined twice. Use one name per pipeline.");
- }
- configurePipeline(pipeline);
- }
- }
-
- private void configurePipeline(IngestPipelineDefinition pipeline) {
- requireParserComponent(pipeline);
- String storeRef = bindInstance(pipeline.embeddingStore(),
pipeline.embeddingStoreRef(), pipeline, "store");
- String modelRef = bindInstance(pipeline.embeddingModel(),
pipeline.embeddingModelRef(), pipeline, "model");
- String splitterRef = pipeline.documentSplitterRef();
-
- if (pipeline.directory() != null) {
- directoryRoute(pipeline, storeRef, modelRef, splitterRef);
- log.info("Ingestion pipeline '{}': source=file:{}",
pipeline.name(), pipeline.directory());
- } else {
- consumerRoute(pipeline, storeRef, modelRef, splitterRef);
- log.info("Ingestion pipeline '{}': source={}", pipeline.name(),
URISupport.sanitizeUri(pipeline.uri()));
- }
- }
-
- //
********************************************************************************
- // The two route topologies
- //
********************************************************************************
-
- /**
- * The route: watch the directory → resolve the id → (parse) → split,
embed, store. The register in the file
- * endpoint keeps unchanged files from re-ingesting; an edited file gets a
new key and re-ingests, its old segments
- * remain.
- */
- private void directoryRoute(IngestPipelineDefinition pipeline, String
storeRef, String modelRef, String splitterRef) {
- String registerRef = repositoryRef(pipeline, true);
- Expression documentId = documentIdExpression(pipeline.name(),
pipeline.documentId(), Exchange.FILE_NAME);
-
- ProcessorDefinition<?> route = from(fileEndpointUri(pipeline,
registerRef))
- .routeId(routeId(pipeline))
- .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
documentId);
- // no duplicate pre-check: the endpoint register already filtered
duplicates out
- route = parseSteps(route, pipeline, null);
- // the file consumer discards the reply, so no repository is passed to
the producer: the
- // endpoint register already keeps the same file version from being
ingested twice
- ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline,
storeRef, modelRef, splitterRef, null, null));
- if (pipeline.parser() != null) {
- // the discarded reply would otherwise hide it: a parse to nothing
typically means a
- // missing Tika parser module or an image-only document, and the
file's register key
- // is committed, so it is not retried until the file changes
- tail.process(exchange -> {
- IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
- if (result != null && result.outcome() ==
IngestResult.Outcome.EMPTY) {
- log.warn("Ingestion pipeline '{}': document '{}' parsed to
no text and was skipped; its key is"
- + " committed, so it is not retried until the file
changes (missing parser module?"
- + " image-only document?)",
- pipeline.name(), result.documentId());
- }
- });
- } else {
- // the discarded reply would otherwise hide even the trace of a
blank file
- tail.process(exchange -> {
- IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
- if (result != null && result.outcome() ==
IngestResult.Outcome.EMPTY) {
- log.debug("Ingestion pipeline '{}': document '{}'
contained no text, nothing was written",
- pipeline.name(), result.documentId());
- }
- });
- }
- }
-
- /** The route: consume → resolve the id → (parse) → split, embed, store;
the reply is the result. */
- private void consumerRoute(IngestPipelineDefinition pipeline, String
storeRef, String modelRef, String splitterRef) {
- String scheme = StringHelper.before(pipeline.uri(), ":");
- requireComponent(scheme, pipeline, "its source");
- String registerRef = repositoryRef(pipeline, false);
- Expression documentId = documentIdExpression(pipeline.name(),
pipeline.documentId(),
- LangChain4jIngestHeaders.DOCUMENT_ID);
- IdempotentRepository register = registerRef == null
- ? null
- : getContext().getRegistry().lookupByNameAndType(registerRef,
IdempotentRepository.class);
-
- ProcessorDefinition<?> route = from(pipeline.uri())
- .routeId(routeId(pipeline))
- .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
documentId);
- route = parseSteps(route, pipeline, register);
- // deduplication by document id happens inside the producer, so a
duplicate is answered
- // SKIPPED, a blank delivery releases its claim and a failed write can
be retried
- // when the id comes from a plain header, the producer is told its
name: the property set
- // above always wins, but the producer's missing-id error then names
the header the route
- // actually reads instead of the component default
- String configured = pipeline.documentId();
- String documentIdHeader = configured != null &&
!configured.contains("${") && !configured.contains("$simple{")
- ? configured
- : null;
- route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef,
registerRef, documentIdHeader));
- }
-
- /**
- * The optional parse stage; the route is returned unchanged when the
pipeline has no parser. Tika parses in-process
- * to its plain-text output — body subtree only, block boundaries
newline-separated by Tika's own handlers, the
- * encoding pinned — and {@link TikaTextDecode} then decodes those bytes
as the pinned charset, dodging the exchange
- * charset heuristic that the parsed document's own Content-Type header
could steer. Docling hands the payload to a
- * Docling Serve instance and answers finished markdown; the body is
pinned to bytes and the {@code CamelDocling*}
- * control headers are swept, because none of them may be decided by a
consumer-delivered payload.
- */
- private ProcessorDefinition<?> parseSteps(ProcessorDefinition<?> route,
IngestPipelineDefinition pipeline,
- IdempotentRepository register) {
- if (pipeline.parser() == null) {
- return route;
- }
- // a parser must never run without a captured id: the property would
be unset, and the
- // producer's header fallback would read headers a parsed document may
have written -
- // the previous engine failed such a delivery, and so does this guard
- route = route.process(exchange -> {
- String id =
exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class);
- if (id == null || id.isBlank()) {
- throw new IllegalArgumentException("Ingestion pipeline '" +
pipeline.name()
- + "': the document id resolved to nothing, and a
parser pipeline requires it before"
- + " the parse - a parsed document must not supply its
own identity");
- }
- });
- if (register != null) {
- // advisory: a known duplicate is answered SKIPPED before the
(possibly remote) parse
- // is paid for; the producer's eager claim stays authoritative, so
a duplicate racing
- // this check is still caught there
- route = route.choice()
- .when(exchange -> register
-
.contains(exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
String.class)))
- .process(exchange -> exchange.getMessage().setBody(new
IngestResult(pipeline.name(),
-
exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class), 0,
- IngestResult.Outcome.SKIPPED)))
- .stop()
- .end();
- }
- if (pipeline.maxDocumentSize() > 0) {
- // the endpoint's own cap counts extracted characters, which
protects the splitter and
- // the model but not the parse: this guard rejects the raw payload
first, before tika
- // or docling materialize it
- route = route.process(rawSizeGuard(pipeline));
- }
- return switch (pipeline.parser()) {
- case TIKA ->
route.to("tika:parse?tikaParseOutputFormat=text&tikaParseOutputEncoding=UTF-8")
- .process(new TikaTextDecode());
- case DOCLING -> route.convertBodyTo(byte[].class)
- .removeHeaders("CamelDocling*")
-
.to("docling:convert?operation=CONVERT_TO_MARKDOWN&contentInBody=true");
- };
- }
-
- private static Processor rawSizeGuard(IngestPipelineDefinition pipeline) {
- // the declared-length header spares even the read, but only the
directory pipeline's own
- // file consumer is trusted to have set it: on a consumer pipeline
every header may be
- // attacker-supplied along with the payload, so its body is always
measured - a forged
- // CamelFileLength must not talk an oversized payload past the guard
and into the parser
- boolean trustDeclaredLength = pipeline.directory() != null;
- return exchange -> {
- Long declared = trustDeclaredLength
- ? exchange.getMessage().getHeader(Exchange.FILE_LENGTH,
Long.class)
- : null;
- long size;
- byte[] bounded = null;
- if (declared != null) {
- size = declared;
- } else {
- // every body is measured through a stream - a GenericFile
streams from disk, a
- // byte[] merely wraps - so an attacker-sized payload is
rejected after
- // maxDocumentSize + 1 bytes instead of being materialized
whole in the heap just
- // to be measured; an accepted stream is consumed here, so the
bytes replace it
- // as the body. A null body carries no bytes to guard; it
flows on and becomes
- // the EMPTY outcome
- InputStream stream =
exchange.getMessage().getBody(InputStream.class);
- if (stream == null) {
- size = 0;
- } else {
- int limit = pipeline.maxDocumentSize() == Integer.MAX_VALUE
- ? Integer.MAX_VALUE
- : pipeline.maxDocumentSize() + 1;
- bounded = stream.readNBytes(limit);
- size = bounded.length;
- }
- }
- if (size > pipeline.maxDocumentSize()) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + pipeline.name() + "':
document '"
- +
exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
- String.class)
- + "' exceeds maxDocumentSize (" + size + " > "
- + pipeline.maxDocumentSize() + " bytes)");
- }
- if (bounded != null) {
- exchange.getMessage().setBody(bounded);
- }
- };
- }
-
- //
********************************************************************************
- // Helper methods
- //
********************************************************************************
-
- /**
- * The directory consumer. Built through {@code createQueryString} rather
than concatenated, so no option can be
- * injected through a crafted value. {@code noop} leaves the documents
where they are, the register keeps the same
- * file version from being ingested twice, and the {@code changed} read
lock waits for a file still being copied in.
- */
- private static String fileEndpointUri(IngestPipelineDefinition pipeline,
String registerRef) {
- Map<String, Object> options = new LinkedHashMap<>();
- options.put("noop", "true");
- options.put("idempotent", "true");
- options.put("idempotentRepository", "#bean:" + registerRef);
- // an edited file gets a new key and re-ingests; old segments remain
(append)
- options.put("idempotentKey",
"${file:absolute.path}:${file:modified}:${file:size}");
- options.put("recursive", String.valueOf(pipeline.recursive()));
- options.put("readLock", "changed");
- if (pipeline.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
- options.put("charset", "UTF-8");
- }
- return "file:" + pipeline.directory() + "?" + queryString(options);
- }
-
- private static String ingestEndpointUri(
- IngestPipelineDefinition pipeline, String storeRef, String
modelRef, String splitterRef, String registerRef,
- String documentIdHeader) {
- Map<String, Object> options = new LinkedHashMap<>();
- if (documentIdHeader != null) {
- options.put("documentIdHeader", documentIdHeader);
- }
- options.put("maxSegmentSize",
String.valueOf(pipeline.maxSegmentSize()));
- options.put("maxOverlapSize",
String.valueOf(pipeline.maxOverlapSize()));
- options.put("embeddingBatchSize",
String.valueOf(pipeline.embeddingBatchSize()));
- if (pipeline.maxDocumentSize() > 0) {
- options.put("maxDocumentSize",
String.valueOf(pipeline.maxDocumentSize()));
- }
- if (splitterRef != null) {
- options.put("documentSplitter", "#bean:" + splitterRef);
- }
- if (storeRef != null) {
- options.put("embeddingStore", "#bean:" + storeRef);
- }
- if (modelRef != null) {
- options.put("embeddingModel", "#bean:" + modelRef);
- }
- if (registerRef != null) {
- options.put("idempotentRepository", "#bean:" + registerRef);
- }
- return LangChain4jIngest.SCHEME + ":" + pipeline.name() + "?" +
queryString(options);
- }
-
- private static String queryString(Map<String, Object> options) {
- return URISupport.createQueryString(options);
- }
-
- /**
- * A bare header name is read as a header directly rather than parsed: a
dotted header name would send the simple
- * parser into OGNL. A value containing {@code ${} or {@code $simple{} is
a simple-language expression. The
- * expression is initialised here, at route build time — left to reify
lazily it would race on the first concurrent
- * exchanges.
- */
- /**
- * When the id lives in the default {@code
CamelLangChain4jIngestDocumentId} header, the 3.39
- * name is still read as a fallback — each name keeping Camel's
header-then-exchange-property
- * lookup — and the deprecation is warned once per pipeline.
- */
- @SuppressWarnings("deprecation")
- private Expression documentIdExpression(String name, String configured,
String defaultHeader) {
- if (configured != null && (configured.contains("${") ||
configured.contains("$simple{"))) {
- Expression expression =
ExpressionBuilder.simpleExpression(configured);
- expression.init(getContext());
- return expression;
- }
- String header = configured != null ? configured : defaultHeader;
- if (!IngestHeaders.DOCUMENT_ID.equals(header)) {
- Expression expression = ExpressionBuilder.headerExpression(header);
- expression.init(getContext());
- return expression;
- }
- Expression current =
ExpressionBuilder.headerExpression(IngestHeaders.DOCUMENT_ID);
- Expression legacy =
ExpressionBuilder.headerExpression(IngestHeaders.LEGACY_DOCUMENT_ID);
- current.init(getContext());
- legacy.init(getContext());
- return new ExpressionAdapter() {
- private final AtomicBoolean warned = new AtomicBoolean();
-
- @Override
- public Object evaluate(Exchange exchange) {
- Object id = current.evaluate(exchange, Object.class);
- if (id != null) {
- return id;
- }
- id = legacy.evaluate(exchange, Object.class);
- if (id != null && warned.compareAndSet(false, true)) {
- log.warn("Ingestion pipeline '{}': document id read via
the deprecated {} name"
- + " - switch the producer to {}",
- name, IngestHeaders.LEGACY_DOCUMENT_ID,
IngestHeaders.DOCUMENT_ID);
- }
- return id;
- }
- };
- }
-
- /**
- * Resolves the pipeline's duplicate register to a registry reference: a
named bean, an auto-created in-memory
- * register — the existing bean wins — or, when a directory pipeline names
none, the built-in one.
- */
- private String repositoryRef(IngestPipelineDefinition pipeline, boolean
required) {
- String ref = pipeline.idempotentRepositoryRef();
- if (pipeline.idempotentRepositoryAutoCreate()) {
- if (ref == null) {
- throw new IllegalStateException(
- "Ingestion pipeline '" + pipeline.name() + "' sets
idempotentRepositoryAutoCreate but no"
- + " idempotentRepository name to create the
register under.");
- }
- if (getContext().getRegistry().lookupByNameAndType(ref,
IdempotentRepository.class) == null) {
- getContext().getRegistry().bind(ref,
-
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
- }
- }
- if (ref == null && required) {
- ref = generatedName(pipeline, "register");
- if (getContext().getRegistry().lookupByNameAndType(ref,
IdempotentRepository.class) == null) {
- getContext().getRegistry().bind(ref,
-
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
- }
- }
- return ref;
- }
-
- /**
- * Binds a definition-supplied instance to the registry, so the endpoint
can reference it;
- * called from {@code configure()}, which Camel runs single-threaded
during startup.
- */
- private String bindInstance(Object instance, String ref,
IngestPipelineDefinition pipeline, String what) {
- if (instance == null) {
- return ref;
- }
- String name = ref != null ? ref : generatedName(pipeline, what);
- getContext().getRegistry().bind(name, instance);
- return name;
- }
-
- private void requireParserComponent(IngestPipelineDefinition pipeline) {
- if (pipeline.parser() == null) {
- return;
- }
- switch (pipeline.parser()) {
- case TIKA -> requireComponent("tika", pipeline, "its parser");
- case DOCLING -> requireComponent("docling", pipeline, "its parser");
- }
- }
-
- /**
- * Fails a pipeline whose component is not on the classpath with the
artifact to add, instead of Camel's bare
- * failed-to-resolve message much later.
- */
- private void requireComponent(String scheme, IngestPipelineDefinition
pipeline, String purpose) {
- Component component = null;
- try {
- component = getContext().getComponent(scheme, true, false);
- } catch (Exception e) {
- // fall through to the actionable message
- }
- if (component == null) {
- throw new IllegalStateException(
- "Ingestion pipeline '" + pipeline.name() + "' needs the '"
+ scheme + "' component for " + purpose
- + ", which is not on the classpath. Add the
extension that provides it, e.g. "
- + "org.apache.camel.quarkus:camel-quarkus-" +
scheme);
- }
- }
-
- private static String generatedName(IngestPipelineDefinition pipeline,
String what) {
- return LangChain4jIngest.SCHEME + "-" + pipeline.name() + "-" + what;
- }
-
- private static String routeId(IngestPipelineDefinition pipeline) {
- return LangChain4jIngest.SCHEME + "-" + pipeline.name();
- }
-}
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 88468bf595..0c895937b2 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
@@ -16,11 +16,17 @@
*/
package org.apache.camel.quarkus.component.langchain4j.ingest;
-import java.util.ArrayList;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
+import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import dev.langchain4j.data.segment.TextSegment;
@@ -32,18 +38,47 @@ import jakarta.enterprise.inject.Instance;
import jakarta.enterprise.inject.literal.NamedLiteral;
import jakarta.inject.Inject;
import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.langchain4j.ingest.IngestResult;
+import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
+import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders;
+import org.apache.camel.model.ProcessorDefinition;
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.util.URISupport;
import org.jboss.logging.Logger;
/**
* Translates the extension's configuration model — build-time and runtime
properties and
- * {@code @Ingest} builder methods — into upstream {@link
IngestPipelineDefinition}s; the routes
- * themselves are generated by the inherited {@link
IngestPipelineRouteBuilder}, and the engine
+ * {@code @Ingest} builder methods — into one route per pipeline, composed
from the catalog's
+ * langchain4j-ingest Kamelets, each adding a route named after the pipeline.
Their engine
* lives in the {@code camel-langchain4j-ingest} component. What stays here is
the Quarkus DX:
* CDI bean resolution with its actionable messages, and the
configuration-level validations.
*/
@ApplicationScoped
-public class IngestRoutes extends IngestPipelineRouteBuilder {
+public class IngestRoutes extends RouteBuilder {
+
+ /**
+ * The built-in register capacity, sized above Camel's 1000-entry default
so eviction does
+ * not re-ingest large directories during normal operation; in-memory, so
lost on restart.
+ */
+ static final int DEFAULT_REGISTER_CAPACITY = 100_000;
+
+ /** The pipeline name charset: checked at build time for {@code @Ingest}
names, at startup for all. */
+ public static final String NAME_PATTERN = "[A-Za-z0-9._-]+";
+
+ private static final String FILE_SOURCE_KAMELET =
"langchain4j-ingest-file-source";
+ private static final String SINK_KAMELET = "langchain4j-ingest-sink";
+
+ /** Every catalog Kamelet a pipeline can use, by id; the deployment module
embeds these in a native image. */
+ public static final List<String> KAMELETS =
Stream.concat(Stream.of(FILE_SOURCE_KAMELET, SINK_KAMELET),
+ Arrays.stream(Parser.values()).map(parser -> parser.actionKamelet))
+ .toList();
private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
@@ -67,9 +102,7 @@ public class IngestRoutes extends IngestPipelineRouteBuilder
{
Instance<EmbeddingModel> modelCandidates;
@Override
- protected List<IngestPipelineDefinition> pipelines() {
- List<IngestPipelineDefinition> definitions = new ArrayList<>();
-
+ public void configure() {
// a pipeline may be declared entirely through runtime properties -
the documented
// minimum is a directory and nothing else - so the two config roots
are unioned. Keying
// off the build-time map alone would make that configuration a silent
no-op, since
@@ -80,6 +113,7 @@ public class IngestRoutes extends IngestPipelineRouteBuilder
{
Set<String> names = new
TreeSet<>(buildTimeConfig.pipelines().keySet());
names.addAll(runTimeConfig.pipelines().keySet());
names.removeAll(builderDeclared);
+ rejectRouteIdClashes(names, builderDeclared);
for (String name : names) {
IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline =
buildTimeConfig.pipelines().get(name);
@@ -98,7 +132,7 @@ public class IngestRoutes extends IngestPipelineRouteBuilder
{
+ "reads one source: keep the URI, or drop it to read
the directory.");
}
- definitions.add(definition(name, uri, runtime,
+ compositionRoute(name, uri, runtime,
resolveStore(name, pipeline == null ? null :
pipeline.embeddingStore().orElse(null)),
resolveModel(name, pipeline == null ? null :
pipeline.embeddingModel().orElse(null)),
pipeline == null ?
IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE : pipeline.maxSegmentSize(),
@@ -106,27 +140,47 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
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));
}
for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
- IngestPipelineDefinition definition = builderPipeline(entry);
- if (definition != null) {
- definitions.add(definition);
+ builderPipeline(entry);
+ }
+ }
+
+ /**
+ * A pipeline named after a step of another one - {@code x} and {@code
x-source}, {@code x-parser} or
+ * {@code x-sink} - would share a route id with it; both are named here,
before any route exists, instead
+ * of the obscure failure Camel reports for the second route. A disabled
pipeline adds no route.
+ */
+ private void rejectRouteIdClashes(Set<String> configured, Set<String>
builderDeclared) {
+ Set<String> enabled = new TreeSet<>();
+ Stream.concat(configured.stream(), builderDeclared.stream())
+ .filter(name -> {
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime =
runTimeConfig.pipelines().get(name);
+ return runtime == null || runtime.enabled();
+ })
+ .forEach(enabled::add);
+ for (String name : enabled) {
+ for (String step : List.of("source", "parser", "sink")) {
+ String other = name + "-" + step;
+ if (enabled.contains(other)) {
+ throw new IllegalStateException("Ingestion pipelines '" +
name + "' and '" + other
+ + "' would both use the route id '" + id(other) +
"'. Rename one of them.");
+ }
}
}
- return definitions;
}
/** An {@code @Ingest}-declared pipeline: the builder twin of the
configuration path. */
- private IngestPipelineDefinition
builderPipeline(IngestBuilderPipelines.Entry entry) {
+ private void builderPipeline(IngestBuilderPipelines.Entry entry) {
String name = entry.name();
// configuration can still switch a builder-declared pipeline off, and
the check precedes
// the invocation so a disabled pipeline's method never runs
IngestRunTimeConfig.PipelineRunTimeConfig external =
runTimeConfig.pipelines().get(name);
if (external != null && !external.enabled()) {
LOG.infof("Ingestion pipeline '%s' (builder) is disabled", name);
- return null;
+ return;
}
// enabled is the one thing 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
@@ -143,7 +197,7 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
IngestPipeline definition = builderPipelines.definition(entry);
String uri = "file".equals(definition.sourceType()) ? null :
definition.sourceUri();
- return definition(name, uri, definition.asRunTimeConfig(),
+ compositionRoute(name, uri, definition.asRunTimeConfig(),
resolveStore(name,
definition.embeddingStoreName().orElse(null)),
resolveModel(name,
definition.embeddingModelName().orElse(null)),
definition.maxSegmentSize(),
@@ -158,44 +212,278 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
* One creation path for both declaration styles, mirroring how a builder
pipeline provides
* the same runtime-configuration view the configuration path reads.
*/
- private IngestPipelineDefinition definition(String name, String uri,
+ private void compositionRoute(String name, String uri,
IngestRunTimeConfig.PipelineRunTimeConfig runtime,
EmbeddingStore<TextSegment> store, EmbeddingModel model,
int maxSegmentSize, int maxOverlapSize, int embeddingBatchSize,
int maxDocumentSize,
String documentSplitterName, String parser) {
- IngestPipelineDefinition definition;
+ // 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
+ if (!name.matches(NAME_PATTERN)) {
+ throw new IllegalStateException(
+ "Ingestion pipeline name '" + name + "' may only contain
letters, digits, '.', '_' and '-'");
+ }
+ String storeRef = bindInstance(name, "store", store);
+ String modelRef = bindInstance(name, "model", model);
+ String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
+ String repositoryRef = repositoryRef(name, runtime, uri == null);
+
+ // options at their defaults are passed all the same, here and in the
other steps: an
+ // application-wide camel.kamelet.<kamelet>.* property would otherwise
fill them in too.
+ // An unset string option goes as "", which the component reads as
unset, and an unset
+ // documentFilter as an always-true predicate. An unset
documentSplitter or
+ // idempotentRepository is left out on purpose: "" converts to no bean
and fails the start,
+ // and neither has a neutral bean, so such a property still reaches
it, as usage.adoc warns
+ Map<String, Object> sink = new LinkedHashMap<>();
+ sink.put("pipelineName", name);
+ sink.put("documentIdHeader", LangChain4jIngestHeaders.DOCUMENT_ID);
+ 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", "");
+ // 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)));
+ sink.put("embeddingStore", "#bean:" + storeRef);
+ sink.put("embeddingModel", "#bean:" + modelRef);
+
if (uri == null) {
String directory = required(name, runtime == null ? null :
runtime.source().directory().orElse(null),
"source.directory");
- definition = IngestPipelineDefinition.directory(name, directory)
- .recursive(runtime.source().recursive());
+ // substituted into the file-source Kamelet's endpoint URI: a '?'
or '#' could
+ // inject consumer options - delete=true would consume the user's
documents
+ if (directory.contains("?") || directory.contains("#")) {
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "': the directory must not contain '?' or '#' (got
'" + directory + "')");
+ }
+ Map<String, Object> source = new LinkedHashMap<>();
+ source.put("directory", directory);
+ 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
+ source.put("charset", "UTF-8");
+ }
+ source.put("idempotentRepository", "#bean:" + repositoryRef);
+
+ ProcessorDefinition<?> route = from(kameletUri(name,
FILE_SOURCE_KAMELET, "source", source))
+ .routeId(id(name));
+ 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);
+ // 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
+ route.to(kameletUri(name, SINK_KAMELET, "sink",
sink)).process(exchange -> {
+ IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
+ if (result == null || result.outcome() !=
IngestResult.Outcome.EMPTY) {
+ return;
+ }
+ if (parser != null) {
+ LOG.warnf("Ingestion pipeline '%s': document '%s' parsed
to no text and was skipped; its key is"
+ + " committed, so it is not retried until the file
changes (missing parser module?"
+ + " image-only document?)", name,
result.documentId());
+ } else {
+ LOG.debugf("Ingestion pipeline '%s': document '%s'
contained no text, nothing was written",
+ name, result.documentId());
+ }
+ });
+ LOG.infof("Ingestion pipeline '%s': source=file:%s", name,
directory);
} else {
- definition = IngestPipelineDefinition.consumer(name, uri);
+ if (!uri.contains(":")) {
+ throw new IllegalStateException(
+ "Ingestion pipeline '" + name + "': '" + uri + "' is
not a consumer URI");
+ }
+ ProcessorDefinition<?> route = from(uri).routeId(id(name));
+ if (documentId != null &&
!IngestHeaders.DOCUMENT_ID.equals(documentId)) {
+ // copied into the canonical header the actions read, as on a
directory pipeline:
+ // evaluated against the exchange the consumer delivered,
before any parse, and a
+ // plain header name is read directly, never substituted into
an expression
+ route = route.setHeader(LangChain4jIngestHeaders.DOCUMENT_ID,
documentIdExpression(documentId));
+ if (!isSimpleExpression(documentId)) {
+ // the sink's missing-id error names the header the route
reads, as before; the
+ // parser steps keep the canonical one
+ sink.put("documentIdHeader", documentId);
+ }
+ } else {
+ // unset, or the component's canonical header (the 3.40
rename, #9162): read with the
+ // previous fallbacks, its exchange property and the 3.39 name
+ route = route.process(documentIdFallback(name));
+ }
+ // captured into the exchange property before any parse, as by the
previous builder: the
+ // sink reads the property first, so the id resolved here wins
over any header
+ route = route.process(exchange ->
exchange.setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
+
exchange.getMessage().getHeader(LangChain4jIngestHeaders.DOCUMENT_ID)));
+ IdempotentRepository register = repositoryRef == null
+ ? null
+ :
getContext().getRegistry().lookupByNameAndType(repositoryRef,
IdempotentRepository.class);
+ route = parseSteps(route, name, parser, maxDocumentSize, false,
register);
+ if (repositoryRef != null) {
+ // deduplication by document id happens inside the sink's
producer: a duplicate
+ // is answered SKIPPED, a blank delivery releases its claim
+ sink.put("idempotentRepository", "#bean:" + repositoryRef);
+ }
+ route.to(kameletUri(name, SINK_KAMELET, "sink", sink));
+ LOG.infof("Ingestion pipeline '%s': source=%s", name,
URISupport.sanitizeUri(uri));
}
- definition.embeddingStore(store)
- .embeddingModel(model)
- .splitter(maxSegmentSize, maxOverlapSize)
- .embeddingBatchSize(embeddingBatchSize);
- if (maxDocumentSize > 0) {
- definition.maxDocumentSize(maxDocumentSize);
+ }
+
+ /**
+ * 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.
+ */
+ private static ProcessorDefinition<?> parseSteps(ProcessorDefinition<?>
route, String name, String parser,
+ int maxDocumentSize, boolean directory, IdempotentRepository
register) {
+ if (parser == null) {
+ return route;
}
- if (documentSplitterName != null) {
- definition.documentSplitter(documentSplitterName);
+ String documentIdHeader = LangChain4jIngestHeaders.DOCUMENT_ID;
+ // a parser must never run without a captured id: the header the
action reads would be
+ // absent, and headers written by a parsed document could take its
place - the previous
+ // engine failed such a delivery, and so does this guard
+ route = route.process(exchange -> {
+ String id = exchange.getMessage().getHeader(documentIdHeader,
String.class);
+ if (id == null || id.isBlank()) {
+ throw new IllegalArgumentException("Ingestion pipeline '" +
name
+ + "': the document id resolved to nothing, and a
parser pipeline requires it before"
+ + " the parse - a parsed document must not supply its
own identity");
+ }
+ });
+ 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
+ // racing this check is still caught there
+ route = route.choice()
+ .when(exchange ->
register.contains(exchange.getMessage().getHeader(documentIdHeader,
String.class)))
+ .process(exchange -> exchange.getMessage().setBody(new
IngestResult(name,
+ exchange.getMessage().getHeader(documentIdHeader,
String.class), 0,
+ IngestResult.Outcome.SKIPPED)))
+ .stop()
+ .end();
}
- if (parser != null) {
- definition.parser(parser);
+ if (maxDocumentSize > 0) {
+ // the endpoint's own cap counts extracted characters, which
protects the splitter and
+ // the model but not the parse: this guard rejects the raw payload
first, before tika
+ // or docling materialize it
+ route = route.process(rawSizeGuard(name, maxDocumentSize,
directory, documentIdHeader));
}
+ Map<String, Object> action = new LinkedHashMap<>();
+ action.put("documentIdHeader", documentIdHeader);
+ // the guard above caps the raw payload; the action's own cap would
read each file whole
+ // to measure it again, so it is pinned off
+ action.put("maxDocumentSize", "0");
+ return route.to(kameletUri(name,
Parser.valueOf(parser.toUpperCase(Locale.ROOT)).actionKamelet, "parser",
action));
+ }
- String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
- if (documentId != null) {
- definition.documentId(documentId);
- } else if (uri != null) {
- // the component's own default since the 3.40 rename (#9162); the
3.39 name is still
- // read as a fallback in the route builder's documentIdExpression
- definition.documentId(IngestHeaders.DOCUMENT_ID);
+ private static Processor rawSizeGuard(String name, int maxDocumentSize,
boolean trustDeclaredLength,
+ String documentIdHeader) {
+ // the declared-length header spares even the read, but only the
directory pipeline's own
+ // file consumer is trusted to have set it: on a consumer pipeline
every header may be
+ // attacker-supplied along with the payload, so its body is always
measured - a forged
+ // CamelFileLength must not talk an oversized payload past the guard
and into the parser
+ return exchange -> {
+ Long declared = trustDeclaredLength
+ ? exchange.getMessage().getHeader(Exchange.FILE_LENGTH,
Long.class)
+ : null;
+ long size;
+ byte[] bounded = null;
+ if (declared != null) {
+ size = declared;
+ } else {
+ // every body is measured through a stream - a GenericFile
streams from disk, a
+ // byte[] merely wraps - so an attacker-sized payload is
rejected after
+ // maxDocumentSize + 1 bytes instead of being materialized
whole in the heap just
+ // to be measured; an accepted stream is consumed here, so the
bytes replace it
+ // as the body. A null body carries no bytes to guard; it
flows on and becomes
+ // the EMPTY outcome
+ InputStream stream =
exchange.getMessage().getBody(InputStream.class);
+ if (stream == null) {
+ size = 0;
+ } else {
+ int limit = maxDocumentSize == Integer.MAX_VALUE ?
Integer.MAX_VALUE : maxDocumentSize + 1;
+ bounded = stream.readNBytes(limit);
+ size = bounded.length;
+ }
+ }
+ if (size > maxDocumentSize) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': document '"
+ +
exchange.getMessage().getHeader(documentIdHeader, String.class)
+ + "' exceeds maxDocumentSize (" + size + " > "
+ maxDocumentSize + " bytes)");
+ }
+ if (bounded != null) {
+ exchange.getMessage().setBody(bounded);
+ }
+ };
+ }
+
+ /**
+ * 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
+ * Kamelet's own URI parsing unwraps them, so a {@code #bean:} prefix, an
Ant pattern's slashes and commas, or
+ * a directory path arrive at the template verbatim — percent-encoded they
would survive the substitution
+ * literally. The route the Kamelet adds is named {@code
langchain4j-ingest-<name>-<step>} rather than generated.
+ * The endpoint's own {@code location} loads the catalog Kamelet from the
classpath whatever
+ * {@code camel.component.kamelet.location} says; it is appended as is,
since the component reads it neither
+ * decoded nor unwrapped from a RAW token.
+ */
+ private static String kameletUri(String name, String kamelet, String step,
Map<String, Object> properties) {
+ Map<String, Object> raw = new LinkedHashMap<>();
+ properties.forEach((key, value) -> raw.put(key, raw(name,
String.valueOf(value))));
+ return "kamelet:" + kamelet + "/" + id(name, step) + "?"
+ + URISupport.createQueryString(raw) +
"&location=classpath:kamelets";
+ }
+
+ /** A pipeline's route id: the component's scheme name, then the pipeline
name. */
+ private static String id(String name) {
+ return LangChain4jIngest.SCHEME + "-" + name;
+ }
+
+ /** The id of a pipeline's Kamelet route or registry bean, {@code
langchain4j-ingest-<name>-<suffix>}. */
+ private static String id(String name, String suffix) {
+ return id(name) + "-" + suffix;
+ }
+
+ private static String raw(String name, String value) {
+ if (!value.contains(")")) {
+ return "RAW(" + value + ")";
+ }
+ if (!value.contains("}")) {
+ return "RAW{" + value + "}";
}
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "': a Kamelet property value containing both ')' and '}'
cannot be passed: " + value);
+ }
+
+ private static boolean isSimpleExpression(String value) {
+ return value.contains("${") || value.contains("$simple{");
+ }
+ /**
+ * Resolves the pipeline's duplicate register to a registry reference: a
named bean (existence
+ * 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.
+ */
+ private String repositoryRef(String name,
IngestRunTimeConfig.PipelineRunTimeConfig runtime, boolean directory) {
String repositoryName = runtime == null ? null :
runtime.source().idempotentRepository().orElse(null);
boolean autoCreate = runtime != null &&
runtime.source().idempotentRepositoryAutoCreate();
if (autoCreate && repositoryName == null) {
@@ -206,21 +494,79 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
if (repositoryName != null) {
IdempotentRepository repository =
getContext().getRegistry().lookupByNameAndType(repositoryName,
IdempotentRepository.class);
- if (repository == null && !autoCreate) {
- // the existence check up front, with the configuration-level
message
- throw new IllegalStateException("Ingestion pipeline '" + name
- + "' references idempotent repository '" +
repositoryName + "' but no such bean exists");
- }
- if (repository != null) {
- // whether auto-create is set or not, a pre-existing
CDI-produced repository wins
- // and does not pass through the registry's bind hook, so a
CamelContextAware
- // implementation would otherwise run contextless
+ if (repository == null) {
+ if (!autoCreate) {
+ throw new IllegalStateException("Ingestion pipeline '" +
name
+ + "' references idempotent repository '" +
repositoryName + "' but no such bean exists");
+ }
+ getContext().getRegistry().bind(repositoryName,
+
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
+ } 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());
}
- definition.idempotentRepository(repositoryName)
- .idempotentRepositoryAutoCreate(autoCreate);
+ return repositoryName;
+ }
+ if (directory) {
+ String ref = id(name, "register");
+ if (getContext().getRegistry().lookupByNameAndType(ref,
IdempotentRepository.class) == null) {
+ getContext().getRegistry().bind(ref,
+
MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
+ }
+ return ref;
}
- return definition;
+ return null;
+ }
+
+ /**
+ * When the id lives in the default {@code
CamelLangChain4jIngestDocumentId} header, it is looked
+ * up as the previous builder did: the header, then the exchange property
of that name, then the
+ * 3.39 name (header, then property), warned once per pipeline. The id
found goes into the header.
+ */
+ @SuppressWarnings("deprecation")
+ private static Processor documentIdFallback(String name) {
+ AtomicBoolean warned = new AtomicBoolean();
+ return exchange -> {
+ if (exchange.getMessage().getHeader(IngestHeaders.DOCUMENT_ID) !=
null) {
+ return;
+ }
+ Object id = exchange.getProperty(IngestHeaders.DOCUMENT_ID);
+ if (id == null) {
+ id =
exchange.getMessage().getHeader(IngestHeaders.LEGACY_DOCUMENT_ID);
+ if (id == null) {
+ id =
exchange.getProperty(IngestHeaders.LEGACY_DOCUMENT_ID);
+ }
+ if (id != null && warned.compareAndSet(false, true)) {
+ LOG.warnf("Ingestion pipeline '%s': document id read via
the deprecated %s name"
+ + " - switch the producer to %s",
+ name, IngestHeaders.LEGACY_DOCUMENT_ID,
IngestHeaders.DOCUMENT_ID);
+ }
+ }
+ if (id != null) {
+ exchange.getMessage().setHeader(IngestHeaders.DOCUMENT_ID, id);
+ }
+ };
+ }
+
+ /** Binds a CDI-resolved instance to the registry, so the Kamelet can
reference it. */
+ private String bindInstance(String name, String what, Object instance) {
+ String ref = id(name, what);
+ getContext().getRegistry().bind(ref, instance);
+ return ref;
+ }
+
+ /**
+ * A bare header name is read as a header directly rather than parsed: a
dotted header name
+ * would send the simple parser into OGNL. The expression is initialised
here, at route build
+ * time — left to reify lazily it would race on the first concurrent
exchanges.
+ */
+ private Expression documentIdExpression(String configured) {
+ Expression expression = isSimpleExpression(configured)
+ ? ExpressionBuilder.simpleExpression(configured)
+ : ExpressionBuilder.headerExpression(configured);
+ expression.init(getContext());
+ return expression;
}
private static String required(String name, String value, String property)
{
@@ -232,11 +578,11 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
}
private EmbeddingStore<TextSegment> resolveStore(String name, String
configured) {
- return resolve(name, storeCandidates, configured, "embedding store",
"embedding-store");
+ return resolve(name, storeCandidates, configured, "embedding store",
"embedding-store", "embeddingStore");
}
private EmbeddingModel resolveModel(String name, String configured) {
- return resolve(name, modelCandidates, configured, "embedding model",
"embedding-model");
+ return resolve(name, modelCandidates, configured, "embedding model",
"embedding-model", "embeddingModel");
}
/**
@@ -246,7 +592,8 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
* discovered first. A raw-typed registry search cannot serve here: it
never matches a bean
* typed {@code EmbeddingStore<TextSegment>}.
*/
- private <T> T resolve(String name, Instance<T> candidates, String
configured, String what, String property) {
+ private <T> T resolve(String name, Instance<T> candidates, String
configured, String what, String property,
+ String setter) {
if (configured != null) {
Instance<T> named = candidates.select(NamedLiteral.of(configured));
if (named.isUnsatisfied()) {
@@ -262,9 +609,12 @@ public class IngestRoutes extends
IngestPipelineRouteBuilder {
+ ", but no bean of that type exists. Define one, for
example with a @Produces method.");
}
if (handles.size() > 1) {
+ // the build rejects configuration keys for the name of a pipeline
declared in Java
+ boolean java = builderPipelines.entries().stream().anyMatch(entry
-> entry.name().equals(name));
throw new IllegalStateException("Ingestion pipeline '" + name + "'
found " + handles.size() + " "
- + what + " beans. Name the one to use with
quarkus.camel.langchain4j.ingest." + name + "."
- + property);
+ + what + " beans. Name the one to use with " + (java
+ ? "IngestPipeline." + setter + "(...) in its
@Ingest method"
+ : "quarkus.camel.langchain4j.ingest." + name + "."
+ property));
}
return handles.get(0).get();
}
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Parser.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Parser.java
new file mode 100644
index 0000000000..f70dc3cf80
--- /dev/null
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Parser.java
@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** The supported document parsers, each with the action Kamelet performing
the parse. */
+enum Parser {
+ TIKA("tika-extract-text-action"),
+ DOCLING("docling-convert-action");
+
+ // derived from the constants, so a parser added here cannot diverge from
the validated set
+ static final Set<String> NAMES = Arrays.stream(values())
+ .map(parser -> parser.name().toLowerCase(Locale.ROOT))
+ .collect(Collectors.toUnmodifiableSet());
+
+ final String actionKamelet;
+
+ Parser(String actionKamelet) {
+ this.actionKamelet = actionKamelet;
+ }
+}
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 c93e1fae3f..d407e2b599 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
@@ -55,6 +55,13 @@ public class IngestItProducers {
return new InMemoryEmbeddingStore<>();
}
+ @Produces
+ @Singleton
+ @Named("dottedfeed-store")
+ EmbeddingStore<TextSegment> dottedfeedStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
@Produces
@Singleton
@Named("reports-store")
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 af14281da2..5db526584b 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
@@ -35,6 +35,7 @@ import jakarta.ws.rs.Produces;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.langchain4j.ingest.IngestResult;
import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
@@ -61,6 +62,10 @@ public class IngestResource {
@Named("htmlfeed-store")
EmbeddingStore<TextSegment> htmlfeedStore;
+ @Inject
+ @Named("dottedfeed-store")
+ EmbeddingStore<TextSegment> dottedfeedStore;
+
@Inject
@Named("datasheets-store")
EmbeddingStore<TextSegment> datasheetsStore;
@@ -149,6 +154,7 @@ public class IngestResource {
EmbeddingStore<TextSegment> store = switch (storeName == null ?
"products" : storeName) {
case "custom" -> customStore;
case "htmlfeed" -> htmlfeedStore;
+ case "dottedfeed" -> dottedfeedStore;
case "datasheets" -> datasheetsStore;
case "s3" -> s3Store;
case "events" -> eventsStore;
@@ -178,17 +184,20 @@ public class IngestResource {
public record SearchHit(String text, String pipeline, String documentId) {
}
- /** Feeds a pipeline synchronously; the reply carries the outcome, so
tests can assert skipped and failures. */
+ /**
+ * Feeds a pipeline synchronously; the reply carries the outcome, so tests
can assert skipped and failures.
+ * The id travels in the {@code header} query parameter's header, the
canonical one by default.
+ */
@POST
@jakarta.ws.rs.Path("/feed/{pipeline}/{documentId:.+}")
@Consumes(MediaType.TEXT_PLAIN)
@Produces(MediaType.TEXT_PLAIN)
public String feed(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
- String content) {
+ @QueryParam("header") String header, String content) {
// every consumer-fed test pipeline reads direct:<pipeline>-feed
String uri = "direct:" + pipeline + "-feed";
- IngestResult result = producerTemplate.requestBodyAndHeader(uri,
content, IngestHeaders.DOCUMENT_ID,
- documentId, IngestResult.class);
+ IngestResult result = producerTemplate.requestBodyAndHeader(uri,
content,
+ header == null ? IngestHeaders.DOCUMENT_ID : header,
documentId, IngestResult.class);
return result.outcome().label();
}
@@ -219,6 +228,22 @@ public class IngestResource {
return result.outcome().label();
}
+ /** Feeds a pipeline carrying the id only as the exchange property;
answers the outcome or the failure message. */
+ @POST
+ @jakarta.ws.rs.Path("/feed-property/{pipeline}/{documentId:.+}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ public String feedProperty(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
+ String content) {
+ Exchange exchange = producerTemplate.request("direct:" + pipeline +
"-feed", e -> {
+ e.setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId);
+ e.getMessage().setBody(content);
+ });
+ return exchange.getException() != null
+ ? exchange.getException().getMessage()
+ :
exchange.getMessage().getBody(IngestResult.class).outcome().label();
+ }
+
/** Feeds a pipeline without any document id, so tests can assert the
pre-parse id guard. */
@POST
@jakarta.ws.rs.Path("/feed-anonymous/{pipeline}")
diff --git
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
index 274beec070..2440083c87 100644
---
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
+++
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
@@ -21,6 +21,11 @@ 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
+camel.kamelet.tika-extract-text-action.maxDocumentSize=1
+camel.kamelet.docling-convert-action.maxDocumentSize=1
+
quarkus.camel.langchain4j.ingest.products.source.directory=${ingest.test.directory}
quarkus.camel.langchain4j.ingest.products.embedding-store=products-store
quarkus.camel.langchain4j.ingest.products.embedding-model=test-model
@@ -74,6 +79,14 @@
quarkus.camel.langchain4j.ingest.htmlfeed.embedding-model=test-model
quarkus.camel.langchain4j.ingest.htmlfeed.max-segment-size=120
quarkus.camel.langchain4j.ingest.htmlfeed.max-overlap-size=20
+# consumer-fed parser pipeline reading its id from a dotted header name, which
the parser action
+# must never see: simple would read it as OGNL, needing camel-bean, and fail
the start
+quarkus.camel.langchain4j.ingest.dottedfeed.source.uri=direct:dottedfeed-feed
+quarkus.camel.langchain4j.ingest.dottedfeed.source.document-id=doc.id
+quarkus.camel.langchain4j.ingest.dottedfeed.parser=tika
+quarkus.camel.langchain4j.ingest.dottedfeed.embedding-store=dottedfeed-store
+quarkus.camel.langchain4j.ingest.dottedfeed.embedding-model=test-model
+
# consumer-fed pipeline; the document-id exercises the $simple{...} form and
the register gives
# first-write-wins dedup by document id
quarkus.camel.langchain4j.ingest.custom.source.uri=direct:custom-feed
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 a660a90e9d..5e06dc33c0 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
@@ -162,6 +162,24 @@ class Langchain4jIngestTikaTest {
"nothing may be ingested from an id-less delivery");
}
+ /**
+ * A consumer-fed parser pipeline still accepts the id carried only as the
+ * {@code CamelLangChain4jIngestDocumentId} exchange property, as the
previous builder did: it is
+ * copied into the header the pre-parse guard and the parser read.
+ */
+ @Test
+ void consumerFedParserReadsIdFromExchangeProperty() {
+ RestAssured.given().contentType(ContentType.TEXT)
+ .body("<html><body><p>The TAU-6 valve arrives with an exchange
property.</p></body></html>")
+ .post("/langchain4j-ingest/feed-property/htmlfeed/valve.html")
+
.then().statusCode(200).body(org.hamcrest.Matchers.is("ingested"));
+
+ Map<String, String> hit = Langchain4jIngestTest.hit("What arrives with
an exchange property?", "htmlfeed",
+ "TAU-6");
+ assertNotNull(hit, "the HTML must be parsed and its text ingested");
+ assertEquals("valve.html", hit.get("documentId"), "the id must come
from the exchange property");
+ }
+
/** The same anti-spoofing contract on a directory pipeline: the file name
wins. */
@Test
void directoryDocumentCannotSpoofItsId() {
@@ -181,4 +199,23 @@ class Langchain4jIngestTikaTest {
"the id must be the file name, not the document's
forgery");
});
}
+
+ /**
+ * A consumer-fed parser pipeline whose {@code source.document-id} is a
dotted header name
+ * ({@code doc.id}): the name is read as a header and copied into the
canonical one, so the
+ * parser action never substitutes it into its expression, where it would
fail the start.
+ */
+ @Test
+ void consumerFedParserReadsDottedIdHeader() {
+ RestAssured.given().contentType(ContentType.TEXT)
+ .queryParam("header", "doc.id")
+ .body("<html><body><p>The SIGMA-2 gauge arrives under a dotted
header.</p></body></html>")
+ .post("/langchain4j-ingest/feed/dottedfeed/gauge.html")
+
.then().statusCode(200).body(org.hamcrest.Matchers.is("ingested"));
+
+ Map<String, String> hit = Langchain4jIngestTest.hit("What arrives
under a dotted header?", "dottedfeed",
+ "SIGMA-2");
+ assertNotNull(hit, "the HTML must be parsed and its text ingested");
+ assertEquals("gauge.html", hit.get("documentId"), "the id must come
from the doc.id header");
+ }
}
diff --git a/pom.xml b/pom.xml
index 17c328fd19..9e4c70e65a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -44,7 +44,7 @@
<camel.docs.components.version>${camel.major.minor}.x</camel.docs.components.version><!--
the version in Camel's docs/components/antora.yml -->
<camel.docs.components.xref>${camel.docs.components.version}@components</camel.docs.components.xref><!--
the version in Camel's docs/components/antora.yml -->
<camel.docs.branch>camel-${camel.major.minor}.x</camel.docs.branch><!-- The
stable camel branch on which our Antora docs depends -->
- <camel-kamelets.version>4.22.1</camel-kamelets.version>
+ <camel-kamelets.version>4.23.0</camel-kamelets.version>
<cassandra-quarkus.version>1.4.1</cassandra-quarkus.version><!-- This
should be in sync with quarkus-platform
https://repo1.maven.org/maven2/com/datastax/oss/quarkus/cassandra-quarkus-bom/
-->
<debezium.version>3.7.0.Final</debezium.version> <!-- This should be
in sync with quarkus-platform https://github.com/quarkusio/quarkus-platform-->
<rocketmq.version>${rocketmq-version}</rocketmq.version>
diff --git a/poms/bom/src/main/generated/flattened-full-pom.xml
b/poms/bom/src/main/generated/flattened-full-pom.xml
index 37c22d1909..2f0d490966 100644
--- a/poms/bom/src/main/generated/flattened-full-pom.xml
+++ b/poms/bom/src/main/generated/flattened-full-pom.xml
@@ -7934,7 +7934,7 @@
<dependency>
<groupId>org.apache.camel.kamelets</groupId>
<artifactId>camel-kamelets</artifactId>
- <version>4.22.1</version>
+ <version>4.23.0</version>
</dependency>
<dependency>
<groupId>org.apache.cassandra</groupId>
diff --git a/poms/bom/src/main/generated/flattened-reduced-pom.xml
b/poms/bom/src/main/generated/flattened-reduced-pom.xml
index 758f986c42..43c7f9b0df 100644
--- a/poms/bom/src/main/generated/flattened-reduced-pom.xml
+++ b/poms/bom/src/main/generated/flattened-reduced-pom.xml
@@ -4796,7 +4796,7 @@
<dependency>
<groupId>org.apache.camel.kamelets</groupId>
<artifactId>camel-kamelets</artifactId>
- <version>4.22.1</version>
+ <version>4.23.0</version>
</dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
diff --git a/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
b/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
index 989532e4fe..77f99b10fe 100644
--- a/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
+++ b/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
@@ -4796,7 +4796,7 @@
<dependency>
<groupId>org.apache.camel.kamelets</groupId>
<artifactId>camel-kamelets</artifactId>
- <version>4.22.1</version>
+ <version>4.23.0</version>
</dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
