jamesnetherton commented on code in PR #9339:
URL: https://github.com/apache/camel-quarkus/pull/9339#discussion_r4374915581
##########
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 as the migration guide comment: the `-source` route isn't visible in
JMX by default.
##########
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:
The pinning policy here doesn't cover everything the released 4.23.0
Kamelets expose. The sink now has `modality` and `contentType`, and both parser
actions have `maxDocumentSize`, none of which the extension sets. So an
application-wide `camel.kamelet.langchain4j-ingest-sink.modality=media` (or the
`CAMEL_KAMELET_*` env var), set for an app's own media route, switches every
extension pipeline into media mode.
Suggest pinning `modality=text` here and passing the pipeline's cap (or `0`)
into the parser `action` map.
`IngestConsumerDocumentIdTest.appWideSinkPropertiesDoNotReachThePipeline` only
covers the two older sink options, so it would be worth extending it.
##########
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:
Please drop the comment. The reasoning belongs in the commit message rather
than above the `<dependency>` entries.
##########
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:
This re-implements the pre-parse byte cap that `tika-extract-text-action`
and `docling-convert-action` now provide via their own `maxDocumentSize`
property, so the same invariant lives in two places with different measurement
(this one trusts `CamelFileLength` on directory pipelines and bounds the stream
read; the Kamelet materialises the stream as `byte[]`).
Either delegate to the Kamelet option and drop this guard, or keep it and
pin the Kamelet's `maxDocumentSize` to `0` so the two can never both be active.
If the bounded read is the reason to keep it, that's probably worth an upstream
camel-kamelets issue so the Java copy can eventually go.
##########
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._-]+")) {
Review Comment:
Minor: this regex is duplicated verbatim in
`Langchain4jIngestProcessor.discoverBuilderPipelines`, and the
`"langchain4j-ingest-" + name` prefix is now spelled out five times in this
class after the old `routeId()` helpers went. A single `NAME_PATTERN` constant
and an `id(name, suffix)` helper based on `LangChain4jIngest.SCHEME` would keep
build-time and startup validation from drifting apart.
##########
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:
Camel doesn't register Kamelet-created routes in JMX by default
(`DefaultManagementAgent.registerRoutesCreateByKamelet` is `false`), so an
operator following this from a JMX console only sees
`langchain4j-ingest-<name>`, stops it, and lands in exactly the 30 s timeout
situation this paragraph warns about. The same sentence is in `usage.adoc`.
Either mention the `registerRoutesCreateByKamelet` switch, or point at the
`RouteController` API instead of JMX.
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java:
##########
@@ -26,7 +26,7 @@
public final class IngestPipeline {
/** The values {@link #parser(String)} and the {@code parser}
configuration property accept. */
- public static final Set<String> SUPPORTED_PARSERS =
IngestPipelineDefinition.SUPPORTED_PARSERS;
+ public static final Set<String> SUPPORTED_PARSERS =
IngestRoutes.SUPPORTED_PARSERS;
Review Comment:
Minor: this now reads a constant off the `@ApplicationScoped` route builder,
so the public builder API and the deployment processor class-initialise
`IngestRoutes` (RouteBuilder, logger, Camel model) just to get a `Set<String>`.
Hoisting the `Parser` enum into its own package-private type, and deriving both
`SUPPORTED_PARSERS` and `KAMELETS` from it, would avoid 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]
