CAMEL-20968 : Camel LangChain4j : add the RAG feature

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/langchain4j-chat.json

@@ -30,7 +30,8 @@
     "chatModel": { "index": 4, "kind": "property", "displayName": "Chat Model", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "dev.langchain4j.model.chat.ChatLanguageModel", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.langchain4j.chat.LangChain4jChatConfiguration", "configurationField": "configuration", "description": "Chat Language Model of type dev.langchain4j.model.chat.ChatLanguageModel" }
   },
   "headers": {
-    "CamelLangChain4jChatPromptTemplate": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The prompt Template.", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#PROMPT_TEMPLATE" }
+    "CamelLangChain4jChatPromptTemplate": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The prompt Template.", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#PROMPT_TEMPLATE" },
+    "CamelLangChain4jChatAugmentedData": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Augmented Data for RAG", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#AUGMENTED_DATA" }
   },
   "properties": {
     "chatId": { "index": 0, "kind": "path", "displayName": "Chat Id", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The id" },

components/camel-ai/camel-langchain4j-chat/src/generated/resources/META-INF/org/apache/camel/component/langchain4j/chat/langchain4j-chat.json

@@ -30,7 +30,8 @@
     "chatModel": { "index": 4, "kind": "property", "displayName": "Chat Model", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "dev.langchain4j.model.chat.ChatLanguageModel", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.langchain4j.chat.LangChain4jChatConfiguration", "configurationField": "configuration", "description": "Chat Language Model of type dev.langchain4j.model.chat.ChatLanguageModel" }
   },
   "headers": {
-    "CamelLangChain4jChatPromptTemplate": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The prompt Template.", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#PROMPT_TEMPLATE" }
+    "CamelLangChain4jChatPromptTemplate": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The prompt Template.", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#PROMPT_TEMPLATE" },
+    "CamelLangChain4jChatAugmentedData": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Augmented Data for RAG", "constantName": "org.apache.camel.component.langchain4j.chat.LangChain4jChat$Headers#AUGMENTED_DATA" }
   },
   "properties": {
     "chatId": { "index": 0, "kind": "path", "displayName": "Chat Id", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The id" },

components/camel-ai/camel-langchain4j-chat/src/main/docs/langchain4j-chat-component.adoc

@@ -180,3 +180,64 @@ List<ChatMessage> messages = new ArrayList<>();
 
         Exchange message = fluentTemplate.to("direct:test").withBody(messages).request(Exchange.class);
 ----
+
+== RAG (Retrieval Augmented Generation)
+Use the RAG feature to enrich exchanges with data retrieved from any type of Camel endpoint. The feature is compatible with all LangChain4 Chat operations and is ideal for orchestrating the RAG workflow, utilizing the extensive library of components and Enterprise Integration Patterns (EIPs) available in Apache Camel.
+
+There are two ways for utilizing the RAG feature:
+
+=== Using RAG with Content Enricher and LangChain4jRagAggregatorStrategy
+Enrich the exchange by retrieving a list of strings using any Camel producer. The `LangChain4jRagAggregatorStrategy` is specifically designed to augment data within LangChain4j chat producers.
+
+Example of usage:
+[source, java]
+----
+// Create an instance of the RAG aggregator strategy
+LangChain4jRagAggregatorStrategy aggregatorStrategy = new LangChain4jRagAggregatorStrategy();
+
+from("direct:test")
+     .enrich("direct:rag", aggregatorStrategy)
+     .to("langchain4j-chat:test1?chatOperation=CHAT_SIMPLE_MESSAGE");
+
+  from("direct:rag")
+          .process(exchange -> {
+                List<String> augmentedData = List.of("data 1", "data 2" );
+                exchange.getIn().setBody(augmentedData);
+              });
+----
+
+[NOTE]
+====
+This method leverages a separate Camel route to fetch and process the augmented data.
+====
+
+It is possible to enrich the message from multiple sources within the same exchange.
+
+Example of usage:
+[source, java]
+----
+// Create an instance of the RAG aggregator strategy
+LangChain4jRagAggregatorStrategy aggregatorStrategy = new LangChain4jRagAggregatorStrategy();
+
+from("direct:test")
+     .enrich("direct:rag-from-source-1", aggregatorStrategy)
+     .enrich("direct:rag-from-source-2", aggregatorStrategy)
+     .to("langchain4j-chat:test1?chatOperation=CHAT_SIMPLE_MESSAGE");
+----
+
+=== Using RAG with Header
+Directly add augmented data into the header. This method is particularly efficient for straightforward use cases where the augmented data is predefined or static.
+You must add augmented data as a List of `dev.langchain4j.rag.content.Content` directly inside the header `CamelLangChain4jChatAugmentedData`.
+
+Example of usage:
+[source, java]
+----
+import dev.langchain4j.rag.content.Content;
+
+...
+
+Content augmentedContent = new Content("data test");
+List<Content> contents = List.of(augmentedContent);
+
+String response = template.requestBodyAndHeader("direct:send-multiple", messages, LangChain4jChat.Headers.AUGMENTED_DATA , contents, String.class);
+----

components/camel-ai/camel-langchain4j-chat/src/main/java/org/apache/camel/component/langchain4j/chat/LangChain4jChat.java

@@ -28,5 +28,8 @@ private LangChain4jChat() {
     public static class Headers {
         @Metadata(description = "The prompt Template.", javaType = "String")
         public static final String PROMPT_TEMPLATE = "CamelLangChain4jChatPromptTemplate";
+
+        @Metadata(description = "Augmented Data for RAG", javaType = "String")
+        public static final String AUGMENTED_DATA = "CamelLangChain4jChatAugmentedData";
     }
 }

components/camel-ai/camel-langchain4j-chat/src/main/java/org/apache/camel/component/langchain4j/chat/LangChain4jChatProducer.java

@@ -27,10 +27,14 @@
 import dev.langchain4j.data.message.AiMessage;
 import dev.langchain4j.data.message.ChatMessage;
 import dev.langchain4j.data.message.ToolExecutionResultMessage;
+import dev.langchain4j.data.message.UserMessage;
 import dev.langchain4j.model.chat.ChatLanguageModel;
 import dev.langchain4j.model.input.Prompt;
 import dev.langchain4j.model.input.PromptTemplate;
 import dev.langchain4j.model.output.Response;
+import dev.langchain4j.rag.content.Content;
+import dev.langchain4j.rag.content.injector.ContentInjector;
+import dev.langchain4j.rag.content.injector.DefaultContentInjector;
 import org.apache.camel.Exchange;
 import org.apache.camel.InvalidPayloadException;
 import org.apache.camel.NoSuchHeaderException;
@@ -39,6 +43,8 @@
 import org.apache.camel.support.DefaultProducer;
 import org.apache.camel.util.ObjectHelper;
 
+import static org.apache.camel.component.langchain4j.chat.LangChain4jChat.Headers.AUGMENTED_DATA;
+
 public class LangChain4jChatProducer extends DefaultProducer {
 
     private final LangChain4jChatEndpoint endpoint;
@@ -76,19 +82,19 @@ private void processSingleMessageWithPrompt(Exchange exchange) throws NoSuchHead
 
         Map<String, Object> variables = (Map<String, Object>) exchange.getIn().getMandatoryBody(Map.class);
 
-        var response = sendWithPromptTemplate(promptTemplate, variables);
+        var response = sendWithPromptTemplate(promptTemplate, variables, exchange);
 
         populateResponse(response, exchange);
     }
 
     private void processSingleMessage(Exchange exchange) throws InvalidPayloadException {
+        // Retrieve the mandatory body from the exchange
         final var message = exchange.getIn().getMandatoryBody();
 
-        if (message instanceof String text) {
-            populateResponse(sendMessage(text), exchange);
-        } else if (message instanceof ChatMessage chatMessage) {
-            populateResponse(sendChatMessage(chatMessage), exchange);
-        }
+        // Use pattern matching with instanceof to streamline type checks and assignments
+        ChatMessage userMessage = (message instanceof String) ? new UserMessage((String) message) : (ChatMessage) message;
+
+        populateResponse(sendChatMessage(userMessage, exchange), exchange);
 
     }
 
@@ -110,24 +116,36 @@ private void populateResponse(String response, Exchange exchange) {
     }
 
     /**
-     * Send one simple message
+     * Send a ChatMessage
      *
-     * @param  message
+     * @param  chatMessage
      * @return
      */
-    public String sendMessage(String message) {
-        return this.chatLanguageModel.generate(message);
+    private String sendChatMessage(ChatMessage chatMessage, Exchange exchange) {
+        var augmentedChatMessage = addAugmentedData(chatMessage, exchange);
+
+        Response<AiMessage> response = this.chatLanguageModel.generate(augmentedChatMessage);
+        return extractAiResponse(response);
     }
 
     /**
-     * Send a ChatMessage
+     * Augment the message for RAG if the header is specified
      *
-     * @param  chatMessage
-     * @return
+     * @param chatMessage
+     * @param exchange
      */
-    private String sendChatMessage(ChatMessage chatMessage) {
-        Response<AiMessage> response = this.chatLanguageModel.generate(chatMessage);
-        return extractAiResponse(response);
+    private ChatMessage addAugmentedData(ChatMessage chatMessage, Exchange exchange) {
+        // check if there's any augmented data
+        List<Content> augmentedData = exchange.getIn().getHeader(AUGMENTED_DATA, List.class);
+
+        // inject data
+        if (augmentedData != null && augmentedData.size() != 0) {
+            ContentInjector contentInjector = new DefaultContentInjector();
+            // inject with new List of Contents
+            return contentInjector.inject(augmentedData, chatMessage);
+        } else {
+            return chatMessage;
+        }
     }
 
     /**
@@ -140,6 +158,18 @@ private String sendListChatMessage(List<ChatMessage> chatMessages, Exchange exch
         LangChain4jChatEndpoint langChain4jChatEndpoint = (LangChain4jChatEndpoint) getEndpoint();
 
         Response<AiMessage> response;
+
+        // Check if the last message is a UserMessage and if there's a need to augment the message for RAG
+        int size = chatMessages.size();
+        if (size != 0) {
+            int lastIndex = size - 1;
+            ChatMessage lastUserMessage = chatMessages.get(lastIndex);
+            if (lastUserMessage instanceof UserMessage) {
+                chatMessages.set(lastIndex, addAugmentedData(lastUserMessage, exchange));
+            }
+
+        }
+
         if (CamelToolExecutorCache.getInstance().getTools().containsKey(langChain4jChatEndpoint.getChatId())) {
             List<ToolSpecification> toolSpecifications = CamelToolExecutorCache.getInstance().getTools()
                     .get(langChain4jChatEndpoint.getChatId()).stream()
@@ -191,10 +221,10 @@ private String extractAiResponse(Response<AiMessage> response) {
         return message == null ? null : message.text();
     }
 
-    public String sendWithPromptTemplate(String promptTemplate, Map<String, Object> variables) {
+    public String sendWithPromptTemplate(String promptTemplate, Map<String, Object> variables, Exchange exchange) {
         PromptTemplate template = PromptTemplate.from(promptTemplate);
         Prompt prompt = template.apply(variables);
-        return this.sendMessage(prompt.text());
+        return this.sendChatMessage(new UserMessage(prompt.text()), exchange);
     }
 
 }

components/camel-ai/camel-langchain4j-chat/src/main/java/org/apache/camel/component/langchain4j/chat/rag/LangChain4jRagAggregatorStrategy.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.chat.rag;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import dev.langchain4j.rag.content.Content;
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+
+import static org.apache.camel.component.langchain4j.chat.LangChain4jChat.Headers.AUGMENTED_DATA;
+
+public class LangChain4jRagAggregatorStrategy implements AggregationStrategy {
+    @Override
+    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+        // In theory old exchange shouldn't be null
+        if (oldExchange == null) {
+            return newExchange;
+        }
+
+        // check that we got new Augmented Data
+        Optional<List<String>> newAugmentedData = Optional.ofNullable(newExchange.getIn().getBody(List.class));
+        if (newAugmentedData.isEmpty()) {
+            return oldExchange;
+        }
+
+        // create a list of contents from the retrieved Strings
+        List<Content> newContents = newAugmentedData.get().stream()
+                .map(Content::new)
+                .collect(Collectors.toList());
+
+        // Get or create the augmented data list from the old exchange
+        List<Content> augmentedData = Optional.ofNullable(oldExchange.getIn().getHeader(AUGMENTED_DATA, List.class))
+                .orElse(new ArrayList<Content>());
+        augmentedData.addAll(newContents);
+
+        // add the retrieved data in the body, langchain4j-chat will know it has to add inside the body
+        oldExchange.getIn().setHeader(AUGMENTED_DATA, augmentedData);
+
+        return oldExchange;
+    }
+}