jiriOndrusek-agent commented on code in PR #9340:
URL: https://github.com/apache/camel-quarkus/pull/9340#discussion_r4387709043
##########
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:
True, reproduced with a `FileIdempotentRepository`: the excluded file's key
was persisted, and after removing the filter and restarting, the file was never
ingested.
Fixed in 92c3c12003. Releasing the key would re-read the file on every poll,
so a named register on a directory pipeline is now wrapped (`SourceRegister`),
which holds a filtered file's key in memory only. The file is skipped while the
app runs and checked again after a restart. Test:
`Langchain4jIngestFilterTest.filteredFileStaysOutOfTheNamedRegister`.
usage.adoc updated.
This also removes my #9290 objection to answering `filtered` for an
untypable media file.
_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:
##########
@@ -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:
Right, reproduced both ways: with `UTF-8`, WAV and PDF files fail on every
poll. With `ISO-8859-1` and `max-document-size` set, the WAV is stored under a
wrong vector and nothing is logged.
It can't be pinned: the file endpoint has no neutral charset (`""` fails its
validation). Fixed in 92c3c12003: parser and media directory pipelines drop the
charset right after the source, before anything reads the file. Text pipelines
keep `UTF-8` pinned.
The IT poison set now has
`camel.kamelet.langchain4j-ingest-file-source.charset=UTF-16`. The audio, Tika
and text tests pass with it.
_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:
##########
@@ -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:
Done in 92c3c12003, with the same message: `Ingestion pipeline 'docs'
references document filter 'no-such-filter' but no such bean exists`. Test:
`IngestMissingDocumentFilterTest`.
_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]
