JiriOndrusek commented on code in PR #9143: URL: https://github.com/apache/camel-quarkus/pull/9143#discussion_r4003908169
########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java: ########## @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.langchain4j.ingest; + +import java.io.InputStream; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Component; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.langchain4j.ingest.IngestResult; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders; +import org.apache.camel.component.langchain4j.ingest.TikaTextDecode; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.spi.IdempotentRepository; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; + +/** + * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse, + * then split, embed and store through the {@code langchain4j-ingest} producer. + * + * <p> + * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged + * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied + * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is + * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies + * document metadata over the headers, so a crafted document could otherwise forge its own identity. + * + * <p> + * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel + * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the + * community adopts one of them. + * + * <p> + * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public + * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction. + */ +abstract class IngestPipelineRouteBuilder extends RouteBuilder { + + /** + * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large + * directories during normal operation; in-memory, so lost on restart. + */ + static final int DEFAULT_REGISTER_CAPACITY = 100_000; + + /** + * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a + * configuration model, say. + */ + protected abstract List<IngestPipelineDefinition> pipelines(); + + @Override + public void configure() { + Set<String> names = new HashSet<>(); + for (IngestPipelineDefinition pipeline : pipelines()) { + if (!names.add(pipeline.name())) { + throw new IllegalArgumentException( + "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline."); + } + configurePipeline(pipeline); + } + } + + private void configurePipeline(IngestPipelineDefinition pipeline) { + requireParserComponent(pipeline); + String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store"); + String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model"); + String splitterRef = pipeline.documentSplitterRef(); + + if (pipeline.directory() != null) { + directoryRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory()); + } else { + consumerRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri())); + } + } + + // ******************************************************************************** + // The two route topologies + // ******************************************************************************** + + /** + * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file + * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments + * remain. + */ + private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String registerRef = repositoryRef(pipeline, true); + Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME); + + ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef)) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // the file consumer discards the reply, so no repository is passed to the producer: the + // endpoint register already keeps the same file version from being ingested twice + ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null)); + if (pipeline.parser() != null) { + // the discarded reply would otherwise hide it: a parse to nothing typically means a + // missing Tika parser module or an image-only document, and the file's register key + // is committed, so it is not retried until the file changes + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is" + + " committed, so it is not retried until the file changes (missing parser module?" + + " image-only document?)", + pipeline.name(), result.documentId()); + } + }); + } else { + // the discarded reply would otherwise hide even the trace of a blank file + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written", + pipeline.name(), result.documentId()); + } + }); + } + } + + /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */ + private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String scheme = StringHelper.before(pipeline.uri(), ":"); + requireComponent(scheme, pipeline, "its source"); + String registerRef = repositoryRef(pipeline, false); + Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID); + + ProcessorDefinition<?> route = from(pipeline.uri()) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); Review Comment: Fixed in the amended commit: a parser pipeline now fails the exchange before the parse when the captured id resolved to nothing — same behaviour as the previous engine — with a test (`consumerDeliveryWithoutIdFailsBeforeTheParse`). Worth noting camel-tika filters `Camel*` metadata names since CAMEL-24423, so the `CamelIngestDocumentId` forgery specifically is blocked either way — but a custom non-Camel header name was genuinely forgeable, and the null-id delivery silently ingesting was a regression. Both closed by the pre-parse guard. _Claude Code on behalf of @JiriOndrusek_ ########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java: ########## @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.langchain4j.ingest; + +import java.io.InputStream; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Component; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.langchain4j.ingest.IngestResult; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders; +import org.apache.camel.component.langchain4j.ingest.TikaTextDecode; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.spi.IdempotentRepository; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; + +/** + * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse, + * then split, embed and store through the {@code langchain4j-ingest} producer. + * + * <p> + * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged + * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied + * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is + * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies + * document metadata over the headers, so a crafted document could otherwise forge its own identity. + * + * <p> + * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel + * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the + * community adopts one of them. + * + * <p> + * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public + * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction. + */ +abstract class IngestPipelineRouteBuilder extends RouteBuilder { + + /** + * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large + * directories during normal operation; in-memory, so lost on restart. + */ + static final int DEFAULT_REGISTER_CAPACITY = 100_000; + + /** + * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a + * configuration model, say. + */ + protected abstract List<IngestPipelineDefinition> pipelines(); + + @Override + public void configure() { + Set<String> names = new HashSet<>(); + for (IngestPipelineDefinition pipeline : pipelines()) { + if (!names.add(pipeline.name())) { + throw new IllegalArgumentException( + "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline."); + } + configurePipeline(pipeline); + } + } + + private void configurePipeline(IngestPipelineDefinition pipeline) { + requireParserComponent(pipeline); + String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store"); + String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model"); + String splitterRef = pipeline.documentSplitterRef(); + + if (pipeline.directory() != null) { + directoryRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory()); + } else { + consumerRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri())); + } + } + + // ******************************************************************************** + // The two route topologies + // ******************************************************************************** + + /** + * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file + * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments + * remain. + */ + private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String registerRef = repositoryRef(pipeline, true); + Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME); + + ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef)) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // the file consumer discards the reply, so no repository is passed to the producer: the + // endpoint register already keeps the same file version from being ingested twice + ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null)); + if (pipeline.parser() != null) { + // the discarded reply would otherwise hide it: a parse to nothing typically means a + // missing Tika parser module or an image-only document, and the file's register key + // is committed, so it is not retried until the file changes + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is" + + " committed, so it is not retried until the file changes (missing parser module?" + + " image-only document?)", + pipeline.name(), result.documentId()); + } + }); + } else { + // the discarded reply would otherwise hide even the trace of a blank file + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written", + pipeline.name(), result.documentId()); + } + }); + } + } + + /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */ + private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String scheme = StringHelper.before(pipeline.uri(), ":"); + requireComponent(scheme, pipeline, "its source"); + String registerRef = repositoryRef(pipeline, false); + Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID); + + ProcessorDefinition<?> route = from(pipeline.uri()) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); Review Comment: Fixed in the amended commit: consumer pipelines with a register now answer a known duplicate `SKIPPED` before the parse via an advisory `repository.contains(id)` check — the producer's eager claim stays authoritative, so a duplicate racing the check is still caught there; the check only saves the parse cost, which covers the S3 re-listing scenario. _Claude Code on behalf of @JiriOndrusek_ -- 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]
