LangChain4j Ingest

Point a pipeline at a folder and the files in it become a knowledge base. No route, no ingestion code:

Each file is read as UTF-8 text — there is no format parsing, so convert a PDF or DOCX before it reaches the pipeline — split into overlapping segments (max-segment-size, max-overlap-size), embedded in batches and written to the store; a document is held in memory whole while it is split. embedding-store and embedding-model name CDI beans and may be omitted when the application has exactly one of each. Every segment carries camel_quarkus_pipeline and camel_quarkus_document_id metadata, so retrieval can cite which document an answer came from. 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.

A directory pipeline registers what it ingested, keyed on the file’s path, modification time and size: unchanged files are skipped, edited files re-ingest (their previous segments remain, see the note above). The default register is in-memory (100,000 keys) and is there for the poll loop, not for restarts: the consumer re-scans the directory on every poll, so without a register a running application would re-embed the whole directory every few seconds. A restart loses it and re-ingests the directory once; a persistent repository makes restarts free. source.idempotent-repository names an IdempotentRepository bean to use instead:

The bean can be provided three ways.

Auto-created: an in-memory register (100,000 keys) is created and bound under the configured name — no bean definition needed, still lost on restart. If a bean with the name already exists, the existing bean wins:

Defined in properties through Camel’s camel.beans. syntax — here a file-backed register that survives restarts (camel-core, no extra dependency). Use a name without dashes; Camel normalises dashed camel.beans. keys to camelCase:

#class: beans are created reflectively. For native mode the extension registers the camel-core repositories shown here and every IdempotentRepository implementation found in the Jandex index — application classes always, component-provided ones through their jar’s index. A class from a jar without an index needs @RegisterForReflection(targets = …​) or a quarkus.index-dependency.* entry.

Defined as a CDI producer — needed when construction takes other beans. A JDBC register also deduplicates across instances: reliably for consumer-fed pipelines keyed on the document id, for directory pipelines only when all instances see identical paths and modification times:

More repository types — Caffeine, Infinispan, MongoDB, Cassandra, Hazelcast, Kafka and others, each provided by its component’s extension — along with the pattern’s details are covered in the Idempotent Consumer EIP guide.

Sizing. An in-memory register smaller than the directory evicts keys and re-ingests those files during normal operation — hence the 100,000 default. FileIdempotentRepository falls back to its file store on a cache miss, so cacheSize is a performance setting; the correctness limit is maxFileStoreSize: the default of about 1 MB holds roughly 10–17 thousand of these keys, beyond it the oldest 1000 are dropped with a warning. A JDBC register has no size limit. Every edit adds a new key and old keys are never removed, so a persistent register grows under churn until the synchronising engine arrives. There is no size option; size the bean. Use one register per pipeline — keys are source-derived, and pipelines sharing a register over the same directory skip each other’s work.

The register cannot be switched off: the consumer leaves files in place (noop), so without a register every poll would re-ingest the directory. To force a full re-ingest, restart (default register) or clear the persistent one. A pipeline that should consume its source — move or delete files after ingestion — uses source.uri (for example file:/inbox?delete=true&idempotent=false with source.document-id=CamelFileName), where all consumer options are available.

On a consumer-fed pipeline (source.uri or @Ingest), a configured register deduplicates deliveries by document id: a redelivered record or re-listed object ingests once, and a request-reply caller receives skipped instead of ingested. First write wins per id — an update with the same id is skipped, not replaced; streams that carry updates need a version-aware source.document-id. Only a delivery that wrote segments claims its id: a blank document answers empty and releases the claim, so an object created empty and populated later under the same id still ingests the content. With eager idempotency, a duplicate racing an in-flight first delivery is answered skipped even if that delivery then fails and releases the key.

Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation. Such a pipeline is declared in Java with @Ingest and the Camel Endpoint DSL:

Typing dsl. lists a factory for every component, each completing its own typed options. The method runs once at startup; it must return IngestPipeline, take no parameters and use a name no configured pipeline uses. enabled=false in configuration switches a Java-declared pipeline off. A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.

Ingestion needs a stable id per document, and where it lives is the consumer’s business: documentId names the header — the record key CamelKafkaKey above, CamelAwsS3Key for S3 — or gives a simple-language expression. Without it, the pipeline expects the CamelIngestDocumentId header and fails the exchange when it is absent. Mind each component’s own defaults, too: the aws2-s3 consumer deletes objects after reading them unless deleteAfterRead(false) is set — a knowledge base reads its source, it does not consume it.

After such a pipeline ingests a document, the exchange body is replaced with the IngestResult, so a request-reply caller receives the outcome of its call.

The same pipeline can be declared purely in properties: source.uri takes the consumer URI as written, source.document-id the header name — or a simple-language expression written $simple{...}, the one form MicroProfile Config leaves untouched. The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration. Treat runtime configuration as the trust boundary it is: the directory and the id expression decide what the process reads into an often external store.

A failure while splitting, embedding or storing — a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment. For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly. For a consumer-fed pipeline the component’s own error handling applies: a request-reply caller receives the exception, while a Kafka consumer with default settings logs the failure and commits the offset, so the record is dropped — and since this engine keeps no record either, nothing remembers it. A record whose configured document-id resolves to nothing (a Kafka record without a key, say) fails the same way, one exchange at a time.

In order to ensure alignment across all Quarkus and LangChain4j related dependencies, it is recommended to import the LangChain4j BOM as below:

Note that the import order is paramount when using maven dependencyManagement. As such, one might need to import the langchain4j-bom before other related Camel and Quarkus BOMs.

This extension is designed and tested to work together with the Quarkus LangChain4j extensions. The EmbeddingStore and EmbeddingModel beans a pipeline writes through are ordinary CDI beans shared by both stacks — including stores declared through Quarkus LangChain4j configuration — and with the RAG augmentor bridge an @RegisterAiService interface answers from the store a pipeline filled, without glue code. The integration tests run both stacks in one application.