You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "JiriOndrusek (via GitHub)" <gi...@apache.org> on 2026/08/18 08:49:26 UTC
[PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
JiriOndrusek opened a new pull request, #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018
fixes https://github.com/apache/camel-quarkus/issues/9015
Declares the RAG ingestion pipeline instead of coding it: documents are read, split, embedded in batches and written to an EmbeddingStore, each segment carrying camel_quarkus_pipeline and camel_quarkus_document_id so retrieval can cite its source. With the RAG bridge (#9013), an @RegisterAiService answers from the store a pipeline filled — no glue code.
Declaration. Properties or Java, as strict twins: @Ingest methods (discovered via Jandex, validated at build time, invoked once at startup) convert into the configuration model, so both paths share one route-building implementation.
Sources. A directory (file consumer with guarded defaults: noop, sized idempotent register, changed read lock, UTF-8), or any of the ~300 Camel consumers — source.uri in properties, Source.endpoint(dsl -> dsl.aws2S3(...)) in Java, where dsl. completes every component's typed options. No hand-curated source types. Identity is the consumer's business: document-id names the header (CamelAwsS3Key, CamelKafkaKey) or gives an expression.
Config split. source.uri is build-time-fixed — a runtime-overridable consumer URI would be arbitrary component invocation. Per-deployment values (directory, document-id, enabled) stay runtime; placeholders in the URI resolve at startup, so credentials do too. Logged URIs are sanitized.
Engine. Deliberately naive and shipped as preview: it writes what it is given and keeps no record, so re-ingestion duplicates and restarts re-read — documented, with the synchronising engine (on tracker #9003) following separately behind the same API.
Validation. Everything decidable at build time fails there with the fix in the message: unknown component scheme (plus the add-extension command), malformed @Ingest methods, splitter bounds, conflicting source declarations — produced as ValidationErrorBuildItem so dev/test mode check too.
Tests. 15 deployment tests lock in every validation message; 6 integration tests cover directory, pushed endpoint, builder pipeline, S3 (MinIO) and Kafka (container), both scoped per class; native CI category added.
Out of scope: synchronisation, document parsing (text only), retry/DLC hooks, automatic pipeline→augmentor pairing.
<!-- Uncomment and fill this section if your PR is not trivial
[ ] An issue should be filed for the change unless this is a trivial change (fixing a typo or similar). One issue should ideally be fixed by not more than one commit and the other way round, each commit should fix just one issue, without pulling in other changes.
[ ] Each commit in the pull request should have a meaningful and properly spelled subject line and body. Copying the title of the associated issue is typically enough. Please include the issue number in the commit message prefixed by #.
[ ] The pull request description should explain what the pull request does, how, and why. If the info is available in the associated issue or some other external document, a link is enough.
[ ] Phrases like Fix #<issueNumber> or Fixes #<issueNumber> will auto-close the named issue upon merging the pull request. Using them is typically a good idea.
[ ] Please run mvn process-resources -Pformat (and amend the changes if necessary) before sending the pull request.
[ ] Contributor guide is your good friend: https://camel.apache.org/camel-quarkus/latest/contributor-guide.html
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805268502
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+import io.quarkus.runtime.annotations.ConfigDocMapKey;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithParentName;
+
+/**
+ * The shape of an ingestion pipeline: what it reads and where it writes. Locations that differ
+ * per deployment are runtime configuration, see {@link IngestRunTimeConfig}.
+ */
+@ConfigMapping(prefix = "quarkus.camel.ai.ingest")
Review Comment:
Done — renamed to quarkus.camel.langchain4j.ingest.*, route ids follow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805491096
##########
tooling/scripts/test-categories.yaml:
##########
@@ -143,6 +143,7 @@ group-07:
- ldap
- langchain4j-embeddings
- langchain4j-rag-bridge-ql4j
+ - langchain4j-ingest
Review Comment:
If we need to rebalance groups, there's a maven mojo that does it.
I can run it in isolation later at some point. So no further action needed here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805356051
##########
integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/MinioTestResource.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.net.URI;
+import java.util.Map;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+
+/**
+ * S3-compatible store for the {@code s3docs} pipeline. Enables the pipeline (disabled by
+ * default) and provides the endpoint/credentials as runtime configuration.
+ */
+public class MinioTestResource implements QuarkusTestResourceLifecycleManager {
Review Comment:
Tried it, but it doesn't fit this module: Aws2TestResource is final and module-global by design, and with restrictToAnnotatedClass it fails (classloader cast; a delegating wrapper then hits a SmallRye config ServiceConfigurationError inside start()). Using it unrestricted would start the container for every test class, losing the file-test coverage on the Windows job that the selective excludes keep — so MinIO stays for now. Happy to revisit if the support grows a restrict-friendly mode
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3804067257
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+import io.quarkus.runtime.annotations.ConfigDocMapKey;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithParentName;
+
+/**
+ * The shape of an ingestion pipeline: what it reads and where it writes. Locations that differ
+ * per deployment are runtime configuration, see {@link IngestRunTimeConfig}.
+ */
+@ConfigMapping(prefix = "quarkus.camel.ai.ingest")
Review Comment:
`quarkus.camel.ai.ingest` sits oddly next to `quarkus.camel.langchain4j.rag`, which #9013 introduced a day ago for the same feature family — and next to this extension's own name, `camel-quarkus-langchain4j-ingest`.
Worth picking one namespace now. Config keys are public API and this is much cheaper to change before the preview ships than after.
##########
integration-tests/langchain4j-ingest/pom.xml:
##########
@@ -0,0 +1,183 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-build-parent-it</artifactId>
+ <version>3.39.0-SNAPSHOT</version>
+ <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>camel-quarkus-integration-test-langchain4j-ingest</artifactId>
+ <name>Camel Quarkus :: Integration Tests :: LangChain4j Ingest</name>
+ <description>Integration tests for the declarative AI document ingestion extension</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>dev.langchain4j</groupId>
+ <artifactId>langchain4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest-jackson</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka</artifactId>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.rest-assured</groupId>
+ <artifactId>rest-assured</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-integration-tests-support-kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>native</id>
+ <activation>
+ <property>
+ <name>native</name>
+ </property>
+ </activation>
+ <properties>
+ <quarkus.native.enabled>true</quarkus.native.enabled>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>virtualDependencies</id>
+ <activation>
+ <property>
+ <name>!noVirtualDependencies</name>
+ </property>
+ </activation>
+ <dependencies>
+ <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ </profile>
+ </profiles>
Review Comment:
This module needs a `skip-testcontainers-tests` profile. The `integration-tests-alternative-platform (windows-latest)` job runs `mvn clean verify -Dskip-testcontainers-tests`, and the S3 and Kafka tests here need Docker. 74 other IT modules declare:
```xml
<profile>
<id>skip-testcontainers-tests</id>
<activation>
<property>
<name>skip-testcontainers-tests</name>
</property>
</activation>
<properties>
<skipTests>true</skipTests>
</properties>
</profile>
```
I believe this is why that check is currently failing.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
Review Comment:
The union of the two config roots works, and the comment explains why it is needed. Worth calling out the consequence somewhere user-visible, though: a pipeline declared purely through runtime properties never appears in the build-time map, so it is invisible to every build-time validation — including the duplicate-name check in `discoverBuilderPipelines`, which tests `config.pipelines().containsKey(name)` on the build-time map only.
That case is caught later by `configureBuilderPipeline`, so it is not a hole, but the two paths report it very differently.
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
Review Comment:
`private` and `static` are rejected because the method is invoked on a client proxy — but a `final` method (or a `final` declaring class) on a normal-scoped bean fails in exactly the same way, and silently: the call is not delegated, so it runs against the proxy with null injected fields.
Worth extending the check so that case gets the same clear message.
Separately, `CDI.current().select(declaringClass).get()` in `IngestBuilderPipelines` creates an instance that is never destroyed if the declaring bean happens to be `@Dependent`.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.util.URISupport;
+
+/**
+ * Fails a pipeline whose consumer component is not on the classpath with the same add-extension
+ * hint the build gives for configured URIs.
+ *
+ * <p>
+ * The build-time check cannot see a builder-declared pipeline — its URI is composed at startup —
+ * nor a configured URI whose scheme hides behind a property placeholder. And a check inside the
+ * route builder would run too late whenever a {@code camel.component.<scheme>.*} property
+ * references the missing component, because Camel Main binds those properties before any route
+ * builder runs and fails first with its bare classpath message. This check therefore runs as a
+ * pre-start task, recorded ahead of the Camel runtime assembly, which beats Camel Main's
+ * property binding.
+ */
+final class IngestComponentPresence {
+
+ private IngestComponentPresence() {
+ }
+
+ /** Walks every pipeline that consumes from a component and requires the component present. */
+ static void check(CamelContext context, IngestBuildTimeConfig buildTimeConfig,
+ IngestRunTimeConfig runTimeConfig, IngestBuilderPipelines builderPipelines) {
+ for (var entry : buildTimeConfig.pipelines().entrySet()) {
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(entry.getKey());
+ if (runtime != null && !runtime.enabled()) {
+ continue;
+ }
+ String uri = entry.getValue().source().uri().orElse(null);
+ if (uri != null) {
+ require(context, entry.getKey(), uri);
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ IngestRunTimeConfig.PipelineRunTimeConfig external = runTimeConfig.pipelines().get(entry.name());
+ if (external != null && !external.enabled()) {
+ // a disabled pipeline's @Ingest method must not run at all
+ continue;
+ }
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ // the route builder refuses this conflict with its own error; invoking the
+ // method here first would change which failure the user sees
+ continue;
+ }
+ IngestPipeline definition = builderPipelines.definition(entry);
+ if ("endpoint".equals(definition.sourceType())) {
+ require(context, entry.name(), definition.sourceUri());
+ }
+ }
+ }
+
+ private static void require(CamelContext context, String name, String uri) {
+ String resolved;
+ try {
+ resolved = context.resolvePropertyPlaceholders(uri);
+ } catch (Exception e) {
Review Comment:
Swallowing this without a trace means a typo'd `{{...}}` in `source.uri` produces Camel's later error with no indication that the presence check bailed out early. A debug log here would make that much easier to diagnose.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
+ .map(IngestBuilderPipelines.Entry::name)
+ .collect(Collectors.toSet());
+ Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet());
+ names.addAll(runTimeConfig.pipelines().keySet());
+ names.removeAll(builderDeclared);
+
+ for (String name : names) {
+ IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name);
+
+ if (runtime != null && !runtime.enabled()) {
+ LOG.infof("Ingestion pipeline '%s' is disabled", name);
+ continue;
+ }
+
+ IngestService service = new IngestService(
+ name,
+ 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(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
+
+ // a consumer URI says "consume from this"; its absence says "read that directory"
+ String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
+ if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri
+ + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ + "reads one source: keep the URI, or drop it to read the directory.");
+ }
+ if (uri == null) {
+ configureFileSource(name, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=file", name);
+ } else {
+ configureEndpointSource(name, uri, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ configureBuilderPipeline(entry);
+ }
+ }
+
+ /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
+ private void configureBuilderPipeline(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;
+ }
+ // 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
+ // (source.recursive cannot be told apart from its default, so it alone goes undetected -
+ // Source.recursive() is its builder twin)
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source "
+ + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or "
+ + "declare the pipeline in configuration instead.");
+ }
+
+ IngestPipeline definition = builderPipelines.definition(entry);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
+
+ IngestService service = new IngestService(
+ name,
+ resolveStore(name, definition.embeddingStoreName().orElse(null)),
+ resolveModel(name, definition.embeddingModelName().orElse(null)),
+ definition.maxSegmentSize(),
+ definition.maxOverlapSize());
+
+ switch (definition.sourceType()) {
+ case "file" -> configureFileSource(name, runtime, service);
+ case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
+ default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
+ }
+
+ LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
+ URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
+ }
+
+ private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ IngestService service) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
+ // space would otherwise mis-parse, and a crafted one could inject options - delete=true
+ // is honoured ahead of noop and would delete the user's documents after reading them.
+ // noop leaves the documents where they are (a knowledge base reads its source, it does
+ // not consume it), idempotent keeps the same file from being ingested twice, and the
+ // changed read lock waits for a file still being copied in rather than embedding half of
+ // it - the truncation would be permanent, since idempotent keys on the path. The register
+ // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction
+ // would re-ingest a large directory steadily during normal operation, not just on restart
+ Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
+ from(file(directory)
+ .noop(true)
+ .idempotent(true)
+ .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .recursive(runtime.source().recursive())
+ .readLock("changed")
+ .charset(StandardCharsets.UTF_8.name()))
+ .routeId(routeId(name))
+ .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class),
+ exchange.getIn().getBody(String.class)));
+ }
+
+ /**
+ * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
+ * identifies the document is the consumer's business, so {@code source.document-id} says it —
+ * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ */
+ private void configureEndpointSource(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
+ Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
+ from(uri)
+ .routeId(routeId(name))
+ .process(exchange -> {
+ String id = documentId.evaluate(exchange, String.class);
+ if (id == null) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
+ + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
+ + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the "
+ + "consumer puts it.");
+ }
+ exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
+ });
+ }
+
+ private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ String defaultHeader) {
+ String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (configured == null) {
+ return ExpressionBuilder.headerExpression(defaultHeader);
+ }
+ // a bare header name is read as a header directly rather than parsed: a dotted name such
+ // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is
+ // MicroProfile Config expansion, which would consume an expression before Camel saw it
+ return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured);
Review Comment:
`simple(configured)` returns a `ValueBuilder` wrapping a `SimpleExpression`, which is an `ExpressionDefinition`. It is never `init(CamelContext)`-ed here, so `ExpressionDefinition.evaluate` reifies it lazily on the first exchange:
```java
public <T> T evaluate(Exchange exchange, Class<T> type) {
if (expressionValue == null) {
expressionValue = createExpression(exchange.getContext());
}
...
}
```
`expressionValue` is not volatile, so two concurrent exchanges on the first poll race on it. Calling `expression.init(getContext())` while building the route removes both the race and the first-exchange reification cost.
(The `headerExpression` branch is fine — it degrades to a constant expression, which needs no init.)
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.core;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import dev.langchain4j.data.document.Document;
+import dev.langchain4j.data.document.DocumentSplitter;
+import dev.langchain4j.data.document.Metadata;
+import dev.langchain4j.data.document.splitter.DocumentSplitters;
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.jboss.logging.Logger;
+
+/**
+ * Splits a document, embeds the segments and writes them to the store.
+ *
+ * <p>
+ * Deliberately naive: it writes whatever it is given and remembers nothing, so ingesting a
+ * document twice leaves two copies. Keeping a store in step with a changing source — skipping
+ * unchanged documents, replacing changed ones, removing deleted ones — needs a record of what was
+ * written, and that engine replaces this class.
+ */
+public class IngestService {
+
+ private static final Logger LOG = Logger.getLogger(IngestService.class);
+
+ public static final String METADATA_PIPELINE = "camel_quarkus_pipeline";
Review Comment:
`camel_quarkus_pipeline` and `camel_quarkus_document_id` are the feature the PR description leads with, and they are not asserted by any test — they appear only here and in `usage.adoc`.
An assertion on the segment metadata would also pin down the `documentId("CamelAwsS3Key")` / `documentId("CamelKafkaKey")` resolution, which is currently only tested to the extent that it produces something non-null (a wrong but non-null id passes).
##########
integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/MinioTestResource.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.net.URI;
+import java.util.Map;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+
+/**
+ * S3-compatible store for the {@code s3docs} pipeline. Enables the pipeline (disabled by
+ * default) and provides the endpoint/credentials as runtime configuration.
+ */
+public class MinioTestResource implements QuarkusTestResourceLifecycleManager {
Review Comment:
`integration-tests-support/aws2` already provides `Aws2TestResource` (LocalStack), `Aws2Client` and a `DockerSocketAvailable` guard, which every other aws2 integration test uses. Reusing it would keep this module consistent with the rest of the repo, and the Docker guard would have covered the Windows job too.
##########
tooling/scripts/test-categories.yaml:
##########
@@ -143,6 +143,7 @@ group-07:
- ldap
- langchain4j-embeddings
- langchain4j-rag-bridge-ql4j
+ - langchain4j-ingest
Review Comment:
group-07 already carries `langchain4j-embeddings` and `langchain4j-rag-bridge-ql4j`. This module builds one native image and restarts it three times against MinIO and Kafka containers, so it is worth checking what it does to that group's wall time.
Minor: the entry also breaks the local alphabetical order — it belongs before `langchain4j-rag-bridge-ql4j`.
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
Review Comment:
These `@Ingest` shape checks throw `ConfigurationException` directly, while `validatePipelines` and `validateConnectorsPresent` produce `ValidationErrorBuildItem`.
The PR description says validation is produced as `ValidationErrorBuildItem` so dev and test mode check too — but these are the checks users are most likely to hit, and they take the other path.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
+ .map(IngestBuilderPipelines.Entry::name)
+ .collect(Collectors.toSet());
+ Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet());
+ names.addAll(runTimeConfig.pipelines().keySet());
+ names.removeAll(builderDeclared);
+
+ for (String name : names) {
+ IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name);
+
+ if (runtime != null && !runtime.enabled()) {
+ LOG.infof("Ingestion pipeline '%s' is disabled", name);
+ continue;
+ }
+
+ IngestService service = new IngestService(
+ name,
+ 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(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
+
+ // a consumer URI says "consume from this"; its absence says "read that directory"
+ String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
+ if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri
+ + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ + "reads one source: keep the URI, or drop it to read the directory.");
+ }
+ if (uri == null) {
+ configureFileSource(name, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=file", name);
+ } else {
+ configureEndpointSource(name, uri, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ configureBuilderPipeline(entry);
+ }
+ }
+
+ /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
+ private void configureBuilderPipeline(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;
+ }
+ // 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
+ // (source.recursive cannot be told apart from its default, so it alone goes undetected -
+ // Source.recursive() is its builder twin)
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source "
+ + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or "
+ + "declare the pipeline in configuration instead.");
+ }
+
+ IngestPipeline definition = builderPipelines.definition(entry);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
+
+ IngestService service = new IngestService(
+ name,
+ resolveStore(name, definition.embeddingStoreName().orElse(null)),
+ resolveModel(name, definition.embeddingModelName().orElse(null)),
+ definition.maxSegmentSize(),
+ definition.maxOverlapSize());
+
+ switch (definition.sourceType()) {
+ case "file" -> configureFileSource(name, runtime, service);
+ case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
+ default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
+ }
+
+ LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
+ URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
+ }
+
+ private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ IngestService service) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
+ // space would otherwise mis-parse, and a crafted one could inject options - delete=true
+ // is honoured ahead of noop and would delete the user's documents after reading them.
+ // noop leaves the documents where they are (a knowledge base reads its source, it does
+ // not consume it), idempotent keeps the same file from being ingested twice, and the
+ // changed read lock waits for a file still being copied in rather than embedding half of
+ // it - the truncation would be permanent, since idempotent keys on the path. The register
+ // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction
+ // would re-ingest a large directory steadily during normal operation, not just on restart
+ Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
+ from(file(directory)
+ .noop(true)
+ .idempotent(true)
+ .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .recursive(runtime.source().recursive())
+ .readLock("changed")
+ .charset(StandardCharsets.UTF_8.name()))
+ .routeId(routeId(name))
+ .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class),
+ exchange.getIn().getBody(String.class)));
+ }
+
+ /**
+ * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
+ * identifies the document is the consumer's business, so {@code source.document-id} says it —
+ * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ */
+ private void configureEndpointSource(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
+ Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
+ from(uri)
+ .routeId(routeId(name))
+ .process(exchange -> {
+ String id = documentId.evaluate(exchange, String.class);
+ if (id == null) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
+ + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
+ + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the "
+ + "consumer puts it.");
+ }
+ exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
+ });
+ }
+
+ private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ String defaultHeader) {
+ String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (configured == null) {
+ return ExpressionBuilder.headerExpression(defaultHeader);
+ }
+ // a bare header name is read as a header directly rather than parsed: a dotted name such
+ // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is
+ // MicroProfile Config expansion, which would consume an expression before Camel saw it
+ return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured);
+ }
+
+ private static String routeId(String name) {
+ return "camel-quarkus-ai-ingest-" + name;
+ }
+
+ private static String required(String name, String value, String property) {
+ if (value == null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' has no " + property
+ + ". Set quarkus.camel.ai.ingest." + name + "." + property);
+ }
+ return value;
+ }
+
+ private EmbeddingStore<TextSegment> resolveStore(String name, String configured) {
Review Comment:
Named lookup goes through the Camel registry here, while the unnamed path in `single()` goes through the CDI `Instance`. The two see different sets: a store bound only via `@BindToRegistry` resolves by name but is invisible to the "exactly one" count, and the reverse holds for a CDI bean that never reaches the registry.
Worth using one mechanism for both.
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
+ }
+ if (!names.add(name) || config.pipelines().containsKey(name)) {
+ throw new ConfigurationException("Ingestion pipeline '" + name + "' is declared more than once "
+ + "(builder and/or configuration). Pipeline names must be unique.");
+ }
+
+ flatEntries.add(name);
+ flatEntries.add(method.declaringClass().name().toString());
+ flatEntries.add(method.name());
+ beanClasses.add(method.declaringClass().name().toString());
+ }
+
+ if (!beanClasses.isEmpty()) {
+ beans.produce(AdditionalBeanBuildItem.builder()
+ .addBeanClasses(beanClasses.toArray(new String[0]))
+ .setUnremovable()
+ .build());
+ reflectiveClasses.produce(ReflectiveClassBuildItem.builder(beanClasses.toArray(new String[0]))
+ .methods()
+ .build());
+ }
+
+ syntheticBeans.produce(SyntheticBeanBuildItem.configure(IngestBuilderPipelines.class)
+ .scope(Singleton.class)
+ .unremovable()
+ .runtimeValue(recorder.createBuilderPipelines(flatEntries))
+ .done());
+ }
+
+ /**
+ * The pre-start half of the component-presence check: recorded as a Camel runtime task, it
+ * runs after ArC is fully initialised but before the Camel runtime is assembled — and thus
+ * before Camel Main binds {@code camel.component.*} properties, whose failure for a missing
+ * component would otherwise preempt the friendlier add-extension hint.
+ */
+ @BuildStep
+ @Record(ExecutionTime.RUNTIME_INIT)
+ @Consume(SyntheticBeansRuntimeInitBuildItem.class)
+ CamelRuntimeTaskBuildItem checkComponentsPresent(Langchain4jIngestRecorder recorder,
+ CamelContextBuildItem camelContext) {
+ recorder.checkComponentsPresent(camelContext.getCamelContext());
+ return new CamelRuntimeTaskBuildItem("langchain4j-ingest-components");
+ }
+
+ /**
+ * A pipeline whose consumer URI names a component that is not on the classpath stops the
+ * build, with the command that fixes it rather than a startup failure. Only configured URIs
+ * can be checked: a builder-declared pipeline composes its URI at startup, where the pre-start
+ * task above applies the same hint. Component services are REGISTRY-destination, so they are
+ * read from the application archives directly — they never appear among the DISCOVERY
+ * {@code CamelServiceBuildItem}s.
+ */
+ @BuildStep
+ void validateConnectorsPresent(IngestBuildTimeConfig config, ApplicationArchivesBuildItem applicationArchives,
+ BuildProducer<ValidationErrorBuildItem> validationErrors) {
+ PathFilter pathFilter = new PathFilter.Builder()
+ .include("META-INF/services/org/apache/camel/component/*")
+ .build();
+ Set<String> components = CamelSupport.services(applicationArchives, pathFilter)
+ .map(CamelServiceBuildItem::getName)
+ .collect(Collectors.toSet());
+
+ for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig> entry : config.pipelines().entrySet()) {
+ String uri = entry.getValue().source().uri().orElse(null);
+ if (uri == null) {
+ continue;
+ }
+ int colon = uri.indexOf(':');
+ String scheme = colon < 1 ? uri : uri.substring(0, colon);
+ // a placeholder resolves at startup, so its scheme cannot be known here
+ if (scheme.contains("{{") || scheme.contains("$")) {
+ continue;
+ }
+ if (!components.contains(scheme)) {
+ // the URI is sanitized: a consumer URI may legitimately carry credentials, and a
+ // build log is no place for them. The artifact hint is a heuristic - multi-scheme
+ // components (smtp -> camel-quarkus-mail) name their extension differently
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' consumes from '"
+ + URISupport.sanitizeUri(uri) + "', but the Camel component '" + scheme
+ + "' is not on the classpath.\nAdd the extension that provides it, usually:"
+ + " ./mvnw quarkus:add-extension -Dextensions=camel-quarkus-" + scheme)));
Review Comment:
Three things about this hint (it also appears in `IngestComponentPresence`):
- `./mvnw` assumes a wrapper in the user's project. Every user-facing page under `docs/modules/ROOT/pages/user-guide/` uses plain `mvn`; `./mvnw` appears only in the contributor guide, i.e. for building Camel Quarkus itself. This is the first extension in the repo to emit such a hint, so it sets the precedent.
- It assumes Maven. Gradle and JBang are both first-class Quarkus paths and would get a command that cannot work.
- `camel-quarkus-<scheme>` is a heuristic, as the comment above acknowledges — `smtp`/`imaps` are `camel-quarkus-mail`, `https` is `camel-quarkus-http`.
Naming the dependency rather than prescribing a command avoids all three, e.g. "Add the extension that provides it, e.g. `org.apache.camel.quarkus:camel-quarkus-<scheme>`".
Also, the `\n` is on the wrong side of the concatenation — this currently renders as `usually: ./mvnw ...` on one line.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+import io.quarkus.runtime.annotations.ConfigDocMapKey;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithParentName;
+
+/**
+ * Runtime configuration of ingestion pipelines: concrete locations and switches that may differ
+ * per deployment. The pipeline topology is build-time, see {@link IngestBuildTimeConfig}.
+ */
+@ConfigMapping(prefix = "quarkus.camel.ai.ingest")
+@ConfigRoot(phase = ConfigPhase.RUN_TIME)
+public interface IngestRunTimeConfig {
+
+ /**
+ * Ingestion pipelines by name.
+ */
+ @WithParentName
+ @ConfigDocMapKey("pipeline-name")
+ Map<String, PipelineRunTimeConfig> pipelines();
+
+ interface PipelineRunTimeConfig {
+
+ /**
+ * Whether this pipeline starts. Useful to switch ingestion off in dev mode.
+ */
+ @WithDefault("true")
+ boolean enabled();
+
+ /**
+ * The document source.
+ */
+ SourceRunTimeConfig source();
+
+ interface SourceRunTimeConfig {
+
+ /**
+ * The directory to ingest documents from, for a pipeline that has no `source.uri`. A
+ * path is a deployment concern, so unlike the URI it stays runtime configuration.
+ * Setting both is an error.
+ */
+ Optional<String> directory();
+
+ /**
+ * Whether subdirectories are ingested too, when reading a directory.
+ */
+ @WithDefault("true")
+ boolean recursive();
+
+ /**
+ * Where the document id lives in the exchange the consumer delivers: normally the
+ * name of a header, such as `CamelAwsS3Key` for an S3 consumer or `CamelKafkaKey` for
+ * a Kafka one. A value containing a dollar-brace placeholder is taken as a
Review Comment:
This documents a syntax the option cannot actually accept: MicroProfile Config expands `${...}` before Camel ever sees the value, as the javadoc goes on to admit. So the simple-expression branch is only reachable from `Source.documentId(...)` in Java, not from a properties file at all.
Either drop it from the config option's contract, or use a marker Config will not expand (`simple:...`, or Camel's `$simple{...}`) and document `\${...}` escaping.
##########
integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingSearchRequest;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestHeaders;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+@jakarta.ws.rs.Path("/langchain4j-ingest")
+public class IngestResource {
+
+ @Inject
+ @Named("products-store")
+ EmbeddingStore<TextSegment> productsStore;
+
+ @Inject
+ @Named("test-model")
+ EmbeddingModel model;
+
+ @Inject
+ @Named("custom-store")
+ EmbeddingStore<TextSegment> customStore;
+
+ @Inject
+ @Named("built-store")
+ EmbeddingStore<TextSegment> builtStore;
+
+ @Inject
+ @Named("s3-store")
+ EmbeddingStore<TextSegment> s3Store;
+
+ @Inject
+ @Named("events-store")
+ EmbeddingStore<TextSegment> eventsStore;
+
+ @Inject
+ ProducerTemplate producerTemplate;
+
+ @ConfigProperty(name = "ingest.test.directory")
+ String directory;
+
+ /** Writes a document into the watched directory — app-side, so native mode shares the path. */
+ @POST
+ @jakarta.ws.rs.Path("/file/{name}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ public void writeFile(@PathParam("name") String name, String content) throws Exception {
+ Path dir = Path.of(directory);
+ Files.createDirectories(dir);
+ Files.writeString(dir.resolve(name), content);
+ }
+
+ @GET
+ @jakarta.ws.rs.Path("/search")
+ @Produces(MediaType.APPLICATION_JSON)
+ public List<String> search(@QueryParam("q") String query, @QueryParam("store") String storeName) {
+ EmbeddingStore<TextSegment> store = switch (storeName == null ? "products" : storeName) {
+ case "custom" -> customStore;
+ case "built" -> builtStore;
+ case "s3" -> s3Store;
+ case "events" -> eventsStore;
+ default -> productsStore;
+ };
+ var result = store.search(EmbeddingSearchRequest.builder()
+ .queryEmbedding(model.embed(query).content())
+ .maxResults(20)
+ .minScore(0.0)
Review Comment:
`DeterministicEmbeddingModel` produces hash-seeded random unit vectors, so a query embedding has no semantic relationship to the document embeddings. With `maxResults(20)` and `minScore(0.0)`, this search effectively returns the whole store.
That is fine as an ingestion assertion, but it means tests named `...AndRetrievable` are really asserting "the text reached the store" — and the hardcoded 20 will silently start dropping matches once a shared store holds more segments than that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805320641
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
+ }
+ if (!names.add(name) || config.pipelines().containsKey(name)) {
+ throw new ConfigurationException("Ingestion pipeline '" + name + "' is declared more than once "
+ + "(builder and/or configuration). Pipeline names must be unique.");
+ }
+
+ flatEntries.add(name);
+ flatEntries.add(method.declaringClass().name().toString());
+ flatEntries.add(method.name());
+ beanClasses.add(method.declaringClass().name().toString());
+ }
+
+ if (!beanClasses.isEmpty()) {
+ beans.produce(AdditionalBeanBuildItem.builder()
+ .addBeanClasses(beanClasses.toArray(new String[0]))
+ .setUnremovable()
+ .build());
+ reflectiveClasses.produce(ReflectiveClassBuildItem.builder(beanClasses.toArray(new String[0]))
+ .methods()
+ .build());
+ }
+
+ syntheticBeans.produce(SyntheticBeanBuildItem.configure(IngestBuilderPipelines.class)
+ .scope(Singleton.class)
+ .unremovable()
+ .runtimeValue(recorder.createBuilderPipelines(flatEntries))
+ .done());
+ }
+
+ /**
+ * The pre-start half of the component-presence check: recorded as a Camel runtime task, it
+ * runs after ArC is fully initialised but before the Camel runtime is assembled — and thus
+ * before Camel Main binds {@code camel.component.*} properties, whose failure for a missing
+ * component would otherwise preempt the friendlier add-extension hint.
+ */
+ @BuildStep
+ @Record(ExecutionTime.RUNTIME_INIT)
+ @Consume(SyntheticBeansRuntimeInitBuildItem.class)
+ CamelRuntimeTaskBuildItem checkComponentsPresent(Langchain4jIngestRecorder recorder,
+ CamelContextBuildItem camelContext) {
+ recorder.checkComponentsPresent(camelContext.getCamelContext());
+ return new CamelRuntimeTaskBuildItem("langchain4j-ingest-components");
+ }
+
+ /**
+ * A pipeline whose consumer URI names a component that is not on the classpath stops the
+ * build, with the command that fixes it rather than a startup failure. Only configured URIs
+ * can be checked: a builder-declared pipeline composes its URI at startup, where the pre-start
+ * task above applies the same hint. Component services are REGISTRY-destination, so they are
+ * read from the application archives directly — they never appear among the DISCOVERY
+ * {@code CamelServiceBuildItem}s.
+ */
+ @BuildStep
+ void validateConnectorsPresent(IngestBuildTimeConfig config, ApplicationArchivesBuildItem applicationArchives,
+ BuildProducer<ValidationErrorBuildItem> validationErrors) {
+ PathFilter pathFilter = new PathFilter.Builder()
+ .include("META-INF/services/org/apache/camel/component/*")
+ .build();
+ Set<String> components = CamelSupport.services(applicationArchives, pathFilter)
+ .map(CamelServiceBuildItem::getName)
+ .collect(Collectors.toSet());
+
+ for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig> entry : config.pipelines().entrySet()) {
+ String uri = entry.getValue().source().uri().orElse(null);
+ if (uri == null) {
+ continue;
+ }
+ int colon = uri.indexOf(':');
+ String scheme = colon < 1 ? uri : uri.substring(0, colon);
+ // a placeholder resolves at startup, so its scheme cannot be known here
+ if (scheme.contains("{{") || scheme.contains("$")) {
+ continue;
+ }
+ if (!components.contains(scheme)) {
+ // the URI is sanitized: a consumer URI may legitimately carry credentials, and a
+ // build log is no place for them. The artifact hint is a heuristic - multi-scheme
+ // components (smtp -> camel-quarkus-mail) name their extension differently
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' consumes from '"
+ + URISupport.sanitizeUri(uri) + "', but the Camel component '" + scheme
+ + "' is not on the classpath.\nAdd the extension that provides it, usually:"
+ + " ./mvnw quarkus:add-extension -Dextensions=camel-quarkus-" + scheme)));
Review Comment:
Done — the message now names org.apache.camel.quarkus:camel-quarkus-<scheme> and prescribes no command in both checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805310340
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
Review Comment:
Done — the @Ingest shape checks now produce ValidationErrorBuildItems and report together.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek merged PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805279853
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
+ .map(IngestBuilderPipelines.Entry::name)
+ .collect(Collectors.toSet());
+ Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet());
+ names.addAll(runTimeConfig.pipelines().keySet());
+ names.removeAll(builderDeclared);
+
+ for (String name : names) {
+ IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name);
+
+ if (runtime != null && !runtime.enabled()) {
+ LOG.infof("Ingestion pipeline '%s' is disabled", name);
+ continue;
+ }
+
+ IngestService service = new IngestService(
+ name,
+ 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(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
+
+ // a consumer URI says "consume from this"; its absence says "read that directory"
+ String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
+ if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri
+ + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ + "reads one source: keep the URI, or drop it to read the directory.");
+ }
+ if (uri == null) {
+ configureFileSource(name, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=file", name);
+ } else {
+ configureEndpointSource(name, uri, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ configureBuilderPipeline(entry);
+ }
+ }
+
+ /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
+ private void configureBuilderPipeline(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;
+ }
+ // 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
+ // (source.recursive cannot be told apart from its default, so it alone goes undetected -
+ // Source.recursive() is its builder twin)
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source "
+ + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or "
+ + "declare the pipeline in configuration instead.");
+ }
+
+ IngestPipeline definition = builderPipelines.definition(entry);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
+
+ IngestService service = new IngestService(
+ name,
+ resolveStore(name, definition.embeddingStoreName().orElse(null)),
+ resolveModel(name, definition.embeddingModelName().orElse(null)),
+ definition.maxSegmentSize(),
+ definition.maxOverlapSize());
+
+ switch (definition.sourceType()) {
+ case "file" -> configureFileSource(name, runtime, service);
+ case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
+ default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
+ }
+
+ LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
+ URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
+ }
+
+ private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ IngestService service) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
+ // space would otherwise mis-parse, and a crafted one could inject options - delete=true
+ // is honoured ahead of noop and would delete the user's documents after reading them.
+ // noop leaves the documents where they are (a knowledge base reads its source, it does
+ // not consume it), idempotent keeps the same file from being ingested twice, and the
+ // changed read lock waits for a file still being copied in rather than embedding half of
+ // it - the truncation would be permanent, since idempotent keys on the path. The register
+ // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction
+ // would re-ingest a large directory steadily during normal operation, not just on restart
+ Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
+ from(file(directory)
+ .noop(true)
+ .idempotent(true)
+ .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .recursive(runtime.source().recursive())
+ .readLock("changed")
+ .charset(StandardCharsets.UTF_8.name()))
+ .routeId(routeId(name))
+ .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class),
+ exchange.getIn().getBody(String.class)));
+ }
+
+ /**
+ * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
+ * identifies the document is the consumer's business, so {@code source.document-id} says it —
+ * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ */
+ private void configureEndpointSource(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
+ Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
+ from(uri)
+ .routeId(routeId(name))
+ .process(exchange -> {
+ String id = documentId.evaluate(exchange, String.class);
+ if (id == null) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
+ + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
+ + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the "
+ + "consumer puts it.");
+ }
+ exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
+ });
+ }
+
+ private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ String defaultHeader) {
+ String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (configured == null) {
+ return ExpressionBuilder.headerExpression(defaultHeader);
+ }
+ // a bare header name is read as a header directly rather than parsed: a dotted name such
+ // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is
+ // MicroProfile Config expansion, which would consume an expression before Camel saw it
+ return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured);
Review Comment:
Done — the expression is built via ExpressionBuilder.simpleExpression and init()-ed at route build time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805265665
##########
integration-tests/langchain4j-ingest/pom.xml:
##########
@@ -0,0 +1,183 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-build-parent-it</artifactId>
+ <version>3.39.0-SNAPSHOT</version>
+ <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>camel-quarkus-integration-test-langchain4j-ingest</artifactId>
+ <name>Camel Quarkus :: Integration Tests :: LangChain4j Ingest</name>
+ <description>Integration tests for the declarative AI document ingestion extension</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>dev.langchain4j</groupId>
+ <artifactId>langchain4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest-jackson</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka</artifactId>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.rest-assured</groupId>
+ <artifactId>rest-assured</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-integration-tests-support-kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>native</id>
+ <activation>
+ <property>
+ <name>native</name>
+ </property>
+ </activation>
+ <properties>
+ <quarkus.native.enabled>true</quarkus.native.enabled>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>virtualDependencies</id>
+ <activation>
+ <property>
+ <name>!noVirtualDependencies</name>
+ </property>
+ </activation>
+ <dependencies>
+ <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ </profile>
+ </profiles>
Review Comment:
Done in 2541fc3a5b — profile added; only the two container-backed test classes are excluded, so the file and endpoint tests keep running on that job (the pom comment explains the deviation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805293834
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
Review Comment:
Done — the startup-only validation of runtime-only pipelines is now called out in usage.adoc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805306580
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
+ .map(IngestBuilderPipelines.Entry::name)
+ .collect(Collectors.toSet());
+ Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet());
+ names.addAll(runTimeConfig.pipelines().keySet());
+ names.removeAll(builderDeclared);
+
+ for (String name : names) {
+ IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name);
+
+ if (runtime != null && !runtime.enabled()) {
+ LOG.infof("Ingestion pipeline '%s' is disabled", name);
+ continue;
+ }
+
+ IngestService service = new IngestService(
+ name,
+ 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(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
+
+ // a consumer URI says "consume from this"; its absence says "read that directory"
+ String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
+ if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri
+ + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ + "reads one source: keep the URI, or drop it to read the directory.");
+ }
+ if (uri == null) {
+ configureFileSource(name, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=file", name);
+ } else {
+ configureEndpointSource(name, uri, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ configureBuilderPipeline(entry);
+ }
+ }
+
+ /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
+ private void configureBuilderPipeline(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;
+ }
+ // 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
+ // (source.recursive cannot be told apart from its default, so it alone goes undetected -
+ // Source.recursive() is its builder twin)
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source "
+ + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or "
+ + "declare the pipeline in configuration instead.");
+ }
+
+ IngestPipeline definition = builderPipelines.definition(entry);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
+
+ IngestService service = new IngestService(
+ name,
+ resolveStore(name, definition.embeddingStoreName().orElse(null)),
+ resolveModel(name, definition.embeddingModelName().orElse(null)),
+ definition.maxSegmentSize(),
+ definition.maxOverlapSize());
+
+ switch (definition.sourceType()) {
+ case "file" -> configureFileSource(name, runtime, service);
+ case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
+ default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
+ }
+
+ LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
+ URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
+ }
+
+ private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ IngestService service) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
+ // space would otherwise mis-parse, and a crafted one could inject options - delete=true
+ // is honoured ahead of noop and would delete the user's documents after reading them.
+ // noop leaves the documents where they are (a knowledge base reads its source, it does
+ // not consume it), idempotent keeps the same file from being ingested twice, and the
+ // changed read lock waits for a file still being copied in rather than embedding half of
+ // it - the truncation would be permanent, since idempotent keys on the path. The register
+ // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction
+ // would re-ingest a large directory steadily during normal operation, not just on restart
+ Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
+ from(file(directory)
+ .noop(true)
+ .idempotent(true)
+ .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .recursive(runtime.source().recursive())
+ .readLock("changed")
+ .charset(StandardCharsets.UTF_8.name()))
+ .routeId(routeId(name))
+ .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class),
+ exchange.getIn().getBody(String.class)));
+ }
+
+ /**
+ * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
+ * identifies the document is the consumer's business, so {@code source.document-id} says it —
+ * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ */
+ private void configureEndpointSource(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
+ Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
+ from(uri)
+ .routeId(routeId(name))
+ .process(exchange -> {
+ String id = documentId.evaluate(exchange, String.class);
+ if (id == null) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
+ + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
+ + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the "
+ + "consumer puts it.");
+ }
+ exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
+ });
+ }
+
+ private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ String defaultHeader) {
+ String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (configured == null) {
+ return ExpressionBuilder.headerExpression(defaultHeader);
+ }
+ // a bare header name is read as a header directly rather than parsed: a dotted name such
+ // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is
+ // MicroProfile Config expansion, which would consume an expression before Camel saw it
+ return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured);
+ }
+
+ private static String routeId(String name) {
+ return "camel-quarkus-ai-ingest-" + name;
+ }
+
+ private static String required(String name, String value, String property) {
+ if (value == null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' has no " + property
+ + ". Set quarkus.camel.ai.ingest." + name + "." + property);
+ }
+ return value;
+ }
+
+ private EmbeddingStore<TextSegment> resolveStore(String name, String configured) {
Review Comment:
Done — the unnamed lookup now goes through Registry.findByType, the same mechanism as the named one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805315777
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
Review Comment:
Done — final methods and final declaring classes are rejected at build time (new test), and @Dependent beans are destroyed after the invocation via Instance.getHandle().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#issuecomment-5330191594
Thanks for the thorough review. Everything is addressed in 2541fc3a5b — kept deliberately as a separate commit so the delta stays reviewable against the original; I'll squash it into the main commit on merge if you prefer a single-commit history.
The two settle-before-merge items:
- `skip-testcontainers-tests` is declared. I excluded only the two container-backed test classes instead of the module-wide `skipTests`, so the file and endpoint tests still run on the alternative-platform job — a pom comment explains the deviation from the convention.
- The config namespace is now `quarkus.camel.langchain4j.ingest.*`, aligned with #9013 and the extension name; route ids follow suit.
The inline findings are applied — short confirmation on each thread — with one decline: reusing the shared aws2 test support. `Aws2TestResource` is final and module-global by design and does not survive `restrictToAnnotatedClass`, which this module needs to keep its Docker-free tests running on Windows; details on that thread.
On the scope question: agreed it deserves an explicit record — I'll follow up on #9015 with the plan. Short version: the engine core is deliberately Quarkus-free, the intent is to propose a `camel-langchain4j-ingest` component upstream and have this extension delegate to it while keeping the declarative surface (properties, `@Ingest`, build-time validation) here; the metadata keys will be parameterized so upstream can use runtime-neutral names.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805324377
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.util.URISupport;
+
+/**
+ * Fails a pipeline whose consumer component is not on the classpath with the same add-extension
+ * hint the build gives for configured URIs.
+ *
+ * <p>
+ * The build-time check cannot see a builder-declared pipeline — its URI is composed at startup —
+ * nor a configured URI whose scheme hides behind a property placeholder. And a check inside the
+ * route builder would run too late whenever a {@code camel.component.<scheme>.*} property
+ * references the missing component, because Camel Main binds those properties before any route
+ * builder runs and fails first with its bare classpath message. This check therefore runs as a
+ * pre-start task, recorded ahead of the Camel runtime assembly, which beats Camel Main's
+ * property binding.
+ */
+final class IngestComponentPresence {
+
+ private IngestComponentPresence() {
+ }
+
+ /** Walks every pipeline that consumes from a component and requires the component present. */
+ static void check(CamelContext context, IngestBuildTimeConfig buildTimeConfig,
+ IngestRunTimeConfig runTimeConfig, IngestBuilderPipelines builderPipelines) {
+ for (var entry : buildTimeConfig.pipelines().entrySet()) {
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(entry.getKey());
+ if (runtime != null && !runtime.enabled()) {
+ continue;
+ }
+ String uri = entry.getValue().source().uri().orElse(null);
+ if (uri != null) {
+ require(context, entry.getKey(), uri);
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ IngestRunTimeConfig.PipelineRunTimeConfig external = runTimeConfig.pipelines().get(entry.name());
+ if (external != null && !external.enabled()) {
+ // a disabled pipeline's @Ingest method must not run at all
+ continue;
+ }
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ // the route builder refuses this conflict with its own error; invoking the
+ // method here first would change which failure the user sees
+ continue;
+ }
+ IngestPipeline definition = builderPipelines.definition(entry);
+ if ("endpoint".equals(definition.sourceType())) {
+ require(context, entry.name(), definition.sourceUri());
+ }
+ }
+ }
+
+ private static void require(CamelContext context, String name, String uri) {
+ String resolved;
+ try {
+ resolved = context.resolvePropertyPlaceholders(uri);
+ } catch (Exception e) {
Review Comment:
Done — the skipped check is debug-logged with the pipeline name and the cause.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805330784
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+import io.quarkus.runtime.annotations.ConfigDocMapKey;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithParentName;
+
+/**
+ * Runtime configuration of ingestion pipelines: concrete locations and switches that may differ
+ * per deployment. The pipeline topology is build-time, see {@link IngestBuildTimeConfig}.
+ */
+@ConfigMapping(prefix = "quarkus.camel.ai.ingest")
+@ConfigRoot(phase = ConfigPhase.RUN_TIME)
+public interface IngestRunTimeConfig {
+
+ /**
+ * Ingestion pipelines by name.
+ */
+ @WithParentName
+ @ConfigDocMapKey("pipeline-name")
+ Map<String, PipelineRunTimeConfig> pipelines();
+
+ interface PipelineRunTimeConfig {
+
+ /**
+ * Whether this pipeline starts. Useful to switch ingestion off in dev mode.
+ */
+ @WithDefault("true")
+ boolean enabled();
+
+ /**
+ * The document source.
+ */
+ SourceRunTimeConfig source();
+
+ interface SourceRunTimeConfig {
+
+ /**
+ * The directory to ingest documents from, for a pipeline that has no `source.uri`. A
+ * path is a deployment concern, so unlike the URI it stays runtime configuration.
+ * Setting both is an error.
+ */
+ Optional<String> directory();
+
+ /**
+ * Whether subdirectories are ingested too, when reading a directory.
+ */
+ @WithDefault("true")
+ boolean recursive();
+
+ /**
+ * Where the document id lives in the exchange the consumer delivers: normally the
+ * name of a header, such as `CamelAwsS3Key` for an S3 consumer or `CamelKafkaKey` for
+ * a Kafka one. A value containing a dollar-brace placeholder is taken as a
Review Comment:
Done — $simple{...} is now detected and documented as the properties form (exercised end-to-end); ${...} stays for the Java builder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805334680
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.core;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import dev.langchain4j.data.document.Document;
+import dev.langchain4j.data.document.DocumentSplitter;
+import dev.langchain4j.data.document.Metadata;
+import dev.langchain4j.data.document.splitter.DocumentSplitters;
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.jboss.logging.Logger;
+
+/**
+ * Splits a document, embeds the segments and writes them to the store.
+ *
+ * <p>
+ * Deliberately naive: it writes whatever it is given and remembers nothing, so ingesting a
+ * document twice leaves two copies. Keeping a store in step with a changing source — skipping
+ * unchanged documents, replacing changed ones, removing deleted ones — needs a record of what was
+ * written, and that engine replaces this class.
+ */
+public class IngestService {
+
+ private static final Logger LOG = Logger.getLogger(IngestService.class);
+
+ public static final String METADATA_PIPELINE = "camel_quarkus_pipeline";
Review Comment:
Done — the search endpoint returns (text, pipeline, documentId) and all five pipelines assert the metadata, pinning the CamelAwsS3Key/CamelKafkaKey resolution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805508069
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.URISupport;
+import org.jboss.logging.Logger;
+
+import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
+
+/**
+ * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
+ * they are the implementation of the configuration.
+ */
+@ApplicationScoped
+public class IngestRoutes extends RouteBuilder {
+
+ private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
+
+ @Inject
+ IngestBuildTimeConfig buildTimeConfig;
+
+ @Inject
+ IngestRunTimeConfig runTimeConfig;
+
+ @Inject
+ IngestBuilderPipelines builderPipelines;
+
+ @Inject
+ @Any
+ Instance<EmbeddingStore<TextSegment>> storeCandidates;
+
+ @Inject
+ @Any
+ Instance<EmbeddingModel> modelCandidates;
+
+ @Override
+ 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
+ // SmallRye only materialises a map key for the mapping whose structure a property matches
+ Set<String> builderDeclared = builderPipelines.entries().stream()
+ .map(IngestBuilderPipelines.Entry::name)
+ .collect(Collectors.toSet());
+ Set<String> names = new TreeSet<>(buildTimeConfig.pipelines().keySet());
+ names.addAll(runTimeConfig.pipelines().keySet());
+ names.removeAll(builderDeclared);
+
+ for (String name : names) {
+ IngestBuildTimeConfig.PipelineBuildTimeConfig pipeline = buildTimeConfig.pipelines().get(name);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = runTimeConfig.pipelines().get(name);
+
+ if (runtime != null && !runtime.enabled()) {
+ LOG.infof("Ingestion pipeline '%s' is disabled", name);
+ continue;
+ }
+
+ IngestService service = new IngestService(
+ name,
+ 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(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
+
+ // a consumer URI says "consume from this"; its absence says "read that directory"
+ String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
+ if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' sets both source.uri ('" + uri
+ + "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ + "reads one source: keep the URI, or drop it to read the directory.");
+ }
+ if (uri == null) {
+ configureFileSource(name, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=file", name);
+ } else {
+ configureEndpointSource(name, uri, runtime, service);
+ LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
+ }
+ }
+
+ for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
+ configureBuilderPipeline(entry);
+ }
+ }
+
+ /** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
+ private void configureBuilderPipeline(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;
+ }
+ // 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
+ // (source.recursive cannot be told apart from its default, so it alone goes undetected -
+ // Source.recursive() is its builder twin)
+ if (external != null && (external.source().directory().isPresent()
+ || external.source().documentId().isPresent())) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' is declared in Java, so its source "
+ + "comes from the @Ingest method. Remove quarkus.camel.ai.ingest." + name + ".source.* , or "
+ + "declare the pipeline in configuration instead.");
+ }
+
+ IngestPipeline definition = builderPipelines.definition(entry);
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
+
+ IngestService service = new IngestService(
+ name,
+ resolveStore(name, definition.embeddingStoreName().orElse(null)),
+ resolveModel(name, definition.embeddingModelName().orElse(null)),
+ definition.maxSegmentSize(),
+ definition.maxOverlapSize());
+
+ switch (definition.sourceType()) {
+ case "file" -> configureFileSource(name, runtime, service);
+ case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
+ default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
+ }
+
+ LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
+ URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
+ }
+
+ private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ IngestService service) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
+ // space would otherwise mis-parse, and a crafted one could inject options - delete=true
+ // is honoured ahead of noop and would delete the user's documents after reading them.
+ // noop leaves the documents where they are (a knowledge base reads its source, it does
+ // not consume it), idempotent keeps the same file from being ingested twice, and the
+ // changed read lock waits for a file still being copied in rather than embedding half of
+ // it - the truncation would be permanent, since idempotent keys on the path. The register
+ // is sized explicitly: Camel's default caps at 1000 entries, and beyond that eviction
+ // would re-ingest a large directory steadily during normal operation, not just on restart
+ Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
+ from(file(directory)
+ .noop(true)
+ .idempotent(true)
+ .idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .recursive(runtime.source().recursive())
+ .readLock("changed")
+ .charset(StandardCharsets.UTF_8.name()))
+ .routeId(routeId(name))
+ .process(exchange -> service.ingest(documentId.evaluate(exchange, String.class),
+ exchange.getIn().getBody(String.class)));
+ }
+
+ /**
+ * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
+ * identifies the document is the consumer's business, so {@code source.document-id} says it —
+ * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ */
+ private void configureEndpointSource(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
+ Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
+ from(uri)
+ .routeId(routeId(name))
+ .process(exchange -> {
+ String id = documentId.evaluate(exchange, String.class);
+ if (id == null) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
+ + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
+ + "quarkus.camel.ai.ingest." + name + ".source.document-id at where the "
+ + "consumer puts it.");
+ }
+ exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
+ });
+ }
+
+ private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ String defaultHeader) {
+ String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (configured == null) {
+ return ExpressionBuilder.headerExpression(defaultHeader);
+ }
+ // a bare header name is read as a header directly rather than parsed: a dotted name such
+ // as a dotted header name sends the simple parser into OGNL, and ${...} in a properties file is
+ // MicroProfile Config expansion, which would consume an expression before Camel saw it
+ return configured.contains("${") ? simple(configured) : ExpressionBuilder.headerExpression(configured);
+ }
+
+ private static String routeId(String name) {
+ return "camel-quarkus-ai-ingest-" + name;
+ }
+
+ private static String required(String name, String value, String property) {
+ if (value == null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name + "' has no " + property
+ + ". Set quarkus.camel.ai.ingest." + name + "." + property);
+ }
+ return value;
+ }
+
+ private EmbeddingStore<TextSegment> resolveStore(String name, String configured) {
Review Comment:
Reopening this one — the change works for the *named* path but breaks the unnamed one.
`getRegistry().findByType(EmbeddingStore.class)` reaches `RuntimeBeanRepository.findByType`, which is `Arc.container().listAll(type)`. That is a raw required type, and CDI treats a parameterized bean type as assignable to a raw one only when its type parameters are unbounded variables or `Object` — so a bean typed `EmbeddingStore<TextSegment>` never matches. On this branch:
```
registry.findByType(EmbeddingStore) = 0
registry.findByTypeWithName(EmbeddingStore) = []
Arc.listAll(EmbeddingStore) = 0
Arc.listAll(EmbeddingModel) = 1 <- not generic, so it does match
```
with `@Named("store")` set and `quarkus.arc.remove-unused-beans=false`, so neither naming nor bean removal is the variable.
Separately, the `@Inject @Any Instance<...>` fields that went away were the injection points keeping unnamed store and model beans unremovable. Even for the non-generic `EmbeddingModel`, an unnamed producer is now removed as unused; it resolves again with `quarkus.arc.remove-unused-beans=false`.
Net effect: a configuration-declared pipeline with exactly one unnamed store and model — the arrangement `usage.adoc` documents — starts at `dd193707` and fails at `2541fc3a` with `needs an embedding store, but no bean of that type exists`.
Going the other way would fix both at once: keep the `Instance` fields and resolve the named case through them, `candidates.select(NamedLiteral.of(configured))`. If the registry is the mechanism you would rather keep, it needs an `UnremovableBeanBuildItem` for both types and a `TypeLiteral<EmbeddingStore<TextSegment>>` lookup for the store.
Worth a test either way — `IngestNoStoreBeanTest` asserts the "no bean of that type" message, so it currently passes whether or not resolution works.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+import io.quarkus.runtime.annotations.ConfigDocMapKey;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithParentName;
+
+/**
+ * Runtime configuration of ingestion pipelines: concrete locations and switches that may differ
+ * per deployment. The pipeline topology is build-time, see {@link IngestBuildTimeConfig}.
+ */
+@ConfigMapping(prefix = "quarkus.camel.ai.ingest")
+@ConfigRoot(phase = ConfigPhase.RUN_TIME)
+public interface IngestRunTimeConfig {
+
+ /**
+ * Ingestion pipelines by name.
+ */
+ @WithParentName
+ @ConfigDocMapKey("pipeline-name")
+ Map<String, PipelineRunTimeConfig> pipelines();
+
+ interface PipelineRunTimeConfig {
+
+ /**
+ * Whether this pipeline starts. Useful to switch ingestion off in dev mode.
+ */
+ @WithDefault("true")
+ boolean enabled();
+
+ /**
+ * The document source.
+ */
+ SourceRunTimeConfig source();
+
+ interface SourceRunTimeConfig {
+
+ /**
+ * The directory to ingest documents from, for a pipeline that has no `source.uri`. A
+ * path is a deployment concern, so unlike the URI it stays runtime configuration.
+ * Setting both is an error.
+ */
+ Optional<String> directory();
+
+ /**
+ * Whether subdirectories are ingested too, when reading a directory.
+ */
+ @WithDefault("true")
+ boolean recursive();
+
+ /**
+ * Where the document id lives in the exchange the consumer delivers: normally the
+ * name of a header, such as `CamelAwsS3Key` for an S3 consumer or `CamelKafkaKey` for
+ * a Kafka one. A value containing a dollar-brace placeholder is taken as a
Review Comment:
`$simple{...}` is the right call, and the end-to-end coverage on the `custom` pipeline is good.
One narrowing though: the check is `configured.startsWith("$simple{")`, while Camel's own `LanguageSupport.hasSimpleFunction` uses `contains` for both tokens (`SIMPLE_FUNCTION_START = {"${", "$simple{"}`). So `id-$simple{header.Foo}` matches neither branch here and is silently taken as a header name, even though Camel would happily parse it. The `${` branch already uses `contains`, so the two halves disagree with each other too — `contains("$simple{")` lines both up with Camel.
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
Review Comment:
Thanks — the check and the test do what I was after.
One small over-reach: it rejects `final` on pseudo-scoped beans too. `@Singleton` and `@Dependent` get no client proxy, so a final method there is harmless. Narrowing the check to normal-scoped declaring beans would keep the message for the case that actually breaks. Not a blocker, and erring strict is defensible — just noting the false positive.
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.deployment;
+
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.Consume;
+import io.quarkus.deployment.annotations.ExecutionTime;
+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.ReflectiveClassBuildItem;
+import io.quarkus.runtime.configuration.ConfigurationException;
+import jakarta.inject.Singleton;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuildTimeConfig;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestBuilderPipelines;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestRoutes;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Langchain4jIngestRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeTaskBuildItem;
+import org.apache.camel.quarkus.core.deployment.spi.CamelServiceBuildItem;
+import org.apache.camel.quarkus.core.deployment.util.CamelSupport;
+import org.apache.camel.quarkus.core.deployment.util.PathFilter;
+import org.apache.camel.util.URISupport;
+import org.jboss.jandex.AnnotationInstance;
+import org.jboss.jandex.AnnotationTarget;
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.MethodInfo;
+
+class Langchain4jIngestProcessor {
+
+ private static final String FEATURE = "camel-langchain4j-ingest";
+
+ @BuildStep
+ FeatureBuildItem feature() {
+ return new FeatureBuildItem(FEATURE);
+ }
+
+ @BuildStep
+ AdditionalBeanBuildItem beans() {
+ return AdditionalBeanBuildItem.builder()
+ .addBeanClasses(IngestRoutes.class)
+ .setUnremovable()
+ .build();
+ }
+
+ /**
+ * Discovers {@code @Ingest} builder methods: validated here (return type, no parameters,
+ * unique names, no collision with configuration-declared pipelines), invoked reflectively once
+ * at startup.
+ */
+ @BuildStep
+ @Record(ExecutionTime.STATIC_INIT)
+ void discoverBuilderPipelines(
+ CombinedIndexBuildItem combinedIndex,
+ IngestBuildTimeConfig config,
+ Langchain4jIngestRecorder recorder,
+ BuildProducer<AdditionalBeanBuildItem> beans,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+ BuildProducer<SyntheticBeanBuildItem> syntheticBeans) {
+
+ DotName ingestAnnotation = DotName.createSimple(Ingest.class.getName());
+ DotName pipelineType = DotName.createSimple(IngestPipeline.class.getName());
+
+ List<String> flatEntries = new ArrayList<>();
+ Set<String> names = new HashSet<>();
+ Set<String> beanClasses = new HashSet<>();
+
+ for (AnnotationInstance annotation : combinedIndex.getIndex().getAnnotations(ingestAnnotation)) {
+ if (annotation.target().kind() != AnnotationTarget.Kind.METHOD) {
+ continue;
+ }
+ MethodInfo method = annotation.target().asMethod();
+ String name = annotation.value().asString();
+ String location = method.declaringClass().name() + "#" + method.name();
+
+ if (name.isBlank()) {
+ throw new ConfigurationException("@Ingest on " + location + " has a blank pipeline name");
+ }
+ if (!method.returnType().name().equals(pipelineType)) {
+ throw new ConfigurationException("@Ingest method " + location + " must return "
+ + IngestPipeline.class.getSimpleName());
+ }
+ if (!method.parameters().isEmpty()) {
+ throw new ConfigurationException("@Ingest method " + location + " must take no parameters");
+ }
+ // the method is invoked on a CDI bean instance, which a static method would bypass
+ // and a private one would run against the client proxy, seeing null injected fields
+ if (Modifier.isPrivate(method.flags()) || Modifier.isStatic(method.flags())) {
+ throw new ConfigurationException("@Ingest method " + location + " must not be private or static");
+ }
+ if (!names.add(name) || config.pipelines().containsKey(name)) {
+ throw new ConfigurationException("Ingestion pipeline '" + name + "' is declared more than once "
+ + "(builder and/or configuration). Pipeline names must be unique.");
+ }
+
+ flatEntries.add(name);
+ flatEntries.add(method.declaringClass().name().toString());
+ flatEntries.add(method.name());
+ beanClasses.add(method.declaringClass().name().toString());
+ }
+
+ if (!beanClasses.isEmpty()) {
+ beans.produce(AdditionalBeanBuildItem.builder()
+ .addBeanClasses(beanClasses.toArray(new String[0]))
+ .setUnremovable()
+ .build());
+ reflectiveClasses.produce(ReflectiveClassBuildItem.builder(beanClasses.toArray(new String[0]))
+ .methods()
+ .build());
+ }
+
+ syntheticBeans.produce(SyntheticBeanBuildItem.configure(IngestBuilderPipelines.class)
+ .scope(Singleton.class)
+ .unremovable()
+ .runtimeValue(recorder.createBuilderPipelines(flatEntries))
+ .done());
+ }
+
+ /**
+ * The pre-start half of the component-presence check: recorded as a Camel runtime task, it
+ * runs after ArC is fully initialised but before the Camel runtime is assembled — and thus
+ * before Camel Main binds {@code camel.component.*} properties, whose failure for a missing
+ * component would otherwise preempt the friendlier add-extension hint.
+ */
+ @BuildStep
+ @Record(ExecutionTime.RUNTIME_INIT)
+ @Consume(SyntheticBeansRuntimeInitBuildItem.class)
+ CamelRuntimeTaskBuildItem checkComponentsPresent(Langchain4jIngestRecorder recorder,
+ CamelContextBuildItem camelContext) {
+ recorder.checkComponentsPresent(camelContext.getCamelContext());
+ return new CamelRuntimeTaskBuildItem("langchain4j-ingest-components");
+ }
+
+ /**
+ * A pipeline whose consumer URI names a component that is not on the classpath stops the
+ * build, with the command that fixes it rather than a startup failure. Only configured URIs
+ * can be checked: a builder-declared pipeline composes its URI at startup, where the pre-start
+ * task above applies the same hint. Component services are REGISTRY-destination, so they are
+ * read from the application archives directly — they never appear among the DISCOVERY
+ * {@code CamelServiceBuildItem}s.
+ */
+ @BuildStep
+ void validateConnectorsPresent(IngestBuildTimeConfig config, ApplicationArchivesBuildItem applicationArchives,
+ BuildProducer<ValidationErrorBuildItem> validationErrors) {
+ PathFilter pathFilter = new PathFilter.Builder()
+ .include("META-INF/services/org/apache/camel/component/*")
+ .build();
+ Set<String> components = CamelSupport.services(applicationArchives, pathFilter)
+ .map(CamelServiceBuildItem::getName)
+ .collect(Collectors.toSet());
+
+ for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig> entry : config.pipelines().entrySet()) {
+ String uri = entry.getValue().source().uri().orElse(null);
+ if (uri == null) {
+ continue;
+ }
+ int colon = uri.indexOf(':');
+ String scheme = colon < 1 ? uri : uri.substring(0, colon);
+ // a placeholder resolves at startup, so its scheme cannot be known here
+ if (scheme.contains("{{") || scheme.contains("$")) {
+ continue;
+ }
+ if (!components.contains(scheme)) {
+ // the URI is sanitized: a consumer URI may legitimately carry credentials, and a
+ // build log is no place for them. The artifact hint is a heuristic - multi-scheme
+ // components (smtp -> camel-quarkus-mail) name their extension differently
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' consumes from '"
+ + URISupport.sanitizeUri(uri) + "', but the Camel component '" + scheme
+ + "' is not on the classpath.\nAdd the extension that provides it, usually:"
+ + " ./mvnw quarkus:add-extension -Dextensions=camel-quarkus-" + scheme)));
Review Comment:
The message reads well now, in both checks.
The class javadoc on `IngestComponentPresence` still describes the old form though — "Fails a pipeline whose consumer component is not on the classpath with the same add-extension hint the build gives for configured URIs".
##########
integration-tests/langchain4j-ingest/pom.xml:
##########
@@ -0,0 +1,183 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-build-parent-it</artifactId>
+ <version>3.39.0-SNAPSHOT</version>
+ <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>camel-quarkus-integration-test-langchain4j-ingest</artifactId>
+ <name>Camel Quarkus :: Integration Tests :: LangChain4j Ingest</name>
+ <description>Integration tests for the declarative AI document ingestion extension</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>dev.langchain4j</groupId>
+ <artifactId>langchain4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-rest-jackson</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka</artifactId>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.rest-assured</groupId>
+ <artifactId>rest-assured</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-integration-tests-support-kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>native</id>
+ <activation>
+ <property>
+ <name>native</name>
+ </property>
+ </activation>
+ <properties>
+ <quarkus.native.enabled>true</quarkus.native.enabled>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>virtualDependencies</id>
+ <activation>
+ <property>
+ <name>!noVirtualDependencies</name>
+ </property>
+ </activation>
+ <dependencies>
+ <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-aws2-s3-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-kafka-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-langchain4j-ingest-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ </profile>
+ </profiles>
Review Comment:
Confirmed locally: `mvn verify -Dskip-testcontainers-tests` on this module runs the four non-container tests, starts no container, and passes.
The surefire-only exclusion is fine as things stand — failsafe is bound only in the `native` profile, and no workflow combines `-Dnative` with `-Dskip-testcontainers-tests`, so the `*IT` classes never run on those jobs. Worth keeping in mind if one ever does, since the 74 modules using `<skipTests>true</skipTests>` would be covered there and this one would not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805339727
##########
integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingSearchRequest;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestHeaders;
+import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+@jakarta.ws.rs.Path("/langchain4j-ingest")
+public class IngestResource {
+
+ @Inject
+ @Named("products-store")
+ EmbeddingStore<TextSegment> productsStore;
+
+ @Inject
+ @Named("test-model")
+ EmbeddingModel model;
+
+ @Inject
+ @Named("custom-store")
+ EmbeddingStore<TextSegment> customStore;
+
+ @Inject
+ @Named("built-store")
+ EmbeddingStore<TextSegment> builtStore;
+
+ @Inject
+ @Named("s3-store")
+ EmbeddingStore<TextSegment> s3Store;
+
+ @Inject
+ @Named("events-store")
+ EmbeddingStore<TextSegment> eventsStore;
+
+ @Inject
+ ProducerTemplate producerTemplate;
+
+ @ConfigProperty(name = "ingest.test.directory")
+ String directory;
+
+ /** Writes a document into the watched directory — app-side, so native mode shares the path. */
+ @POST
+ @jakarta.ws.rs.Path("/file/{name}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ public void writeFile(@PathParam("name") String name, String content) throws Exception {
+ Path dir = Path.of(directory);
+ Files.createDirectories(dir);
+ Files.writeString(dir.resolve(name), content);
+ }
+
+ @GET
+ @jakarta.ws.rs.Path("/search")
+ @Produces(MediaType.APPLICATION_JSON)
+ public List<String> search(@QueryParam("q") String query, @QueryParam("store") String storeName) {
+ EmbeddingStore<TextSegment> store = switch (storeName == null ? "products" : storeName) {
+ case "custom" -> customStore;
+ case "built" -> builtStore;
+ case "s3" -> s3Store;
+ case "events" -> eventsStore;
+ default -> productsStore;
+ };
+ var result = store.search(EmbeddingSearchRequest.builder()
+ .queryEmbedding(model.embed(query).content())
+ .maxResults(20)
+ .minScore(0.0)
Review Comment:
Done — search now returns the whole store, with a comment noting the test model has no ranking semantics, and the test name no longer claims retrievability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#discussion_r3805342862
##########
tooling/scripts/test-categories.yaml:
##########
@@ -143,6 +143,7 @@ group-07:
- ldap
- langchain4j-embeddings
- langchain4j-rag-bridge-ql4j
+ - langchain4j-ingest
Review Comment:
Done — order fixed; native wall time is not measurable locally, so I will watch CI and move the module to a lighter group if it drags
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #9015. Add the langchain4j-ingest extension: declarative doc ingestion [camel-quarkus]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on PR #9018:
URL: https://github.com/apache/camel-quarkus/pull/9018#issuecomment-5339611261
The issue is fixed.
Regarding the scope question about @Ingest, IngestPipeline, Source and IngestService - only the ingestSergvice is "portable" into Camel (but brings a quite small benefit from doing it), the rest are working only for Camel-quarkus (they configures camel routes under the hood, without needing a knowledge of Camel)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org