jamesnetherton commented on code in PR #9340:
URL: https://github.com/apache/camel-quarkus/pull/9340#discussion_r4386073912
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -241,17 +262,29 @@ private void compositionRoute(String name, String uri,
sink.put("maxSegmentSize", String.valueOf(maxSegmentSize));
sink.put("maxOverlapSize", String.valueOf(maxOverlapSize));
sink.put("embeddingBatchSize", String.valueOf(embeddingBatchSize));
- sink.put("modality", "text");
- sink.put("contentType", "");
+ // lower-cased, as the Kamelet declares it, rather than relying on a
case-insensitive enum conversion
+ sink.put("modality", modality.toLowerCase(Locale.ROOT));
+ sink.put("contentType", contentType == null ? "" : contentType);
// 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("includeId", "");
- sink.put("excludeId", "");
- sink.put("documentFilter", "#bean:" + bindInstance(name,
"document-filter", PredicateBuilder.constant(true)));
+ // filter.* comes from configuration for both declaration styles - a
builder pipeline's view
+ // delegates to it, the builder having no filter API - and the
component enforces it
+ IngestRunTimeConfig.PipelineRunTimeConfig.FilterRunTimeConfig filter =
runtime == null
+ ? IngestPipeline.NO_FILTER
+ : runtime.filter();
+ String includeId = filter.includeId().orElse("");
+ String excludeId = filter.excludeId().orElse("");
+ sink.put("includeId", includeId);
+ sink.put("excludeId", excludeId);
+ if (filter.minDocumentSize() != 0) {
+ sink.put("minDocumentSize",
String.valueOf(filter.minDocumentSize()));
+ }
+ sink.put("documentFilter", "#bean:" + filter.documentFilter()
Review Comment:
A misspelled `filter.document-filter` bean fails inside Kamelet property
binding, with an error that names the sink route rather than the pipeline or
the config key. `repositoryRef()` already checks the bean up front with a clear
"references ... but no such bean exists" message. Could this get the same check?
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -283,14 +316,18 @@ private void compositionRoute(String name, String uri,
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);
+ route = parseSteps(route, name, parser, maxDocumentSize, true,
null, includeId, excludeId);
// 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
+ // image-only document, and the file's register key is committed.
A filtered one is
+ // logged at DEBUG
route.to(kameletUri(name, SINK_KAMELET, "sink",
sink)).process(exchange -> {
IngestResult result =
exchange.getMessage().getBody(IngestResult.class);
+ if (result != null && result.outcome() ==
IngestResult.Outcome.FILTERED) {
Review Comment:
On a directory pipeline a filtered file's exchange succeeds, so its key is
committed in the source register. With a persistent
`source.idempotent-repository`, a file filtered by `filter.exclude-id` stays
skipped after the filter is removed and the app restarted, unless the file is
touched. `filter.*` is runtime config, so changing it is expected.
That is the same problem given in #9290 as the reason not to commit an
untypable media file ("a committed key would keep the file skipped even after a
config fix"). Should a filtered file roll back its key instead, or should
usage.adoc at least warn about it alongside the persistent register docs?
##########
extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java:
##########
@@ -310,6 +310,36 @@ void validatePipelines(IngestBuildTimeConfig config,
BuildProducer<ValidationErr
"Ingestion pipeline '" + entry.getKey() + "':
max-document-size must not be negative, 0 "
+ "meaning no limit (got " +
pipeline.maxDocumentSize() + ")")));
}
+
+ // the component and IngestRoutes check these at startup; a
configured pipeline fails the build instead
+ boolean media = "media".equalsIgnoreCase(pipeline.modality());
Review Comment:
Minor: the `max-segment-size`/`max-overlap-size`/`embedding-batch-size`
checks above still run for `modality=media`, although the new Javadoc says they
don't apply there. A media pipeline with settings copied from a text one (e.g.
overlap 600, default segment 500) fails the build over options that are
ignored. Skip them when `media` is true?
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -269,9 +302,9 @@ private void compositionRoute(String name, String uri,
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
+ if (parser == null && !media) {
Review Comment:
With `!media` added here, a media pipeline now leaves `charset` out, so an
application-wide `camel.kamelet.langchain4j-ingest-file-source.charset` fills
it in. That goes against the "options at their defaults are passed all the
same" rule at line 253. If I read `GenericFileConverter` right, a charset on
the exchange makes the byte conversion go through a reader, which would corrupt
WAV/PNG bytes with no error. The same applies to parser pipelines, though that
was there before. Can it be pinned to a neutral value, or covered by the IT
poison set like `documentFilter`?
##########
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java:
##########
@@ -241,17 +262,29 @@ private void compositionRoute(String name, String uri,
sink.put("maxSegmentSize", String.valueOf(maxSegmentSize));
sink.put("maxOverlapSize", String.valueOf(maxOverlapSize));
sink.put("embeddingBatchSize", String.valueOf(embeddingBatchSize));
- sink.put("modality", "text");
- sink.put("contentType", "");
+ // lower-cased, as the Kamelet declares it, rather than relying on a
case-insensitive enum conversion
+ sink.put("modality", modality.toLowerCase(Locale.ROOT));
+ sink.put("contentType", contentType == null ? "" : contentType);
// 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("includeId", "");
- sink.put("excludeId", "");
- sink.put("documentFilter", "#bean:" + bindInstance(name,
"document-filter", PredicateBuilder.constant(true)));
+ // filter.* comes from configuration for both declaration styles - a
builder pipeline's view
+ // delegates to it, the builder having no filter API - and the
component enforces it
+ IngestRunTimeConfig.PipelineRunTimeConfig.FilterRunTimeConfig filter =
runtime == null
+ ? IngestPipeline.NO_FILTER
+ : runtime.filter();
+ String includeId = filter.includeId().orElse("");
+ String excludeId = filter.excludeId().orElse("");
+ sink.put("includeId", includeId);
+ sink.put("excludeId", excludeId);
+ if (filter.minDocumentSize() != 0) {
Review Comment:
Nit: `minDocumentSize` is put as `"0"` at line 270 and then conditionally
overwritten here. A single `sink.put("minDocumentSize",
String.valueOf(filter.minDocumentSize()))` would do.
--
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]
