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; + [email protected]("/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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
