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.

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.