jiriOndrusek-agent commented on code in PR #9339:
URL: https://github.com/apache/camel-quarkus/pull/9339#discussion_r4375812032
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -158,44 +213,256 @@ private IngestPipelineDefinition
builderPipeline(IngestBuilderPipelines.Entry en
* One creation path for both declaration styles, mirroring how a builder
pipeline provides
* the same runtime-configuration view the configuration path reads.
*/
- private IngestPipelineDefinition definition(String name, String uri,
+ private void compositionRoute(String name, String uri,
IngestRunTimeConfig.PipelineRunTimeConfig runtime,
EmbeddingStore<TextSegment> store, EmbeddingModel model,
int maxSegmentSize, int maxOverlapSize, int embeddingBatchSize,
int maxDocumentSize,
String documentSplitterName, String parser) {
- IngestPipelineDefinition definition;
+ // the name is substituted into Kamelet URIs and registry references,
so both declaration
+ // styles are held to one charset; @Ingest names are already checked
at build time
+ if (!name.matches("[A-Za-z0-9._-]+")) {
+ throw new IllegalStateException(
+ "Ingestion pipeline name '" + name + "' may only contain
letters, digits, '.', '_' and '-'");
+ }
+ String storeRef = bindInstance(name, "store", store);
+ String modelRef = bindInstance(name, "model", model);
+ String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
+ String repositoryRef = repositoryRef(name, runtime, uri == null);
+
+ // options at their defaults are passed all the same, here and in the
other steps: an
+ // application-wide camel.kamelet.<kamelet>.* property would otherwise
fill them in too
+ Map<String, Object> sink = new LinkedHashMap<>();
+ sink.put("pipelineName", name);
+ sink.put("documentIdHeader", LangChain4jIngestHeaders.DOCUMENT_ID);
+ sink.put("maxSegmentSize", String.valueOf(maxSegmentSize));
+ sink.put("maxOverlapSize", String.valueOf(maxOverlapSize));
+ sink.put("embeddingBatchSize", String.valueOf(embeddingBatchSize));
+ // 0, no limit, is the endpoint's default too
+ sink.put("maxDocumentSize", String.valueOf(maxDocumentSize));
+ sink.put("minDocumentSize", "0");
Review Comment:
Agreed, and confirmed: with an app-wide `modality=media` the pipelines
failed to start (text-only model). Fixed in this PR (moved down from #9340):
the sink now always gets `modality=text`, `contentType=""`, `includeId=""`,
`excludeId=""` and an always-true `documentFilter` bean, and both parser
actions get `maxDocumentSize=0` (the extension's guard stays, see the next
thread). The component reads `""` as unset.
`appWideSinkPropertiesDoNotReachThePipeline` now sets all five sink options
app-wide, and the IT sets both actions' cap to 1 app-wide, so a missing pin
fails JVM and native. #9340 now only passes the configured values instead.
A few options have no neutral value and stay unset (`documentSplitter`,
`idempotentRepository`, Docling's serve options); usage.adoc keeps the warning
for those.
_Generated with [Claude Code](https://claude.com/claude-code) on behalf of
@JiriOndrusek_
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -158,44 +213,256 @@ private IngestPipelineDefinition
builderPipeline(IngestBuilderPipelines.Entry en
* One creation path for both declaration styles, mirroring how a builder
pipeline provides
* the same runtime-configuration view the configuration path reads.
*/
- private IngestPipelineDefinition definition(String name, String uri,
+ private void compositionRoute(String name, String uri,
IngestRunTimeConfig.PipelineRunTimeConfig runtime,
EmbeddingStore<TextSegment> store, EmbeddingModel model,
int maxSegmentSize, int maxOverlapSize, int embeddingBatchSize,
int maxDocumentSize,
String documentSplitterName, String parser) {
- IngestPipelineDefinition definition;
+ // the name is substituted into Kamelet URIs and registry references,
so both declaration
+ // styles are held to one charset; @Ingest names are already checked
at build time
+ if (!name.matches("[A-Za-z0-9._-]+")) {
+ throw new IllegalStateException(
+ "Ingestion pipeline name '" + name + "' may only contain
letters, digits, '.', '_' and '-'");
+ }
+ String storeRef = bindInstance(name, "store", store);
+ String modelRef = bindInstance(name, "model", model);
+ String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
+ String repositoryRef = repositoryRef(name, runtime, uri == null);
+
+ // options at their defaults are passed all the same, here and in the
other steps: an
+ // application-wide camel.kamelet.<kamelet>.* property would otherwise
fill them in too
+ Map<String, Object> sink = new LinkedHashMap<>();
+ sink.put("pipelineName", name);
+ sink.put("documentIdHeader", LangChain4jIngestHeaders.DOCUMENT_ID);
+ sink.put("maxSegmentSize", String.valueOf(maxSegmentSize));
+ sink.put("maxOverlapSize", String.valueOf(maxOverlapSize));
+ sink.put("embeddingBatchSize", String.valueOf(embeddingBatchSize));
+ // 0, no limit, is the endpoint's default too
+ sink.put("maxDocumentSize", String.valueOf(maxDocumentSize));
+ sink.put("minDocumentSize", "0");
+ if (documentSplitterName != null) {
+ sink.put("documentSplitter", "#bean:" + documentSplitterName);
+ }
+ sink.put("embeddingStore", "#bean:" + storeRef);
+ sink.put("embeddingModel", "#bean:" + modelRef);
+
if (uri == null) {
String directory = required(name, runtime == null ? null :
runtime.source().directory().orElse(null),
"source.directory");
- definition = IngestPipelineDefinition.directory(name, directory)
- .recursive(runtime.source().recursive());
+ // substituted into the file-source Kamelet's endpoint URI: a '?'
or '#' could
+ // inject consumer options - delete=true would consume the user's
documents
+ if (directory.contains("?") || directory.contains("#")) {
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "': the directory must not contain '?' or '#' (got
'" + directory + "')");
+ }
+ Map<String, Object> source = new LinkedHashMap<>();
+ source.put("directory", directory);
+ source.put("recursive",
String.valueOf(runtime.source().recursive()));
+ // the file consumer's own default poll delay
+ source.put("delay", "500");
+ if (parser == null) {
+ // text is read as UTF-8; a parser receives the raw bytes
instead - the format is
+ // its business, and a charset conversion would corrupt a
binary document
+ source.put("charset", "UTF-8");
+ }
+ source.put("idempotentRepository", "#bean:" + repositoryRef);
+
+ ProcessorDefinition<?> route = from(kameletUri(name,
FILE_SOURCE_KAMELET, "source", source))
+ .routeId("langchain4j-ingest-" + name);
+ if (documentId != null) {
+ // override the source's file-name default; captured before
any further step
+ route = route.setHeader(LangChain4jIngestHeaders.DOCUMENT_ID,
documentIdExpression(documentId));
+ }
+ // no duplicate pre-check: the source register already filtered
duplicates out
+ route = parseSteps(route, name, parser, maxDocumentSize, true,
null);
+ // the file consumer discards the reply and the source register
already keeps the
+ // same file version from being ingested twice, so no repository
goes to the sink.
+ // The discarded reply would hide an empty outcome, so it is
logged: warned with a
+ // parser, since a parse to nothing typically means a missing Tika
parser module or an
+ // image-only document, and the file's register key is committed
+ route.to(kameletUri(name, SINK_KAMELET, "sink",
sink)).process(exchange -> {
+ IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
+ if (result == null || result.outcome() !=
IngestResult.Outcome.EMPTY) {
+ return;
+ }
+ if (parser != null) {
+ LOG.warnf("Ingestion pipeline '%s': document '%s' parsed
to no text and was skipped; its key is"
+ + " committed, so it is not retried until the file
changes (missing parser module?"
+ + " image-only document?)", name,
result.documentId());
+ } else {
+ LOG.debugf("Ingestion pipeline '%s': document '%s'
contained no text, nothing was written",
+ name, result.documentId());
+ }
+ });
+ LOG.infof("Ingestion pipeline '%s': source=file:%s", name,
directory);
} else {
- definition = IngestPipelineDefinition.consumer(name, uri);
+ if (!uri.contains(":")) {
+ throw new IllegalStateException(
+ "Ingestion pipeline '" + name + "': '" + uri + "' is
not a consumer URI");
+ }
+ ProcessorDefinition<?> route =
from(uri).routeId("langchain4j-ingest-" + name);
+ if (documentId != null &&
!IngestHeaders.DOCUMENT_ID.equals(documentId)) {
+ // copied into the canonical header the actions read, as on a
directory pipeline:
+ // evaluated against the exchange the consumer delivered,
before any parse, and a
+ // plain header name is read directly, never substituted into
an expression
+ route = route.setHeader(LangChain4jIngestHeaders.DOCUMENT_ID,
documentIdExpression(documentId));
+ if (!isSimpleExpression(documentId)) {
+ // the sink's missing-id error names the header the route
reads, as before; the
+ // parser steps keep the canonical one
+ sink.put("documentIdHeader", documentId);
+ }
+ } else {
+ // unset, or the component's canonical header (the 3.40
rename, #9162): read with the
+ // previous fallbacks, its exchange property and the 3.39 name
+ route = route.process(documentIdFallback(name));
+ }
+ // captured into the exchange property before any parse, as by the
previous builder: the
+ // sink reads the property first, so the id resolved here wins
over any header
+ route = route.process(exchange ->
exchange.setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
+
exchange.getMessage().getHeader(LangChain4jIngestHeaders.DOCUMENT_ID)));
+ IdempotentRepository register = repositoryRef == null
+ ? null
+ :
getContext().getRegistry().lookupByNameAndType(repositoryRef,
IdempotentRepository.class);
+ route = parseSteps(route, name, parser, maxDocumentSize, false,
register);
+ if (repositoryRef != null) {
+ // deduplication by document id happens inside the sink's
producer: a duplicate
+ // is answered SKIPPED, a blank delivery releases its claim
+ sink.put("idempotentRepository", "#bean:" + repositoryRef);
+ }
+ route.to(kameletUri(name, SINK_KAMELET, "sink", sink));
+ LOG.infof("Ingestion pipeline '%s': source=%s", name,
URISupport.sanitizeUri(uri));
}
- definition.embeddingStore(store)
- .embeddingModel(model)
- .splitter(maxSegmentSize, maxOverlapSize)
- .embeddingBatchSize(embeddingBatchSize);
- if (maxDocumentSize > 0) {
- definition.maxDocumentSize(maxDocumentSize);
+ }
+
+ /**
+ * The optional parse stage: the id guard, the advisory duplicate check
and the raw-size
+ * guard, then the parser action Kamelet, which captures the document id
from the canonical
+ * header into the exchange property before the parse. The route is
returned unchanged when
+ * the pipeline has no parser.
+ */
+ private static ProcessorDefinition<?> parseSteps(ProcessorDefinition<?>
route, String name, String parser,
+ int maxDocumentSize, boolean directory, IdempotentRepository
register) {
+ if (parser == null) {
+ return route;
}
- if (documentSplitterName != null) {
- definition.documentSplitter(documentSplitterName);
+ String documentIdHeader = LangChain4jIngestHeaders.DOCUMENT_ID;
+ // a parser must never run without a captured id: the header the
action reads would be
+ // absent, and headers written by a parsed document could take its
place - the previous
+ // engine failed such a delivery, and so does this guard
+ route = route.process(exchange -> {
+ String id = exchange.getMessage().getHeader(documentIdHeader,
String.class);
+ if (id == null || id.isBlank()) {
+ throw new IllegalArgumentException("Ingestion pipeline '" +
name
+ + "': the document id resolved to nothing, and a
parser pipeline requires it before"
+ + " the parse - a parsed document must not supply its
own identity");
+ }
+ });
+ if (register != null) {
+ // advisory: a known duplicate is answered SKIPPED before the
(possibly remote) parse
+ // is paid for; the sink producer's eager claim stays
authoritative, so a duplicate
+ // racing this check is still caught there
+ route = route.choice()
+ .when(exchange ->
register.contains(exchange.getMessage().getHeader(documentIdHeader,
String.class)))
+ .process(exchange -> exchange.getMessage().setBody(new
IngestResult(name,
+ exchange.getMessage().getHeader(documentIdHeader,
String.class), 0,
+ IngestResult.Outcome.SKIPPED)))
+ .stop()
+ .end();
}
- if (parser != null) {
- definition.parser(parser);
+ if (maxDocumentSize > 0) {
+ // the endpoint's own cap counts extracted characters, which
protects the splitter and
+ // the model but not the parse: this guard rejects the raw payload
first, before tika
+ // or docling materialize it
+ route = route.process(rawSizeGuard(name, maxDocumentSize,
directory, documentIdHeader));
}
+ Map<String, Object> action = new LinkedHashMap<>();
+ action.put("documentIdHeader", documentIdHeader);
+ return route.to(kameletUri(name,
Parser.valueOf(parser.toUpperCase(Locale.ROOT)).actionKamelet, "parser",
action));
+ }
- String documentId = runtime == null ? null :
runtime.source().documentId().orElse(null);
- if (documentId != null) {
- definition.documentId(documentId);
- } else if (uri != null) {
- // the component's own default since the 3.40 rename (#9162); the
3.39 name is still
- // read as a fallback in the route builder's documentIdExpression
- definition.documentId(IngestHeaders.DOCUMENT_ID);
+ private static Processor rawSizeGuard(String name, int maxDocumentSize,
boolean trustDeclaredLength,
Review Comment:
Went with your second option: the guard stays and the actions'
`maxDocumentSize` is pinned to `0` (in this PR now, with the pinning above), so
the two are never both active.
Why keep it: on a directory pipeline it trusts the file consumer's
`CamelFileLength` and reads nothing, while the Kamelet's `${length()}` reads
each file to the end just to count it. On a consumer pipeline it reads at most
max + 1 bytes, while the Kamelet first turns an uncached stream into a full
`byte[]`. `length()` also counts in an `int`.
Agreed it belongs upstream: I'll open a camel-kamelets issue for a bounded
check (and `CamelFileLength` for files), so this copy can go later.
_Generated with [Claude Code](https://claude.com/claude-code) on behalf of
@JiriOndrusek_
##########
docs/modules/ROOT/pages/migration-guide/4.0.0.adoc:
##########
@@ -89,9 +89,16 @@ The declaration surface is unchanged — the same
`quarkus.camel.langchain4j.ing
Java code reading the request-reply outcome as `IngestResult` now imports
`org.apache.camel.component.langchain4j.ingest.IngestResult` — the extension's
`org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult` is
gone with the engine, so this surfaces at compile time.
The same applies to the segment metadata keys
`IngestService.METADATA_DOCUMENT_ID` and `IngestService.METADATA_PIPELINE`: use
`org.apache.camel.component.langchain4j.ingest.LangChain4jIngest.METADATA_DOCUMENT_ID`
and `METADATA_PIPELINE` instead; the values are unchanged.
-A pipeline's route id is now `langchain4j-ingest-<pipeline-name>` rather than
`camel-quarkus-langchain4j-ingest-<pipeline-name>`.
+A pipeline's own route id is now `langchain4j-ingest-<pipeline-name>` rather
than `camel-quarkus-langchain4j-ingest-<pipeline-name>`.
+The pipeline is composed from Kamelets now, and each one it uses adds a route:
`langchain4j-ingest-<pipeline-name>-source`, `-parser` or `-sink`.
Tooling that addresses the routes by id — JMX, the management endpoints,
`camel.main.*` route policies keyed on ids — must use the new names.
The prefix is the component's scheme name, so it follows the component.
+To pause a directory pipeline, stop its
`langchain4j-ingest-<pipeline-name>-source` route: with
`langchain4j-ingest-<pipeline-name>` alone stopped, the directory is still
polled and each new file fails after the Kamelet's 30 s timeout.
Review Comment:
Good catch, done both: the sentence now points at
`camelContext.getRouteController().stopRoute(...)`, and a new line says Camel
registers the Kamelet routes in JMX only with
`camel.main.jmxManagementRegisterRoutesCreateByKamelet=true`.
_Generated with [Claude Code](https://claude.com/claude-code) on behalf of
@JiriOndrusek_
##########
extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc:
##########
@@ -15,6 +15,10 @@ Each file is read as UTF-8 text — unless a `parser` turns
binary formats into
Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id`
metadata, so retrieval can cite which document an answer came from (up to
3.39.x the keys were `camel_quarkus_pipeline` and `camel_quarkus_document_id` —
see the migration guide).
Apart from `enabled` and the `source.*` settings shown here, properties are
fixed at build time.
A pipeline declared through runtime properties alone — nothing but a
`source.directory`, say — is invisible to build-time validation; its checks,
including the clash with an equally named `@Ingest` pipeline, report at startup
instead.
+Each pipeline is composed from the catalog's `langchain4j-ingest-file-source`,
parser action and `langchain4j-ingest-sink` Kamelets: its route
`langchain4j-ingest-<name>` plus one route per Kamelet it uses,
`langchain4j-ingest-<name>-source` (directory pipelines), `-parser` (with a
`parser`) and `-sink`.
+The pipelines always load these Kamelets from the classpath, whatever
`camel.component.kamelet.location` says; shipping an equally named Kamelet
replaces internal steps — including the identity capture — and is advanced
surgery.
+Application-wide `camel.kamelet.<kamelet>.*` properties and `CAMEL_KAMELET_*`
environment variables for these four Kamelets reach the pipelines too, for
every option the extension leaves unset, so do not set them.
+To pause a directory pipeline, stop its `langchain4j-ingest-<name>-source`
route, or both routes: with `langchain4j-ingest-<name>` alone stopped, the
directory is still polled and each new file fails after the Kamelet's 30 s
timeout.
Review Comment:
Same change here (the generated reference page follows).
_Generated with [Claude Code](https://claude.com/claude-code) on behalf of
@JiriOndrusek_
##########
extensions/langchain4j-ingest/runtime/pom.xml:
##########
@@ -45,6 +45,20 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-ingest</artifactId>
</dependency>
+ <!-- the pipeline topology lives in the langchain4j-ingest Kamelets:
the catalog carries
Review Comment:
Dropped. The commit message already names the new dependencies.
_Generated with [Claude Code](https://claude.com/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]
